diff --git a/docker/docker-compose-base.yml b/docker/docker-compose-base.yml index 36cca3db77..a2654cf733 100644 --- a/docker/docker-compose-base.yml +++ b/docker/docker-compose-base.yml @@ -255,7 +255,7 @@ services: - ragflow-go image: nats:2.14.2 ports: - - ${NATS_PORT}:4222 + - ${NATS_PORT:-4222}:4222 - "8222:8222" volumes: - nats_data:/data @@ -265,10 +265,10 @@ services: - ragflow restart: unless-stopped healthcheck: - test: ["CMD", "nc", "-z", "localhost", "${NATS_PORT}"] + test: ["CMD", "nats-server", "--health-check"] interval: 10s timeout: 10s - retries: 120 + retries: 30 tei-cpu: profiles: diff --git a/internal/admin/handler.go b/internal/admin/handler.go index 3977834b5c..6ca1402c39 100644 --- a/internal/admin/handler.go +++ b/internal/admin/handler.go @@ -56,9 +56,25 @@ type ErrorResponse struct { Message string `json:"message"` } -// Health check -func (h *Handler) Health(c *gin.Context) { - c.JSON(200, gin.H{"status": "ok"}) +// Healthz to get system health +func (h *Handler) Healthz(c *gin.Context) { + result, allOK := service.GetComponentsHealthz(c.Request.Context()) + if allOK { + c.JSON(http.StatusOK, gin.H{ + "code": 0, + "data": result, + }) + } else { + c.JSON(http.StatusInternalServerError, gin.H{ + "code": common.CodeServerError, + "data": result, + }) + } +} + +// Live endpoint +func (h *Handler) Live(c *gin.Context) { + common.SuccessNoData(c, "") } // Ping ping endpoint diff --git a/internal/admin/router.go b/internal/admin/router.go index 89aaa6d059..1576a4ee38 100644 --- a/internal/admin/router.go +++ b/internal/admin/router.go @@ -34,8 +34,10 @@ func NewRouter(handler *Handler) *Router { // Setup setup routes func (r *Router) Setup(engine *gin.Engine) { - // Health check - engine.GET("/health", r.handler.Health) + // Healthz to get system health + engine.GET("/healthz", r.handler.Healthz) + engine.GET("/", r.handler.Live) + engine.GET("/live", r.handler.Live) // Admin API routes with prefix /api/v1/admin admin := engine.Group("/api/v1/admin") diff --git a/internal/cli/admin_command.go b/internal/cli/admin_command.go index af27ea1ccb..c31a8de8cd 100644 --- a/internal/cli/admin_command.go +++ b/internal/cli/admin_command.go @@ -87,6 +87,24 @@ func (c *CLI) AdminPingCacheCommand(cmd *Command) (ResponseIf, error) { return HandleSimpleResponse(resp, fmt.Sprintf("ping cache")) } +// AdminLiveServerCommand GET /live +func (c *CLI) AdminLiveServerCommand(cmd *Command) (ResponseIf, error) { + resp, err := c.AdminServerClient.Request("GET", "/live", "none", nil, nil) + if err != nil { + return nil, fmt.Errorf("failed to show live server: %w", err) + } + return HandleSimpleResponse(resp, fmt.Sprintf("show live server")) +} + +// AdminHealthServerCommand GET /healthz +func (c *CLI) AdminHealthServerCommand(cmd *Command) (ResponseIf, error) { + resp, err := c.AdminServerClient.Request("GET", "/healthz", "none", nil, nil) + if err != nil { + return nil, fmt.Errorf("failed to show health server: %w", err) + } + return HandleCommonDataResponse(resp, fmt.Sprintf("show health server")) +} + // AdminShowVersionCommand show RAGFlow admin version func (c *CLI) AdminShowVersionCommand(cmd *Command) (ResponseIf, error) { resp, err := c.AdminServerClient.Request("GET", "/admin/version", "web", nil, nil) diff --git a/internal/cli/admin_parser.go b/internal/cli/admin_parser.go index 6995fc891f..984ae40ee5 100644 --- a/internal/cli/admin_parser.go +++ b/internal/cli/admin_parser.go @@ -97,6 +97,18 @@ func (p *Parser) parseAdminPingServer() (*Command, error) { return cmd, nil } +func (p *Parser) parseAdminLiveServer() (*Command, error) { + p.nextToken() // consume LIVE + cmd := NewCommand("admin_live_server") + return cmd, nil +} + +func (p *Parser) parseAdminHealthServer() (*Command, error) { + p.nextToken() // consume HEALTH + cmd := NewCommand("admin_health_server") + return cmd, nil +} + // endregion // region LIST commands diff --git a/internal/cli/cli_http.go b/internal/cli/cli_http.go index 836f4927b6..97ef1b3bb7 100644 --- a/internal/cli/cli_http.go +++ b/internal/cli/cli_http.go @@ -51,6 +51,10 @@ func (c *CLI) ExecuteAdminCommand(cmd *Command) (ResponseIf, error) { return c.AdminPingCacheCommand(cmd) case "admin_ping_server": return c.PingServerByCommand(cmd) + case "admin_live_server": + return c.AdminLiveServerCommand(cmd) + case "admin_health_server": + return c.AdminHealthServerCommand(cmd) case "benchmark": return c.RunBenchmark(cmd) case "admin_list_services": diff --git a/internal/cli/http_client.go b/internal/cli/http_client.go index bf62d450b4..1c2aba974c 100644 --- a/internal/cli/http_client.go +++ b/internal/cli/http_client.go @@ -67,12 +67,19 @@ func (c *HTTPClient) APIBase() string { // NonAPIBase returns the non-API base URL func (c *HTTPClient) NonAPIBase() string { - return fmt.Sprintf("%s:%d/%s", c.Host, c.Port, c.APIVersion) + return fmt.Sprintf("%s:%d", c.Host, c.Port) } // BuildURL builds the full URL for a given path -func (c *HTTPClient) BuildURL(path string) string { - base := c.APIBase() +func (c *HTTPClient) BuildURL(path string, authKind string) string { + var base string + + if authKind == "none" { + base = c.NonAPIBase() + } else { + base = c.APIBase() + } + if c.VerifySSL { return fmt.Sprintf("https://%s%s", base, path) } @@ -126,7 +133,7 @@ func (c *HTTPClient) Request(method, path string, authKind string, headers map[s return nil, fmt.Errorf("HTTP Client is nil") } - url := c.BuildURL(path) + url := c.BuildURL(path, authKind) mergedHeaders := c.Headers(authKind, headers) var body io.Reader @@ -196,7 +203,7 @@ func (c *HTTPClient) RequestWithIterations(method, path string, authKind string, return response, nil } - url := c.BuildURL(path) + url := c.BuildURL(path, authKind) mergedHeaders := c.Headers(authKind, headers) var body io.Reader @@ -278,7 +285,7 @@ func (c *HTTPClient) RequestJSON(method, path string, authKind string, headers m // UploadMultipart uploads data using multipart/form-data func (c *HTTPClient) UploadMultipart(path string, contentType string, body io.Reader) error { - url := c.BuildURL(path) + url := c.BuildURL(path, "api") req, err := http.NewRequest("POST", url, body) if err != nil { @@ -322,7 +329,7 @@ func (c *HTTPClient) UploadMultipart(path string, contentType string, body io.Re // RequestStream makes an HTTP request for SSE streaming and returns the response body reader func (c *HTTPClient) RequestStream(method, path string, authKind string, headers map[string]string, jsonBody map[string]interface{}) (io.ReadCloser, error) { - url := c.BuildURL(path) + url := c.BuildURL(path, authKind) mergedHeaders := c.Headers(authKind, headers) var body io.Reader diff --git a/internal/cli/lexer.go b/internal/cli/lexer.go index 7d9174f5ac..8952c98cbe 100644 --- a/internal/cli/lexer.go +++ b/internal/cli/lexer.go @@ -227,6 +227,10 @@ func (l *Lexer) lookupIdent(ident string) Token { return Token{Type: TokenAPI, Value: ident} case "ADD": return Token{Type: TokenAdd, Value: ident} + case "LIVE": + return Token{Type: TokenLive, Value: ident} + case "HEALTH": + return Token{Type: TokenHealth, Value: ident} case "HOST": return Token{Type: TokenHost, Value: ident} case "DELETE": diff --git a/internal/cli/parser.go b/internal/cli/parser.go index aff4ff100c..7b83183d92 100644 --- a/internal/cli/parser.go +++ b/internal/cli/parser.go @@ -86,6 +86,10 @@ func (p *Parser) parseAdminCommand() (*Command, error) { return p.parseAdminLogout() case TokenPing: return p.parseAdminPingServer() + case TokenLive: + return p.parseAdminLiveServer() + case TokenHealth: + return p.parseAdminHealthServer() case TokenList: return p.parseAdminListCommands() case TokenShow: diff --git a/internal/cli/types.go b/internal/cli/types.go index bba51e27fc..661d71c8c4 100644 --- a/internal/cli/types.go +++ b/internal/cli/types.go @@ -45,6 +45,8 @@ const ( TokenServer TokenAPI TokenAdd + TokenLive + TokenHealth TokenHost TokenDelete TokenPassword diff --git a/internal/common/system.go b/internal/common/system_settings.go similarity index 100% rename from internal/common/system.go rename to internal/common/system_settings.go diff --git a/internal/service/system.go b/internal/service/system.go index 25ef64a092..729971d73b 100644 --- a/internal/service/system.go +++ b/internal/service/system.go @@ -75,12 +75,13 @@ type HealthzMeta struct { } type HealthzResponse struct { - DB string `json:"db"` - Redis string `json:"redis"` - DocEngine string `json:"doc_engine"` - Storage string `json:"storage"` - Status string `json:"status"` - Meta map[string]HealthzMeta `json:"_meta,omitempty"` + DB string `json:"db"` + Redis string `json:"redis"` + DocEngine string `json:"doc_engine"` + Storage string `json:"storage"` + MessageQueue string `json:"message_queue"` + Status string `json:"status"` + Meta map[string]HealthzMeta `json:"_meta,omitempty"` } // GetVersion get RAGFlow version @@ -312,8 +313,7 @@ func timedHealthCheck(check func() error) (bool, HealthzMeta) { return true, meta } -// Healthz runs lightweight dependency checks for /api/v1/system/healthz. -func (s *SystemService) Healthz(ctx context.Context) (*HealthzResponse, bool) { +func GetComponentsHealthz(ctx context.Context) (*HealthzResponse, bool) { meta := map[string]HealthzMeta{} dbOK, dbMeta := timedHealthCheck(func() error { @@ -363,13 +363,32 @@ func (s *SystemService) Healthz(ctx context.Context) (*HealthzResponse, bool) { meta["storage"] = storageMeta } - allOK := dbOK && redisOK && docOK && storageOK + messageQueueOK, messageQueueMeta := timedHealthCheck(func() error { + + msgQueueEngine := engine.GetMessageQueueEngine() + if msgQueueEngine == nil { + return fmt.Errorf("message queue is not initialized") + } + + status := msgQueueEngine.CheckStatus() + + if msgQueueEngine == nil || status != "CONNECTED" { + return fmt.Errorf("message queue is not healthy") + } + return nil + }) + if !messageQueueOK { + meta["message_queue"] = messageQueueMeta + } + + allOK := dbOK && redisOK && docOK && storageOK && messageQueueOK result := &HealthzResponse{ - DB: okNok(dbOK), - Redis: okNok(redisOK), - DocEngine: okNok(docOK), - Storage: okNok(storageOK), - Status: okNok(allOK), + DB: okNok(dbOK), + Redis: okNok(redisOK), + DocEngine: okNok(docOK), + Storage: okNok(storageOK), + MessageQueue: okNok(messageQueueOK), + Status: okNok(allOK), } if len(meta) > 0 { result.Meta = meta @@ -377,6 +396,11 @@ func (s *SystemService) Healthz(ctx context.Context) (*HealthzResponse, bool) { return result, allOK } +// Healthz runs lightweight dependency checks for /api/v1/system/healthz. +func (s *SystemService) Healthz(ctx context.Context) (*HealthzResponse, bool) { + return GetComponentsHealthz(ctx) +} + // ListAllVariables list all variables // Returns all system settings from database func (s *SystemService) ListAllVariables() ([]map[string]interface{}, error) { diff --git a/internal/service/system_test.go b/internal/service/system_test.go index 564b293f62..c06dccc92a 100644 --- a/internal/service/system_test.go +++ b/internal/service/system_test.go @@ -10,7 +10,7 @@ func TestSystemServiceHealthzReportsUnhealthyDependencies(t *testing.T) { if result.Status != "nok" { t.Fatalf("status=%q, want nok", result.Status) } - if result.DB != "nok" || result.Redis != "nok" || result.DocEngine != "nok" || result.Storage != "nok" { + if result.DB != "nok" || result.Redis != "nok" || result.DocEngine != "nok" || result.Storage != "nok" || result.MessageQueue != "nok" { t.Fatalf("unexpected health result: %+v", result) } if len(result.Meta) == 0 {