Report WAL history lost before a restart and harden redaction

Review found that a crash while WAL writes were failing left no trace:
the restarted process has no failure counters and the restored history
has no interior gap, so the lost tail was invisible. The restore now
records its start time and the tail gap between the newest restored
object and the restart, exposed as opencost_wal_restore_tail_gap_seconds
and opencost_wal_restore_newest_timestamp_seconds.

opencost_wal_restore_errors_total is renamed opencost_wal_restore_errors
and is a gauge, since it is set once at startup and increase() over it
would never fire.

RedactURLs also redacts well known signature and credential parameters
(sig, X-Amz-*, X-Goog-*, AccountKey, SharedAccessSignature, tokens)
wherever they appear, including URLs without a scheme and connection
strings, and strips passwords containing @. WAL error logs are redacted
too, and a changed error during an ongoing outage is logged at error
level.

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

core/pkg/source/walstatus.go

@@ -3,7 +3,8 @@ package source
 import "time"
 
 // WALStatus reports the health of a data source's write-ahead log: whether updates are being
-// persisted, and how complete the most recent restore was.
+// persisted, and how complete the most recent restore was. Durations serialize to JSON as integer
+// nanoseconds.
 type WALStatus struct {
 	// Enabled is false when the data source has no WAL configured; all other fields are zero.
 	Enabled bool `json:"enabled"`
@@ -22,6 +23,8 @@ type WALStatus struct {
 
 	// RestoreCompleted is true once the startup restore has finished, whether or not it had errors.
 	RestoreCompleted bool `json:"restoreCompleted"`
+	// RestoreStartedAt is the time the startup restore began.
+	RestoreStartedAt time.Time `json:"restoreStartedAt"`
 	// RestoreListError is set when the WAL objects could not be listed, meaning nothing was restored.
 	RestoreListError string `json:"restoreListError,omitempty"`
 	// RestoreObjectsSeen is the number of WAL objects inside the retention window.
@@ -40,6 +43,10 @@ type WALStatus struct {
 	// was never persisted or could not be restored.
 	RestoreLargestGap      time.Duration `json:"restoreLargestGap"`
 	RestoreLargestGapStart time.Time     `json:"restoreLargestGapStart"`
+	// RestoreTailGap is the interval between the newest applied object (or the start of the retention
+	// window, if nothing was applied) and the start of the restore: history that was not persisted
+	// before the restart, whether because the process was down or because writes were failing.
+	RestoreTailGap time.Duration `json:"restoreTailGap"`
 }
 
 // WALStatusProvider is optionally implemented by an OpenCostDataSource that persists its state

core/pkg/util/stringutil/redact.go

@@ -2,25 +2,31 @@ package stringutil
 
 import (
 	"regexp"
+	"strings"
 )
 
 // urlPattern matches scheme://... URLs up to the next whitespace or quote.
 var urlPattern = regexp.MustCompile(`[a-zA-Z][a-zA-Z0-9+.-]*://[^\s"']+`)
 
-// userInfoPattern matches the user:password@ portion following a URL scheme.
-var userInfoPattern = regexp.MustCompile(`^([a-zA-Z][a-zA-Z0-9+.-]*://)[^/@]*@`)
+// userInfoPattern matches the user:password@ portion following a URL scheme, up to the last @ before
+// the path, so passwords containing @ are fully removed.
+var userInfoPattern = regexp.MustCompile(`^([a-zA-Z][a-zA-Z0-9+.-]*://)[^/]*@`)
 
-// RedactURLs removes query strings, fragments and user info from any URLs contained in s, so
-// that error messages from storage clients (which may embed presigned URLs or credentials) are
-// safe to expose through status endpoints and diagnostics.
+// secretParamPattern matches the values of well known signature and credential parameters used by
+// cloud storage presigned URLs and connection strings, wherever they appear.
+var secretParamPattern = regexp.MustCompile(`(?i)\b(sig|signature|x-amz-signature|x-amz-credential|x-amz-security-token|x-goog-signature|x-goog-credential|accountkey|sharedaccesssignature|access_token|token)=[^&;\s"']+`)
+
+// RedactURLs removes query strings, fragments and user info from any URLs contained in s, and the
+// values of well known signature and credential parameters anywhere in s, so that error messages
+// from storage clients (which may embed presigned URLs or credentials) are safe to expose through
+// logs, status endpoints and diagnostics.
 func RedactURLs(s string) string {
-	return urlPattern.ReplaceAllStringFunc(s, func(u string) string {
+	s = urlPattern.ReplaceAllStringFunc(s, func(u string) string {
 		u = userInfoPattern.ReplaceAllString(u, "${1}REDACTED@")
-		for i, r := range u {
-			if r == '?' || r == '#' {
-				return u[:i] + "?REDACTED"
-			}
+		if i := strings.IndexAny(u, "?#"); i >= 0 {
+			return u[:i] + "?REDACTED"
 		}
 		return u
 	})
+	return secretParamPattern.ReplaceAllString(s, "${1}=REDACTED")
 }

core/pkg/util/stringutil/redact_test.go

@@ -1,18 +1,32 @@
 package stringutil
 
-import "testing"
+import (
+	"strings"
+	"testing"
+)
 
 func TestRedactURLs(t *testing.T) {
 	cases := map[string]string{
 		"plain error": "plain error",
 		`Put "https://b.s3.amazonaws.com/k?X-Amz-Signature=abc&X-Amz-Credential=AKIA": EOF`: `Put "https://b.s3.amazonaws.com/k?REDACTED": EOF`,
 		"dial https://user:secret@host/path failed":                                         "dial https://REDACTED@host/path failed",
+		"dial https://user:p@ss@host/path failed":                                           "dial https://REDACTED@host/path failed",
 		"see https://host/a#frag and s3://b/k?sig=1":                                        "see https://host/a?REDACTED and s3://b/k?REDACTED",
 		"no query https://host/path/object.gz":                                              "no query https://host/path/object.gz",
+		"Put bucket.s3.amazonaws.com/k?X-Amz-Signature=abc&x=1":                             "Put bucket.s3.amazonaws.com/k?X-Amz-Signature=REDACTED&x=1",
+		`https:\/\/h\/k?sv=2020&sig=SECRET`:                                                 `https:\/\/h\/k?sv=2020&sig=REDACTED`,
+		"AccountName=a;AccountKey=SECRET;EndpointSuffix=core":                               "AccountName=a;AccountKey=REDACTED;EndpointSuffix=core",
+		"SharedAccessSignature=SECRET;BlobEndpoint=x":                                       "SharedAccessSignature=REDACTED;BlobEndpoint=x",
+		"https://h/k?a=1 &sig=SECRET":                                                       "https://h/k?REDACTED &sig=REDACTED",
+		"https://storage.googleapis.com/b/o?X-Goog-Signature=SECRET":                        "https://storage.googleapis.com/b/o?REDACTED",
 	}
 	for in, want := range cases {
-		if got := RedactURLs(in); got != want {
+		got := RedactURLs(in)
+		if got != want {
 			t.Errorf("RedactURLs(%q) = %q, want %q", in, got, want)
 		}
+		if strings.Contains(got, "SECRET") {
+			t.Errorf("RedactURLs(%q) leaked a secret: %q", in, got)
+		}
 	}
 }

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

@@ -172,8 +172,12 @@ const (
 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))
+		since := "no successful write since start"
+		if !status.LastExportSuccess.IsZero() {
+			since = "last successful write at " + status.LastExportSuccess.Format(time.RFC3339)
+		}
+		problems = append(problems, fmt.Sprintf("%d consecutive write failures, %s (last error: %s)",
+			status.ConsecutiveExportFailures, since, status.LastExportError))
 	}
 	if status.RestoreListError != "" {
 		problems = append(problems, fmt.Sprintf("restore could not list objects: %s", status.RestoreListError))
@@ -195,6 +199,7 @@ func walDiagnosticDetails(status source.WALStatus) (map[string]any, error) {
 		"restoreNewest":          status.RestoreNewest,
 		"restoreLargestGap":      status.RestoreLargestGap.String(),
 		"restoreLargestGapStart": status.RestoreLargestGapStart,
+		"restoreTailGap":         status.RestoreTailGap.String(),
 	}, nil
 }
 

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

@@ -97,13 +97,13 @@ type restoreResult struct {
 
 // restore applies updates from wal files to restore the state of the previous updater(repo)
 func (w *Walinator) restore() {
-	startTime := time.Now()
+	startTime := time.Now().UTC()
 	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())
+		log.Errorf("failed to retrieve updates files: %s", listErr)
 	}
 	limit := w.limitResolution.Limit()
 
@@ -117,7 +117,7 @@ func (w *Walinator) restore() {
 	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())
+			log.Errorf("failed to load file contents for '%s': %s", fi.name, stringutil.RedactURLs(err.Error()))
 			return restoreResult{fi: fi}
 		}
 
@@ -158,15 +158,21 @@ func (w *Walinator) restore() {
 	worker.ConcurrentOrderedProcessWith(worker.OptimalWorkerCount(), workerFn, inRange, processFn)
 
 	duration := time.Since(startTime)
+	tailFrom := newest
+	if tailFrom.IsZero() {
+		tailFrom = limit
+	}
+	tailGap := max(startTime.Sub(tailFrom), 0)
 	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)
+		log.Infof("wal restore complete: %d objects applied in %s, largest gap %s, %s since newest object", applied, duration, largestGap, tailGap)
 	}
 
 	w.statusLock.Lock()
 	defer w.statusLock.Unlock()
 	w.status.RestoreCompleted = true
+	w.status.RestoreStartedAt = startTime
 	w.status.RestoreListError = listErr
 	w.status.RestoreObjectsSeen = len(inRange)
 	w.status.RestoreObjectsApplied = applied
@@ -176,6 +182,7 @@ func (w *Walinator) restore() {
 	w.status.RestoreNewest = newest
 	w.status.RestoreLargestGap = largestGap
 	w.status.RestoreLargestGapStart = gapStart
+	w.status.RestoreTailGap = tailGap
 }
 
 // Status returns the current export and restore status of the wal
@@ -253,12 +260,14 @@ func (w *Walinator) recordExport(err error) {
 		return
 	}
 
-	if w.status.ConsecutiveExportFailures == 0 {
-		log.Errorf("failed to export update results: %s", err.Error())
+	msg := stringutil.RedactURLs(err.Error())
+	// log at error level when writes start failing or the cause changes, not on every scrape
+	if w.status.ConsecutiveExportFailures == 0 || msg != w.status.LastExportError {
+		log.Errorf("failed to export update results: %s", msg)
 	} else {
-		log.Debugf("failed to export update results: %s", err.Error())
+		log.Debugf("failed to export update results: %s", msg)
 	}
-	w.status.LastExportError = stringutil.RedactURLs(err.Error())
+	w.status.LastExportError = msg
 	w.status.LastExportErrorAt = now
 	w.status.ConsecutiveExportFailures++
 	w.status.ExportFailuresTotal++

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

@@ -204,3 +204,57 @@ func TestWalinator_RestoreFailuresAreObservable(t *testing.T) {
 		}
 	})
 }
+
+// TestWalinator_OutageUntilRestartIsObservable covers the most likely F-34 sequence: writes fail and the
+// process restarts before they recover. The lost tail of history must show up in the restored status
+// even though the restarted process has no record of the failed writes.
+func TestWalinator_OutageUntilRestartIsObservable(t *testing.T) {
+	store := &flakyStorage{MemoryStorage: storage.NewMemoryStorage()}
+	wal := newTestWalinator(t, store)
+
+	scrape := 30 * time.Second
+	now := time.Now().UTC()
+	lastPersisted := now.Add(-30 * time.Minute)
+
+	// scrapes persist until 30 minutes ago, then every write fails until the restart
+	for ts := lastPersisted.Add(-10 * scrape); !ts.After(lastPersisted); ts = ts.Add(scrape) {
+		wal.Update(testUpdateSet(ts))
+	}
+	store.set(true, false, false)
+	for ts := lastPersisted.Add(scrape); ts.Before(now); ts = ts.Add(scrape) {
+		wal.Update(testUpdateSet(ts))
+	}
+	store.set(false, false, false)
+
+	restarted := newTestWalinator(t, store)
+	restarted.restore()
+
+	rs := restarted.Status()
+	if rs.ConsecutiveExportFailures != 0 {
+		t.Fatalf("expected a fresh process to have no export failures, got %d", rs.ConsecutiveExportFailures)
+	}
+	if rs.RestoreLargestGap != scrape {
+		t.Errorf("expected no interior gap, got %s", rs.RestoreLargestGap)
+	}
+	if !rs.RestoreNewest.Equal(lastPersisted.Truncate(time.Second)) && !rs.RestoreNewest.Equal(lastPersisted) {
+		t.Errorf("expected newest restored object at %s, got %s", lastPersisted, rs.RestoreNewest)
+	}
+	if rs.RestoreTailGap < 30*time.Minute {
+		t.Errorf("expected a tail gap of at least 30m, got %s", rs.RestoreTailGap)
+	}
+	if !rs.RestoreStartedAt.After(rs.RestoreNewest) {
+		t.Errorf("expected restore start %s after newest object %s", rs.RestoreStartedAt, rs.RestoreNewest)
+	}
+}
+
+// TestWalinator_EmptyRestoreTailGap reports the whole retention window as missing when nothing was
+// restored.
+func TestWalinator_EmptyRestoreTailGap(t *testing.T) {
+	wal := newTestWalinator(t, storage.NewMemoryStorage())
+	wal.restore()
+
+	rs := wal.Status()
+	if rs.RestoreObjectsApplied != 0 || rs.RestoreTailGap < 2*timeutil.Day {
+		t.Errorf("expected nothing applied and a tail gap covering retention, got applied=%d tail=%s", rs.RestoreObjectsApplied, rs.RestoreTailGap)
+	}
+}

pkg/metrics/walmetrics.go

@@ -49,12 +49,18 @@ var walMetrics = []walMetric{
 	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,
+	newWALMetric("opencost_wal_restore_errors",
+		"Collector WAL objects that could not be read or decoded during the startup restore.", prometheus.GaugeValue,
 		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_newest_timestamp_seconds",
+		"Unix time of the newest collector WAL object applied during the startup restore, 0 if none.", prometheus.GaugeValue,
+		func(s source.WALStatus) float64 { return unixOrZero(s.RestoreNewest.Unix(), s.RestoreNewest.IsZero()) }),
+	newWALMetric("opencost_wal_restore_tail_gap_seconds",
+		"Interval between the newest restored collector WAL object and the start of the restore: history not persisted before the restart.", prometheus.GaugeValue,
+		func(s source.WALStatus) float64 { return s.RestoreTailGap.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() }),

pkg/metrics/walmetrics_test.go

@@ -27,6 +27,8 @@ func TestWALStatusCollector(t *testing.T) {
 		RestoreErrors:             3,
 		RestoreDuration:           1500 * time.Millisecond,
 		RestoreLargestGap:         10 * time.Minute,
+		RestoreNewest:             time.Unix(1699999000, 0),
+		RestoreTailGap:            30 * time.Minute,
 	}}
 	collector := WALStatusCollector{
 		provider:      provider,
@@ -43,12 +45,18 @@ 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_errors Collector WAL objects that could not be read or decoded during the startup restore.
+# TYPE opencost_wal_restore_errors gauge
+opencost_wal_restore_errors 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_newest_timestamp_seconds Unix time of the newest collector WAL object applied during the startup restore, 0 if none.
+# TYPE opencost_wal_restore_newest_timestamp_seconds gauge
+opencost_wal_restore_newest_timestamp_seconds 1.699999e+09
+# HELP opencost_wal_restore_tail_gap_seconds Interval between the newest restored collector WAL object and the start of the restore: history not persisted before the restart.
+# TYPE opencost_wal_restore_tail_gap_seconds gauge
+opencost_wal_restore_tail_gap_seconds 1800
 # 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