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 <haijin.chn@gmail.com>

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)
 	}
 

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:

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
 	}
 

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,
 			}