From cf71c7193a4558c2c7f0d03b9b1b78cf0eee9f64 Mon Sep 17 00:00:00 2001 From: Jin Hai Date: Fri, 17 Jul 2026 13:54:19 +0800 Subject: [PATCH] Go: update server config (#17027) ### Summary ``` RAGFlow(admin)> list services; +--------+--------------+----+---------------------+------+---------------+-----------+ | enable | host | id | name | port | service_type | status | +--------+--------------+----+---------------------+------+---------------+-----------+ | | localhost | 0 | redis | 6379 | cache | alive | | | localhost | 1 | minio | 9000 | file_store | alive | | | localhost | 2 | elasticsearch | 1200 | retrieval | alive | | | localhost | 3 | mysql | 3306 | meta_data | alive | | false | localhost | 4 | jaeger | 4318 | tracing | unknown | | | localhost | 5 | clickhouse | 9900 | olap | unknown | | | localhost | 6 | nats | 4222 | message_queue | CONNECTED | | | 192.168.1.68 | 7 | ragflow-server-9384 | 9384 | api_server | alive | +--------+--------------+----+---------------------+------+---------------+-----------+ ``` Signed-off-by: Jin Hai --- cmd/ragflow_server.go | 2 +- conf/service_conf.yaml | 5 ++++- internal/admin/service.go | 21 ++++++--------------- internal/server/config.go | 16 +++++++++++----- 4 files changed, 22 insertions(+), 22 deletions(-) diff --git a/cmd/ragflow_server.go b/cmd/ragflow_server.go index 76fc0d9de5..6aa8f9ac74 100644 --- a/cmd/ragflow_server.go +++ b/cmd/ragflow_server.go @@ -359,7 +359,7 @@ func main() { } defer storage.CloseStorage() - if err = engine.InitMessageQueueEngine(config.TaskExecutor.MessageQueueType); err != nil { + if err = engine.InitMessageQueueEngine(config.Ingestor.MQType); err != nil { common.Error("Failed to initialize message queue engine", err) } diff --git a/conf/service_conf.yaml b/conf/service_conf.yaml index cf6f7acd56..a0fc3fdc0c 100644 --- a/conf/service_conf.yaml +++ b/conf/service_conf.yaml @@ -62,7 +62,10 @@ otel: port: 4318 secure: false sample_ratio: 1.0 - stdout: true + stdout: false + enable: false +ingestor: + mq_type: 'nats' task_executor: message_queue_type: 'nats' file_syncer: diff --git a/internal/admin/service.go b/internal/admin/service.go index d9a29fe879..d4cbd283d2 100644 --- a/internal/admin/service.go +++ b/internal/admin/service.go @@ -1022,21 +1022,12 @@ func (s *Service) GetServiceDetails(configDict map[string]interface{}) (map[stri switch serviceType { case "meta_data": return s.getMySQLStatus(name) + case "cache": + return s.getRedisInfo(name) case "message_queue": - 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 - } + 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 { @@ -1396,7 +1387,7 @@ func (s *Service) checkOlapAlive(name string) (map[string]interface{}, error) { if clickhouseDriver == nil { return map[string]interface{}{ "service_name": name, - "status": "ClickHouse engine not initialized", + "status": "unknown", }, nil } diff --git a/internal/server/config.go b/internal/server/config.go index cc9a0c4371..486d55f2f0 100644 --- a/internal/server/config.go +++ b/internal/server/config.go @@ -52,6 +52,7 @@ type Config struct { DefaultSuperUser DefaultSuperUser `mapstructure:"default_super_user"` Language string `mapstructure:"language"` TaskExecutor TaskExecutorConfig `mapstructure:"task_executor"` + Ingestor IngestorConfig `mapstructure:"ingestor"` FileSyncer FileSyncerConfig `mapstructure:"file_syncer"` OTel OtelConfig `mapstructure:"otel"` Clickhouse ClickhouseConfig `mapstructure:"clickhouse"` @@ -74,6 +75,10 @@ type DefaultSuperUser struct { Nickname string `mapstructure:"nickname"` } +type IngestorConfig struct { + MQType string `mapstructure:"mq_type"` +} + type TaskExecutorConfig struct { MessageQueueType string `mapstructure:"message_queue_type"` } @@ -89,6 +94,7 @@ type OtelConfig struct { SampleRatio float64 `mapstructure:"sample_ratio"` Secure bool `mapstructure:"secure"` Stdout bool `mapstructure:"stdout"` + Enable bool `mapstructure:"enable"` } type ClickhouseConfig struct { @@ -397,7 +403,7 @@ func Init(configPath string) error { configDict["name"] = "redis" configDict["host"] = host configDict["port"] = port - configDict["service_type"] = "message_queue" + configDict["service_type"] = "cache" configDict["extra"] = map[string]interface{}{ "mq_type": "redis", "database": db, @@ -425,11 +431,11 @@ func Init(configPath string) error { delete(configDict, "max_allowed_packet") delete(configDict, "user") delete(configDict, "password") - case "task_executor": - mqType := getString(configDict, "message_queue_type") + case "ingestor": + mqType := getString(configDict, "mq_type") configDict["id"] = id - configDict["name"] = "task_executor" - configDict["service_type"] = "task_executor" + configDict["name"] = "ingestor" + configDict["service_type"] = "ingestor" configDict["extra"] = map[string]interface{}{ "message_queue_type": mqType, }