Expose collector WAL status via diagnostics and metrics

The collector data source implements source.WALStatusProvider and
registers a "Collector WAL" diagnostic that fails while writes are failing
or when the startup restore was incomplete.

Add unlabelled opencost_wal_* metrics, registered once alongside the other
cost model metrics when the data source provides a WAL status. They respect
the disabled metrics config, and an existing registration (for example by
an application embedding OpenCost) is tolerated rather than panicking.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Signed-off-by: peatey <warwick@automatic.systems>

modules/collector-source/pkg/collector/datasource.go

@@ -2,6 +2,8 @@ package collector
 
 import (
 	"context"
+	"fmt"
+	"strings"
 	"time"
 
 	"github.com/julienschmidt/httprouter"
@@ -25,6 +27,7 @@ type collectorDataSource struct {
 	clusterInfo       clusters.ClusterInfoProvider
 	config            CollectorConfig
 	diagnosticsModule *metric.DiagnosticsModule
+	wal               *metric.Walinator
 }
 
 func NewDefaultCollectorDataSource(
@@ -70,6 +73,7 @@ func NewCollectorDataSource(
 	)
 	var updater metric.Updater
 	updater = repo
+	var walinator *metric.Walinator
 	if store != nil {
 		wal, err := metric.NewWalinator(
 			config.ClusterName,
@@ -83,6 +87,7 @@ func NewCollectorDataSource(
 		} else {
 			wal.Start()
 			updater = wal
+			walinator = wal
 		}
 	}
 
@@ -121,6 +126,7 @@ func NewCollectorDataSource(
 		clusterInfo:       clusterInfo,
 		clusterMap:        clusterMap,
 		diagnosticsModule: diagnosticsModule,
+		wal:               walinator,
 	}
 }
 
@@ -145,6 +151,60 @@ func (c *collectorDataSource) RegisterDiagnostics(diagService diagnostics.Diagno
 			log.Warnf("Failed to register collector diagnostic %s: %s", dd.ID, err.Error())
 		}
 	}
+
+	if c.wal != nil {
+		err := diagService.Register(WALDiagnosticName, WALDiagnosticDescription, CollectorDiagnosticCategory, func(ctx context.Context) (map[string]any, error) {
+			return walDiagnosticDetails(c.wal.Status())
+		})
+		if err != nil {
+			log.Warnf("Failed to register collector diagnostic %s: %s", WALDiagnosticName, err.Error())
+		}
+	}
+}
+
+const (
+	WALDiagnosticName        = "Collector WAL"
+	WALDiagnosticDescription = "Collector write-ahead log is persisting scrapes to storage and was fully restored at startup."
+)
+
+// walDiagnosticDetails converts a WAL status into diagnostic details, returning an error describing
+// the failure when writes are currently failing or the startup restore was incomplete.
+func walDiagnosticDetails(status source.WALStatus) (map[string]any, error) {
+	var problems []string
+	if status.ConsecutiveExportFailures > 0 {
+		problems = append(problems, fmt.Sprintf("%d consecutive write failures since %s (last error: %s)",
+			status.ConsecutiveExportFailures, status.LastExportSuccess.Format(time.RFC3339), status.LastExportError))
+	}
+	if status.RestoreListError != "" {
+		problems = append(problems, fmt.Sprintf("restore could not list objects: %s", status.RestoreListError))
+	}
+	if status.RestoreErrors > 0 {
+		problems = append(problems, fmt.Sprintf("restore failed to read %d of %d objects", status.RestoreErrors, status.RestoreObjectsSeen))
+	}
+	if len(problems) > 0 {
+		return nil, fmt.Errorf("%s", strings.Join(problems, "; "))
+	}
+
+	return map[string]any{
+		"lastExportSuccess":      status.LastExportSuccess,
+		"exportFailuresTotal":    status.ExportFailuresTotal,
+		"restoreCompleted":       status.RestoreCompleted,
+		"restoreObjectsApplied":  status.RestoreObjectsApplied,
+		"restoreDuration":        status.RestoreDuration.String(),
+		"restoreOldest":          status.RestoreOldest,
+		"restoreNewest":          status.RestoreNewest,
+		"restoreLargestGap":      status.RestoreLargestGap.String(),
+		"restoreLargestGapStart": status.RestoreLargestGapStart,
+	}, nil
+}
+
+// WALStatus implements source.WALStatusProvider, reporting the export and restore health of the
+// collector's write-ahead log.
+func (c *collectorDataSource) WALStatus() source.WALStatus {
+	if c.wal == nil {
+		return source.WALStatus{}
+	}
+	return c.wal.Status()
 }
 
 func (c *collectorDataSource) Metrics() source.MetricsQuerier {

modules/collector-source/pkg/collector/datasource_wal_test.go

@@ -0,0 +1,44 @@
+package collector
+
+import (
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/opencost/opencost/core/pkg/source"
+)
+
+var _ source.WALStatusProvider = (*collectorDataSource)(nil)
+
+func TestWALDiagnosticDetails(t *testing.T) {
+	healthy := source.WALStatus{Enabled: true, LastExportSuccess: time.Now(), RestoreCompleted: true, RestoreObjectsSeen: 3, RestoreObjectsApplied: 3}
+	details, err := walDiagnosticDetails(healthy)
+	if err != nil {
+		t.Fatalf("expected healthy wal to pass, got %s", err)
+	}
+	if details["restoreObjectsApplied"] != 3 {
+		t.Errorf("unexpected details: %v", details)
+	}
+
+	failing := healthy
+	failing.ConsecutiveExportFailures = 4
+	failing.LastExportError = "503"
+	failing.RestoreErrors = 1
+	failing.RestoreListError = "denied"
+	_, err = walDiagnosticDetails(failing)
+	if err == nil {
+		t.Fatalf("expected failing wal to report an error")
+	}
+	for _, want := range []string{"4 consecutive write failures", "could not list", "failed to read 1 of 3"} {
+		if !strings.Contains(err.Error(), want) {
+			t.Errorf("expected %q in %q", want, err)
+		}
+	}
+}
+
+func TestCollectorDataSource_WALStatusWithoutWAL(t *testing.T) {
+	c := &collectorDataSource{}
+	if c.WALStatus().Enabled {
+		t.Errorf("expected disabled status without a wal")
+	}
+}

pkg/costmodel/metrics.go

@@ -372,6 +372,10 @@ func NewCostModelMetricsEmitter(clusterCache clustercache.ClusterCache, provider
 
 	metrics.InitOpencostTelemetry(metricsConfig)
 
+	if model != nil {
+		metrics.InitWALMetrics(model.DataSource, metricsConfig)
+	}
+
 	return &CostModelMetricsEmitter{
 		KubeClusterCache:                 clusterCache,
 		CloudProvider:                    provider,

pkg/metrics/walmetrics.go

@@ -0,0 +1,133 @@
+package metrics
+
+import (
+	"errors"
+	"sync"
+
+	"github.com/opencost/opencost/core/pkg/log"
+	"github.com/opencost/opencost/core/pkg/source"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+var walMetricsInit sync.Once
+
+// walMetric describes a single WAL status metric and how to derive its value from a WALStatus.
+type walMetric struct {
+	name      string
+	desc      *prometheus.Desc
+	valueType prometheus.ValueType
+	value     func(source.WALStatus) float64
+}
+
+func newWALMetric(name, help string, valueType prometheus.ValueType, value func(source.WALStatus) float64) walMetric {
+	return walMetric{
+		name:      name,
+		desc:      prometheus.NewDesc(name, help, nil, nil),
+		valueType: valueType,
+		value:     value,
+	}
+}
+
+var walMetrics = []walMetric{
+	newWALMetric("opencost_wal_last_export_success_timestamp_seconds",
+		"Unix time of the most recent successful collector WAL write, 0 if none.", prometheus.GaugeValue,
+		func(s source.WALStatus) float64 {
+			return unixOrZero(s.LastExportSuccess.Unix(), s.LastExportSuccess.IsZero())
+		}),
+	newWALMetric("opencost_wal_export_failures_total",
+		"Total failed collector WAL writes since start.", prometheus.CounterValue,
+		func(s source.WALStatus) float64 { return float64(s.ExportFailuresTotal) }),
+	newWALMetric("opencost_wal_consecutive_export_failures",
+		"Collector WAL writes that have failed since the last successful write.", prometheus.GaugeValue,
+		func(s source.WALStatus) float64 { return float64(s.ConsecutiveExportFailures) }),
+	newWALMetric("opencost_wal_restore_completed",
+		"1 once the collector WAL startup restore has finished, 0 before.", prometheus.GaugeValue,
+		func(s source.WALStatus) float64 { return boolToFloat(s.RestoreCompleted) }),
+	newWALMetric("opencost_wal_restore_list_failed",
+		"1 if the collector WAL startup restore could not list stored objects, meaning nothing was restored.", prometheus.GaugeValue,
+		func(s source.WALStatus) float64 { return boolToFloat(s.RestoreListError != "") }),
+	newWALMetric("opencost_wal_restore_objects_applied",
+		"Collector WAL objects applied during the startup restore.", prometheus.GaugeValue,
+		func(s source.WALStatus) float64 { return float64(s.RestoreObjectsApplied) }),
+	newWALMetric("opencost_wal_restore_errors_total",
+		"Collector WAL objects that could not be read or decoded during the startup restore.", prometheus.CounterValue,
+		func(s source.WALStatus) float64 { return float64(s.RestoreErrors) }),
+	newWALMetric("opencost_wal_restore_duration_seconds",
+		"Duration of the collector WAL startup restore.", prometheus.GaugeValue,
+		func(s source.WALStatus) float64 { return s.RestoreDuration.Seconds() }),
+	newWALMetric("opencost_wal_restore_largest_gap_seconds",
+		"Largest interval between consecutive restored collector WAL objects. Values well above the scrape interval indicate history that was never persisted or could not be restored.", prometheus.GaugeValue,
+		func(s source.WALStatus) float64 { return s.RestoreLargestGap.Seconds() }),
+}
+
+// WALStatusCollector is a prometheus collector that reports the status of a data source's write-ahead log.
+type WALStatusCollector struct {
+	provider      source.WALStatusProvider
+	metricsConfig MetricsConfig
+}
+
+// Describe sends the descriptors of all enabled WAL metrics.
+func (wc WALStatusCollector) Describe(ch chan<- *prometheus.Desc) {
+	disabled := wc.metricsConfig.GetDisabledMetricsMap()
+	for _, m := range walMetrics {
+		if _, ok := disabled[m.name]; !ok {
+			ch <- m.desc
+		}
+	}
+}
+
+// Collect reads the current WAL status and emits all enabled WAL metrics. Nothing is emitted when the
+// data source has no WAL configured.
+func (wc WALStatusCollector) Collect(ch chan<- prometheus.Metric) {
+	status := wc.provider.WALStatus()
+	if !status.Enabled {
+		return
+	}
+
+	disabled := wc.metricsConfig.GetDisabledMetricsMap()
+	for _, m := range walMetrics {
+		if _, ok := disabled[m.name]; ok {
+			continue
+		}
+		ch <- prometheus.MustNewConstMetric(m.desc, m.valueType, m.value(status))
+	}
+}
+
+// InitWALMetrics registers WAL status metrics for the data source if it implements
+// source.WALStatusProvider. Registration happens at most once per process, and an existing
+// registration of the same metrics (for example by an embedding application) is left in place.
+func InitWALMetrics(dataSource source.OpenCostDataSource, metricsConfig *MetricsConfig) {
+	provider, ok := dataSource.(source.WALStatusProvider)
+	if !ok || metricsConfig == nil {
+		return
+	}
+
+	walMetricsInit.Do(func() {
+		err := prometheus.Register(WALStatusCollector{
+			provider:      provider,
+			metricsConfig: *metricsConfig,
+		})
+		if err != nil {
+			var already prometheus.AlreadyRegisteredError
+			if errors.As(err, &already) {
+				log.Debugf("WAL status metrics already registered")
+				return
+			}
+			log.Warnf("Failed to register WAL status metrics: %s", err)
+		}
+	})
+}
+
+func unixOrZero(unix int64, zero bool) float64 {
+	if zero {
+		return 0
+	}
+	return float64(unix)
+}
+
+func boolToFloat(b bool) float64 {
+	if b {
+		return 1
+	}
+	return 0
+}

pkg/metrics/walmetrics_test.go

@@ -0,0 +1,89 @@
+package metrics
+
+import (
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/opencost/opencost/core/pkg/source"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+)
+
+type fakeWALStatusProvider struct {
+	status source.WALStatus
+}
+
+func (f *fakeWALStatusProvider) WALStatus() source.WALStatus { return f.status }
+
+func TestWALStatusCollector(t *testing.T) {
+	provider := &fakeWALStatusProvider{status: source.WALStatus{
+		Enabled:                   true,
+		LastExportSuccess:         time.Unix(1700000000, 0),
+		ExportFailuresTotal:       7,
+		ConsecutiveExportFailures: 2,
+		RestoreCompleted:          true,
+		RestoreObjectsApplied:     100,
+		RestoreErrors:             3,
+		RestoreDuration:           1500 * time.Millisecond,
+		RestoreLargestGap:         10 * time.Minute,
+	}}
+	collector := WALStatusCollector{
+		provider:      provider,
+		metricsConfig: MetricsConfig{DisabledMetrics: []string{"opencost_wal_restore_duration_seconds"}},
+	}
+
+	expected := `
+# HELP opencost_wal_consecutive_export_failures Collector WAL writes that have failed since the last successful write.
+# TYPE opencost_wal_consecutive_export_failures gauge
+opencost_wal_consecutive_export_failures 2
+# HELP opencost_wal_export_failures_total Total failed collector WAL writes since start.
+# TYPE opencost_wal_export_failures_total counter
+opencost_wal_export_failures_total 7
+# HELP opencost_wal_last_export_success_timestamp_seconds Unix time of the most recent successful collector WAL write, 0 if none.
+# TYPE opencost_wal_last_export_success_timestamp_seconds gauge
+opencost_wal_last_export_success_timestamp_seconds 1.7e+09
+# HELP opencost_wal_restore_errors_total Collector WAL objects that could not be read or decoded during the startup restore.
+# TYPE opencost_wal_restore_errors_total counter
+opencost_wal_restore_errors_total 3
+# HELP opencost_wal_restore_largest_gap_seconds Largest interval between consecutive restored collector WAL objects. Values well above the scrape interval indicate history that was never persisted or could not be restored.
+# TYPE opencost_wal_restore_largest_gap_seconds gauge
+opencost_wal_restore_largest_gap_seconds 600
+# HELP opencost_wal_restore_list_failed 1 if the collector WAL startup restore could not list stored objects, meaning nothing was restored.
+# TYPE opencost_wal_restore_list_failed gauge
+opencost_wal_restore_list_failed 0
+# HELP opencost_wal_restore_objects_applied Collector WAL objects applied during the startup restore.
+# TYPE opencost_wal_restore_objects_applied gauge
+opencost_wal_restore_objects_applied 100
+# HELP opencost_wal_restore_completed 1 once the collector WAL startup restore has finished, 0 before.
+# TYPE opencost_wal_restore_completed gauge
+opencost_wal_restore_completed 1
+`
+	if err := testutil.CollectAndCompare(collector, strings.NewReader(expected)); err != nil {
+		t.Errorf("unexpected metrics: %s", err)
+	}
+
+	provider.status = source.WALStatus{}
+	if n := testutil.CollectAndCount(collector); n != 0 {
+		t.Errorf("expected no metrics without a wal, got %d", n)
+	}
+}
+
+type fakeWALDataSource struct {
+	source.OpenCostDataSource
+	fakeWALStatusProvider
+}
+
+// An embedding application that already registered the WAL collector on the default registry must
+// not cause InitWALMetrics to panic.
+func TestInitWALMetrics_AlreadyRegistered(t *testing.T) {
+	existing := WALStatusCollector{provider: &fakeWALStatusProvider{}}
+	if err := prometheus.Register(existing); err != nil {
+		t.Fatalf("failed to pre-register collector: %s", err)
+	}
+	t.Cleanup(func() { prometheus.Unregister(existing) })
+
+	ds := &fakeWALDataSource{fakeWALStatusProvider: fakeWALStatusProvider{status: source.WALStatus{Enabled: true}}}
+	InitWALMetrics(ds, &MetricsConfig{})
+	InitWALMetrics(ds, &MetricsConfig{})
+}