diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index b51b2d57dc..7c84b24cf5 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -728,6 +728,12 @@ into monitoring's models. Absent evidence has to carry its declared sentinel rather than a zero value that reads as a real measurement: an absent physical-disk view projects `Wearout` as `unifiedresources.WearoutUnreported`, never `0`, which would announce a spent disk the agent never reported. +When the host heartbeat lease expires, `State.ExpireHostTelemetry` keeps that +agent's SMART temperature and I/O counters as last-known values but marks them +`unavailable` ("host agent stopped reporting"); the agent's next accepted +report replaces them. Expiry is compare-and-set on the report time the offline +sweep judged stale, so a report admitted between that judgement and the expiry +is never expired. An enabled availability target assigned to a host agent creates an agent-lifecycle lease for that exact target/agent pairing. First assignment diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 9ae9a5d83a..10dbab5822 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -3684,7 +3684,21 @@ transiently `unavailable`, provider/controller `unsupported`, or unexpectedly `missing`. Normalization may retain the last known value when the current observation is not available, but it must preserve the current state and reason so API and UI consumers do not present retained evidence as freshly -collected. Unified-resource physical-disk round trips must retain named +collected. A host agent's SMART temperature and I/O counters follow its +reporting lease: once `State.ExpireHostTelemetry` expires the agent, they stay +as last-known values marked `unavailable` ("host agent stopped reporting"), +and the linked Proxmox disk merge and the registry let the agent's own later +state supersede the availability it supplied earlier +(`diskinventory.MergeReportedStatus`), so neither a skipped disk poll nor a +host that is never disk-polled again can carry it forward. Expiry is +compare-and-set on the report time the evaluation judged stale, so a report +accepted in between is never expired. Every SMART temperature history writer +records a temperature only when its current collection state is available (or +predates collection state), so a retained reading has to keep its non-available +state to stay out of history; a path that relabels a carried reading as +available, such as a node-temperature carry stamped `proxmox_node_smart`, is +not covered by this rule. +Unified-resource physical-disk round trips must retain named `StorageGroup` membership rather than degrading it to the generic `Used` filesystem label. That same host-agent temperature boundary must prefer a recent linked host-agent diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index b91304e4d5..452757ae82 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -3516,6 +3516,13 @@ the five field statuses for serial, temperature, I/O, controller, and pool. `unavailable`, `unsupported`, or `missing` state may retain a prior value for continuity, but the state and reason must survive so the consumer cannot claim fresh evidence or synthesize controller-level activity for one member. +The one exception is a source withdrawing its own evidence: when the agent +row reports a field as no longer available, that supersedes an `available` +state the Proxmox row still carries from the same source +(`diskinventory.MergeReportedStatus`), in either ingest order. A host agent +past its reporting lease must not stay "collected" through the Proxmox row's +copy of the agent's own state, including when its host is down and no disk +poll refreshes that copy. Cross-source correlation compares normalized serial and WWN values across fields without truncation, allowing a PVE bare-hex array-volume serial to join diff --git a/internal/models/host_transient_expiry_test.go b/internal/models/host_transient_expiry_test.go index 4aab4e2d9e..d91c8bfd28 100644 --- a/internal/models/host_transient_expiry_test.go +++ b/internal/models/host_transient_expiry_test.go @@ -1,6 +1,11 @@ package models -import "testing" +import ( + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/pkg/diskinventory" +) func TestExpireHostTelemetryClearsTransientStorageOperationsOnly(t *testing.T) { state := NewState() @@ -24,7 +29,7 @@ func TestExpireHostTelemetryClearsTransientStorageOperationsOnly(t *testing.T) { }, }) - host, changed := state.ExpireHostTelemetry("storage-host") + host, changed := state.ExpireHostTelemetry("storage-host", time.Time{}) if !changed { t.Fatal("expected host telemetry to change") } @@ -47,3 +52,92 @@ func TestExpireHostTelemetryClearsTransientStorageOperationsOnly(t *testing.T) { t.Fatalf("RAID last-known health was lost: %+v", host.RAID[0]) } } + +func TestExpireHostTelemetryMarksSMARTReadingsNoLongerCollected(t *testing.T) { + state := NewState() + judged := time.Now().Add(-time.Hour) + state.UpsertHost(Host{ + ID: "agent", + Status: "online", + LastSeen: judged, + Sensors: HostSensorSummary{SMART: []HostDiskSMART{ + { + Device: "sda", Serial: "CURRENT", Temperature: 41, Health: "PASSED", + IO: &DiskIO{Device: "sda", ReadBytes: 1000}, + Collection: &diskinventory.CollectionStatus{ + Serial: diskinventory.Available("smartctl"), + Temperature: diskinventory.Available("smartctl"), + IO: diskinventory.Available("kernel_diskstats"), + }, + }, + // Agents before collection provenance report only what they collected. + {Device: "sdb", Serial: "LEGACY", Temperature: 38, IO: &DiskIO{Device: "sdb"}}, + { + Device: "sdc", Serial: "ASLEEP", Standby: true, + Collection: &diskinventory.CollectionStatus{ + Temperature: diskinventory.Unavailable("smartctl", "disk is in standby"), + IO: diskinventory.Unsupported("controller", "per-member counters unavailable"), + }, + }, + }}, + }) + + host, changed := state.ExpireHostTelemetry("agent", judged) + if !changed { + t.Fatal("expected host telemetry to change") + } + stopped := diskinventory.Unavailable("smartctl", "host agent stopped reporting") + current := host.Sensors.SMART[0] + if current.Temperature != 41 || current.IO == nil || current.IO.ReadBytes != 1000 { + t.Fatalf("last-known SMART readings were dropped: %+v", current) + } + if current.Collection.Temperature != stopped || + current.Collection.IO != diskinventory.Unavailable("kernel_diskstats", "host agent stopped reporting") { + t.Fatalf("expired readings still claim to be collected: %+v", current.Collection) + } + if current.Collection.Serial != diskinventory.Available("smartctl") { + t.Fatalf("disk identity evidence must stay last-known: %+v", current.Collection.Serial) + } + legacy := host.Sensors.SMART[1] + if legacy.Temperature != 38 || legacy.Collection == nil || + legacy.Collection.Temperature.State != diskinventory.FieldUnavailable || + legacy.Collection.IO.State != diskinventory.FieldUnavailable { + t.Fatalf("provenance-less readings not expired: %+v", legacy) + } + asleep := host.Sensors.SMART[2].Collection + if asleep.Temperature != diskinventory.Unavailable("smartctl", "disk is in standby") || + asleep.IO.State != diskinventory.FieldUnsupported { + t.Fatalf("readings that were already not collected must keep their state: %+v", asleep) + } + + if _, changed := state.ExpireHostTelemetry("agent", judged); changed { + t.Fatal("re-evaluating an expired host must not report a change") + } + stored, _ := state.GetHost("agent") + if stored.Sensors.SMART[0].Collection.Temperature != stopped { + t.Fatalf("expiry was not kept in state: %+v", stored.Sensors.SMART[0].Collection) + } +} + +// Expiry is judged on a host snapshot. A report accepted before the expiry +// takes the state lock must not be expired on the strength of that judgement. +func TestExpireHostTelemetryLeavesHostThatReportedSinceJudgement(t *testing.T) { + state := NewState() + judged := time.Now().Add(-time.Hour) + fresh := Host{ + ID: "agent", Status: "online", LastSeen: time.Now(), + Sensors: HostSensorSummary{SMART: []HostDiskSMART{{ + Device: "sda", Temperature: 41, + Collection: &diskinventory.CollectionStatus{Temperature: diskinventory.Available("smartctl")}, + }}}, + } + state.UpsertHost(fresh) + + if host, changed := state.ExpireHostTelemetry("agent", judged); changed || host.ID != "" { + t.Fatalf("a host that reported after the stale judgement was expired: %+v changed=%v", host, changed) + } + stored, _ := state.GetHost("agent") + if stored.Status != "online" || stored.Sensors.SMART[0].Collection.Temperature.State != diskinventory.FieldAvailable { + t.Fatalf("fresh report was altered: %+v", stored) + } +} diff --git a/internal/models/models.go b/internal/models/models.go index 5f1f2af5b8..d20dd7c66f 100644 --- a/internal/models/models.go +++ b/internal/models/models.go @@ -5634,7 +5634,13 @@ func (s *State) SetHostStatus(hostID, status string) bool { // ExpireHostTelemetry marks a host offline and clears transient operation // claims that cannot remain authoritative after the reporting lease expires. // Static topology and health counters remain available as last-known context. -func (s *State) ExpireHostTelemetry(hostID string) (Host, bool) { +// SMART temperature and I/O readings stay as last-known values, but their +// collection state says they are no longer collected. +// +// lastSeen is the report time the caller judged stale. A host that has +// reported since then is left untouched and nothing is returned, so a report +// accepted between that judgement and this call is never expired. +func (s *State) ExpireHostTelemetry(hostID string, lastSeen time.Time) (Host, bool) { s.mu.Lock() defer s.mu.Unlock() @@ -5642,6 +5648,9 @@ func (s *State) ExpireHostTelemetry(hostID string) (Host, bool) { if host.ID != hostID { continue } + if !host.LastSeen.Equal(lastSeen) { + return Host{}, false + } changed := host.Status != "offline" host.Status = "offline" for idx := range host.RAID { @@ -5658,6 +5667,11 @@ func (s *State) ExpireHostTelemetry(hostID string) (Host, bool) { host.Unraid.SyncProgress = 0 changed = true } + for idx := range host.Sensors.SMART { + if expireHostSMARTReadings(&host.Sensors.SMART[idx]) { + changed = true + } + } s.Hosts[i] = host if changed { s.LastUpdate = time.Now() @@ -5667,6 +5681,40 @@ func (s *State) ExpireHostTelemetry(hostID string) (Host, bool) { return Host{}, false } +const hostAgentStoppedReportingReason = "host agent stopped reporting" + +// expireHostSMARTReadings marks the temperature and I/O counters of one SMART +// row as no longer collected, keeping their values as last-known evidence. +// Reports from agents that predate collection provenance carry no state, so a +// present reading is treated as collected. It reports whether anything changed. +func expireHostSMARTReadings(disk *HostDiskSMART) bool { + collection := diskinventory.CloneStatus(disk.Collection) + if collection == nil { + collection = &diskinventory.CollectionStatus{} + } + temperature := expireCollectedReading(&collection.Temperature, disk.Temperature > 0) + io := expireCollectedReading(&collection.IO, disk.IO != nil) + if !temperature && !io { + return false + } + disk.Collection = collection + return true +} + +func expireCollectedReading(status *diskinventory.FieldStatus, hasValue bool) bool { + switch status.State { + case diskinventory.FieldAvailable: + case "": + if !hasValue { + return false + } + default: + return false + } + *status = diskinventory.Unavailable(status.Source, hostAgentStoppedReportingReason) + return true +} + // TouchHost updates the last seen timestamp for a host. func (s *State) TouchHost(hostID string, ts time.Time) bool { s.mu.Lock() diff --git a/internal/monitoring/issue1595_collection_trust_test.go b/internal/monitoring/issue1595_collection_trust_test.go index 83587605fd..2c4835d3c9 100644 --- a/internal/monitoring/issue1595_collection_trust_test.go +++ b/internal/monitoring/issue1595_collection_trust_test.go @@ -10,6 +10,7 @@ import ( "github.com/rcourtman/pulse-go-rewrite/internal/models" "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" "github.com/rcourtman/pulse-go-rewrite/pkg/diskinventory" + "github.com/rcourtman/pulse-go-rewrite/pkg/metrics" ) type issue1595TopologyFixture struct { @@ -316,3 +317,46 @@ func hasIssue1595Source(sources []unifiedresources.DataSource, want unifiedresou } return false } + +// An agent report's SMART row may carry a temperature its collector did not +// collect this time (retained across a standby or failed probe). Only a +// collected temperature becomes a history sample. +func TestWriteHostSMARTMetricsRecordsOnlyCollectedTemperatures(t *testing.T) { + storeConfig := metrics.DefaultConfig(t.TempDir()) + storeConfig.FlushInterval = time.Hour + store, err := metrics.NewStore(storeConfig) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = store.Close() }) + m := &Monitor{metricsStore: store} + + host := models.Host{ID: "agent-1", Hostname: "node1", Sensors: models.HostSensorSummary{SMART: []models.HostDiskSMART{ + {Device: "sda", Serial: "COLLECTED1", Temperature: 41, + Collection: &diskinventory.CollectionStatus{Temperature: diskinventory.Available("smartctl")}}, + {Device: "sdb", Serial: "LEGACY1", Temperature: 38}, + {Device: "sdc", Serial: "RETAINED1", Temperature: 36, + Collection: &diskinventory.CollectionStatus{Temperature: diskinventory.Unavailable("smartctl", "disk is in standby")}}, + }}} + now := time.Now() + m.writeHostSMARTMetrics(host, now) + store.Flush() + + for _, tc := range []struct { + disk models.HostDiskSMART + want int + }{ + {host.Sensors.SMART[0], 1}, + {host.Sensors.SMART[1], 1}, + {host.Sensors.SMART[2], 0}, + } { + id := unifiedresources.HostSMARTDiskSourceID(host, tc.disk) + points, err := store.Query("disk", id, "smart_temp", now.Add(-time.Minute), now.Add(time.Minute), 0) + if err != nil { + t.Fatal(err) + } + if len(points) != tc.want { + t.Fatalf("%s: temperature samples = %d, want %d", tc.disk.Serial, len(points), tc.want) + } + } +} diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 53182cf222..e5c5c89632 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -648,7 +648,10 @@ func mergeHostAgentSMARTIntoDisks(disks []models.PhysicalDisk, nodes []models.No ioCopy := *matched.IO updated[i].IO = &ioCopy } - updated[i].Collection = diskinventory.MergeStatus(updated[i].Collection, matched.Collection) + // The agent's later word on a field it supplied (its lease expired, or + // the disk went into standby) must win over the copy a skipped disk + // poll carries forward from the previous merge. + updated[i].Collection = diskinventory.MergeReportedStatus(updated[i].Collection, hostAgentSMARTCollection(*matched)) if shouldUseHostAgentPhysicalDiskHealth(updated[i].Health, matched.Health) { updated[i].Health = matched.Health @@ -658,6 +661,28 @@ func mergeHostAgentSMARTIntoDisks(disks []models.PhysicalDisk, nodes []models.No return updated } +// hostAgentLegacySource is the provenance recorded for a temperature from an +// agent that predates collection provenance (before 6.2). Those agents only +// send readings they collected. +const hostAgentLegacySource = "host_agent" + +// hostAgentSMARTCollection returns the agent's collection state for one SMART +// row, giving a provenance-less temperature the legacy agent source so that +// history and later lease expiry treat it like any other agent reading. +func hostAgentSMARTCollection(smart models.HostDiskSMART) *diskinventory.CollectionStatus { + collection := diskinventory.CloneStatus(smart.Collection) + if collection == nil { + collection = &diskinventory.CollectionStatus{} + } + switch { + case collection.Temperature.State == "" && smart.Temperature > 0 && !smart.Standby: + collection.Temperature = diskinventory.Available(hostAgentLegacySource) + case collection.Temperature.State != "" && strings.TrimSpace(collection.Temperature.Source) == "": + collection.Temperature.Source = hostAgentLegacySource + } + return collection +} + func shouldUseHostAgentPhysicalDiskHealth(existing, incoming string) bool { incoming = strings.ToUpper(strings.TrimSpace(incoming)) if incoming == "" || incoming == "UNKNOWN" { @@ -914,8 +939,7 @@ func (m *Monitor) writeSMARTMetrics(disk models.PhysicalDisk, now time.Time) { return } - // Temperature (always write if > 0) - if disk.Temperature > 0 && m.metricsHistory != nil { + if diskTemperatureCollected(disk.Temperature, disk.Collection) && m.metricsHistory != nil { m.metricsHistory.AddDiskMetric(resourceID, "smart_temp", float64(disk.Temperature), now) } @@ -929,6 +953,22 @@ func (m *Monitor) writeSMARTMetrics(disk models.PhysicalDisk, now time.Time) { } } +// diskTemperatureCollected reports whether a disk temperature was collected by +// its current observation. Normalization may carry a last-known temperature +// when it was not (standby, an agent past its reporting lease); that value must +// not be recorded as a new history sample. A temperature without collection +// state predates the contract and keeps the old behavior. +func diskTemperatureCollected(temperature int, collection *diskinventory.CollectionStatus) bool { + if temperature <= 0 { + return false + } + if collection == nil { + return true + } + state := collection.Temperature.State + return state == "" || state == diskinventory.FieldAvailable +} + // smartMetricStoreWrites builds the persisted SMART writes for one physical // disk observation. The caller owns batching so the unified sync can anchor the // observation time and drop exact replays (#1966). @@ -949,8 +989,7 @@ func (m *Monitor) smartMetricStoreWrites(disk models.PhysicalDisk, resourceID st }) } - // Temperature (always write if > 0) - if disk.Temperature > 0 { + if diskTemperatureCollected(disk.Temperature, disk.Collection) { appendWrite("smart_temp", float64(disk.Temperature)) } @@ -5984,6 +6023,7 @@ func (m *Monitor) syncUnifiedPhysicalDiskMetrics(store ResourceStoreInterface) { RPM: resource.PhysicalDisk.RPM, Used: resource.PhysicalDisk.Used, SmartAttributes: smartAttributesFromUnifiedMeta(resource.PhysicalDisk.SMART), + Collection: diskinventory.CloneStatus(resource.PhysicalDisk.Collection), LastChecked: resource.LastSeen, } if disk.Serial == "" { diff --git a/internal/monitoring/monitor_agents.go b/internal/monitoring/monitor_agents.go index 4ab39a48ea..5fcf15d549 100644 --- a/internal/monitoring/monitor_agents.go +++ b/internal/monitoring/monitor_agents.go @@ -4086,7 +4086,7 @@ func (m *Monitor) writeHostSMARTMetrics(host models.Host, now time.Time) { continue } - if disk.Temperature > 0 { + if diskTemperatureCollected(disk.Temperature, disk.Collection) { m.metricsStore.Write("disk", resourceID, "smart_temp", float64(disk.Temperature), now) } @@ -5090,7 +5090,7 @@ func (m *Monitor) evaluateHostAgents(now time.Time) { Bool("lastSeenZero", host.LastSeen.IsZero()). Msg("Host agent appears offline") } - if expiredHost, changed := m.state.ExpireHostTelemetry(host.ID); expiredHost.ID != "" { + if expiredHost, changed := m.state.ExpireHostTelemetry(host.ID, host.LastSeen); expiredHost.ID != "" { hostCopy = expiredHost resourceRefreshNeeded = resourceRefreshNeeded || changed } else { diff --git a/internal/monitoring/physical_disk_roundtrip_test.go b/internal/monitoring/physical_disk_roundtrip_test.go index 64060eaa9a..19c9f8a187 100644 --- a/internal/monitoring/physical_disk_roundtrip_test.go +++ b/internal/monitoring/physical_disk_roundtrip_test.go @@ -11,6 +11,9 @@ import ( "github.com/rcourtman/pulse-go-rewrite/internal/models" "github.com/rcourtman/pulse-go-rewrite/internal/notifications" "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" + "github.com/rcourtman/pulse-go-rewrite/pkg/diskinventory" + "github.com/rcourtman/pulse-go-rewrite/pkg/metrics" + "github.com/rcourtman/pulse-go-rewrite/pkg/proxmox" ) // Monitor construction restores notification destinations before admitting @@ -326,3 +329,228 @@ func TestMergeHostAgentSMARTIntoDisks_AgentWearoutFillsUnreportedPVELife(t *test t.Fatalf("agent endurance did not fill unreported PVE life: wearout=%d", result[0].Wearout) } } + +type silentAgentDiskPVEClient struct { + fakeStorageClient +} + +// The Proxmox inventory carries no temperature or I/O counters; for this disk +// both come only from the node's linked host agent. +func (*silentAgentDiskPVEClient) GetDisks(ctx context.Context, node string) ([]proxmox.Disk, error) { + return []proxmox.Disk{{ + DevPath: "/dev/sdb", + Model: "WDC WD40EFRX", + Serial: "WD-SILENT1", + Type: "hdd", + Health: "PASSED", + Wearout: 100, + Size: 4000787030016, + }}, nil +} + +// A host agent that stops reporting keeps its last SMART rows in state. Once +// its reporting lease expires, the linked Proxmox disk may retain the agent's +// last temperature and I/O counters, but the disk state, the canonical +// resource and temperature history must stop treating them as collected. +func TestSilentLinkedAgentSMARTIsRetainedButNotPresentedAsCollected(t *testing.T) { + t.Setenv("PULSE_DATA_DIR", t.TempDir()) + now := time.Now() + state := models.NewState() + state.UpdateNodesForInstance("pve1", []models.Node{{ + ID: "pve1-node1", Name: "node1", Instance: "pve1", Status: "online", + LastSeen: now, LinkedAgentID: "agent-1", + }}) + report := func(seen time.Time, temperature int, readBytes uint64) { + state.UpsertHost(models.Host{ + ID: "agent-1", Hostname: "node1", LinkedNodeID: "pve1-node1", + Status: "online", IntervalSeconds: 30, LastSeen: seen, + Sensors: models.HostSensorSummary{SMART: []models.HostDiskSMART{{ + Device: "/dev/sdb", Serial: "WD-SILENT1", Type: "sata", Health: "PASSED", + Temperature: temperature, + IO: &models.DiskIO{Device: "sdb", ReadBytes: readBytes}, + Collection: &diskinventory.CollectionStatus{ + Serial: diskinventory.Available("smartctl"), + Temperature: diskinventory.Available("smartctl"), + IO: diskinventory.Available("kernel_diskstats"), + }, + }}}, + }) + } + + alertManager := alerts.NewManager() + t.Cleanup(alertManager.Stop) + storeConfig := metrics.DefaultConfig(t.TempDir()) + storeConfig.FlushInterval = time.Hour + store, err := metrics.NewStore(storeConfig) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = store.Close() }) + adapter := unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(nil)) + history := NewMetricsHistory(100, time.Hour) + m := &Monitor{ + state: state, resourceStore: adapter, metricsHistory: history, metricsStore: store, + alertManager: alertManager, startTime: now.Add(-time.Hour), + lastPhysicalDiskPoll: make(map[string]time.Time), + } + pveNodes := []proxmox.Node{{Node: "node1", Status: "online"}} + nodeStatus := map[string]string{"node1": "online"} + + disk := func() models.PhysicalDisk { + t.Helper() + disks := state.GetSnapshot().PhysicalDisks + if len(disks) != 1 { + t.Fatalf("physical disks = %+v, want one", disks) + } + return disks[0] + } + fullPoll := func() models.PhysicalDisk { + t.Helper() + adapter.PopulateFromSnapshot(state.GetSnapshot()) + started := time.Now() + delete(m.lastPhysicalDiskPoll, "pve1") + m.maybePollPhysicalDisksAsync(context.Background(), "pve1", &config.PVEInstance{}, &silentAgentDiskPVEClient{}, pveNodes, nodeStatus, nil) + deadline := time.Now().Add(3 * time.Second) + for { + if disks := state.GetSnapshot().PhysicalDisks; len(disks) == 1 && !disks[0].LastChecked.Before(started) { + return disks[0] + } + if time.Now().After(deadline) { + t.Fatal("full physical disk poll did not land in state") + } + time.Sleep(10 * time.Millisecond) + } + } + skippedPoll := func() models.PhysicalDisk { + t.Helper() + adapter.PopulateFromSnapshot(state.GetSnapshot()) + m.lastPhysicalDiskPoll["pve1"] = time.Now() + m.maybePollPhysicalDisksAsync(context.Background(), "pve1", &config.PVEInstance{}, nil, nil, nil, nil) + return disk() + } + // Chart samples, and the persisted values. The store keys samples by + // second, so polls run back to back in this test can share one row. + temperatureSamples := func(d models.PhysicalDisk) (int, []float64) { + t.Helper() + id := unifiedresources.PhysicalDiskMetricID(d) + chart := len(history.GetDiskMetrics(id, "smart_temp", time.Hour)) + store.Flush() + points, err := store.Query("disk", id, "smart_temp", now.Add(-time.Hour), time.Now().Add(time.Hour), 0) + if err != nil { + t.Fatal(err) + } + stored := make([]float64, 0, len(points)) + for _, point := range points { + stored = append(stored, point.Value) + } + return chart, stored + } + canonicalCollection := func() *diskinventory.CollectionStatus { + t.Helper() + adapter.PopulateFromSnapshot(state.GetSnapshot()) + views := adapter.PhysicalDisks() + if len(views) != 1 { + t.Fatalf("canonical physical disks = %d, want one", len(views)) + } + if views[0].Temperature() != 41 { + t.Fatalf("canonical disk temperature = %d, want the retained 41", views[0].Temperature()) + } + return views[0].Collection() + } + assertRetainedNotCollected := func(step string, got models.PhysicalDisk) { + t.Helper() + if got.Temperature != 41 || got.IO == nil || got.IO.ReadBytes != 1000 { + t.Fatalf("%s: last-known temperature and I/O were not retained: temp=%d io=%+v", step, got.Temperature, got.IO) + } + if got.Collection == nil || + got.Collection.Temperature.State != diskinventory.FieldUnavailable || + got.Collection.IO.State != diskinventory.FieldUnavailable { + t.Fatalf("%s: silent agent's readings presented as collected: %+v", step, got.Collection) + } + } + + report(now, 41, 1000) + got := fullPoll() + if got.Temperature != 41 || got.Collection == nil || got.Collection.Temperature.State != diskinventory.FieldAvailable { + t.Fatalf("reporting agent temperature not merged: temp=%d collection=%+v", got.Temperature, got.Collection) + } + if chart, stored := temperatureSamples(got); chart != 1 || len(stored) != 1 || stored[0] != 41 { + t.Fatalf("reporting agent temperature history: chart=%d stored=%v, want one 41 sample", chart, stored) + } + + // The agent stops reporting; its lease expires on the next evaluation. + state.TouchHost("agent-1", now.Add(-hostAgentHealthWindow(30)-time.Minute)) + m.evaluateHostAgents(now) + + // No disk poll runs when the whole host is down (the PVE poll fails before + // reaching it), so the canonical disk must reflect the expiry on its own + // even though the Proxmox disk row still carries the earlier merge. + if collection := canonicalCollection(); collection == nil || + collection.Temperature.State != diskinventory.FieldUnavailable || + collection.IO.State != diskinventory.FieldUnavailable { + t.Fatalf("canonical disk presents the silent agent's readings as collected before any disk poll: %+v", collection) + } + assertRetainedNotCollected("skipped poll", skippedPoll()) + got = fullPoll() + assertRetainedNotCollected("full poll", got) + if chart, stored := temperatureSamples(got); chart != 1 || len(stored) != 1 { + t.Fatalf("silent agent's retained temperature was recorded as new history: chart=%d stored=%v", chart, stored) + } + if collection := canonicalCollection(); collection == nil || + collection.Temperature.State != diskinventory.FieldUnavailable || + collection.IO.State != diskinventory.FieldUnavailable { + t.Fatalf("canonical disk presents the silent agent's readings as collected: %+v", collection) + } + + // A resumed agent is collected again. + report(time.Now(), 39, 2000) + m.evaluateHostAgents(time.Now()) + got = fullPoll() + if got.Temperature != 39 || got.Collection.Temperature.State != diskinventory.FieldAvailable || + got.IO == nil || got.IO.ReadBytes != 2000 || got.Collection.IO.State != diskinventory.FieldAvailable { + t.Fatalf("resumed agent readings not collected: temp=%d io=%+v collection=%+v", got.Temperature, got.IO, got.Collection) + } + if chart, stored := temperatureSamples(got); chart != 2 || len(stored) == 0 || stored[len(stored)-1] != 39 { + t.Fatalf("resumed agent temperature history: chart=%d stored=%v, want a new 39 sample", chart, stored) + } +} + +// Agents before 6.2 send SMART rows without collection provenance. Their +// temperature is still a collected reading (history records it), and once the +// lease expires the disk must stop presenting it as collected even when a +// skipped disk poll carries the previous merge forward. +func TestMergeHostAgentSMARTIntoDisks_LegacyAgentTemperatureFollowsLease(t *testing.T) { + proxmoxDisk := models.PhysicalDisk{ + ID: "pve1-node1-sdb", Node: "node1", Instance: "pve1", DevPath: "/dev/sdb", Serial: "LEGACY1", + Collection: &diskinventory.CollectionStatus{ + Temperature: diskinventory.Unsupported("proxmox_disks", "Proxmox disk inventory does not expose temperature"), + }, + } + nodes := []models.Node{{Name: "node1", LinkedAgentID: "agent-1"}} + state := models.NewState() + state.UpsertHost(models.Host{ID: "agent-1", Sensors: models.HostSensorSummary{SMART: []models.HostDiskSMART{{ + Device: "/dev/sdb", Serial: "LEGACY1", Temperature: 38, + }}}}) + + reporting := mergeHostAgentSMARTIntoDisks([]models.PhysicalDisk{proxmoxDisk}, nodes, state.GetHosts())[0] + if reporting.Temperature != 38 || reporting.Collection.Temperature != diskinventory.Available(hostAgentLegacySource) { + t.Fatalf("legacy agent temperature not recorded as collected: temp=%d collection=%+v", reporting.Temperature, reporting.Collection) + } + if !diskTemperatureCollected(reporting.Temperature, reporting.Collection) { + t.Fatal("a reporting legacy agent's temperature must reach history") + } + + state.ExpireHostTelemetry("agent-1", time.Time{}) + for name, disk := range map[string]models.PhysicalDisk{ + "full poll": proxmoxDisk, + "skipped poll": reporting, + } { + got := mergeHostAgentSMARTIntoDisks([]models.PhysicalDisk{disk}, nodes, state.GetHosts())[0] + if got.Temperature != 38 || got.Collection.Temperature.State != diskinventory.FieldUnavailable { + t.Fatalf("%s: silent legacy agent temperature presented as collected: temp=%d collection=%+v", name, got.Temperature, got.Collection) + } + if diskTemperatureCollected(got.Temperature, got.Collection) { + t.Fatalf("%s: a silent legacy agent's retained temperature must not reach history", name) + } + } +} diff --git a/internal/unifiedresources/registry.go b/internal/unifiedresources/registry.go index 34992538c1..cb2f10a48a 100644 --- a/internal/unifiedresources/registry.go +++ b/internal/unifiedresources/registry.go @@ -3719,11 +3719,19 @@ func (rr *ResourceRegistry) mergeInto(existing *Resource, incoming Resource, sou if previous.IO != nil { existing.PhysicalDisk.IO = clonePhysicalDiskIOMeta(previous.IO) } - existing.PhysicalDisk.Collection = diskinventory.MergeStatus( + // The agent's own report is the later word on fields it supplied; + // the Proxmox row may still carry an earlier copy of them. + existing.PhysicalDisk.Collection = diskinventory.MergeReportedStatus( incoming.PhysicalDisk.Collection, previous.Collection, ) } + if source == SourceAgent && previous != nil { + existing.PhysicalDisk.Collection = diskinventory.MergeReportedStatus( + previous.Collection, + incoming.PhysicalDisk.Collection, + ) + } } if existing.PhysicalDisk != nil && (mergedPhysicalDisk || len(incoming.Incidents) > 0) { existing.PhysicalDisk.Risk = physicalDiskRiskFromMeta(existing.PhysicalDisk, existing.Incidents) diff --git a/internal/unifiedresources/registry_merge_policy_test.go b/internal/unifiedresources/registry_merge_policy_test.go index 51b27e416b..83c8fda61a 100644 --- a/internal/unifiedresources/registry_merge_policy_test.go +++ b/internal/unifiedresources/registry_merge_policy_test.go @@ -6,6 +6,7 @@ import ( "time" "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/pkg/diskinventory" ) func TestLinkedMergeAllowsOneSidedNodeHostLinkWhenHostnameCorroborates(t *testing.T) { @@ -557,3 +558,79 @@ func TestProxmoxOneWayLinkRejectsHostLocalEndpointCorroboration(t *testing.T) { t.Fatal("explicit private report-IP corroboration lost") } } + +// The Proxmox disk row carries a copy of the agent's collection state from the +// last disk poll. When the agent's own row later withdraws a reading (its +// reporting lease expired), the canonical disk must follow the agent in either +// ingest order, even if no disk poll has refreshed the Proxmox copy (a host that +// is down entirely is never disk-polled). +func TestPhysicalDiskMergeFollowsAgentWithdrawalOfItsOwnReadings(t *testing.T) { + proxmoxDisk := func(agent diskinventory.FieldStatus) Resource { + return Resource{ + Type: ResourceTypePhysicalDisk, Name: "WDC", Status: StatusOnline, + PhysicalDisk: &PhysicalDiskMeta{ + DevPath: "/dev/sdb", Serial: "WD-SILENT1", Temperature: 41, + Collection: &diskinventory.CollectionStatus{ + Serial: diskinventory.Available("proxmox_disks"), + Temperature: agent, + Pool: diskinventory.Available("proxmox_zfs"), + }, + }, + } + } + agentDisk := func(temperature diskinventory.FieldStatus) Resource { + return Resource{ + Type: ResourceTypePhysicalDisk, Name: "WDC", Status: StatusOnline, + PhysicalDisk: &PhysicalDiskMeta{ + DevPath: "/dev/sdb", Serial: "WD-SILENT1", Temperature: 41, + Collection: &diskinventory.CollectionStatus{ + Serial: diskinventory.Available("smartctl"), + Temperature: temperature, + }, + }, + } + } + identity := ResourceIdentity{MachineID: "WD-SILENT1", Hostnames: []string{"node1"}} + collected := diskinventory.Available("smartctl") + expired := diskinventory.Unavailable("smartctl", "host agent stopped reporting") + + for _, tc := range []struct { + name string + proxmoxFirst bool + agent diskinventory.FieldStatus + want diskinventory.FieldStatus + }{ + {"proxmox then expired agent", true, expired, expired}, + {"expired agent then proxmox", false, expired, expired}, + {"proxmox then reporting agent", true, collected, collected}, + {"reporting agent then proxmox", false, collected, collected}, + } { + t.Run(tc.name, func(t *testing.T) { + registry := NewRegistry(nil) + // The Proxmox row still holds the earlier "collected" copy. + ingestProxmox := func() { registry.ingest(SourceProxmox, "pve1-node1-sdb", proxmoxDisk(collected), identity) } + ingestAgent := func() { registry.ingest(SourceAgent, "agent-1-sdb", agentDisk(tc.agent), identity) } + if tc.proxmoxFirst { + ingestProxmox() + ingestAgent() + } else { + ingestAgent() + ingestProxmox() + } + disks := registry.ListByType(ResourceTypePhysicalDisk) + if len(disks) != 1 || disks[0].PhysicalDisk == nil || disks[0].PhysicalDisk.Collection == nil { + t.Fatalf("expected one merged disk, got %+v", disks) + } + collection := disks[0].PhysicalDisk.Collection + if collection.Temperature != tc.want { + t.Fatalf("temperature collection = %+v, want %+v", collection.Temperature, tc.want) + } + if disks[0].PhysicalDisk.Temperature != 41 { + t.Fatalf("last-known temperature was not retained: %d", disks[0].PhysicalDisk.Temperature) + } + if collection.Pool != diskinventory.Available("proxmox_zfs") { + t.Fatalf("Proxmox-owned evidence changed: %+v", collection.Pool) + } + }) + } +} diff --git a/pkg/diskinventory/status.go b/pkg/diskinventory/status.go index 68e98b4839..7041ab9883 100644 --- a/pkg/diskinventory/status.go +++ b/pkg/diskinventory/status.go @@ -89,6 +89,35 @@ func MergeStatus(existing, incoming *CollectionStatus) *CollectionStatus { return &merged } +// MergeReportedStatus merges a source's own report into existing state that +// may still hold an earlier copy of that report. It behaves like MergeStatus, +// except that a field the report no longer marks available supersedes an +// available state that existing carries from the same source: the source's +// later word on its own field wins, so a stale copy cannot keep claiming the +// field was collected. +func MergeReportedStatus(existing, reported *CollectionStatus) *CollectionStatus { + merged := MergeStatus(existing, reported) + if existing == nil || reported == nil { + return merged + } + merged.Serial = supersedeFieldStatus(merged.Serial, existing.Serial, reported.Serial) + merged.Temperature = supersedeFieldStatus(merged.Temperature, existing.Temperature, reported.Temperature) + merged.IO = supersedeFieldStatus(merged.IO, existing.IO, reported.IO) + merged.Controller = supersedeFieldStatus(merged.Controller, existing.Controller, reported.Controller) + merged.Pool = supersedeFieldStatus(merged.Pool, existing.Pool, reported.Pool) + return merged +} + +func supersedeFieldStatus(merged, existing, reported FieldStatus) FieldStatus { + source := strings.TrimSpace(existing.Source) + if existing.State == FieldAvailable && + reported.State != "" && reported.State != FieldAvailable && + source != "" && strings.EqualFold(source, strings.TrimSpace(reported.Source)) { + return reported + } + return merged +} + func mergeFieldStatus(existing, incoming FieldStatus) FieldStatus { if incoming.State == "" { return existing diff --git a/pkg/diskinventory/status_test.go b/pkg/diskinventory/status_test.go index 4c0bba2883..238e4b7097 100644 --- a/pkg/diskinventory/status_test.go +++ b/pkg/diskinventory/status_test.go @@ -24,3 +24,44 @@ func TestMergeStatusPrefersAvailableEvidencePerField(t *testing.T) { t.Fatalf("missing I/O state was not retained: %+v", got.IO) } } + +func TestMergeReportedStatusLetsTheSourceWithdrawItsOwnAvailability(t *testing.T) { + existing := &CollectionStatus{ + Serial: Available("smartctl"), + Temperature: Available("smartctl"), + IO: Available("kernel_diskstats"), + Pool: Available("proxmox_zfs"), + } + reported := &CollectionStatus{ + Serial: Available("smartctl"), + Temperature: Unavailable("smartctl", "host agent stopped reporting"), + IO: Unavailable("kernel_diskstats", "host agent stopped reporting"), + Pool: Unavailable("zpool", "command failed"), + } + + got := MergeReportedStatus(existing, reported) + if got.Temperature != reported.Temperature || got.IO != reported.IO { + t.Fatalf("the source's later word on its own fields was ignored: %+v", got) + } + if got.Serial != Available("smartctl") { + t.Fatalf("still-available field changed: %+v", got.Serial) + } + if got.Pool != Available("proxmox_zfs") { + t.Fatalf("another source's available evidence was withdrawn: %+v", got.Pool) + } + + // A copy that is already weaker never overrides a current report, and the + // rule is plain MergeStatus when either side is absent. + if got := MergeReportedStatus(reported, existing); got.Temperature != Available("smartctl") { + t.Fatalf("current report lost to a weaker copy: %+v", got.Temperature) + } + if got := MergeReportedStatus(nil, reported); got.Temperature != reported.Temperature { + t.Fatalf("nil existing: %+v", got) + } + if got := MergeReportedStatus(existing, nil); got.Temperature != existing.Temperature { + t.Fatalf("nil report: %+v", got) + } + if existing.Temperature != Available("smartctl") { + t.Fatal("MergeReportedStatus mutated its input") + } +}