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