diff --git a/collector/collector.go b/collector/collector.go index 5890d79d..989f4ecf 100644 --- a/collector/collector.go +++ b/collector/collector.go @@ -386,7 +386,25 @@ func (p *PostgresCollector) Close() error { func execute(ctx context.Context, name string, c Collector, instance *instance, ch chan<- prometheus.Metric, logger *slog.Logger, includeDatname bool) { begin := time.Now() - err := c.Update(ctx, instance, ch) + // Buffer the collector's metrics so that nothing is exposed if Update fails + // part way through, e.g. after some rows were already processed. + buf := make(chan prometheus.Metric) + collected := make(chan []prometheus.Metric) + go func() { + var metrics []prometheus.Metric + for m := range buf { + metrics = append(metrics, m) + } + collected <- metrics + }() + err := c.Update(ctx, instance, buf) + close(buf) + metrics := <-collected + if err == nil { + for _, m := range metrics { + ch <- m + } + } duration := time.Since(begin) var success float64 diff --git a/collector/collector_test.go b/collector/collector_test.go index eafd5d57..22d245ae 100644 --- a/collector/collector_test.go +++ b/collector/collector_test.go @@ -14,6 +14,7 @@ package collector import ( "context" + "errors" "strings" "sync" "testing" @@ -326,3 +327,50 @@ func TestNewPGStatStatementsCollectorUsesConfig(t *testing.T) { t.Fatalf("excludedUsers[0] = %q, want monitor", got.excludedUsers[0]) } } + +type fakeCollector struct { + err error +} + +func (f fakeCollector) Update(ctx context.Context, inst *instance, ch chan<- prometheus.Metric) error { + ch <- prometheus.MustNewConstMetric( + prometheus.NewDesc("fake_metric", "help", nil, nil), prometheus.GaugeValue, 1) + return f.err +} + +func TestExecuteDropsMetricsOnError(t *testing.T) { + tests := []struct { + name string + err error + want []string + }{ + {"success", nil, []string{"fake_metric", "pg_scrape_collector_duration_seconds", "pg_scrape_collector_success"}}, + {"error", errors.New("boom"), []string{"pg_scrape_collector_duration_seconds", "pg_scrape_collector_success"}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ch := make(chan prometheus.Metric, 10) + execute(context.Background(), "fake", fakeCollector{err: tt.err}, nil, ch, promslog.NewNopLogger(), false) + close(ch) + var got []string + success := -1.0 + for m := range ch { + name := m.Desc().String() + for _, w := range []string{"pg_scrape_collector_duration_seconds", "pg_scrape_collector_success", "fake_metric"} { + if strings.Contains(name, `"`+w+`"`) { + got = append(got, w) + if w == "pg_scrape_collector_success" { + success = readMetric(m).value + } + } + } + } + if strings.Join(got, ",") != strings.Join(tt.want, ",") { + t.Fatalf("got metrics %v, want %v", got, tt.want) + } + if (tt.err == nil) != (success == 1) { + t.Fatalf("unexpected success value %v", success) + } + }) + } +}