Explain why prometheus-exporter marks an endpoint unavailable

docs/release-notes/change-log.md

@@ -98,6 +98,49 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
 - `substreams-tier1` now names the usage marker it writes in every module cache folder after the request's plan tier: `last_used_<plan>` (lowercase, e.g. `last_used_pro`), still plain `last_used` when unauthenticated. `firecore tools substreams purge` reads the plan back from that name to apply a retention per plan.
 
+### Tools
+
+- `substreams tools prometheus-exporter` now says *why* an endpoint is down. Every failure is classified into a
+  `reason` -- `invalid_config`, `connect_failed`, `connect_timeout`, `invalid_request`, `request_timeout`,
+  `stream_error`, `stale_block`, `invalid_response` or `no_data` -- exposed on the new
+  `substreams_healthcheck_failure_count{reason,grpc_code}` counter and included in the logs. An alert firing on
+  `substreams_healthcheck_status` no longer requires guessing whether the endpoint was unreachable, unauthenticated,
+  overloaded or merely late. A dial that fails outright is reported as `connect_failed`, carrying the dial
+  error where gRPC exposes it (`connection refused`); a hostname that does not resolve surfaces as the balancer's
+  own `no children to pick from`, since it replaces the resolver error. `connect_timeout` is reserved for a
+  connection that is merely slow to come up and never failed a dial.
+
+- Connection establishment gets its own budget, `--connect-timeout` (default 10s), separate from `--timeout`, which
+  now covers the `Blocks` request alone. gRPC dials lazily, so DNS, TLS and load-balancer resolution used to be
+  charged to the request timeout and a slow connection was reported as an endpoint failure -- this is what produced
+  the `received context error while waiting for new LB policy update: context deadline exceeded` errors. The exporter
+  now waits for the channel to be `READY` before issuing the request, and reports the two phases separately as
+  `substreams_healthcheck_connect_duration_ms` and `substreams_healthcheck_stream_duration_ms`.
+  `substreams_healthcheck_duration_ms` keeps its previous meaning of the two combined.
+
+- Every failed poll is logged, not just the transition into `unavailable`. An endpoint that fails repeatedly, or one
+  that flaps between two Prometheus scrapes, previously produced a single line and then nothing. Failure logs carry
+  the reason, the gRPC code, both durations and the consecutive failure count; the recovery log carries how long the
+  endpoint was down and how many polls failed meanwhile. A block age crossing half of `--max-freshness` is reported
+  too, so an alert on `substreams_healthcheck_block_age_ms` is no longer silent. That one is edge-triggered and only
+  after three consecutive polls agree, so a chain whose block interval straddles the threshold stays quiet.
+
+- New `substreams_healthcheck_consecutive_failures` gauge, meant to be alerted on instead of
+  `substreams_healthcheck_status` when single-poll hiccups should be ignored.
+
+- `substreams_healthcheck_block_age_ms` is reset to `NaN` when a poll returns no block, instead of keeping the age of
+  the last block ever seen -- which silently under-reported staleness for as long as an endpoint stayed broken.
+
+- Fixed: endpoints configured with different sets of query-parameter labels (e.g. one with `?namespace=x&region=y`
+  and one with only `?namespace=z`) made the exporter panic on inconsistent label cardinality. Missing labels are now
+  filled with an empty value.
+
+- **Breaking** The exporter now speaks `sf.substreams.rpc.v4.Stream/Blocks` only. The v3-to-v2 fallback is gone --
+  it closed the connection and then kept reading from it, double-counting the failure -- and
+  `--force-protocol-version` accepts only `4` (or `0`), the flag being kept for the protocol versions to come. An
+  invalid value used to be parsed and then silently ignored, it is now rejected at startup, so an invocation passing
+  `--force-protocol-version 2` or `3` must drop the flag.
+
 ### Dependencies
 
 - `google.golang.org/grpc` is at v1.83.1, which clears GHSA-vp52-pcj8-j9qc, reported as HIGH: a peer could exhaust

tools/log_test.go

@@ -0,0 +1,9 @@
+package tools
+
+import (
+	"github.com/streamingfast/logging"
+)
+
+func init() {
+	logging.InstantiateLoggers()
+}

tools/metrics.go

@@ -0,0 +1,141 @@
+package tools
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"slices"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/streamingfast/dgrpc"
+	"github.com/streamingfast/dmetrics"
+	"google.golang.org/grpc/codes"
+)
+
+// This file holds the metrics of the `prometheus-exporter` command: the failure taxonomy
+// that gives them their labels, their declaration and the label plumbing. The polling loop
+// that feeds them lives in `prometheus-exporter.go`.
+
+// failureReason categorizes *where* a poll failed, so that an alert firing on
+// `substreams_healthcheck_status == 0` can be traced back to a cause without
+// having to correlate it with the logs.
+type failureReason string
+
+const (
+	reasonInvalidConfig   failureReason = "invalid_config"
+	reasonConnectFailed   failureReason = "connect_failed"
+	reasonConnectTimeout  failureReason = "connect_timeout"
+	reasonInvalidRequest  failureReason = "invalid_request"
+	reasonRequestTimeout  failureReason = "request_timeout"
+	reasonStreamError     failureReason = "stream_error"
+	reasonStaleBlock      failureReason = "stale_block"
+	reasonInvalidResponse failureReason = "invalid_response"
+	reasonNoData          failureReason = "no_data"
+)
+
+// noGRPCCode is the value of the `grpc_code` label for failures that did not carry a gRPC status.
+const noGRPCCode = "none"
+
+const (
+	// endpointLabel is the one label every endpoint always carries, the others come from the
+	// query parameters of the endpoint specification.
+	endpointLabel = "endpoint"
+	// reasonLabel and grpcCodeLabel are carried by the failure counter only.
+	reasonLabel   = "reason"
+	grpcCodeLabel = "grpc_code"
+)
+
+var healthcheckMetrics = dmetrics.NewSet(dmetrics.PrefixNameWith("substreams_healthcheck"))
+
+var (
+	status              *dmetrics.GaugeVec
+	requestDurationMs   *dmetrics.GaugeVec
+	connectDurationMs   *dmetrics.GaugeVec
+	streamDurationMs    *dmetrics.GaugeVec
+	blockAgeMs          *dmetrics.GaugeVec
+	consecutiveFailures *dmetrics.GaugeVec
+	failureCount        *dmetrics.CounterVec
+)
+
+// initHealthcheckMetrics declares every healthcheck metric over the given label names, which
+// are the union of the labels across all the polled endpoints, and returns the collectors so
+// that the caller can register them on its own registry.
+func initHealthcheckMetrics(labelNames []string) []prometheus.Collector {
+	failureLabelNames := append(slices.Clone(labelNames), reasonLabel, grpcCodeLabel)
+
+	status = healthcheckMetrics.NewGaugeVec("status", labelNames, "Either 1 for successful subtreams request, or 0 for failure")
+	requestDurationMs = healthcheckMetrics.NewGaugeVec("duration_ms", labelNames, "Request full processing time in millisecond, connection establishment included")
+	connectDurationMs = healthcheckMetrics.NewGaugeVec("connect_duration_ms", labelNames, "Time spent establishing the gRPC connection in millisecond, this excludes the Blocks request itself")
+	streamDurationMs = healthcheckMetrics.NewGaugeVec("stream_duration_ms", labelNames, "Time spent on the Blocks request in millisecond, from an established connection to the first block")
+	blockAgeMs = healthcheckMetrics.NewGaugeVec("block_age_ms", labelNames, "Age of returned block, NaN when the last poll did not return a block")
+	consecutiveFailures = healthcheckMetrics.NewGaugeVec("consecutive_failures", labelNames, "Number of consecutive failed polls, 0 when the last poll succeeded. Alert on this instead of 'status' to ignore single-poll hiccups")
+	failureCount = healthcheckMetrics.NewCounterVec("failure_count", failureLabelNames, "Number of failed polls, broken down by 'reason' and by gRPC status code")
+
+	return []prometheus.Collector{status, requestDurationMs, connectDurationMs, streamDurationMs, blockAgeMs, consecutiveFailures, failureCount}
+}
+
+// endpointLabelValues orders an endpoint's parameters along labelNames, filling in an empty
+// value for the labels this endpoint was not given.
+func endpointLabelValues(url string, params map[string]string, labelNames []string) []string {
+	values := make([]string, len(labelNames))
+	for i, name := range labelNames {
+		if name == endpointLabel {
+			values[i] = url
+			continue
+		}
+		values[i] = params[name]
+	}
+	return values
+}
+
+// failureLabelValues returns the endpoint label values augmented with the failure
+// classification, used by the `substreams_healthcheck_failure_count` counter only.
+func failureLabelValues(endpoint string, reason failureReason, grpcCode string) []string {
+	return append(slices.Clone(endpointMap[endpoint].labelValues), string(reason), grpcCode)
+}
+
+// grpcCodeOf returns the gRPC status code carried by err, or `noGRPCCode` when there is none.
+func grpcCodeOf(err error) string {
+	if grpcError := dgrpc.AsGRPCError(err); grpcError != nil {
+		return grpcError.Code().String()
+	}
+	return noGRPCCode
+}
+
+// streamFailure attributes an error returned by the Blocks call. A channel that never became
+// ready makes gRPC answer with the dial error it was holding, so the failure belongs to the
+// connection rather than to the endpoint's own answer, and `connect_failed` says so.
+func streamFailure(ctx context.Context, connectFailed bool, err error) (failureReason, error) {
+	err = withDeadlineCause(ctx, err)
+	if connectFailed {
+		return reasonConnectFailed, err
+	}
+	return classifyStreamError(err), err
+}
+
+// withDeadlineCause appends the reason ctx was cut short. gRPC answers a request that
+// outlived its deadline with its own "context deadline exceeded" status and drops the cause
+// attached to the context, so without this the error never names which budget expired.
+func withDeadlineCause(ctx context.Context, err error) error {
+	if cause := context.Cause(ctx); cause != nil {
+		return fmt.Errorf("%w: %s", err, cause)
+	}
+	return err
+}
+
+// classifyStreamError maps an error returned while talking to the endpoint onto a
+// failureReason, distinguishing a timeout from an outright refusal, and an empty
+// stream from a stream that errored out.
+func classifyStreamError(err error) failureReason {
+	if errors.Is(err, io.EOF) {
+		return reasonNoData
+	}
+	if errors.Is(err, context.DeadlineExceeded) {
+		return reasonRequestTimeout
+	}
+	if grpcError := dgrpc.AsGRPCError(err); grpcError != nil && grpcError.Code() == codes.DeadlineExceeded {
+		return reasonRequestTimeout
+	}
+	return reasonStreamError
+}

tools/metrics_test.go

@@ -0,0 +1,165 @@
+package tools
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"maps"
+	"slices"
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
+	grpcstatus "google.golang.org/grpc/status"
+)
+
+func TestClassifyStreamError(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+		want failureReason
+	}{
+		{
+			name: "stream completed without any block",
+			err:  io.EOF,
+			want: reasonNoData,
+		},
+		{
+			name: "wrapped end of stream",
+			err:  fmt.Errorf("receiving message: %w", io.EOF),
+			want: reasonNoData,
+		},
+		{
+			name: "local context deadline",
+			err:  context.DeadlineExceeded,
+			want: reasonRequestTimeout,
+		},
+		{
+			name: "grpc deadline exceeded",
+			err:  grpcstatus.Error(codes.DeadlineExceeded, "context deadline exceeded"),
+			want: reasonRequestTimeout,
+		},
+		{
+			name: "endpoint unavailable",
+			err:  grpcstatus.Error(codes.Unavailable, "no healthy upstream"),
+			want: reasonStreamError,
+		},
+		{
+			name: "endpoint refused our credentials",
+			err:  grpcstatus.Error(codes.Unauthenticated, "invalid token"),
+			want: reasonStreamError,
+		},
+		{
+			name: "plain error",
+			err:  errors.New("boom"),
+			want: reasonStreamError,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, classifyStreamError(tt.err))
+		})
+	}
+}
+
+func TestGRPCCodeOf(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+		want string
+	}{
+		{"no error", nil, noGRPCCode},
+		{"not a grpc error", errors.New("boom"), noGRPCCode},
+		{"grpc error", grpcstatus.Error(codes.ResourceExhausted, "overloaded"), "ResourceExhausted"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, grpcCodeOf(tt.err))
+		})
+	}
+}
+
+// Endpoints configured with different query parameters must all produce the same number of
+// label values, otherwise the prometheus `WithLabelValues` calls panic on inconsistent cardinality.
+func TestEndpointLabelValues(t *testing.T) {
+	endpoints := []string{
+		"a.domain:443?namespace=eth-mainnet",
+		"b.domain:443?namespace=sol-mainnet&region=us-east",
+		"c.domain:443",
+	}
+
+	allLabels := map[string]bool{endpointLabel: true}
+	params := map[string]map[string]string{}
+
+	for _, endpoint := range endpoints {
+		url, _, endpointParams, err := parseEndpoint(endpoint)
+		require.NoError(t, err)
+
+		for k := range endpointParams {
+			allLabels[k] = true
+		}
+		params[url] = endpointParams
+	}
+
+	labelNames := slices.Sorted(maps.Keys(allLabels))
+	require.Equal(t, []string{"endpoint", "namespace", "region"}, labelNames)
+
+	values := map[string][]string{}
+	for url, endpointParams := range params {
+		values[url] = endpointLabelValues(url, endpointParams, labelNames)
+	}
+
+	assert.Equal(t, []string{"a.domain:443", "eth-mainnet", ""}, values["a.domain:443"])
+	assert.Equal(t, []string{"b.domain:443", "sol-mainnet", "us-east"}, values["b.domain:443"])
+	assert.Equal(t, []string{"c.domain:443", "", ""}, values["c.domain:443"])
+
+	gauge := prometheus.NewGaugeVec(prometheus.GaugeOpts{Name: "test_status"}, labelNames)
+	for _, labelValues := range values {
+		assert.NotPanics(t, func() { gauge.WithLabelValues(labelValues...).Set(1) })
+	}
+}
+
+func TestWithDeadlineCause(t *testing.T) {
+	streamErr := grpcstatus.Error(codes.DeadlineExceeded, "context deadline exceeded")
+
+	t.Run("live context leaves the error alone", func(t *testing.T) {
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		assert.Equal(t, streamErr, withDeadlineCause(ctx, streamErr))
+	})
+
+	t.Run("expired context names the budget that ran out", func(t *testing.T) {
+		ctx, cancel := context.WithTimeoutCause(context.Background(), time.Nanosecond, errors.New("request timeout of 15s reached"))
+		defer cancel()
+		<-ctx.Done()
+
+		err := withDeadlineCause(ctx, streamErr)
+		assert.ErrorContains(t, err, "request timeout of 15s reached")
+		assert.ErrorIs(t, err, streamErr)
+	})
+}
+
+func TestStreamFailure(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// A channel that never became ready answers through the request, so the gRPC error carries
+	// the dial message and belongs to the connection, not to the endpoint's own answer.
+	refused := grpcstatus.Error(codes.Unavailable, `connection error: desc = "transport: Error while dialing: dial tcp 127.0.0.1:19999: connect: connection refused"`)
+	reason, err := streamFailure(ctx, true, refused)
+	assert.Equal(t, reasonConnectFailed, reason)
+	assert.ErrorContains(t, err, "connection refused")
+
+	reason, _ = streamFailure(ctx, false, refused)
+	assert.Equal(t, reasonStreamError, reason)
+
+	reason, _ = streamFailure(ctx, false, grpcstatus.Error(codes.DeadlineExceeded, "context deadline exceeded"))
+	assert.Equal(t, reasonRequestTimeout, reason)
+}