From 649011efc7b2ac25afc2860ad9ede6efab9b6b95 Mon Sep 17 00:00:00 2001 From: Silvano Ravotto Date: Mon, 13 Jul 2026 11:11:32 -0400 Subject: [PATCH] http: stop writing errors back to disconnected metrics clients The metrics handler called errorResponse after streaming of the response had already begun. Once a scrape client (for example Prometheus) closes the connection mid-response, the HTTP/2 server returns "client disconnected" on the next write, so trying to send an error status and body back to that client is futile and just fails again. The write loop also lacked a return, so a single disconnect logged one error line per remaining metric family, making one benign disconnect look like many failures. Stop calling errorResponse from the streaming path. On a write failure to the client, log at debug (a disconnect is an expected event) and return after the first error. The registry.Gather() failure path still uses errorResponse, since it fails before any body is written and a 500 there is legitimate. While here, handle two previously ignored errors on the --prometheus proxy path: metricsWriter.Copy's return value in the handler (logged at error level, since a Copy failure usually means the upstream source is unreachable or returned malformed metrics) and translator.Translate's return value inside Writer.Copy. Extract the handler into metricsHandler so it can be unit tested, and add tests for a normal scrape, a client disconnect, and a Copy failure. Co-Authored-By: roachdev-claude --- internal/http/server.go | 54 ++++++++++++------ internal/http/server_test.go | 103 +++++++++++++++++++++++++++++++++++ internal/metric/writer.go | 4 +- 3 files changed, 142 insertions(+), 19 deletions(-) diff --git a/internal/http/server.go b/internal/http/server.go index 831cd3d..eb81b2f 100644 --- a/internal/http/server.go +++ b/internal/http/server.go @@ -129,24 +129,7 @@ func (s *serverImpl) Start(ctx *stopper.Context) error { if err != nil { return err } - handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - defer r.Body.Close() - if s.metricsWriter != nil { - s.metricsWriter.Copy(ctx, w) - } - metrics, err := s.registry.Gather() - if err != nil { - s.errorResponse(w, "Error gathering metrics", err) - return - } - for _, m := range metrics { - _, err = expfmt.MetricFamilyToText(w, m) - if err != nil { - s.errorResponse(w, "Error gathering metrics", err) - } - } - - }) + handler := http.HandlerFunc(s.metricsHandler(ctx)) http.Handle(s.config.Endpoint, gziphandler.GzipHandler(handler)) s.debugInfo() @@ -180,6 +163,41 @@ func (s *serverImpl) Start(ctx *stopper.Context) error { return nil } +// metricsHandler returns the handler that serves the metrics endpoint. It first +// copies the metrics fetched from the upstream source (applying any histogram +// translators) and then appends the metrics gathered from the local registry. +func (s *serverImpl) metricsHandler(ctx *stopper.Context) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + defer r.Body.Close() + if s.metricsWriter != nil { + // A Copy failure is usually the upstream source being unreachable + // or returning malformed metrics, which is actionable, so it is + // logged at error level. It can also be the client disconnecting + // mid-response, in which case the stream is already broken and we + // cannot report anything back either way. + if err := s.metricsWriter.Copy(ctx, w); err != nil { + log.Errorf("Error copying metrics from source: %s", err.Error()) + return + } + } + metrics, err := s.registry.Gather() + if err != nil { + s.errorResponse(w, "Error gathering metrics", err) + return + } + for _, m := range metrics { + // A write failure here means the client has disconnected, which is + // an expected event (for example a scrape timeout). We stop writing + // and log at debug level rather than trying to send an error to a + // client that is no longer listening. + if _, err := expfmt.MetricFamilyToText(w, m); err != nil { + log.Debugf("Stopped writing metrics to client: %s", err.Error()) + return + } + } + } +} + func (s *serverImpl) errorResponse(w http.ResponseWriter, msg string, err error) { log.Errorf("%s: %s", msg, err.Error()) w.WriteHeader(http.StatusInternalServerError) diff --git a/internal/http/server_test.go b/internal/http/server_test.go index 18e970d..474721c 100644 --- a/internal/http/server_test.go +++ b/internal/http/server_test.go @@ -19,6 +19,8 @@ import ( "context" "crypto/tls" "crypto/x509" + "net/http" + "net/http/httptest" _ "net/http/pprof" "os" "testing" @@ -29,10 +31,40 @@ import ( "github.com/cockroachlabs/visus/internal/metric" "github.com/cockroachlabs/visus/internal/server" "github.com/cockroachlabs/visus/internal/store" + "github.com/prometheus/client_golang/prometheus" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) +// disconnectedResponseWriter simulates a client that has disconnected: every +// write fails, mimicking the "client disconnected" error returned by the HTTP/2 +// server once the peer is gone. It records whether WriteHeader was called so +// tests can assert the handler does not try to send an error status back to a +// client that is no longer listening. +type disconnectedResponseWriter struct { + header http.Header + wroteHeader bool + statusCode int +} + +var _ http.ResponseWriter = &disconnectedResponseWriter{} + +func (d *disconnectedResponseWriter) Header() http.Header { + if d.header == nil { + d.header = http.Header{} + } + return d.header +} + +func (d *disconnectedResponseWriter) Write([]byte) (int, error) { + return 0, errors.New("client disconnected") +} + +func (d *disconnectedResponseWriter) WriteHeader(statusCode int) { + d.wroteHeader = true + d.statusCode = statusCode +} + // TestRefreshHistograms verifies we reload the histograms from the store. func TestRefreshHistograms(t *testing.T) { r := require.New(t) @@ -198,3 +230,74 @@ func TestRefreshTLSConfig(t *testing.T) { r.NoError(err) a.Equal(expectedCert.Certificate, cert.Certificate) } + +// newTestRegistry returns a registry with a single registered gauge so the +// metrics handler has something to gather. +func newTestRegistry(t *testing.T) *prometheus.Registry { + t.Helper() + registry := prometheus.NewRegistry() + gauge := prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "visus_test_gauge", + Help: "gauge used by the metrics handler tests", + }) + gauge.Set(1) + require.NoError(t, registry.Register(gauge)) + return registry +} + +// TestMetricsHandler verifies that a normal scrape returns the gathered metrics. +func TestMetricsHandler(t *testing.T) { + a := assert.New(t) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + stop := stopper.WithContext(ctx) + server := &serverImpl{ + registry: newTestRegistry(t), + } + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/metrics", nil) + server.metricsHandler(stop)(rec, req) + a.Equal(http.StatusOK, rec.Code) + a.Contains(rec.Body.String(), "visus_test_gauge") +} + +// TestMetricsHandlerClientDisconnect verifies that when the client disconnects +// mid-response the handler stops writing and does not try to send an error +// status back to the client that is no longer listening. +func TestMetricsHandlerClientDisconnect(t *testing.T) { + a := assert.New(t) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + stop := stopper.WithContext(ctx) + server := &serverImpl{ + registry: newTestRegistry(t), + } + w := &disconnectedResponseWriter{} + req := httptest.NewRequest(http.MethodGet, "/metrics", nil) + // The handler must not panic and must not attempt to write an error status + // back to the disconnected client. + server.metricsHandler(stop)(w, req) + a.False(w.wroteHeader, "handler should not write a status back to a disconnected client") +} + +// TestMetricsHandlerCopyError verifies that when copying from the upstream +// source fails the handler returns early without writing gathered metrics. +func TestMetricsHandlerCopyError(t *testing.T) { + r := require.New(t) + a := assert.New(t) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + stop := stopper.WithContext(ctx) + // A file source that does not exist makes Copy fail on open. + writer, err := metric.NewWriter("file:///nonexistent-source.txt", nil, nil) + r.NoError(err) + server := &serverImpl{ + registry: newTestRegistry(t), + metricsWriter: writer, + } + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/metrics", nil) + server.metricsHandler(stop)(rec, req) + // Copy failed, so we return before gathering registry metrics. + a.NotContains(rec.Body.String(), "visus_test_gauge") +} diff --git a/internal/metric/writer.go b/internal/metric/writer.go index a2b2aaf..f4b3ace 100644 --- a/internal/metric/writer.go +++ b/internal/metric/writer.go @@ -115,7 +115,9 @@ func (w *Writer) Copy(ctx context.Context, out io.Writer) error { for _, mf := range metricFamilies { if mf.GetType() == dto.MetricType_HISTOGRAM && translators != nil { for _, translator := range translators { - translator.Translate(ctx, mf, out) + if err := translator.Translate(ctx, mf, out); err != nil { + return err + } } } else { _, err := expfmt.MetricFamilyToText(out, mf)