Add app and runner metrics and operational observability

.dockerignore

@@ -11,6 +11,7 @@
 !destination/
 !discovery/
 !eth/
+!internal/
 !nodeconfig/
 !orchestrator/
 !pm/

destination/policy.go

@@ -14,6 +14,9 @@ import (
 	"time"
 )
 
+// ErrDeniedAddress identifies a destination blocked by the grants policy.
+var ErrDeniedAddress = errors.New("destination contains a denied address; configure grants for this purpose")
+
 var deniedV4 = mustPrefixes(
 	"0.0.0.0/8", "10.0.0.0/8", "100.64.0.0/10", "127.0.0.0/8",
 	"169.254.0.0/16", "172.16.0.0/12", "192.0.0.0/24", "192.0.2.0/24",
@@ -149,7 +152,7 @@ func (p Policy) DialContext(ctx context.Context, network, address string) (net.C
 	}
 	for _, ip := range ips {
 		if !Public(ip) && !p.granted(net.JoinHostPort(ip.String(), port)) {
-			return nil, fmt.Errorf("%s destination contains a denied address; configure grants for this purpose", p.Purpose)
+			return nil, fmt.Errorf("%s %w", p.Purpose, ErrDeniedAddress)
 		}
 	}
 	dialer := net.Dialer{Timeout: 10 * time.Second, KeepAlive: 30 * time.Second}

discovery/app.go

@@ -86,11 +86,9 @@ func Serve(parent context.Context, p Params, logOut io.Writer) (result error) {
 	}
 	defer metricsListener.Close()
 	metrics := http.NewServeMux()
-	for path, body := range map[string]string{"/healthz": "ok\n", "/readyz": "ready\n", "/metrics": "livepeer_discovery_up 1\n"} {
+	metrics.Handle("GET /metrics", app.metrics.Handler())
+	for path, body := range map[string]string{"/healthz": "ok\n", "/readyz": "ready\n"} {
 		metrics.HandleFunc("GET "+path, func(w http.ResponseWriter, _ *http.Request) {
-			if path == "/metrics" {
-				w.Header().Set("Content-Type", "text/plain; version=0.0.4")
-			}
 			_, _ = io.WriteString(w, body)
 		})
 	}

discovery/app_test.go

@@ -10,6 +10,7 @@ import (
 	"net/url"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
 	"time"
 
@@ -70,7 +71,11 @@ func TestServeWithoutRunnerOrPaymentDependencies(t *testing.T) {
 				require.NoError(t, err)
 				require.NoError(t, response.Body.Close())
 				require.Equal(t, 200, response.StatusCode)
-				require.Equal(t, want, string(body))
+				if strings.HasSuffix(path, "/metrics") {
+					require.Contains(t, string(body), want)
+				} else {
+					require.Equal(t, want, string(body))
+				}
 			}
 		})
 	}

discovery/server.go

@@ -13,13 +13,15 @@ import (
 
 	"github.com/livepeer/node/destination"
 	discovery "github.com/livepeer/node/discovery/shared"
+	"github.com/livepeer/node/internal/observability"
 )
 
 type discoveryNode struct {
 	URL       string    `json:"url"`
 	QueriedAt time.Time `json:"queried_at"`
 }
 type Server struct {
+	metrics                          *observability.Metrics
 	logger                           *slog.Logger
 	mux                              *http.ServeMux
 	urls                             []*url.URL
@@ -71,10 +73,11 @@ func NewServer(p Params, logger *slog.Logger) (*Server, error) {
 	client := policy.Client()
 	client.Timeout = 3 * time.Second
 	client.CheckRedirect = func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }
-	s := &Server{logger: logger, mux: http.NewServeMux(),
+	metrics := observability.New("discovery")
+	s := &Server{metrics: metrics, logger: logger.With("component", "discovery"), mux: http.NewServeMux(),
 		urls: urls, interval: p.RefreshInterval,
 		advertiseNodes: p.AdvertiseNodes, advertiseRunners: p.AdvertiseRunners,
-		collector: &discovery.Collector[discoveryRunner]{Client: client, MaxFailures: max(1, p.MaxFailedRefreshes), Logger: logger}}
+		collector: &discovery.Collector[discoveryRunner]{Client: client, MaxFailures: max(1, p.MaxFailedRefreshes), Logger: logger.With("component", "discovery"), Observe: metrics.Discovery}}
 	endpoints := make([]*url.URL, 0, len(urls))
 	for _, base := range urls {
 		endpoint := base.Clone()
@@ -86,10 +89,18 @@ func NewServer(p Params, logger *slog.Logger) (*Server, error) {
 	}
 	s.collector.SetSources(endpoints)
 	s.mux.HandleFunc("GET /discovery", s.discovery)
+	metrics.Snapshot(s.collector.CollectMetrics)
+	metrics.Legacy(func(emit observability.Emit) { emit("discovery_up", 1, nil) })
 	return s, nil
 }
 
-func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { s.mux.ServeHTTP(w, r) }
+func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	s.metrics.HTTP(w, r, func(w http.ResponseWriter, r *http.Request) {
+		_, pattern := s.mux.Handler(r)
+		observability.Operation(r, pattern)
+		s.mux.ServeHTTP(w, r)
+	})
+}
 
 func (s *Server) Close() { s.collector.Close() }
 

discovery/shared/collector.go

@@ -5,20 +5,25 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"net"
 	"net/http"
 	"net/url"
 	"slices"
 	"strings"
 	"sync"
 	"time"
+
+	"github.com/livepeer/node/destination"
+	"github.com/livepeer/node/internal/observability"
 )
 
 // Snapshot is an immutable successful inventory observation.
 type Snapshot[R Runner] struct {
-	URL       string
-	QueriedAt time.Time
-	Entries   []Entry[R]
-	Failures  int
+	URL         string
+	QueriedAt   time.Time
+	Entries     []Entry[R]
+	Failures    int
+	LastSuccess time.Time
 }
 
 // Collector polls complete discovery endpoints and retains per-source inventories.
@@ -27,6 +32,7 @@ type Collector[R Runner] struct {
 	Client         *http.Client
 	Logger         *slog.Logger
 	MaxFailures    int
+	Observe        func(int, time.Duration, error)
 	mu             sync.Mutex
 	sources        []*Snapshot[R]
 	pending        chan struct{}
@@ -141,7 +147,11 @@ feedLoop:
 		}
 		workers.Go(func() {
 			defer func() { <-slots }()
+			started := time.Now()
 			entries, err := c.fetch(ctx, source.URL)
+			if c.Observe != nil {
+				c.Observe(index, time.Since(started), err)
+			}
 			if ctx.Err() != nil {
 				return
 			}
@@ -150,6 +160,7 @@ feedLoop:
 				if err == nil {
 					source.Entries = entries
 					source.QueriedAt = time.Now().UTC()
+					source.LastSuccess = source.QueriedAt
 					source.Failures = 0
 					c.initialized = true
 				} else {
@@ -166,13 +177,24 @@ feedLoop:
 				if logger == nil {
 					logger = slog.Default()
 				}
-				logger.WarnContext(ctx, "discovery refresh failed", "source_index", index, "error", err)
+				kind := observability.ErrorKind(err)
+				var dns *net.DNSError
+				if errors.Is(err, destination.ErrDeniedAddress) {
+					kind = "denied_destination"
+				} else if errors.As(err, &dns) {
+					kind = "dns"
+				} else if errors.Is(err, errInvalidRedirect) {
+					kind = "invalid_redirect"
+				}
+				logger.WarnContext(ctx, "discovery refresh failed", "event", "discovery_refresh", "operation", "refresh", "outcome", "error", "source_index", index, "error_kind", kind)
 			}
 		})
 	}
 	workers.Wait()
 }
 
+var errInvalidRedirect = errors.New("invalid discovery redirect URL")
+
 func (c *Collector[R]) fetch(ctx context.Context, endpoint string) ([]Entry[R], error) {
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
 	if err != nil {
@@ -186,7 +208,7 @@ func (c *Collector[R]) fetch(ctx context.Context, endpoint string) ([]Entry[R],
 		}
 		// Go includes the raw Location URL in this error, even after unwrapping.
 		if strings.HasPrefix(err.Error(), "failed to parse Location header ") {
-			err = errors.New("invalid discovery redirect URL")
+			err = errInvalidRedirect
 		}
 		return nil, err
 	}

discovery/shared/collector_test.go

@@ -14,6 +14,8 @@ import (
 	"time"
 
 	"github.com/livepeer/node/destination"
+	"github.com/livepeer/node/internal/observability"
+	"github.com/livepeer/node/internal/test"
 	"github.com/stretchr/testify/require"
 )
 
@@ -28,8 +30,30 @@ func TestCollectorCacheLifecycle(t *testing.T) {
 	status, body := 200, envelope(candidate("1999-01-01T00:00:00Z"))
 	c := &Collector[record]{MaxFailures: 6, Client: &http.Client{Transport: collectorTransport(func(*http.Request) (*http.Response, error) { return collectorResponse(status, body), nil })}}
 	c.SetSources([]*url.URL{{Scheme: "https", Host: "source.example"}})
+	m := observability.New("discovery")
+	m.Snapshot(c.CollectMetrics)
+	assertMetrics := func(state string, count, failures int, last time.Time) {
+		t.Helper()
+		require.Len(t, test.MetricSamples(t, m.Registry, "discovery_source_state", nil), 4)
+		for _, candidate := range []string{"uninitialized", "healthy", "retained", "evicted"} {
+			require.Equal(t, observability.Bool(candidate == state), test.MetricSum(t, m.Registry, "discovery_source_state", map[string]string{"state": candidate}))
+		}
+		wantLast := 0.0
+		if !last.IsZero() {
+			wantLast = float64(last.Unix())
+		}
+		require.Equal(t, wantLast, test.MetricSum(t, m.Registry, "discovery_last_success_timestamp_seconds", nil))
+		require.EqualValues(t, count, test.MetricSum(t, m.Registry, "discovery_cached_runners", nil))
+		require.EqualValues(t, failures, test.MetricSum(t, m.Registry, "discovery_source_failures", nil))
+	}
+	assertMetrics("uninitialized", 0, 0, time.Time{})
+	status = 500
+	c.Refresh(t.Context(), false)
+	assertMetrics("uninitialized", 0, 1, time.Time{})
+	status = 200
 	c.Refresh(t.Context(), false)
 	original := c.Snapshots()[0]
+	assertMetrics("healthy", 1, 0, original.LastSuccess)
 	for i, failure := range []struct {
 		status int
 		body   string
@@ -40,22 +64,27 @@ func TestCollectorCacheLifecycle(t *testing.T) {
 		c.Refresh(t.Context(), false)
 		got := c.Snapshots()[0]
 		require.Equal(t, original.Entries, got.Entries)
+		require.Equal(t, original.LastSuccess, got.LastSuccess)
+		assertMetrics("retained", 1, i+1, original.LastSuccess)
 		require.Equal(t, original.QueriedAt, got.QueriedAt)
 		require.Equal(t, i+1, got.Failures)
 	}
 	c.Refresh(t.Context(), false)
 	require.Empty(t, c.Snapshots())
+	assertMetrics("evicted", 0, 6, original.LastSuccess)
 	require.True(t, c.Initialized(), "eviction does not reset startup")
 	status, body = 200, envelope(candidate("1999-02-01T00:00:00Z"))
 	c.Refresh(t.Context(), false)
 	require.Equal(t, "1999-02-01T00:00:00Z", c.Snapshots()[0].Entries[0].Runners[0].Observed)
 	require.Zero(t, c.Snapshots()[0].Failures)
+	assertMetrics("healthy", 1, 0, c.Snapshots()[0].LastSuccess)
 	status = 500
 	c.Refresh(t.Context(), false)
 	status, body = 200, "[]"
 	c.Refresh(t.Context(), false)
 	require.Len(t, c.Snapshots(), 1, "empty success is still a usable source")
 	require.Empty(t, c.Snapshots()[0].Entries)
+	assertMetrics("healthy", 0, 0, c.Snapshots()[0].LastSuccess)
 	require.Zero(t, c.Snapshots()[0].Failures)
 }
 
@@ -136,10 +165,10 @@ func TestCollectorFailureDiagnostics(t *testing.T) {
 		name, endpoint, want string
 		transport            http.RoundTripper
 	}{
-		{"grant", "http://127.0.0.1/private-token", "denied address", policy.Client().Transport},
-		{"DNS", "https://missing.example/private-token", "no such host", policy.Client().Transport},
-		{"timeout", "https://source.example/private-token", "context deadline exceeded", collectorTransport(func(*http.Request) (*http.Response, error) { return nil, context.DeadlineExceeded })},
-		{"redirect", "https://source.example/private-token", "invalid discovery redirect URL", collectorTransport(func(*http.Request) (*http.Response, error) {
+		{"grant", "http://127.0.0.1/private-token", "denied_destination", policy.Client().Transport},
+		{"DNS", "https://missing.example/private-token", "dns", policy.Client().Transport},
+		{"timeout", "https://source.example/private-token", "timeout", collectorTransport(func(*http.Request) (*http.Response, error) { return nil, context.DeadlineExceeded })},
+		{"redirect", "https://source.example/private-token", "invalid_redirect", collectorTransport(func(*http.Request) (*http.Response, error) {
 			response := collectorResponse(302, "")
 			response.Header = http.Header{"Location": {"https://redirect.example/private-token/%zz"}}
 			return response, nil

discovery/shared/metrics.go

@@ -0,0 +1,40 @@
+package shared
+
+import (
+	"strconv"
+
+	"github.com/livepeer/node/internal/observability"
+)
+
+// CollectMetrics includes configured sources that have never succeeded and
+// sources whose inventory has been evicted. Labels never contain source URLs.
+func (c *Collector[R]) CollectMetrics(emit observability.Emit) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	for index, source := range c.sources {
+		labels := observability.Labels{"source_index": strconv.Itoa(index)}
+		state := "healthy"
+		if source.QueriedAt.IsZero() {
+			state = "uninitialized"
+			if !source.LastSuccess.IsZero() {
+				state = "evicted"
+			}
+		} else if source.Failures > 0 {
+			state = "retained"
+		}
+		for _, candidate := range []string{"healthy", "uninitialized", "evicted", "retained"} {
+			emit("discovery_source_state", observability.Bool(state == candidate), labels.With("state", candidate))
+		}
+		emit("discovery_source_failures", float64(source.Failures), labels)
+		last := 0.0
+		if !source.LastSuccess.IsZero() {
+			last = float64(source.LastSuccess.Unix())
+		}
+		emit("discovery_last_success_timestamp_seconds", last, labels)
+		count := 0
+		for _, entry := range source.Entries {
+			count += len(entry.Runners)
+		}
+		emit("discovery_cached_runners", float64(count), labels)
+	}
+}

docs/README.md

@@ -13,6 +13,7 @@ Start with [installation](../README.md#install) and the
 | [Signer](signer.md) | Configure price limits, authorization, discovery, and Kafka accounting. |
 | [Chain commands](chain.md) | Inspect accounts, manage stake and funds, and submit transactions. |
 | [Payment recovery](payment-recovery.md) | Back up state, upgrade or roll back, and recover accounting/redemption work. |
+| [Metrics and events](metrics.md) | Scrape app/runner metrics, investigate payments, and configure operational alerts. |
 
 ## Developing and integrating
 

docs/configuration.md

@@ -157,7 +157,5 @@ curl -fsS http://127.0.0.1:8940/readyz
 ```
 
 All metrics listeners serve `/healthz`, `/readyz`, and `/metrics`.
-Orchestrator and discovery readiness report that their listeners are running;
-they do not check runner health, upstream availability, or payment dependencies.
-Signer readiness checks its payment dependencies and, when enabled, local
-accounting storage. See the component guides for those checks.
+See [metrics and operational events](metrics.md) for readiness requirements,
+scrape configuration, measurements, and query/alert examples.

docs/discovery.md

@@ -51,8 +51,10 @@ omits the metadata entry or runner entries, respectively. Disabling both returns
 signers reject responses larger than 1 MiB.
 
 The loopback metrics listener defaults to `127.0.0.1:8940` and serves `/healthz`,
-`/readyz`, and `/metrics` (`livepeer_discovery_up`). Readiness indicates that the
-service is running, even with no available sources. The cache is not persisted.
+`/readyz`, and `/metrics`, including source health, cached inventory, refresh
+results, and `livepeer_discovery_up`. See [metrics and events](metrics.md).
+Readiness indicates that the service is running, even with no available sources.
+The cache is not persisted.
 
 ## Connect a signer