From a31647f652f093827b2670cc78b05e70923bf552 Mon Sep 17 00:00:00 2001 From: Lukasz Raczylo Date: Thu, 3 Sep 2026 01:42:49 +0100 Subject: [PATCH] feat(target-health): expose target health on admin API and metrics Add a health object to each row of GET /admin/api/targets (adminTargetHealthView, admin.go): state, an RFC3339 lastCheck, lastError, source, and latencyMs, every optional field omitted while the target has never been observed. Both serveAdminTargets and the MCP/agent listing routes (handleMCPServers, handleAgents) trigger the active probe sweep before responding, so admin polling and client listings double as the heartbeat that drives it. Add llmgateway_target_healthy to /metrics (metrics.go): a 1/0 gauge labeled by kind (mcp or agent) and target name, with no sample at all for a target the tracker has never observed. serveMetrics triggers the sweep too, so a Prometheus scrape becomes a heartbeat as well. Claude-Session: https://claude.ai/code/session_01EebV5jyWvYocByNQ3oss1w --- admin.go | 44 +++++++++++++++++ admin_test.go | 127 ++++++++++++++++++++++++++++++++++++++++++++++++ metrics.go | 49 +++++++++++++++++++ metrics_test.go | 75 ++++++++++++++++++++++++++++ 4 files changed, 295 insertions(+) diff --git a/admin.go b/admin.go index a1c66b7..748b8f2 100644 --- a/admin.go +++ b/admin.go @@ -1235,13 +1235,54 @@ type adminTargetCountersView struct { // convention (adminUsageEntryView) — rather than always listing every // group name, which would grow with the group catalog for no reason once // nothing is actually restricted. +// +// Health is feat/target-health's own readout (targetHealthView, below). +// Field order (Health first, the struct-typed field, then the strings/ +// slice, then Counters last) is fieldalignment-sensitive, derived +// against a scratch copy the same way this package's other structs +// already document (Config's own doc comment, llmgateway.go). type adminTargetView struct { + Health adminTargetHealthView `json:"health"` Name string `json:"name"` URL string `json:"url"` Access []string `json:"access,omitempty"` Counters adminTargetCountersView `json:"counters"` } +// adminTargetHealthView is one target's feat/target-health readout in +// GET /admin/api/targets — the exact JSON contract the webui's MCP & +// Agents panel codes against. State and ConsecutiveFailures are always +// present; every other field is omitted while State is +// targetHealthUnknown (this tracker has never observed the target), +// since none of them carry a meaningful value yet. LatencyMs is a +// pointer so a genuine 0ms observation still serializes as +// "latencyMs":0 rather than being indistinguishable from "omitted". +// Field order is fieldalignment-derived, the same convention +// adminTargetView's own doc comment above explains. +type adminTargetHealthView struct { + LatencyMs *int64 `json:"latencyMs,omitempty"` + State targetHealthState `json:"state"` + LastCheck string `json:"lastCheck,omitempty"` + LastError string `json:"lastError,omitempty"` + Source string `json:"source,omitempty"` + ConsecutiveFailures int `json:"consecutiveFailures"` +} + +// targetHealthView converts one targetHealthTracker.snapshot result +// (target_health.go) into its JSON view. +func targetHealthView(snap targetHealthSnapshot) adminTargetHealthView { + view := adminTargetHealthView{State: snap.state, ConsecutiveFailures: snap.consecutiveFailures} + if snap.state == targetHealthUnknown { + return view + } + view.LastCheck = snap.lastCheck.UTC().Format(time.RFC3339) + view.LastError = snap.lastError + view.Source = snap.source + ms := snap.latency.Milliseconds() + view.LatencyMs = &ms + return view +} + // adminTargetsResponse is the full body of GET /admin/api/targets (Feature // B, v0.21): every configured MCP server and every configured agent, each // with its own access list and current-window request counters. Both @@ -1339,6 +1380,7 @@ func (g *Gateway) buildAdminTargets() adminTargetsResponse { mcpServers := make([]adminTargetView, len(mcpNames)) for i, name := range mcpNames { mcpServers[i] = adminTargetView{ + Health: targetHealthView(g.targetHealth.snapshot(targetKindMCP, name)), Name: name, URL: sanitizeBaseURL(g.cfg.MCPServers[name].URL), Access: adminTargetAccess(groupSummaries, func(gs groupSummary) []string { return gs.mcpServers }, name), @@ -1348,6 +1390,7 @@ func (g *Gateway) buildAdminTargets() adminTargetsResponse { agents := make([]adminTargetView, len(agentNames)) for i, name := range agentNames { agents[i] = adminTargetView{ + Health: targetHealthView(g.targetHealth.snapshot(targetKindAgent, name)), Name: name, URL: sanitizeBaseURL(g.cfg.Agents[name].URL), Access: adminTargetAccess(groupSummaries, func(gs groupSummary) []string { return gs.agents }, name), @@ -1360,6 +1403,7 @@ func (g *Gateway) buildAdminTargets() adminTargetsResponse { // serveAdminTargets writes buildAdminTargets's result as JSON. func (g *Gateway) serveAdminTargets(w http.ResponseWriter) { + g.maybeSweepTargetHealth() setAdminJSONHeaders(w) _ = json.NewEncoder(w).Encode(g.buildAdminTargets()) } diff --git a/admin_test.go b/admin_test.go index 93b26c2..92bb80f 100644 --- a/admin_test.go +++ b/admin_test.go @@ -6,6 +6,7 @@ import ( "crypto/sha256" "encoding/hex" "encoding/json" + "errors" "fmt" "math" "net/http" @@ -2436,3 +2437,129 @@ func TestParseHistoryScope_ModelIDKeepsLaterSeparators(t *testing.T) { t.Errorf("parseHistoryScope = (%q, %q, %v), want the full model id preserved", kind, id, ok) } } + +// --- feat/target-health: GET /admin/api/targets "health" field --- + +// TestAdminTargets_Health_NeverObserved_UnknownWithOmittedFields proves +// the exact JSON contract for a target this tracker has never observed: +// "state":"unknown", consecutiveFailures present as 0, and every other +// health field (lastCheck/lastError/latencyMs/source) omitted entirely. +func TestAdminTargets_Health_NeverObserved_UnknownWithOmittedFields(t *testing.T) { + t.Parallel() + cfg := newAdminTestConfig() + cfg.MCPServers = map[string]*TargetConfig{"alpha": {URL: "http://mcp-alpha.internal"}} + h, _ := newAdminGatewayHandle(t, cfg) + + rec := httptest.NewRecorder() + h.ServeHTTP(rec, adminRequest(http.MethodGet, adminTargetsPath, "sk-admin1")) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200, body=%s", rec.Code, rec.Body.String()) + } + body := rec.Body.String() + if !strings.Contains(body, `"health":{"state":"unknown","consecutiveFailures":0}`) { + t.Errorf("body = %s, want an unknown target's health object to carry only state and consecutiveFailures", body) + } +} + +// TestAdminTargets_Health_Recorded_AllFieldsPresent proves a target with +// at least one recorded observation carries every health field, and that +// state/source/consecutiveFailures reflect the tracker exactly. +func TestAdminTargets_Health_Recorded_AllFieldsPresent(t *testing.T) { + t.Parallel() + cfg := newAdminTestConfig() + cfg.MCPServers = map[string]*TargetConfig{"alpha": {URL: "http://mcp-alpha.internal"}} + h, gw := newAdminGatewayHandle(t, cfg) + + gw.targetHealth.record(targetKindMCP, "alpha", true, nil, 12*time.Millisecond, targetHealthSourceTraffic) + + rec := httptest.NewRecorder() + h.ServeHTTP(rec, adminRequest(http.MethodGet, adminTargetsPath, "sk-admin1")) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200, body=%s", rec.Code, rec.Body.String()) + } + var got adminTargetsResponse + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatalf("decode: %v", err) + } + if len(got.MCPServers) != 1 { + t.Fatalf("mcpServers = %+v, want exactly 1", got.MCPServers) + } + health := got.MCPServers[0].Health + if health.State != targetHealthHealthy { + t.Errorf("state = %q, want healthy", health.State) + } + if health.LastCheck == "" { + t.Error("want a non-empty lastCheck (RFC3339)") + } + if _, err := time.Parse(time.RFC3339, health.LastCheck); err != nil { + t.Errorf("lastCheck = %q, want RFC3339: %v", health.LastCheck, err) + } + if health.Source != targetHealthSourceTraffic { + t.Errorf("source = %q, want %q", health.Source, targetHealthSourceTraffic) + } + if health.LatencyMs == nil || *health.LatencyMs != 12 { + t.Errorf("latencyMs = %v, want *12", health.LatencyMs) + } + if health.ConsecutiveFailures != 0 { + t.Errorf("consecutiveFailures = %d, want 0", health.ConsecutiveFailures) + } + if health.LastError != "" { + t.Errorf("lastError = %q, want empty (omitted) after a success", health.LastError) + } +} + +// TestAdminTargets_Health_Unhealthy_LastErrorPresent proves an unhealthy +// target's lastError is populated and non-empty. +func TestAdminTargets_Health_Unhealthy_LastErrorPresent(t *testing.T) { + t.Parallel() + cfg := newAdminTestConfig() + cfg.MCPServers = map[string]*TargetConfig{"alpha": {URL: "http://mcp-alpha.internal"}} + cfg.TargetHealth = TargetHealthConfig{FailureThreshold: 1} + h, gw := newAdminGatewayHandle(t, cfg) + + gw.targetHealth.record(targetKindMCP, "alpha", false, errors.New("dial tcp: connection refused"), 3*time.Millisecond, targetHealthSourceProbe) + + rec := httptest.NewRecorder() + h.ServeHTTP(rec, adminRequest(http.MethodGet, adminTargetsPath, "sk-admin1")) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200, body=%s", rec.Code, rec.Body.String()) + } + var got adminTargetsResponse + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatalf("decode: %v", err) + } + health := got.MCPServers[0].Health + if health.State != targetHealthUnhealthy { + t.Errorf("state = %q, want unhealthy", health.State) + } + if health.LastError != "dial tcp: connection refused" { + t.Errorf("lastError = %q, want the recorded error text", health.LastError) + } + if health.Source != targetHealthSourceProbe { + t.Errorf("source = %q, want %q", health.Source, targetHealthSourceProbe) + } +} + +// TestAdminTargets_Health_AgentUsesAgentKind proves an agent row's health +// is read under targetKindAgent, not the mcp kind. +func TestAdminTargets_Health_AgentUsesAgentKind(t *testing.T) { + t.Parallel() + cfg := newAdminTestConfig() + cfg.Agents = map[string]*AgentConfig{"bot1": {URL: "http://agent-bot1.internal"}} + h, gw := newAdminGatewayHandle(t, cfg) + + gw.targetHealth.record(targetKindAgent, "bot1", true, nil, time.Millisecond, targetHealthSourceTraffic) + + rec := httptest.NewRecorder() + h.ServeHTTP(rec, adminRequest(http.MethodGet, adminTargetsPath, "sk-admin1")) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200, body=%s", rec.Code, rec.Body.String()) + } + var got adminTargetsResponse + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatalf("decode: %v", err) + } + if len(got.Agents) != 1 || got.Agents[0].Health.State != targetHealthHealthy { + t.Fatalf("agents = %+v, want bot1 healthy", got.Agents) + } +} diff --git a/metrics.go b/metrics.go index f8207ce..b39ad8b 100644 --- a/metrics.go +++ b/metrics.go @@ -170,6 +170,7 @@ func (g *Gateway) metricsSourceAllowed(r *http.Request) bool { // disk cache, and its declared Content-Type must never be second-guessed // by a client's MIME sniffer. func (g *Gateway) serveMetrics(w http.ResponseWriter) { + g.maybeSweepTargetHealth() w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8") w.Header().Set("X-Content-Type-Options", "nosniff") w.Header().Set("Cache-Control", "no-store") @@ -448,6 +449,16 @@ const aggregationNotePerProcess = " AGGREGATION ACROSS REPLICAS: this counter li // instant (e.g. one mid-backoff, one already recovered). const aggregationNoteProviderHealthy = " AGGREGATION ACROSS REPLICAS: each replica runs its own discovery circuit breaker in-process, never shared via Redis, so this can legitimately differ per replica. Read it per-instance where possible; if you must aggregate, use min() to surface \"at least one replica sees this provider as unhealthy\" — sum() is meaningless for a 0/1 gauge." +// aggregationNoteTargetHealthy is appended to llmgateway_target_healthy's +// HELP text (feat/target-health): the tracker behind this gauge +// (target_health.go) is per-process, in-memory state, fed by each +// replica's own passive traffic and, when enabled, its own active probe +// sweep — never shared via Redis. A sibling to +// aggregationNoteProviderHealthy above, not a reuse of it: this +// tracker has no discovery step of its own, so that note's own +// reference to "discovery circuit breaker" would not apply here. +const aggregationNoteTargetHealthy = " AGGREGATION ACROSS REPLICAS: each replica tracks this target's health independently from its own passive traffic and, if enabled, its own active probes — never shared via Redis — so this can legitimately differ per replica. Read it per-instance where possible; if you must aggregate, use min() to surface \"at least one replica sees this target as unhealthy\" — sum() is meaningless for a 0/1 gauge." + // aggregationNoteStoreHealth is appended to llmgateway_limit_store_up's // HELP text: whether THIS replica's own connection to the configured // limit store is currently healthy is, by definition, discovered @@ -470,6 +481,7 @@ func (g *Gateway) renderMetrics() []byte { var m metricWriter g.writeUsageMetrics(&m) g.writeProviderMetrics(&m) + g.writeTargetMetrics(&m) g.writeLatencyMetrics(&m) g.writeProvenanceMetrics(&m) g.writeRejectionMetrics(&m) @@ -673,6 +685,43 @@ func (g *Gateway) writeProviderMetrics(m *metricWriter) { } } +// writeTargetMetrics emits llmgateway_target_healthy (feat/target-health): +// 1 for a healthy MCP server or A2A agent, 0 for an unhealthy one, and NO +// sample at all for a target this tracker has never observed — the +// identical "a gap is honest, a fabricated value is not" convention +// writeUsageMetrics/writeProviderMetrics already apply to a storeDown +// read, applied here to "never yet observed" instead of "currently +// unreadable". Iterates every CONFIGURED MCP server and agent (not just +// ones the tracker happens to have an entry for), so a target removed +// from config since its last observation contributes no stale series. +func (g *Gateway) writeTargetMetrics(m *metricWriter) { + m.family("llmgateway_target_healthy", "gauge", + "1 when an MCP server or A2A agent's health (last passive observation or active probe) is healthy, 0 when its consecutive-failure count has reached targetHealth.failureThreshold. No sample while never yet observed."+aggregationNoteTargetHealthy) + + for _, name := range sortedMCPServerNames(g.cfg.MCPServers) { + writeTargetHealthySample(m, g.targetHealth.snapshot(targetKindMCP, name), targetScopeKind(targetKindMCP), name) + } + for _, name := range sortedAgentNames(g.cfg.Agents) { + writeTargetHealthySample(m, g.targetHealth.snapshot(targetKindAgent, name), targetScopeKind(targetKindAgent), name) + } +} + +// writeTargetHealthySample writes llmgateway_target_healthy's one sample +// for (kindLabel, name), skipped entirely while snap's state is unknown +// — see writeTargetMetrics' own doc comment for why. kindLabel is the +// external "mcp"/"agent" spelling (targetScopeKind), never the internal +// routing kind ("a2a") targetHealthTracker itself keys on. +func writeTargetHealthySample(m *metricWriter, snap targetHealthSnapshot, kindLabel, name string) { + if snap.state == targetHealthUnknown { + return + } + healthy := int64(1) + if snap.state == targetHealthUnhealthy { + healthy = 0 + } + m.sampleInt("llmgateway_target_healthy", []metricLabel{{"kind", kindLabel}, {"target", name}}, healthy) +} + // --- upstream latency (feat: instrument upstream latency) --- // // The task: "priority-based throttling that engages when providers are diff --git a/metrics_test.go b/metrics_test.go index 3a0c4f2..6cc675e 100644 --- a/metrics_test.go +++ b/metrics_test.go @@ -3,6 +3,7 @@ package traefikllmgateway import ( "context" "encoding/json" + "errors" "fmt" "math" "net/http" @@ -1494,3 +1495,77 @@ func TestRecordLatency_MetricsDisabled_NoObservationCollected(t *testing.T) { t.Errorf("g.latency has %d entries after a request with metrics disabled, want 0 — no latency observation should ever be collected when metrics is off", len(snaps)) } } + +// --- feat/target-health: llmgateway_target_healthy --- + +// TestMetrics_TargetHealthy_NoSampleWhileUnknown proves a configured but +// never-observed target contributes NO sample at all — the same +// "a gap is honest, a fabricated value is not" rule the store-backed +// families apply to storeDown, applied here to "never yet observed". +func TestMetrics_TargetHealthy_NoSampleWhileUnknown(t *testing.T) { + t.Parallel() + cfg := newMetricsTestConfig() + cfg.MCPServers = map[string]*TargetConfig{"alpha": {URL: "http://mcp-alpha.internal"}} + h, _ := newMetricsGatewayHandle(t, cfg) + + rec := httptest.NewRecorder() + h.ServeHTTP(rec, metricsRequest(http.MethodGet, metricsPathDefault, "sk-admin1")) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200", rec.Code) + } + _, samples := parsePrometheusText(t, rec.Body.Bytes()) + for _, s := range samples { + if s.name == "llmgateway_target_healthy" { + t.Errorf("unexpected sample %v — a never-observed target must emit nothing", s) + } + } +} + +// TestMetrics_TargetHealthy_ReflectsRecordedState proves the gauge +// renders 1 for a healthy MCP server and 0 for an unhealthy agent, with +// the documented kind label values ("mcp"/"agent" — NOT the internal +// routing kind "a2a"). +func TestMetrics_TargetHealthy_ReflectsRecordedState(t *testing.T) { + t.Parallel() + cfg := newMetricsTestConfig() + cfg.MCPServers = map[string]*TargetConfig{"alpha": {URL: "http://mcp-alpha.internal"}} + cfg.Agents = map[string]*AgentConfig{"bot1": {URL: "http://agent-bot1.internal"}} + cfg.TargetHealth = TargetHealthConfig{FailureThreshold: 1} + h, gw := newMetricsGatewayHandle(t, cfg) + + gw.targetHealth.record(targetKindMCP, "alpha", true, nil, time.Millisecond, targetHealthSourceTraffic) + gw.targetHealth.record(targetKindAgent, "bot1", false, errors.New("boom"), time.Millisecond, targetHealthSourceProbe) + + rec := httptest.NewRecorder() + h.ServeHTTP(rec, metricsRequest(http.MethodGet, metricsPathDefault, "sk-admin1")) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200", rec.Code) + } + _, samples := parsePrometheusText(t, rec.Body.Bytes()) + assertSample(t, samples, "llmgateway_target_healthy", map[string]string{"kind": "mcp", "target": "alpha"}, "1") + assertSample(t, samples, "llmgateway_target_healthy", map[string]string{"kind": "agent", "target": "bot1"}, "0") +} + +// TestMetrics_TargetHealthy_UnaffectedByStoreOutage proves this gauge is +// unaffected by a limit-store outage — it reads the in-process +// target-health tracker only, never the store, mirroring +// llmgateway_provider_healthy's own identical independence +// (TestMetrics_StoreDown_SkipsRatherThanFabricatesZero's own doc +// comment explains why that one is asserted present during an outage; +// this is the same claim for this feature's gauge). +func TestMetrics_TargetHealthy_UnaffectedByStoreOutage(t *testing.T) { + t.Parallel() + cfg := newMetricsTestConfig() + cfg.MCPServers = map[string]*TargetConfig{"alpha": {URL: "http://mcp-alpha.internal"}} + h, gw := newMetricsGatewayHandle(t, cfg) + gw.targetHealth.record(targetKindMCP, "alpha", true, nil, time.Millisecond, targetHealthSourceTraffic) + gw.limiter.store = alwaysErrStore{} + + rec := httptest.NewRecorder() + h.ServeHTTP(rec, metricsRequest(http.MethodGet, metricsPathDefault, "sk-admin1")) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200", rec.Code) + } + _, samples := parsePrometheusText(t, rec.Body.Bytes()) + assertSample(t, samples, "llmgateway_target_healthy", map[string]string{"kind": "mcp", "target": "alpha"}, "1") +}