Track collector WAL export and restore status

Walinator export and restore failures were only logged, so a consumer could
not tell that scrapes were not being persisted, or that a restart restored
incomplete history.

Walinator now records consecutive and total write failures, the last
success and (redacted) last error, and for the startup restore the number
of objects seen, applied and failed, any list error, the duration, the
restored time range and the largest gap between restored objects. A gap
well above the scrape interval shows history that was never persisted.
Write failures are logged at error level on the first failure and on
recovery rather than on every scrape.

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

modules/collector-source/pkg/metric/walinator.go

@@ -8,13 +8,16 @@ import (
 	"path"
 	"sort"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/opencost/opencost/core/pkg/exporter"
 	"github.com/opencost/opencost/core/pkg/exporter/pathing"
 	"github.com/opencost/opencost/core/pkg/log"
+	"github.com/opencost/opencost/core/pkg/source"
 	"github.com/opencost/opencost/core/pkg/storage"
 	"github.com/opencost/opencost/core/pkg/util/json"
+	"github.com/opencost/opencost/core/pkg/util/stringutil"
 	"github.com/opencost/opencost/core/pkg/util/worker"
 	"github.com/opencost/opencost/modules/collector-source/pkg/util"
 )
@@ -33,6 +36,9 @@ type Walinator struct {
 	exporter        exporter.EventExporter[UpdateSet]
 	limitResolution *util.Resolution
 	updater         Updater
+
+	statusLock sync.Mutex
+	status     source.WALStatus
 }
 
 func NewWalinator(
@@ -66,6 +72,7 @@ func NewWalinator(
 		exporter:        exp,
 		limitResolution: limitResolution,
 		updater:         updater,
+		status:          source.WALStatus{Enabled: true},
 	}, nil
 }
 
@@ -82,42 +89,100 @@ func (w *Walinator) Start() {
 	}()
 }
 
+// restoreResult is the outcome of reading a single wal file during restore
+type restoreResult struct {
+	fi        fileInfo
+	updateSet *UpdateSet
+}
+
 // restore applies updates from wal files to restore the state of the previous updater(repo)
 func (w *Walinator) restore() {
+	startTime := time.Now()
+	var listErr string
+
 	fileInfos, err := w.getFileInfos()
 	if err != nil {
 		log.Errorf("failed to retrieve updates files: %s", err.Error())
+		listErr = stringutil.RedactURLs(err.Error())
 	}
 	limit := w.limitResolution.Limit()
 
-	workerFn := func(fi fileInfo) *UpdateSet {
-		if fi.timestamp.Before(limit) {
-			return nil
+	var inRange []fileInfo
+	for _, fi := range fileInfos {
+		if !fi.timestamp.Before(limit) {
+			inRange = append(inRange, fi)
 		}
+	}
 
+	workerFn := func(fi fileInfo) restoreResult {
 		b, err := w.storage.Read(fi.name)
 		if err != nil {
 			log.Errorf("failed to load file contents for '%s': %s", fi.name, err.Error())
-			return nil
+			return restoreResult{fi: fi}
 		}
 
 		updateSet, err := deserializeUpdateSet(fi.ext, b)
 		if err != nil {
 			log.Errorf("failed to deserialize file contents for '%s': %s", fi.name, err.Error())
-			return nil
+			return restoreResult{fi: fi}
 		}
 
 		if updateSet.Timestamp.IsZero() {
 			updateSet.Timestamp = fi.timestamp
 		}
 
-		return updateSet
+		return restoreResult{fi: fi, updateSet: updateSet}
+	}
+
+	// processFn is called in file order from a single goroutine
+	var applied, errs int
+	var oldest, newest, gapStart time.Time
+	var largestGap time.Duration
+	processFn := func(res restoreResult) {
+		if res.updateSet == nil {
+			errs++
+			return
+		}
+		w.updater.Update(res.updateSet)
+		applied++
+
+		ts := res.fi.timestamp
+		if oldest.IsZero() {
+			oldest = ts
+		} else if gap := ts.Sub(newest); gap > largestGap {
+			largestGap = gap
+			gapStart = newest
+		}
+		newest = ts
 	}
+	worker.ConcurrentOrderedProcessWith(worker.OptimalWorkerCount(), workerFn, inRange, processFn)
 
-	processFn := func(updateSet *UpdateSet) {
-		w.updater.Update(updateSet)
+	duration := time.Since(startTime)
+	if listErr != "" || errs > 0 {
+		log.Errorf("wal restore incomplete: %d of %d objects applied, %d errors, list error: %q", applied, len(inRange), errs, listErr)
+	} else {
+		log.Infof("wal restore complete: %d objects applied in %s, largest gap %s", applied, duration, largestGap)
 	}
-	worker.ConcurrentOrderedProcessWith(worker.OptimalWorkerCount(), workerFn, fileInfos, processFn)
+
+	w.statusLock.Lock()
+	defer w.statusLock.Unlock()
+	w.status.RestoreCompleted = true
+	w.status.RestoreListError = listErr
+	w.status.RestoreObjectsSeen = len(inRange)
+	w.status.RestoreObjectsApplied = applied
+	w.status.RestoreErrors = errs
+	w.status.RestoreDuration = duration
+	w.status.RestoreOldest = oldest
+	w.status.RestoreNewest = newest
+	w.status.RestoreLargestGap = largestGap
+	w.status.RestoreLargestGapStart = gapStart
+}
+
+// Status returns the current export and restore status of the wal
+func (w *Walinator) Status() source.WALStatus {
+	w.statusLock.Lock()
+	defer w.statusLock.Unlock()
+	return w.status
 }
 
 func deserializeUpdateSet(ext string, b []byte) (*UpdateSet, error) {
@@ -169,9 +234,34 @@ func (w *Walinator) Update(
 	w.updater.Update(updateSet)
 
 	err := w.exporter.Export(updateSet.Timestamp, updateSet)
-	if err != nil {
+	w.recordExport(err)
+}
+
+// recordExport updates the export status, logging only when the export state changes so that a
+// long outage does not produce an error log per scrape
+func (w *Walinator) recordExport(err error) {
+	w.statusLock.Lock()
+	defer w.statusLock.Unlock()
+
+	now := time.Now().UTC()
+	if err == nil {
+		if w.status.ConsecutiveExportFailures > 0 {
+			log.Infof("wal export recovered after %d failed writes", w.status.ConsecutiveExportFailures)
+		}
+		w.status.LastExportSuccess = now
+		w.status.ConsecutiveExportFailures = 0
+		return
+	}
+
+	if w.status.ConsecutiveExportFailures == 0 {
 		log.Errorf("failed to export update results: %s", err.Error())
+	} else {
+		log.Debugf("failed to export update results: %s", err.Error())
 	}
+	w.status.LastExportError = stringutil.RedactURLs(err.Error())
+	w.status.LastExportErrorAt = now
+	w.status.ConsecutiveExportFailures++
+	w.status.ExportFailuresTotal++
 }
 
 // getFileInfos returns a sorted slice of fileInfo

modules/collector-source/pkg/metric/walinator_status_test.go

@@ -0,0 +1,206 @@
+package metric
+
+import (
+	"errors"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/opencost/opencost/core/pkg/storage"
+	"github.com/opencost/opencost/core/pkg/util/timeutil"
+	"github.com/opencost/opencost/modules/collector-source/pkg/util"
+)
+
+// flakyStorage wraps a MemoryStorage and fails writes, reads or lists while the corresponding flag is set.
+type flakyStorage struct {
+	*storage.MemoryStorage
+
+	mu        sync.Mutex
+	failWrite bool
+	failRead  bool
+	failList  bool
+}
+
+var errBucketUnavailable = errors.New(`Put "https://bucket.s3.amazonaws.com/wal?X-Amz-Credential=AKIAEXAMPLE&X-Amz-Signature=deadbeef": 503 Service Unavailable`)
+
+func (fs *flakyStorage) set(write, read, list bool) {
+	fs.mu.Lock()
+	defer fs.mu.Unlock()
+	fs.failWrite, fs.failRead, fs.failList = write, read, list
+}
+
+func (fs *flakyStorage) Write(path string, data []byte) error {
+	fs.mu.Lock()
+	fail := fs.failWrite
+	fs.mu.Unlock()
+	if fail {
+		return errBucketUnavailable
+	}
+	return fs.MemoryStorage.Write(path, data)
+}
+
+func (fs *flakyStorage) Read(path string) ([]byte, error) {
+	fs.mu.Lock()
+	fail := fs.failRead
+	fs.mu.Unlock()
+	if fail {
+		return nil, errBucketUnavailable
+	}
+	return fs.MemoryStorage.Read(path)
+}
+
+func (fs *flakyStorage) List(path string) ([]*storage.StorageInfo, error) {
+	fs.mu.Lock()
+	fail := fs.failList
+	fs.mu.Unlock()
+	if fail {
+		return nil, errBucketUnavailable
+	}
+	return fs.MemoryStorage.List(path)
+}
+
+func newTestWalinator(t *testing.T, store storage.Storage) *Walinator {
+	t.Helper()
+	res1d, err := util.NewResolution(util.ResolutionConfiguration{Interval: "1d", Retention: 3})
+	if err != nil {
+		t.Fatalf("failed to create resolution: %s", err)
+	}
+	resolutions := []*util.Resolution{res1d}
+	wal, err := NewWalinator("test", "test", store, resolutions, NewMetricRepository(resolutions, testMetricCollector))
+	if err != nil {
+		t.Fatalf("failed to create walinator: %s", err)
+	}
+	return wal
+}
+
+func testUpdateSet(ts time.Time) *UpdateSet {
+	return &UpdateSet{
+		Timestamp: ts,
+		Updates: []Update{{
+			Name:   TestMetric,
+			Labels: map[string]string{"test": "test"},
+			Value:  1,
+		}},
+	}
+}
+
+// TestWalinator_ExportOutageIsObservable reproduces F-34: WAL writes fail for a period, then the
+// process restarts. Both the write failures and the resulting hole in restored history must be
+// visible through Status(), not only in logs.
+func TestWalinator_ExportOutageIsObservable(t *testing.T) {
+	store := &flakyStorage{MemoryStorage: storage.NewMemoryStorage()}
+	wal := newTestWalinator(t, store)
+
+	base := time.Now().UTC().Truncate(timeutil.Day).Add(-12 * time.Hour)
+	scrape := 30 * time.Second
+
+	// 10 good scrapes, 20 failed scrapes (a 10 minute outage), 10 good scrapes
+	ts := base
+	for i := 0; i < 10; i++ {
+		wal.Update(testUpdateSet(ts))
+		ts = ts.Add(scrape)
+	}
+	outageStart := ts.Add(-scrape)
+
+	store.set(true, false, false)
+	for i := 0; i < 20; i++ {
+		wal.Update(testUpdateSet(ts))
+		ts = ts.Add(scrape)
+	}
+
+	status := wal.Status()
+	if !status.Enabled {
+		t.Errorf("expected wal status to be enabled")
+	}
+	if status.ConsecutiveExportFailures != 20 || status.ExportFailuresTotal != 20 {
+		t.Errorf("expected 20 consecutive and total export failures, got %d and %d", status.ConsecutiveExportFailures, status.ExportFailuresTotal)
+	}
+	if status.LastExportErrorAt.IsZero() || status.LastExportError == "" {
+		t.Errorf("expected last export error to be recorded, got %+v", status)
+	}
+	if strings.Contains(status.LastExportError, "AKIAEXAMPLE") || strings.Contains(status.LastExportError, "X-Amz-Signature") {
+		t.Errorf("last export error leaks credentials: %q", status.LastExportError)
+	}
+
+	store.set(false, false, false)
+	for i := 0; i < 10; i++ {
+		wal.Update(testUpdateSet(ts))
+		ts = ts.Add(scrape)
+	}
+
+	status = wal.Status()
+	if status.ConsecutiveExportFailures != 0 {
+		t.Errorf("expected consecutive failures to reset after recovery, got %d", status.ConsecutiveExportFailures)
+	}
+	if status.ExportFailuresTotal != 20 {
+		t.Errorf("expected total failures to be retained after recovery, got %d", status.ExportFailuresTotal)
+	}
+	if status.LastExportSuccess.Before(status.LastExportErrorAt) {
+		t.Errorf("expected last success after last error once recovered")
+	}
+
+	// restart: a new walinator over the same storage
+	restarted := newTestWalinator(t, store)
+	restarted.restore()
+
+	rs := restarted.Status()
+	if !rs.RestoreCompleted {
+		t.Fatalf("expected restore to be completed")
+	}
+	if rs.RestoreObjectsSeen != 20 || rs.RestoreObjectsApplied != 20 || rs.RestoreErrors != 0 {
+		t.Errorf("expected 20 objects seen and applied with no errors, got seen=%d applied=%d errors=%d",
+			rs.RestoreObjectsSeen, rs.RestoreObjectsApplied, rs.RestoreErrors)
+	}
+	wantGap := 21 * scrape
+	if rs.RestoreLargestGap != wantGap {
+		t.Errorf("expected largest restored gap %s, got %s", wantGap, rs.RestoreLargestGap)
+	}
+	if !rs.RestoreLargestGapStart.Equal(outageStart) {
+		t.Errorf("expected largest gap to start at %s, got %s", outageStart, rs.RestoreLargestGapStart)
+	}
+	if !rs.RestoreOldest.Equal(base) || !rs.RestoreNewest.Equal(ts.Add(-scrape)) {
+		t.Errorf("unexpected restored range %s - %s", rs.RestoreOldest, rs.RestoreNewest)
+	}
+}
+
+// TestWalinator_RestoreFailuresAreObservable covers restore errors that were previously log-only:
+// unreadable objects and an unlistable bucket.
+func TestWalinator_RestoreFailuresAreObservable(t *testing.T) {
+	store := &flakyStorage{MemoryStorage: storage.NewMemoryStorage()}
+	wal := newTestWalinator(t, store)
+
+	base := time.Now().UTC().Truncate(timeutil.Day).Add(-12 * time.Hour)
+	for i := 0; i < 5; i++ {
+		wal.Update(testUpdateSet(base.Add(time.Duration(i) * time.Minute)))
+	}
+
+	t.Run("unreadable objects", func(t *testing.T) {
+		store.set(false, true, false)
+		defer store.set(false, false, false)
+
+		restarted := newTestWalinator(t, store)
+		restarted.restore()
+
+		rs := restarted.Status()
+		if !rs.RestoreCompleted || rs.RestoreObjectsSeen != 5 || rs.RestoreObjectsApplied != 0 || rs.RestoreErrors != 5 {
+			t.Errorf("expected 5 seen, 0 applied, 5 errors; got %+v", rs)
+		}
+	})
+
+	t.Run("unlistable bucket", func(t *testing.T) {
+		store.set(false, false, true)
+		defer store.set(false, false, false)
+
+		restarted := newTestWalinator(t, store)
+		restarted.restore()
+
+		rs := restarted.Status()
+		if !rs.RestoreCompleted || rs.RestoreListError == "" {
+			t.Errorf("expected list error to be reported, got %+v", rs)
+		}
+		if strings.Contains(rs.RestoreListError, "AKIAEXAMPLE") {
+			t.Errorf("restore list error leaks credentials: %q", rs.RestoreListError)
+		}
+	})
+}