Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ae06d33ccc | ||
|
|
c7b57daaf8 | ||
|
|
8aa0123047 | ||
|
|
4a3d348e39 | ||
|
|
4086b75224 | ||
|
|
4f89c46545 | ||
|
|
5d5afc8888 |
@@ -18,6 +18,7 @@ type fleetEndpoint struct {
|
||||
Tier capability.FleetTier
|
||||
URL string
|
||||
Headers map[string]string
|
||||
Client *http.Client
|
||||
}
|
||||
|
||||
type fleetStatusResponse struct {
|
||||
@@ -41,7 +42,11 @@ func fleetReadinessHandler(client *http.Client, preferred capability.FleetTier,
|
||||
for name, value := range endpoint.Headers {
|
||||
request.Header.Set(name, value)
|
||||
}
|
||||
response, requestErr := client.Do(request)
|
||||
probeClient := endpoint.Client
|
||||
if probeClient == nil {
|
||||
probeClient = client
|
||||
}
|
||||
response, requestErr := probeClient.Do(request)
|
||||
if requestErr == nil {
|
||||
response.Body.Close()
|
||||
if response.StatusCode >= 200 && response.StatusCode < 300 {
|
||||
|
||||
@@ -85,21 +85,26 @@ func main() {
|
||||
gateway := cognitiveGateway{resolver: resolver, providers: providers, builtinLocal: builtinLocal, externalRecovery: externalRecovery, now: time.Now, allowCanary: os.Getenv("COGNITIVE_ALLOW_CANARY") == "true", canaryMaxOutputBytes: 4096, admission: newHTTPAdmissionClient(envOr("CORE_RUNTIME_URL", "http://core-runtime:8080"), &http.Client{Timeout: 5 * time.Second})}
|
||||
mux.Handle("/healthz", health.Handler{Runtime: "cognitive-worker", SemanticGeneration: 1, ArchitectureGeneration: 1})
|
||||
readinessClient := &http.Client{Timeout: 2 * time.Second}
|
||||
cpuReadinessClient := &http.Client{Timeout: 10 * 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.OptionalHTTPProbe(readinessClient, strings.TrimRight(envOr("LOCAL_CPU_PROVIDER_URL", "http://docker.cin.su:18082/v1"), "/")+"/models"),
|
||||
health.HTTPProbe(readinessClient, lifeboatURL),
|
||||
health.OptionalHTTPProbe(readinessClient, strings.TrimRight(envOr("LOCAL_GPU_PROVIDER_URL", "http://192.168.220.91:11434"), "/")+"/api/tags"),
|
||||
}})
|
||||
fleetClient := &http.Client{Timeout: 2 * time.Second}
|
||||
preferredFleet := capability.FleetTier(envOr("COGNITIVE_PREFERRED_FLEET_TIER", string(capability.FleetTierHeavy)))
|
||||
cpuFleetHeaders := map[string]string{}
|
||||
if cpuAPIKey != "" {
|
||||
cpuFleetHeaders["Authorization"] = "Bearer " + cpuAPIKey
|
||||
}
|
||||
cpuEndpoint := strings.TrimRight(envOr("LOCAL_CPU_PROVIDER_URL", "http://docker.cin.su:18082/v1"), "/") + "/models"
|
||||
cpuHealthProbe := health.OptionalHTTPProbeWithHeaders(cpuReadinessClient, cpuEndpoint, cpuFleetHeaders)
|
||||
cpuHealthProbe.Timeout = 10 * time.Second
|
||||
readinessProbes := []health.Probe{
|
||||
health.HTTPProbe(readinessClient, strings.TrimRight(envOr("CORE_RUNTIME_URL", "http://core-runtime:8080"), "/")+"/healthz"),
|
||||
cpuHealthProbe,
|
||||
health.HTTPProbe(readinessClient, lifeboatURL),
|
||||
health.OptionalHTTPProbe(readinessClient, strings.TrimRight(envOr("LOCAL_GPU_PROVIDER_URL", "http://192.168.220.91:11434"), "/")+"/api/tags"),
|
||||
}
|
||||
mux.Handle("/readyz", health.ReadinessHandler{Runtime: "cognitive-worker", Probes: readinessProbes})
|
||||
mux.Handle("/v1/fleet/readiness", fleetReadinessHandler(fleetClient, preferredFleet, []fleetEndpoint{
|
||||
{ProviderID: "local-cpu-primary", ModelID: "local-cpu-model", Tier: capability.FleetTierFast, URL: strings.TrimRight(envOr("LOCAL_CPU_PROVIDER_URL", "http://docker.cin.su:18082/v1"), "/") + "/models", Headers: cpuFleetHeaders},
|
||||
{ProviderID: "local-cpu-primary", ModelID: "local-cpu-model", Tier: capability.FleetTierFast, URL: cpuEndpoint, Headers: cpuFleetHeaders, Client: &http.Client{Timeout: 10 * time.Second}},
|
||||
{ProviderID: "local-medium-unconfigured", ModelID: "medium-unconfigured", Tier: capability.FleetTierMedium},
|
||||
{ProviderID: "local-gpu-opportunistic", ModelID: "gpu-opportunistic", Tier: capability.FleetTierHeavy, URL: strings.TrimRight(envOr("LOCAL_GPU_PROVIDER_URL", "http://192.168.220.91:11434"), "/") + "/api/tags"},
|
||||
{ProviderID: "cognitive-lifeboat", ModelID: "qwen3-8b-lifeboat", Tier: capability.FleetTierLifeboat, URL: lifeboatURL},
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
{
|
||||
"task_file": "dev/tasks/DEV-000164-authorized-fleet-readiness.json",
|
||||
"task_id": "DEV-000164",
|
||||
"task_file": "dev/tasks/DEV-000167-cpu-readiness-client-timeout.json",
|
||||
"task_id": "DEV-000167",
|
||||
"task_revision": 1
|
||||
}
|
||||
|
||||
@@ -0,0 +1,23 @@
|
||||
{
|
||||
"task_id": "DEV-000165",
|
||||
"task_revision": 1,
|
||||
"architecture_revision": "ADR-0018",
|
||||
"base_repo_revision": "5d5afc88883c7f79e3b051c790642e41386666ec",
|
||||
"risk_tier": "E",
|
||||
"intent": "Use the existing file-backed CPU provider API key in cognitive-worker readiness probes so /readyz and fleet readiness report consistent provider state.",
|
||||
"requirement_ids": ["WP-601", "Model fleet/failover"],
|
||||
"allowed_change_scope": {"modules": ["cognitive-worker", "platform-health"], "paths": ["runtime/health/server.go", "runtime/health/server_test.go", "cmd/cognitive-worker/main.go", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000165-authorized-cpu-health-probe.json"]},
|
||||
"forbidden_change_scope": {"paths": ["architecture/**", "security/**", "database/**", "deploy/**", "contracts/**", "adapters/**", ".gitea/**"]},
|
||||
"allowed_architecture_delta": {"new_cross_module_dependencies": 0, "new_canonical_owners": 0, "new_privileges": 0, "new_third_party_dependencies": 0, "new_database_owner_roles": 0},
|
||||
"contracts": {"consume": ["existing health Probe contract", "existing provider API key resolver"], "may_change": ["health probe header support"]},
|
||||
"semantic_owner": "cognitive-worker/platform health",
|
||||
"canonical_state": [],
|
||||
"state_transitions": ["authenticated CPU probe -> consistent readiness evidence"],
|
||||
"side_effects_and_privileges": ["resolved key is used only in outbound probe header; never logged or persisted"],
|
||||
"persistence_and_migrations": ["none"],
|
||||
"failure_retry_restart": ["provider timeout or rejected key remains degraded; no fallback behavior changed"],
|
||||
"assumptions": ["the local provider uses Bearer authentication"],
|
||||
"acceptance_scenarios": ["health probe sends configured header", "cognitive /readyz uses authenticated CPU probe", "GPU remains optional", "no key is sent when absent"],
|
||||
"required_tests": ["go test ./...", "go vet ./...", "preflight", "check-diff", "check-architecture", "independent", "git diff --check", "CI", "production /readyz and fleet readiness"],
|
||||
"terminal_gap_states": ["BLOCKED_BY_MISSING_CAPABILITY", "BLOCKED_BY_CONFLICT"]
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
{
|
||||
"task_id": "DEV-000166",
|
||||
"task_revision": 1,
|
||||
"architecture_revision": "ADR-0018",
|
||||
"base_repo_revision": "4086b7522460a78afc341d996ca7874505018871",
|
||||
"risk_tier": "E",
|
||||
"intent": "Give the authenticated CPU readiness probe the provider adapter timeout while retaining short opportunistic GPU probing.",
|
||||
"requirement_ids": ["WP-601", "Model fleet/failover"],
|
||||
"allowed_change_scope": {"modules": ["cognitive-worker", "platform-health"], "paths": ["runtime/health/server.go", "runtime/health/server_test.go", "cmd/cognitive-worker/main.go", "cmd/cognitive-worker/fleet_status.go", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000166-per-provider-readiness-timeouts.json"]},
|
||||
"forbidden_change_scope": {"paths": ["architecture/**", "security/**", "database/**", "deploy/**", "contracts/**", "adapters/**", ".gitea/**"]},
|
||||
"allowed_architecture_delta": {"new_cross_module_dependencies": 0, "new_canonical_owners": 0, "new_privileges": 0, "new_third_party_dependencies": 0, "new_database_owner_roles": 0},
|
||||
"contracts": {"consume": ["existing health Probe contract", "existing fleet readiness probe"], "may_change": ["per-probe timeout metadata"]},
|
||||
"semantic_owner": "cognitive-worker/platform health",
|
||||
"canonical_state": [],
|
||||
"state_transitions": ["provider-specific probe deadline -> truthful readiness result"],
|
||||
"side_effects_and_privileges": ["no new privilege; existing provider auth only"],
|
||||
"persistence_and_migrations": ["none"],
|
||||
"failure_retry_restart": ["CPU gets bounded 10-second probe; GPU retains 2-second opportunistic timeout"],
|
||||
"assumptions": ["CPU provider catalog may take longer than opportunistic probes while remaining within adapter bounds"],
|
||||
"acceptance_scenarios": ["per-probe health timeout is honored", "CPU fleet client uses 10 seconds", "GPU remains bounded", "Lifeboat fallback remains unchanged"],
|
||||
"required_tests": ["go test ./...", "go vet ./...", "preflight", "check-diff", "check-architecture", "independent", "git diff --check", "CI", "production /readyz and fleet readiness"],
|
||||
"terminal_gap_states": ["BLOCKED_BY_MISSING_CAPABILITY", "BLOCKED_BY_CONFLICT"]
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
{
|
||||
"task_id": "DEV-000167",
|
||||
"task_revision": 1,
|
||||
"architecture_revision": "ADR-0018",
|
||||
"base_repo_revision": "c7b57daaf8233b0f0e79c77c02e92040378921d5",
|
||||
"risk_tier": "E",
|
||||
"intent": "Use the bounded CPU provider HTTP client for the authenticated readiness probe so /readyz reports the same truthful result as fleet readiness.",
|
||||
"requirement_ids": ["WP-601", "Model fleet/failover"],
|
||||
"allowed_change_scope": {"modules": ["cognitive-worker"], "paths": ["cmd/cognitive-worker/main.go", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000167-cpu-readiness-client-timeout.json"]},
|
||||
"forbidden_change_scope": {"paths": ["architecture/**", "security/**", "database/**", "deploy/**", "contracts/**", "adapters/**", ".gitea/**"]},
|
||||
"allowed_architecture_delta": {"new_cross_module_dependencies": 0, "new_canonical_owners": 0, "new_privileges": 0, "new_third_party_dependencies": 0, "new_database_owner_roles": 0},
|
||||
"contracts": {"consume": ["existing readiness Probe contract", "existing authenticated CPU fleet endpoint"], "may_change": []},
|
||||
"semantic_owner": "cognitive-worker",
|
||||
"canonical_state": [],
|
||||
"state_transitions": ["CPU provider response -> consistent /readyz and fleet readiness state"],
|
||||
"side_effects_and_privileges": ["no new privilege; existing provider auth only"],
|
||||
"persistence_and_migrations": ["none"],
|
||||
"failure_retry_restart": ["CPU readiness remains bounded at 10 seconds; opportunistic GPU remains 2 seconds"],
|
||||
"assumptions": ["The CPU provider is authenticated with the existing file-backed API key"],
|
||||
"acceptance_scenarios": ["/readyz can report CPU AVAILABLE when fleet readiness reports CPU AVAILABLE", "GPU remains bounded", "Lifeboat fallback remains unchanged"],
|
||||
"required_tests": ["go test ./...", "go vet ./...", "preflight", "check-diff", "check-architecture", "independent", "git diff --check", "CI", "production /readyz and fleet readiness"],
|
||||
"terminal_gap_states": ["BLOCKED_BY_MISSING_CAPABILITY", "BLOCKED_BY_CONFLICT"]
|
||||
}
|
||||
@@ -29,6 +29,7 @@ type Probe struct {
|
||||
Name string
|
||||
Check func(context.Context) error
|
||||
Optional bool
|
||||
Timeout time.Duration
|
||||
}
|
||||
|
||||
type Result struct {
|
||||
@@ -56,7 +57,11 @@ func (h ReadinessHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
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)
|
||||
probeTimeout := timeout
|
||||
if probe.Timeout > 0 {
|
||||
probeTimeout = probe.Timeout
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(r.Context(), probeTimeout)
|
||||
err := probe.Check(ctx)
|
||||
cancel()
|
||||
if err != nil {
|
||||
@@ -82,11 +87,18 @@ func (h ReadinessHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
func HTTPProbe(client *http.Client, endpoint string) Probe {
|
||||
return HTTPProbeWithHeaders(client, endpoint, nil)
|
||||
}
|
||||
|
||||
func HTTPProbeWithHeaders(client *http.Client, endpoint string, headers map[string]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
|
||||
}
|
||||
for name, value := range headers {
|
||||
req.Header.Set(name, value)
|
||||
}
|
||||
response, err := client.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -108,6 +120,12 @@ func OptionalHTTPProbe(client *http.Client, endpoint string) Probe {
|
||||
return probe
|
||||
}
|
||||
|
||||
func OptionalHTTPProbeWithHeaders(client *http.Client, endpoint string, headers map[string]string) Probe {
|
||||
probe := HTTPProbeWithHeaders(client, endpoint, headers)
|
||||
probe.Optional = true
|
||||
return probe
|
||||
}
|
||||
|
||||
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))
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestReadinessReturns503WithTypedDependencyReason(t *testing.T) {
|
||||
@@ -27,6 +28,31 @@ func TestReadinessReturns200WhenAllProbesPass(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestHTTPProbeWithHeadersSendsHeaders(t *testing.T) {
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Header.Get("Authorization") != "Bearer health-key" {
|
||||
t.Fatalf("authorization=%q", r.Header.Get("Authorization"))
|
||||
}
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}))
|
||||
defer server.Close()
|
||||
probe := HTTPProbeWithHeaders(server.Client(), server.URL, map[string]string{"Authorization": "Bearer health-key"})
|
||||
if err := probe.Check(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestReadinessUsesPerProbeTimeout(t *testing.T) {
|
||||
handler := ReadinessHandler{Runtime: "test", Timeout: time.Millisecond, Probes: []Probe{{
|
||||
Name: "slow-but-required", Timeout: time.Second, Check: func(context.Context) error { return nil },
|
||||
}}}
|
||||
recorder := httptest.NewRecorder()
|
||||
handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/readyz", nil))
|
||||
if recorder.Code != http.StatusOK {
|
||||
t.Fatalf("readiness response = %d %s", recorder.Code, recorder.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestReadinessStaysAvailableWithOptionalDependencyDegraded(t *testing.T) {
|
||||
handler := ReadinessHandler{Runtime: "test", Probes: []Probe{
|
||||
{Name: "lifeboat", Check: func(context.Context) error { return nil }},
|
||||
|
||||
Reference in New Issue
Block a user