From deaed7ce9b9c9611380187822e000c0c629ad30b Mon Sep 17 00:00:00 2001 From: Warwick Date: Mon, 13 Apr 2026 13:58:51 -0500 Subject: [PATCH 1/3] fix: persist snapshot state across restarts to prevent incomplete window exports - Add SnapshotState struct to persist lastSnapshot timestamp - Implement persistState() and recoverState() methods in ConcurrentSnapshotProvider - Add ScratchDir configuration to SnapshotConfig (from SCRATCH_DIR env var) - Recover persisted state on provider initialization - Persist state after successful export in exporter loop - Add PersistState() method to SnapshotProvider interface - Update test mocks to implement new interface method This fixes the in-memory state loss issue where the first export after restart would feed allocation/asset pipelines with incomplete window data. State is persisted to {SCRATCH_DIR}/snapshot-state.json after successful export, ensuring window continuity across restarts. --- pkg/emitter/exporter.go | 5 +++ pkg/emitter/exporter_test.go | 4 ++ pkg/emitter/snapshot.go | 81 ++++++++++++++++++++++++++++++++++- pkg/emitter/snapshotconfig.go | 15 +++++++ 4 files changed, 104 insertions(+), 1 deletion(-) diff --git a/pkg/emitter/exporter.go b/pkg/emitter/exporter.go index f5e25777..8e98a1bd 100644 --- a/pkg/emitter/exporter.go +++ b/pkg/emitter/exporter.go @@ -113,6 +113,11 @@ func (de *defaultExporter) Start(interval time.Duration) bool { // wait for all emit tasks to complete before continuing emitTasks.Wait() + + // persist snapshot state after successful export to maintain window continuity across restarts + if err := de.snapshotProvider.PersistState(); err != nil { + log.Warnf("failed to persist snapshot state: %v", err) + } } }() diff --git a/pkg/emitter/exporter_test.go b/pkg/emitter/exporter_test.go index 40634bbe..7d4f040b 100644 --- a/pkg/emitter/exporter_test.go +++ b/pkg/emitter/exporter_test.go @@ -172,6 +172,10 @@ func (e *emptySnapshotProvider) SnapshotOf(ds core.DataSource) (*ClusterSnapshot return &ClusterSnapshot{}, nil } +func (e *emptySnapshotProvider) PersistState() error { + return nil +} + type emptyDataSource struct{} func (e *emptyDataSource) OpenCostSource() source.OpenCostDataSource { diff --git a/pkg/emitter/snapshot.go b/pkg/emitter/snapshot.go index e7e43bd8..ae518b24 100644 --- a/pkg/emitter/snapshot.go +++ b/pkg/emitter/snapshot.go @@ -1,8 +1,11 @@ package emitter import ( + "encoding/json" "errors" "fmt" + "os" + "path/filepath" "time" "github.com/hashicorp/go-multierror" @@ -21,12 +24,22 @@ type SnapshotProvider interface { // SnapshotOf generates a `ClusterSnapshot` from the provided `core.DataSource` and returns it. // If the snapshot generation fails, an error is returned. SnapshotOf(core.DataSource) (*ClusterSnapshot, error) + + // PersistState persists the current snapshot state to disk for recovery after restart. + // This should be called after successful export to ensure state consistency. + PersistState() error } // FIXME: (bolt) use a metrics summary cache duration of 5 minutes while we're using a prometheus data source. // FIXME: (bolt) this should be fine to run on a much faster frequency with a non-promethues metrics querier. var metricsSummaryCacheDuration time.Duration = 5 * time.Minute +// SnapshotState represents the persisted state of the snapshot provider to maintain +// continuity across restarts. +type SnapshotState struct { + LastSnapshot time.Time `json:"lastSnapshot"` +} + // ConcurrentSnapshotProvider is a struct that implements the `SnapshotProvider` interface and executes the // snapshot generation process concurrently. type ConcurrentSnapshotProvider struct { @@ -48,10 +61,76 @@ func NewConcurrentSnapshotProvider(config *SnapshotConfig) SnapshotProvider { now = defaultNow } - return &ConcurrentSnapshotProvider{ + csp := &ConcurrentSnapshotProvider{ now: now, config: config, } + + // Attempt to recover persisted state from previous runs + csp.recoverState() + + return csp +} + +// persistState writes the current lastSnapshot timestamp to disk for recovery after restart. +func (csp *ConcurrentSnapshotProvider) persistState() error { + if csp.config.ScratchDir == "" { + return fmt.Errorf("scratch directory not configured") + } + + stateFile := filepath.Join(csp.config.ScratchDir, "snapshot-state.json") + state := SnapshotState{ + LastSnapshot: csp.lastSnapshot, + } + + data, err := json.Marshal(state) + if err != nil { + return fmt.Errorf("failed to marshal snapshot state: %w", err) + } + + if err := os.MkdirAll(csp.config.ScratchDir, 0755); err != nil { + return fmt.Errorf("failed to create scratch directory: %w", err) + } + + if err := os.WriteFile(stateFile, data, 0644); err != nil { + return fmt.Errorf("failed to write snapshot state: %w", err) + } + + return nil +} + +// recoverState attempts to read the persisted snapshot state from disk. +// If the state file is missing or corrupt, it treats this as a cold start and logs a warning. +func (csp *ConcurrentSnapshotProvider) recoverState() { + if csp.config.ScratchDir == "" { + log.Warnf("Scratch directory not configured, starting with empty snapshot state") + return + } + + stateFile := filepath.Join(csp.config.ScratchDir, "snapshot-state.json") + data, err := os.ReadFile(stateFile) + if err != nil { + if os.IsNotExist(err) { + log.Infof("No previous snapshot state found, starting fresh") + } else { + log.Warnf("Failed to read snapshot state file: %v, starting fresh", err) + } + return + } + + var state SnapshotState + if err := json.Unmarshal(data, &state); err != nil { + log.Warnf("Failed to unmarshal snapshot state: %v, starting fresh", err) + return + } + + csp.lastSnapshot = state.LastSnapshot + log.Infof("Recovered snapshot state: lastSnapshot=%s", state.LastSnapshot.Format(time.RFC3339)) +} + +// PersistState implements the SnapshotProvider interface method to persist state. +func (csp *ConcurrentSnapshotProvider) PersistState() error { + return csp.persistState() } // SnapshotOf generates a `ClusterSnapshot` from the provided `core.DataSource` and returns it. diff --git a/pkg/emitter/snapshotconfig.go b/pkg/emitter/snapshotconfig.go index 18090d03..35a67ef1 100644 --- a/pkg/emitter/snapshotconfig.go +++ b/pkg/emitter/snapshotconfig.go @@ -1,6 +1,7 @@ package emitter import ( + "os" "time" "github.com/ibm/finops-agent/pkg/env" @@ -181,6 +182,15 @@ func defaultNow() time.Time { return time.Now().UTC() } +// getScratchDir returns the scratch directory path from the SCRATCH_DIR environment variable, +// or the default path if not set. +func getScratchDir() string { + if dir := os.Getenv("SCRATCH_DIR"); dir != "" { + return dir + } + return "/opt/finops-agent" +} + // SnapshotConfig holds the configuration for general snapshotting options. type SnapshotConfig struct { // UseMetricsCache indicates whether or not to use a cache for metrics query results. @@ -194,6 +204,9 @@ type SnapshotConfig struct { // KubernetesSnapshotConfig holds the configuration for Kubernetes resources to snapshot. KubernetesSnapshot *KubernetesSnapshotConfig + // ScratchDir is the directory path where snapshot state will be persisted. + ScratchDir string + // Now is the func used to determine the current time. Now Now } @@ -215,6 +228,7 @@ func NewSnapshotConfigFromEnv() *SnapshotConfig { return &SnapshotConfig{ UseMetricsCache: !env.IsCollectorDataSourceEnabled(), MinutelyMetricsEnabled: env.IsMinuteMetricsEnabled(), + ScratchDir: getScratchDir(), Now: defaultNow, } } @@ -225,6 +239,7 @@ func DefaultSnapshotConfig() *SnapshotConfig { UseMetricsCache: false, MinutelyMetricsEnabled: false, KubernetesSnapshot: NewKubernetesSnapshotConfig().EnableAll(), + ScratchDir: "/opt/finops-agent", Now: defaultNow, } } From c31f0379bf92f808cee6836e047ec2a7a8215830 Mon Sep 17 00:00:00 2001 From: Warwick Date: Mon, 13 Apr 2026 14:04:51 -0500 Subject: [PATCH 2/3] fix: persist snapshot state across restarts to prevent incomplete window exports - Add SnapshotState struct to persist lastSnapshot timestamp - Implement persistState() and recoverState() methods in ConcurrentSnapshotProvider - Add ScratchDir configuration to SnapshotConfig (from SCRATCH_DIR env var) - Recover persisted state on provider initialization - Track emit errors and only persist state after successful export - Add PersistState() method to SnapshotProvider interface - Update test mocks to implement new interface method - Add comprehensive unit tests for state persistence and recovery - Extract state filename as constant This fixes the in-memory state loss issue where the first export after restart would feed allocation/asset pipelines with incomplete window data. State is persisted to {SCRATCH_DIR}/snapshot-state.json only after all emitters succeed, ensuring window continuity across restarts. --- pkg/emitter/exporter.go | 15 ++- pkg/emitter/snapshot.go | 6 +- pkg/emitter/snapshot_state_test.go | 186 +++++++++++++++++++++++++++++ 3 files changed, 200 insertions(+), 7 deletions(-) create mode 100644 pkg/emitter/snapshot_state_test.go diff --git a/pkg/emitter/exporter.go b/pkg/emitter/exporter.go index 8e98a1bd..51110bb2 100644 --- a/pkg/emitter/exporter.go +++ b/pkg/emitter/exporter.go @@ -5,12 +5,13 @@ import ( "fmt" "runtime/debug" "sync" + "sync/atomic" "time" "github.com/ibm/finops-agent/pkg/core" "github.com/ibm/finops-agent/pkg/util" + ocatomic "github.com/opencost/opencost/core/pkg/util/atomic" "github.com/opencost/opencost/core/pkg/log" - "github.com/opencost/opencost/core/pkg/util/atomic" ) // Exporter is an interface that defines a data emission management system and facilitates the @@ -31,7 +32,7 @@ type Exporter interface { // defaultExporter is the default implementation of the Exporter interface. It's a straight-forward // snapshot and emission loop that runs on a specified interval. type defaultExporter struct { - runState atomic.AtomicRunState + runState ocatomic.AtomicRunState ds core.DataSource snapshotProvider SnapshotProvider emitters []Emitter @@ -100,6 +101,7 @@ func (de *defaultExporter) Start(interval time.Duration) bool { } var emitTasks sync.WaitGroup + var emitErrors atomic.Int32 // sandbox each emitter.Emit() call to it's own goroutine and trap any panics that occur, logging // the error and emitter id @@ -107,6 +109,7 @@ func (de *defaultExporter) Start(interval time.Duration) bool { emitTasks.Go(func() { if err := emit(runContext, emitter, snapshot); err != nil { log.Errorf("[%s] failed to emit snapshot: %v", emitter.ID(), err) + emitErrors.Add(1) } }) } @@ -114,9 +117,11 @@ func (de *defaultExporter) Start(interval time.Duration) bool { // wait for all emit tasks to complete before continuing emitTasks.Wait() - // persist snapshot state after successful export to maintain window continuity across restarts - if err := de.snapshotProvider.PersistState(); err != nil { - log.Warnf("failed to persist snapshot state: %v", err) + // only persist snapshot state if all emitters succeeded to avoid recording timestamps for data that never shipped + if emitErrors.Load() == 0 { + if err := de.snapshotProvider.PersistState(); err != nil { + log.Warnf("failed to persist snapshot state: %v", err) + } } } }() diff --git a/pkg/emitter/snapshot.go b/pkg/emitter/snapshot.go index ae518b24..ecf3ca3a 100644 --- a/pkg/emitter/snapshot.go +++ b/pkg/emitter/snapshot.go @@ -34,6 +34,8 @@ type SnapshotProvider interface { // FIXME: (bolt) this should be fine to run on a much faster frequency with a non-promethues metrics querier. var metricsSummaryCacheDuration time.Duration = 5 * time.Minute +const snapshotStateFilename = "snapshot-state.json" + // SnapshotState represents the persisted state of the snapshot provider to maintain // continuity across restarts. type SnapshotState struct { @@ -78,7 +80,7 @@ func (csp *ConcurrentSnapshotProvider) persistState() error { return fmt.Errorf("scratch directory not configured") } - stateFile := filepath.Join(csp.config.ScratchDir, "snapshot-state.json") + stateFile := filepath.Join(csp.config.ScratchDir, snapshotStateFilename) state := SnapshotState{ LastSnapshot: csp.lastSnapshot, } @@ -107,7 +109,7 @@ func (csp *ConcurrentSnapshotProvider) recoverState() { return } - stateFile := filepath.Join(csp.config.ScratchDir, "snapshot-state.json") + stateFile := filepath.Join(csp.config.ScratchDir, snapshotStateFilename) data, err := os.ReadFile(stateFile) if err != nil { if os.IsNotExist(err) { diff --git a/pkg/emitter/snapshot_state_test.go b/pkg/emitter/snapshot_state_test.go new file mode 100644 index 00000000..d633ff81 --- /dev/null +++ b/pkg/emitter/snapshot_state_test.go @@ -0,0 +1,186 @@ +package emitter + +import ( + "encoding/json" + "os" + "path/filepath" + "testing" + "time" +) + +func TestSnapshotStatePersistAndRecover(t *testing.T) { + // Create a temporary directory for testing + tempDir, err := os.MkdirTemp("", "snapshot-state-test-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tempDir) + + // Create a snapshot config with the temp directory + config := &SnapshotConfig{ + ScratchDir: tempDir, + Now: defaultNow, + } + + // Create first provider and set a timestamp + provider1 := NewConcurrentSnapshotProvider(config).(*ConcurrentSnapshotProvider) + testTime := time.Date(2024, 1, 15, 10, 30, 0, 0, time.UTC) + provider1.lastSnapshot = testTime + + // Persist the state + if err := provider1.PersistState(); err != nil { + t.Fatalf("Failed to persist state: %v", err) + } + + // Verify the state file was created + stateFile := filepath.Join(tempDir, snapshotStateFilename) + if _, err := os.Stat(stateFile); os.IsNotExist(err) { + t.Fatalf("State file was not created") + } + + // Create a new provider and verify it recovers the state + provider2 := NewConcurrentSnapshotProvider(config).(*ConcurrentSnapshotProvider) + if !provider2.lastSnapshot.Equal(testTime) { + t.Errorf("Expected recovered timestamp %v, got %v", testTime, provider2.lastSnapshot) + } +} + +func TestSnapshotStateRecoverMissingFile(t *testing.T) { + // Create a temporary directory for testing + tempDir, err := os.MkdirTemp("", "snapshot-state-test-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tempDir) + + config := &SnapshotConfig{ + ScratchDir: tempDir, + Now: defaultNow, + } + + // Create provider without any existing state file + provider := NewConcurrentSnapshotProvider(config).(*ConcurrentSnapshotProvider) + + // Verify lastSnapshot is zero (cold start) + if !provider.lastSnapshot.IsZero() { + t.Errorf("Expected zero timestamp on cold start, got %v", provider.lastSnapshot) + } +} + +func TestSnapshotStateRecoverCorruptFile(t *testing.T) { + // Create a temporary directory for testing + tempDir, err := os.MkdirTemp("", "snapshot-state-test-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tempDir) + + // Write corrupt data to the state file + stateFile := filepath.Join(tempDir, snapshotStateFilename) + if err := os.WriteFile(stateFile, []byte("invalid json {{{"), 0644); err != nil { + t.Fatalf("Failed to write corrupt state file: %v", err) + } + + config := &SnapshotConfig{ + ScratchDir: tempDir, + Now: defaultNow, + } + + // Create provider with corrupt state file + provider := NewConcurrentSnapshotProvider(config).(*ConcurrentSnapshotProvider) + + // Verify lastSnapshot is zero (cold start due to corrupt file) + if !provider.lastSnapshot.IsZero() { + t.Errorf("Expected zero timestamp on corrupt file, got %v", provider.lastSnapshot) + } +} + +func TestSnapshotStatePersistWithoutScratchDir(t *testing.T) { + config := &SnapshotConfig{ + ScratchDir: "", // No scratch directory configured + Now: defaultNow, + } + + provider := NewConcurrentSnapshotProvider(config).(*ConcurrentSnapshotProvider) + provider.lastSnapshot = time.Now() + + // Attempt to persist should return an error + err := provider.PersistState() + if err == nil { + t.Error("Expected error when persisting without scratch directory, got nil") + } +} + +func TestSnapshotStateFileFormat(t *testing.T) { + // Create a temporary directory for testing + tempDir, err := os.MkdirTemp("", "snapshot-state-test-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tempDir) + + config := &SnapshotConfig{ + ScratchDir: tempDir, + Now: defaultNow, + } + + provider := NewConcurrentSnapshotProvider(config).(*ConcurrentSnapshotProvider) + testTime := time.Date(2024, 1, 15, 10, 30, 0, 0, time.UTC) + provider.lastSnapshot = testTime + + // Persist the state + if err := provider.PersistState(); err != nil { + t.Fatalf("Failed to persist state: %v", err) + } + + // Read and verify the file format + stateFile := filepath.Join(tempDir, snapshotStateFilename) + data, err := os.ReadFile(stateFile) + if err != nil { + t.Fatalf("Failed to read state file: %v", err) + } + + var state SnapshotState + if err := json.Unmarshal(data, &state); err != nil { + t.Fatalf("Failed to unmarshal state file: %v", err) + } + + if !state.LastSnapshot.Equal(testTime) { + t.Errorf("Expected timestamp %v in file, got %v", testTime, state.LastSnapshot) + } +} + +func TestSnapshotStateMultiplePersists(t *testing.T) { + // Create a temporary directory for testing + tempDir, err := os.MkdirTemp("", "snapshot-state-test-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tempDir) + + config := &SnapshotConfig{ + ScratchDir: tempDir, + Now: defaultNow, + } + + provider := NewConcurrentSnapshotProvider(config).(*ConcurrentSnapshotProvider) + + // Persist multiple times with different timestamps + time1 := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC) + provider.lastSnapshot = time1 + if err := provider.PersistState(); err != nil { + t.Fatalf("Failed to persist state (first): %v", err) + } + + time2 := time.Date(2024, 1, 15, 11, 0, 0, 0, time.UTC) + provider.lastSnapshot = time2 + if err := provider.PersistState(); err != nil { + t.Fatalf("Failed to persist state (second): %v", err) + } + + // Create new provider and verify it has the latest timestamp + provider2 := NewConcurrentSnapshotProvider(config).(*ConcurrentSnapshotProvider) + if !provider2.lastSnapshot.Equal(time2) { + t.Errorf("Expected recovered timestamp %v, got %v", time2, provider2.lastSnapshot) + } +} From e9c7b5ed150db90c1e093554e75fe6508c383ecb Mon Sep 17 00:00:00 2001 From: Warwick Date: Mon, 13 Apr 2026 14:13:49 -0500 Subject: [PATCH 3/3] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- pkg/emitter/snapshotconfig.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/pkg/emitter/snapshotconfig.go b/pkg/emitter/snapshotconfig.go index 35a67ef1..66ee8f9f 100644 --- a/pkg/emitter/snapshotconfig.go +++ b/pkg/emitter/snapshotconfig.go @@ -183,11 +183,14 @@ func defaultNow() time.Time { } // getScratchDir returns the scratch directory path from the SCRATCH_DIR environment variable, -// or the default path if not set. +// falling back to CLOUDABILITY_SCRATCH_DIR, or the default path if neither is set. func getScratchDir() string { if dir := os.Getenv("SCRATCH_DIR"); dir != "" { return dir } + if dir := os.Getenv("CLOUDABILITY_SCRATCH_DIR"); dir != "" { + return dir + } return "/opt/finops-agent" }