From 1f2ff618b644188a2033058a599661678ec2fa34 Mon Sep 17 00:00:00 2001 From: rcourtman <8825017+rcourtman@users.noreply.github.com> Date: Wed, 7 Oct 2026 11:16:31 +0100 Subject: [PATCH] Own PVE disk alert history rows by recorded hardware PVE disk health and wearout alerts reference the disk by instance, node and device path, so a physical disk's history never showed its own SMART and wearout alerts. Binding the path reference failed review: a binding belongs to the reference string, so any disk that takes over the path inherits every row and every read of it. Own each lifecycle row instead. The alert now records the disk's WWN next to its serial, and the history writer puts the row under the disk the registry knows by that hardware identity, wherever it sits now. The registry only translates identity into a canonical ID, so a generation that still places another disk at the path cannot claim the row, and no check precedes a later write. A disk not yet in inventory, or removed, gets the ID the registry mints for its identity. Rows without usable or unambiguous identity stay under the reference, as do rows journaled before this change; journal rows are never rewritten. Reads by the reference still return every row raised under it, through the alert identifier index, and the incident projection keeps owned rows under their alert reference, so the Alerts history Resource action and incident timelines show the same occurrences as before. A read by the disk returns its own occurrences with their saved shells and notes. --- .../v6/internal/subsystems/ai-runtime.md | 22 ++ .../v6/internal/subsystems/alerts.md | 11 + .../v6/internal/subsystems/monitoring.md | 14 + .../internal/subsystems/unified-resources.md | 77 +++-- internal/ai/memory/incident_query.go | 84 +++++- .../memory/incident_query_disk_owner_test.go | 236 +++++++++++++++ internal/alerts/alerts_test.go | 31 ++ internal/alerts/disk_health.go | 14 +- .../monitoring/monitor_alert_handling_test.go | 70 +++++ internal/unifiedresources/history_identity.go | 9 +- .../unifiedresources/history_identity_test.go | 271 ++++++++++++++++-- internal/unifiedresources/incident_changes.go | 4 + internal/unifiedresources/monitor_adapter.go | 13 +- .../pve_disk_alert_history.go | 173 +++++++++++ internal/unifiedresources/store.go | 54 +++- 15 files changed, 1014 insertions(+), 69 deletions(-) create mode 100644 internal/ai/memory/incident_query_disk_owner_test.go create mode 100644 internal/unifiedresources/pve_disk_alert_history.go 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 1806e08bbc..9acdce84f7 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 d810dc84e0..734db0e1c2 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -5893,34 +5893,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 }