diff --git a/docs/release-control/v6/internal/subsystems/ai-runtime.md b/docs/release-control/v6/internal/subsystems/ai-runtime.md index ca23f1367c..47e5c4897e 100644 --- a/docs/release-control/v6/internal/subsystems/ai-runtime.md +++ b/docs/release-control/v6/internal/subsystems/ai-runtime.md @@ -138,6 +138,28 @@ status does not establish current resource health. Assistant handoff includes the latest bounded events, their source provenance and query coverage. It preserves the existing exclusion of raw command arguments/output. An absent closure does not acquire a duration measured up to the current time. +A PVE disk alert row written under the physical disk its hardware identity +names (`unifiedresources.OwnedAlertReference`) is matched to occurrences under +its alert's own reference, as when it was journaled there, while its evidence +keeps the owning disk. Reads by alert identifier and by that reference +therefore return the same occurrences as before ownership, including one +occurrence whose rows two disks own after one took over the path while +failing. The one difference is an owned row whose occurrence's firing the +read does not hold (another disk owns it, or retention removed it): before +any event is assigned, its recorded start takes the occurrence that starts +there, else the closest saved shell within the start tolerance, else opens +the occurrence under the ID the alert-centric read gives it, instead of +leaving the row as partial evidence. That boundary is not canonical, so a +truncated read still keeps events without a start away from it. A read by a +disk returns the occurrences its rows belong to and admits their saved +shells, so a note recorded from a disk read reaches the occurrence the +alert-centric read shows. The exception is two occurrences of one alert +starting within the shell tolerance: the alert-centric read assigns a saved +shell by firing order, which a disk read missing a firing cannot reproduce. +`TestIncidentQueryKeepsOccurrencesOfHardwareOwnedDiskAlerts` and +`TestIncidentQueryTruncatedReadKeepsOwnedOccurrencesGuarded` in +`internal/ai/memory/incident_query_disk_owner_test.go` compare both layouts, +the first on the memory and SQLite stores. ### Canonical Patrol and Assistant continuation, 2026-09-07 diff --git a/docs/release-control/v6/internal/subsystems/alerts.md b/docs/release-control/v6/internal/subsystems/alerts.md index 1e96e73aae..140987af27 100644 --- a/docs/release-control/v6/internal/subsystems/alerts.md +++ b/docs/release-control/v6/internal/subsystems/alerts.md @@ -2170,6 +2170,17 @@ wearout recovery rules. `TestProxmoxDiskCanonicalResourceIDTrimsIdentity` pins the persisted identity shape; monitoring's registry-backed `TestProxmoxPhysicalDiskMuteResolvesAndSuppressesWearoutAlert` pins the policy path. These are source invariants, not evidence of installed field relief. +Each PVE disk alert records the evaluated disk's serial and WWN +(`unifiedresources.MetadataDiskSerial`, `MetadataDiskWWN`), and resource +history owns each lifecycle row by that identity rather than by the path. A +row without a usable serial or WWN names only an identity-less disk at its +path, and a row whose identity is ambiguous, or names no disk but one another +resource already holds, stays under the path reference. +Reads by the path reference find those rows through the alert identifiers +`ProxmoxPhysicalDiskAlertIdentifiers` lists, so spec IDs and occurrence IDs +must keep matching it; +`TestCheckDiskHealthAlertsCarryHistoryOwnershipIdentity` in +`internal/alerts/alerts_test.go` pins both. Shared metric threshold runtime now lives in `internal/alerts/metric_runtime.go`. That file owns metric threshold lookup, per-metric delay and intent resolution, reducer input composition, active-alert diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 5ef5a18add..1019d8b72f 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -1075,6 +1075,20 @@ that changes one of these reference shapes must change Swarm service, ID-less container and service, ZFS pool and device, and host filesystem shapes against the real alert manager. +PVE disk health and wearout lifecycle events pass through the same writer but +never bind their path reference: each row goes to the physical disk its +recorded serial or WWN names, or stays under the reference, under the +unified-resources history identity clause. When a reboot reorders devices +before the registry catches up, a row that records a serial or WWN goes to the +disk the stale registry places at the path only if that disk reports the same +identity (the unified-resources clause lists where inventory merges or keys +disks differently); a row without either names only an identity-less disk at +its path. +`TestProxmoxDiskAlertTimelineFollowsRecordedHardware` in +`internal/monitoring/monitor_alert_handling_test.go` pins that path, and the +reference's incidents for the Alerts history Resource action, against the real +alert manager. + TrueNAS native alert projection preserves the trimmed, uppercase provider level in ResourceIncident.NativeSeverity. INFO and NOTICE retain the same canonical monitor risk; consumers must not lose their distinct actionability when projecting provider evidence. Native CRITICAL, ALERT, and EMERGENCY all project to canonical critical severity; EMERGENCY must not be discarded as unknown or make a still-active condition appear recovered. WARNING remains warning, and INFO and NOTICE remain informational at this projection boundary. Verification: `TestIncidentProjectionPreservesNativeSeverity` in `internal/truenas/provider_pool_health_contract_test.go` covers all seven native levels and case/whitespace normalization. `TestTrueNASNativeSeverityDispatch` in `internal/alerts/truenas_native_dispatch_test.go` verifies downstream INFO suppression, NOTICE preservation, notification severity, duplicate-poll retention, and confirmed recovery callback identity. The TrueNAS lifecycle tests in `internal/alerts/unified_incidents_test.go` require repeated EMERGENCY evidence to interrupt recovery confirmation. These are fixture-based projection and manager checks, not appliance ingestion or external notification-provider receipt proof. diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index 06c392b82e..bc3d781246 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -5927,34 +5927,63 @@ and `TestHistoryIdentityMonitorAdapterResolvesSubResourceReferences` in `internal/monitoring/monitor_alert_handling_test.go` against the real alert producers. PVE disk health and wearout alerts (`ProxmoxPhysicalDiskAlertResourceID`, -`::disk:`) stay unbound, although +`::disk:`) never bind, although `ResolveReferenceID` resolves the same reference to the physical disk at that path for operator mutes (#2112). A device path is not hardware identity: a replacement disk in the same slot, or a reboot that reorders devices, takes over the path, and disk alerts are evaluated before their poll's disks reach -the registry, so a lifecycle event can resolve against a generation that still -places another disk there. A binding belongs to the reference string and -moves to its latest target, carrying every row journaled under the reference -and every read of it. Bound, the path would put rows a replaced disk journaled -into its successor's history, and the Alerts history Resource action, which -reads incidents by the alert's own reference, would show the successor's -incidents for its predecessor's alert. Checking the serial each disk alert -records (`disk_serial`) narrows this but does not close it: an event rejected -for naming other hardware is still journaled under the reference and joins the -disk the reference is bound to, the check and the journal write are not -atomic, and an atomic check would still leave one binding owning every row -under the path. A disk's history therefore omits its own SMART health and -wearout alerts; they stay under the alert's reference, where incident -timelines by alert identifier and start time find them. The Resource action -reads the whole reference, so it lists every disk that has held the path, and -alert identifiers derive from the path too. Binding them needs ownership per -row, not per reference: each event written under the disk its recorded -hardware identity names, rows with missing, unusable or ambiguous identity -left under the reference rather than inferred from the current path, and a -decided read for the Resource action. `proxmoxDiskAlertMetadata` records the -serial but not the WWN, so WWN ownership also needs a producer change. Proof: -`TestHistoryIdentityLeavesProxmoxDiskAlertReferencesUnbound` in -`internal/unifiedresources/history_identity_test.go`. +the registry. A binding belongs to the reference string and would carry every +row and every read of the path to whichever disk held it last. Each lifecycle +row is owned on its own instead (`proxmoxDiskAlertOwner` in +`internal/unifiedresources/pve_disk_alert_history.go`). The alert records the +evaluated disk's serial and WWN (`disk_serial`, `disk_wwn`), and +`MonitorAdapter.RecordChange` writes the row under the one physical disk the +registry knows by that hardware identity (`diskinventory.HardwareIdentityMatch`), +wherever it sits now. A differing WWN does not rule a disk out, because the +registry keeps a merged agent observation's WWN, which smartctl frames +differently from PVE; of two matching disks, a WWN only one of them shares +with the row decides. The registry only translates identity into a +canonical ID: the path never breaks a tie, so a generation that still places +another disk at the path never gains a row that disk does not match by +identity, and no check precedes a later write. With no such disk in inventory +(a new disk before its first poll reaches the registry, or a removed one), the +row takes the canonical ID the registry mints for an identity it +knows by nothing else (`MachineIdentityCanonicalID`), unless another resource +holds that ID. A row without usable identity names only the one identity-less +disk at its path. Every other row, missing, unusable or ambiguous, stays under +the reference and is never retried. An owned row records its alert's +reference as `alert_resource_id`. A read by the reference returns the rows +journaled under it plus the rows owned away from it, matched through the +indexed alert identifier (`ProxmoxPhysicalDiskAlertIdentifiers`) and that +recorded reference, and every count uses the same predicate. The Alerts +history Resource action therefore still lists every occurrence raised under +the path, across the disks that held it, and alert-centric reads keep their +occurrences (see the AI runtime contract's incident-history queries). Rows +journaled under the reference before ownership existed stay there, +unrewritten, until the journal's 30-day retention removes them. The disk +drawer requests no change history, so ownership reaches the facets, timeline +and intelligence APIs, Assistant resource context and Patrol's scoped change +feed. History follows inventory where inventory merges: the registry merges +disks that report the same serial, and a disk whose poll reports no serial +keeps the serial last seen at its path (`preserveUnavailablePhysicalDiskEvidence`). +A disk the registry correlates with its agent's observation by path (an +identity-less or SAS agent disk) keeps that observation's ID and its serial +and WWN. When the identifiers PVE records for it share nothing with the +agent's (a SAS address reported as the serial, with no common WWN), or the +alert fired before the PVE observation reached the registry, the row carries +the minted ID: the disk's own history omits it, and reads by the reference +still return it. A disk the registry keys by its agent's WWN alone also +carries the serial PVE reports, so another disk reporting that serial, before +its own first poll reaches the registry, has its row written under the first +disk. Proof: +`TestHistoryIdentityNeverBindsProxmoxDiskAlertReferences`, +`TestProxmoxDiskAlertRowsFollowRecordedHardwareIdentity`, +`TestProxmoxDiskAlertOwnerDecisions` and +`TestProxmoxDiskAlertReferenceReadUsesIndexes` in +`internal/unifiedresources/history_identity_test.go`, and +`TestProxmoxDiskAlertTimelineFollowsRecordedHardware` in +`internal/monitoring/monitor_alert_handling_test.go` against the real alert +manager. That same shared timeline vocabulary now includes the `activity` change kind for provider-read breadcrumbs such as VMware tasks and events, plus the `vmware_adapter` source-adapter token for canonical provenance drill-down. diff --git a/internal/ai/memory/incident_query.go b/internal/ai/memory/incident_query.go index 8612ed4787..73350747e9 100644 --- a/internal/ai/memory/incident_query.go +++ b/internal/ai/memory/incident_query.go @@ -98,6 +98,40 @@ func (s *IncidentStore) QueryIncidents(query IncidentQuery) (IncidentPage, error } } } + // A lifecycle row that hardware ownership wrote under a physical disk + // stands for its alert's own reference in occurrence matching, as it did + // when journaled under it, so alert-centric reads keep their occurrences. + // Its evidence keeps the disk. A read by the disk admits the saved shells + // of the occurrences its rows belong to. + alertResource := func(change unifiedresources.ResourceChange) string { + if ref := unifiedresources.OwnedAlertReference(change); ref != "" { + return ref + } + return change.ResourceID + } + projected := func(change unifiedresources.ResourceChange) unifiedresources.ResourceChange { + change.ResourceID = alertResource(change) + return change + } + ownedStarts := make(map[string][]time.Time) + for _, change := range changes { + ref := unifiedresources.OwnedAlertReference(change) + if ref == "" { + continue + } + if started := incidentChangeStartedAt(change); !started.IsZero() { + key := ref + "\x00" + projectedAlertIdentifier(change) + ownedStarts[key] = append(ownedStarts[key], started) + } + } + ownedShell := func(shell *incidentShell) bool { + for _, started := range ownedStarts[shell.ResourceID+"\x00"+shell.AlertIdentifier] { + if incidentStartsMatch(shell.OpenedAt, started) { + return true + } + } + return false + } // Resource aliases are read identities. A shell for another resource must // not become an occurrence boundary merely because its alert ID matches. resourceKeys := make(map[string]string) @@ -106,7 +140,7 @@ func (s *IncidentStore) QueryIncidents(query IncidentQuery) (IncidentPage, error identityIDs = append(identityIDs, shell.ResourceID) } for _, change := range changes { - identityIDs = append(identityIDs, change.ResourceID) + identityIDs = append(identityIDs, alertResource(change)) } for _, id := range identityIDs { if _, known := resourceKeys[id]; known { @@ -133,7 +167,7 @@ func (s *IncidentStore) QueryIncidents(query IncidentQuery) (IncidentPage, error // Keep a stable existing ID and all local notes, not competing lifecycles. sort.Slice(shells, func(i, j int) bool { return shells[i].ID < shells[j].ID }) for _, shell := range shells { - if query.ResourceID != "" && !resourceIDs[shell.ResourceID] { + if query.ResourceID != "" && !resourceIDs[shell.ResourceID] && !ownedShell(shell) { continue } if query.AlertIdentifier != "" && shell.AlertIdentifier != query.AlertIdentifier { @@ -211,7 +245,7 @@ func (s *IncidentStore) QueryIncidents(query IncidentQuery) (IncidentPage, error var occurrence *Incident for _, candidate := range byAlert[identifier] { if canonicalBoundary[candidate] { - if sameResource(candidate.ResourceID, change.ResourceID) && boundaries[candidate].Equal(started) { + if sameResource(candidate.ResourceID, alertResource(change)) && boundaries[candidate].Equal(started) { occurrence = candidate break } @@ -225,11 +259,45 @@ func (s *IncidentStore) QueryIncidents(query IncidentQuery) (IncidentPage, error occurrence = &Incident{ID: canonicalIncidentID(identifier, started), AlertIdentifier: identifier, OpenedAt: started, Status: IncidentStatusUnknown} byAlert[identifier] = append(byAlert[identifier], occurrence) } - hydrateIncidentFromCanonicalChange(occurrence, change) + hydrateIncidentFromCanonicalChange(occurrence, projected(change)) boundaries[occurrence] = started firedOwners[change.ID] = occurrence canonicalBoundary[occurrence] = true } + // An owned row can belong to an occurrence whose firing this read does + // not hold: another disk owns the firing, or retention removed it. Its + // recorded start names the occurrence, so take the occurrence that + // starts there, else the closest saved shell within the start tolerance, + // else open it under the ID the alert-centric read gives it, before any + // event is assigned. The boundary is not canonical: a truncated read must + // not attach events without a start to it. + for _, change := range changes { + ref := unifiedresources.OwnedAlertReference(change) + identifier := projectedAlertIdentifier(change) + started := incidentChangeStartedAt(change) + if ref == "" || identifier == "" || started.IsZero() || change.Kind == unifiedresources.ChangeAlertFired { + continue + } + var occurrence *Incident + for _, candidate := range byAlert[identifier] { + if !sameResource(candidate.ResourceID, ref) { + continue + } + if boundaries[candidate].Equal(started) { + occurrence = candidate + break + } + if !canonicalBoundary[candidate] && incidentStartsMatch(candidate.OpenedAt, started) && + (occurrence == nil || incidentStartDelta(candidate.OpenedAt, started) < incidentStartDelta(occurrence.OpenedAt, started)) { + occurrence = candidate + } + } + if occurrence == nil { + occurrence = &Incident{ID: canonicalIncidentID(identifier, started), AlertIdentifier: identifier, ResourceID: ref, OpenedAt: started, Status: IncidentStatusUnknown} + byAlert[identifier] = append(byAlert[identifier], occurrence) + } + boundaries[occurrence] = started + } for _, occurrences := range byAlert { sort.Slice(occurrences, func(i, j int) bool { return boundaries[occurrences[i]].Before(boundaries[occurrences[j]]) }) } @@ -247,7 +315,7 @@ func (s *IncidentStore) QueryIncidents(query IncidentQuery) (IncidentPage, error explicitStart := incidentChangeStartedAt(change) if occurrence == nil && !explicitStart.IsZero() { for _, candidate := range byAlert[identifier] { - if sameResource(candidate.ResourceID, change.ResourceID) && boundaries[candidate].Equal(explicitStart) { + if sameResource(candidate.ResourceID, alertResource(change)) && boundaries[candidate].Equal(explicitStart) { occurrence = candidate break } @@ -255,7 +323,7 @@ func (s *IncidentStore) QueryIncidents(query IncidentQuery) (IncidentPage, error } if occurrence == nil && explicitStart.IsZero() { for _, candidate := range byAlert[identifier] { - if change.Kind != unifiedresources.ChangeCommandExecuted && change.Kind != unifiedresources.ChangeRunbookExecuted && !sameResource(candidate.ResourceID, change.ResourceID) { + if change.Kind != unifiedresources.ChangeCommandExecuted && change.Kind != unifiedresources.ChangeRunbookExecuted && !sameResource(candidate.ResourceID, alertResource(change)) { continue } // A capped read may have omitted a later firing after a saved shell. @@ -273,7 +341,7 @@ func (s *IncidentStore) QueryIncidents(query IncidentQuery) (IncidentPage, error // An event without a retained start cannot establish when the alert // opened. Preserve it as incomplete evidence with an unknown lifecycle. for _, candidate := range byAlert[identifier] { - if candidate.OpenedAt.IsZero() && sameResource(candidate.ResourceID, change.ResourceID) { + if candidate.OpenedAt.IsZero() && sameResource(candidate.ResourceID, alertResource(change)) { occurrence = candidate break } @@ -284,7 +352,7 @@ func (s *IncidentStore) QueryIncidents(query IncidentQuery) (IncidentPage, error } } openedAt := occurrence.OpenedAt - hydrateIncidentFromCanonicalChange(occurrence, change) + hydrateIncidentFromCanonicalChange(occurrence, projected(change)) occurrence.OpenedAt = openedAt canonicalEvents[occurrence] = append(canonicalEvents[occurrence], event) } diff --git a/internal/ai/memory/incident_query_disk_owner_test.go b/internal/ai/memory/incident_query_disk_owner_test.go new file mode 100644 index 0000000000..66d60dfc51 --- /dev/null +++ b/internal/ai/memory/incident_query_disk_owner_test.go @@ -0,0 +1,236 @@ +package memory + +import ( + "fmt" + "strings" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" + "github.com/stretchr/testify/require" +) + +// PVE disk alert lifecycle rows are written under the physical disk their +// recorded hardware identity names, recording the alert's own reference. +// Reads by the alert identifier and by the reference (the Alerts history +// Resource action) must project exactly what they projected when every row +// sat under the reference, while a read by a disk returns its own +// occurrences, with their saved shells. +func TestIncidentQueryKeepsOccurrencesOfHardwareOwnedDiskAlerts(t *testing.T) { + // The occurrence two disks own may have a saved shell: one opened at the + // alert's start, as lifecycle shells are, or one a little off it, as an + // imported shell can be, which reads match within a tolerance. + noShell := time.Duration(-1) + for _, tc := range []struct { + backend string + shellOffset time.Duration + }{{"memory", noShell}, {"sqlite", noShell}, {"memory", 0}, {"sqlite", 0}, {"memory", 500 * time.Millisecond}, {"sqlite", 500 * time.Millisecond}} { + backend, withSwappedShell, shellOffset := tc.backend, tc.shellOffset != noShell, tc.shellOffset + shell := "none" + if withSwappedShell { + shell = shellOffset.String() + } + t.Run(fmt.Sprintf("%s/swapped-shell=%s", backend, shell), func(t *testing.T) { + ref := unifiedresources.ProxmoxPhysicalDiskAlertResourceID("lab", "pve1", "/dev/sda") + identifier := unifiedresources.ProxmoxPhysicalDiskAlertIdentifiers(ref)[0] + base := time.Now().UTC().Add(-6 * time.Hour).Truncate(time.Second) + legacy, first, swapped, latest := base, base.Add(time.Hour), base.Add(2*time.Hour), base.Add(3*time.Hour) + type row struct { + id string + owner string + kind unifiedresources.ChangeKind + started time.Time + at time.Time + } + rows := []row{ + // Journaled under the reference before ownership existed. + {"legacy-fired", "", unifiedresources.ChangeAlertFired, legacy, legacy}, + {"legacy-resolved", "", unifiedresources.ChangeAlertResolved, legacy, legacy.Add(time.Minute)}, + {"first-fired", "physical_disk-a", unifiedresources.ChangeAlertFired, first, first}, + {"first-resolved", "physical_disk-a", unifiedresources.ChangeAlertResolved, first, first.Add(time.Minute)}, + // One occurrence spans two disks: B took the path while failing. + {"swapped-fired", "physical_disk-a", unifiedresources.ChangeAlertFired, swapped, swapped}, + {"swapped-resolved", "physical_disk-b", unifiedresources.ChangeAlertResolved, swapped, swapped.Add(5 * time.Minute)}, + {"latest-fired", "physical_disk-b", unifiedresources.ChangeAlertFired, latest, latest}, + {"latest-ack", "physical_disk-b", unifiedresources.ChangeAlertAcknowledged, latest, latest.Add(time.Minute)}, + } + newStore := func(owned bool) *IncidentStore { + var canonical unifiedresources.ResourceStore = unifiedresources.NewMemoryStore() + if backend == "sqlite" { + sqlite, err := unifiedresources.NewSQLiteResourceStore(t.TempDir(), "default") + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, sqlite.Close()) }) + canonical = sqlite + } + for _, r := range rows { + change := unifiedresources.BuildAlertTimelineChange(ref, r.kind, r.at, "", unifiedresources.AlertTimelineChange{ + AlertIdentifier: identifier, AlertStartedAt: r.started, AlertType: "disk-health", AlertLevel: "critical", + }) + change.ID = r.id + if owned && r.owner != "" { + change.ResourceID = r.owner + change.Metadata[unifiedresources.MetadataAlertResourceID] = ref + } + require.NoError(t, canonical.RecordChange(*change)) + } + // A command run on disk B for the latest occurrence, without a start. + command := unifiedresources.BuildCommandExecutionChange("physical_disk-b", identifier, "operator", "smartctl -a /dev/sda", true, "", nil) + command.ID = "latest-command" + command.ObservedAt = latest.Add(2 * time.Minute) + require.NoError(t, canonical.RecordChange(*command)) + store := NewIncidentStore(IncidentStoreConfig{}) + store.SetResourceTimelineStore(canonical) + store.incidents = []*incidentShell{{ + ID: "saved-first", AlertIdentifier: identifier, ResourceID: ref, OpenedAt: first, + Events: []IncidentEvent{{ID: "note", Type: IncidentEventNote, Timestamp: first.Add(30 * time.Second), Details: map[string]any{"note": "reseated the cable"}}}, + }} + if withSwappedShell { + store.incidents = append(store.incidents, &incidentShell{ID: "saved-swapped", AlertIdentifier: identifier, ResourceID: ref, OpenedAt: swapped.Add(shellOffset), + Events: []IncidentEvent{{ID: "swapped-note", Type: IncidentEventNote, Timestamp: swapped.Add(time.Minute), Details: map[string]any{"note": "disk B went in failing"}}}}) + } + return store + } + journaled, owned := newStore(false), newStore(true) + + type occurrence struct { + ID, ResourceID string + Status IncidentStatus + OpenedAt time.Time + Closed bool + Acknowledged bool + Events int + } + project := func(store *IncidentStore, query IncidentQuery) []occurrence { + page, err := store.QueryIncidents(query) + require.NoError(t, err) + out := make([]occurrence, 0, len(page.Incidents)) + for _, incident := range page.Incidents { + out = append(out, occurrence{incident.ID, incident.ResourceID, incident.Status, incident.OpenedAt, incident.ClosedAt != nil, incident.Acknowledged, len(incident.Events)}) + } + return out + } + for _, query := range []IncidentQuery{ + {AlertIdentifier: identifier, Limit: 10}, + {AlertIdentifier: identifier, StartedAt: swapped, Limit: 1}, + {ResourceID: ref, Limit: 10}, + } { + want := project(journaled, query) + require.Len(t, want, 1+3*boolInt(query.StartedAt.IsZero())) + require.Equal(t, want, project(owned, query), "%+v", query) + } + + starts := func(store *IncidentStore, resourceID string) map[time.Time]occurrence { + out := make(map[time.Time]occurrence) + for _, incident := range project(store, IncidentQuery{ResourceID: resourceID, Limit: 10}) { + out[incident.OpenedAt] = incident + } + return out + } + diskA := starts(owned, "physical_disk-a") + require.Len(t, diskA, 2) + require.Equal(t, "saved-first", diskA[first].ID, "the disk reads the saved occurrence, not a second projection") + require.Equal(t, IncidentStatusResolved, diskA[first].Status) + require.Equal(t, 3, diskA[first].Events, "fired, note, resolved") + require.Equal(t, ref, diskA[first].ResourceID) + diskAIDs := map[string]bool{} + for _, incident := range diskA { + diskAIDs[incident.ID] = true + } + diskB := starts(owned, "physical_disk-b") + require.Contains(t, diskB, latest) + require.True(t, diskB[latest].Acknowledged) + require.Equal(t, 3, diskB[latest].Events, "fired, acknowledged, and the command run after it") + require.NotContains(t, diskB, first) + require.NotContains(t, diskB, legacy) + require.NotContains(t, diskA, legacy) + + // B owns only the resolution of the occurrence it took over. Its read + // names that occurrence as the alert-centric read does, so a note + // recorded from the disk reaches the same occurrence. + swappedOccurrence := project(owned, IncidentQuery{AlertIdentifier: identifier, StartedAt: swapped, Limit: 1})[0] + if withSwappedShell { + require.Equal(t, "saved-swapped", swappedOccurrence.ID) + } + diskBSwapped := project(owned, IncidentQuery{ResourceID: "physical_disk-b", Limit: 10}) + require.Len(t, diskBSwapped, 2, "the latest occurrence and the swapped one, never a second copy of it") + require.Equal(t, swappedOccurrence.ID, diskBSwapped[1].ID) + require.Equal(t, 1+boolInt(withSwappedShell), diskBSwapped[1].Events, "B's resolution, with the saved note when there is a shell") + require.True(t, diskAIDs[swappedOccurrence.ID], "disk A reads the swapped occurrence under the same ID") + require.True(t, owned.RecordNote(identifier, diskBSwapped[1].ID, "replaced the failing disk", "operator")) + noted, err := owned.QueryIncidents(IncidentQuery{AlertIdentifier: identifier, StartedAt: swapped, Limit: 1}) + require.NoError(t, err) + require.Len(t, noted.Incidents, 1) + owners := map[string]string{} + for _, event := range noted.Incidents[0].Events { + if event.Evidence != nil { + owners[event.ID] = event.Evidence.ResourceID + } + } + require.Equal(t, map[string]string{"swapped-fired": "physical_disk-a", "swapped-resolved": "physical_disk-b"}, owners, + "evidence keeps the disk that owns each row") + require.Len(t, noted.Incidents[0].Events, 3+boolInt(withSwappedShell), "fired, resolved, the new note, and the saved note") + }) + } +} + +func boolInt(value bool) int { + if value { + return 1 + } + return 0 +} + +// A truncated read keeps the guard that stops events without a start from +// joining an occurrence a later saved shell may own. An owned row naming an +// older occurrence must not give such an event a destination it lacked when +// the rows sat under the reference. +func TestIncidentQueryTruncatedReadKeepsOwnedOccurrencesGuarded(t *testing.T) { + ref := unifiedresources.ProxmoxPhysicalDiskAlertResourceID("lab", "pve1", "/dev/sda") + identifier := unifiedresources.ProxmoxPhysicalDiskAlertIdentifiers(ref)[0] + older := time.Now().UTC().Add(-3 * time.Hour).Truncate(time.Second) + later := older.Add(time.Hour) + project := func(owned bool) []string { + canonical := unifiedresources.NewMemoryStore() + record := func(id, owner string, kind unifiedresources.ChangeKind, started, at time.Time) { + change := unifiedresources.BuildAlertTimelineChange(ref, kind, at, "", unifiedresources.AlertTimelineChange{AlertIdentifier: identifier, AlertStartedAt: started}) + change.ID = id + if owned { + change.ResourceID = owner + change.Metadata[unifiedresources.MetadataAlertResourceID] = ref + } + require.NoError(t, canonical.RecordChange(*change)) + } + record("older-fired", "physical_disk-a", unifiedresources.ChangeAlertFired, older, older) + record("later-fired", "physical_disk-a", unifiedresources.ChangeAlertFired, later, later) + command := unifiedresources.BuildCommandExecutionChange("node-pve1", identifier, "operator", "smartctl -a /dev/sda", true, "", nil) + command.ID = "command" + command.ObservedAt = later.Add(time.Minute) + require.NoError(t, canonical.RecordChange(*command)) + record("older-ack", "physical_disk-b", unifiedresources.ChangeAlertAcknowledged, older, later.Add(2*time.Minute)) + store := NewIncidentStore(IncidentStoreConfig{}) + store.SetResourceTimelineStore(canonical) + store.incidents = []*incidentShell{ + {ID: "saved-older", AlertIdentifier: identifier, ResourceID: ref, OpenedAt: older}, + {ID: "saved-later", AlertIdentifier: identifier, ResourceID: ref, OpenedAt: later}, + } + page, err := store.QueryIncidents(IncidentQuery{AlertIdentifier: identifier, ChangeLimit: 2, Limit: 10}) + require.NoError(t, err) + require.True(t, page.History.HasMoreChanges) + var out []string + for _, incident := range page.Incidents { + events := make([]string, 0, len(incident.Events)) + for _, event := range incident.Events { + events = append(events, event.ID) + } + out = append(out, fmt.Sprintf("%s %v", incident.ID, events)) + } + return out + } + journaled := project(false) + require.Equal(t, journaled, project(true)) + for _, incident := range journaled { + if strings.HasPrefix(incident, "saved-older") { + require.NotContains(t, incident, "command") + } + } +} diff --git a/internal/alerts/alerts_test.go b/internal/alerts/alerts_test.go index f67cbee911..f4c4463c5e 100644 --- a/internal/alerts/alerts_test.go +++ b/internal/alerts/alerts_test.go @@ -22111,3 +22111,34 @@ func TestStaleCleanupKeepsThresholdAlertStillBeingEvaluated(t *testing.T) { t.Fatal("cleanup kept an alert nobody has evaluated for over a day") } } + +// Resource history owns each PVE disk alert row by the hardware identity the +// alert records, and reads by the path reference find those rows by alert +// identifier. Both must match what CheckDiskHealth produces. +func TestCheckDiskHealthAlertsCarryHistoryOwnershipIdentity(t *testing.T) { + m := newTestManager(t) + m.ClearActiveAlerts() + disk := proxmox.Disk{DevPath: "/dev/sdb", Model: "Crucial MX500", Serial: "2117E59AB123", WWN: "0x5002538f12345678", + Type: "ssd", Health: "FAILED", Wearout: 3} + for i := 0; i < 3; i++ { + m.CheckDiskHealth("lab", "pve1", disk) + } + + ref := unifiedresources.ProxmoxPhysicalDiskAlertResourceID("lab", "pve1", disk.DevPath) + want := map[string]bool{} + for _, identifier := range unifiedresources.ProxmoxPhysicalDiskAlertIdentifiers(ref) { + want[identifier] = true + } + active := m.GetActiveAlerts() + if len(active) != len(want) { + t.Fatalf("active alerts = %d, want %d", len(active), len(want)) + } + for _, alert := range active { + if !want[alert.ID] || alert.ResourceID != ref { + t.Fatalf("alert %q on %q is not a history identifier of %q", alert.ID, alert.ResourceID, ref) + } + if alert.Metadata[unifiedresources.MetadataDiskSerial] != disk.Serial || alert.Metadata[unifiedresources.MetadataDiskWWN] != disk.WWN { + t.Fatalf("alert %q records serial %v and WWN %v", alert.ID, alert.Metadata[unifiedresources.MetadataDiskSerial], alert.Metadata[unifiedresources.MetadataDiskWWN]) + } + } +} diff --git a/internal/alerts/disk_health.go b/internal/alerts/disk_health.go index 5864403747..ae0e3d1610 100644 --- a/internal/alerts/disk_health.go +++ b/internal/alerts/disk_health.go @@ -16,13 +16,17 @@ func proxmoxDiskCanonicalResourceID(instance, node, devPath string) string { return unifiedresources.ProxmoxPhysicalDiskAlertResourceID(instance, node, devPath) } +// proxmoxDiskAlertMetadata records the evaluated disk's hardware identity +// with each alert. The alert reference names only the device path, so +// resource history owns each lifecycle row by this serial or WWN. func proxmoxDiskAlertMetadata(disk proxmox.Disk) map[string]interface{} { return map[string]interface{}{ - "disk_path": disk.DevPath, - "disk_model": disk.Model, - "disk_serial": disk.Serial, - "disk_type": disk.Type, - "disk_size": disk.Size, + "disk_path": disk.DevPath, + "disk_model": disk.Model, + unifiedresources.MetadataDiskSerial: disk.Serial, + unifiedresources.MetadataDiskWWN: disk.WWN, + "disk_type": disk.Type, + "disk_size": disk.Size, } } diff --git a/internal/monitoring/monitor_alert_handling_test.go b/internal/monitoring/monitor_alert_handling_test.go index 3a5b27f43f..4e2b4cf48a 100644 --- a/internal/monitoring/monitor_alert_handling_test.go +++ b/internal/monitoring/monitor_alert_handling_test.go @@ -1771,3 +1771,73 @@ func TestSilentLinkedAgentStopsRefreshingNodeTemperatureAlert(t *testing.T) { require.NotNil(t, poll()) require.True(t, temperatureAlert().LastSeen.After(held.LastSeen), "a fresh agent reading is evaluated") } + +// PVE disk alerts reference the disk by device path and record the evaluated +// disk's serial and WWN. Through the real alert manager, each lifecycle row +// joins the disk that identity names, even when a reboot has reordered the +// devices and the registry still places the other disk at the path. The +// Alerts history Resource action, which reads incidents by the alert's own +// reference, still lists that reference's occurrences. +func TestProxmoxDiskAlertTimelineFollowsRecordedHardware(t *testing.T) { + store, err := unifiedresources.NewSQLiteResourceStore(t.TempDir(), "default") + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, store.Close()) }) + manager := alerts.NewManagerWithDataDir(t.TempDir()) + t.Cleanup(manager.Stop) + config := manager.GetConfig() + config.Enabled = true + config.ActivationState = alerts.ActivationPending + config.SuppressionWindow = 0 + manager.UpdateConfig(config) + now := time.Now() + disk := func(device, serial, health string) models.PhysicalDisk { + return models.PhysicalDisk{ID: unifiedresources.ProxmoxPhysicalDiskSourceID("lab", "pve1", device, "", ""), Node: "pve1", Instance: "lab", + DevPath: device, Model: "Disk " + serial, Serial: serial, WWN: "0x5000c500" + serial, Type: "sata", Health: health, Wearout: -1, LastChecked: now} + } + registry := unifiedresources.NewRegistry(store) + registry.IngestSnapshot(models.StateSnapshot{PhysicalDisks: []models.PhysicalDisk{disk("/dev/sda", "ZA1A2B3C", "FAILED"), disk("/dev/sdb", "ZA4D5E6F", "PASSED")}}) + incidents := memory.NewIncidentStore(memory.IncidentStoreConfig{}) + incidents.SetResourceTimelineStore(store) + monitor := &Monitor{alertManager: manager, resourceStore: unifiedresources.NewMonitorAdapter(registry), incidentStore: incidents} + manager.SubscribeLifecycleCallback(monitor.handleAlertLifecycleEvent) + + manager.CheckDiskHealth("lab", "pve1", proxmoxDiskFromPhysicalDisk(disk("/dev/sda", "ZA1A2B3C", "FAILED"))) + // A reboot swaps the devices before the registry catches up: the failing + // disk now raises its alert at /dev/sdb, and the healthy one at /dev/sda + // resolves the alert the failing disk raised there. + manager.CheckDiskHealth("lab", "pve1", proxmoxDiskFromPhysicalDisk(disk("/dev/sdb", "ZA1A2B3C", "FAILED"))) + manager.CheckDiskHealth("lab", "pve1", proxmoxDiskFromPhysicalDisk(disk("/dev/sda", "ZA4D5E6F", "PASSED"))) + + diskIDs := map[string]string{} + for _, resource := range registry.List() { + diskIDs[resource.Name] = resource.ID + } + failing, healthy := diskIDs["Disk ZA1A2B3C"], diskIDs["Disk ZA4D5E6F"] + filters := unifiedresources.ResourceChangeFilters{Kinds: []unifiedresources.ChangeKind{unifiedresources.ChangeAlertFired, unifiedresources.ChangeAlertResolved}} + changes, err := store.GetRecentChangesFiltered(failing, time.Time{}, 10, filters) + require.NoError(t, err) + require.Len(t, changes, 3) + sda := unifiedresources.ProxmoxPhysicalDiskAlertResourceID("lab", "pve1", "/dev/sda") + sdb := unifiedresources.ProxmoxPhysicalDiskAlertResourceID("lab", "pve1", "/dev/sdb") + refs := map[string]int{} + for _, change := range changes { + require.Equal(t, failing, change.ResourceID) + refs[string(change.Kind)+" "+unifiedresources.OwnedAlertReference(change)]++ + } + require.Equal(t, map[string]int{"alert_fired " + sda: 1, "alert_resolved " + sda: 1, "alert_fired " + sdb: 1}, refs) + changes, err = store.GetRecentChangesFiltered(healthy, time.Time{}, 10, filters) + require.NoError(t, err) + require.Empty(t, changes) + + page, err := incidents.QueryIncidents(memory.IncidentQuery{ResourceID: sda, Limit: 10}) + require.NoError(t, err) + require.Len(t, page.Incidents, 1) + require.Equal(t, memory.IncidentStatusResolved, page.Incidents[0].Status) + require.Equal(t, sda, page.Incidents[0].ResourceID) + page, err = incidents.QueryIncidents(memory.IncidentQuery{ResourceID: failing, Limit: 10}) + require.NoError(t, err) + require.Len(t, page.Incidents, 2) + page, err = incidents.QueryIncidents(memory.IncidentQuery{ResourceID: healthy, Limit: 10}) + require.NoError(t, err) + require.Empty(t, page.Incidents) +} diff --git a/internal/unifiedresources/history_identity.go b/internal/unifiedresources/history_identity.go index c6d71b126b..3cb351ee8d 100644 --- a/internal/unifiedresources/history_identity.go +++ b/internal/unifiedresources/history_identity.go @@ -170,6 +170,10 @@ func (rr *ResourceRegistry) resolveHistoryReference(ref string) (resourceID stri resourceID = rr.dockerHistoryOwnerLocked(ref) return resourceID, resourceID != "" } + if isProxmoxPhysicalDiskAlertReference(ref) { + // A device path is not durable identity (proxmoxDiskAlertOwner). + return "", false + } matches := rr.durableHistoryMatchesLocked(ref) if ownerRef, ownerType, ok := historySubResourceOwner(ref); ok && len(matches) == 0 { // The owner must resolve uniquely and have the expected type. An owner @@ -326,7 +330,8 @@ func (b *legacyHistoryBackfill) add(ref string, now time.Time) { // unboundHistoryReferences lists alert journal references without a binding. // Docker container references and hostless service names are excluded: the // store migration binds full container IDs, and names or shortened IDs must -// never bind. +// never bind. PVE disk alert references never bind either; their rows are +// owned row by row (proxmoxDiskAlertOwner). func (s *SQLiteResourceStore) unboundHistoryReferences() ([]string, error) { rows, err := s.db.Query(`SELECT DISTINCT canonical_id FROM resource_changes WHERE kind GLOB 'alert_*' @@ -342,7 +347,7 @@ func (s *SQLiteResourceStore) unboundHistoryReferences() ([]string, error) { if err := rows.Scan(&ref); err != nil { return nil, err } - if !isDockerNameHistoryReference(ref) { + if !isDockerNameHistoryReference(ref) && !isProxmoxPhysicalDiskAlertReference(ref) { refs = append(refs, ref) } } diff --git a/internal/unifiedresources/history_identity_test.go b/internal/unifiedresources/history_identity_test.go index ee4304499c..03758b80bf 100644 --- a/internal/unifiedresources/history_identity_test.go +++ b/internal/unifiedresources/history_identity_test.go @@ -589,41 +589,272 @@ func TestHistoryIdentityMonitorAdapterResolvesSubResourceReferences(t *testing.T // device path. Operator mutes resolve that reference to the disk at the path // (#2112), but history never binds it: the path passes to a replacement disk // in the same slot, and a binding would carry every row journaled under the -// reference, and every read of the reference, to whichever disk held the path -// last. -func TestHistoryIdentityLeavesProxmoxDiskAlertReferencesUnbound(t *testing.T) { +// reference, and every read of it, to whichever disk held the path last. A +// live row goes to the disk its recorded hardware identity names; a row +// journaled under the reference before stays there, unrewritten. +func TestHistoryIdentityNeverBindsProxmoxDiskAlertReferences(t *testing.T) { store, err := NewSQLiteResourceStore(t.TempDir(), "default") require.NoError(t, err) t.Cleanup(func() { require.NoError(t, store.Close()) }) now := time.Now().UTC().Truncate(time.Second) ref := ProxmoxPhysicalDiskAlertResourceID("lab", "pve1", "/dev/sda") - event := func(id string, at time.Time) ResourceChange { - return ResourceChange{ID: id, ResourceID: ref, Kind: ChangeAlertFired, SourceType: SourceHeuristic, ObservedAt: at, - Metadata: map[string]any{"alert_type": "disk-health", "disk_serial": "ZA1A2B3C"}} - } // Journaled before this process started, then raised live. - require.NoError(t, store.RecordChange(event("legacy", now.Add(-time.Hour)))) + require.NoError(t, store.RecordChange(pveDiskAlertChange("legacy", "pve1", "/dev/sda", ChangeAlertFired, "ZA1A2B3C", "", now.Add(-time.Hour)))) snapshot := proxmoxHistoryIdentitySnapshot(now) - snapshot.PhysicalDisks = []models.PhysicalDisk{{ID: ProxmoxPhysicalDiskSourceID("lab", "pve1", "/dev/sda", "", ""), Node: "pve1", Instance: "lab", - DevPath: "/dev/sda", Model: "Seagate ST2000DM008", Serial: "ZA1A2B3C", Type: "hdd", Health: "FAILED", Wearout: -1, LastChecked: now}} + snapshot.PhysicalDisks = []models.PhysicalDisk{pveDiskHistoryTestDisk("pve1", "/dev/sda", "ZA1A2B3C", "", now)} adapter := NewMonitorAdapter(NewRegistry(store)) adapter.PopulateFromSnapshot(snapshot) registry := adapter.currentRegistry() diskID, ok := registry.ResolveReferenceID(ref) require.True(t, ok, "operator mutes resolve the reference to the disk") require.Equal(t, ResourceTypePhysicalDisk, registry.resources[diskID].Type) - require.NoError(t, adapter.RecordChange(event("live", now))) + require.NotContains(t, adapter.legacyHistory.pending, ref, "the reference is never retried for a binding") + require.NoError(t, adapter.RecordChange(pveDiskAlertChange("live", "pve1", "/dev/sda", ChangeAlertFired, "ZA1A2B3C", "", now))) _, found, err := store.ResolveHistorySourceIdentity(ref) require.NoError(t, err) require.False(t, found) - got, err := store.GetRecentChanges(ref, time.Time{}, 10) - require.NoError(t, err) - require.Len(t, got, 2) - for _, change := range got { - require.Equal(t, ref, change.ResourceID, change.ID) - } - got, err = store.GetRecentChangesFiltered(diskID, time.Time{}, 10, ResourceChangeFilters{Kinds: []ChangeKind{ChangeAlertFired}}) - require.NoError(t, err) - require.Empty(t, got) + byRef := pveDiskHistoryRows(t, store, ref) + require.ElementsMatch(t, []string{"legacy", "live"}, pveDiskHistoryIDs(byRef)) + require.Equal(t, ref, byRef["legacy"].ResourceID) + require.Equal(t, diskID, byRef["live"].ResourceID) + require.ElementsMatch(t, []string{"live"}, pveDiskHistoryIDs(pveDiskHistoryRows(t, store, diskID))) +} + +func pveDiskHistoryTestDisk(node, device, serial, wwn string, now time.Time) models.PhysicalDisk { + return models.PhysicalDisk{ID: ProxmoxPhysicalDiskSourceID("lab", node, device, "", ""), Node: node, Instance: "lab", + DevPath: device, Model: "Disk " + node + device, Serial: serial, WWN: wwn, Type: "sata", Health: "FAILED", Wearout: -1, LastChecked: now} +} + +// pveDiskAlertChange is a lifecycle row as the alert manager journals it: +// under the path reference, with the evaluated disk's hardware identity. +func pveDiskAlertChange(id, node, device string, kind ChangeKind, serial, wwn string, at time.Time) ResourceChange { + ref := ProxmoxPhysicalDiskAlertResourceID("lab", node, device) + change := BuildAlertTimelineChange(ref, kind, at, "", AlertTimelineChange{ + AlertIdentifier: ProxmoxPhysicalDiskAlertIdentifiers(ref)[0], AlertStartedAt: at, AlertType: "disk-health", + AlertMetadata: map[string]any{"disk_path": device, MetadataDiskSerial: serial, MetadataDiskWWN: wwn}, + }) + change.ID = id + return *change +} + +func pveDiskHistoryRows(t *testing.T, store ResourceStore, resourceID string) map[string]ResourceChange { + t.Helper() + changes, err := store.GetRecentChangesFiltered(resourceID, time.Time{}, 100, ResourceChangeFilters{Kinds: []ChangeKind{ChangeAlertFired}}) + require.NoError(t, err) + rows := make(map[string]ResourceChange, len(changes)) + for _, change := range changes { + rows[change.ID] = change + } + return rows +} + +func pveDiskHistoryIDs(rows map[string]ResourceChange) []string { + ids := make([]string, 0, len(rows)) + for id := range rows { + ids = append(ids, id) + } + return ids +} + +// Each PVE disk alert row belongs to the disk its recorded serial or WWN +// names, wherever that disk sits in the registry generation the event is +// evaluated against, and never to the disk the path names: a reboot reorder +// or a replacement disk before its first poll reaches the registry cannot +// claim another disk's history. Reads by the path reference still return +// every row raised under it. +func TestProxmoxDiskAlertRowsFollowRecordedHardwareIdentity(t *testing.T) { + for _, backend := range []string{"sqlite", "memory"} { + t.Run(backend, func(t *testing.T) { + var store ResourceStore = NewMemoryStore() + if backend == "sqlite" { + sqlite, err := NewSQLiteResourceStore(t.TempDir(), "default") + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, sqlite.Close()) }) + store = sqlite + } + now := time.Now().UTC().Truncate(time.Second) + const wwn = "0x5000c500a1b2c3d4" + snapshot := proxmoxHistoryIdentitySnapshot(now) + snapshot.PhysicalDisks = []models.PhysicalDisk{ + pveDiskHistoryTestDisk("pve1", "/dev/sda", "ZA1A2B3C", "", now), + pveDiskHistoryTestDisk("pve1", "/dev/sdb", "ZA4D5E6F", "", now), + pveDiskHistoryTestDisk("pve1", "/dev/sdc", "", wwn, now), + pveDiskHistoryTestDisk("pve1", "/dev/sdd", "", "", now), + } + adapter := NewMonitorAdapter(NewRegistry(store)) + adapter.PopulateFromSnapshot(snapshot) + registry := adapter.currentRegistry() + diskA := historyIdentityResourceID(t, registry, ResourceTypePhysicalDisk, "Disk pve1/dev/sda") + diskB := historyIdentityResourceID(t, registry, ResourceTypePhysicalDisk, "Disk pve1/dev/sdb") + diskW := historyIdentityResourceID(t, registry, ResourceTypePhysicalDisk, "Disk pve1/dev/sdc") + diskP := historyIdentityResourceID(t, registry, ResourceTypePhysicalDisk, "Disk pve1/dev/sdd") + sda := ProxmoxPhysicalDiskAlertResourceID("lab", "pve1", "/dev/sda") + sdc := ProxmoxPhysicalDiskAlertResourceID("lab", "pve1", "/dev/sdc") + sdd := ProxmoxPhysicalDiskAlertResourceID("lab", "pve1", "/dev/sdd") + replacement := MachineIdentityCanonicalID(ResourceTypePhysicalDisk, "ZC9X8Y7Z") + replacementWWN := MachineIdentityCanonicalID(ResourceTypePhysicalDisk, "0x5000c500ffeeddcc") + + want := map[string]string{} + record := func(change ResourceChange, owner string) { + t.Helper() + require.NoError(t, adapter.RecordChange(change)) + want[change.ID] = owner + } + record(pveDiskAlertChange("own", "pve1", "/dev/sda", ChangeAlertFired, "ZA1A2B3C", "", now), diskA) + // A reboot put B at /dev/sda; the registry still places it at /dev/sdb. + record(pveDiskAlertChange("reordered", "pve1", "/dev/sda", ChangeAlertFired, "ZA4D5E6F", "", now), diskB) + // A replacement disk fired before its first poll reached the registry. + record(pveDiskAlertChange("replacement", "pve1", "/dev/sda", ChangeAlertFired, "ZC9X8Y7Z", "", now), replacement) + record(pveDiskAlertChange("wwn", "pve1", "/dev/sdc", ChangeAlertFired, "", wwn, now), diskW) + record(pveDiskAlertChange("wwn-replacement", "pve1", "/dev/sdc", ChangeAlertFired, "", "0x5000c500ffeeddcc", now), replacementWWN) + record(pveDiskAlertChange("path-only", "pve1", "/dev/sdd", ChangeAlertFired, "", "", now), diskP) + // No usable identity, and the disk at the path reports one: the row + // stays under the reference rather than following the path. + record(pveDiskAlertChange("unidentified", "pve1", "/dev/sda", ChangeAlertFired, "", "", now), sda) + record(pveDiskAlertChange("placeholder", "pve1", "/dev/sda", ChangeAlertFired, "To Be Filled By O.E.M.", "", now), sda) + + all := pveDiskHistoryRows(t, store, "") + for id, owner := range want { + row, ok := all[id] + require.True(t, ok, id) + require.Equal(t, owner, row.ResourceID, id) + if strings.Contains(owner, ":disk:") { + require.Empty(t, OwnedAlertReference(row), id) + } else { + require.Equal(t, ProxmoxPhysicalDiskAlertResourceID("lab", "pve1", row.Metadata["disk_path"].(string)), OwnedAlertReference(row), id) + } + } + for _, ref := range []string{sda, sdc, sdd} { + _, bound, err := store.(resourceHistoryIdentityWriter).ResolveHistorySourceIdentity(ref) + require.NoError(t, err) + require.False(t, bound, "a path reference never binds: %s", ref) + } + require.ElementsMatch(t, []string{"own"}, pveDiskHistoryIDs(pveDiskHistoryRows(t, store, diskA))) + require.ElementsMatch(t, []string{"reordered"}, pveDiskHistoryIDs(pveDiskHistoryRows(t, store, diskB))) + require.ElementsMatch(t, []string{"wwn"}, pveDiskHistoryIDs(pveDiskHistoryRows(t, store, diskW))) + require.ElementsMatch(t, []string{"path-only"}, pveDiskHistoryIDs(pveDiskHistoryRows(t, store, diskP))) + + byRef := pveDiskHistoryRows(t, store, sda) + require.ElementsMatch(t, []string{"own", "reordered", "replacement", "unidentified", "placeholder"}, pveDiskHistoryIDs(byRef)) + require.Equal(t, diskB, byRef["reordered"].ResourceID, "a read by the reference returns rows where they are owned") + count, err := store.CountRecentChangesFiltered(sda, time.Time{}, ResourceChangeFilters{Kinds: []ChangeKind{ChangeAlertFired}}) + require.NoError(t, err) + require.Equal(t, len(byRef), count) + byKind, err := store.CountRecentChangesByKind(sda, time.Time{}) + require.NoError(t, err) + require.Equal(t, len(byRef), byKind[ChangeAlertFired]) + + // Once the replacement reaches the registry it owns the row it raised. + snapshot.PhysicalDisks[0] = pveDiskHistoryTestDisk("pve1", "/dev/sda", "ZC9X8Y7Z", "", now) + adapter.PopulateFromSnapshot(snapshot) + require.Equal(t, replacement, historyIdentityResourceID(t, adapter.currentRegistry(), ResourceTypePhysicalDisk, "Disk pve1/dev/sda")) + require.ElementsMatch(t, []string{"replacement"}, pveDiskHistoryIDs(pveDiskHistoryRows(t, store, replacement))) + }) + } +} + +// The registry only translates a row's recorded hardware identity into a +// canonical ID. A WWN only one of two matching disks shares decides between +// them, the path never breaks a tie, and a row without usable identity names +// only the one identity-less disk at its path. +func TestProxmoxDiskAlertOwnerDecisions(t *testing.T) { + registry := NewRegistry(nil) + disk := func(id, node, device, serial, wwn string) { + registry.resources[id] = &Resource{ID: id, Type: ResourceTypePhysicalDisk, + PhysicalDisk: &PhysicalDiskMeta{DevPath: device, Serial: serial, WWN: wwn}, + Proxmox: &ProxmoxData{Instance: "lab", NodeName: node}} + } + serialDisk := MachineIdentityCanonicalID(ResourceTypePhysicalDisk, "ZA1A2B3C") + disk(serialDisk, "pve1", "/dev/sda", "ZA1A2B3C", "") + disk("physical_disk-dup-pve1", "pve1", "/dev/sdb", "DUP00001", "") + disk("physical_disk-dup-pve2", "pve2", "/dev/sdb", "DUP00001", "") + disk("physical_disk-twin-1", "pve1", "/dev/sdg", "TWIN0001", "0x5000c500000000a1") + disk("physical_disk-twin-2", "pve1", "/dev/sdh", "TWIN0001", "0x5000c500000000b2") + // A RAID volume: PVE reports the full NAA as its serial and a truncated + // udev WWN, and the merged agent observation keeps the full naa. WWN. + disk("physical_disk-raid", "pve1", "/dev/sdi", "600508b1001c5c7a1b2c3d4e5f607080", "naa.600508b1001c5c7a1b2c3d4e5f607080") + // Merged with its agent's observation, whose smartctl WWN the registry + // keeps in a framing PVE does not use. + disk(MachineIdentityCanonicalID(ResourceTypePhysicalDisk, "ZB5C6D7E"), "pve1", "/dev/sdj", "ZB5C6D7E", "5-c500-da60ca43") + disk("physical_disk-wwn", "pve1", "/dev/sdc", "", "naa.5000c500a1b2c3d4") + disk("physical_disk-path", "pve1", "/dev/sdd", "", "") + disk("physical_disk-member-1", "pve1", "/dev/sde", "", "") + disk("physical_disk-member-2", "pve1", "/dev/sde", "", "") + disk(MachineIdentityCanonicalID(ResourceTypePhysicalDisk, "HELD0001"), "pve1", "/dev/sdf", "OTHER001", "") + + for _, tc := range []struct { + name, node, device, serial, wwn, want string + }{ + {"serial", "pve1", "/dev/sda", "ZA1A2B3C", "", serialDisk}, + {"serial after a move to another node", "pve2", "/dev/sdz", "ZA1A2B3C", "", serialDisk}, + // The path never breaks a tie: a stale generation could still place + // the other disk there. + {"shared serial", "pve1", "/dev/sdb", "DUP00001", "", ""}, + {"shared serial told apart by WWN", "pve1", "/dev/sdg", "TWIN0001", "naa.5000c500000000b2", "physical_disk-twin-2"}, + {"shared serial without WWN", "pve1", "/dev/sdg", "TWIN0001", "", ""}, + {"RAID volume with a truncated WWN", "pve1", "/dev/sdi", "600508b1001c5c7a1b2c3d4e5f607080", "0x600508b1001c5c7a", "physical_disk-raid"}, + {"agent-framed WWN on the merged disk", "pve1", "/dev/sdj", "ZB5C6D7E", "0x5000c500da60ca43", MachineIdentityCanonicalID(ResourceTypePhysicalDisk, "ZB5C6D7E")}, + {"WWN framing differs between reporters", "pve1", "/dev/sdc", "", "0x5000c500a1b2c3d4", "physical_disk-wwn"}, + {"unknown hardware", "pve1", "/dev/sda", "ZC9X8Y7Z", "", MachineIdentityCanonicalID(ResourceTypePhysicalDisk, "ZC9X8Y7Z")}, + {"minted ID held by other hardware", "pve1", "/dev/sda", "HELD0001", "", ""}, + {"no identity, identity-less disk at path", "pve1", "/dev/sdd", "", "", "physical_disk-path"}, + {"no identity, two disks at path", "pve1", "/dev/sde", "", "", ""}, + {"no identity, disk at path reports one", "pve1", "/dev/sda", "", "", ""}, + {"no identity, placeholder serial", "pve1", "/dev/sda", "0000000000000000", "", ""}, + {"no identity, nothing at path", "pve1", "/dev/sdz", "", "", ""}, + } { + ref := ProxmoxPhysicalDiskAlertResourceID("lab", tc.node, tc.device) + got := registry.proxmoxDiskAlertOwner(ref, map[string]any{MetadataDiskSerial: tc.serial, MetadataDiskWWN: tc.wwn}) + require.Equal(t, tc.want, got, tc.name) + } +} + +// A read by a PVE disk alert reference reaches the rows owned away from it +// through the alert identity index, not a journal scan. +func TestProxmoxDiskAlertReferenceReadUsesIndexes(t *testing.T) { + store, err := NewSQLiteResourceStore(t.TempDir(), "default") + require.NoError(t, err) + defer store.Close() + before := time.Now() + // The incident projection's read (memory.IncidentStore.QueryIncidents). + filters := ResourceChangeFilters{ObservedBefore: &before, Kinds: []ChangeKind{ + ChangeAlertFired, ChangeAlertResolved, ChangeAlertAcknowledged, ChangeAlertUnacknowledged, + ChangeAlertSnoozed, ChangeAlertUnsnoozed, ChangeCommandExecuted, ChangeRunbookExecuted, + }} + ref := ProxmoxPhysicalDiskAlertResourceID("lab", "pve1", "/dev/sda") + query, args := buildRecentChangeCountQuery([]string{ref}, before.Add(-resourceChangesRetention), filters, "EXPLAIN QUERY PLAN SELECT id FROM resource_changes", store.resourceChangesObservedAtExpr(), store.resourceChangesSourceTypeExpr(), store.resourceChangesSourceAdapterExpr()) + rows, err := store.db.Query(query, args...) + require.NoError(t, err) + defer rows.Close() + var plans []string + for rows.Next() { + var id, parent, unused int + var detail string + require.NoError(t, rows.Scan(&id, &parent, &unused, &detail)) + plans = append(plans, detail) + } + require.NoError(t, rows.Err()) + plan := strings.Join(plans, "\n") + require.Contains(t, plan, "idx_resource_changes_canonical_time") + require.Contains(t, plan, "idx_resource_changes_alert_time") + require.NotContains(t, plan, "SCAN resource_changes") +} + +func TestProxmoxPhysicalDiskAlertReferenceShape(t *testing.T) { + for _, ref := range []string{ + ProxmoxPhysicalDiskAlertResourceID("lab", "pve1", "/dev/sda"), + ProxmoxPhysicalDiskAlertResourceID("Production West", "pve5", "/dev/disk/by-id/ATA_DISK"), + ProxmoxPhysicalDiskAlertResourceID("lab", "pve1", "/"), + } { + require.True(t, isProxmoxPhysicalDiskAlertReference(ref), ref) + require.Len(t, ProxmoxPhysicalDiskAlertIdentifiers(ref), 2, ref) + } + for _, ref := range []string{ + ProxmoxPhysicalDiskAlertResourceID("lab", "pve1", ""), + "agent:host-1/disk:sda", "lab-pve1", "lab:pve1:101", "docker:host/disk:sda", "pve1:disk:dev-sda", "lab:pve1:disk:Dev-SDA", + } { + require.False(t, isProxmoxPhysicalDiskAlertReference(ref), ref) + require.Nil(t, ProxmoxPhysicalDiskAlertIdentifiers(ref), ref) + } } diff --git a/internal/unifiedresources/incident_changes.go b/internal/unifiedresources/incident_changes.go index 8d39473401..cd5f7ea3ea 100644 --- a/internal/unifiedresources/incident_changes.go +++ b/internal/unifiedresources/incident_changes.go @@ -19,6 +19,10 @@ const ( // MetadataAlertResolution is the reason code of a resolve that was not a // recovery, such as "moved_to_agent". Ordinary recoveries omit it. MetadataAlertResolution = "alert_resolution" + // MetadataAlertResourceID is the alert's own resource reference, recorded + // on a lifecycle row written under another resource because the row's + // hardware identity names it (see OwnedAlertReference). + MetadataAlertResourceID = "alert_resource_id" MetadataCommand = "command" MetadataSuccess = "success" MetadataOutputExcerpt = "output_excerpt" diff --git a/internal/unifiedresources/monitor_adapter.go b/internal/unifiedresources/monitor_adapter.go index 26a39e0958..6abf5dfdbf 100644 --- a/internal/unifiedresources/monitor_adapter.go +++ b/internal/unifiedresources/monitor_adapter.go @@ -199,13 +199,24 @@ func (a *MonitorAdapter) currentRegistry() *ResourceRegistry { // resource store when monitoring has a durable store attached. Alert lifecycle // events carry source-native references (Proxmox node and guest IDs, agent // IDs, Docker host and Swarm service IDs, sub-resource references); they -// resolve here so history reads by canonical ID include them. +// resolve here so history reads by canonical ID include them. PVE disk alert +// rows are owned row by row instead (proxmoxDiskAlertOwner). func (a *MonitorAdapter) RecordChange(change ResourceChange) error { registry := a.currentRegistry() if registry == nil || registry.store == nil { return nil } sourceRef := change.ResourceID + if ref, ok := proxmoxDiskAlertRowReference(change); ok { + // A PVE disk alert reference never binds: the row goes to the disk its + // own hardware identity names, or stays under the reference. + if owner := registry.proxmoxDiskAlertOwner(ref, change.Metadata); owner != "" && owner != ref { + change.Metadata = cloneChangeMetadata(change.Metadata) + change.Metadata[MetadataAlertResourceID] = ref + change.ResourceID = owner + } + return registry.store.RecordChange(change) + } history, hasHistory := registry.store.(resourceHistoryIdentityWriter) unbound := false if sourceID, derivedID, ok := legacyDockerHistoryIdentity(sourceRef); ok { diff --git a/internal/unifiedresources/pve_disk_alert_history.go b/internal/unifiedresources/pve_disk_alert_history.go new file mode 100644 index 0000000000..e2040b2eaf --- /dev/null +++ b/internal/unifiedresources/pve_disk_alert_history.go @@ -0,0 +1,173 @@ +package unifiedresources + +import ( + "strings" + + "github.com/rcourtman/pulse-go-rewrite/pkg/diskinventory" +) + +// PVE disk health and wearout alerts reference the disk by instance, node and +// device path (ProxmoxPhysicalDiskAlertResourceID). A device path is not +// hardware identity: a replacement disk in the same slot, or a reboot that +// reorders devices, takes it over, and the alerts are evaluated before their +// poll's disks reach the registry. History therefore never binds the +// reference. Each lifecycle row is owned instead by the disk its own recorded +// hardware identity names, and reads by the reference find those rows by alert +// identifier. +const ( + // MetadataDiskSerial and MetadataDiskWWN carry the evaluated disk's + // hardware identity on PVE disk alerts and their lifecycle rows. + MetadataDiskSerial = "disk_serial" + MetadataDiskWWN = "disk_wwn" +) + +// proxmoxDiskAlertSpecSuffixes name the PVE disk health and wearout alert +// specs. The alerts manager identifies each alert as +// "::" (buildCanonicalStateID over the spec ID). +var proxmoxDiskAlertSpecSuffixes = []string{"-health", "-wearout"} + +// ProxmoxPhysicalDiskAlertIdentifiers lists the alert identifiers of the PVE +// disk alerts raised under ref, or nil when ref is not shaped like a PVE disk +// alert reference. +func ProxmoxPhysicalDiskAlertIdentifiers(ref string) []string { + if !isProxmoxPhysicalDiskAlertReference(ref) { + return nil + } + identifiers := make([]string, 0, len(proxmoxDiskAlertSpecSuffixes)) + for _, suffix := range proxmoxDiskAlertSpecSuffixes { + identifiers = append(identifiers, ref+"::"+ref+suffix) + } + return identifiers +} + +// isProxmoxPhysicalDiskAlertReference matches "::disk:" +// with a non-empty device key (physicalDiskAlertKey). Instance names are free +// text; node names and device keys never contain a colon. +func isProxmoxPhysicalDiskAlertReference(ref string) bool { + scope, key, found := cutLast(ref, ":disk:") + if !found || key == "" { + return false + } + instance, node, found := cutLast(scope, ":") + if !found || strings.TrimSpace(instance) == "" || node == "" || strings.TrimSpace(node) != node { + return false + } + for _, r := range key { + if (r < 'a' || r > 'z') && (r < '0' || r > '9') && r != '.' && r != '-' { + return false + } + } + return true +} + +func cutLast(value, sep string) (before, after string, found bool) { + i := strings.LastIndex(value, sep) + if i < 0 { + return value, "", false + } + return value[:i], value[i+len(sep):], true +} + +// proxmoxDiskAlertRowReference reports the PVE disk alert reference of an +// alert lifecycle row journaled under it: the row's resource ID is the +// reference and its alert identifier is one of that reference's alerts. +func proxmoxDiskAlertRowReference(change ResourceChange) (string, bool) { + if !strings.HasPrefix(string(change.Kind), "alert_") { + return "", false + } + ref := CanonicalResourceID(change.ResourceID) + identifier, _ := change.Metadata[MetadataAlertIdentifier].(string) + for _, candidate := range ProxmoxPhysicalDiskAlertIdentifiers(ref) { + if identifier == candidate { + return ref, true + } + } + return "", false +} + +// OwnedAlertReference returns the alert's own resource reference recorded on +// a lifecycle row that hardware ownership wrote under another resource, or +// "" for every other row. The row's alert identifier must be one of that +// reference's alerts. +func OwnedAlertReference(change ResourceChange) string { + ref, _ := change.Metadata[MetadataAlertResourceID].(string) + if ref == "" || ref == CanonicalResourceID(change.ResourceID) { + return "" + } + identifier, _ := change.Metadata[MetadataAlertIdentifier].(string) + for _, candidate := range ProxmoxPhysicalDiskAlertIdentifiers(ref) { + if identifier == candidate { + return ref + } + } + return "" +} + +// proxmoxDiskAlertOwner names the physical disk that owns one PVE disk alert +// lifecycle row, from the hardware identity the row records, or "" to leave +// the row under its reference. The registry only translates that identity to +// a canonical ID; it never supplies identity from the path, so a generation +// that still places another disk at the path cannot claim the row. +// +// - A usable serial or WWN names the one disk the registry knows by the +// same hardware identity (diskinventory.HardwareIdentityMatch), wherever +// it sits now. A differing WWN does not rule a disk out: the registry +// keeps a merged agent observation's WWN, which smartctl frames +// differently from PVE, and merges disks that report the same serial. +// Of two matching disks, a WWN only one of them shares with the row +// decides; otherwise they are ambiguous. With none in inventory (a +// new disk before its first poll reaches the registry, or a removed one), +// the row goes to the canonical ID the registry mints for that identity, +// unless another resource already holds that ID. +// - A row without usable identity names a disk only by its path: the one +// disk at the path that reports no usable identity either. +func (rr *ResourceRegistry) proxmoxDiskAlertOwner(ref string, metadata map[string]any) string { + if rr == nil { + return "" + } + serial, _ := metadata[MetadataDiskSerial].(string) + wwn, _ := metadata[MetadataDiskWWN].(string) + serial, wwn = strings.TrimSpace(serial), strings.TrimSpace(wwn) + machineID := "" + if diskinventory.IsUsableHardwareID(serial) { + machineID = serial + } else if diskinventory.IsUsableHardwareID(wwn) { + machineID = wwn + } + + rr.mu.RLock() + defer rr.mu.RUnlock() + var matches, sameWWN []string + for id, resource := range rr.resources { + if resource == nil || CanonicalResourceType(resource.Type) != ResourceTypePhysicalDisk || resource.PhysicalDisk == nil { + continue + } + disk := resource.PhysicalDisk + if machineID != "" { + if !diskinventory.HardwareIdentityMatch(serial, wwn, disk.Serial, disk.WWN) { + continue + } + if diskinventory.HardwareIdentityMatch("", wwn, "", disk.WWN) { + sameWWN = append(sameWWN, id) + } + } else if resource.Proxmox == nil || strings.TrimSpace(disk.DevPath) == "" || + ProxmoxPhysicalDiskAlertResourceID(resource.Proxmox.Instance, resource.Proxmox.NodeName, disk.DevPath) != ref || + diskinventory.IsUsableHardwareID(disk.Serial) || diskinventory.IsUsableHardwareID(disk.WWN) { + continue + } + matches = append(matches, id) + } + switch { + case len(matches) == 1: + return matches[0] + case len(matches) > 1 && len(sameWWN) == 1: + return sameWWN[0] + case len(matches) > 1 || machineID == "": + return "" + } + derived := MachineIdentityCanonicalID(ResourceTypePhysicalDisk, machineID) + if rr.resources[derived] != nil { + return "" + } + return derived +} diff --git a/internal/unifiedresources/store.go b/internal/unifiedresources/store.go index 61c63ebfe5..06b03e9332 100644 --- a/internal/unifiedresources/store.go +++ b/internal/unifiedresources/store.go @@ -3745,17 +3745,50 @@ func appendRecentChangeResourceCondition(conditions []string, args []any, canoni args = append(args, id) } inClause := "(" + strings.Join(placeholders, ", ") + ")" - if !includeRelated { - return append(conditions, "canonical_id IN "+inClause), args - } - for _, id := range canonicalIDs { - args = append(args, id) - } - return append(conditions, `(canonical_id IN `+inClause+` OR EXISTS ( + condition := "canonical_id IN " + inClause + if includeRelated { + for _, id := range canonicalIDs { + args = append(args, id) + } + condition = `(canonical_id IN ` + inClause + ` OR EXISTS ( SELECT 1 FROM json_each(CASE WHEN json_valid(resource_changes.related_resources) THEN resource_changes.related_resources ELSE '[]' END) - WHERE TRIM(json_each.value) IN `+inClause+` - ))`), args + WHERE TRIM(json_each.value) IN ` + inClause + ` + ))` + } + if owned, ownedArgs := ownedAlertRowsCondition(canonicalIDs); owned != "" { + condition = "(" + condition + " OR " + owned + ")" + args = append(args, ownedArgs...) + } + return append(conditions, condition), args +} + +// ownedAlertRowsCondition matches the lifecycle rows hardware ownership wrote +// away from a PVE disk alert reference in the read's identity set. It filters +// on the indexed alert identifier first, then on the reference each owned row +// records, so a read by the reference returns exactly the rows journaled +// under it before. +func ownedAlertRowsCondition(canonicalIDs []string) (string, []any) { + metadata := "CASE WHEN json_valid(metadata_json) THEN metadata_json ELSE '{}' END" + var terms []string + var args []any + for _, id := range canonicalIDs { + identifiers := ProxmoxPhysicalDiskAlertIdentifiers(id) + if len(identifiers) == 0 { + continue + } + for _, identifier := range identifiers { + args = append(args, identifier) + } + args = append(args, id) + terms = append(terms, "(json_type("+metadata+", '$.alert_identifier') = 'text' AND "+resourceChangesAlertIdentifierExpr()+ + " IN ("+strings.TrimSuffix(strings.Repeat("?, ", len(identifiers)), ", ")+") AND json_extract("+metadata+", '$."+ + MetadataAlertResourceID+"') = ?)") + } + if len(terms) == 0 { + return "", nil + } + return "(" + strings.Join(terms, " OR ") + ")", args } func changeMatchesResource(change ResourceChange, canonicalIDs []string, includeRelated bool) bool { @@ -3771,6 +3804,9 @@ func changeMatchesResource(change ResourceChange, canonicalIDs []string, include if matches(change.ResourceID) { return true } + if ref := OwnedAlertReference(change); ref != "" && matches(ref) { + return true + } if !includeRelated { return false }