diff --git a/README.md b/README.md index 320a737..6212f7a 100644 --- a/README.md +++ b/README.md @@ -652,7 +652,7 @@ Recently cached: | Endpoint | Description | |----------|-------------| | `GET /` | Dashboard (web UI) | -| `GET /health` | Health check (JSON; HTTP 200 healthy, 503 unhealthy) | +| `GET /health` | Health check and upstream circuit breaker state (JSON; HTTP 200 healthy, 503 unhealthy) | | `GET /stats` | Cache statistics (JSON) | | `GET /metrics` | Prometheus metrics | | `GET /npm/*` | npm registry protocol | @@ -895,8 +895,14 @@ The proxy exposes Prometheus metrics at `GET /metrics`. All metric names are pre | `proxy_storage_errors_total` | counter | `operation` | Storage read/write failures | | `proxy_active_requests` | gauge | | In-flight requests | | `proxy_health_probe_failures_total` | counter | `step` | Storage health probe failures by failing step (`write`, `size`, `read`, `verify`, `delete`). | +| `proxy_circuit_breaker_state` | gauge | `registry` | Artifact-fetch circuit breaker state per upstream host (0 closed, 2 open). Published once that host's breaker has tripped. | +| `proxy_circuit_breaker_trips_total` | counter | `registry` | Circuit breaker trips per upstream host. | -Cache size and artifact count are refreshed every 60 seconds. The remaining metrics update on each request. +Cache size and artifact count are refreshed every 60 seconds. Circuit breaker state is read from the fetcher on each scrape of `/metrics` and each `/health` request, so `proxy_circuit_breaker_trips_total` counts the trips visible between those reads — a breaker that opens and recovers entirely between two scrapes is not counted. The remaining metrics update on each request. + +The breaker metrics carry one series per upstream host, but only for hosts whose breaker has tripped at least once since startup. A breaker is created per host the proxy fetches artifacts from, and for some ecosystems that host comes from upstream metadata rather than from configuration (composer takes it from a package's `dist.url`, helm from the chart URLs in `index.yaml`), so publishing every host would let upstream content grow the series count for the lifetime of the process. Once a host has tripped it keeps reporting, so a recovery still shows up as a transition to 0 rather than as a series that vanishes. `/health` is not a persistent time series and lists every breaker, tripped or not. + +Alert on `proxy_circuit_breaker_state == 2` sustained for more than a few minutes: while a breaker is open, artifact downloads for that upstream fail with HTTP 502 on every cache miss, and only a single probe request per backoff interval reaches the upstream. Cached artifacts keep serving, and so does metadata for the same ecosystem (metadata does not go through the circuit breaker), so installs fail in a way that looks like a partial upstream outage. ### Health Check @@ -908,12 +914,20 @@ Cache size and artifact count are refreshed every 60 seconds. The remaining metr "checks": { "database": {"status": "ok"}, "storage": {"status": "ok"} + }, + "circuit_breakers": { + "registry.npmjs.org": "closed", + "static.crates.io": "open" } } ``` Failing checks include an `"error"` field. Storage failures also include a `"step"` field identifying which probe step failed (`write`, `size`, `read`, `verify`, `delete`). When the database check fails, the storage entry reports `{"status": "skipped"}` so the response always carries the same key set. +`circuit_breakers` reports the state of each upstream's artifact-fetch circuit breaker (`"open"` or `"closed"`), keyed by upstream host. The key is omitted until the proxy has fetched an artifact from at least one upstream, and a host appears only once a breaker has been created for it. Breakers trip after repeated upstream failures and retry the upstream after an exponential backoff. While one is open, artifact downloads for that host return HTTP 502 on a cache miss without contacting the upstream; already-cached artifacts are still served from storage, since the cache is checked before the fetcher. A breaker is reported as `"open"` throughout its backoff, including the half-open window in which it admits one probe request to test recovery. Breaker state is per process and in memory, so a restart always clears it — worth knowing when a breaker stays open after the upstream has recovered. + +An open breaker does **not** set `status` to `"error"` or change the HTTP status code: it reports a specific upstream refusing to serve, not this proxy being unfit to receive traffic, and failing the readiness probe over one unhealthy upstream would pull the pod out of rotation for every other ecosystem too. Use `proxy_circuit_breaker_state` for alerting on it. + Storage probe results are cached for `health.storage_probe_interval` (default 30s) to bound the cost of probing remote backends. A probe holds an internal mutex for up to 10 seconds (the hardcoded per-probe timeout), so `/health` is intended as a Kubernetes **readiness** probe rather than a liveness probe — a slow S3 round-trip should pull the pod from rotation, not restart it. Scrape config for Prometheus: diff --git a/docs/architecture.md b/docs/architecture.md index 6d9bfda..0d2c6ab 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -279,7 +279,7 @@ HTTP server setup, web UI, and API handlers. - Web UI under `/ui`: dashboard, package browser, source browser, version comparison - Templates are embedded in the binary via `//go:embed` - Enrichment API for package metadata, vulnerability scanning, and outdated detection -- Health, stats, and Prometheus metrics endpoints. `/health` runs an active write → size-check → read → verify → delete probe against the storage backend and returns a structured JSON response (`HealthResponse`) with `"ok"` / `"error"` status per subsystem. Probe results are cached (default 30 s, configurable via `health.storage_probe_interval`) to avoid overwhelming remote backends. +- Health, stats, and Prometheus metrics endpoints. `/health` runs an active write → size-check → read → verify → delete probe against the storage backend and returns a structured JSON response (`HealthResponse`) with `"ok"` / `"error"` status per subsystem. Probe results are cached (default 30 s, configurable via `health.storage_probe_interval`) to avoid overwhelming remote backends. The response also carries a `circuit_breakers` map reporting each upstream host's artifact-fetch breaker as `"open"` or `"closed"`; the same state is published as the `proxy_circuit_breaker_state` gauge on each `/metrics` scrape. An open breaker leaves the overall status `"ok"` — it describes an upstream, not this proxy. ### `internal/metrics` diff --git a/docs/swagger/docs.go b/docs/swagger/docs.go index c4b21f3..1f0334f 100644 --- a/docs/swagger/docs.go +++ b/docs/swagger/docs.go @@ -538,6 +538,13 @@ const docTemplate = `{ "$ref": "#/definitions/server.HealthCheck" } }, + "circuit_breakers": { + "description": "CircuitBreakers reports the state (\"open\" or \"closed\") of each upstream\nregistry's artifact-fetch circuit breaker, omitted when no breaker has\nbeen created yet. An open breaker fails every artifact fetch for that\nhost without contacting it, but says nothing about this proxy's own\nhealth, so it does not change Status.", + "type": "object", + "additionalProperties": { + "type": "string" + } + }, "status": { "type": "string" } diff --git a/docs/swagger/swagger.json b/docs/swagger/swagger.json index 898f580..4f0622c 100644 --- a/docs/swagger/swagger.json +++ b/docs/swagger/swagger.json @@ -531,6 +531,13 @@ "$ref": "#/definitions/server.HealthCheck" } }, + "circuit_breakers": { + "description": "CircuitBreakers reports the state (\"open\" or \"closed\") of each upstream\nregistry's artifact-fetch circuit breaker, omitted when no breaker has\nbeen created yet. An open breaker fails every artifact fetch for that\nhost without contacting it, but says nothing about this proxy's own\nhealth, so it does not change Status.", + "type": "object", + "additionalProperties": { + "type": "string" + } + }, "status": { "type": "string" } diff --git a/internal/server/breakers.go b/internal/server/breakers.go new file mode 100644 index 0000000..ec69969 --- /dev/null +++ b/internal/server/breakers.go @@ -0,0 +1,118 @@ +package server + +import ( + "log/slog" + "sync" + + "github.com/git-pkgs/proxy/internal/metrics" +) + +// Gauge values for proxy_circuit_breaker_state. The fetcher reports only open +// or closed, so half-open (1) is never published. +const ( + breakerGaugeClosed = 0 + breakerGaugeOpen = 2 +) + +const ( + breakerStateOpen = "open" + breakerStateClosed = "closed" +) + +// breakerStateSource reports circuit breaker state per registry host, keyed by +// host, with values breakerStateOpen or breakerStateClosed. Implemented by +// fetch.CircuitBreakerFetcher. +type breakerStateSource interface { + GetBreakerState() map[string]string +} + +// breakerMonitor mirrors the artifact fetcher's per-registry circuit breaker +// state into Prometheus metrics, the health report, and the log. +// +// Breaker state lives only in the fetcher's memory. While a breaker is open +// every artifact fetch for that host that misses the cache fails without +// reaching the upstream, which looks identical to an upstream outage from the +// outside: metadata still serves (it does not go through the fetcher), other +// registries still serve, and /health reports the database and storage as fine. +// Publishing the state makes that distinguishable. +type breakerMonitor struct { + source breakerStateSource + logger *slog.Logger + + // mu serializes snapshots. It guards seen, which holds one entry per + // registry that has tripped at least once in this process — the only + // registries published as metrics — and keeps each state read paired with + // the updates it produces. + mu sync.Mutex + seen map[string]string +} + +func newBreakerMonitor(source breakerStateSource, logger *slog.Logger) *breakerMonitor { + if logger == nil { + logger = slog.Default() + } + return &breakerMonitor{ + source: source, + logger: logger, + seen: map[string]string{}, + } +} + +// snapshot returns the current state of every breaker the fetcher has created, +// keyed by registry host, and mirrors it into the breaker metrics as a side +// effect. It returns nil for a nil monitor so callers that build a Server +// without a fetcher (tests) need no special case. +// +// Only registries that have tripped at least once are published as metrics. +// The fetcher creates a breaker per host it fetches from, and for some +// ecosystems that host comes from upstream metadata rather than configuration +// (composer takes it from a package's dist.url, helm from the chart URLs in +// index.yaml), so publishing every host would let upstream content grow the +// series count for the life of the process. A host that has never tripped +// carries no information a series could convey; once it trips it keeps +// reporting, including the 0 that marks its recovery. /health is a per-request +// response rather than a persistent series, so it reports every breaker. +// +// Trips are counted on the closed→open transitions observed between calls, +// because the fetcher exposes current state rather than trip events: a breaker +// that opens and recovers entirely between two calls is not counted. +func (m *breakerMonitor) snapshot() map[string]string { + if m == nil || m.source == nil { + return nil + } + + m.mu.Lock() + defer m.mu.Unlock() + + // Read under the lock. Two concurrent snapshots — a /health request and a + // /metrics scrape landing during a transition — can otherwise apply their + // reads to seen in the opposite order, counting one trip twice, logging a + // close for a breaker that is still open, and leaving the gauge at 0 until + // the next call. + states := m.source.GetBreakerState() + + for registry, state := range states { + previous, published := m.seen[registry] + + switch { + case state == breakerStateOpen && previous != breakerStateOpen: + metrics.RecordCircuitBreakerTrip(registry) + m.logger.Error("circuit breaker open, artifact fetches for this registry "+ + "fail without contacting it", "registry", registry) + case state == breakerStateClosed && previous == breakerStateOpen: + m.logger.Info("circuit breaker closed", "registry", registry) + case state == breakerStateClosed && !published: + // Never tripped: nothing to publish. + continue + } + + gauge := breakerGaugeClosed + if state == breakerStateOpen { + gauge = breakerGaugeOpen + } + metrics.UpdateCircuitBreakerState(registry, gauge) + m.seen[registry] = state + } + + return states +} diff --git a/internal/server/breakers_test.go b/internal/server/breakers_test.go new file mode 100644 index 0000000..c366103 --- /dev/null +++ b/internal/server/breakers_test.go @@ -0,0 +1,262 @@ +package server + +import ( + "context" + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/git-pkgs/proxy/internal/metrics" + "github.com/git-pkgs/registries/fetch" + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" +) + +// cbThreshold in registries/fetch: consecutive failures needed to trip. +const breakerTripFailures = 5 + +// fakeBreakerSource reports breaker state without a real fetcher, so tests can +// drive transitions that would otherwise need to wait out a 30s backoff. +type fakeBreakerSource struct { + states map[string]string +} + +func (f *fakeBreakerSource) GetBreakerState() map[string]string { + states := make(map[string]string, len(f.states)) + for registry, state := range f.states { + states[registry] = state + } + return states +} + +// newTrippedMonitor returns a monitor over a real circuit-breaker fetcher whose +// breaker for the test server's host has been tripped by repeated 5xx +// responses, plus that host. +func newTrippedMonitor(t *testing.T) (*breakerMonitor, string) { + t.Helper() + + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusBadGateway) + })) + t.Cleanup(upstream.Close) + + monitor, fetcher, host := newMonitorFor(t, upstream) + for range breakerTripFailures { + if _, err := fetcher.Fetch(context.Background(), upstream.URL+"/artifact.tgz"); err == nil { + t.Fatal("fetch against a 502 upstream should fail") + } + } + return monitor, host +} + +// newMonitorFor builds a monitor over a circuit-breaker fetcher that talks to +// srv through its own client, bypassing the SSRF dial gate that would otherwise +// refuse the loopback address. +func newMonitorFor(t *testing.T, srv *httptest.Server) ( + monitor *breakerMonitor, fetcher *fetch.CircuitBreakerFetcher, host string, +) { + t.Helper() + + base := fetch.NewFetcher( + fetch.WithHTTPClient(srv.Client()), + fetch.WithMaxRetries(0), + ) + t.Cleanup(func() { _ = base.Close() }) + + fetcher = fetch.NewCircuitBreakerFetcher(base) + return newBreakerMonitor(fetcher, slog.New(slog.DiscardHandler)), + fetcher, + strings.TrimPrefix(srv.URL, "http://") +} + +// metricValue returns the value of the series carrying registry=want, and +// whether such a series exists at all. It collects rather than calling +// WithLabelValues, which would create the series it is looking for. +func metricValue(t *testing.T, collector prometheus.Collector, want string) (value float64, found bool) { + t.Helper() + + ch := make(chan prometheus.Metric, 64) + go func() { + collector.Collect(ch) + close(ch) + }() + + for metric := range ch { + var parsed dto.Metric + if err := metric.Write(&parsed); err != nil { + t.Fatalf("writing metric: %v", err) + } + for _, label := range parsed.GetLabel() { + if label.GetName() != "registry" || label.GetValue() != want { + continue + } + if gauge := parsed.GetGauge(); gauge != nil { + return gauge.GetValue(), true + } + return parsed.GetCounter().GetValue(), true + } + } + return 0, false +} + +// resetSeries drops any series for host left behind by an earlier test, since +// httptest ports can be reused within a process and the metrics registry is +// global. Absolute trip counts stay meaningful after it. +func resetSeries(host string) { + metrics.CircuitBreakerState.DeleteLabelValues(host) + metrics.CircuitBreakerTrips.DeleteLabelValues(host) +} + +func TestBreakerMonitor_OpenBreakerReportedAndCounted(t *testing.T) { + monitor, host := newTrippedMonitor(t) + resetSeries(host) + + if state := monitor.snapshot()[host]; state != breakerStateOpen { + t.Fatalf("state for %s = %q, want open", host, state) + } + + gauge, found := metricValue(t, metrics.CircuitBreakerState, host) + if !found { + t.Fatalf("no state gauge published for %s", host) + } + if gauge != breakerGaugeOpen { + t.Errorf("state gauge = %v, want %v", gauge, breakerGaugeOpen) + } + if trips, _ := metricValue(t, metrics.CircuitBreakerTrips, host); trips != 1 { + t.Errorf("trips = %v, want 1", trips) + } + + // A breaker that stays open is one trip, not one per scrape. + monitor.snapshot() + monitor.snapshot() + if trips, _ := metricValue(t, metrics.CircuitBreakerTrips, host); trips != 1 { + t.Errorf("trips after further snapshots = %v, want 1", trips) + } +} + +// A breaker per fetched host is created even for hosts that come from upstream +// metadata (composer dist.url, helm chart URLs), so publishing a series for +// every one of them would let upstream content grow the series count without +// bound. Only registries that have actually tripped are published. +func TestBreakerMonitor_HealthyRegistryPublishesNoSeries(t *testing.T) { + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte("artifact")) + })) + defer upstream.Close() + + monitor, fetcher, host := newMonitorFor(t, upstream) + resetSeries(host) + + artifact, err := fetcher.Fetch(context.Background(), upstream.URL+"/artifact.tgz") + if err != nil { + t.Fatalf("fetch: %v", err) + } + _, _ = io.Copy(io.Discard, artifact.Body) + _ = artifact.Body.Close() + + // The breaker exists and is reported to /health... + if state := monitor.snapshot()[host]; state != breakerStateClosed { + t.Fatalf("state for %s = %q, want closed", host, state) + } + // ...but carries no metric series. + if _, found := metricValue(t, metrics.CircuitBreakerState, host); found { + t.Errorf("state gauge published for %s, want none until it trips", host) + } + if _, found := metricValue(t, metrics.CircuitBreakerTrips, host); found { + t.Errorf("trip counter published for %s, want none until it trips", host) + } +} + +// Once a registry has tripped it keeps reporting, so recovery is visible as a +// transition to 0 rather than as a series that disappears. +func TestBreakerMonitor_RecoveryReportedAfterTrip(t *testing.T) { + const host = "recovering.example.com" + + resetSeries(host) + + source := &fakeBreakerSource{states: map[string]string{host: breakerStateOpen}} + monitor := newBreakerMonitor(source, slog.New(slog.DiscardHandler)) + + monitor.snapshot() + if gauge, _ := metricValue(t, metrics.CircuitBreakerState, host); gauge != breakerGaugeOpen { + t.Fatalf("state gauge = %v, want %v", gauge, breakerGaugeOpen) + } + + source.states[host] = breakerStateClosed + if state := monitor.snapshot()[host]; state != breakerStateClosed { + t.Fatalf("state for %s = %q, want closed", host, state) + } + gauge, found := metricValue(t, metrics.CircuitBreakerState, host) + if !found { + t.Fatal("state gauge disappeared after recovery, want 0") + } + if gauge != breakerGaugeClosed { + t.Errorf("state gauge = %v, want %v", gauge, breakerGaugeClosed) + } + if trips, _ := metricValue(t, metrics.CircuitBreakerTrips, host); trips != 1 { + t.Errorf("trips = %v, want 1", trips) + } + + // Tripping again after recovery counts a second trip. + source.states[host] = breakerStateOpen + monitor.snapshot() + if trips, _ := metricValue(t, metrics.CircuitBreakerTrips, host); trips != 2 { + t.Errorf("trips after re-trip = %v, want 2", trips) + } +} + +func TestBreakerMonitor_NilSafe(t *testing.T) { + var nilMonitor *breakerMonitor + if states := nilMonitor.snapshot(); states != nil { + t.Errorf("nil monitor snapshot = %v, want nil", states) + } + if states := newBreakerMonitor(nil, nil).snapshot(); states != nil { + t.Errorf("snapshot without a state source = %v, want nil", states) + } +} + +func TestHealthEndpoint_ReportsOpenBreaker(t *testing.T) { + ts := newTestServer(t) + defer ts.close() + + monitor, host := newTrippedMonitor(t) + ts.server.breakers = monitor + + req := httptest.NewRequest("GET", "/health", nil) + w := httptest.NewRecorder() + ts.handler.ServeHTTP(w, req) + + // An unreachable upstream is not this proxy being unhealthy: the breaker + // state is reported, but the probe still passes. + if w.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body: %s", w.Code, w.Body.String()) + } + + var resp HealthResponse + if err := json.NewDecoder(w.Body).Decode(&resp); err != nil { + t.Fatalf("decoding response: %v", err) + } + if resp.Status != "ok" { + t.Errorf("status = %q, want ok", resp.Status) + } + if got := resp.CircuitBreakers[host]; got != breakerStateOpen { + t.Errorf("circuit_breakers[%s] = %q, want open", host, got) + } +} + +func TestHealthEndpoint_OmitsBreakersWhenNoneExist(t *testing.T) { + ts := newTestServer(t) + defer ts.close() + + req := httptest.NewRequest("GET", "/health", nil) + w := httptest.NewRecorder() + ts.handler.ServeHTTP(w, req) + + if strings.Contains(w.Body.String(), "circuit_breakers") { + t.Errorf("body should omit circuit_breakers when no breaker exists: %s", w.Body.String()) + } +} diff --git a/internal/server/health.go b/internal/server/health.go index f4e4847..483e9dc 100644 --- a/internal/server/health.go +++ b/internal/server/health.go @@ -30,6 +30,12 @@ const ( type HealthResponse struct { Status string `json:"status"` Checks map[string]HealthCheck `json:"checks"` + // CircuitBreakers reports the state ("open" or "closed") of each upstream + // registry's artifact-fetch circuit breaker, omitted when no breaker has + // been created yet. An open breaker fails every artifact fetch for that + // host without contacting it, but says nothing about this proxy's own + // health, so it does not change Status. + CircuitBreakers map[string]string `json:"circuit_breakers,omitempty"` } // HealthCheck reports the status of a single subsystem check. diff --git a/internal/server/server.go b/internal/server/server.go index bb964e8..357ffa0 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -98,6 +98,7 @@ type Server struct { cancel context.CancelFunc healthCache *healthCache accessLog *accesslog.Logger + breakers *breakerMonitor } // New creates a new Server with the given configuration. @@ -198,6 +199,7 @@ func (s *Server) Start() error { // Create shared components with circuit breaker. baseFetcher := fetch.NewFetcher(fetch.WithHTTPClient(&artifactClient)) fetcher := fetch.NewCircuitBreakerFetcher(baseFetcher) + s.breakers = newBreakerMonitor(fetcher, s.logger) resolver := fetch.NewResolver() cd := &cooldown.Config{ Default: s.cfg.Cooldown.Default, @@ -291,6 +293,8 @@ func (s *Server) Start() error { r.Get("/stats", s.handleStats) r.Get("/openapi.json", s.handleOpenAPIJSON) r.Get("/metrics", func(w http.ResponseWriter, r *http.Request) { + // Breaker state is only held in the fetcher, so publish it on scrape. + s.breakers.snapshot() metrics.Handler().ServeHTTP(w, r) }) @@ -907,7 +911,11 @@ func (s *Server) showComparePage(w http.ResponseWriter, r *http.Request, ecosyst func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") - resp := HealthResponse{Status: "ok", Checks: map[string]HealthCheck{}} + resp := HealthResponse{ + Status: "ok", + Checks: map[string]HealthCheck{}, + CircuitBreakers: s.breakers.snapshot(), + } // Database check (short-circuit; do not waste a storage probe call when DB is down). // On DB failure the storage entry reports "skipped" rather than being omitted so diff --git a/internal/server/server_test.go b/internal/server/server_test.go index 77c32ae..1f4b202 100644 --- a/internal/server/server_test.go +++ b/internal/server/server_test.go @@ -30,6 +30,7 @@ import ( type testServer struct { handler http.Handler + server *Server db *database.DB storage storage.Storage tempDir string @@ -129,6 +130,7 @@ func newTestServer(t *testing.T) *testServer { return &testServer{ handler: r, + server: s, db: db, storage: store, tempDir: tempDir,