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.
This commit is contained in:
rcourtman
2026-10-07 11:16:31 +01:00
parent 8b36dec0a2
commit 1f2ff618b6
15 changed files with 1014 additions and 69 deletions
@@ -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
@@ -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
@@ -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.
@@ -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`,
`<instance>:<node>:disk:<device key>`) stay unbound, although
`<instance>:<node>:disk:<device key>`) 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.
+76 -8
View File
@@ -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)
}
@@ -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")
}
}
}
+31
View File
@@ -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])
}
}
}
+9 -5
View File
@@ -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,
}
}
@@ -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)
}
@@ -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)
}
}
@@ -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)
}
}
@@ -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"
+12 -1
View File
@@ -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 {
@@ -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
// "<reference>::<reference><suffix>" (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 "<instance>:<node>:disk:<key>"
// 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
}
+45 -9
View File
@@ -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
}