From b0c1eca4cdace3cb798a77b2fa0c01a220f76981 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Thu, 27 Aug 2026 20:29:17 +0100 Subject: [PATCH] Add predictive storage capacity alerts --- .../v6/internal/subsystems/alerts.md | 17 + .../v6/internal/subsystems/monitoring.md | 13 + internal/alerts/alerts_test.go | 24 + internal/alerts/canonical_lifecycle.go | 50 ++- internal/alerts/canonical_metric.go | 5 + internal/alerts/capacity_forecast.go | 424 ++++++++++++++++++ internal/alerts/capacity_forecast_test.go | 227 ++++++++++ internal/alerts/metric_runtime.go | 10 +- internal/alerts/storage.go | 15 +- internal/alerts/unified_eval.go | 22 +- internal/alerts/unified_eval_test.go | 22 + .../monitoring/canonical_guardrails_test.go | 2 +- internal/monitoring/ceph.go | 2 +- internal/monitoring/metrics_history.go | 3 + internal/monitoring/metrics_history_test.go | 26 ++ internal/monitoring/monitor_alert_sync.go | 3 +- internal/monitoring/monitor_alerts.go | 2 +- .../monitoring/monitor_polling_storage.go | 2 +- .../monitoring/storage_capacity_forecast.go | 115 +++++ .../storage_capacity_forecast_test.go | 53 +++ 20 files changed, 1002 insertions(+), 35 deletions(-) create mode 100644 internal/alerts/capacity_forecast.go create mode 100644 internal/alerts/capacity_forecast_test.go create mode 100644 internal/monitoring/storage_capacity_forecast.go create mode 100644 internal/monitoring/storage_capacity_forecast_test.go diff --git a/docs/release-control/v6/internal/subsystems/alerts.md b/docs/release-control/v6/internal/subsystems/alerts.md index 3143d73c7..79076d6b6 100644 --- a/docs/release-control/v6/internal/subsystems/alerts.md +++ b/docs/release-control/v6/internal/subsystems/alerts.md @@ -193,6 +193,22 @@ lifecycle when the agent heartbeat is unhealthy, and resolves only when a fresh result from the currently assigned agent arrives. Notification delivery, acknowledgement, history, and recovery reuse the normal alert pipeline. +Storage capacity forecasting is an alert-grade evidence boundary, not an AI +prose feature. It requires at least 24 hours of valid history, hourly median +normalization, a fresh terminal sample, agreement between full-window and +recent positive slopes, and confidence of at least 0.80 before it can fire. +Raw poll count alone must never manufacture confidence. Forecast risk opens +only when projected exhaustion is within seven days (critical within one day) +and recovers only after the projection moves beyond fourteen days or trusted +evidence proves growth stopped. Missing, shallow, stale, or low-confidence +history is unknown rather than recovery for an already-active forecast. +Predictive and static usage policy share `metric-threshold:usage` as one +canonical occurrence: a forecast that later crosses the percentage threshold +keeps its start time, acknowledgement, history, and timeline instead of +emitting a forecast recovery plus a second capacity alert. The same contract +applies to Proxmox, Ceph, TrueNAS pools/datasets, and vSphere datastores through +their existing platform thresholds and disable policy. + Active-alert restore is opt-out at construction. `NewManagerWithDataDir` accepts `ManagerOption` values, and `WithoutPersistedAlertRestore` starts the manager with an empty active-alert set instead of reading `active-alerts.json`. Mock @@ -252,6 +268,7 @@ default construction path still restores. 45. `internal/alerts/docker.go` 46. `internal/alerts/pbs.go` 47. `internal/alerts/storage.go` +47a. `internal/alerts/capacity_forecast.go` 48. `internal/alerts/node.go` 49. `internal/alerts/host.go` 50. `internal/alerts/backup_snapshot.go` diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 4a644be5d..b94276350 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -420,6 +420,18 @@ lifecycle, and treats a fresh result from that same assignment as recovery. Assignment trackers are removed with their targets and reset when agent identity changes. +Storage capacity forecasting consumes monitoring-owned percentage history +through `internal/monitoring/storage_capacity_forecast.go`. The bridge combines +the durable SQLite series with the in-memory tail and current observation, then +caches the alert-owned trend result for a bounded interval. Durable history is +required in the read path so restart cannot erase a previously earned +confidence floor. Canonical storage resources must resolve their metrics target +before trend evaluation, keeping TrueNAS and vSphere forecast identity aligned +with the series written by `syncUnifiedStorageMetrics`; Proxmox and Ceph retain +their source-native storage history IDs. Monitoring supplies evidence only and +must not choose forecast horizons, severity, hysteresis, notification routing, +or lifecycle identity. + Monitoring ingest keeps mock mode hermetic. The unified read path already substitutes the mock snapshot wholesale, so anything that runs after that substitution has to be suppressed explicitly rather than assumed hidden. Server @@ -510,6 +522,7 @@ cleanup so readers cannot retain orphaned runtime or alert projections. 50. `internal/models/ceph_cluster_identity.go` 51. `internal/truenas/types.go` 52. `internal/monitoring/monitor_alert_sync.go` +52a. `internal/monitoring/storage_capacity_forecast.go` 53. `internal/monitoring/platform_poller_shared.go` 54. `internal/monitoring/monitor_backups.go` 55. `internal/monitoring/resource_stale_thresholds.go` diff --git a/internal/alerts/alerts_test.go b/internal/alerts/alerts_test.go index b50a2aac8..60f8ec2f5 100644 --- a/internal/alerts/alerts_test.go +++ b/internal/alerts/alerts_test.go @@ -20878,3 +20878,27 @@ func TestCleanupRemovesOnlyStaleSMARTCounterSnapshots(t *testing.T) { t.Fatal("current SMART counter snapshot was removed") } } + +func TestCheckStorageForecastUsesCanonicalUsageAlertIdentity(t *testing.T) { + m := newTestManager(t) + m.mu.Lock() + m.config.TimeThresholds = map[string]int{} + m.config.StorageDefault = HysteresisThreshold{Trigger: 80, Clear: 70} + m.mu.Unlock() + + storage := models.Storage{ID: "forecast-proof", Name: "archive", Status: "active", Usage: 72} + trend := CapacityTrendObservation{ + Ready: true, Reason: "increasing", ObservedAt: time.Now(), DailyChange: 5, + Confidence: 0.99, SampleCount: 300, BucketCount: 48, CoverageSpan: 48 * time.Hour, + } + m.CheckStorageWithCapacityTrend(storage, trend) + m.CheckStorageWithCapacityTrend(storage, trend) + + alert := testRequireActiveAlert(t, m, canonicalMetricStateID(storage.ID, "usage")) + if alert.CanonicalSpecID != canonicalMetricSpecID(storage.ID, "usage") { + t.Fatalf("CanonicalSpecID = %q, want canonical usage spec", alert.CanonicalSpecID) + } + if got := alert.Metadata[capacityAlertOriginKey]; got != capacityAlertOriginForecast { + t.Fatalf("capacity origin = %v, want forecast", got) + } +} diff --git a/internal/alerts/canonical_lifecycle.go b/internal/alerts/canonical_lifecycle.go index 73d35c45d..b978dd0f2 100644 --- a/internal/alerts/canonical_lifecycle.go +++ b/internal/alerts/canonical_lifecycle.go @@ -13,23 +13,27 @@ import ( ) type canonicalLifecycleAlertParams struct { - Spec alertspecs.ResourceAlertSpec - Evidence alertspecs.AlertEvidence - IntentSignal string - PolicyDisabledNoLock func() bool - AlertID string - AlertType string - ResourceID string - ResourceName string - Node string - Instance string - Message string - Metadata map[string]interface{} - AddToRecent bool - AddToHistory bool - RateLimit bool - DispatchAsync bool - IntentBackup BackupIntentContext + Spec alertspecs.ResourceAlertSpec + Evidence alertspecs.AlertEvidence + IntentSignal string + PolicyDisabledNoLock func() bool + AlertID string + AlertType string + ResourceID string + ResourceName string + Node string + Instance string + Message string + Value float64 + Threshold float64 + Metadata map[string]interface{} + AddToRecent bool + AddToHistory bool + RateLimit bool + DispatchAsync bool + NotifyOnSeverityChange bool + AddToHistoryOnSeverityChange bool + IntentBackup BackupIntentContext } type canonicalStatefulAlertParams struct { @@ -460,8 +464,8 @@ func (m *Manager) evaluateCanonicalLifecycleAlert(params canonicalLifecycleAlert Node: params.Node, Instance: params.Instance, Message: params.Message, - Value: 0, - Threshold: 0, + Value: params.Value, + Threshold: params.Threshold, StartTime: incident.StartedAt, LastSeen: params.Evidence.ObservedAt, Metadata: cloneMetadata(params.Metadata), @@ -501,6 +505,14 @@ func (m *Manager) evaluateCanonicalLifecycleAlert(params canonicalLifecycleAlert if existing != nil { if primary == reducer.EventSeverityChanged { result.Transition = transition(alertspecs.EvaluationTransitionSeverityChanged, alertspecs.AlertStateFiring, alertspecs.AlertStateFiring) + if params.AddToHistoryOnSeverityChange { + m.historyManager.AddAlertTransition(*alert) + } + if params.NotifyOnSeverityChange { + if !params.RateLimit || m.checkRateLimit(trackingKey) { + m.dispatchAlert(alert, params.DispatchAsync) + } + } } return result, true } diff --git a/internal/alerts/canonical_metric.go b/internal/alerts/canonical_metric.go index a91161469..e8422355c 100644 --- a/internal/alerts/canonical_metric.go +++ b/internal/alerts/canonical_metric.go @@ -327,6 +327,11 @@ func (m *Manager) evaluateCanonicalMetricAlert(spec alertspecs.ResourceAlertSpec if existingAlert.Metadata == nil { existingAlert.Metadata = map[string]interface{}{} } + if opts != nil { + for _, key := range opts.RemoveMetadata { + delete(existingAlert.Metadata, key) + } + } for k, v := range alertMetadata { existingAlert.Metadata[k] = v } diff --git a/internal/alerts/capacity_forecast.go b/internal/alerts/capacity_forecast.go new file mode 100644 index 000000000..b435829b3 --- /dev/null +++ b/internal/alerts/capacity_forecast.go @@ -0,0 +1,424 @@ +package alerts + +import ( + "fmt" + "math" + "sort" + "strconv" + "time" + + alertspecs "github.com/rcourtman/pulse-go-rewrite/internal/alerts/specs" + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rs/zerolog/log" +) + +const ( + capacityForecastLookback = 7 * 24 * time.Hour + capacityForecastBucketWidth = time.Hour + capacityForecastMinimumSpan = 24 * time.Hour + capacityForecastFreshness = 2 * time.Hour + capacityForecastMinimumBuckets = 12 + capacityForecastMinimumRate = 0.1 + capacityForecastMinConfidence = 0.80 + capacityForecastWarningHorizon = 7 * 24 * time.Hour + capacityForecastCriticalWindow = 24 * time.Hour + capacityForecastRecoveryWindow = 14 * 24 * time.Hour +) + +// CapacityMetricPoint is one percentage-utilization observation used to +// estimate when a capacity-backed resource will fill. Callers may provide raw +// samples at any cadence; the estimator normalizes them into hourly medians so +// a fast poller cannot manufacture confidence by repeating nearly identical +// observations. +type CapacityMetricPoint struct { + Timestamp time.Time + Value float64 +} + +// CapacityTrendObservation is detector evidence, not an alert decision. +// Ready means the evidence coverage is sufficient to say whether a trend is +// actionable. A non-ready observation must not be interpreted as recovery. +type CapacityTrendObservation struct { + Ready bool + Reason string + ObservedAt time.Time + DailyChange float64 + Confidence float64 + SampleCount int + BucketCount int + CoverageSpan time.Duration +} + +type capacityBucketPoint struct { + timestamp time.Time + value float64 +} + +// EstimateCapacityTrend produces a conservative, time-aware capacity trend. +// Alert policy (warning/critical horizons, threshold coexistence, and +// hysteresis) remains in Manager; this function only establishes trustworthy +// trend evidence. +func EstimateCapacityTrend(points []CapacityMetricPoint, now time.Time) CapacityTrendObservation { + if now.IsZero() { + now = time.Now() + } + result := CapacityTrendObservation{Reason: "insufficient-history"} + if len(points) == 0 { + return result + } + + cutoff := now.Add(-capacityForecastLookback) + buckets := make(map[int64][]float64) + validSamples := 0 + latest := time.Time{} + for _, point := range points { + if point.Timestamp.IsZero() || point.Timestamp.Before(cutoff) || point.Timestamp.After(now.Add(5*time.Minute)) { + continue + } + if math.IsNaN(point.Value) || math.IsInf(point.Value, 0) || point.Value < 0 || point.Value > 100 { + continue + } + bucket := point.Timestamp.Unix() / int64(capacityForecastBucketWidth/time.Second) + buckets[bucket] = append(buckets[bucket], point.Value) + validSamples++ + if point.Timestamp.After(latest) { + latest = point.Timestamp + } + } + result.SampleCount = validSamples + if validSamples == 0 || latest.IsZero() { + result.Reason = "no-valid-samples" + return result + } + result.ObservedAt = latest + if now.Sub(latest) > capacityForecastFreshness { + result.Reason = "stale-history" + return result + } + + normalized := make([]capacityBucketPoint, 0, len(buckets)) + for bucket, values := range buckets { + sort.Float64s(values) + median := values[len(values)/2] + if len(values)%2 == 0 { + median = (values[len(values)/2-1] + values[len(values)/2]) / 2 + } + normalized = append(normalized, capacityBucketPoint{ + timestamp: time.Unix(bucket*int64(capacityForecastBucketWidth/time.Second), 0), + value: median, + }) + } + sort.Slice(normalized, func(i, j int) bool { + return normalized[i].timestamp.Before(normalized[j].timestamp) + }) + result.BucketCount = len(normalized) + if len(normalized) < capacityForecastMinimumBuckets { + result.Reason = "insufficient-hourly-coverage" + return result + } + result.CoverageSpan = normalized[len(normalized)-1].timestamp.Sub(normalized[0].timestamp) + if result.CoverageSpan < capacityForecastMinimumSpan { + result.Reason = "history-window-too-short" + return result + } + + overallSlope, overallR2 := capacityLinearRegression(normalized) + recentStart := len(normalized) / 2 + if len(normalized)-recentStart < 8 { + recentStart = len(normalized) - 8 + } + recentSlope, _ := capacityLinearRegression(normalized[recentStart:]) + overallDaily := overallSlope * 24 + recentDaily := recentSlope * 24 + result.DailyChange = overallDaily + + // A historic rise that has flattened or reversed is not an impending + // exhaustion signal. Requiring both windows to rise also rejects one-off + // capacity jumps caused by a resize or a telemetry discontinuity. + if overallDaily <= capacityForecastMinimumRate || recentDaily <= capacityForecastMinimumRate { + result.Ready = true + result.Reason = "not-increasing" + return result + } + + spanFactor := math.Min(1, result.CoverageSpan.Hours()/48) + agreement := math.Min(overallDaily, recentDaily) / math.Max(overallDaily, recentDaily) + result.Confidence = clampCapacityConfidence(overallR2 * spanFactor * agreement) + result.Ready = true + if result.Confidence < capacityForecastMinConfidence { + result.Reason = "low-confidence" + return result + } + result.Reason = "increasing" + return result +} + +func capacityLinearRegression(points []capacityBucketPoint) (slopePerHour, rSquared float64) { + if len(points) < 2 { + return 0, 0 + } + start := points[0].timestamp + n := float64(len(points)) + var sumX, sumY, sumXY, sumX2 float64 + for _, point := range points { + x := point.timestamp.Sub(start).Hours() + sumX += x + sumY += point.value + sumXY += x * point.value + sumX2 += x * x + } + denominator := n*sumX2 - sumX*sumX + if denominator == 0 { + return 0, 0 + } + slopePerHour = (n*sumXY - sumX*sumY) / denominator + intercept := (sumY - slopePerHour*sumX) / n + meanY := sumY / n + var residual, total float64 + for _, point := range points { + x := point.timestamp.Sub(start).Hours() + predicted := intercept + slopePerHour*x + residual += math.Pow(point.value-predicted, 2) + total += math.Pow(point.value-meanY, 2) + } + if total == 0 { + return slopePerHour, 0 + } + return slopePerHour, clampCapacityConfidence(1 - residual/total) +} + +func clampCapacityConfidence(value float64) float64 { + if value < 0 { + return 0 + } + if value > 1 { + return 1 + } + return value +} + +const ( + capacityAlertOriginKey = "capacityAlertOrigin" + capacityAlertOriginThreshold = "threshold" + capacityAlertOriginForecast = "forecast" +) + +var capacityForecastMetadataKeys = []string{ + "forecastConfidence", + "forecastDailyChangePct", + "forecastDaysToFull", + "forecastObservedAt", + "forecastSampleCount", + "forecastBucketCount", + "forecastCoverageSeconds", +} + +func (m *Manager) evaluateStorageCapacity(storage models.Storage, thresholds ThresholdConfig, trend CapacityTrendObservation) { + input := &UnifiedResourceInput{ + ID: storage.ID, + Type: "storage", + Name: storage.Name, + Node: storage.Node, + Instance: storage.Instance, + Disk: &UnifiedResourceMetric{Percent: storage.Usage}, + } + m.evaluateUnifiedCapacity(input, thresholds, trend, func() bool { + if !m.config.Enabled { + return true + } + allDisabled, _ := m.alertPolicyTypeSwitchesNoLock("storage") + return allDisabled || m.resolveStorageThresholdsNoLock(storage).Disabled + }) +} + +func (m *Manager) evaluateUnifiedCapacity(input *UnifiedResourceInput, thresholds ThresholdConfig, trend CapacityTrendObservation, policyDisabledNoLock func() bool) { + if input == nil || input.Disk == nil { + return + } + threshold := thresholds.Usage + if threshold == nil || threshold.Trigger <= 0 { + m.evaluateUnifiedMetrics(input, thresholds, nil) + return + } + + current := input.DiskValue() + active, origin := m.activeStorageCapacityOrigin(input.ID) + staticTriggered := current >= threshold.Trigger + staticLatched := active && origin == capacityAlertOriginThreshold && threshold.Clear > 0 && current >= threshold.Clear + if staticTriggered || staticLatched { + m.evaluateUnifiedMetrics(input, thresholds, &metricOptions{ + Metadata: map[string]interface{}{ + capacityAlertOriginKey: capacityAlertOriginThreshold, + }, + RemoveMetadata: capacityForecastMetadataKeys, + }) + return + } + + // Insufficient or low-confidence history is unknown, not recovery. Keep an + // already-firing forecast occurrence untouched until trustworthy evidence + // says the risk has receded; normal stale-alert cleanup remains the final + // bound if telemetry never becomes usable again. + recoveryFloor := threshold.Clear + if recoveryFloor <= 0 { + recoveryFloor = threshold.Trigger + } + if active && origin == capacityAlertOriginForecast && current >= recoveryFloor && (!trend.Ready || trend.Reason == "low-confidence") { + return + } + + eta, hasETA := capacityTimeToFull(current, trend.DailyChange) + forecastTrusted := trend.Ready && trend.Confidence >= capacityForecastMinConfidence && hasETA + forecastTriggered := forecastTrusted && eta <= capacityForecastWarningHorizon + forecastLatched := active && origin == capacityAlertOriginForecast && forecastTrusted && eta <= capacityForecastRecoveryWindow + if forecastTriggered || forecastLatched { + m.evaluateCapacityForecast(input, thresholds, trend, eta, policyDisabledNoLock) + return + } + + // No predictive risk remains. Run the ordinary metric evaluator so a + // forecast recovery and every existing static hysteresis rule retain the + // same canonical state, history, and notification behavior. + m.evaluateUnifiedMetrics(input, thresholds, &metricOptions{ + Metadata: map[string]interface{}{ + capacityAlertOriginKey: capacityAlertOriginThreshold, + }, + RemoveMetadata: capacityForecastMetadataKeys, + }) +} + +func (m *Manager) activeStorageCapacityOrigin(resourceID string) (bool, string) { + trackingKey := canonicalMetricStateID(resourceID, "usage") + m.mu.RLock() + defer m.mu.RUnlock() + alert, exists := m.getActiveAlertNoLock(trackingKey) + if !exists || alert == nil { + return false, "" + } + if alert.Metadata == nil { + return true, capacityAlertOriginThreshold + } + origin, _ := alert.Metadata[capacityAlertOriginKey].(string) + if origin == "" { + origin = capacityAlertOriginThreshold + } + return true, origin +} + +func capacityTimeToFull(current, dailyChange float64) (time.Duration, bool) { + if current < 0 || current >= 100 || dailyChange <= capacityForecastMinimumRate || math.IsNaN(dailyChange) || math.IsInf(dailyChange, 0) { + return 0, false + } + days := (100 - current) / dailyChange + if days <= 0 || math.IsNaN(days) || math.IsInf(days, 0) { + return 0, false + } + return time.Duration(days * float64(24*time.Hour)), true +} + +func (m *Manager) evaluateCapacityForecast(input *UnifiedResourceInput, thresholds ThresholdConfig, trend CapacityTrendObservation, eta time.Duration, policyDisabledNoLock func() bool) { + resourceID := input.ID + specID := canonicalMetricSpecID(resourceID, "usage") + resourceType, ok := unifiedMetricResourceType(input.Type) + if !ok { + return + } + spec, err := buildCanonicalSeverityThresholdSpec( + specID, + resourceID, + input.Name, + resourceType, + "capacity-risk", + 1, + 2, + false, + ) + if err != nil { + log.Warn().Err(err).Str("resourceID", input.ID).Msg("Skipping invalid storage capacity forecast spec") + return + } + spec.ConfirmationsRequired = 2 + if err := spec.Validate(); err != nil { + log.Warn().Err(err).Str("resourceID", input.ID).Msg("Skipping invalid confirmed storage capacity forecast spec") + return + } + + riskScore := 1.0 + if eta <= capacityForecastCriticalWindow { + riskScore = 2 + } + observedAt := m.policyNow() + daysToFull := eta.Hours() / 24 + message := fmt.Sprintf( + "%s projected to fill in %s (%.1f%% used, +%.2f%%/day)", + unifiedAlertType(input.Type), + formatCapacityETA(eta), + input.DiskValue(), + trend.DailyChange, + ) + attributes := map[string]string{ + "capacity_alert_origin": capacityAlertOriginForecast, + "confidence": strconv.FormatFloat(trend.Confidence, 'f', 3, 64), + "current_usage_percent": strconv.FormatFloat(input.DiskValue(), 'f', 2, 64), + "daily_change_percent": strconv.FormatFloat(trend.DailyChange, 'f', 3, 64), + "forecast_days_to_full": strconv.FormatFloat(daysToFull, 'f', 2, 64), + "history_bucket_count": strconv.Itoa(trend.BucketCount), + "history_coverage_seconds": strconv.FormatInt(int64(trend.CoverageSpan/time.Second), 10), + } + metadata := map[string]interface{}{ + "resourceType": input.Type, + "clearThreshold": thresholds.Usage.Clear, + capacityAlertOriginKey: capacityAlertOriginForecast, + "forecastConfidence": trend.Confidence, + "forecastDailyChangePct": trend.DailyChange, + "forecastDaysToFull": daysToFull, + "forecastObservedAt": trend.ObservedAt, + "forecastSampleCount": trend.SampleCount, + "forecastBucketCount": trend.BucketCount, + "forecastCoverageSeconds": int64(trend.CoverageSpan / time.Second), + } + _, _ = m.evaluateCanonicalLifecycleAlert(canonicalLifecycleAlertParams{ + Spec: spec, + Evidence: alertspecs.AlertEvidence{ + ObservedAt: observedAt, + Summary: message, + Attributes: attributes, + SeverityThreshold: &alertspecs.SeverityThresholdEvidence{ + Metric: "capacity-risk", + Direction: alertspecs.ThresholdDirectionAbove, + Observed: riskScore, + }, + }, + IntentSignal: MetricAlertIntentSignal("usage"), + PolicyDisabledNoLock: policyDisabledNoLock, + AlertID: canonicalMetricStateID(resourceID, "usage"), + AlertType: "usage", + ResourceID: resourceID, + ResourceName: input.Name, + Node: input.Node, + Instance: input.Instance, + Message: message, + Value: input.DiskValue(), + Threshold: 100, + Metadata: metadata, + AddToRecent: true, + AddToHistory: true, + RateLimit: true, + NotifyOnSeverityChange: true, + AddToHistoryOnSeverityChange: true, + }) +} + +func formatCapacityETA(eta time.Duration) string { + if eta < 2*time.Hour { + return "about 1 hour" + } + if eta < 24*time.Hour { + return fmt.Sprintf("about %.0f hours", math.Ceil(eta.Hours())) + } + days := int(math.Ceil(eta.Hours() / 24)) + if days == 1 { + return "about 1 day" + } + return fmt.Sprintf("about %d days", days) +} diff --git a/internal/alerts/capacity_forecast_test.go b/internal/alerts/capacity_forecast_test.go new file mode 100644 index 000000000..234b0f7bd --- /dev/null +++ b/internal/alerts/capacity_forecast_test.go @@ -0,0 +1,227 @@ +package alerts + +import ( + "math" + "strings" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/models" +) + +func TestEstimateCapacityTrendRequiresCoverageAndRecentAgreement(t *testing.T) { + now := time.Date(2026, time.August, 27, 12, 0, 0, 0, time.UTC) + + t.Run("trusts a clean multi-day fill trend", func(t *testing.T) { + points := make([]CapacityMetricPoint, 0, 73) + for hour := 72; hour >= 0; hour-- { + age := time.Duration(hour) * time.Hour + points = append(points, CapacityMetricPoint{ + Timestamp: now.Add(-age), + Value: 55 + float64(72-hour)*0.08, + }) + } + trend := EstimateCapacityTrend(points, now) + if !trend.Ready || trend.Reason != "increasing" { + t.Fatalf("trend = %+v, want trusted increasing evidence", trend) + } + if math.Abs(trend.DailyChange-1.92) > 0.02 { + t.Fatalf("DailyChange = %.3f, want about 1.92", trend.DailyChange) + } + if trend.Confidence < capacityForecastMinConfidence { + t.Fatalf("Confidence = %.3f, want >= %.2f", trend.Confidence, capacityForecastMinConfidence) + } + }) + + t.Run("rejects dense samples without a full-day span", func(t *testing.T) { + points := make([]CapacityMetricPoint, 0, 300) + for i := 0; i < 300; i++ { + points = append(points, CapacityMetricPoint{ + Timestamp: now.Add(-time.Duration(300-i) * time.Minute), + Value: 50 + float64(i)*0.02, + }) + } + trend := EstimateCapacityTrend(points, now) + if trend.Ready || trend.Reason != "insufficient-hourly-coverage" { + t.Fatalf("trend = %+v, want insufficient hourly coverage", trend) + } + }) + + t.Run("does not extrapolate a historic jump after growth stops", func(t *testing.T) { + points := make([]CapacityMetricPoint, 0, 73) + for hour := 72; hour >= 0; hour-- { + elapsed := 72 - hour + value := 50.0 + if elapsed >= 24 { + value = 70 + } + points = append(points, CapacityMetricPoint{Timestamp: now.Add(-time.Duration(hour) * time.Hour), Value: value}) + } + trend := EstimateCapacityTrend(points, now) + if !trend.Ready || trend.Reason != "not-increasing" { + t.Fatalf("trend = %+v, want a ready non-increasing decision", trend) + } + }) +} + +func TestStorageForecastSharesLifecycleWithStaticUsageAlert(t *testing.T) { + m := newTestManager(t) + m.ClearActiveAlerts() + m.mu.Lock() + m.config.TimeThresholds = map[string]int{} + m.config.SuppressionWindow = 0 + m.config.MinimumDelta = 0 + m.config.ActivationState = ActivationActive + m.config.StorageDefault = HysteresisThreshold{Trigger: 80, Clear: 70} + m.mu.Unlock() + + deliveries := 0 + resolved := 0 + m.SetAlertCallback(func(*Alert) { deliveries++ }) + m.SetResolvedCallback(func(string) { resolved++ }) + + storage := models.Storage{ + ID: "storage-forecast-1", + Name: "archive", + Node: "pve1", + Instance: "lab", + Status: "active", + Usage: 72, + } + trend := CapacityTrendObservation{ + Ready: true, + Reason: "increasing", + ObservedAt: time.Now(), + DailyChange: 5, + Confidence: 0.98, + SampleCount: 400, + BucketCount: 72, + CoverageSpan: 72 * time.Hour, + } + + // Forecast alerts require two independent evaluation cycles. + m.CheckStorageWithCapacityTrend(storage, trend) + if testHasActiveAlert(t, m, canonicalMetricStateID(storage.ID, "usage")) { + t.Fatal("forecast activated before its confirmation floor") + } + m.CheckStorageWithCapacityTrend(storage, trend) + alert := testRequireActiveAlert(t, m, canonicalMetricStateID(storage.ID, "usage")) + if alert.Type != "usage" || alert.Level != AlertLevelWarning { + t.Fatalf("forecast alert = type %q level %q, want usage warning", alert.Type, alert.Level) + } + if got := alert.Metadata[capacityAlertOriginKey]; got != capacityAlertOriginForecast { + t.Fatalf("capacity origin = %v, want forecast", got) + } + if !strings.Contains(alert.Message, "projected to fill") { + t.Fatalf("message = %q, want predictive explanation", alert.Message) + } + if deliveries != 1 { + t.Fatalf("deliveries = %d, want one forecast activation", deliveries) + } + start := alert.StartTime + if err := m.AcknowledgeAlert(alert.ID, "operator"); err != nil { + t.Fatalf("acknowledge forecast: %v", err) + } + + // A real threshold breach upgrades the same occurrence. It must not emit a + // forecast recovery or a second unacknowledged activation. + storage.Usage = 91 + m.CheckStorageWithCapacityTrend(storage, trend) + upgraded := testRequireActiveAlert(t, m, canonicalMetricStateID(storage.ID, "usage")) + if upgraded.StartTime != start { + t.Fatalf("occurrence start changed from %s to %s", start, upgraded.StartTime) + } + if !upgraded.Acknowledged || upgraded.AckUser != "operator" { + t.Fatalf("acknowledgement was not preserved: %+v", upgraded) + } + if upgraded.Level != AlertLevelCritical { + t.Fatalf("upgraded level = %q, want critical", upgraded.Level) + } + if got := upgraded.Metadata[capacityAlertOriginKey]; got != capacityAlertOriginThreshold { + t.Fatalf("capacity origin = %v, want threshold", got) + } + if _, exists := upgraded.Metadata["forecastDaysToFull"]; exists { + t.Fatal("stale forecast metadata survived the static threshold transition") + } + if deliveries != 1 { + t.Fatalf("deliveries = %d, want acknowledged transition not to page again", deliveries) + } + if resolved != 0 { + t.Fatalf("resolved callbacks = %d, forecast-to-threshold transition must stay one incident", resolved) + } + if got := len(m.GetActiveAlerts()); got != 1 { + t.Fatalf("active alerts = %d, want one canonical capacity incident", got) + } +} + +func TestStorageForecastTreatsUncertainEvidenceAsUnknown(t *testing.T) { + m := newTestManager(t) + m.ClearActiveAlerts() + m.mu.Lock() + m.config.TimeThresholds = map[string]int{} + m.config.StorageDefault = HysteresisThreshold{Trigger: 80, Clear: 70} + m.mu.Unlock() + + storage := models.Storage{ID: "storage-forecast-unknown", Name: "media", Status: "active", Usage: 70} + trusted := CapacityTrendObservation{ + Ready: true, Reason: "increasing", ObservedAt: time.Now(), DailyChange: 6, + Confidence: 0.99, SampleCount: 300, BucketCount: 48, CoverageSpan: 48 * time.Hour, + } + m.CheckStorageWithCapacityTrend(storage, trusted) + m.CheckStorageWithCapacityTrend(storage, trusted) + before := testRequireActiveAlert(t, m, canonicalMetricStateID(storage.ID, "usage")) + + uncertain := trusted + uncertain.Confidence = 0.3 + uncertain.Reason = "low-confidence" + m.CheckStorageWithCapacityTrend(storage, uncertain) + after := testRequireActiveAlert(t, m, canonicalMetricStateID(storage.ID, "usage")) + if after.StartTime != before.StartTime { + t.Fatal("uncertain evidence replaced the active forecast occurrence") + } + + recovered := trusted + recovered.DailyChange = 0 + recovered.Confidence = 0 + recovered.Reason = "not-increasing" + m.CheckStorageWithCapacityTrend(storage, recovered) + if testHasActiveAlert(t, m, canonicalMetricStateID(storage.ID, "usage")) { + t.Fatal("positive non-increasing evidence did not recover the forecast") + } +} + +func TestUnifiedStorageForecastUsesPlatformPolicyAndCanonicalIdentity(t *testing.T) { + m := newTestManager(t) + m.ClearActiveAlerts() + usage := &HysteresisThreshold{Trigger: 85, Clear: 75} + m.mu.Lock() + m.config.TimeThresholds = map[string]int{} + m.config.TrueNASDefaults.Usage = usage + m.mu.Unlock() + + input := &UnifiedResourceInput{ + ID: "truenas:atlas/pool:tank", + Type: "truenas-pool", + Name: "tank", + Node: "atlas", + Instance: "TrueNAS", + Disk: &UnifiedResourceMetric{Percent: 70}, + } + trend := CapacityTrendObservation{ + Ready: true, Reason: "increasing", ObservedAt: time.Now(), DailyChange: 7, + Confidence: 0.99, SampleCount: 300, BucketCount: 48, CoverageSpan: 48 * time.Hour, + } + m.CheckUnifiedResourceWithCapacityTrend(input, trend) + m.CheckUnifiedResourceWithCapacityTrend(input, trend) + + alert := testRequireActiveAlert(t, m, canonicalMetricStateID(input.ID, "usage")) + if alert.ResourceID != input.ID || alert.Type != "usage" { + t.Fatalf("alert identity = resource %q type %q", alert.ResourceID, alert.Type) + } + if got := alert.Metadata["resourceType"]; got != "truenas-pool" { + t.Fatalf("resourceType metadata = %v, want truenas-pool", got) + } + if got := alert.Metadata[capacityAlertOriginKey]; got != capacityAlertOriginForecast { + t.Fatalf("capacity origin = %v, want forecast", got) + } +} diff --git a/internal/alerts/metric_runtime.go b/internal/alerts/metric_runtime.go index fb5412c6c..e941f354f 100644 --- a/internal/alerts/metric_runtime.go +++ b/internal/alerts/metric_runtime.go @@ -166,8 +166,9 @@ func (m *Manager) getGlobalMetricTimeThreshold(metricType string) (int, bool) { // checkMetric checks a single metric against its threshold with hysteresis. type metricOptions struct { - Metadata map[string]interface{} - Message string + Metadata map[string]interface{} + RemoveMetadata []string + Message string // MonitorOnly suppresses external notifications while still tracking the alert. MonitorOnly bool } @@ -333,6 +334,11 @@ func (m *Manager) checkMetric(resourceID, resourceName, node, instance, resource if existingAlert.Metadata == nil { existingAlert.Metadata = map[string]interface{}{} } + if opts != nil { + for _, key := range opts.RemoveMetadata { + delete(existingAlert.Metadata, key) + } + } existingAlert.Metadata["resourceType"] = resourceType existingAlert.Metadata["clearThreshold"] = threshold.Clear existingAlert.Metadata["monitorOnly"] = monitorOnly diff --git a/internal/alerts/storage.go b/internal/alerts/storage.go index 93618ca21..20e4e70ba 100644 --- a/internal/alerts/storage.go +++ b/internal/alerts/storage.go @@ -15,6 +15,12 @@ import ( // CheckStorage checks storage against thresholds func (m *Manager) CheckStorage(storage models.Storage) { + m.CheckStorageWithCapacityTrend(storage, CapacityTrendObservation{}) +} + +// CheckStorageWithCapacityTrend evaluates static storage policy and optional +// predictive evidence as one canonical capacity lifecycle. +func (m *Manager) CheckStorageWithCapacityTrend(storage models.Storage, trend CapacityTrendObservation) { m.mu.RLock() if !m.config.Enabled { m.mu.RUnlock() @@ -97,14 +103,7 @@ func (m *Manager) CheckStorage(storage models.Storage) { // Check usage if storage is online - checkMetric will skip if threshold is nil or <= 0 if storage.Status != "offline" && storage.Status != "unavailable" && storage.Usage > 0 { - m.evaluateUnifiedMetrics(&UnifiedResourceInput{ - ID: storage.ID, - Type: "storage", - Name: storage.Name, - Node: storage.Node, - Instance: storage.Instance, - Disk: &UnifiedResourceMetric{Percent: storage.Usage}, - }, thresholds, nil) + m.evaluateStorageCapacity(storage, thresholds, trend) } // Check ZFS pool status if this is ZFS storage diff --git a/internal/alerts/unified_eval.go b/internal/alerts/unified_eval.go index 681e1f13c..7e065cb9b 100644 --- a/internal/alerts/unified_eval.go +++ b/internal/alerts/unified_eval.go @@ -462,6 +462,12 @@ func unifiedMetricResourceType(typeKey string) (unifiedresources.ResourceType, b } func (m *Manager) CheckUnifiedResourceMetrics(resources []unifiedresources.Resource) { + m.CheckUnifiedResourceMetricsWithCapacityTrends(resources, nil) +} + +// CheckUnifiedResourceMetricsWithCapacityTrends evaluates canonical resources +// with optional alert-grade capacity evidence keyed by canonical resource ID. +func (m *Manager) CheckUnifiedResourceMetricsWithCapacityTrends(resources []unifiedresources.Resource, capacityTrends map[string]CapacityTrendObservation) { if m == nil { return } @@ -470,7 +476,7 @@ func (m *Manager) CheckUnifiedResourceMetrics(resources []unifiedresources.Resou if !ok { continue } - m.CheckUnifiedResource(input) + m.CheckUnifiedResourceWithCapacityTrend(input, capacityTrends[input.ID]) } } @@ -730,6 +736,13 @@ func (i *UnifiedResourceInput) TemperatureValue() float64 { // available metric. Discrete event alerts (offline, RAID, backup age, etc.) // are NOT evaluated here — they remain in the typed Check* methods. func (m *Manager) CheckUnifiedResource(input *UnifiedResourceInput) { + m.CheckUnifiedResourceWithCapacityTrend(input, CapacityTrendObservation{}) +} + +// CheckUnifiedResourceWithCapacityTrend keeps predictive and static capacity +// evaluation on the same canonical lifecycle for every admitted storage +// platform. +func (m *Manager) CheckUnifiedResourceWithCapacityTrend(input *UnifiedResourceInput, trend CapacityTrendObservation) { if input == nil { return } @@ -757,5 +770,12 @@ func (m *Manager) CheckUnifiedResource(input *UnifiedResourceInput) { Str("resourceType", unifiedAlertType(input.Type)). Msg("Evaluating unified resource metrics") + if unifiedStorageUsageResourceType(input.Type) && input.Disk != nil { + m.evaluateUnifiedCapacity(input, thresholds, trend, func() bool { + return !m.config.Enabled || m.unifiedPlatformAlertsDisabledNoLock(input.Type) || m.resolveResourceThresholds(input.Type, input.ID).Disabled + }) + return + } + m.evaluateUnifiedMetrics(input, thresholds, nil) } diff --git a/internal/alerts/unified_eval_test.go b/internal/alerts/unified_eval_test.go index 4a808e7e5..9584df70d 100644 --- a/internal/alerts/unified_eval_test.go +++ b/internal/alerts/unified_eval_test.go @@ -1537,3 +1537,25 @@ func TestUnifiedResourceInputExcludesDockerAppContainers(t *testing.T) { t.Fatalf("truenas app containers must stay unified-eval owned, got %+v ok=%v", input, ok) } } + +func TestUnifiedStorageForecastKeepsCanonicalMetricSpecAcrossPlatforms(t *testing.T) { + m := newTestManager(t) + configureUnifiedEvalManager(t, m, unifiedEvalBaseConfig()) + input := &UnifiedResourceInput{ + ID: "vmware:lab/datastore:archive", + Type: "vmware-datastore", + Name: "archive", + Disk: &UnifiedResourceMetric{Percent: 70}, + } + trend := CapacityTrendObservation{ + Ready: true, Reason: "increasing", ObservedAt: time.Now(), DailyChange: 7, + Confidence: 0.99, SampleCount: 300, BucketCount: 48, CoverageSpan: 48 * time.Hour, + } + m.CheckUnifiedResourceWithCapacityTrend(input, trend) + m.CheckUnifiedResourceWithCapacityTrend(input, trend) + + alert := testRequireActiveAlert(t, m, canonicalMetricStateID(input.ID, "usage")) + if alert.CanonicalSpecID != canonicalMetricSpecID(input.ID, "usage") { + t.Fatalf("CanonicalSpecID = %q, want canonical usage spec", alert.CanonicalSpecID) + } +} diff --git a/internal/monitoring/canonical_guardrails_test.go b/internal/monitoring/canonical_guardrails_test.go index 118a5ab9a..7ca105988 100644 --- a/internal/monitoring/canonical_guardrails_test.go +++ b/internal/monitoring/canonical_guardrails_test.go @@ -494,7 +494,7 @@ func TestUnifiedResourceAlertSyncEvaluatesMetricsBeforeIncidents(t *testing.T) { } source := string(data) - metricIndex := strings.Index(source, "CheckUnifiedResourceMetrics(resources)") + metricIndex := strings.Index(source, "CheckUnifiedResourceMetricsWithCapacityTrends(resources") incidentIndex := strings.Index(source, "SyncUnifiedResourceIncidents(resources)") if metricIndex < 0 { t.Fatalf("monitor alert sync must run unified resource metric evaluation") diff --git a/internal/monitoring/ceph.go b/internal/monitoring/ceph.go index 354cce2d2..6d609721e 100644 --- a/internal/monitoring/ceph.go +++ b/internal/monitoring/ceph.go @@ -93,7 +93,7 @@ func (m *Monitor) checkCephPoolStorage(cluster models.CephCluster) { } } if m.alertManager != nil { - m.alertManager.CheckStorage(storage) + m.alertManager.CheckStorageWithCapacityTrend(storage, m.storageCapacityTrend(storage, timestamp)) } } } diff --git a/internal/monitoring/metrics_history.go b/internal/monitoring/metrics_history.go index 3fee8361b..ea7accbb8 100644 --- a/internal/monitoring/metrics_history.go +++ b/internal/monitoring/metrics_history.go @@ -53,6 +53,9 @@ type MetricsHistory struct { diskMetrics map[string]*DiskMetrics // key: disk metrics resource ID maxDataPoints int retentionTime time.Duration + + capacityForecastMu sync.Mutex + capacityForecastCache map[string]storageCapacityForecastCacheEntry } // NewMetricsHistory creates a new metrics history tracker diff --git a/internal/monitoring/metrics_history_test.go b/internal/monitoring/metrics_history_test.go index 3954ceaa9..153e949a9 100644 --- a/internal/monitoring/metrics_history_test.go +++ b/internal/monitoring/metrics_history_test.go @@ -3,6 +3,8 @@ package monitoring import ( "testing" "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/models" ) func TestNewMetricsHistory(t *testing.T) { @@ -1395,3 +1397,27 @@ func TestCleanupMetricsReturnsNilForExpiredData(t *testing.T) { t.Errorf("cleanupMetrics should return nil for fully expired data, got slice with len=%d cap=%d", len(result), cap(result)) } } + +func TestMetricsHistorySuppliesAlertGradeStorageCapacityTrend(t *testing.T) { + now := time.Now().UTC().Truncate(time.Hour) + history := NewMetricsHistory(1000, 7*24*time.Hour) + for hour := 72; hour >= 0; hour-- { + history.AddStorageMetric( + "forecast-history-proof", + "usage", + 60+float64(72-hour)*0.08, + now.Add(-time.Duration(hour)*time.Hour), + ) + } + + monitor := &Monitor{metricsHistory: history} + trend := monitor.storageCapacityTrend(models.Storage{ + ID: "forecast-history-proof", Name: "archive", Status: "active", Usage: 65.76, + }, now) + if !trend.Ready || trend.Reason != "increasing" { + t.Fatalf("trend = %+v, want trusted increasing evidence", trend) + } + if trend.Confidence < 0.8 { + t.Fatalf("confidence = %.3f, want alert-grade evidence", trend.Confidence) + } +} diff --git a/internal/monitoring/monitor_alert_sync.go b/internal/monitoring/monitor_alert_sync.go index 98a037e41..c8fe2bc3c 100644 --- a/internal/monitoring/monitor_alert_sync.go +++ b/internal/monitoring/monitor_alert_sync.go @@ -2,6 +2,7 @@ package monitoring import ( "strings" + "time" "github.com/rcourtman/pulse-go-rewrite/internal/alerts" "github.com/rcourtman/pulse-go-rewrite/internal/logging" @@ -132,7 +133,7 @@ func (m *Monitor) syncUnifiedResourceAlertsToState(resources []unifiedresources. m.migrateAvailabilityLinksToCanonicalIDs(resources) - m.alertManager.CheckUnifiedResourceMetrics(resources) + m.alertManager.CheckUnifiedResourceMetricsWithCapacityTrends(resources, m.unifiedStorageCapacityTrends(resources, time.Now())) m.alertManager.SyncUnifiedResourceIncidents(resources) m.syncAlertsToState() } diff --git a/internal/monitoring/monitor_alerts.go b/internal/monitoring/monitor_alerts.go index d2dc6c13e..2000eea3e 100644 --- a/internal/monitoring/monitor_alerts.go +++ b/internal/monitoring/monitor_alerts.go @@ -549,7 +549,7 @@ func (m *Monitor) checkMockAlerts() { Str("name", storage.Name). Float64("usage", storage.Usage). Msg("Checking storage for alerts") - m.alertManager.CheckStorage(storage) + m.alertManager.CheckStorageWithCapacityTrend(storage, m.storageCapacityTrend(storage, time.Now())) } // Check alerts for PBS instances diff --git a/internal/monitoring/monitor_polling_storage.go b/internal/monitoring/monitor_polling_storage.go index c2adaeb87..718a89dbe 100644 --- a/internal/monitoring/monitor_polling_storage.go +++ b/internal/monitoring/monitor_polling_storage.go @@ -710,7 +710,7 @@ func (m *Monitor) pollStorageWithNodes(ctx context.Context, instanceName string, } if m.alertManager != nil { - m.alertManager.CheckStorage(storage) + m.alertManager.CheckStorageWithCapacityTrend(storage, m.storageCapacityTrend(storage, time.Now())) } } diff --git a/internal/monitoring/storage_capacity_forecast.go b/internal/monitoring/storage_capacity_forecast.go new file mode 100644 index 000000000..d1ff8c5f9 --- /dev/null +++ b/internal/monitoring/storage_capacity_forecast.go @@ -0,0 +1,115 @@ +package monitoring + +import ( + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/alerts" + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" + "github.com/rs/zerolog/log" +) + +const ( + storageCapacityForecastLookback = 7 * 24 * time.Hour + storageCapacityForecastRefresh = 15 * time.Minute +) + +type storageCapacityForecastCacheEntry struct { + trend alerts.CapacityTrendObservation + expiresAt time.Time +} + +// storageCapacityTrend bridges monitoring-owned history into alert-owned +// predictive policy. SQLite history is included so a process restart does not +// erase confidence; the in-memory tail and current observation cover buffered +// writes and installations where durable metrics are unavailable. +func (m *Monitor) storageCapacityTrend(storage models.Storage, now time.Time) alerts.CapacityTrendObservation { + if m == nil || storage.ID == "" || storage.Usage <= 0 { + return alerts.CapacityTrendObservation{Reason: "invalid-current-usage"} + } + return m.storageCapacityTrendFor(storage.ID, storage.ID, storage.Usage, now) +} + +func (m *Monitor) storageCapacityTrendFor(cacheID, historyID string, currentUsage float64, now time.Time) alerts.CapacityTrendObservation { + if m == nil || cacheID == "" || historyID == "" || currentUsage <= 0 { + return alerts.CapacityTrendObservation{Reason: "invalid-current-usage"} + } + if now.IsZero() { + now = time.Now() + } + + if m.metricsHistory != nil { + m.metricsHistory.capacityForecastMu.Lock() + if cached, ok := m.metricsHistory.capacityForecastCache[cacheID]; ok && now.Before(cached.expiresAt) { + m.metricsHistory.capacityForecastMu.Unlock() + return cached.trend + } + m.metricsHistory.capacityForecastMu.Unlock() + } + + points := make([]alerts.CapacityMetricPoint, 0, 256) + if m.metricsStore != nil { + stored, err := m.metricsStore.Query( + "storage", + historyID, + "usage", + now.Add(-storageCapacityForecastLookback), + now, + int64(time.Hour/time.Second), + ) + if err != nil { + log.Debug().Err(err).Str("storage", cacheID).Msg("Persistent capacity history unavailable; using in-memory tail") + } else { + for _, point := range stored { + points = append(points, alerts.CapacityMetricPoint{Timestamp: point.Timestamp, Value: point.Value}) + } + } + } + if m.metricsHistory != nil { + for _, point := range m.metricsHistory.GetAllStorageMetrics(historyID, storageCapacityForecastLookback)["usage"] { + points = append(points, alerts.CapacityMetricPoint{Timestamp: point.Timestamp, Value: point.Value}) + } + } + points = append(points, alerts.CapacityMetricPoint{Timestamp: now, Value: currentUsage}) + + trend := alerts.EstimateCapacityTrend(points, now) + if m.metricsHistory != nil { + m.metricsHistory.capacityForecastMu.Lock() + if m.metricsHistory.capacityForecastCache == nil { + m.metricsHistory.capacityForecastCache = make(map[string]storageCapacityForecastCacheEntry) + } + m.metricsHistory.capacityForecastCache[cacheID] = storageCapacityForecastCacheEntry{ + trend: trend, + expiresAt: now.Add(storageCapacityForecastRefresh), + } + m.metricsHistory.capacityForecastMu.Unlock() + } + return trend +} + +func (m *Monitor) unifiedStorageCapacityTrends(resources []unifiedresources.Resource, now time.Time) map[string]alerts.CapacityTrendObservation { + trends := make(map[string]alerts.CapacityTrendObservation) + var resolver MetricsTargetResourceStore + if candidate, ok := m.resourceStore.(MetricsTargetResourceStore); ok { + resolver = candidate + } + for _, resource := range resources { + input, ok := alerts.UnifiedResourceInputFromResource(resource) + if !ok || input.Disk == nil { + continue + } + switch input.Type { + case "truenas-pool", "truenas-dataset", "vmware-datastore": + default: + continue + } + historyID := input.ID + if resolver != nil { + if target := resolver.MetricsTargetForResource(resource.ID); target != nil && target.ResourceType == "storage" && target.ResourceID != "" { + historyID = target.ResourceID + } + } + trends[input.ID] = m.storageCapacityTrendFor(input.ID, historyID, input.DiskValue(), now) + } + return trends +} diff --git a/internal/monitoring/storage_capacity_forecast_test.go b/internal/monitoring/storage_capacity_forecast_test.go new file mode 100644 index 000000000..08a1c3f3a --- /dev/null +++ b/internal/monitoring/storage_capacity_forecast_test.go @@ -0,0 +1,53 @@ +package monitoring + +import ( + "path/filepath" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/pkg/metrics" +) + +func TestStorageCapacityTrendUsesDurableHistoryAfterRestart(t *testing.T) { + now := time.Now().UTC().Truncate(time.Hour) + config := metrics.DefaultConfig(t.TempDir()) + config.DBPath = filepath.Join(t.TempDir(), "capacity-metrics.db") + store, err := metrics.NewStore(config) + if err != nil { + t.Fatalf("new metrics store: %v", err) + } + t.Cleanup(func() { _ = store.Close() }) + + writes := make([]metrics.WriteMetric, 0, 72) + for hour := 72; hour >= 1; hour-- { + writes = append(writes, metrics.WriteMetric{ + ResourceType: "storage", + ResourceID: "durable-storage", + MetricType: "usage", + Value: 60 + float64(72-hour)*0.08, + Timestamp: now.Add(-time.Duration(hour) * time.Hour), + Tier: metrics.TierHourly, + }) + } + store.WriteBatchSync(writes) + + // Deliberately omit MetricsHistory: this models the process immediately + // after restart, when the in-memory ring is empty but SQLite remains. + monitor := &Monitor{metricsStore: store} + trend := monitor.storageCapacityTrend(models.Storage{ + ID: "durable-storage", + Name: "archive", + Status: "active", + Usage: 65.76, + }, now) + if !trend.Ready || trend.Reason != "increasing" { + t.Fatalf("trend = %+v, want trusted increasing evidence from durable history", trend) + } + if trend.CoverageSpan < 48*time.Hour { + t.Fatalf("coverage = %s, want at least 48h", trend.CoverageSpan) + } + if trend.Confidence < 0.8 { + t.Fatalf("confidence = %.3f, want alert-grade evidence", trend.Confidence) + } +}