m/Assistant
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Add dependency-aware runtime readiness

Browse Source
This commit is contained in:
m
2026-08-22 11:20:53 +03:00
parent 470fdf8c03
commit 3f2b008109
6 changed files with 142 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/cognitive-worker/main.go
+9
View File
@@ -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)
cmd/commit-runtime/main.go
+7
View File
@@ -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,
cmd/continuity-runtime/main.go
+7
View File
@@ -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")
cmd/core-runtime/main.go
+6
View File
@@ -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)
runtime/health/server.go
+85
View File
@@ -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()
}}
}
runtime/health/server_test.go
+28
View File
@@ -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())
}
}