diff --git a/internal/monitoring/mock_metrics_history.go b/internal/monitoring/mock_metrics_history.go index 117ef3e8d..19242c621 100644 --- a/internal/monitoring/mock_metrics_history.go +++ b/internal/monitoring/mock_metrics_history.go @@ -6,6 +6,7 @@ import ( "hash/fnv" "math" "math/rand" + "sort" "strings" "time" @@ -27,11 +28,17 @@ const ( type mockMetricsSamplerConfig struct { SeedDuration time.Duration SampleInterval time.Duration + // SeedStore opts the sampler into persisting mock history to the metrics + // store. Only the dev harness sets it (alongside pointing PULSE_DATA_DIR + // at the isolated tmp/mock-data dir), because seeded mock rows must never + // land in a metrics.db that also holds real history. + SeedStore bool } func mockMetricsSamplerConfigFromEnv() mockMetricsSamplerConfig { seedDuration := parseDurationEnv("PULSE_MOCK_TRENDS_SEED_DURATION", defaultMockSeedDuration) sampleInterval := parseDurationEnv("PULSE_MOCK_TRENDS_SAMPLE_INTERVAL", defaultMockSampleInterval) + seedStore := parseBoolEnv("PULSE_MOCK_SEED_METRICS_STORE", false) // Guardrails to keep memory and CPU bounded in demo mode. if seedDuration < 5*time.Minute { @@ -55,9 +62,66 @@ func mockMetricsSamplerConfigFromEnv() mockMetricsSamplerConfig { return mockMetricsSamplerConfig{ SeedDuration: seedDuration, SampleInterval: sampleInterval, + SeedStore: seedStore, } } +// Store-tier backfill plans for mock mode. Each tier is seeded over its own +// window at its own spacing: report queries read exactly one tier (picked by +// range, falling back only when it is empty), so each tier needs enough rows +// of its own for charts without exploding total row count: +// +// daily: 30d window, 4h spacing -> 180 rows/series, serves >7d queries +// hourly: 7d window, 2h spacing -> 84 rows/series, serves 24h-7d queries +// minute: 24h window, 15m spacing -> 96 rows/series, serves 2h-24h queries +// +// That is 360 rows/series; at the default mock estate (~750 series) the seed +// stays under ~100MB of SQLite. The raw tier is left to the live mock tick. +// Mock mode extends hourly/daily retention to 90d (see monitoring.New), so +// seeded rows outlive their windows. Timestamps sit on the spacing grid, so +// restarts target the same rows (the store upserts on timestamp+tier) and +// only the gap since the previous boot is written. +type mockStoreSeedPlan struct { + tier metrics.Tier + window time.Duration + spacing time.Duration +} + +var mockStoreSeedPlans = []mockStoreSeedPlan{ + {tier: metrics.TierDaily, window: 30 * 24 * time.Hour, spacing: 4 * time.Hour}, + {tier: metrics.TierHourly, window: 7 * 24 * time.Hour, spacing: 2 * time.Hour}, + {tier: metrics.TierMinute, window: 24 * time.Hour, spacing: 15 * time.Minute}, +} + +const mockStoreSeedBatchSize = 8192 + +// mockStoreSeedTimestamps returns the spacing-aligned timestamps covering +// [now-window, now], oldest first. +func mockStoreSeedTimestamps(now time.Time, window, spacing time.Duration) []time.Time { + if window <= 0 || spacing <= 0 { + return nil + } + windowStart := now.Add(-window) + start := windowStart.Truncate(spacing) + if start.Before(windowStart) { + start = start.Add(spacing) + } + var out []time.Time + for ts := start; !ts.After(now); ts = ts.Add(spacing) { + out = append(out, ts) + } + return out +} + +// timestampsAfter returns the suffix of the sorted slice strictly after the +// given time. +func timestampsAfter(timestamps []time.Time, after time.Time) []time.Time { + idx := sort.Search(len(timestamps), func(i int) bool { + return timestamps[i].After(after) + }) + return timestamps[idx:] +} + // HashSeed produces a deterministic uint64 from the given parts. func HashSeed(parts ...string) uint64 { h := fnv.New64a() @@ -437,35 +501,103 @@ func seedMockMetricsHistory(mh *MetricsHistory, ms *metrics.Store, graph mock.Fi // 2h–24h: 2min intervals (~660 points) // 24h–90d: ~65min intervals (~1920 points) seedTimestamps := buildTieredTimestamps(now, seedDuration) - const seedBatchSize = 5000 numPoints := len(seedTimestamps) - var seedBatch []metrics.WriteMetric - queueMetric := func(resourceType, resourceID, metricType string, value float64, ts time.Time) { - if ms == nil { + + // Store backfill state: aligned timestamps per tier plan (capped by the + // seed duration) plus existing coverage so restarts only fill the gap. + storeSeedStart := time.Now() + storeRows := 0 + var storePlanTimestamps map[metrics.Tier][]time.Time + var storeCoverage map[metrics.Tier]map[metrics.SeriesKey]time.Time + var storeBatch []metrics.WriteMetric + if ms != nil { + storePlanTimestamps = make(map[metrics.Tier][]time.Time, len(mockStoreSeedPlans)) + storeCoverage = make(map[metrics.Tier]map[metrics.SeriesKey]time.Time, len(mockStoreSeedPlans)) + for _, plan := range mockStoreSeedPlans { + window := plan.window + if window > seedDuration { + window = seedDuration + } + storePlanTimestamps[plan.tier] = mockStoreSeedTimestamps(now, window, plan.spacing) + coverage, err := ms.MaxTimestampsForTier(plan.tier) + if err != nil { + log.Warn().Err(err).Str("tier", string(plan.tier)).Msg("mock seeding: failed to read store coverage; reseeding full window") + coverage = nil + } + storeCoverage[plan.tier] = coverage + } + } + + flushStoreBatch := func() { + if ms == nil || len(storeBatch) == 0 { return } - seedBatch = append(seedBatch, - metrics.WriteMetric{ - ResourceType: resourceType, - ResourceID: resourceID, - MetricType: metricType, - Value: value, - Timestamp: ts, - Tier: metrics.TierHourly, - }, - metrics.WriteMetric{ - ResourceType: resourceType, - ResourceID: resourceID, - MetricType: metricType, - Value: value, - Timestamp: ts, - Tier: metrics.TierDaily, - }, - ) + ms.WriteBatchSync(storeBatch) + storeBatch = storeBatch[:0] + } - if len(seedBatch) >= seedBatchSize { - ms.WriteBatchSync(seedBatch) - seedBatch = seedBatch[:0] + queueStorePoint := func(resourceType, resourceID, metricType string, value float64, ts time.Time, tier metrics.Tier) { + storeBatch = append(storeBatch, metrics.WriteMetric{ + ResourceType: resourceType, + ResourceID: resourceID, + MetricType: metricType, + Value: value, + Timestamp: ts, + Tier: tier, + }) + storeRows++ + if len(storeBatch) >= mockStoreSeedBatchSize { + flushStoreBatch() + } + } + + // seedStoreGapTimestamps returns the plan timestamps still missing for the + // series whose coverage is tracked under coverageKey. + seedStoreGapTimestamps := func(plan mockStoreSeedPlan, coverageKey metrics.SeriesKey) []time.Time { + timestamps := storePlanTimestamps[plan.tier] + if len(timestamps) == 0 { + return nil + } + if coverage := storeCoverage[plan.tier]; coverage != nil { + if maxTs, ok := coverage[coverageKey]; ok { + timestamps = timestampsAfter(timestamps, maxTs) + } + } + return timestamps + } + + seedStoreSeries := func(resourceType, resourceID string, metricNames ...string) { + if ms == nil || strings.TrimSpace(resourceID) == "" { + return + } + for _, metricType := range metricNames { + for _, plan := range mockStoreSeedPlans { + coverageKey := metrics.NormalizedSeriesKey(resourceType, resourceID, metricType) + for _, ts := range seedStoreGapTimestamps(plan, coverageKey) { + queueStorePoint(resourceType, resourceID, metricType, mock.SampleMetric(resourceType, resourceID, metricType, ts), ts, plan.tier) + } + } + } + } + + // seedStoreStorageSeries derives the used/avail/total companions from the + // sampled usage percentage, mirroring how live storage ticks persist + // capacity metrics. + seedStoreStorageSeries := func(storageID string, currentTotal float64) { + if ms == nil || strings.TrimSpace(storageID) == "" || currentTotal <= 0 { + return + } + for _, plan := range mockStoreSeedPlans { + coverageKey := metrics.NormalizedSeriesKey("storage", storageID, "usage") + for _, ts := range seedStoreGapTimestamps(plan, coverageKey) { + usage := clampFloat(mock.SampleMetric("storage", storageID, "usage", ts), 0, 100) + used := currentTotal * (usage / 100.0) + avail := math.Max(0, currentTotal-used) + queueStorePoint("storage", storageID, "usage", usage, ts, plan.tier) + queueStorePoint("storage", storageID, "used", used, ts, plan.tier) + queueStorePoint("storage", storageID, "avail", avail, ts, plan.tier) + queueStorePoint("storage", storageID, "total", currentTotal, ts, plan.tier) + } } } recordStorageTimeline := func(storageID string, currentTotal float64) { @@ -483,12 +615,8 @@ func seedMockMetricsHistory(mh *MetricsHistory, ms *metrics.Store, graph mock.Fi mh.AddStorageMetric(storageID, "used", used, ts) mh.AddStorageMetric(storageID, "avail", avail, ts) mh.AddStorageMetric(storageID, "total", currentTotal, ts) - queueMetric("storage", storageID, "usage", usage, ts) - queueMetric("storage", storageID, "used", used, ts) - queueMetric("storage", storageID, "avail", avail, ts) - queueMetric("storage", storageID, "total", currentTotal, ts) } - + seedStoreStorageSeries(storageID, currentTotal) } recordNode := func(node models.Node) { @@ -505,11 +633,8 @@ func seedMockMetricsHistory(mh *MetricsHistory, ms *metrics.Store, graph mock.Fi mh.AddNodeMetric(node.ID, "cpu", cpuSeries[i], ts) mh.AddNodeMetric(node.ID, "memory", memSeries[i], ts) mh.AddNodeMetric(node.ID, "disk", diskSeries[i], ts) - queueMetric("node", node.ID, "cpu", cpuSeries[i], ts) - queueMetric("node", node.ID, "memory", memSeries[i], ts) - queueMetric("node", node.ID, "disk", diskSeries[i], ts) } - + seedStoreSeries("node", node.ID, "cpu", "memory", "disk") } recordGuest := func( @@ -560,32 +685,36 @@ func seedMockMetricsHistory(mh *MetricsHistory, ms *metrics.Store, graph mock.Fi mh.AddGuestMetric(metricID, "cpu", cpuSeries[i], ts) mh.AddGuestMetric(metricID, "memory", memSeries[i], ts) } - queueMetric(storeType, storeID, "cpu", cpuSeries[i], ts) - queueMetric(storeType, storeID, "memory", memSeries[i], ts) if includeDisk { for _, metricID := range uniqueMetricIDs { mh.AddGuestMetric(metricID, "disk", diskSeries[i], ts) } - queueMetric(storeType, storeID, "disk", diskSeries[i], ts) } if includeDiskIO { for _, metricID := range uniqueMetricIDs { mh.AddGuestMetric(metricID, "diskread", diskReadSeries[i], ts) mh.AddGuestMetric(metricID, "diskwrite", diskWriteSeries[i], ts) } - queueMetric(storeType, storeID, "diskread", diskReadSeries[i], ts) - queueMetric(storeType, storeID, "diskwrite", diskWriteSeries[i], ts) } if includeNetwork { for _, metricID := range uniqueMetricIDs { mh.AddGuestMetric(metricID, "netin", netInSeries[i], ts) mh.AddGuestMetric(metricID, "netout", netOutSeries[i], ts) } - queueMetric(storeType, storeID, "netin", netInSeries[i], ts) - queueMetric(storeType, storeID, "netout", netOutSeries[i], ts) } } + storeMetrics := []string{"cpu", "memory"} + if includeDisk { + storeMetrics = append(storeMetrics, "disk") + } + if includeDiskIO { + storeMetrics = append(storeMetrics, "diskread", "diskwrite") + } + if includeNetwork { + storeMetrics = append(storeMetrics, "netin", "netout") + } + seedStoreSeries(storeType, storeID, storeMetrics...) } log.Debug().Int("count", len(state.Nodes)).Msg("mock seeding: processing nodes") @@ -693,11 +822,8 @@ func seedMockMetricsHistory(mh *MetricsHistory, ms *metrics.Store, graph mock.Fi mh.AddDiskMetric(resourceID, "disk", busySeries[i], ts) mh.AddDiskMetric(resourceID, "diskread", diskReadSeries[i], ts) mh.AddDiskMetric(resourceID, "diskwrite", diskWriteSeries[i], ts) - queueMetric("disk", resourceID, "smart_temp", tempSeries[i], ts) - queueMetric("disk", resourceID, "disk", busySeries[i], ts) - queueMetric("disk", resourceID, "diskread", diskReadSeries[i], ts) - queueMetric("disk", resourceID, "diskwrite", diskWriteSeries[i], ts) } + seedStoreSeries("disk", resourceID, "smart_temp", "disk", "diskread", "diskwrite") } log.Debug().Int("count", len(state.PhysicalDisks)).Msg("mock seeding: processing physical disks") @@ -725,12 +851,7 @@ func seedMockMetricsHistory(mh *MetricsHistory, ms *metrics.Store, graph mock.Fi continue } - usageSeries := canonicalMetricSeries("ceph", cephID, "usage", seedTimestamps) - for i := 0; i < numPoints; i++ { - ts := seedTimestamps[i] - queueMetric("ceph", cephID, "usage", usageSeries[i], ts) - } - + seedStoreSeries("ceph", cephID, "usage") } log.Debug().Int("count", len(state.DockerHosts)).Msg("mock seeding: processing docker hosts") @@ -893,8 +1014,12 @@ func seedMockMetricsHistory(mh *MetricsHistory, ms *metrics.Store, graph mock.Fi recordStorageTimeline(sourceID, float64(total)) } - if ms != nil && len(seedBatch) > 0 { - ms.WriteBatchSync(seedBatch) + flushStoreBatch() + if ms != nil { + log.Info(). + Int("rows", storeRows). + Dur("took", time.Since(storeSeedStart)). + Msg("mock seeding: metrics store backfill completed") } log.Debug().Msg("mock seeding: completed") } @@ -1535,9 +1660,21 @@ func (m *Monitor) startMockMetricsSampler(ctx context.Context) { Dur("seedDuration", seedDuration). Dur("sampleInterval", cfg.SampleInterval). Msg("Mock metrics sampler: seeding historical data") - // Keep mock trend generation in-memory only so production history in the - // persistent metrics store remains untouched while mock mode is active. - seedMockMetricsHistory(m.metricsHistory, nil, graph, normalizeMockMetricTimestamp(time.Now(), cfg.SampleInterval), seedDuration, cfg.SampleInterval) + // Persist mock history into the metrics store only when the dev harness + // opted in via PULSE_MOCK_SEED_METRICS_STORE — scripts/hot-dev.sh and + // scripts/toggle-mock.sh set it exactly where they point PULSE_DATA_DIR at + // the isolated tmp/mock-data dir. Without the opt-in (e.g. a production + // install flipping PULSE_MOCK_MODE on its real data dir) the store may + // hold real history, which seeded mock rows must never touch, so trend + // generation stays in-memory only. + var storeSink *metrics.Store + if cfg.SeedStore { + storeSink = m.metricsStore + if storeSink == nil { + log.Warn().Msg("PULSE_MOCK_SEED_METRICS_STORE is set but the metrics store is unavailable; mock report history will stay empty") + } + } + seedMockMetricsHistory(m.metricsHistory, storeSink, graph, normalizeMockMetricTimestamp(time.Now(), cfg.SampleInterval), seedDuration, cfg.SampleInterval) m.invalidateMockChartCaches() m.prewarmMockDashboardChartCaches() @@ -1556,7 +1693,7 @@ func (m *Monitor) startMockMetricsSampler(ctx context.Context) { if !mock.IsMockEnabled() { continue } - recordMockStateToMetricsHistory(m.metricsHistory, nil, mock.CurrentFixtureGraph(), normalizeMockMetricTimestamp(time.Now(), cfg.SampleInterval)) + recordMockStateToMetricsHistory(m.metricsHistory, storeSink, mock.CurrentFixtureGraph(), normalizeMockMetricTimestamp(time.Now(), cfg.SampleInterval)) m.invalidateMockChartCaches() m.prewarmMockDashboardChartCaches() } diff --git a/internal/monitoring/mock_metrics_history_test.go b/internal/monitoring/mock_metrics_history_test.go index 052072277..c191ea566 100644 --- a/internal/monitoring/mock_metrics_history_test.go +++ b/internal/monitoring/mock_metrics_history_test.go @@ -23,6 +23,10 @@ func setMockSamplerTestEnv(t *testing.T, seedDuration, sampleInterval time.Durat t.Helper() t.Setenv("PULSE_MOCK_TRENDS_SEED_DURATION", seedDuration.String()) t.Setenv("PULSE_MOCK_TRENDS_SAMPLE_INTERVAL", sampleInterval.String()) + // Default to the production-safe path even when the dev shell (hot-dev + // exports this) leaks the store-seeding opt-in into the test run; tests + // that exercise the opt-in re-set it to true explicitly. + t.Setenv("PULSE_MOCK_SEED_METRICS_STORE", "false") } func compactMockFixtureConfig() mock.MockConfig { @@ -43,34 +47,11 @@ func compactMockFixtureConfig() mock.MockConfig { const boundedMockHistoryProofWindow = 4 * time.Hour -func expectedCanonicalStoreBucketAverage(t *testing.T, resourceType, resourceID, metricType string, now time.Time, seedDuration, interval time.Duration, bucket time.Time, stepSecs int64) float64 { - t.Helper() - if stepSecs <= 0 { - t.Fatalf("expected positive step seconds, got %d", stepSecs) - } - - alignedNow := normalizeMockMetricTimestamp(now, interval) - startUnix := now.Add(-seedDuration).Unix() - endUnix := now.Unix() - bucketUnix := bucket.Unix() - var sum float64 - var count int - - for _, ts := range buildTieredTimestamps(alignedNow, seedDuration) { - tsUnix := ts.Unix() - if tsUnix < startUnix || tsUnix > endUnix { - continue - } - if (tsUnix/stepSecs)*stepSecs+(stepSecs/2) != bucketUnix { - continue - } - sum += mock.SampleMetric(resourceType, resourceID, metricType, ts) - count++ - } - if count == 0 { - t.Fatalf("expected seeded samples in %s bucket at %s", metricType, bucket.Format(time.RFC3339)) - } - return sum / float64(count) +// expectedCanonicalStoreUsageSample returns the value the store seeder writes +// for a storage usage series at the given aligned timestamp: the canonical +// mock runtime sample, clamped to a valid percentage. +func expectedCanonicalStoreUsageSample(resourceID string, ts time.Time) float64 { + return clampFloat(mock.SampleMetric("storage", resourceID, "usage", ts), 0, 100) } func compactMockChartFixtureConfig() mock.MockConfig { @@ -482,6 +463,82 @@ func TestSeedMockMetricsHistory_SeedsMetricsStore(t *testing.T) { } } +func TestSeedMockMetricsHistory_StoreSeedGapFillsWithoutDuplicates(t *testing.T) { + now := time.Now().UTC().Truncate(time.Minute) + seedDuration := boundedMockHistoryProofWindow + + state := models.StateSnapshot{ + VMs: []models.VM{ + { + ID: "mock-cluster-pve1-100", + Status: "running", + CPU: 0.21, + Memory: models.Memory{Usage: 47, Total: 8 * 1024 * 1024 * 1024}, + Disk: models.Disk{Usage: 28, Total: 1024, Used: 256}, + }, + }, + } + + cfg := metrics.DefaultConfig(t.TempDir()) + cfg.RetentionRaw = 90 * 24 * time.Hour + cfg.RetentionMinute = 90 * 24 * time.Hour + cfg.RetentionHourly = 90 * 24 * time.Hour + cfg.RetentionDaily = 90 * 24 * time.Hour + cfg.WriteBufferSize = 500 + + store, err := metrics.NewStore(cfg) + if err != nil { + t.Fatalf("failed to create metrics store: %v", err) + } + defer store.Close() + + // The query window stays under 24h so it always reads the minute tier, + // which has the densest grid (10m spacing) and therefore the most + // sensitive duplicate / gap-fill signal. + countPoints := func(end time.Time) int { + t.Helper() + points, err := store.Query("vm", "mock-cluster-pve1-100", "cpu", now.Add(-seedDuration-time.Hour), end, 0) + if err != nil { + t.Fatalf("failed to query seeded vm cpu points: %v", err) + } + return len(points) + } + + seedMockMetricsHistory(NewMetricsHistory(1000, seedDuration), store, fixtureGraphWithState(state), now, seedDuration, time.Minute) + first := countPoints(now) + if first == 0 { + t.Fatal("expected initial store seed to write minute-tier points") + } + + // Re-seeding at the same instant must not add rows: timestamps sit on the + // tier spacing grids and existing coverage short-circuits the backfill. + seedMockMetricsHistory(NewMetricsHistory(1000, seedDuration), store, fixtureGraphWithState(state), now, seedDuration, time.Minute) + if got := countPoints(now); got != first { + t.Fatalf("expected idempotent reseed to keep %d points, got %d", first, got) + } + + // A later boot fills only the gap between previous coverage and the new + // now: exactly the minute-tier grid points inside (now, later]. + later := now.Add(30 * time.Minute) + var minutePlan mockStoreSeedPlan + for _, plan := range mockStoreSeedPlans { + if plan.tier == metrics.TierMinute { + minutePlan = plan + } + } + if minutePlan.spacing <= 0 { + t.Fatal("expected a minute-tier store seed plan") + } + gapPoints := len(timestampsAfter(mockStoreSeedTimestamps(later, seedDuration, minutePlan.spacing), now)) + if gapPoints == 0 { + t.Fatal("expected the advanced clock to expose new minute-tier grid points") + } + seedMockMetricsHistory(NewMetricsHistory(1000, seedDuration), store, fixtureGraphWithState(state), later, seedDuration, time.Minute) + if got, want := countPoints(later), first+gapPoints; got != want { + t.Fatalf("expected gap fill to append exactly the missing grid points (%d), got %d", want, got) + } +} + func TestSeedMockMetricsHistory_SeedsDiskTemperatureMetricsStore(t *testing.T) { now := time.Now() seedDuration := boundedMockHistoryProofWindow @@ -566,7 +623,7 @@ func TestSeedMockMetricsHistory_SeedsVMwareMetricsStore(t *testing.T) { t.Fatal("expected metrics store to have seeded VMware VM cpu points") } - storagePoints, err := store.Query("storage", "vc-mock-1:datastore:datastore-201", "usage", now.Add(-seedDuration), now, 3600) + storagePoints, err := store.Query("storage", "vc-mock-1:datastore:datastore-201", "usage", now.Add(-seedDuration), now, 0) if err != nil { t.Fatalf("failed to query VMware datastore usage metrics: %v", err) } @@ -574,8 +631,8 @@ func TestSeedMockMetricsHistory_SeedsVMwareMetricsStore(t *testing.T) { t.Fatal("expected metrics store to have seeded VMware datastore usage points") } lastStoragePoint := storagePoints[len(storagePoints)-1] - if got, want := lastStoragePoint.Value, expectedCanonicalStoreBucketAverage(t, "storage", "vc-mock-1:datastore:datastore-201", "usage", now, seedDuration, time.Minute, lastStoragePoint.Timestamp, 3600); math.Abs(got-want) > 1e-9 { - t.Fatalf("expected VMware datastore usage seed at %s to match canonical downsample bucket, got=%v want=%v", lastStoragePoint.Timestamp.Format(time.RFC3339), got, want) + if got, want := lastStoragePoint.Value, expectedCanonicalStoreUsageSample("vc-mock-1:datastore:datastore-201", lastStoragePoint.Timestamp); math.Abs(got-want) > 1e-9 { + t.Fatalf("expected VMware datastore usage seed at %s to match canonical mock runtime sample, got=%v want=%v", lastStoragePoint.Timestamp.Format(time.RFC3339), got, want) } storageUsedPoints, err := store.Query("storage", "vc-mock-1:datastore:datastore-201", "used", now.Add(-seedDuration), now, 3600) @@ -638,7 +695,7 @@ func TestSeedMockMetricsHistory_SeedsTrueNASMetricsStore(t *testing.T) { } datasetID := mock.TrueNASDatasetMetricID(fixtures.System.Hostname, fixtures.Datasets[0].Name) - datasetPoints, err := store.Query("storage", datasetID, "usage", now.Add(-seedDuration), now, 3600) + datasetPoints, err := store.Query("storage", datasetID, "usage", now.Add(-seedDuration), now, 0) if err != nil { t.Fatalf("failed to query canonical TrueNAS dataset usage metrics: %v", err) } @@ -646,8 +703,8 @@ func TestSeedMockMetricsHistory_SeedsTrueNASMetricsStore(t *testing.T) { t.Fatal("expected metrics store to have seeded canonical TrueNAS dataset usage points") } lastDatasetPoint := datasetPoints[len(datasetPoints)-1] - if got, want := lastDatasetPoint.Value, expectedCanonicalStoreBucketAverage(t, "storage", datasetID, "usage", now, seedDuration, time.Minute, lastDatasetPoint.Timestamp, 3600); math.Abs(got-want) > 1e-9 { - t.Fatalf("expected TrueNAS dataset usage seed at %s to match canonical downsample bucket, got=%v want=%v", lastDatasetPoint.Timestamp.Format(time.RFC3339), got, want) + if got, want := lastDatasetPoint.Value, expectedCanonicalStoreUsageSample(datasetID, lastDatasetPoint.Timestamp); math.Abs(got-want) > 1e-9 { + t.Fatalf("expected TrueNAS dataset usage seed at %s to match canonical mock runtime sample, got=%v want=%v", lastDatasetPoint.Timestamp.Format(time.RFC3339), got, want) } poolID := mock.TrueNASPoolMetricID(fixtures.System.Hostname, fixtures.Pools[0].Name) @@ -826,6 +883,82 @@ func TestStartMockMetricsSampler_DoesNotClearExistingMetricsStoreData(t *testing if !found { t.Fatal("expected to find original production metric value after mock sampler start") } + + // Without the PULSE_MOCK_SEED_METRICS_STORE opt-in the sampler must not + // persist any mock series to the store. + graph := mock.CurrentFixtureGraph() + if len(graph.State.Nodes) == 0 { + t.Fatal("expected mock fixture graph to include a node") + } + mockPoints, err := store.Query("node", graph.State.Nodes[0].ID, "cpu", now.Add(-48*time.Hour), now.Add(time.Hour), 0) + if err != nil { + t.Fatalf("failed to query mock node metrics: %v", err) + } + if len(mockPoints) != 0 { + t.Fatalf("expected no mock node store history without the seed opt-in, got %d points", len(mockPoints)) + } +} + +func TestStartMockMetricsSampler_SeedsMetricsStoreWhenOptedIn(t *testing.T) { + setMockSamplerTestEnv(t, time.Hour, 5*time.Minute) + t.Setenv("PULSE_MOCK_SEED_METRICS_STORE", "true") + + previousEnabled := mock.IsMockEnabled() + previousConfig := mock.GetConfig() + t.Cleanup(func() { + mustSetMockEnabled(t, false) + mock.SetMockConfig(previousConfig) + if previousEnabled { + mustSetMockEnabled(t, true) + mock.SetMockConfig(previousConfig) + } + }) + + cfg := metrics.DefaultConfig(t.TempDir()) + cfg.RetentionRaw = 90 * 24 * time.Hour + cfg.RetentionMinute = 90 * 24 * time.Hour + cfg.RetentionHourly = 90 * 24 * time.Hour + cfg.RetentionDaily = 90 * 24 * time.Hour + cfg.WriteBufferSize = 100 + + store, err := metrics.NewStore(cfg) + if err != nil { + t.Fatalf("failed to create metrics store: %v", err) + } + defer store.Close() + + mockCfg := compactMockFixtureConfig() + mockCfg.VMsPerNode = 1 + + mustSetMockEnabled(t, false) + mock.SetMockConfig(mockCfg) + mustSetMockEnabled(t, true) + + monitor := &Monitor{ + metricsHistory: NewMetricsHistory(1000, 24*time.Hour), + metricsStore: store, + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + monitor.startMockMetricsSampler(ctx) + t.Cleanup(func() { monitor.stopMockMetricsSampler() }) + + graph := mock.CurrentFixtureGraph() + if len(graph.State.VMs) == 0 { + t.Fatal("expected mock fixture graph to include a PVE guest") + } + vmID := graph.State.VMs[0].ID + + now := time.Now() + points, err := store.Query("vm", vmID, "cpu", now.Add(-2*time.Hour), now, 0) + if err != nil { + t.Fatalf("failed to query mock PVE guest store history: %v", err) + } + if len(points) < 2 { + t.Fatalf("expected seeded mock PVE guest store history when opted in, got %d points", len(points)) + } } func TestStartAndStopMockMetricsSampler_ClearStaleMockChartCaches(t *testing.T) { diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 9d66dd58e..9f07024de 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -1447,11 +1447,12 @@ func New(cfg *config.Config) (*Monitor, error) { metricsStoreConfig.RetentionDaily = time.Duration(cfg.MetricsRetentionDailyDays) * 24 * time.Hour } - // In mock mode, extend hourly/daily retention to 90 days to match the - // seeded data range (seeds write directly to hourly+daily tiers). - // Raw and minute tiers keep production defaults — seeded data doesn't - // use them, and live mock ticks at 2s intervals would bloat the DB - // (the old 90-day raw retention caused metrics.db to grow to ~2 GB). + // In mock mode, extend hourly/daily retention to 90 days so seeded mock + // backfill (daily tier 30d, hourly tier 7d — see mockStoreSeedPlans) + // outlives its windows. Raw and minute tiers keep production defaults — + // the minute seed window (24h) matches the default retention, and short + // raw retention keeps live mock ticks from bloating the DB (the old + // 90-day raw retention caused metrics.db to grow to ~2 GB). if mock.IsMockEnabled() { metricsStoreConfig.WriteBufferSize = 2000 metricsStoreConfig.RetentionHourly = 90 * 24 * time.Hour diff --git a/internal/monitoring/monitor_helpers.go b/internal/monitoring/monitor_helpers.go index 80ebdb31a..e1151b996 100644 --- a/internal/monitoring/monitor_helpers.go +++ b/internal/monitoring/monitor_helpers.go @@ -61,6 +61,24 @@ func makeGuestID(instanceName string, node string, vmid int) string { return fmt.Sprintf("%s:%s:%d", instanceName, node, vmid) } +// parseBoolEnv parses a boolean from an environment variable, returning defaultVal if not set or invalid +func parseBoolEnv(key string, defaultVal bool) bool { + val := strings.TrimSpace(os.Getenv(key)) + if val == "" { + return defaultVal + } + parsed, err := strconv.ParseBool(val) + if err != nil { + log.Warn(). + Str("key", key). + Str("value", val). + Bool("default", defaultVal). + Msg("Failed to parse boolean from environment variable, using default") + return defaultVal + } + return parsed +} + // parseDurationEnv parses a duration from an environment variable, returning defaultVal if not set or invalid func parseDurationEnv(key string, defaultVal time.Duration) time.Duration { val := os.Getenv(key) diff --git a/pkg/metrics/store.go b/pkg/metrics/store.go index 2e03a8f2d..86bf29573 100644 --- a/pkg/metrics/store.go +++ b/pkg/metrics/store.go @@ -158,6 +158,24 @@ type WriteMetric struct { Tier Tier } +// SeriesKey identifies one stored metric series in the normalized form the +// write path persists (lower-cased resource/metric types, trimmed id). +type SeriesKey struct { + ResourceType string + ResourceID string + MetricType string +} + +// NormalizedSeriesKey builds the SeriesKey the write path would store for the +// given identifiers, so callers can match MaxTimestampsForTier results. +func NormalizedSeriesKey(resourceType, resourceID, metricType string) SeriesKey { + return SeriesKey{ + ResourceType: normalizeMetricResourceType(resourceType), + ResourceID: normalizeMetricIdentifier(resourceID), + MetricType: normalizeMetricType(metricType), + } +} + type maintenanceRequest struct { run func() done chan struct{} @@ -1707,6 +1725,33 @@ func (s *Store) getMaxTimestampForTier(tier Tier) (int64, bool) { return maxTs.Int64, true } +// MaxTimestampsForTier returns the newest stored timestamp for every series +// present in the given tier. Backfill seeders use it to write only the gap +// between existing coverage and now instead of re-writing whole windows. +func (s *Store) MaxTimestampsForTier(tier Tier) (map[SeriesKey]time.Time, error) { + rows, err := s.db.Query(` + SELECT resource_type, resource_id, metric_type, MAX(timestamp) + FROM metrics + WHERE tier = ? + GROUP BY resource_type, resource_id, metric_type + `, string(tier)) + if err != nil { + return nil, fmt.Errorf("failed to query metrics coverage for tier %q: %w", tier, err) + } + defer rows.Close() + + coverage := make(map[SeriesKey]time.Time) + for rows.Next() { + var key SeriesKey + var ts int64 + if err := rows.Scan(&key.ResourceType, &key.ResourceID, &key.MetricType, &ts); err != nil { + return nil, fmt.Errorf("failed to scan metrics coverage row: %w", err) + } + coverage[key] = time.Unix(ts, 0).UTC() + } + return coverage, rows.Err() +} + // runRetention deletes data older than retention period func (s *Store) runRetention() { start := time.Now() diff --git a/pkg/metrics/store_series_coverage_test.go b/pkg/metrics/store_series_coverage_test.go new file mode 100644 index 000000000..d0165bf9a --- /dev/null +++ b/pkg/metrics/store_series_coverage_test.go @@ -0,0 +1,49 @@ +package metrics + +import ( + "testing" + "time" +) + +func TestMaxTimestampsForTier(t *testing.T) { + store, err := NewStore(DefaultConfig(t.TempDir())) + if err != nil { + t.Fatalf("failed to create metrics store: %v", err) + } + defer store.Close() + + base := time.Now().UTC().Truncate(time.Hour) + store.WriteBatchSync([]WriteMetric{ + {ResourceType: "vm", ResourceID: "vm-1", MetricType: "cpu", Value: 1, Timestamp: base.Add(-2 * time.Hour), Tier: TierHourly}, + {ResourceType: "vm", ResourceID: "vm-1", MetricType: "cpu", Value: 2, Timestamp: base, Tier: TierHourly}, + {ResourceType: " VM ", ResourceID: " vm-2 ", MetricType: " CPU ", Value: 3, Timestamp: base.Add(-time.Hour), Tier: TierHourly}, + {ResourceType: "vm", ResourceID: "vm-1", MetricType: "cpu", Value: 4, Timestamp: base.Add(time.Hour), Tier: TierRaw}, + }) + + coverage, err := store.MaxTimestampsForTier(TierHourly) + if err != nil { + t.Fatalf("failed to read hourly coverage: %v", err) + } + if len(coverage) != 2 { + t.Fatalf("expected 2 hourly series, got %d: %v", len(coverage), coverage) + } + if got := coverage[NormalizedSeriesKey("vm", "vm-1", "cpu")]; !got.Equal(base) { + t.Fatalf("expected vm-1 hourly coverage at %s, got %s", base, got) + } + // Coverage keys are stored normalized, so messy caller identifiers must + // resolve through NormalizedSeriesKey. + if got := coverage[NormalizedSeriesKey(" VM ", " vm-2 ", " CPU ")]; !got.Equal(base.Add(-time.Hour)) { + t.Fatalf("expected normalized vm-2 hourly coverage at %s, got %s", base.Add(-time.Hour), got) + } + + rawCoverage, err := store.MaxTimestampsForTier(TierRaw) + if err != nil { + t.Fatalf("failed to read raw coverage: %v", err) + } + if len(rawCoverage) != 1 { + t.Fatalf("expected 1 raw series, got %d: %v", len(rawCoverage), rawCoverage) + } + if got := rawCoverage[NormalizedSeriesKey("vm", "vm-1", "cpu")]; !got.Equal(base.Add(time.Hour)) { + t.Fatalf("expected vm-1 raw coverage at %s, got %s", base.Add(time.Hour), got) + } +} diff --git a/scripts/hot-dev.sh b/scripts/hot-dev.sh index 174407a90..e923eea8c 100755 --- a/scripts/hot-dev.sh +++ b/scripts/hot-dev.sh @@ -381,6 +381,10 @@ else if [[ "${HOT_DEV_MOCK_FLAG}" == "true" ]]; then export PULSE_DATA_DIR="${ROOT_DIR}/tmp/mock-data" mkdir -p "$PULSE_DATA_DIR" + # The mock data dir holds no real history, so the backend may safely + # backfill mock metrics history into its metrics.db (powers reports + # and long-range charts in mock mode). + export PULSE_MOCK_SEED_METRICS_STORE=true log_info "Mock mode: using isolated mock data directory: ${PULSE_DATA_DIR}" else DEV_CONFIG_DIR="${ROOT_DIR}/tmp/dev-config" diff --git a/scripts/toggle-mock.sh b/scripts/toggle-mock.sh index 690d7f0be..0b09e3612 100755 --- a/scripts/toggle-mock.sh +++ b/scripts/toggle-mock.sh @@ -604,6 +604,10 @@ start_standalone_runtime() { load_env_file "${PULSE_DATA_DIR}/.env" if [[ "$PULSE_DATA_DIR" == "$MOCK_DATA_DIR" ]]; then + # The isolated mock data dir holds no real history, so the backend may + # safely backfill mock metrics history into its metrics.db (powers + # reports and long-range charts in mock mode). + export PULSE_MOCK_SEED_METRICS_STORE=true log_warn "Using isolated mock data dir (${MOCK_DATA_DIR}); production metrics history will pause while mock mode is on" else ensure_dev_key