From 26a5af8b83ffe664f733921f986ebbe1de2f9d0b Mon Sep 17 00:00:00 2001 From: Jin Hai Date: Sat, 1 Aug 2026 14:52:37 +0800 Subject: [PATCH] Go: fix list services (#17664) ### Summary 1. Fix admin list services; 2. Fix admin show service 'service_name'; Signed-off-by: Jin Hai --- internal/admin/handler.go | 63 ++- internal/admin/router.go | 8 +- internal/admin/service.go | 443 ++++++++------------- internal/admin/service_ee.go | 4 + internal/cli/admin_command.go | 28 +- internal/cli/admin_parser.go | 16 +- internal/engine/engine.go | 5 +- internal/engine/nats/nats.go | 4 + internal/engine/redis/redis.go | 2 +- internal/ingestion/testutil/mocks.go | 2 + internal/service/chunk/chunk_test.go | 1 + internal/service/document/document_test.go | 1 + internal/service/file/file_test.go | 2 + internal/storage/gcs.go | 2 + internal/storage/memory.go | 2 + internal/storage/minio.go | 2 + internal/storage/oss.go | 2 + internal/storage/s3.go | 2 + internal/storage/types.go | 2 + 19 files changed, 262 insertions(+), 329 deletions(-) diff --git a/internal/admin/handler.go b/internal/admin/handler.go index 37ad042f60..91bf959c38 100644 --- a/internal/admin/handler.go +++ b/internal/admin/handler.go @@ -490,7 +490,8 @@ func (h *Handler) DeleteUserAPIToken(c *gin.Context) { // GetServices handle get all services func (h *Handler) GetServices(c *gin.Context) { - services, err := h.service.ListServices() + ctx := c.Request.Context() + services, err := h.service.ListServices(ctx) if err != nil { common.ResponseWithHttpCodeData(c, http.StatusInternalServerError, common.CodeBadRequest, nil, err.Error()) return @@ -518,48 +519,38 @@ func (h *Handler) GetServicesByType(c *gin.Context) { // GetService handle get service details func (h *Handler) GetService(c *gin.Context) { - serviceID := c.Param("service_id") - if serviceID == "" { - common.ErrorWithCode(c, common.CodeBadRequest, "Service ID is required") + serviceName := c.Param("service_name") + if serviceName == "" { + common.ErrorWithCode(c, common.CodeBadRequest, "Service name is required") return } - // Get all services and find the one with matching ID - allConfigs := server.GetAllConfigs() - - var targetService map[string]interface{} - for _, config := range allConfigs { - if id, ok := config["id"]; ok { - if strconv.Itoa(id.(int)) == serviceID { - targetService = config - break - } - } - } - - if targetService == nil { - common.ErrorWithCode(c, common.CodeNotFound, "Service not found") - return - } - - serviceStatus, err := h.service.GetServiceDetails(targetService) + ctx := c.Request.Context() + services, err := h.service.ListServices(ctx) if err != nil { common.ErrorWithCode(c, common.CodeServerError, err.Error()) return } - common.SuccessWithData(c, serviceStatus, "") + for _, serviceInstance := range services { + if serviceInstance.Name == serviceName { + common.SuccessWithData(c, serviceInstance, "") + return + } + } + + common.ErrorWithCode(c, common.CodeNotFound, "Service not found") } // ShutdownService handle shutdown service func (h *Handler) ShutdownService(c *gin.Context) { - serviceID := c.Param("service_id") - if serviceID == "" { - common.ErrorWithCode(c, common.CodeBadRequest, "Service ID is required") + serviceName := c.Param("service_name") + if serviceName == "" { + common.ErrorWithCode(c, common.CodeBadRequest, "Service name is required") return } - result, err := h.service.ShutdownService(serviceID) + result, err := h.service.ShutdownService(serviceName) if err != nil { common.ErrorWithCode(c, common.CodeServerError, err.Error()) return @@ -570,13 +561,13 @@ func (h *Handler) ShutdownService(c *gin.Context) { // StartService handle start service func (h *Handler) StartService(c *gin.Context) { - serviceID := c.Param("service_id") - if serviceID == "" { - common.ErrorWithCode(c, common.CodeBadRequest, "Service ID is required") + serviceName := c.Param("service_name") + if serviceName == "" { + common.ErrorWithCode(c, common.CodeBadRequest, "Service name is required") return } - result, err := h.service.StartService(serviceID) + result, err := h.service.StartService(serviceName) if err != nil { common.ErrorWithCode(c, common.CodeServerError, err.Error()) return @@ -587,13 +578,13 @@ func (h *Handler) StartService(c *gin.Context) { // RestartService handle restart service func (h *Handler) RestartService(c *gin.Context) { - serviceID := c.Param("service_id") - if serviceID == "" { - common.ErrorWithCode(c, common.CodeBadRequest, "Service ID is required") + serviceName := c.Param("service_name") + if serviceName == "" { + common.ErrorWithCode(c, common.CodeBadRequest, "Service name is required") return } - result, err := h.service.RestartService(serviceID) + result, err := h.service.RestartService(serviceName) if err != nil { common.ErrorWithCode(c, common.CodeServerError, err.Error()) return diff --git a/internal/admin/router.go b/internal/admin/router.go index e6ee394d7f..e778bd01e7 100644 --- a/internal/admin/router.go +++ b/internal/admin/router.go @@ -75,10 +75,10 @@ func (r *Router) Setup(engine *gin.Engine) { // Service management protected.GET("/services", r.handler.GetServices) protected.GET("/service_types/:service_type", r.handler.GetServicesByType) - protected.GET("/services/:service_id", r.handler.GetService) - protected.DELETE("/services/:service_id", r.handler.ShutdownService) - protected.PUT("/services/:service_id", r.handler.RestartService) - protected.POST("/services/:service_id", r.handler.StartService) + protected.GET("/services/:service_name", r.handler.GetService) + protected.DELETE("/services/:service_name", r.handler.ShutdownService) + protected.PUT("/services/:service_name", r.handler.RestartService) + protected.POST("/services/:service_name", r.handler.StartService) // Variables/Settings protected.GET("/variables", r.handler.ListVariables) diff --git a/internal/admin/service.go b/internal/admin/service.go index 389b828c49..e884c3c172 100644 --- a/internal/admin/service.go +++ b/internal/admin/service.go @@ -19,7 +19,6 @@ package admin import ( "context" "crypto/rand" - "crypto/tls" "encoding/base64" "encoding/hex" "errors" @@ -36,9 +35,9 @@ import ( "ragflow/internal/server" "ragflow/internal/server/config" servicepkg "ragflow/internal/service" + "ragflow/internal/storage" "ragflow/internal/utility" "regexp" - "strconv" "time" "go.uber.org/zap" @@ -963,60 +962,88 @@ func (s *Service) DeleteUserAPIToken(ctx context.Context, username, key string) return nil } -// ListServices get all services -func (s *Service) ListServices() ([]map[string]interface{}, error) { - allConfigs := server.GetAllConfigs() +type ServiceStatus struct { + Type string `json:"type"` + Name string `json:"name"` + Status string `json:"status"` + Elapsed string `json:"elapsed"` + Message string `json:"message"` +} - var results []map[string]interface{} - for _, configDict := range allConfigs { - serviceType := configDict["service_type"] - if serviceType != "ragflow_server" { - // Get service details to check status - serviceDetail, err := s.GetServiceDetails(configDict) - if err == nil { - if status, ok := serviceDetail["status"]; ok { - configDict["status"] = status - } else { - configDict["status"] = "timeout" - } - } else { - configDict["status"] = "timeout" - } - if serviceDetail != nil { - // delete element of configDict - delete(configDict, "extra") - delete(configDict, "database") - delete(configDict, "password") - delete(configDict, "sample_ratio") - delete(configDict, "secure") - delete(configDict, "stdout") - delete(configDict, "user") - results = append(results, configDict) - } - } +func newServiceStatus(typeStr, nameStr, statusStr string, startTime time.Time, messageStr string) ServiceStatus { + return ServiceStatus{ + Type: typeStr, + Name: nameStr, + Status: statusStr, + Elapsed: fmt.Sprintf("%.1d", time.Since(startTime).Milliseconds()), + Message: messageStr, } +} + +// ListServices get all services +func (s *Service) ListServices(ctx context.Context) ([]ServiceStatus, error) { + + var results []ServiceStatus + + globalConfig := server.GetConfig() + // Database + databaseType := globalConfig.DatabaseType() + switch databaseType { + case "mysql": + mysqlStatus := s.getMySQLStatus(ctx) + results = append(results, mysqlStatus) + default: + results = append(results, newServiceStatus("database", databaseType, "not available", time.Now(), "not supported database type")) + } + + // Doc engine + docEngineImpl := engine.Get() + err := docEngineImpl.Ping(ctx) + if err == nil { + results = append(results, newServiceStatus("doc_engine", docEngineImpl.GetType(), "alive", time.Now(), "")) + } else { + results = append(results, newServiceStatus("doc_engine", docEngineImpl.GetType(), "timeout", time.Now(), err.Error())) + } + + // storage engine + storageImpl := storage.GetStorageFactory().GetStorage() + storageHealth := storageImpl.Health() + if storageHealth { + results = append(results, newServiceStatus("storage_engine", storageImpl.Type(), "alive", time.Now(), "")) + } else { + results = append(results, newServiceStatus("storage_engine", storageImpl.Type(), "timeout", time.Now(), "")) + } + + // cache engine + cacheType := globalConfig.CacheEngineType() + switch cacheType { + case "redis": + mysqlStatus := s.getRedisInfo(ctx) + results = append(results, mysqlStatus) + default: + results = append(results, newServiceStatus("database", databaseType, "not available", time.Now(), "not supported database type")) + } + + // message queue + messageQueueImpl := engine.GetMessageQueueEngine() + messageQueueStatus := messageQueueImpl.CheckStatus() + results = append(results, newServiceStatus("message_queue", messageQueueImpl.Type(), messageQueueStatus, time.Now(), "")) + + results = append(results, s.GetEEServicesStatus()...) serverList := GlobalServerStore.ListInfos() - now := time.Now() for _, serverStatus := range serverList { - serverItem := make(map[string]interface{}) - serverItem["name"] = serverStatus.ServerName - serverItem["service_type"] = serverStatus.ServerType - serverItem["host"] = serverStatus.Host - serverItem["port"] = serverStatus.Port + now := time.Now() + serverItem := newServiceStatus(string(serverStatus.ServerType), serverStatus.ServerName, "", serverStatus.Timestamp, "") // the difference between now and serverStatus.Timestamp is less than 5 seconds, then the server is alive - if now.Sub(serverStatus.Timestamp) < 30*time.Second { - serverItem["status"] = "alive" + if now.Sub(serverStatus.Timestamp) < 45*time.Second { + serverItem.Status = "alive" } else { - serverItem["status"] = "timeout" + serverItem.Status = "timeout" } results = append(results, serverItem) } - for id, result := range results { - result["id"] = id - } - return results, nil } @@ -1026,169 +1053,143 @@ func (s *Service) GetServicesByType(serviceType string) ([]map[string]interface{ } // GetServiceDetails get service details -func (s *Service) GetServiceDetails(configDict map[string]interface{}) (map[string]interface{}, error) { - serviceType, _ := configDict["service_type"].(string) - name, _ := configDict["name"].(string) - - // Call detail function based on service type - switch serviceType { - case "meta_data": - return s.getMySQLStatus(name) - case "cache": - return s.getRedisInfo(name) - case "message_queue": - host := configDict["host"].(string) - port := configDict["port"].(int) - return s.checkNatsAlive(name, host, port) - case "retrieval": - // Check the extra.retrieval_type to determine which retrieval service - if extra, ok := configDict["extra"].(map[string]interface{}); ok { - if retrievalType, ok := extra["retrieval_type"].(string); ok { - if retrievalType == "infinity" { - return s.getInfinityStatus(name) - } - } - } - return s.getESClusterStats(name) - case "ragflow_server": - return s.checkRAGFlowServerAlive(name) - case "file_store": - return s.checkMinioAlive(name) - case "olap": - return s.checkOlapAlive(name) - case "tracing": - return s.checkTracingAlive(name) - default: - return nil, nil - } +func (s *Service) GetServiceDetails(configDict map[string]interface{}) ([]ServiceStatus, error) { + //serviceType, _ := configDict["service_type"].(string) + //name, _ := configDict["name"].(string) + //ctx := context.Background() + // + //// Call detail function based on service type + //switch serviceType { + //case "meta_data": + // return s.getMySQLStatus(ctx), nil + //case "cache": + // return s.getRedisInfo(ctx), nil + //case "message_queue": + // host := configDict["host"].(string) + // port := configDict["port"].(int) + // return s.checkNatsAlive(name, host, port) + //case "retrieval": + // // Check the extra.retrieval_type to determine which retrieval service + // if extra, ok := configDict["extra"].(map[string]interface{}); ok { + // if retrievalType, ok := extra["retrieval_type"].(string); ok { + // if retrievalType == "infinity" { + // return s.getInfinityStatus(name) + // } + // } + // } + // return s.getESClusterStats(name) + //case "ragflow_server": + // return s.checkRAGFlowServerAlive(name) + //case "file_store": + // return s.checkMinioAlive(name) + //case "olap": + // return s.checkOlapAlive(name) + //case "tracing": + // return s.checkTracingAlive(name) + //default: + // return nil, nil + //} + return nil, nil } // getMySQLStatus gets MySQL service status -func (s *Service) getMySQLStatus(name string) (map[string]interface{}, error) { +func (s *Service) getMySQLStatus(ctx context.Context) ServiceStatus { + + serviceType := "database" + name := "mysql" startTime := time.Now() - // Check basic connectivity with SELECT 1 sqlDB, err := dao.DB.DB() if err != nil { - return map[string]interface{}{ - "service_name": name, - "status": "timeout", - "elapsed": fmt.Sprintf("%.1d", time.Since(startTime).Milliseconds()), - "message": err.Error(), - }, nil + return newServiceStatus(serviceType, name, "not connected", startTime, err.Error()) } // Execute SELECT 1 to check connectivity - _, err = sqlDB.Exec("SELECT 1") + err = sqlDB.PingContext(ctx) if err != nil { - return map[string]interface{}{ - "service_name": name, - "status": "timeout", - "elapsed": fmt.Sprintf("%.1d", time.Since(startTime).Milliseconds()), - "message": err.Error(), - }, nil + return newServiceStatus(serviceType, name, "timeout", startTime, err.Error()) } - return map[string]interface{}{ - "service_name": name, - "status": "alive", - "elapsed": fmt.Sprintf("%.1d", time.Since(startTime).Milliseconds()), - "message": "MySQL connection successful", - }, nil + return newServiceStatus(serviceType, name, "alive", startTime, "") } // getRedisInfo gets Redis service info -func (s *Service) getRedisInfo(name string) (map[string]interface{}, error) { +func (s *Service) getRedisInfo(ctx context.Context) ServiceStatus { + + serviceType := "cache" + name := "redis" + startTime := time.Now() redisClient := redis.Get() - if redisClient == nil { - return map[string]interface{}{ - "service_name": name, - "status": "timeout", - "elapsed": fmt.Sprintf("%.1d", time.Since(startTime).Milliseconds()), - "error": "Redis client not initialized", - }, nil + if redisClient.Health() { + return newServiceStatus(serviceType, name, "alive", startTime, "") } - // Check health - if !redisClient.Health() { - return map[string]interface{}{ - "service_name": name, - "status": "timeout", - "elapsed": fmt.Sprintf("%.1d", time.Since(startTime).Milliseconds()), - "error": "Redis health check failed", - }, nil - } - - return map[string]interface{}{ - "service_name": name, - "status": "alive", - "elapsed": fmt.Sprintf("%.1d", time.Since(startTime).Milliseconds()), - "message": "Redis connection successful", - }, nil + return newServiceStatus(serviceType, name, "timeout", startTime, "Redis health check failed") } // getESClusterStats gets Elasticsearch cluster stats -func (s *Service) getESClusterStats(name string) (map[string]interface{}, error) { - // Check if Elasticsearch is the doc engine - docEngine := common.GetEnv(common.EnvDocEngine) - if docEngine == "" { - docEngine = "elasticsearch" - } - if docEngine != "elasticsearch" { - return map[string]interface{}{ - "service_name": name, - "status": "timeout", - "message": "error: Elasticsearch is not in use.", - }, nil - } +func (s *Service) getESClusterStats(serviceType string) map[string]interface{} { + name := "elasticsearch" + startTime := time.Now() - // Get ES config from server config cfg := server.GetConfig() if cfg == nil || !cfg.IsElasticConfigured() { return map[string]interface{}{ - "service_name": name, - "status": "timeout", - "message": "error: Elasticsearch configuration not found", - }, nil + "type": serviceType, + "name": name, + "status": "timeout", + "elapsed": fmt.Sprintf("%.1d", time.Since(startTime).Milliseconds()), + "message": "error: Elasticsearch configuration not found", + } } // Create ES engine and get cluster stats esEngine, err := elasticsearch.NewEngine(cfg.GetElasticsearchConfig()) if err != nil { return map[string]interface{}{ - "service_name": name, - "status": "timeout", - "message": fmt.Sprintf("error: %s", err.Error()), - }, nil + "type": serviceType, + "name": name, + "status": "timeout", + "elapsed": fmt.Sprintf("%.1d", time.Since(startTime).Milliseconds()), + "message": fmt.Sprintf("error: %s", err.Error()), + } } defer esEngine.Close() clusterStats, err := esEngine.GetClusterStats() if err != nil { return map[string]interface{}{ - "service_name": name, - "status": "timeout", - "message": fmt.Sprintf("error: %s", err.Error()), - }, nil + "type": serviceType, + "name": name, + "status": "timeout", + "elapsed": fmt.Sprintf("%.1d", time.Since(startTime).Milliseconds()), + "message": fmt.Sprintf("error: %s", err.Error()), + } } return map[string]interface{}{ - "service_name": name, - "status": "alive", - "message": clusterStats, - }, nil + "type": serviceType, + "name": name, + "status": "alive", + "elapsed": fmt.Sprintf("%.1d", time.Since(startTime).Milliseconds()), + "message": clusterStats, + } } // getInfinityStatus gets Infinity service status -func (s *Service) getInfinityStatus(name string) (map[string]interface{}, error) { +func (s *Service) getInfinityStatus(serviceType string) map[string]interface{} { + name := "infinity" + startTime := time.Now() // TODO: Implement actual Infinity health check return map[string]interface{}{ - "service_name": name, - "status": "unknown", - "message": "Infinity health check not implemented", - }, nil + "type": serviceType, + "name": name, + "status": "unknown", + "elapsed": fmt.Sprintf("%.1d", time.Since(startTime).Milliseconds()), + "message": "Infinity health check not implemented", + } } // checkRAGFlowServerAlive checks if RAGFlow server is alive @@ -1257,104 +1258,6 @@ func (s *Service) checkRAGFlowServerAlive(name string) (map[string]interface{}, }, nil } -// checkMinioAlive checks if MinIO is alive -func (s *Service) checkMinioAlive(name string) (map[string]interface{}, error) { - startTime := time.Now() - - // Get MinIO file store config from allConfigs - var host string - var port int - var secure bool - verify := true - - allConfigs := server.GetAllConfigs() - for _, config := range allConfigs { - if serviceType, ok := config["service_type"].(string); ok && serviceType == "file_store" { - // Get host from config - if h, ok := config["host"].(string); ok { - host = h - } - - if p, ok := config["port"].(int); ok { - port = p - } else if p, ok := config["port"].(float64); ok { - port = int(p) - } else if p, ok := config["port"].(string); ok { - if parsedPort, err := strconv.Atoi(p); err == nil { - port = parsedPort - } - } - // Get secure from extra config - if extra, ok := config["extra"].(map[string]interface{}); ok { - if s, ok := extra["secure"].(bool); ok { - secure = s - } else if s, ok := extra["secure"].(string); ok { - secure = s == "true" || s == "1" || s == "yes" - } - if v, ok := extra["verify"].(bool); ok { - verify = v - } else if v, ok := extra["verify"].(string); ok { - verify = !(v == "false" || v == "0" || v == "no") - } - } - break - } - } - - // Default host - if host == "" { - host = "localhost" - } - if port == 0 { - port = 9000 - } - - // Determine scheme - scheme := "http" - if secure { - scheme = "https" - } - - url := fmt.Sprintf("%s://%s:%d/minio/health/live", scheme, host, port) - - // Create HTTP client with timeout - client := &http.Client{ - Timeout: 10 * time.Second, - } - - // If verify is false, we need to skip SSL verification - if !verify && scheme == "https" { - client.Transport = &http.Transport{ - TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, - } - } - - resp, err := client.Get(url) - if err != nil { - return map[string]interface{}{ - "service_name": name, - "status": "timeout", - "message": fmt.Sprintf("error: %s", err.Error()), - }, nil - } - defer resp.Body.Close() - - elapsed := time.Since(startTime).Milliseconds() - if resp.StatusCode == 200 { - return map[string]interface{}{ - "service_name": name, - "status": "alive", - "message": fmt.Sprintf("Confirm elapsed: %.1f ms.", float64(elapsed)), - }, nil - } - - return map[string]interface{}{ - "service_name": name, - "status": "timeout", - "message": fmt.Sprintf("Confirm elapsed: %.1f ms.", float64(elapsed)), - }, nil -} - // checkTaskExecutorAlive checks if task executor is alive func (s *Service) checkTaskExecutorAlive(name string) (map[string]interface{}, error) { // TODO: Implement actual task executor health check @@ -1420,30 +1323,30 @@ func (s *Service) checkOlapAlive(name string) (map[string]interface{}, error) { } // ShutdownService shutdown service -func (s *Service) ShutdownService(serviceID string) (map[string]interface{}, error) { +func (s *Service) ShutdownService(serviceName string) (map[string]interface{}, error) { // TODO: Implement with proper service manager return map[string]interface{}{ - "command": "shutdown service", - "service_id": serviceID, - "error": "shutdown service not implemented", + "command": "shutdown service", + "service_name": serviceName, + "error": "shutdown service not implemented", }, nil } // StartService start service -func (s *Service) StartService(serviceID string) (map[string]interface{}, error) { +func (s *Service) StartService(serviceName string) (map[string]interface{}, error) { return map[string]interface{}{ - "command": "start service", - "service_id": serviceID, - "error": "command 'start service' isn't implemented", + "command": "start service", + "service_name": serviceName, + "error": "command 'start service' isn't implemented", }, nil } // RestartService restart service -func (s *Service) RestartService(serviceID string) (map[string]interface{}, error) { +func (s *Service) RestartService(serviceName string) (map[string]interface{}, error) { return map[string]interface{}{ - "command": "restart service", - "service_id": serviceID, - "error": "command 'restart service' isn't implemented", + "command": "restart service", + "service_name": serviceName, + "error": "command 'restart service' isn't implemented", }, nil } diff --git a/internal/admin/service_ee.go b/internal/admin/service_ee.go index 2c4bf8f7ed..9702e2a982 100644 --- a/internal/admin/service_ee.go +++ b/internal/admin/service_ee.go @@ -1351,3 +1351,7 @@ func (s *Service) ListLogs(ctx context.Context, userName string, days int) ([]ma } return result, nil } + +func (s *Service) GetEEServicesStatus() []ServiceStatus { + return []ServiceStatus{} +} diff --git a/internal/cli/admin_command.go b/internal/cli/admin_command.go index 27959de0ee..8d11b80706 100644 --- a/internal/cli/admin_command.go +++ b/internal/cli/admin_command.go @@ -586,9 +586,12 @@ func (c *CLI) AdminStartServiceCommand(cmd *Command) (ResponseIf, error) { return nil, fmt.Errorf("this command is only allowed in ADMIN mode or already login") } - serviceIndex := cmd.Params["service_index"].(int) + serviceName, ok := cmd.Params["service_name"].(string) + if !ok { + return nil, fmt.Errorf("service_name not provided") + } - endPoint := fmt.Sprintf("/admin/services/%d", serviceIndex) + endPoint := fmt.Sprintf("/admin/services/%s", serviceName) resp, err := c.AdminServerClient.Request("POST", endPoint, "admin", nil, nil) if err != nil { @@ -604,9 +607,12 @@ func (c *CLI) AdminRestartServiceCommand(cmd *Command) (ResponseIf, error) { return nil, fmt.Errorf("this command is only allowed in ADMIN mode or already login") } - serviceIndex := cmd.Params["service_index"].(int) + serviceName, ok := cmd.Params["service_name"].(string) + if !ok { + return nil, fmt.Errorf("service_name not provided") + } - endPoint := fmt.Sprintf("/admin/services/%d", serviceIndex) + endPoint := fmt.Sprintf("/admin/services/%s", serviceName) resp, err := c.AdminServerClient.Request("PUT", endPoint, "admin", nil, nil) if err != nil { @@ -622,9 +628,12 @@ func (c *CLI) AdminShutdownServiceCommand(cmd *Command) (ResponseIf, error) { return nil, fmt.Errorf("this command is only allowed in ADMIN mode or already login") } - serviceIndex := cmd.Params["service_index"].(int) + serviceName, ok := cmd.Params["service_name"].(string) + if !ok { + return nil, fmt.Errorf("service_name not provided") + } - endPoint := fmt.Sprintf("/admin/services/%d", serviceIndex) + endPoint := fmt.Sprintf("/admin/services/%s", serviceName) resp, err := c.AdminServerClient.Request("DELETE", endPoint, "admin", nil, nil) if err != nil { @@ -640,9 +649,12 @@ func (c *CLI) AdminShowService(cmd *Command) (ResponseIf, error) { return nil, fmt.Errorf("this command is only allowed in ADMIN mode or already login") } - serviceIndex := cmd.Params["service_index"].(int) + serviceName, ok := cmd.Params["service_name"].(string) + if !ok { + return nil, fmt.Errorf("service_name not provided") + } - endPoint := fmt.Sprintf("/admin/services/%d", serviceIndex) + endPoint := fmt.Sprintf("/admin/services/%s", serviceName) resp, err := c.AdminServerClient.Request("GET", endPoint, "admin", nil, nil) if err != nil { diff --git a/internal/cli/admin_parser.go b/internal/cli/admin_parser.go index b87569cc9b..448893870d 100644 --- a/internal/cli/admin_parser.go +++ b/internal/cli/admin_parser.go @@ -435,13 +435,13 @@ func (p *Parser) parseAdminShowCommands() (*Command, error) { func (p *Parser) parseAdminShowService() (*Command, error) { p.nextToken() // consume SERVICE - serviceIndex, err := p.parseNumber() + serviceName, err := p.parseQuotedString() if err != nil { return nil, err } cmd := NewCommand("admin_show_service") - cmd.Params["service_index"] = serviceIndex + cmd.Params["service_name"] = serviceName p.nextToken() // Semicolon is optional @@ -2097,13 +2097,13 @@ func (p *Parser) parseAdminStartService() (*Command, error) { } p.nextToken() - serviceIndex, err := p.parseNumber() + serviceName, err := p.parseQuotedString() if err != nil { return nil, err } cmd := NewCommand("admin_start_service") - cmd.Params["service_index"] = serviceIndex + cmd.Params["service_name"] = serviceName p.nextToken() // Semicolon is optional @@ -2129,13 +2129,13 @@ func (p *Parser) parseAdminShutdownCommands() (*Command, error) { func (p *Parser) parseAdminShutdownService() (*Command, error) { p.nextToken() // consume SERVICE - serviceIndex, err := p.parseNumber() + serviceName, err := p.parseQuotedString() if err != nil { return nil, err } cmd := NewCommand("admin_shutdown_service") - cmd.Params["service_index"] = serviceIndex + cmd.Params["service_name"] = serviceName p.nextToken() // Semicolon is optional @@ -2171,13 +2171,13 @@ func (p *Parser) parseAdminRestart() (*Command, error) { } p.nextToken() - serviceIndex, err := p.parseNumber() + serviceName, err := p.parseQuotedString() if err != nil { return nil, err } cmd := NewCommand("admin_restart_service") - cmd.Params["service_index"] = serviceIndex + cmd.Params["service_name"] = serviceName p.nextToken() // Semicolon is optional diff --git a/internal/engine/engine.go b/internal/engine/engine.go index 066d893e6d..24c5f80e78 100644 --- a/internal/engine/engine.go +++ b/internal/engine/engine.go @@ -66,14 +66,14 @@ type DocEngine interface { GetAggregation(chunks []map[string]interface{}, fieldName string) []map[string]interface{} GetHighlight(chunks []map[string]interface{}, keywords []string, fieldName string) map[string]string - // Run SQL + // RunSQL runs a SQL query RunSQL(ctx context.Context, tableName string, sqlText string, kbIDs []string, format string) ([]map[string]interface{}, error) GetChunkIDs(chunks []map[string]interface{}) []string KNNScores(ctx context.Context, chunks []map[string]interface{}, queryVector []float64, topK int) (map[string]interface{}, error) GetScores(searchResult map[string]interface{}) map[string]float64 - // Health check + // Ping check the engine is alive Ping(ctx context.Context) error Close() error @@ -101,6 +101,7 @@ func Type(docEngine DocEngine) EngineType { type MessageQueue interface { Init() error + Type() string InitConsumer(subject string) error PublishTask(subject string, payload []byte) error GetMessages(messageCount int) ([]common.TaskHandle, error) diff --git a/internal/engine/nats/nats.go b/internal/engine/nats/nats.go index 273ec17bd7..a0269b1cc3 100644 --- a/internal/engine/nats/nats.go +++ b/internal/engine/nats/nats.go @@ -97,6 +97,10 @@ func (n *NatsEngine) Init() error { return nil } +func (n *NatsEngine) Type() string { + return "nats" +} + func (n *NatsEngine) PublishTask(subject string, payload []byte) error { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() diff --git a/internal/engine/redis/redis.go b/internal/engine/redis/redis.go index 00f7023ff8..d32d84d9a7 100644 --- a/internal/engine/redis/redis.go +++ b/internal/engine/redis/redis.go @@ -374,7 +374,7 @@ func (r *Client) SetNX(key string, value string, exp time.Duration) bool { return ok } -// GetOrCreateSecretKey atomically retrieves an existing key or creates a new one +// GetOrCreateKey atomically retrieves an existing key or creates a new one // Uses Redis SETNX command to ensure atomicity across multiple goroutines/processes func (r *Client) GetOrCreateKey(key string, value string) (string, error) { if r.client == nil { diff --git a/internal/ingestion/testutil/mocks.go b/internal/ingestion/testutil/mocks.go index 75ee1a9d6a..a64358c03c 100644 --- a/internal/ingestion/testutil/mocks.go +++ b/internal/ingestion/testutil/mocks.go @@ -100,6 +100,8 @@ func NewMockStorageWithError(err error) *MockStorage { return &MockStorage{Err: err} } +func (m *MockStorage) Type() string { return "mock_storage" } + func (m *MockStorage) Health() bool { return true } func (m *MockStorage) Get(bucket, objectName string, tenantID ...string) ([]byte, error) { diff --git a/internal/service/chunk/chunk_test.go b/internal/service/chunk/chunk_test.go index 02d3a01540..b2555a6f44 100644 --- a/internal/service/chunk/chunk_test.go +++ b/internal/service/chunk/chunk_test.go @@ -1184,6 +1184,7 @@ type chunkImageStorage struct { putCalls int } +func (s *chunkImageStorage) Type() string { return "chunk_image_storage" } func (s *chunkImageStorage) Health() bool { return true } func (s *chunkImageStorage) Put(bucket, fnm string, binary []byte, tenantID ...string) error { s.putCalls++ diff --git a/internal/service/document/document_test.go b/internal/service/document/document_test.go index 29d02202d6..1ab2613baf 100644 --- a/internal/service/document/document_test.go +++ b/internal/service/document/document_test.go @@ -62,6 +62,7 @@ func newFakeUploadStorage() *fakeUploadStorage { return &fakeUploadStorage{objects: map[string][]byte{}} } +func (f *fakeUploadStorage) Type() string { return "fake_upload_storage" } func (f *fakeUploadStorage) Health() bool { return true } func (f *fakeUploadStorage) key(bucket, fnm string) string { return bucket + "/" + fnm } func (f *fakeUploadStorage) Put(bucket, fnm string, binary []byte, tenantID ...string) error { diff --git a/internal/service/file/file_test.go b/internal/service/file/file_test.go index 065a3786e3..65b9af04d7 100644 --- a/internal/service/file/file_test.go +++ b/internal/service/file/file_test.go @@ -45,6 +45,8 @@ func testFileService() *FileService { } } +func (f *fakeStorage) Type() string { return "fake_storage" } + func (f *fakeStorage) Health() bool { return true } diff --git a/internal/storage/gcs.go b/internal/storage/gcs.go index 9d97b67a9f..1310eac49d 100644 --- a/internal/storage/gcs.go +++ b/internal/storage/gcs.go @@ -68,6 +68,8 @@ func (m *GCSStorage) reconnect() { } } +func (m *GCSStorage) Type() string { return "gcs" } + // Health checks GCS service availability func (m *GCSStorage) Health() bool { return m.BucketExists(m.config.Bucket) diff --git a/internal/storage/memory.go b/internal/storage/memory.go index d5bd273b46..145a324f2e 100644 --- a/internal/storage/memory.go +++ b/internal/storage/memory.go @@ -46,6 +46,8 @@ func NewMemoryStorage() Storage { return &MemoryStorage{objects: make(map[string]map[string][]byte)} } +func (m *MemoryStorage) Type() string { return "memory_storage" } + // Health always reports healthy for the in-memory backend. func (m *MemoryStorage) Health() bool { return true diff --git a/internal/storage/minio.go b/internal/storage/minio.go index 41fccb9653..cccf13f450 100644 --- a/internal/storage/minio.go +++ b/internal/storage/minio.go @@ -107,6 +107,8 @@ func (m *MinioStorage) resolveBucketAndPath(bucket, fnm string) (string, string) return actualBucket, actualPath } +func (m *MinioStorage) Type() string { return "minio" } + // Health checks MinIO service availability func (m *MinioStorage) Health() bool { cancelFunction, err := m.client.HealthCheck(time.Second * 5) diff --git a/internal/storage/oss.go b/internal/storage/oss.go index 30936259b9..3adba24b36 100644 --- a/internal/storage/oss.go +++ b/internal/storage/oss.go @@ -103,6 +103,8 @@ func (o *OSSStorage) resolveBucketAndPath(bucket, fnm string) (string, string) { return actualBucket, actualPath } +func (o *OSSStorage) Type() string { return "oss" } + // Health checks OSS service availability func (o *OSSStorage) Health() bool { bucket := o.bucket diff --git a/internal/storage/s3.go b/internal/storage/s3.go index 78a11d9cde..82e381abe0 100644 --- a/internal/storage/s3.go +++ b/internal/storage/s3.go @@ -111,6 +111,8 @@ func (s *S3Storage) resolveBucketAndPath(bucket, fnm string) (string, string) { return actualBucket, actualPath } +func (s *S3Storage) Type() string { return "s3" } + // Health checks S3 service availability func (s *S3Storage) Health() bool { bucket := s.bucket diff --git a/internal/storage/types.go b/internal/storage/types.go index cdee2b71ca..bf778db1fe 100644 --- a/internal/storage/types.go +++ b/internal/storage/types.go @@ -56,6 +56,8 @@ func (s StorageType) String() string { // Storage defines the interface for storage operations type Storage interface { + Type() string + // Health checks the storage service availability Health() bool