Keep PBS metric writes tied to observations

Canonical registry rebuilds can run several times between PBS polls. Reuse the metric source sighting time so unchanged datastore capacity does not create a fresh raw sample on every rebuild.

Contract-Neutral: behavioral regression fix; metric contract and agent lifecycle boundaries unchanged
Change-source: pulse-maintainer

internal/monitoring/monitor.go

@@ -5385,7 +5385,7 @@ func (m *Monitor) syncUnifiedStorageMetrics(store ResourceStoreInterface) {
 
 	now := time.Now()
 	storeWrites := make([]metrics.WriteMetric, 0)
-	appendStoreWrite := func(resourceType, resourceID, metricType string, value float64) {
+	appendStoreWrite := func(resourceType, resourceID, metricType string, value float64, timestamp time.Time) {
 		if m.metricsStore == nil {
 			return
 		}
@@ -5394,7 +5394,7 @@ func (m *Monitor) syncUnifiedStorageMetrics(store ResourceStoreInterface) {
 			ResourceID:   resourceID,
 			MetricType:   metricType,
 			Value:        value,
-			Timestamp:    now,
+			Timestamp:    timestamp,
 			Tier:         metrics.TierRaw,
 		})
 	}
@@ -5434,6 +5434,13 @@ func (m *Monitor) syncUnifiedStorageMetrics(store ResourceStoreInterface) {
 		seenTargets[targetID] = struct{}{}
 
 		disk := resource.Metrics.Disk
+		observedAt := resource.LastSeen
+		if status, ok := resource.SourceStatus[disk.Source]; ok && !status.LastSeen.IsZero() {
+			observedAt = status.LastSeen
+		}
+		if observedAt.IsZero() {
+			observedAt = now
+		}
 		usage := disk.Percent
 		used := int64(0)
 		total := int64(0)
@@ -5452,18 +5459,18 @@ func (m *Monitor) syncUnifiedStorageMetrics(store ResourceStoreInterface) {
 		}
 
 		if m.metricsHistory != nil {
-			m.metricsHistory.AddStorageMetric(targetID, "usage", usage, now)
+			m.metricsHistory.AddStorageMetric(targetID, "usage", usage, observedAt)
 			if total > 0 {
-				m.metricsHistory.AddStorageMetric(targetID, "used", float64(used), now)
-				m.metricsHistory.AddStorageMetric(targetID, "total", float64(total), now)
-				m.metricsHistory.AddStorageMetric(targetID, "avail", float64(free), now)
+				m.metricsHistory.AddStorageMetric(targetID, "used", float64(used), observedAt)
+				m.metricsHistory.AddStorageMetric(targetID, "total", float64(total), observedAt)
+				m.metricsHistory.AddStorageMetric(targetID, "avail", float64(free), observedAt)
 			}
 		}
-		appendStoreWrite("storage", targetID, "usage", usage)
+		appendStoreWrite("storage", targetID, "usage", usage, observedAt)
 		if total > 0 {
-			appendStoreWrite("storage", targetID, "used", float64(used))
-			appendStoreWrite("storage", targetID, "total", float64(total))
-			appendStoreWrite("storage", targetID, "avail", float64(free))
+			appendStoreWrite("storage", targetID, "used", float64(used), observedAt)
+			appendStoreWrite("storage", targetID, "total", float64(total), observedAt)
+			appendStoreWrite("storage", targetID, "avail", float64(free), observedAt)
 		}
 	}
 	if len(storeWrites) > 0 {

internal/monitoring/monitor_polling_test.go

@@ -368,6 +368,78 @@ func TestUpdateResourceStoreSyncsUnifiedIncidentAlerts(t *testing.T) {
 }
 
 func TestUpdateResourceStoreSyncsCanonicalStorageMetrics(t *testing.T) {
+	t.Run("repeated PBS registry rebuild keeps one observed sample", func(t *testing.T) {
+		cfg := metrics.DefaultConfig(t.TempDir())
+		store, err := metrics.NewStore(cfg)
+		if err != nil {
+			t.Fatalf("metrics.NewStore() error = %v", err)
+		}
+		defer func() { _ = store.Close() }()
+
+		observedAt := time.Now().UTC().Add(-10 * time.Second).Truncate(time.Second)
+		snapshot := models.StateSnapshot{
+			PBSInstances: []models.PBSInstance{{
+				ID:       "pbs-backup",
+				Name:     "backup",
+				Status:   "online",
+				LastSeen: observedAt,
+				Datastores: []models.PBSDatastore{{
+					Name:   "main",
+					Status: "available",
+					Total:  1000,
+					Used:   400,
+					Free:   600,
+					Usage:  40,
+				}},
+			}},
+		}
+		resourceStore := unifiedresources.NewMonitorAdapter(nil)
+		monitor := &Monitor{
+			resourceStore:  resourceStore,
+			metricsHistory: NewMetricsHistory(1024, 24*time.Hour),
+			metricsStore:   store,
+		}
+
+		monitor.updateResourceStore(snapshot)
+		monitor.updateResourceStore(snapshot)
+
+		var targetID string
+		for _, resource := range resourceStore.GetAll() {
+			if resource.Type != unifiedresources.ResourceTypeStorage || resource.Storage == nil || resource.Storage.Platform != "pbs" {
+				continue
+			}
+			target := resourceStore.MetricsTargetForResource(resource.ID)
+			if target != nil {
+				targetID = target.ResourceID
+			}
+		}
+		if targetID == "" {
+			t.Fatal("expected a PBS datastore metrics target")
+		}
+
+		memory := monitor.GetStorageMetrics(targetID, time.Hour)
+		for _, metricType := range []string{"usage", "used", "total", "avail"} {
+			points := memory[metricType]
+			if len(points) != 1 {
+				t.Fatalf("in-memory %s points = %d, want one sample for one PBS observation", metricType, len(points))
+			}
+			if !points[0].Timestamp.Equal(observedAt) {
+				t.Fatalf("in-memory %s timestamp = %s, want observation time %s", metricType, points[0].Timestamp, observedAt)
+			}
+
+			persisted, err := store.Query("storage", targetID, metricType, observedAt.Add(-time.Minute), time.Now().Add(time.Minute), 0)
+			if err != nil {
+				t.Fatalf("query persisted %s metrics: %v", metricType, err)
+			}
+			if len(persisted) != 1 {
+				t.Fatalf("persisted %s points = %d, want one sample for one PBS observation", metricType, len(persisted))
+			}
+			if !persisted[0].Timestamp.Equal(observedAt) {
+				t.Fatalf("persisted %s timestamp = %s, want observation time %s", metricType, persisted[0].Timestamp, observedAt)
+			}
+		}
+	})
+
 	t.Run("supplemental truenas storage", func(t *testing.T) {
 		cfg := metrics.DefaultConfig(t.TempDir())
 		store, err := metrics.NewStore(cfg)