From 3f2b00810949c67b086e015ad0b3a0f44255f349 Mon Sep 17 00:00:00 2001 From: Mikhail Date: Sat, 22 Aug 2026 11:20:53 +0300 Subject: [PATCH] Add dependency-aware runtime readiness --- cmd/cognitive-worker/main.go | 9 ++++ cmd/commit-runtime/main.go | 7 +++ cmd/continuity-runtime/main.go | 7 +++ cmd/core-runtime/main.go | 6 +++ runtime/health/server.go | 85 ++++++++++++++++++++++++++++++++++ runtime/health/server_test.go | 28 +++++++++++ 6 files changed, 142 insertions(+) create mode 100644 runtime/health/server_test.go diff --git a/cmd/cognitive-worker/main.go b/cmd/cognitive-worker/main.go index 0847e33..3b9cdd9 100644 --- a/cmd/cognitive-worker/main.go +++ b/cmd/cognitive-worker/main.go @@ -15,6 +15,7 @@ import ( "log" "net/http" "os" + "strings" "time" ) @@ -71,6 +72,14 @@ func main() { } gateway := cognitiveGateway{resolver: resolver, providers: providers, builtinLocal: builtinLocal, externalRecovery: externalRecovery, now: time.Now, allowCanary: os.Getenv("COGNITIVE_ALLOW_CANARY") == "true", canaryMaxOutputBytes: 4096} mux.Handle("/healthz", health.Handler{Runtime: "cognitive-worker", SemanticGeneration: 1, ArchitectureGeneration: 1}) + readinessClient := &http.Client{Timeout: 2 * time.Second} + lifeboatURL := strings.TrimSuffix(strings.TrimRight(envOr("LOCAL_LIFEBOAT_PROVIDER_URL", "http://cognitive-lifeboat:8086/v1"), "/"), "/v1") + "/health" + mux.Handle("/readyz", health.ReadinessHandler{Runtime: "cognitive-worker", Probes: []health.Probe{ + health.HTTPProbe(readinessClient, strings.TrimRight(envOr("CORE_RUNTIME_URL", "http://core-runtime:8080"), "/")+"/healthz"), + health.HTTPProbe(readinessClient, strings.TrimRight(envOr("LOCAL_CPU_PROVIDER_URL", "http://docker.cin.su:18082/v1"), "/")+"/models"), + health.HTTPProbe(readinessClient, lifeboatURL), + health.HTTPProbe(readinessClient, strings.TrimRight(envOr("LOCAL_GPU_PROVIDER_URL", "http://192.168.220.91:11434"), "/")+"/api/tags"), + }}) mux.HandleFunc("POST /v1/proposals/deterministic", deterministicProposal) mux.HandleFunc("POST /v1/advisory-policy", gateway.policyHandler) mux.HandleFunc("POST /v1/advisories", gateway.advisoryHandler) diff --git a/cmd/commit-runtime/main.go b/cmd/commit-runtime/main.go index 2b328c9..6813628 100644 --- a/cmd/commit-runtime/main.go +++ b/cmd/commit-runtime/main.go @@ -84,6 +84,13 @@ func main() { dispatcher := operationDispatcher{continuityExecutor: continuityExecutor, continuityReconciler: continuityExecutor, bindings: creatorCore, secretRef: notificationSecretRef, secret: notificationSecret, client: creatorHTTPClient} mux := http.NewServeMux() mux.Handle("/healthz", health.Handler{Runtime: "commit-runtime", SemanticGeneration: 1, ArchitectureGeneration: 1}) + readinessClient := &http.Client{Timeout: 2 * time.Second} + mux.Handle("/readyz", health.ReadinessHandler{Runtime: "commit-runtime", Probes: []health.Probe{ + health.TCPProbe("commit-postgres", envOr("POSTGRES_HOST", "postgres"), envOr("POSTGRES_PORT", "5432")), + health.TCPProbe("authority-postgres", envOr("POSTGRES_HOST", "postgres"), envOr("POSTGRES_PORT", "5432")), + health.HTTPProbe(readinessClient, strings.TrimRight(coreURL, "/")+"/healthz"), + health.HTTPProbe(readinessClient, strings.TrimRight(envOr("CONTINUITY_RUNTIME_URL", "http://continuity-runtime:8081"), "/")+"/healthz"), + }}) registerExecutionRoutes(mux, repository, dispatcher, dispatcher, executorToken) registerCreatorCommandRoutes(mux, creatorCommandHandler{ core: creatorCore, continuity: creatorContinuity, authority: authorityRepository, repository: repository, diff --git a/cmd/continuity-runtime/main.go b/cmd/continuity-runtime/main.go index 03f79b7..b88adf6 100644 --- a/cmd/continuity-runtime/main.go +++ b/cmd/continuity-runtime/main.go @@ -7,6 +7,7 @@ import ( "log" "net/http" "os" + "strings" "time" "git.cin.su/m/Assistant/adapters/postgres" @@ -31,6 +32,12 @@ func main() { } mux := http.NewServeMux() mux.Handle("/healthz", health.Handler{Runtime: "continuity-runtime", SemanticGeneration: 1, ArchitectureGeneration: 1}) + readinessClient := &http.Client{Timeout: 2 * time.Second} + mux.Handle("/readyz", health.ReadinessHandler{Runtime: "continuity-runtime", Probes: []health.Probe{ + health.TCPProbe("postgres", envOr("POSTGRES_HOST", "postgres"), envOr("POSTGRES_PORT", "5432")), + health.HTTPProbe(readinessClient, strings.TrimRight(envOr("CORE_RUNTIME_URL", "http://core-runtime:8080"), "/")+"/healthz"), + health.HTTPProbe(readinessClient, strings.TrimRight(envOr("COMMIT_RUNTIME_URL", "http://commit-runtime:8082"), "/")+"/healthz"), + }}) executorToken := os.Getenv("COMMIT_EXECUTOR_TOKEN") if executorToken == "" { log.Fatal("COMMIT_EXECUTOR_TOKEN is required") diff --git a/cmd/core-runtime/main.go b/cmd/core-runtime/main.go index 53c236a..f7319d4 100644 --- a/cmd/core-runtime/main.go +++ b/cmd/core-runtime/main.go @@ -135,6 +135,12 @@ func main() { return situationdomain.NewResolver(bindings).Resolve(runtime) } mux.Handle("/healthz", health.Handler{Runtime: "core-runtime", SemanticGeneration: 1, ArchitectureGeneration: 1}) + readinessClient := &http.Client{Timeout: 2 * time.Second} + mux.Handle("/readyz", health.ReadinessHandler{Runtime: "core-runtime", Probes: []health.Probe{ + health.TCPProbe("postgres", envOr("POSTGRES_HOST", "postgres"), envOr("POSTGRES_PORT", "5432")), + health.HTTPProbe(readinessClient, strings.TrimRight(envOr("CONTINUITY_RUNTIME_URL", "http://continuity-runtime:8081"), "/")+"/healthz"), + health.HTTPProbe(readinessClient, strings.TrimRight(envOr("COGNITIVE_RUNTIME_URL", "http://cognitive-worker:8083"), "/")+"/healthz"), + }}) mux.HandleFunc("POST /v1/creator/sessions", creatorSessionHandler(bootstrapToken, creatorUsername, creatorPassword, sessionKey, time.Now)) mux.HandleFunc("GET /v1/creator/sessions/current", func(w http.ResponseWriter, r *http.Request) { session, ok := authenticateCreator(w, r) diff --git a/runtime/health/server.go b/runtime/health/server.go index a440385..c55cee6 100644 --- a/runtime/health/server.go +++ b/runtime/health/server.go @@ -1,8 +1,12 @@ package health import ( + "context" "encoding/json" + "fmt" + "net" "net/http" + "time" ) type Handler struct { @@ -20,3 +24,84 @@ func (h Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { "architecture_generation": h.ArchitectureGeneration, }) } + +type Probe struct { + Name string + Check func(context.Context) error +} + +type Result struct { + Name string `json:"name"` + State string `json:"state"` + Reason string `json:"reason"` + Detail string `json:"detail,omitempty"` +} + +type ReadinessHandler struct { + Runtime string + Probes []Probe + Timeout time.Duration +} + +func (h ReadinessHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + timeout := h.Timeout + if timeout <= 0 { + timeout = 2 * time.Second + } + results := make([]Result, 0, len(h.Probes)) + ready := true + for _, probe := range h.Probes { + result := Result{Name: probe.Name, State: "READY", Reason: "AVAILABLE"} + if probe.Check == nil { + result.State, result.Reason, result.Detail = "DEGRADED", "PROBE_NOT_CONFIGURED", "probe is not configured" + } else { + ctx, cancel := context.WithTimeout(r.Context(), timeout) + err := probe.Check(ctx) + cancel() + if err != nil { + ready = false + result.State, result.Reason, result.Detail = "DEGRADED", "DEPENDENCY_UNAVAILABLE", err.Error() + } + } + if result.State != "READY" { + ready = false + } + results = append(results, result) + } + status := "ready" + code := http.StatusOK + if !ready { + status, code = "degraded", http.StatusServiceUnavailable + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(code) + _ = json.NewEncoder(w).Encode(map[string]any{"status": status, "runtime": h.Runtime, "checks": results}) +} + +func HTTPProbe(client *http.Client, endpoint string) Probe { + return Probe{Name: endpoint, Check: func(ctx context.Context) error { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) + if err != nil { + return err + } + response, err := client.Do(req) + if err != nil { + return err + } + defer response.Body.Close() + if response.StatusCode < 200 || response.StatusCode >= 300 { + return fmt.Errorf("status=%s", response.Status) + } + return nil + }} +} + +func TCPProbe(name, host, port string) Probe { + return Probe{Name: name, Check: func(ctx context.Context) error { + connection, err := (&net.Dialer{}).DialContext(ctx, "tcp", net.JoinHostPort(host, port)) + if err != nil { + return err + } + return connection.Close() + }} +} diff --git a/runtime/health/server_test.go b/runtime/health/server_test.go new file mode 100644 index 0000000..04eb5c1 --- /dev/null +++ b/runtime/health/server_test.go @@ -0,0 +1,28 @@ +package health + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestReadinessReturns503WithTypedDependencyReason(t *testing.T) { + handler := ReadinessHandler{Runtime: "test", Probes: []Probe{{Name: "postgres", Check: func(context.Context) error { return errors.New("connection refused") }}}} + recorder := httptest.NewRecorder() + handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/readyz", nil)) + if recorder.Code != http.StatusServiceUnavailable || !strings.Contains(recorder.Body.String(), `"reason":"DEPENDENCY_UNAVAILABLE"`) { + t.Fatalf("readiness response = %d %s", recorder.Code, recorder.Body.String()) + } +} + +func TestReadinessReturns200WhenAllProbesPass(t *testing.T) { + handler := ReadinessHandler{Runtime: "test", Probes: []Probe{{Name: "postgres", Check: func(context.Context) error { return nil }}}} + recorder := httptest.NewRecorder() + handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/readyz", nil)) + if recorder.Code != http.StatusOK || !strings.Contains(recorder.Body.String(), `"status":"ready"`) { + t.Fatalf("readiness response = %d %s", recorder.Code, recorder.Body.String()) + } +}