From d2152b169cbb78a8607999cb951e5ad68a8380a8 Mon Sep 17 00:00:00 2001 From: Lukasz Raczylo Date: Thu, 3 Sep 2026 02:42:01 +0100 Subject: [PATCH] fix(target-health): scrub lastError URLs and fix unknown-state gaps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Combines three closely-coupled target_health.go fixes that all touch targetHealthTracker.record/stateForLocked in the same commit, since splitting them would leave an intermediate commit that fails to compile or a test asserting behaviour its own commit does not yet implement. F1: record gains a rawURL parameter and scrubs err's text through sanitizeProviderErr before storing lastError, so a target URL configured with an embedded credential can never leak through GET /admin/api/targets. The scrub runs before truncateRunes so a long embedded URL cannot survive truncation unscrubbed. Every probe, the per-server proxy hook (mcp_a2a.go), and both federation record sites (mcp_federation.go) now pass their target's configured URL through. F3: maybeSweepTargetHealth re-checks the elapsed interval after winning the single-flight CAS, closing a window where a caller's own stale interval read could still win the CAS once a separate, faster sweep had already started and finished within the same interval. F4: targetHealthEntry gains everOK, set on record's success branch. stateForLocked now reads a target that has never once succeeded as unknown rather than healthy, even below failureThreshold — the threshold still protects against a single transient first failure, it just does not promote a target this tracker has never seen answer to healthy. targetHealthSnapshot gains observed so the admin view can still tell apart never-observed (omit every other field) from observed-but-never-succeeded (keep them) despite both reading state unknown. The now-redundant lastOK field is removed. Also included: target_health_test.go's own test-hygiene fixes (F6) — an idempotent, t.Cleanup-registered unblock ahead of srv.Close so a failing assertion cannot leave a released server handler blocked until the test binary's own timeout, and an atomic hit counter in the disabled-sweep test — since they land in the same file as the tests above. Claude-Session: https://claude.ai/code/session_01EebV5jyWvYocByNQ3oss1w --- admin.go | 26 ++++-- admin_test.go | 85 +++++++++++++++++- mcp_a2a.go | 2 +- mcp_federation.go | 4 +- metrics_test.go | 6 +- target_health.go | 125 ++++++++++++++++++++------ target_health_test.go | 204 +++++++++++++++++++++++++++++++++++++----- 7 files changed, 387 insertions(+), 65 deletions(-) diff --git a/admin.go b/admin.go index 748b8f2..bda3379 100644 --- a/admin.go +++ b/admin.go @@ -1252,11 +1252,20 @@ type adminTargetView struct { // 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". +// present. State "unknown" (targetHealthUnknown) covers two different +// situations (F4, feat/target-health review) — see that constant's own +// doc comment — and this view tells them apart by whether the tracker +// has ever observed the target at all, NOT by State alone: +// - never observed: every other field is omitted, since none of them +// carry a meaningful value yet. +// - observed, but never once succeeded (still below +// targetHealth.failureThreshold): every field below is still +// present, exactly as for "healthy"/"unhealthy" — there IS a real +// lastCheck/lastError/source/latencyMs to show, the tracker simply +// has not seen this target succeed 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 { @@ -1269,10 +1278,13 @@ type adminTargetHealthView struct { } // targetHealthView converts one targetHealthTracker.snapshot result -// (target_health.go) into its JSON view. +// (target_health.go) into its JSON view. Whether to omit the non-always- +// present fields is decided by snap.observed, NOT snap.state == +// targetHealthUnknown (adminTargetHealthView's own doc comment explains +// why those are different questions since F4). func targetHealthView(snap targetHealthSnapshot) adminTargetHealthView { view := adminTargetHealthView{State: snap.state, ConsecutiveFailures: snap.consecutiveFailures} - if snap.state == targetHealthUnknown { + if !snap.observed { return view } view.LastCheck = snap.lastCheck.UTC().Format(time.RFC3339) diff --git a/admin_test.go b/admin_test.go index 92bb80f..cc49aa1 100644 --- a/admin_test.go +++ b/admin_test.go @@ -2461,6 +2461,51 @@ func TestAdminTargets_Health_NeverObserved_UnknownWithOmittedFields(t *testing.T } } +// TestAdminTargets_Health_NeverSucceeded_UnknownWithFieldsPresent proves +// F4's second "unknown" case: a target that HAS been observed, but never +// once succeeded, and is still below failureThreshold, reads "unknown" — +// yet carries every other health field, unlike a target this tracker has +// never observed at all (the test above). The omit-when-unknown rule +// applies only to a NEVER-OBSERVED entry, not to "unknown because never +// succeeded yet". +func TestAdminTargets_Health_NeverSucceeded_UnknownWithFieldsPresent(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", "", false, errors.New("dial tcp: connection refused"), 7*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 != targetHealthUnknown { + t.Errorf("state = %q, want unknown (never succeeded, still below threshold)", health.State) + } + if health.LastCheck == "" { + t.Error("want a non-empty lastCheck even though state is unknown — this target HAS been observed") + } + 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) + } + if health.LatencyMs == nil || *health.LatencyMs != 7 { + t.Errorf("latencyMs = %v, want *7", health.LatencyMs) + } + if health.ConsecutiveFailures != 1 { + t.Errorf("consecutiveFailures = %d, want 1", health.ConsecutiveFailures) + } +} + // 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. @@ -2470,7 +2515,7 @@ func TestAdminTargets_Health_Recorded_AllFieldsPresent(t *testing.T) { 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) + gw.targetHealth.record(targetKindMCP, "alpha", "", true, nil, 12*time.Millisecond, targetHealthSourceTraffic) rec := httptest.NewRecorder() h.ServeHTTP(rec, adminRequest(http.MethodGet, adminTargetsPath, "sk-admin1")) @@ -2517,7 +2562,7 @@ func TestAdminTargets_Health_Unhealthy_LastErrorPresent(t *testing.T) { 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) + 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")) @@ -2548,7 +2593,7 @@ func TestAdminTargets_Health_AgentUsesAgentKind(t *testing.T) { 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) + gw.targetHealth.record(targetKindAgent, "bot1", "", true, nil, time.Millisecond, targetHealthSourceTraffic) rec := httptest.NewRecorder() h.ServeHTTP(rec, adminRequest(http.MethodGet, adminTargetsPath, "sk-admin1")) @@ -2563,3 +2608,37 @@ func TestAdminTargets_Health_AgentUsesAgentKind(t *testing.T) { t.Fatalf("agents = %+v, want bot1 healthy", got.Agents) } } + +// TestAdminTargets_Health_LastError_ScrubsTargetCredentials proves F1 end +// to end: a target URL configured with an embedded credential never +// reaches GET /admin/api/targets through lastError, even though the +// recorded probe error text embeds the exact configured URL — mirroring +// the credential scrub sanitizeBaseURL already applies to the sibling +// "url" field. +func TestAdminTargets_Health_LastError_ScrubsTargetCredentials(t *testing.T) { + t.Parallel() + rawURL := "https://svc.internal/mcp?api-key=SECRETVALUE" + cfg := newAdminTestConfig() + cfg.MCPServers = map[string]*TargetConfig{"alpha": {URL: rawURL}} + cfg.TargetHealth = TargetHealthConfig{FailureThreshold: 1} + h, gw := newAdminGatewayHandle(t, cfg) + + gw.targetHealth.record(targetKindMCP, "alpha", rawURL, false, errors.New(`Get "`+rawURL+`": 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()) + } + body := rec.Body.String() + if strings.Contains(body, "SECRETVALUE") { + t.Errorf("body = %s, want lastError scrubbed of the target's api-key", body) + } + var got adminTargetsResponse + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatalf("decode: %v", err) + } + if strings.Contains(got.MCPServers[0].Health.LastError, "SECRETVALUE") { + t.Errorf("lastError = %q, want scrubbed", got.MCPServers[0].Health.LastError) + } +} diff --git a/mcp_a2a.go b/mcp_a2a.go index 1fc6c64..32df7e9 100644 --- a/mcp_a2a.go +++ b/mcp_a2a.go @@ -226,7 +226,7 @@ func (g *Gateway) handleTargetProxy(w http.ResponseWriter, r *http.Request, u *u // success/failure rule. start := time.Now() result, ok := g.proxyUpstream(w, r, upstreamURL, g.targetClient, nil, nil, false, kind+" target (name "+name+")", g.targetTimeout) - g.recordTargetProxyHealth(kind, name, result, ok, time.Since(start)) + g.recordTargetProxyHealth(kind, name, targetURL, result, ok, time.Since(start)) } // validateTargetURLs checks every configured MCP-server and agent entry at diff --git a/mcp_federation.go b/mcp_federation.go index e2dd4d3..223bbd2 100644 --- a/mcp_federation.go +++ b/mcp_federation.go @@ -980,7 +980,7 @@ func (g *Gateway) mcpFederatedToolsList(w http.ResponseWriter, format mcpRespons // it is recorded separately, below, once the shape is known. probeStart := time.Now() resp, err := g.mcpBackendCall(fanoutCtx, targetURL, "tools/list", struct{}{}, mcpBackendResponseMaxBytes) - g.targetHealth.record(targetKindMCP, name, err == nil, err, time.Since(probeStart), targetHealthSourceTraffic) + g.targetHealth.record(targetKindMCP, name, targetURL, err == nil, err, time.Since(probeStart), targetHealthSourceTraffic) mu.Lock() defer mu.Unlock() @@ -1103,7 +1103,7 @@ func (g *Gateway) mcpFederatedToolsCall(w http.ResponseWriter, format mcpRespons // not. probeStart := time.Now() resp, err := g.mcpBackendCall(ctx, targetURL, "tools/call", mcpToolCallParams{Name: toolName, Arguments: params.Arguments}, mcpBackendCallResponseMaxBytes) - g.targetHealth.record(targetKindMCP, serverName, err == nil, err, time.Since(probeStart), targetHealthSourceTraffic) + g.targetHealth.record(targetKindMCP, serverName, targetURL, err == nil, err, time.Since(probeStart), targetHealthSourceTraffic) g.limiter.countTargetRequest(targetKindMCP, serverName) if err != nil { g.logf("federated tools/call: server %q: %v", serverName, err) diff --git a/metrics_test.go b/metrics_test.go index 6cc675e..0bf64a3 100644 --- a/metrics_test.go +++ b/metrics_test.go @@ -1533,8 +1533,8 @@ func TestMetrics_TargetHealthy_ReflectsRecordedState(t *testing.T) { 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) + 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")) @@ -1558,7 +1558,7 @@ func TestMetrics_TargetHealthy_UnaffectedByStoreOutage(t *testing.T) { 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.targetHealth.record(targetKindMCP, "alpha", "", true, nil, time.Millisecond, targetHealthSourceTraffic) gw.limiter.store = alwaysErrStore{} rec := httptest.NewRecorder() diff --git a/target_health.go b/target_health.go index 33cdeb7..a16b485 100644 --- a/target_health.go +++ b/target_health.go @@ -32,11 +32,22 @@ import ( type targetHealthState string const ( - // targetHealthUnknown: the tracker has never recorded an observation - // (probe or traffic) for this (kind, name) — a target configured but - // never yet reached. + // targetHealthUnknown covers two distinct situations (F4, + // feat/target-health review), both read the SAME state on purpose — + // neither has ever demonstrated the target actually works: + // 1. never observed at all — the tracker has no entry for this + // (kind, name) yet, a target configured but never yet reached. + // 2. observed, but never once succeeded — every observation so far + // is a failure, and consecutiveFailures has not yet reached + // failureThreshold. Reading "healthy" here (the pre-F4 behavior) + // would be a false claim from a target this tracker has never + // actually seen answer; reading "unhealthy" before the threshold + // is reached would discard the threshold's own protection + // against a single transient first failure. See + // targetHealthSnapshot.observed for how the admin view (admin.go) + // tells these two apart despite sharing this one state. targetHealthUnknown targetHealthState = "unknown" - // targetHealthHealthy: recorded at least once, and consecutiveFailures + // targetHealthHealthy: succeeded at least once, and consecutiveFailures // has not yet reached failureThreshold. This DELIBERATELY includes a // target whose most recent single observation was itself a failure: // 1 or 2 recent failures after a success still read "healthy" until @@ -47,7 +58,10 @@ const ( // state machine of its own (see this file's own package doc comment // for why). targetHealthHealthy targetHealthState = "healthy" - // targetHealthUnhealthy: consecutiveFailures >= failureThreshold. + // targetHealthUnhealthy: consecutiveFailures >= failureThreshold, + // regardless of whether the target has ever succeeded — the + // threshold gate always takes priority over the never-succeeded + // check above. targetHealthUnhealthy targetHealthState = "unhealthy" ) @@ -85,7 +99,11 @@ type targetHealthEntry struct { source string latency time.Duration consecutiveFailures int - lastOK bool + // everOK is set true the first time record observes a success for + // this entry, and never cleared afterward — stateForLocked's F4 + // "never succeeded yet reads unknown, not healthy" rule reads this, + // not consecutiveFailures alone. + everOK bool } // targetHealthTracker is feat/target-health's own per-pod, in-memory @@ -128,9 +146,20 @@ func newTargetHealthTracker(threshold int) *targetHealthTracker { // even on failure — a caller with no Go error to attach, e.g. a bad // HTTP status alone, still leaves lastError at whatever it previously // held rather than clearing it, since there IS a new failure to report). -// source is targetHealthSourceProbe or targetHealthSourceTraffic — -// whichever mechanism produced this observation. -func (t *targetHealthTracker) record(kind, name string, ok bool, err error, latency time.Duration, source string) { +// rawURL is (kind, name)'s configured target URL — the exact string a +// caller dialed, used ONLY to scrub err's text through sanitizeProviderErr +// (admin.go) before it is stored: a Go *url.Error embeds the dialed URL +// verbatim, so an operator-configured target URL carrying credentials +// (an "api-key" query parameter, userinfo) would otherwise leak through +// GET /admin/api/targets' lastError field the same way an unsanitized +// provider error once could (F1, feat/target-health review). The scrub +// runs BEFORE truncateRunes, not after: truncating first could cut a long +// embedded URL in half, leaving sanitizeProviderErr's exact-substring +// match unable to find it and a fragment of the credential in the +// truncated result. source is targetHealthSourceProbe or +// targetHealthSourceTraffic — whichever mechanism produced this +// observation. +func (t *targetHealthTracker) record(kind, name, rawURL string, ok bool, err error, latency time.Duration, source string) { if t == nil || name == "" { return } @@ -143,17 +172,17 @@ func (t *targetHealthTracker) record(kind, name string, ok bool, err error, late t.entries[key] = e } e.lastCheck = t.nowFn() - e.lastOK = ok e.latency = latency e.source = source if ok { + e.everOK = true e.consecutiveFailures = 0 e.lastError = "" return } e.consecutiveFailures++ if err != nil { - e.lastError = truncateRunes(err.Error(), targetHealthMaxErrorRunes) + e.lastError = truncateRunes(sanitizeProviderErr(err.Error(), rawURL), targetHealthMaxErrorRunes) } } @@ -173,11 +202,19 @@ func (t *targetHealthTracker) stateOf(kind, name string) targetHealthState { } // stateForLocked derives e's targetHealthState. Caller must hold t.mu. -// Split out of stateOf/snapshot so both apply the identical rule. +// Split out of stateOf/snapshot so both apply the identical rule. The +// threshold check runs first and always wins (F4): a target that has +// failed failureThreshold times straight reads unhealthy whether or not +// it has ever succeeded. Below threshold, a target that has never once +// succeeded reads unknown rather than healthy — see targetHealthUnknown's +// own doc comment for why. func (t *targetHealthTracker) stateForLocked(e *targetHealthEntry) targetHealthState { if e.consecutiveFailures >= t.failureThreshold { return targetHealthUnhealthy } + if !e.everOK { + return targetHealthUnknown + } return targetHealthHealthy } @@ -191,13 +228,21 @@ type targetHealthSnapshot struct { source string latency time.Duration consecutiveFailures int + // observed is true iff this tracker holds an entry for (kind, name) + // — i.e. record has been called at least once, regardless of + // success or failure. F4, feat/target-health review: state alone + // cannot tell apart targetHealthUnknown's two cases (never observed + // vs. observed-but-never-succeeded), so targetHealthView (admin.go) + // reads this field, not state, to decide whether to omit + // lastCheck/lastError/source/latencyMs — omit only while !observed. + observed bool } // snapshot returns a copy of (kind, name)'s current health state — // buildAdminTargets (admin.go) and writeTargetMetrics (metrics.go) both // read through this, once per configured target, rather than exposing // t.entries directly. A target this tracker has never observed returns -// a zero-valued snapshot with state targetHealthUnknown. +// a zero-valued, !observed snapshot with state targetHealthUnknown. func (t *targetHealthTracker) snapshot(kind, name string) targetHealthSnapshot { if t == nil { return targetHealthSnapshot{state: targetHealthUnknown} @@ -209,6 +254,7 @@ func (t *targetHealthTracker) snapshot(kind, name string) targetHealthSnapshot { return targetHealthSnapshot{state: targetHealthUnknown} } return targetHealthSnapshot{ + observed: true, state: t.stateForLocked(e), lastCheck: e.lastCheck, lastError: e.lastError, @@ -321,26 +367,49 @@ func targetHealthProbeTimeout(requestTimeout time.Duration) time.Duration { // caller reaches this on a live request path and must never block // behind a lock another such call already holds) — a CAS that fails // means either a sweep is already in flight, in which case returning is -// exactly right regardless of how long ago it started, or another -// goroutine just won the race to start this interval's sweep, in which -// case returning is exactly right too. The interval check above the CAS -// is a cheap early-out only; the CAS itself is what actually enforces -// "at most one sweep in flight, and at most one starts per interval". +// exactly right, or another goroutine just won the race to start this +// interval's sweep, in which case returning is exactly right too. The +// CAS alone enforces only single-flight — "at most one sweep in flight +// at a time" — NOT the probe interval: the interval load above it and +// the CAS are not atomic with each other, so a caller whose own interval +// read predates an entirely separate sweep's full start-and-finish could +// still win the CAS once that sweep clears the flag (worst at startup, +// when every caller's first read sees last == 0). F3, feat/target-health +// review: re-check the interval AFTER winning the CAS, against +// whatever targetHealthLastSweepUnixNano holds now — set by the sweep +// that raced ahead, if any — and back out without spawning when that +// sweep already started within the interval. func (g *Gateway) maybeSweepTargetHealth() { if !g.targetHealthCfg.enabled { return } - last := atomic.LoadInt64(&g.targetHealthLastSweepUnixNano) - if last != 0 && time.Duration(time.Now().UnixNano()-last) < g.targetHealthCfg.probeInterval { + if !targetHealthIntervalElapsed(atomic.LoadInt64(&g.targetHealthLastSweepUnixNano), g.targetHealthCfg.probeInterval) { return } if !atomic.CompareAndSwapInt32(&g.targetHealthSweeping, 0, 1) { return } - atomic.StoreInt64(&g.targetHealthLastSweepUnixNano, time.Now().UnixNano()) + now := time.Now().UnixNano() + if !targetHealthIntervalElapsed(atomic.LoadInt64(&g.targetHealthLastSweepUnixNano), g.targetHealthCfg.probeInterval) { + // Lost the race: some other sweep already started (and, since + // the flag was 0 again for this CAS to succeed, already + // finished) within the interval while this call was still + // working from its own, now-stale, interval read above. + atomic.StoreInt32(&g.targetHealthSweeping, 0) + return + } + atomic.StoreInt64(&g.targetHealthLastSweepUnixNano, now) go g.sweepTargetHealth() //nolint:gosec // G118: self-terminating, bounded by mcpFederatedFanoutConcurrency*targetHealthMaxProbeTimeout worst case, and always clears targetHealthSweeping on return (including on panic) — see maybeRefresh's identical reasoning, registry.go, for why this plugin accepts a bounded, self-terminating background goroutine despite having no Close hook to await it } +// targetHealthIntervalElapsed reports whether probeInterval has passed +// since lastSweepUnixNano (0 meaning "never swept", which always counts +// as elapsed) — the exact rule maybeSweepTargetHealth applies both above +// and, again, under the CAS. +func targetHealthIntervalElapsed(lastSweepUnixNano int64, probeInterval time.Duration) bool { + return lastSweepUnixNano == 0 || time.Duration(time.Now().UnixNano()-lastSweepUnixNano) >= probeInterval +} + // sweepTargetHealth performs one active probe sweep: every configured // MCP server and agent, concurrently, behind a semaphore of // mcpFederatedFanoutConcurrency (mcp_federation.go) — reusing federation's @@ -416,7 +485,7 @@ func (g *Gateway) probeMCPTarget(name, rawURL string, timeout time.Duration) { ctx, cancel := context.WithTimeout(context.Background(), timeout) ok, err := g.probeLegacySSE(ctx, rawURL) cancel() - g.targetHealth.record(targetKindMCP, name, ok, err, time.Since(start), targetHealthSourceProbe) + g.targetHealth.record(targetKindMCP, name, rawURL, ok, err, time.Since(start), targetHealthSourceProbe) return } @@ -429,7 +498,7 @@ func (g *Gateway) probeMCPTarget(name, rawURL string, timeout time.Duration) { g.mcpBackendCloseSession(closeCtx, rawURL, sessionID) closeCancel() } - g.targetHealth.record(targetKindMCP, name, ok, err, time.Since(start), targetHealthSourceProbe) + g.targetHealth.record(targetKindMCP, name, rawURL, ok, err, time.Since(start), targetHealthSourceProbe) } // probeLegacySSE probes a legacy HTTP+SSE-transport MCP server (URL path @@ -473,12 +542,12 @@ func (g *Gateway) probeAgentTarget(name, baseURL, cardPath string, timeout time. req, err := http.NewRequestWithContext(ctx, http.MethodGet, baseURL+cardPath, nil) //nolint:gosec // operator-configured target URL, validated at construction (validateTargetURLs, mcp_a2a.go) if err != nil { - g.targetHealth.record(targetKindAgent, name, false, err, time.Since(start), targetHealthSourceProbe) + g.targetHealth.record(targetKindAgent, name, baseURL, false, err, time.Since(start), targetHealthSourceProbe) return } resp, err := g.targetClient.Do(req) if err != nil { - g.targetHealth.record(targetKindAgent, name, false, err, time.Since(start), targetHealthSourceProbe) + g.targetHealth.record(targetKindAgent, name, baseURL, false, err, time.Since(start), targetHealthSourceProbe) return } _ = resp.Body.Close() //nolint:errcheck // read-side close; nothing actionable on failure @@ -487,7 +556,7 @@ func (g *Gateway) probeAgentTarget(name, baseURL, cardPath string, timeout time. if !ok { probeErr = fmt.Errorf("agent probe: upstream returned HTTP %d", resp.StatusCode) } - g.targetHealth.record(targetKindAgent, name, ok, probeErr, time.Since(start), targetHealthSourceProbe) + g.targetHealth.record(targetKindAgent, name, baseURL, ok, probeErr, time.Since(start), targetHealthSourceProbe) } // --- passive recording: handleTargetProxy (mcp_a2a.go) --- @@ -500,7 +569,7 @@ func (g *Gateway) probeAgentTarget(name, baseURL, cardPath string, timeout time. // since a 4xx still proves the target itself answered. A client cancel // (result.clientCanceled) records nothing at all: a caller hanging up // says nothing about the target's own health. -func (g *Gateway) recordTargetProxyHealth(kind, name string, result proxyResult, ok bool, latency time.Duration) { +func (g *Gateway) recordTargetProxyHealth(kind, name, rawURL string, result proxyResult, ok bool, latency time.Duration) { if result.clientCanceled { return } @@ -509,7 +578,7 @@ func (g *Gateway) recordTargetProxyHealth(kind, name string, result proxyResult, if failed { err = targetProxyHealthError(ok, result.status) } - g.targetHealth.record(kind, name, !failed, err, latency, targetHealthSourceTraffic) + g.targetHealth.record(kind, name, rawURL, !failed, err, latency, targetHealthSourceTraffic) } // targetProxyHealthError builds a short, bounded description for a diff --git a/target_health_test.go b/target_health_test.go index fc6f1cc..3ae3424 100644 --- a/target_health_test.go +++ b/target_health_test.go @@ -27,7 +27,7 @@ func TestTargetHealthTracker_NilReceiver_Safe(t *testing.T) { t.Parallel() var tr *targetHealthTracker assert.NotPanics(t, func() { - tr.record(targetKindMCP, "alpha", true, nil, time.Millisecond, targetHealthSourceTraffic) + tr.record(targetKindMCP, "alpha", "", true, nil, time.Millisecond, targetHealthSourceTraffic) }) assert.Equal(t, targetHealthUnknown, tr.stateOf(targetKindMCP, "alpha")) snap := tr.snapshot(targetKindMCP, "alpha") @@ -37,7 +37,7 @@ func TestTargetHealthTracker_NilReceiver_Safe(t *testing.T) { func TestTargetHealthTracker_SingleSuccess_ReadsHealthy(t *testing.T) { t.Parallel() tr := newTargetHealthTracker(3) - tr.record(targetKindMCP, "alpha", true, nil, 5*time.Millisecond, targetHealthSourceTraffic) + tr.record(targetKindMCP, "alpha", "", true, nil, 5*time.Millisecond, targetHealthSourceTraffic) assert.Equal(t, targetHealthHealthy, tr.stateOf(targetKindMCP, "alpha")) } @@ -48,10 +48,10 @@ func TestTargetHealthTracker_SingleSuccess_ReadsHealthy(t *testing.T) { func TestTargetHealthTracker_BelowThreshold_StaysHealthy(t *testing.T) { t.Parallel() tr := newTargetHealthTracker(3) - tr.record(targetKindMCP, "alpha", true, nil, 0, targetHealthSourceTraffic) - tr.record(targetKindMCP, "alpha", false, errors.New("boom"), 0, targetHealthSourceTraffic) + tr.record(targetKindMCP, "alpha", "", true, nil, 0, targetHealthSourceTraffic) + tr.record(targetKindMCP, "alpha", "", false, errors.New("boom"), 0, targetHealthSourceTraffic) assert.Equal(t, targetHealthHealthy, tr.stateOf(targetKindMCP, "alpha"), "1 failure of threshold 3 must still read healthy") - tr.record(targetKindMCP, "alpha", false, errors.New("boom again"), 0, targetHealthSourceTraffic) + tr.record(targetKindMCP, "alpha", "", false, errors.New("boom again"), 0, targetHealthSourceTraffic) assert.Equal(t, targetHealthHealthy, tr.stateOf(targetKindMCP, "alpha"), "2 failures of threshold 3 must still read healthy") } @@ -59,19 +59,50 @@ func TestTargetHealthTracker_ReachesThreshold_ReadsUnhealthy(t *testing.T) { t.Parallel() tr := newTargetHealthTracker(3) for i := 0; i < 3; i++ { - tr.record(targetKindMCP, "alpha", false, errors.New("boom"), 0, targetHealthSourceTraffic) + tr.record(targetKindMCP, "alpha", "", false, errors.New("boom"), 0, targetHealthSourceTraffic) } assert.Equal(t, targetHealthUnhealthy, tr.stateOf(targetKindMCP, "alpha")) } +// TestTargetHealthTracker_FirstObservationFailure_ReadsUnknown proves F4: +// a target whose only observations so far are failures below +// failureThreshold reads "unknown", not "healthy" — the tracker has +// never actually seen it succeed, so "healthy" would be a false claim +// even though the failure-threshold gate has not fired. The threshold +// still protects against a transient first failure: this stays +// "unknown", it does not jump straight to "unhealthy". +func TestTargetHealthTracker_FirstObservationFailure_ReadsUnknown(t *testing.T) { + t.Parallel() + tr := newTargetHealthTracker(3) + tr.record(targetKindMCP, "alpha", "", false, errors.New("boom"), 0, targetHealthSourceTraffic) + assert.Equal(t, targetHealthUnknown, tr.stateOf(targetKindMCP, "alpha"), "never having succeeded must read unknown, not healthy, below threshold") + + tr.record(targetKindMCP, "alpha", "", false, errors.New("boom again"), 0, targetHealthSourceTraffic) + assert.Equal(t, targetHealthUnknown, tr.stateOf(targetKindMCP, "alpha"), "still unknown after a second failure, still below threshold") +} + +// TestTargetHealthTracker_SuccessThenFewFailures_ReadsHealthy re-proves +// TestTargetHealthTracker_BelowThreshold_StaysHealthy's claim in F4 +// terms: once a target HAS succeeded at least once, 1-2 subsequent +// failures below threshold still read "healthy" — F4 only changes the +// NEVER-succeeded case. +func TestTargetHealthTracker_SuccessThenFewFailures_ReadsHealthy(t *testing.T) { + t.Parallel() + tr := newTargetHealthTracker(3) + tr.record(targetKindMCP, "alpha", "", true, nil, 0, targetHealthSourceTraffic) + tr.record(targetKindMCP, "alpha", "", false, errors.New("boom"), 0, targetHealthSourceTraffic) + tr.record(targetKindMCP, "alpha", "", false, errors.New("boom again"), 0, targetHealthSourceTraffic) + assert.Equal(t, targetHealthHealthy, tr.stateOf(targetKindMCP, "alpha")) +} + func TestTargetHealthTracker_SuccessAfterUnhealthy_RecoversToHealthy(t *testing.T) { t.Parallel() tr := newTargetHealthTracker(2) - tr.record(targetKindMCP, "alpha", false, errors.New("boom"), 0, targetHealthSourceTraffic) - tr.record(targetKindMCP, "alpha", false, errors.New("boom"), 0, targetHealthSourceTraffic) + tr.record(targetKindMCP, "alpha", "", false, errors.New("boom"), 0, targetHealthSourceTraffic) + tr.record(targetKindMCP, "alpha", "", false, errors.New("boom"), 0, targetHealthSourceTraffic) require.Equal(t, targetHealthUnhealthy, tr.stateOf(targetKindMCP, "alpha")) - tr.record(targetKindMCP, "alpha", true, nil, 0, targetHealthSourceTraffic) + tr.record(targetKindMCP, "alpha", "", true, nil, 0, targetHealthSourceTraffic) assert.Equal(t, targetHealthHealthy, tr.stateOf(targetKindMCP, "alpha")) snap := tr.snapshot(targetKindMCP, "alpha") assert.Equal(t, 0, snap.consecutiveFailures, "a success must reset consecutiveFailures to 0") @@ -82,7 +113,7 @@ func TestTargetHealthTracker_ErrorString_BoundedLength(t *testing.T) { t.Parallel() tr := newTargetHealthTracker(3) long := strings.Repeat("x", targetHealthMaxErrorRunes+500) - tr.record(targetKindMCP, "alpha", false, errors.New(long), 0, targetHealthSourceProbe) + tr.record(targetKindMCP, "alpha", "", false, errors.New(long), 0, targetHealthSourceProbe) snap := tr.snapshot(targetKindMCP, "alpha") assert.LessOrEqual(t, len([]rune(snap.lastError)), targetHealthMaxErrorRunes) assert.Equal(t, strings.Repeat("x", targetHealthMaxErrorRunes), snap.lastError) @@ -91,15 +122,65 @@ func TestTargetHealthTracker_ErrorString_BoundedLength(t *testing.T) { func TestTargetHealthTracker_ShortError_Unchanged(t *testing.T) { t.Parallel() tr := newTargetHealthTracker(3) - tr.record(targetKindMCP, "alpha", false, errors.New("dial tcp: connection refused"), 0, targetHealthSourceProbe) + tr.record(targetKindMCP, "alpha", "", false, errors.New("dial tcp: connection refused"), 0, targetHealthSourceProbe) snap := tr.snapshot(targetKindMCP, "alpha") assert.Equal(t, "dial tcp: connection refused", snap.lastError) } +// TestTargetHealthTracker_Record_ScrubsCredentialsFromLastError proves F1: +// record scrubs an error's embedded target URL through sanitizeProviderErr +// before storing lastError, so a target URL configured with an embedded +// credential (an "api-key" query parameter, or userinfo) never survives +// into GET /admin/api/targets. +func TestTargetHealthTracker_Record_ScrubsCredentialsFromLastError(t *testing.T) { + t.Parallel() + + t.Run("query parameter credential", func(t *testing.T) { + t.Parallel() + tr := newTargetHealthTracker(3) + rawURL := "https://svc.internal/mcp?api-key=SECRETVALUE" + err := errors.New(`Get "` + rawURL + `": dial tcp: connection refused`) + tr.record(targetKindMCP, "alpha", rawURL, false, err, 0, targetHealthSourceProbe) + snap := tr.snapshot(targetKindMCP, "alpha") + assert.NotContains(t, snap.lastError, "SECRETVALUE") + assert.NotContains(t, snap.lastError, "api-key=") + }) + + t.Run("userinfo credential", func(t *testing.T) { + t.Parallel() + tr := newTargetHealthTracker(3) + rawURL := "https://tok3n@svc.internal/mcp" + err := errors.New(`Get "` + rawURL + `": dial tcp: connection refused`) + tr.record(targetKindMCP, "alpha", rawURL, false, err, 0, targetHealthSourceProbe) + snap := tr.snapshot(targetKindMCP, "alpha") + assert.NotContains(t, snap.lastError, "tok3n") + }) +} + +// TestTargetHealthTracker_Record_ScrubsBeforeTruncate proves F1: the URL +// scrub runs against err.Error() BEFORE truncateRunes bounds it to +// targetHealthMaxErrorRunes — even when the embedded URL straddles the +// truncation point, no fragment of the secret survives. +func TestTargetHealthTracker_Record_ScrubsBeforeTruncate(t *testing.T) { + t.Parallel() + tr := newTargetHealthTracker(3) + rawURL := "https://svc.internal/mcp?api-key=" + strings.Repeat("S", 60) + prefix := strings.Repeat("p", 180) + require.Less(t, len([]rune(prefix)), targetHealthMaxErrorRunes, "prefix must start before the truncation point") + require.Greater(t, len([]rune(prefix))+len([]rune(rawURL)), targetHealthMaxErrorRunes, "the url must straddle the truncation point") + errText := prefix + " " + rawURL + " connection refused" + + tr.record(targetKindMCP, "alpha", rawURL, false, errors.New(errText), 0, targetHealthSourceProbe) + snap := tr.snapshot(targetKindMCP, "alpha") + assert.LessOrEqual(t, len([]rune(snap.lastError)), targetHealthMaxErrorRunes) + assert.NotContains(t, snap.lastError, "SSSS", "no fragment of the secret must survive truncation") + assert.NotContains(t, snap.lastError, "api-key=") +} + func TestTargetHealthTracker_KeyedByKindAndName_Independent(t *testing.T) { t.Parallel() tr := newTargetHealthTracker(1) - tr.record(targetKindMCP, "shared", false, errors.New("boom"), 0, targetHealthSourceTraffic) + tr.record(targetKindMCP, "shared", "", false, errors.New("boom"), 0, targetHealthSourceTraffic) assert.Equal(t, targetHealthUnhealthy, tr.stateOf(targetKindMCP, "shared")) assert.Equal(t, targetHealthUnknown, tr.stateOf(targetKindAgent, "shared"), "an mcp server and an agent sharing a name must track independently") } @@ -109,7 +190,7 @@ func TestTargetHealthTracker_Snapshot_FieldsPopulated(t *testing.T) { now := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) tr := newTargetHealthTracker(3) tr.nowFn = func() time.Time { return now } - tr.record(targetKindAgent, "beta", true, nil, 42*time.Millisecond, targetHealthSourceProbe) + tr.record(targetKindAgent, "beta", "", true, nil, 42*time.Millisecond, targetHealthSourceProbe) snap := tr.snapshot(targetKindAgent, "beta") assert.Equal(t, targetHealthHealthy, snap.state) @@ -127,7 +208,7 @@ func TestTargetHealthTracker_ConcurrentRecord_NoRace(t *testing.T) { wg.Add(1) go func(i int) { defer wg.Done() - tr.record(targetKindMCP, "alpha", i%2 == 0, errors.New("x"), time.Millisecond, targetHealthSourceTraffic) + tr.record(targetKindMCP, "alpha", "", i%2 == 0, errors.New("x"), time.Millisecond, targetHealthSourceTraffic) }(i) } wg.Wait() @@ -226,9 +307,14 @@ func newTargetHealthSweepGateway(t *testing.T, enabled bool, probeInterval time. func TestMaybeSweepTargetHealth_Disabled_NoOutboundRequests(t *testing.T) { t.Parallel() - hits := 0 + // F6, feat/target-health review: atomic even though a disabled sweep + // must never actually invoke this handler — a regression that DID + // invoke it would otherwise race this counter's read below under + // -race, obscuring the real bug (an unexpected outbound probe) behind + // a data-race report instead of the intended assert.Equal failure. + var hits int32 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - hits++ + atomic.AddInt32(&hits, 1) w.WriteHeader(http.StatusOK) })) defer srv.Close() @@ -240,7 +326,7 @@ func TestMaybeSweepTargetHealth_Disabled_NoOutboundRequests(t *testing.T) { // wait on — a brief sleep would only prove absence of a signal we // never expect to see. time.Sleep(20 * time.Millisecond) - assert.Equal(t, 0, hits, "a disabled targetHealth must never send an outbound probe") + assert.Equal(t, int32(0), atomic.LoadInt32(&hits), "a disabled targetHealth must never send an outbound probe") assert.Equal(t, targetHealthUnknown, gw.targetHealth.stateOf(targetKindMCP, "alpha")) } @@ -273,7 +359,20 @@ func TestMaybeSweepTargetHealth_ConcurrentTriggers_SingleFlight(t *testing.T) { w.Header().Set("Content-Type", "application/json") _, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":"federated","result":{}}`)) })) - defer srv.Close() + // F6, feat/target-health review: registered in this order so + // t.Cleanup's LIFO order unblocks the handler (closing release) + // BEFORE srv.Close runs — httptest.Server.Close blocks until + // outstanding requests complete, and the assertions below can + // t.Fatalf/t.Errorf before ever reaching the explicit close(release) + // further down; without this a failing assertion left the handler + // permanently blocked and srv.Close hung until the whole test + // binary's 10-minute timeout, rather than failing fast. unblock is + // idempotent (sync.Once) since the happy path below also calls it + // directly, before waiting on the sweep's resulting state. + var unblockOnce sync.Once + unblock := func() { unblockOnce.Do(func() { close(release) }) } + t.Cleanup(srv.Close) + t.Cleanup(unblock) gw := newTargetHealthSweepGateway(t, true, time.Hour, srv.URL) @@ -292,11 +391,64 @@ func TestMaybeSweepTargetHealth_ConcurrentTriggers_SingleFlight(t *testing.T) { // maybeSweepTargetHealth concurrently, at most one outbound request // must have been sent. waitForCondition(t, func() bool { return atomic.LoadInt32(&hits) > 0 }, time.Second) - close(release) + unblock() waitForCondition(t, func() bool { return gw.targetHealth.stateOf(targetKindMCP, "alpha") != targetHealthUnknown }, time.Second) assert.Equal(t, int32(1), atomic.LoadInt32(&hits), "concurrent triggers within one interval must produce exactly one sweep") } +// TestMaybeSweepTargetHealth_IntervalEnforcedAcrossRapidCalls proves F3: +// the interval load (top of maybeSweepTargetHealth) and the CAS that wins +// the single-flight flag are not atomic with each other. Without a +// re-check under the CAS, a caller whose own interval read predates an +// entirely separate sweep's start-and-finish could still win the CAS +// once that sweep clears the flag, and spawn a second sweep inside the +// same interval — worst at startup, when every caller's first read sees +// last == 0. A fast-completing first sweep, immediately followed by a +// burst of concurrent callers, must still yield exactly one sweep within +// probeInterval; a call after the interval has genuinely elapsed must +// start a second one. +func TestMaybeSweepTargetHealth_IntervalEnforcedAcrossRapidCalls(t *testing.T) { + t.Parallel() + var hits int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&hits, 1) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":"federated","result":{}}`)) + })) + defer srv.Close() + + probeInterval := 150 * time.Millisecond + gw := newTargetHealthSweepGateway(t, true, probeInterval, srv.URL) + + // First sweep: the probe answers immediately, so this completes fast + // and clears targetHealthSweeping well inside probeInterval. + gw.maybeSweepTargetHealth() + waitForCondition(t, func() bool { return atomic.LoadInt32(&hits) > 0 }, time.Second) + waitForCondition(t, func() bool { return gw.targetHealth.stateOf(targetKindMCP, "alpha") != targetHealthUnknown }, time.Second) + + // A burst of concurrent callers immediately after — still well inside + // probeInterval — must not start a second sweep. + var wg sync.WaitGroup + for i := 0; i < 50; i++ { + wg.Add(1) + go func() { + defer wg.Done() + gw.maybeSweepTargetHealth() + }() + } + wg.Wait() + time.Sleep(30 * time.Millisecond) // let any wrongly-spawned sweep reach the server + if got := atomic.LoadInt32(&hits); got != 1 { + t.Errorf("hits = %d, want exactly 1 (a burst of concurrent callers inside probeInterval must not start a second sweep)", got) + } + + // Once probeInterval has genuinely elapsed, a new call starts a + // second sweep. + time.Sleep(probeInterval) + gw.maybeSweepTargetHealth() + waitForCondition(t, func() bool { return atomic.LoadInt32(&hits) == 2 }, time.Second) +} + // TestMaybeSweepTargetHealth_LegacySSE_ClosesBodyWithoutReading proves a // legacy "/sse" MCP server's probe uses a plain GET and closes the body // immediately: a test server that never closes its stream must not hang @@ -304,6 +456,7 @@ func TestMaybeSweepTargetHealth_ConcurrentTriggers_SingleFlight(t *testing.T) { func TestMaybeSweepTargetHealth_LegacySSE_ClosesBodyWithoutReading(t *testing.T) { t.Parallel() headersSent := make(chan struct{}) + release := make(chan struct{}) srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "text/event-stream") w.WriteHeader(http.StatusOK) @@ -312,9 +465,18 @@ func TestMaybeSweepTargetHealth_LegacySSE_ClosesBodyWithoutReading(t *testing.T) if ok { flusher.Flush() } - <-r.Context().Done() // never closes on its own; relies on the probe not reading + <-release // never closes on its own; relies on the probe not reading })) - defer srv.Close() + // F6, feat/target-health review: previously relied on the PROBE'S OWN + // 5s context timeout (targetTimeout below) to unblock this handler, + // so srv.Close (deferred) silently added a real ~5s stall to every + // run of this test. An explicit release channel, unblocked via + // t.Cleanup BEFORE srv.Close (LIFO — see + // TestMaybeSweepTargetHealth_ConcurrentTriggers_SingleFlight's own + // comment for the identical ordering reasoning), makes this test + // fast AND immune to a failing assertion ever leaving it blocked. + t.Cleanup(srv.Close) + t.Cleanup(func() { close(release) }) adapters := map[string]providerAdapter{"openai": newFakeAdapter("openai")} cfg := &Config{