From a4c370c5ba9522248efc1a19b70589d9a8cc912d Mon Sep 17 00:00:00 2001 From: Jin Hai Date: Fri, 3 Jul 2026 15:08:37 +0800 Subject: [PATCH] Go: fix 'list services' (#16598) ### Summary ``` RAGFlow(admin)> list services; +-----------------------------------------------------------------------------+-----------+----+---------------+------+---------------+-----------+ | extra | host | id | name | port | service_type | status | +-----------------------------------------------------------------------------+-----------+----+---------------+------+---------------+-----------+ | map[database:1 mq_type:redis password:infini_rag_flow] | localhost | 0 | redis | 6379 | message_queue | alive | | map[password:infini_rag_flow retrieval_type:elasticsearch username:elastic] | localhost | 1 | elasticsearch | 1200 | retrieval | alive | | | 0.0.0.0 | 2 | nats | 4222 | message_queue | CONNECTED | | map[meta_type:mysql password:infini_rag_flow username:root] | localhost | 3 | mysql | 3306 | meta_data | alive | | map[password:infini_rag_flow store_type:minio user:rag_flow] | localhost | 4 | minio | 9000 | file_store | alive | +-----------------------------------------------------------------------------+-----------+----+---------------+------+---------------+-----------+ ``` Signed-off-by: Jin Hai --- internal/admin/service.go | 55 +++++++++++++++++++++++++----------- internal/admin/state.go | 5 ++++ internal/engine/engine.go | 1 + internal/engine/nats/nats.go | 5 ++++ internal/server/config.go | 1 + 5 files changed, 51 insertions(+), 16 deletions(-) diff --git a/internal/admin/service.go b/internal/admin/service.go index bd559e6393..0dd3c89a4d 100644 --- a/internal/admin/service.go +++ b/internal/admin/service.go @@ -976,7 +976,7 @@ func (s *Service) GenerateUserAPIToken(username string) (map[string]interface{}, } // 4. Save API token - if err := s.apiTokenDAO.Create(apiToken); err != nil { + if err = s.apiTokenDAO.Create(apiToken); err != nil { return nil, fmt.Errorf("failed to generate API key: %w", err) } @@ -1024,7 +1024,7 @@ func (s *Service) DeleteUserAPIToken(username, key string) error { func (s *Service) ListServices() ([]map[string]interface{}, error) { allConfigs := server.GetAllConfigs() - var result []map[string]interface{} + var results []map[string]interface{} for _, configDict := range allConfigs { serviceType := configDict["service_type"] if serviceType != "ragflow_server" { @@ -1039,19 +1039,18 @@ func (s *Service) ListServices() ([]map[string]interface{}, error) { } else { configDict["status"] = "timeout" } - result = append(result, configDict) + if serviceDetail != nil { + results = append(results, configDict) + } } } - id := len(result) 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["id"] = id - id++ serverItem["host"] = serverStatus.Host serverItem["port"] = serverStatus.Port // the difference between now and serverStatus.Timestamp is less than 5 seconds, then the server is alive @@ -1060,9 +1059,14 @@ func (s *Service) ListServices() ([]map[string]interface{}, error) { } else { serverItem["status"] = "timeout" } - result = append(result, serverItem) + results = append(results, serverItem) } - return result, nil + + for id, result := range results { + result["id"] = id + } + + return results, nil } // GetServicesByType get services by type @@ -1080,7 +1084,20 @@ func (s *Service) GetServiceDetails(configDict map[string]interface{}) (map[stri case "meta_data": return s.getMySQLStatus(name) case "message_queue": - return s.getRedisInfo(name) + switch name { + case "redis": + return s.getRedisInfo(name) + case "nats": + host := configDict["host"].(string) + port := configDict["port"].(int) + return s.checkNatsAlive(name, host, port) + default: + return map[string]interface{}{ + "service_name": name, + "status": "unknown", + "message": "Service type not supported", + }, nil + } case "retrieval": // Check the extra.retrieval_type to determine which retrieval service if extra, ok := configDict["extra"].(map[string]interface{}); ok { @@ -1095,14 +1112,8 @@ func (s *Service) GetServiceDetails(configDict map[string]interface{}) (map[stri return s.checkRAGFlowServerAlive(name) case "file_store": return s.checkMinioAlive(name) - case "task_executor": - return s.checkTaskExecutorAlive(name) default: - return map[string]interface{}{ - "service_name": name, - "status": "unknown", - "message": "Service type not supported", - }, nil + return nil, nil } } @@ -1408,6 +1419,18 @@ func (s *Service) checkTaskExecutorAlive(name string) (map[string]interface{}, e }, nil } +// checkNatsAlive checks if NATS is alive +func (s *Service) checkNatsAlive(name string, ip string, port int) (map[string]interface{}, error) { + + msgQueueEngine := engine.GetMessageQueueEngine() + status := msgQueueEngine.CheckStatus() + + return map[string]interface{}{ + "service_name": name, + "status": status, + }, nil +} + // ShutdownService shutdown service func (s *Service) ShutdownService(serviceID string) (map[string]interface{}, error) { // TODO: Implement with proper service manager diff --git a/internal/admin/state.go b/internal/admin/state.go index 0bf94ef15f..45c08397ae 100644 --- a/internal/admin/state.go +++ b/internal/admin/state.go @@ -53,6 +53,11 @@ func (s *ServerStore) UpdateServerInfo(serverName string, status *common.BaseMes defer s.mu.Unlock() s.servers[serverName] = status return + case common.ServerTypeFileSyncer: + s.mu.Lock() + defer s.mu.Unlock() + s.servers[serverName] = status + return } } diff --git a/internal/engine/engine.go b/internal/engine/engine.go index 9a08ce8398..b3d6f8dc2c 100644 --- a/internal/engine/engine.go +++ b/internal/engine/engine.go @@ -99,4 +99,5 @@ type MessageQueue interface { GetMessages(messageCount int) ([]common.TaskHandle, error) ListMessages(messageType string, pending bool) ([]map[string]string, error) ShowMessageQueue() (map[string]string, error) + CheckStatus() string } diff --git a/internal/engine/nats/nats.go b/internal/engine/nats/nats.go index 30d189f9d2..55eaf41c05 100644 --- a/internal/engine/nats/nats.go +++ b/internal/engine/nats/nats.go @@ -222,6 +222,11 @@ func (n *NatsEngine) GetMessages(messageCount int) ([]common.TaskHandle, error) return resultMessages, nil } +func (n *NatsEngine) CheckStatus() string { + n.nc.Stats() + return n.nc.Status().String() +} + type NatsMessageHandle struct { message jetstream.Msg } diff --git a/internal/server/config.go b/internal/server/config.go index d2ac97f64a..61b8a09839 100644 --- a/internal/server/config.go +++ b/internal/server/config.go @@ -417,6 +417,7 @@ func Init(configPath string) error { configDict["name"] = "nats" configDict["host"] = host configDict["port"] = port + configDict["service_type"] = "message_queue" case "admin": // Skip admin section continue