feat: add evidence-based pool health alerts

This commit is contained in:
rcourtman
2026-07-24 12:35:26 +01:00
parent 4a2335ce7f
commit 599c8e6349
51 changed files with 2856 additions and 198 deletions
@@ -656,10 +656,18 @@ declared floor.
path.
5. Storage and disk health:
Supported now for pools, datasets, and physical disks projected into the
shared storage and disk contracts, including SMART/disk-state risk, live
temperature, and recent temperature aggregates when the provider supplies
them. Out of scope: promising deeper TrueNAS-only topology or admin actions
beyond the current shared storage-health floor.
shared storage and disk contracts. The API-native floor preserves pool GUID,
state/detail, structured scrub or resilver progress, pool and vdev
read/write/checksum counters, mirror/RAIDZ/spare topology, path-only leaves,
and explicit missing-device evidence. `ONLINE`, `DEGRADED`, `FAULTED`,
`OFFLINE`, and `UNAVAIL` map into one canonical pool-health envelope while
retaining native evidence. Locked or unmounted datasets, failed app/runtime
state, SMART/disk-state risk, live temperature, and recent temperature
aggregates participate only when the provider supplies the corresponding
evidence. Unknown or absent telemetry never proves failure or recovery.
Out of scope: TrueNAS-only storage administration, automatic replacement,
pool mutation, or device-failure inference from an absent disk inventory
row.
6. Recovery:
Supported now as read-side visibility for TrueNAS snapshots and replication
artifacts in the shared recovery model, filters, rollups, and cross-surface
@@ -667,10 +675,15 @@ declared floor.
flow or promising provider-native restore/control actions that do not yet
exist on the governed shared path.
7. Alerts:
Supported now when TrueNAS systems, disks, and app parents participate in
the shared alert thresholds, incidents, and related-resource handoffs into
infrastructure, workloads, storage, and recovery. Out of scope: a separate
TrueNAS-only alert product surface.
Supported now when TrueNAS systems, pools, vdevs, datasets, disks, apps, and
app containers participate in the shared alert lifecycle, history,
suppression, and related-resource handoffs into infrastructure, workloads,
storage, and recovery. Provider-derived storage and workload conditions
require consecutive confirmation for activation and recovery, retain
connection-local identity across restart, and deduplicate equivalent native
TrueNAS alerts without suppressing distinct error evidence. Out of scope: a
second TrueNAS email-alert path or a separate TrueNAS-only alert product
surface.
8. Assistant read/control:
Supported now for canonical read-side resource access plus bounded app
control. Assistant may read TrueNAS app logs via `pulse_read`, read app
@@ -0,0 +1,78 @@
# Known RC Issue Closure For GA TrueNAS Pool Health Record
- Date: `2026-07-24`
- Gate: `known-rc-issue-closure-for-ga`
- Issue: `#1506`
- Result: `fixed-main-proof`
## Context
TrueNAS already projected pools, datasets, disks, apps, native alerts, SMART
state, temperature, boot-pool state, snapshot and replication evidence, and
connection-local identity. The pool adapter retained only the pool status word,
however. It discarded the native GUID, status detail, structured scrub or
resilver state, read/write/checksum counters, vdev topology, path-only leaves,
and explicit missing-member evidence. Synthetic degraded-pool incidents fired
and cleared on one poll, while locked/unmounted datasets and failed app
runtime state did not participate in the shared incident lifecycle. Native
Ceph health checks were flattened to one message.
## Disposition
TrueNAS `pool.query` and `boot.get_state` now preserve the complete native pool
report needed by the canonical storage-health contract. The provider projects
the full ZFS report plus one additive provider-neutral `PoolHealth` envelope.
The normalized state contract covers `ONLINE`, `DEGRADED`, `FAULTED`,
`OFFLINE`, and `UNAVAIL`; structured scrub and resilver work, terminal scan
errors, pool and vdev counters, path-only leaves, spare/mirror/RAIDZ topology,
and explicit native missing members remain distinct evidence.
Synthetic pool, vdev, dataset, disk, app, and app-container incidents require
two consecutive observations to activate and two healthy observations to
recover. Active state, acknowledgement, suppression, escalation, history, and
restart recovery remain in the shared alerts runtime. Missing resources and
unknown telemetry do not prove recovery. Native TrueNAS alerts suppress only
equivalent synthetic signals; distinct error evidence remains visible. Pulse
does not introduce a second TrueNAS email path.
Locked and unmounted datasets are availability conditions. Readonly alone is
not a fault, and receive-side replication targets classified by native
replication evidence remain healthy. Stopped and crashed apps plus
crashed/exited child containers participate through the canonical workload
incident path and can be suppressed for intentional downtime.
Ceph uses the same provider-neutral envelope only at cluster scope and only
from native `HEALTH_OK`, `HEALTH_WARN`, `HEALTH_ERR`, and native health-check
evidence. It does not infer a failed OSD, disk, ZFS leaf, or replacement target.
## Proof
- complete TrueNAS client/provider, storage-health, unified-resource, alerts,
monitoring, and API package suites
- provider shape coverage for pool GUID/detail, structured scan, errors,
mirror/spare/path-only topology, and explicit native missing-device evidence
- alert lifecycle coverage for transient activation, confirmed recovery,
severity escalation under one identity, acknowledgement/history continuity,
missing telemetry, and manager restart
- two-appliance poller coverage proving identical hostname, pool name, and
restored pool GUID do not merge connection-local resources
- API serialization coverage proving `PoolHealth`, full ZFS evidence, and
confirmation fields survive list pruning and canonical refresh
- frontend model coverage for actionable pool detail and native-alert
deduplication, plus the complete frontend TypeScript check
- current-build Chromium proof in managed mock mode: the TrueNAS storage page
rendered the degraded `archive` pool, and its shared resource drawer showed
canonical and native `DEGRADED`, the evidence-bounded recommendation,
`zfs_pool_state`, and `pool.query`; the browser console had zero errors
- issue-owned Go race tests and the v6 control-plane, status, registry, and
contract audits
## Outcome
The core degraded-pool, native fault-evidence, dataset, and TrueNAS app outcome
is complete on `main` for a future v6 release. It is not part of `v6.1.1`, and
no publication date is claimed. Production-batch disk provenance, guest
`lm-sensors` guidance, NVIDIA temperature collection, and Portainer-managed
container lifecycle remain separate follow-ups because the current TrueNAS API
evidence does not support those claims. Issue `#1506` remains open until a
release containing this change is available for reporter confirmation.
+22 -1
View File
@@ -1823,7 +1823,7 @@
},
{
"id": "RA20",
"summary": "Pulse's TrueNAS support claim is bounded to the current API-first floor: onboarding is through platform connections without requiring unified-agent bootstrap, TrueNAS projects into the canonical resource model and shared infrastructure/workload/storage/recovery/alert surfaces, Assistant support is read-first with bounded app control, and broader TrueNAS administration is not part of the current floor.",
"summary": "Pulse's TrueNAS support claim is bounded to the current API-first floor: onboarding is through platform connections without requiring unified-agent bootstrap; native pool, vdev, dataset, disk, app, and recovery evidence projects into the canonical resource, health, alert-lifecycle, history, and recommendation contracts without inferring failure or recovery from unknown telemetry; Assistant support is read-first with bounded app control; broader TrueNAS administration is not part of the current floor.",
"kind": "invariant",
"blocking_level": "release-ready",
"proof_type": "automated",
@@ -1854,6 +1854,22 @@
"-count=1"
]
},
{
"id": "truenas-pool-health-contract-tests",
"run": [
"go",
"test",
"./internal/alerts",
"./internal/api",
"./internal/monitoring",
"./internal/storagehealth",
"./internal/truenas",
"./internal/unifiedresources",
"-run",
"TestAssessZFSPool|TestCephBuildClusterPreservesNativeHealthChecks|TestCephUnknownHealthDoesNotInventIncident|TestCephUsesNativeClusterEvidenceThroughPoolHealthContract|TestNativeTrueNASPoolAlertSuppressesEquivalentSyntheticSignalsOnly|TestParsePoolStatePreservesScanTopologyErrorsAndNativeMissingEvidence|TestProviderProjectsFullZFSHealthAndActionableDatasetAppIncidents|TestResourceListPreservesCanonicalPoolHealthEvidence|TestTrueNASPollerKeepsPoolHealthConnectionLocalAcrossMatchingAppliances|TestUnifiedProviderIncident",
"-count=1"
]
},
{
"id": "truenas-runtime-floor-tests",
"run": [
@@ -1869,6 +1885,11 @@
}
],
"evidence": [
{
"repo": "pulse",
"path": "docs/release-control/v6/internal/records/known-rc-issue-closure-for-ga-truenas-pool-health-2026-07-24.md",
"kind": "file"
},
{
"repo": "pulse",
"path": "docs/release-control/v6/internal/SOURCE_OF_TRUTH.md",
@@ -140,6 +140,13 @@ tombstones, re-enrollment, or command authority. A linked agent remains
supplemental evidence on the retained canonical Proxmox member; its presence
does not decide provider membership or authorize removal.
The same shared model file may carry monitoring-owned native storage-health
evidence such as structured ZFS scans, vdev members, and Ceph health checks.
Those additive read-state fields do not change Unified Agent enrollment,
transport admission, token binding, command authority, deletion tombstones, or
re-enrollment continuity. Unknown or absent provider storage evidence must not
be reinterpreted as an agent lifecycle transition.
Docker / Podman report transport is shared with `api-contracts`.
`internal/dockeragent/agent.go` measures the gzip-encoded HTTP entity and the
decoded JSON before attempting any destination, while
@@ -1318,3 +1318,34 @@ precedence, normalization, exact zero/positive duration boundaries, VM/LXC
coverage, duplicate and delayed reports, wall-clock changes, suppression
reset, migration/restart continuity, backup hard-cap behavior, preview
immutability, and first-match lifecycle identity.
### Provider-observed storage and workload incident lifecycle
Provider observations use one stable alert identity composed from provider,
connection-local resource identity, native signal identity, and canonical
incident code. Synthetic TrueNAS pool, vdev, dataset, disk, app, and container
conditions require two consecutive observations before activation and two
consecutive healthy observations before recovery. A transient poll therefore
does not fire or clear an alert. Active state, acknowledgement, suppression,
severity changes, recovery, and history remain owned by the shared alerts
runtime and survive restart; an absent resource or unknown collection result is
not recovery evidence.
Native TrueNAS alerts remain authoritative for their equivalent condition.
Their stable native ID suppresses only the duplicate synthetic pool/vdev state,
missing-member, or scan condition; distinct read/write/checksum evidence
continues to participate. Pulse does not send a second TrueNAS email stream.
Intentional stopped apps and other stable conditions remain suppressible
through the canonical per-resource or global alert configuration, and readonly
replication targets do not produce incidents when native replication evidence
classifies them as healthy.
Ceph contributes one cluster-scoped provider incident only for native
`HEALTH_WARN` or `HEALTH_ERR` evidence. It shares the same confirmation,
recovery, history, and suppression machinery but cannot manufacture a disk
incident from cluster health.
`internal/alerts/unified_incident_confirmation_test.go`,
`internal/truenas/provider_pool_health_contract_test.go`, and
`internal/unifiedresources/ceph_pool_health_contract_test.go` prove the
lifecycle and deduplication matrix.
@@ -8375,3 +8375,24 @@ membership or telemetry read as deletion.
`internal/config/pve_instances_test.go` prove unreachable-member persistence,
evidence preservation, duplicate consolidation, and same-name-cluster
isolation.
### Canonical pool-health resource transport
`GET /api/resources` and `GET /api/resources/{id}` preserve the additive
provider-neutral `storage.poolHealth` or `ceph.poolHealth` envelope and the
complete native evidence beside it. TrueNAS storage payloads retain
`storage.zfsPool.scanDetails`, vdev role/parent/GUID/disk/path/state/missing
fields, and pool/device read/write/checksum counters. Ceph payloads retain the
native health-check code, severity, and summary list. Resource incidents retain
their stable provider/native identity plus activation and recovery confirmation
requirements.
List pruning, canonical metadata refresh, cloning, pagination, and frontend
decoding must not flatten those fields back to one status string. Unknown or
absent evidence stays explicit; the transport never synthesizes a failed disk,
replacement recommendation, or healthy recovery. Existing clients remain
compatible because the envelope, structured scan, topology fields, native
checks, and lifecycle counts are additive.
`internal/api/resources_pool_health_contract_test.go` and the frontend
type-check are the focused wire compatibility proofs.
@@ -5297,3 +5297,26 @@ The focused proofs are
`frontend-modern/src/components/Settings/ConnectionEditor/__tests__/AvailabilityTargetSlot.test.tsx`,
and
`frontend-modern/src/utils/__tests__/availabilityProbePresentation.test.ts`.
### Pool-health evidence presentation
TrueNAS storage detail composition presents canonical and native pool state,
structured scrub or resilver progress, pool error totals, affected vdev role
and topology, recommendation, evidence codes, and evidence source from the
canonical resource payload. It does not derive failed-disk claims from a
missing disk row or replace native path-only leaf identity with a guessed
device name.
TrueNAS alert presentation deduplicates a native provider signal projected onto
system, pool, and disk rows by provider, native ID, and code, preferring the
most specific canonical resource. Distinct evidence codes remain distinct
rows. All acknowledgement, suppression, history, and action handoffs continue
through shared alert primitives; no provider-only alert shell or email action
is introduced.
`frontend-modern/src/components/Infrastructure/__tests__/resourceDetailDrawerTrueNASModel.test.ts`
and
`frontend-modern/src/features/truenas/__tests__/truenasPageModel.test.ts`
are the focused presentation proofs. The governed browser proof must open the
shared resource detail surface and verify the same evidence labels in rendered
UI.
@@ -2242,3 +2242,30 @@ allowance is the only path that clears that lineage. Focused proofs are
`internal/monitoring/monitor_host_agents_test.go`, and
`internal/api/host_agent_removal_lifecycle_integration_test.go`; the concurrency
proof must also pass under the Go race detector.
### Native pool-health collection and appliance isolation
TrueNAS monitoring preserves the complete native `pool.query` observation
needed by the shared storage-health contract: pool GUID and status detail,
structured scrub or resilver state, pool and vdev read/write/checksum counters,
mirror/RAIDZ/spare topology, path-only leaves, and explicit native
missing/unavailable members. `disk.query` absence alone is not missing-disk
evidence. Unknown fields remain unknown and may not be converted into a failed
device, a recovered pool, or a zero-error observation.
The poller keys system and child source identity by configured connection.
Appliances with matching hostnames, restored pool GUIDs, or matching pool names
remain separate through refresh, cache rebuild, restart, and registry ingest.
Replication-target readonly classification remains a separate native-evidence
step and cannot hide locked or unmounted dataset state.
Ceph monitoring may enter the provider-neutral pool-health envelope only from
the native cluster health state and native health-check map. It preserves check
codes, severity, and summaries in deterministic order. A cluster-level
`HEALTH_WARN` or `HEALTH_ERR` does not identify a failed OSD or disk unless the
provider supplies that more specific evidence.
`internal/truenas/client_api_shapes_test.go`,
`internal/monitoring/truenas_poller_test.go`, and
`internal/monitoring/ceph_test.go` are the focused collection and identity
proofs.
@@ -2206,6 +2206,7 @@
"internal/alerts/operational_contract_test.go",
"internal/alerts/unified_eval_parity_test.go",
"internal/alerts/unified_eval_test.go",
"internal/alerts/unified_incident_confirmation_test.go",
"internal/alerts/unified_incidents_test.go"
]
},
@@ -2418,6 +2419,7 @@
"internal/alerts/resolved_lock_discipline_test.go",
"internal/alerts/synology_test.go",
"internal/alerts/threshold_resolution_shared_test.go",
"internal/alerts/unified_incident_confirmation_test.go",
"internal/alerts/update_alerts_test.go",
"internal/monitoring/monitor_alert_override_migration_test.go"
]
@@ -5282,8 +5284,10 @@
"exact_files": [
"internal/api/resources_test.go",
"internal/monitoring/monitor_polling_test.go",
"internal/truenas/client_api_shapes_test.go",
"internal/truenas/client_test.go",
"internal/truenas/contract_test.go",
"internal/truenas/provider_pool_health_contract_test.go",
"internal/truenas/provider_test.go",
"internal/truenas/transport_test.go"
]
@@ -5333,7 +5337,8 @@
"exact_files": [
"internal/api/resources_test.go",
"internal/storagehealth/risk_test.go",
"internal/storagehealth/topology_test.go"
"internal/storagehealth/topology_test.go",
"internal/storagehealth/zfs_pool_health_contract_test.go"
]
},
{
@@ -5623,6 +5628,7 @@
"internal/monitoring/availability_poller_test.go",
"internal/monitoring/availability_udp_test.go",
"internal/monitoring/canonical_guardrails_test.go",
"internal/monitoring/ceph_test.go",
"internal/monitoring/issue1485_unraid_lifecycle_test.go",
"internal/monitoring/issue1595_collection_trust_test.go",
"internal/monitoring/issue1613_contract_test.go",
@@ -5636,6 +5642,7 @@
"internal/monitoring/monitor_pve_cluster_refresh_test.go",
"internal/monitoring/monitor_pve_guest_lxc_test.go",
"internal/monitoring/ratetracker_test.go",
"internal/monitoring/truenas_poller_test.go",
"internal/unifiedresources/code_standards_test.go"
]
}
@@ -7150,6 +7157,7 @@
"internal/unifiedresources/availability_projection_test.go",
"internal/unifiedresources/canonical_identity_test.go",
"internal/unifiedresources/canonical_ids_types_test.go",
"internal/unifiedresources/ceph_pool_health_contract_test.go",
"internal/unifiedresources/code_standards_test.go",
"internal/unifiedresources/identity_test.go",
"internal/unifiedresources/ids_test.go",
@@ -7175,6 +7183,7 @@
"internal/monitoring/issue1595_collection_trust_test.go",
"internal/unifiedresources/adapter_coverage_test.go",
"internal/unifiedresources/adapters_test.go",
"internal/unifiedresources/ceph_pool_health_contract_test.go",
"internal/unifiedresources/host_storage_cleanup_test.go",
"internal/unifiedresources/monitor_adapter_read_state_test.go",
"internal/unifiedresources/views_test.go"
@@ -7297,6 +7306,7 @@
"frontend-modern/src/components/Infrastructure/__tests__/ResourceDetailDrawer.history.test.tsx",
"frontend-modern/src/components/Infrastructure/__tests__/resourceDetailDrawerIdentityModel.test.ts",
"frontend-modern/src/components/Infrastructure/__tests__/resourceDetailDrawerOperationalModel.test.ts",
"frontend-modern/src/components/Infrastructure/__tests__/resourceDetailDrawerTrueNASModel.test.ts",
"frontend-modern/src/components/Infrastructure/__tests__/resourceDetailMappers.test.ts",
"frontend-modern/src/components/Infrastructure/__tests__/ResourceOperatorStateSection.test.ts",
"frontend-modern/src/components/Infrastructure/__tests__/ResourcePolicySummary.test.tsx",
@@ -2369,6 +2369,24 @@ while storage detail drawers and filter controls must route summary series IDs,
source tones, and disk metrics through the shared storage helpers instead of
reconstructing them from local table state.
### Evidence-bounded pool and dataset health
Storage consumes the canonical `PoolHealth` envelope for normalized severity
and recommendations while retaining the full native ZFS report for operator
inspection. `ONLINE`, `DEGRADED`, `FAULTED`, `OFFLINE`, and `UNAVAIL`, active
scrub or resilver work, terminal scan errors, pool/vdev counters, path-only
leaves, spare/mirror/RAIDZ topology, and explicit missing members remain
distinct evidence. Storage must not translate unknown telemetry or
`disk.query` absence into failure, replacement, or recovery.
Locked and unmounted datasets are availability conditions. Readonly is not
itself a fault: receive-side replication targets classified by native
replication intent remain healthy. This classification changes storage health
only; it does not widen restore, mutation, replacement, or TrueNAS
administration authority. Ceph may share the normalized pool-health envelope
at cluster scope when native Ceph health evidence exists, but no ZFS leaf or
disk semantics transfer by analogy.
Unified Agent target cache prevention on the shared router is adjacent
lifecycle/API transport. `/api/agent/version` no-store headers and non-secret
request cache keys must not be interpreted as recovery freshness, retained
@@ -4098,6 +4098,31 @@ Shared infrastructure consumers such as the unified resource table and detail
drawer must present that owned metadata through shared helpers instead of
reconstructing privacy posture from display names, source types, or other
incidental runtime hints.
Canonical storage resources now carry a provider-neutral `PoolHealth` envelope
alongside the complete provider-native report. The envelope owns scope,
provider, native identity, canonical and native state, severity, summary,
recommendation, source, evidence codes, and observation time. ZFS
`StorageMeta.ZFSPool` remains the full native authority for scan progress,
topology, per-device states, and read/write/checksum counters; scalar ZFS fields
are compatibility summaries only. Clone and registry boundaries must retain
both shapes without sharing mutable slices or pointers.
The normalized state vocabulary is `ONLINE`, `DEGRADED`, `FAULTED`, `OFFLINE`,
`UNAVAIL`, and `UNKNOWN`. Provider adapters may map equivalent native
pool-health authority into that vocabulary, but may not infer a failed leaf,
missing disk, replacement target, or recovery from a cluster-level or absent
observation. Ceph therefore uses a cluster-scoped `PoolHealth` projection only
when native Ceph health evidence exists and preserves native check codes
separately.
Resource incidents may declare activation and recovery confirmation counts.
These fields describe shared alerts-owned lifecycle eligibility; they are not
provider-local alert state.
`internal/storagehealth/zfs_pool_health_contract_test.go`,
`internal/unifiedresources/ceph_pool_health_contract_test.go`, and
`internal/api/resources_pool_health_contract_test.go` pin the normalized and
wire-level contract.
That same shared-consumer boundary now also owns VMware phase-1 detail
presentation. `frontend-modern/src/components/Infrastructure/`
`resourceDetailDrawerVmwareModel.ts`,
@@ -198,6 +198,92 @@ describe('resourceDetailDrawerTrueNASModel', () => {
);
});
it('renders actionable canonical and native ZFS evidence without inventing a failed disk', () => {
const resource = baseResource({
type: 'storage',
displayName: 'tank',
status: 'degraded',
platformScopes: ['truenas'],
storage: {
type: 'zfs-pool',
topology: 'mirror',
platform: 'truenas',
zfsPoolState: 'DEGRADED',
poolHealth: {
scope: 'pool',
provider: 'truenas',
nativeId: 'pool-guid',
canonicalState: 'DEGRADED',
nativeState: 'DEGRADED',
severity: 'critical',
summary: 'Pool tank is degraded with a missing mirror member',
recommendation: 'Confirm the affected mirror member in TrueNAS before replacement.',
source: 'pool.query',
evidenceCodes: ['zfs_pool_state', 'zfs_device_missing'],
},
zfsPool: {
name: 'tank',
state: 'DEGRADED',
status: 'Degraded',
scan: 'RESILVER SCANNING',
scanDetails: {
function: 'RESILVER',
state: 'SCANNING',
percentage: 33.3,
},
readErrors: 1,
writeErrors: 0,
checksumErrors: 2,
devices: [
{
name: 'missing',
type: 'UNAVAIL_DISK',
role: 'data',
parent: 'mirror-0',
path: '/dev/disk/by-partuuid/missing',
state: 'UNAVAIL',
readErrors: 1,
writeErrors: 0,
checksumErrors: 0,
missing: true,
},
],
},
},
});
const healthRows = buildTrueNASDetailSections(resource).find(
(section) => section.label === 'Health',
)?.rows;
expect(healthRows).toEqual(
expect.arrayContaining([
expect.objectContaining({ label: 'Canonical state', value: 'DEGRADED' }),
expect.objectContaining({ label: 'Native state', value: 'DEGRADED' }),
expect.objectContaining({
label: 'Scan / resilver',
value: 'Resilver Scanning (33.3%)',
}),
expect.objectContaining({
label: 'ZFS errors',
value: 'Read 1 · Write 0 · Checksum 2',
}),
expect.objectContaining({
label: 'Affected vdevs',
value: '/dev/disk/by-partuuid/missing (Data / Unavail Disk): Missing · R 1 W 0 C 0',
}),
expect.objectContaining({
label: 'Recommended',
value: 'Confirm the affected mirror member in TrueNAS before replacement.',
}),
expect.objectContaining({
label: 'Evidence',
value: 'zfs_pool_state, zfs_device_missing',
}),
expect.objectContaining({ label: 'Evidence source', value: 'pool.query' }),
]),
);
});
it('summarizes native TrueNAS physical disk metadata for the detail drawer', () => {
const resource = baseResource({
type: 'physical_disk',
@@ -240,6 +240,55 @@ const storageProtectionLabel = (value?: string): string | null => {
return normalizeDelimitedLabel(protection);
};
const zfsScanLabel = (storage: ResourceStorageMeta): string | null => {
const scan = storage.zfsPool?.scanDetails;
if (!scan) return asString(storage.zfsPool?.scan);
const operation = normalizeDelimitedLabel(scan.function) ?? 'Scan';
const state = normalizeDelimitedLabel(scan.state);
const progress =
typeof scan.percentage === 'number' && Number.isFinite(scan.percentage) && scan.percentage > 0
? ` (${scan.percentage.toFixed(1)}%)`
: '';
const errors =
typeof scan.errors === 'number' && scan.errors > 0 ? ` · ${scan.errors} errors` : '';
return `${operation}${state ? ` ${state}` : ''}${progress}${errors}`;
};
const zfsErrorLabel = (storage: ResourceStorageMeta): string | null => {
const read = storage.zfsReadErrors ?? storage.zfsPool?.readErrors ?? 0;
const write = storage.zfsWriteErrors ?? storage.zfsPool?.writeErrors ?? 0;
const checksum = storage.zfsChecksumErrors ?? storage.zfsPool?.checksumErrors ?? 0;
if (read <= 0 && write <= 0 && checksum <= 0) return null;
return `Read ${read} · Write ${write} · Checksum ${checksum}`;
};
const zfsDeviceEvidenceLabels = (storage: ResourceStorageMeta): string[] =>
(storage.zfsPool?.devices ?? [])
.filter((device) => {
const state = device.state?.trim().toUpperCase();
return (
device.missing === true ||
(state !== '' && !['ONLINE', 'AVAIL', 'INUSE'].includes(state)) ||
device.readErrors > 0 ||
device.writeErrors > 0 ||
device.checksumErrors > 0
);
})
.map((device) => {
const name =
asString(device.disk) ?? asString(device.path) ?? asString(device.name) ?? 'Vdev';
const role = normalizeDelimitedLabel(device.role);
const type = normalizeDelimitedLabel(device.type);
const state = device.missing ? 'Missing' : (asString(device.state) ?? 'Unknown');
const context = [role, type].filter(Boolean).join(' / ');
const errors = [device.readErrors, device.writeErrors, device.checksumErrors].some(
(value) => value > 0,
)
? ` · R ${device.readErrors} W ${device.writeErrors} C ${device.checksumErrors}`
: '';
return `${name}${context ? ` (${context})` : ''}: ${state}${errors}`;
});
const buildTrueNASStorageSections = (
resource: Resource,
storage: ResourceStorageMeta,
@@ -247,6 +296,7 @@ const buildTrueNASStorageSections = (
const riskReasons = (storage.risk?.reasons ?? [])
.map((reason) => asString(reason.summary))
.filter((value): value is string => Boolean(value));
const zfsDeviceEvidence = zfsDeviceEvidenceLabels(storage);
const storageRows = compactRows([
row('Kind', storageKindLabel(storage)),
row('State', storageStateLabel(resource, storage), {
@@ -268,6 +318,17 @@ const buildTrueNASStorageSections = (
]);
const healthRows = compactRows([
row('Canonical state', asString(storage.poolHealth?.canonicalState), {
tone:
storage.poolHealth?.canonicalState === 'ONLINE'
? 'success'
: ['DEGRADED', 'FAULTED', 'OFFLINE', 'UNAVAIL'].includes(
storage.poolHealth?.canonicalState ?? '',
)
? 'warning'
: 'default',
}),
row('Native state', asString(storage.poolHealth?.nativeState)),
row('Risk', normalizeDelimitedLabel(storage.risk?.level), {
tone: storage.risk?.level?.toLowerCase() === 'warning' ? 'warning' : 'default',
}),
@@ -282,10 +343,25 @@ const buildTrueNASStorageSections = (
tone: storage.protectionSummary ? 'warning' : 'default',
}),
row('Rebuild', asString(storage.rebuildSummary)),
row('Scan / resilver', zfsScanLabel(storage), {
tone: storage.zfsPool?.scanDetails?.errors ? 'warning' : 'default',
}),
row('ZFS errors', zfsErrorLabel(storage), { tone: 'warning' }),
row('Affected vdevs', summarizeList(zfsDeviceEvidence, 2), {
title: zfsDeviceEvidence.join(', '),
tone: 'warning',
}),
row('Reasons', summarizeList(riskReasons, 2), {
title: riskReasons.join(', '),
tone: 'warning',
}),
row('Recommended', asString(storage.poolHealth?.recommendation), {
title: asString(storage.poolHealth?.summary) ?? undefined,
}),
row('Evidence', summarizeList(storage.poolHealth?.evidenceCodes ?? [], 4), {
title: (storage.poolHealth?.evidenceCodes ?? []).join(', '),
}),
row('Evidence source', asString(storage.poolHealth?.source)),
]);
return compactSections([
@@ -951,4 +951,42 @@ describe('truenasPageModel', () => {
'older-snapshot',
]);
});
it('deduplicates a native alert projected onto multiple resources and keeps the most specific subject', () => {
const nativeIncident = {
provider: 'truenas',
nativeId: 'native-volume-status',
code: 'truenas_volume_status',
severity: 'critical',
source: 'VolumeStatus',
summary: 'Pool tank is DEGRADED',
};
const rows = buildTrueNASPageModel([
makeResource({
id: 'nas-a',
type: 'agent',
incidents: [nativeIncident],
}),
makeResource({
id: 'tank',
type: 'storage',
incidents: [nativeIncident],
}),
makeResource({
id: 'sdb',
type: 'physical_disk',
displayName: 'sdb',
incidents: [nativeIncident],
}),
]).incidents;
expect(rows).toHaveLength(1);
expect(rows[0]).toEqual(
expect.objectContaining({
resourceId: 'sdb',
resourceName: 'sdb',
code: 'truenas_volume_status',
}),
);
});
});
@@ -995,12 +995,34 @@ const buildRollupIncidentRow = (resource: Resource): TrueNASIncidentRow => {
export function buildTrueNASIncidentRows(resources: Resource[]): TrueNASIncidentRow[] {
const rows: TrueNASIncidentRow[] = [];
const rowByNativeSignal = new Map<string, number>();
const specificity = (resource: Resource): number => {
if (resource.type === 'physical_disk') return 4;
if (resource.type === 'app-container') return 3;
if (resource.type === 'storage') return 2;
return 1;
};
for (const resource of resources) {
const incidents = (resource.incidents ?? []).filter(hasIncidentSignal);
if (incidents.length > 0) {
incidents.forEach((incident, index) =>
rows.push(buildIncidentRow(resource, incident, index)),
);
incidents.forEach((incident, index) => {
const row = buildIncidentRow(resource, incident, index);
const nativeId = asTrimmedString(incident.nativeId);
const provider = asTrimmedString(incident.provider);
const signalKey =
nativeId && provider
? `${provider.toLowerCase()}|${nativeId}|${row.code.toLowerCase()}`
: '';
const existingIndex = signalKey ? rowByNativeSignal.get(signalKey) : undefined;
if (existingIndex === undefined) {
if (signalKey) rowByNativeSignal.set(signalKey, rows.length);
rows.push(row);
return;
}
if (specificity(resource) > specificity(rows[existingIndex]!.resource)) {
rows[existingIndex] = row;
}
});
continue;
}
if (hasIncidentRollup(resource)) {
@@ -8,6 +8,7 @@ import { asTrimmedString } from '@/utils/stringUtils';
import { getGlobalWebSocketStore } from '@/stores/websocket-global';
import type {
Resource,
ResourceCephHealthCheckMeta,
ResourceAgentUnraidMeta,
ResourceCephMeta,
ResourceChange,
@@ -20,6 +21,7 @@ import type {
ResourcePBSMeta,
ResourcePolicyPostureSummary,
ResourcePhysicalDiskMeta,
ResourcePoolHealth,
ResourceStatus,
ResourceStorageMeta,
ResourceStorageRisk,
@@ -259,6 +261,7 @@ type APIResource = {
protectionSummary?: string;
rebuildInProgress?: boolean;
rebuildSummary?: string;
poolHealth?: ResourcePoolHealth;
commandsEnabled?: boolean;
tokenId?: string;
tokenName?: string;
@@ -414,6 +417,8 @@ type APIResource = {
fsid?: string;
healthStatus?: string;
healthMessage?: string;
healthChecks?: ResourceCephHealthCheckMeta[];
poolHealth?: ResourcePoolHealth;
numMons?: number;
numMgrs?: number;
numOsds?: number;
@@ -468,6 +473,8 @@ type APIResource = {
source?: string;
summary?: string;
startedAt?: string;
confirmationsRequired?: number;
recoveryConfirmationsRequired?: number;
}>;
incidentCount?: number;
incidentCode?: string;
@@ -767,6 +774,8 @@ const toResource = (v2: APIResource): Resource => {
source: incident.source,
summary: incident.summary || '',
startedAt: incident.startedAt,
confirmationsRequired: incident.confirmationsRequired,
recoveryConfirmationsRequired: incident.recoveryConfirmationsRequired,
}))
.filter((incident) => incident.code.trim() || incident.summary.trim()),
incidentCount: v2.incidentCount,
+19
View File
@@ -834,19 +834,38 @@ export interface ZFSPool {
state: string; // ONLINE, DEGRADED, FAULTED, OFFLINE, REMOVED, UNAVAIL
status: string; // Healthy, Degraded, Faulted, etc.
scan: string; // Current scan status (scrub, resilver, none)
scanDetails?: ZFSScan;
readErrors: number;
writeErrors: number;
checksumErrors: number;
devices: ZFSDevice[];
}
export interface ZFSScan {
function?: string;
state?: string;
percentage?: number;
errors?: number;
bytesExamined?: number;
bytesToProcess?: number;
totalSecondsRemaining?: number;
startedAt?: string;
endedAt?: string;
}
export interface ZFSDevice {
name: string;
type: string; // disk, mirror, raidz, raidz2, raidz3, spare, log, cache
role?: string;
parent?: string;
guid?: string;
disk?: string;
path?: string;
state: string; // ONLINE, DEGRADED, FAULTED, OFFLINE, REMOVED, UNAVAIL
readErrors: number;
writeErrors: number;
checksumErrors: number;
missing?: boolean;
message?: string;
}
+25
View File
@@ -145,6 +145,8 @@ export interface ResourceIncident {
source?: string;
summary: string;
startedAt?: string;
confirmationsRequired?: number;
recoveryConfirmationsRequired?: number;
}
// Identity information for deduplication
@@ -365,6 +367,20 @@ export interface ResourceStorageRisk {
reasons?: ResourceStorageRiskReason[];
}
export interface ResourcePoolHealth {
scope: string;
provider: string;
nativeId?: string;
canonicalState: string;
nativeState?: string;
severity: string;
summary?: string;
recommendation?: string;
source?: string;
evidenceCodes?: string[];
observedAt?: string;
}
export interface ResourceStorageMeta {
type?: string;
content?: string;
@@ -386,6 +402,7 @@ export interface ResourceStorageMeta {
protectionSummary?: string;
rebuildInProgress?: boolean;
rebuildSummary?: string;
poolHealth?: ResourcePoolHealth;
nodes?: string[];
pool?: string;
path?: string;
@@ -493,10 +510,18 @@ export interface ResourceCephServiceMeta {
total: number;
}
export interface ResourceCephHealthCheckMeta {
code: string;
severity?: string;
summary?: string;
}
export interface ResourceCephMeta {
fsid?: string;
healthStatus: string;
healthMessage?: string;
healthChecks?: ResourceCephHealthCheckMeta[];
poolHealth?: ResourcePoolHealth;
numMons: number;
numMgrs: number;
numOsds: number;
+6
View File
@@ -64,6 +64,9 @@ type Manager struct {
connectionDegradedCount map[string]int // Track consecutive degraded counts for platform connections (pve/pbs/pmg/vmware/truenas)
offlineConfirmations map[string]int // Track consecutive offline counts for all resources
offlineRecoveryConfirmations map[string]int // Track consecutive healthy confirmations before clearing poll-driven offline alerts
unifiedIncidentConfirmations map[string]int // Track consecutive provider-incident observations before activation
unifiedIncidentFirstSeen map[string]time.Time // Preserve the first confirmed observation as lifecycle start
unifiedIncidentRecoveries map[string]int // Track consecutive healthy observations before provider-incident recovery
dockerOfflineCount map[string]int // Track consecutive offline counts for Docker hosts
dockerStateConfirm map[string]int // Track consecutive state confirmations for Docker containers
dockerRestartTracking map[string]*dockerRestartRecord // Track restart counts and times for restart loop detection
@@ -146,6 +149,9 @@ func NewManagerWithDataDir(dataDir string) *Manager {
connectionDegradedCount: make(map[string]int),
offlineConfirmations: make(map[string]int),
offlineRecoveryConfirmations: make(map[string]int),
unifiedIncidentConfirmations: make(map[string]int),
unifiedIncidentFirstSeen: make(map[string]time.Time),
unifiedIncidentRecoveries: make(map[string]int),
dockerOfflineCount: make(map[string]int),
dockerStateConfirm: make(map[string]int),
dockerRestartTracking: make(map[string]*dockerRestartRecord),
+4
View File
@@ -490,6 +490,10 @@ func unifiedAlertResourceType(resource unifiedresources.Resource) (string, bool)
if resource.VMware != nil || resourceHasSource(resource, unifiedresources.SourceVMware) {
return "vmware-vm", true
}
case unifiedresources.ResourceTypeAppContainer:
if resource.TrueNAS != nil || resourceHasSource(resource, unifiedresources.SourceTrueNAS) {
return "truenas-app", true
}
case unifiedresources.ResourceTypeStorage:
if isTrueNASStorageResource(resource) {
switch strings.ToLower(strings.TrimSpace(resource.Storage.Topology)) {
@@ -0,0 +1,165 @@
package alerts
import (
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/storagehealth"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
)
func TestUnifiedProviderIncidentConfirmationRecoveryAndStableIdentity(t *testing.T) {
m := newTestManager(t)
configureUnifiedEvalManager(t, m, unifiedEvalBaseConfig())
observedAt := time.Date(2026, 7, 24, 10, 0, 0, 0, time.UTC)
resource := confirmedProviderIncidentResource(observedAt)
m.SyncUnifiedResourceIncidents([]unifiedresources.Resource{resource})
if alerts := m.GetActiveAlerts(); len(alerts) != 0 {
t.Fatalf("first transient observation must stay pending: %+v", alerts)
}
m.SyncUnifiedResourceIncidents([]unifiedresources.Resource{resource})
alerts := m.GetActiveAlerts()
if len(alerts) != 1 {
t.Fatalf("second observation must activate one alert: %+v", alerts)
}
alertID := alerts[0].ID
canonicalState := alerts[0].CanonicalState
if !alerts[0].StartTime.Equal(observedAt) {
t.Fatalf("start time = %s, want first observation %s", alerts[0].StartTime, observedAt)
}
escalated := resource
escalated.Incidents = append([]unifiedresources.ResourceIncident(nil), resource.Incidents...)
escalated.Incidents[0].Severity = storagehealth.RiskCritical
escalated.Incidents[0].Summary = "TrueNAS app media is crashed"
m.SyncUnifiedResourceIncidents([]unifiedresources.Resource{escalated})
alerts = m.GetActiveAlerts()
if len(alerts) != 1 || alerts[0].ID != alertID || alerts[0].CanonicalState != canonicalState || alerts[0].Level != AlertLevelCritical {
t.Fatalf("severity escalation must update one occurrence: %+v", alerts)
}
if history := m.GetAlertHistory(100); len(history) != 1 {
t.Fatalf("escalation must not create duplicate history: %+v", history)
}
m.SyncUnifiedResourceIncidents(nil)
if alerts := m.GetActiveAlerts(); len(alerts) != 1 {
t.Fatalf("missing provider telemetry is unknown, not recovery: %+v", alerts)
}
recovered := escalated
recovered.Incidents = nil
m.SyncUnifiedResourceIncidents([]unifiedresources.Resource{recovered})
if alerts := m.GetActiveAlerts(); len(alerts) != 1 {
t.Fatalf("first healthy observation must not resolve: %+v", alerts)
}
m.SyncUnifiedResourceIncidents([]unifiedresources.Resource{recovered})
if alerts := m.GetActiveAlerts(); len(alerts) != 0 {
t.Fatalf("second healthy observation must resolve: %+v", alerts)
}
if resolved := m.GetRecentlyResolved(); len(resolved) != 1 {
t.Fatalf("resolved lifecycle entry = %+v", resolved)
}
}
func TestUnifiedProviderIncidentRecoveryConfirmationSurvivesRestart(t *testing.T) {
dataDir := t.TempDir()
cfg := unifiedEvalBaseConfig()
observedAt := time.Date(2026, 7, 24, 10, 0, 0, 0, time.UTC)
resource := confirmedProviderIncidentResource(observedAt)
first := NewManagerWithDataDir(dataDir)
configureUnifiedEvalManager(t, first, cfg)
first.SyncUnifiedResourceIncidents([]unifiedresources.Resource{resource})
first.SyncUnifiedResourceIncidents([]unifiedresources.Resource{resource})
if len(first.GetActiveAlerts()) != 1 {
first.Stop()
t.Fatal("expected active incident before restart")
}
first.Stop()
second := NewManagerWithDataDir(dataDir)
t.Cleanup(second.Stop)
if len(second.GetActiveAlerts()) != 1 {
t.Fatal("persisted incident was not restored")
}
recovered := resource
recovered.Incidents = nil
second.SyncUnifiedResourceIncidents([]unifiedresources.Resource{recovered})
if len(second.GetActiveAlerts()) != 1 {
t.Fatal("restored incident resolved on first healthy poll")
}
second.SyncUnifiedResourceIncidents([]unifiedresources.Resource{recovered})
if len(second.GetActiveAlerts()) != 0 {
t.Fatal("restored incident did not resolve after confirmed recovery")
}
}
func TestUnifiedProviderIncidentPendingActivationRestartsSafely(t *testing.T) {
dataDir := t.TempDir()
cfg := unifiedEvalBaseConfig()
resource := confirmedProviderIncidentResource(time.Date(2026, 7, 24, 10, 0, 0, 0, time.UTC))
first := NewManagerWithDataDir(dataDir)
configureUnifiedEvalManager(t, first, cfg)
first.SyncUnifiedResourceIncidents([]unifiedresources.Resource{resource})
if len(first.GetActiveAlerts()) != 0 {
first.Stop()
t.Fatal("one observation activated an incident before restart")
}
first.Stop()
second := NewManagerWithDataDir(dataDir)
t.Cleanup(second.Stop)
second.SyncUnifiedResourceIncidents([]unifiedresources.Resource{resource})
if len(second.GetActiveAlerts()) != 0 {
t.Fatal("pending state from before restart caused a false activation")
}
second.SyncUnifiedResourceIncidents([]unifiedresources.Resource{resource})
if len(second.GetActiveAlerts()) != 1 {
t.Fatal("incident did not activate after two post-restart observations")
}
}
func TestUnifiedProviderIncidentGlobalDisableClearsImmediately(t *testing.T) {
m := newTestManager(t)
configureUnifiedEvalManager(t, m, unifiedEvalBaseConfig())
resource := confirmedProviderIncidentResource(time.Date(2026, 7, 24, 10, 0, 0, 0, time.UTC))
m.SyncUnifiedResourceIncidents([]unifiedresources.Resource{resource})
m.SyncUnifiedResourceIncidents([]unifiedresources.Resource{resource})
if len(m.GetActiveAlerts()) != 1 {
t.Fatal("expected active incident before global disable")
}
m.mu.Lock()
m.config.Enabled = false
m.mu.Unlock()
m.SyncUnifiedResourceIncidents([]unifiedresources.Resource{resource})
if len(m.GetActiveAlerts()) != 0 {
t.Fatal("global alert disable did not clear the provider incident")
}
}
func confirmedProviderIncidentResource(observedAt time.Time) unifiedresources.Resource {
return unifiedresources.Resource{
ID: "app:media",
Type: unifiedresources.ResourceTypeAppContainer,
Name: "media",
ParentName: "nas-a",
Sources: []unifiedresources.DataSource{unifiedresources.SourceTrueNAS},
TrueNAS: &unifiedresources.TrueNASData{Hostname: "nas-a"},
Incidents: []unifiedresources.ResourceIncident{{
Provider: "truenas",
NativeID: "app:media",
Code: "truenas_app_stopped",
Severity: storagehealth.RiskWarning,
Source: "app.query",
Summary: "TrueNAS app media is stopped",
StartedAt: observedAt,
ConfirmationsRequired: 2,
RecoveryConfirmationsRequired: 2,
}},
}
}
+103 -13
View File
@@ -21,6 +21,7 @@ func (m *Manager) SyncUnifiedResourceIncidents(resources []unifiedresources.Reso
}
desired := make(map[string]*Alert)
observedConditions := make(map[string]struct{})
resourcesByID := make(map[string]unifiedresources.Resource, len(resources))
for _, resource := range resources {
canonical := resource
@@ -55,7 +56,7 @@ func (m *Manager) SyncUnifiedResourceIncidents(resources []unifiedresources.Reso
m.mu.RUnlock()
if enabled {
now := time.Now()
now := m.now()
for _, spec := range canonicalSpecs {
if spec.Kind != alertspecs.AlertSpecKindProviderIncident {
continue
@@ -68,6 +69,20 @@ func (m *Manager) SyncUnifiedResourceIncidents(resources []unifiedresources.Reso
if !resourceSupportsUnifiedIncidentAlerts(resource) {
continue
}
incident, ok := incidentForProviderSpec(resource, spec)
if !ok {
continue
}
level, ok := alertLevelFromCanonicalSeverity(spec.Severity)
if !ok {
continue
}
alert := unifiedIncidentAlert(resource, incident, level, now)
applyCanonicalIdentity(alert, spec.ID, string(spec.Kind))
storageKey := canonicalTrackingKeyForSpec(spec, alert.ID)
observedConditions[storageKey] = struct{}{}
if alertType, ok := unifiedAlertResourceType(resource); ok {
if disableAllKubernetes && isUnifiedKubernetesAlertType(alertType) {
continue
@@ -89,24 +104,13 @@ func (m *Manager) SyncUnifiedResourceIncidents(resources []unifiedresources.Reso
continue
}
incident, ok := incidentForProviderSpec(resource, spec)
if !ok {
continue
}
level, ok := alertLevelFromCanonicalSeverity(spec.Severity)
if !ok {
continue
}
alert := unifiedIncidentAlert(resource, incident, level, now)
applyCanonicalIdentity(alert, spec.ID, string(spec.Kind))
if len(spec.SuppressionKeys) > 0 {
if alert.Metadata == nil {
alert.Metadata = map[string]interface{}{}
}
alert.Metadata["canonicalSuppressionKeys"] = append([]string(nil), spec.SuppressionKeys...)
}
desired[canonicalTrackingKeyForSpec(spec, alert.ID)] = alert
desired[storageKey] = alert
}
}
@@ -160,13 +164,75 @@ func (m *Manager) SyncUnifiedResourceIncidents(resources []unifiedresources.Reso
m.saveActiveAlertsAsync("availability intent state")
}
for storageKey, alert := range desired {
if existing, exists := m.getActiveAlertNoLock(storageKey); exists && existing != nil {
delete(m.unifiedIncidentConfirmations, storageKey)
delete(m.unifiedIncidentFirstSeen, storageKey)
delete(m.unifiedIncidentRecoveries, storageKey)
continue
}
required := alertMetadataInt(alert, "confirmationsRequired")
if required <= 1 {
continue
}
if _, ok := m.unifiedIncidentFirstSeen[storageKey]; !ok {
m.unifiedIncidentFirstSeen[storageKey] = alert.StartTime
}
m.unifiedIncidentConfirmations[storageKey]++
if m.unifiedIncidentConfirmations[storageKey] < required {
delete(desired, storageKey)
continue
}
if firstSeen := m.unifiedIncidentFirstSeen[storageKey]; !firstSeen.IsZero() {
alert.StartTime = firstSeen
}
delete(m.unifiedIncidentConfirmations, storageKey)
delete(m.unifiedIncidentFirstSeen, storageKey)
}
for storageKey := range m.unifiedIncidentConfirmations {
if _, observed := observedConditions[storageKey]; observed {
continue
}
delete(m.unifiedIncidentConfirmations, storageKey)
delete(m.unifiedIncidentFirstSeen, storageKey)
}
for storageKey, alert := range m.activeAlerts {
if !strings.HasPrefix(storageKey, unifiedresources.CanonicalResourceID(alert.ResourceID)+canonicalStateSeparator+"alertspec:provider-incident:") {
continue
}
if _, keep := desired[storageKey]; keep {
delete(m.unifiedIncidentRecoveries, storageKey)
continue
}
if _, stillObserved := observedConditions[storageKey]; stillObserved {
// Explicit alert disable, resource override, or parent/child
// suppression changes eligibility without asserting recovery.
delete(m.unifiedIncidentRecoveries, storageKey)
m.clearAlertNoLock(storageKey)
continue
}
if !enabled {
// Turning provider-incident evaluation off is an explicit policy
// change, not a provider recovery observation.
delete(m.unifiedIncidentRecoveries, storageKey)
m.clearAlertNoLock(storageKey)
continue
}
resourceID := unifiedresources.CanonicalResourceID(alert.ResourceID)
required := alertMetadataInt(alert, "recoveryConfirmationsRequired")
if required > 1 {
if _, resourceObserved := resourcesByID[resourceID]; !resourceObserved {
// Missing provider/resource telemetry is unknown, not recovery
// for incidents that opt into confirmed recovery.
continue
}
m.unifiedIncidentRecoveries[storageKey]++
if m.unifiedIncidentRecoveries[storageKey] < required {
continue
}
}
delete(m.unifiedIncidentRecoveries, storageKey)
m.clearAlertNoLock(storageKey)
}
@@ -194,6 +260,22 @@ func (m *Manager) SyncUnifiedResourceIncidents(resources []unifiedresources.Reso
}
}
func alertMetadataInt(alert *Alert, key string) int {
if alert == nil || alert.Metadata == nil {
return 0
}
switch value := alert.Metadata[key].(type) {
case int:
return value
case int64:
return int(value)
case float64:
return int(value)
default:
return 0
}
}
func isAvailabilityIncidentAlert(alert *Alert) bool {
return alertMetadataString(alert, "incidentCode") == "availability_unreachable"
}
@@ -302,6 +384,8 @@ func resourceSupportsUnifiedIncidentAlerts(resource unifiedresources.Resource) b
switch resource.Type {
case unifiedresources.ResourceTypeStorage, unifiedresources.ResourceTypePhysicalDisk:
return true
case unifiedresources.ResourceTypeCeph, unifiedresources.ResourceTypeAppContainer:
return len(resource.Incidents) > 0
case unifiedresources.ResourceTypeAgent:
return len(resource.Incidents) > 0
case unifiedresources.ResourceTypeVM:
@@ -628,6 +712,12 @@ func unifiedIncidentMetadata(resource unifiedresources.Resource, incident unifie
"incidentNativeID": incident.NativeID,
"incidentSource": incident.Source,
}
if incident.ConfirmationsRequired > 1 {
metadata["confirmationsRequired"] = incident.ConfirmationsRequired
}
if incident.RecoveryConfirmationsRequired > 1 {
metadata["recoveryConfirmationsRequired"] = incident.RecoveryConfirmationsRequired
}
if urgency, action := unifiedresources.IncidentActionForResource(&resource, incident, unifiedresources.IncidentCategoryForResource(&resource, incident)); urgency != "" {
metadata["incidentUrgency"] = urgency
if action != "" {
@@ -0,0 +1,115 @@
package api
import (
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/config"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
"github.com/rcourtman/pulse-go-rewrite/internal/storagehealth"
unified "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
)
func TestResourceListPreservesCanonicalPoolHealthEvidence(t *testing.T) {
observedAt := time.Date(2026, 7, 24, 10, 0, 0, 0, time.UTC)
cfg := &config.Config{DataPath: t.TempDir()}
h := NewResourceHandlers(cfg)
h.SetStateProvider(resourceUnifiedSeedProvider{
snapshot: models.StateSnapshot{LastUpdate: observedAt},
resources: []unified.Resource{{
ID: "truenas:nas-a:pool:pool-guid",
Type: unified.ResourceTypeStorage,
Name: "tank",
Status: unified.StatusWarning,
LastSeen: observedAt,
UpdatedAt: observedAt,
Sources: []unified.DataSource{unified.SourceTrueNAS},
Storage: &unified.StorageMeta{
Platform: "truenas",
Topology: "mirror",
ZFSPoolState: "DEGRADED",
PoolHealth: &unified.PoolHealth{
Scope: "pool",
Provider: "truenas",
NativeID: "pool-guid",
CanonicalState: "DEGRADED",
NativeState: "DEGRADED",
Severity: storagehealth.RiskCritical,
Summary: "Pool tank is degraded with a missing mirror member",
Recommendation: "Confirm the missing mirror member in TrueNAS before replacement.",
Source: "pool.query",
EvidenceCodes: []string{"zfs_pool_state", "zfs_device_missing"},
ObservedAt: observedAt,
},
ZFSPool: &models.ZFSPool{
Name: "tank",
State: "DEGRADED",
ScanDetails: &models.ZFSScan{
Function: "RESILVER",
State: "SCANNING",
Percentage: 33.3,
},
ReadErrors: 1,
Devices: []models.ZFSDevice{{
Name: "missing",
Type: "UNAVAIL_DISK",
Role: "data",
Parent: "mirror-0",
Path: "/dev/disk/by-partuuid/missing",
State: "UNAVAIL",
ReadErrors: 1,
Missing: true,
}},
},
},
Incidents: []unified.ResourceIncident{{
Provider: "truenas",
NativeID: "pool:pool-guid:vdev:missing",
Code: "zfs_device_missing",
Severity: storagehealth.RiskCritical,
Source: "pool.query",
Summary: "Pool tank has a missing data member",
ConfirmationsRequired: 2,
RecoveryConfirmationsRequired: 2,
}},
}},
})
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/api/resources?type=storage&page=1&limit=100", nil)
h.HandleListResources(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, body=%s", rec.Code, rec.Body.String())
}
var response ResourcesResponse
if err := json.NewDecoder(rec.Body).Decode(&response); err != nil {
t.Fatalf("decode response: %v", err)
}
if len(response.Data) != 1 {
t.Fatalf("resources = %+v", response.Data)
}
resource := response.Data[0]
if resource.Storage == nil || resource.Storage.PoolHealth == nil || resource.Storage.ZFSPool == nil {
t.Fatalf("pool-health envelope missing from API: %+v", resource.Storage)
}
if resource.Storage.PoolHealth.CanonicalState != "DEGRADED" ||
resource.Storage.PoolHealth.Source != "pool.query" ||
len(resource.Storage.PoolHealth.EvidenceCodes) != 2 {
t.Fatalf("pool-health evidence = %+v", resource.Storage.PoolHealth)
}
if resource.Storage.ZFSPool.ScanDetails == nil ||
resource.Storage.ZFSPool.ScanDetails.Function != "RESILVER" ||
len(resource.Storage.ZFSPool.Devices) != 1 ||
!resource.Storage.ZFSPool.Devices[0].Missing {
t.Fatalf("native ZFS report = %+v", resource.Storage.ZFSPool)
}
if len(resource.Incidents) != 1 ||
resource.Incidents[0].ConfirmationsRequired != 2 ||
resource.Incidents[0].RecoveryConfirmationsRequired != 2 {
t.Fatalf("incident lifecycle contract = %+v", resource.Incidents)
}
}
+19 -1
View File
@@ -136,7 +136,7 @@ func clonePlatformFixtures(in PlatformFixtures) PlatformFixtures {
func cloneTrueNASFixtureSnapshot(in truenas.FixtureSnapshot) truenas.FixtureSnapshot {
out := in
out.System = cloneTrueNASSystemInfo(in.System)
out.Pools = append([]truenas.Pool(nil), in.Pools...)
out.Pools = cloneTrueNASPools(in.Pools)
out.Datasets = append([]truenas.Dataset(nil), in.Datasets...)
out.Disks = append([]truenas.Disk(nil), in.Disks...)
out.Alerts = append([]truenas.Alert(nil), in.Alerts...)
@@ -146,6 +146,24 @@ func cloneTrueNASFixtureSnapshot(in truenas.FixtureSnapshot) truenas.FixtureSnap
return out
}
func cloneTrueNASPools(in []truenas.Pool) []truenas.Pool {
if in == nil {
return nil
}
out := make([]truenas.Pool, len(in))
for i := range in {
out[i] = in[i]
if in[i].Scan != nil {
scan := *in[i].Scan
out[i].Scan = &scan
}
out[i].VDevs = append([]truenas.PoolVDev(nil), in[i].VDevs...)
out[i].DiskMembers = append([]truenas.PoolDiskMember(nil), in[i].DiskMembers...)
}
return out
}
func cloneTrueNASSystemInfo(in truenas.SystemInfo) truenas.SystemInfo {
out := in
if len(in.TemperatureCelsius) > 0 {
+30
View File
@@ -8,6 +8,7 @@ import (
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/operationaltrust"
"github.com/rcourtman/pulse-go-rewrite/internal/truenas"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
)
@@ -562,3 +563,32 @@ func TestBuildFixtureGraphRefreshesStateMetricsFromCanonicalModel(t *testing.T)
t.Fatalf("expected refreshed physical disk temperature %d, got %d", want, got)
}
}
func TestCloneTrueNASFixtureSnapshotIsolatesPoolHealthEvidence(t *testing.T) {
original := truenas.FixtureSnapshot{Pools: []truenas.Pool{{
Name: "tank",
Scan: &truenas.PoolScan{
Function: "RESILVER",
State: "SCANNING",
},
VDevs: []truenas.PoolVDev{{
GUID: "leaf-guid",
Status: "UNAVAIL",
}},
DiskMembers: []truenas.PoolDiskMember{{
GUID: "leaf-guid",
Status: "UNAVAIL",
}},
}}}
cloned := cloneTrueNASFixtureSnapshot(original)
cloned.Pools[0].Scan.State = "FINISHED"
cloned.Pools[0].VDevs[0].Status = "ONLINE"
cloned.Pools[0].DiskMembers[0].Status = "ONLINE"
if original.Pools[0].Scan.State != "SCANNING" ||
original.Pools[0].VDevs[0].Status != "UNAVAIL" ||
original.Pools[0].DiskMembers[0].Status != "UNAVAIL" {
t.Fatalf("clone mutated original pool evidence: %+v", original.Pools[0])
}
}
+3
View File
@@ -112,6 +112,9 @@ func supplementCephCluster(primary, supplemental CephCluster) CephCluster {
if primary.HealthMessage == "" {
primary.HealthMessage = supplemental.HealthMessage
}
if len(primary.HealthChecks) == 0 && len(supplemental.HealthChecks) > 0 {
primary.HealthChecks = append([]CephHealthCheck(nil), supplemental.HealthChecks...)
}
if primary.TotalBytes == 0 && supplemental.TotalBytes != 0 {
primary.TotalBytes = supplemental.TotalBytes
primary.UsedBytes = supplemental.UsedBytes
+34
View File
@@ -175,3 +175,37 @@ func TestUnknownGuestRatesRemainNumericOnAPIAndWebsocketShapes(t *testing.T) {
t.Fatalf("internal validity leaked into wire payload: %s", wire)
}
}
func TestNativePoolHealthEvidenceNormalizesWithoutChangingIdentity(t *testing.T) {
pool := (ZFSPool{
Name: "tank",
State: "DEGRADED",
ScanDetails: &ZFSScan{
Function: "RESILVER",
State: "SCANNING",
Percentage: 42.5,
},
Devices: []ZFSDevice{{
GUID: "leaf-guid",
Path: "/dev/disk/by-partuuid/example",
State: "UNAVAIL",
Missing: true,
}},
}).NormalizeCollections()
if pool.Name != "tank" || pool.ScanDetails == nil || len(pool.Devices) != 1 {
t.Fatalf("normalized ZFS evidence = %+v", pool)
}
cluster := (CephCluster{
FSID: "cluster-fsid",
Health: "HEALTH_WARN",
HealthChecks: []CephHealthCheck{{
Code: "OSD_DOWN",
Severity: "HEALTH_WARN",
Summary: "one OSD is down",
}},
}).NormalizeCollections()
if cluster.FSID != "cluster-fsid" || len(cluster.HealthChecks) != 1 || cluster.HealthChecks[0].Code != "OSD_DOWN" {
t.Fatalf("normalized Ceph evidence = %+v", cluster)
}
}
+35 -1
View File
@@ -2330,6 +2330,7 @@ type ZFSPool struct {
State string `json:"state"` // ONLINE, DEGRADED, FAULTED, OFFLINE, REMOVED, UNAVAIL
Status string `json:"status"` // Healthy, Degraded, Faulted, etc.
Scan string `json:"scan"` // Current scan status (scrub, resilver, none)
ScanDetails *ZFSScan `json:"scanDetails,omitempty"`
ReadErrors int64 `json:"readErrors"`
WriteErrors int64 `json:"writeErrors"`
ChecksumErrors int64 `json:"checksumErrors"`
@@ -2343,14 +2344,35 @@ func (p ZFSPool) NormalizeCollections() ZFSPool {
return p
}
// ZFSScan is the provider-native scrub or resilver observation for a pool.
// The legacy Scan string remains for compact consumers, while this shape keeps
// progress, terminal errors, and lifecycle timestamps machine-readable.
type ZFSScan struct {
Function string `json:"function,omitempty"`
State string `json:"state,omitempty"`
Percentage float64 `json:"percentage,omitempty"`
Errors int64 `json:"errors,omitempty"`
BytesExamined int64 `json:"bytesExamined,omitempty"`
BytesToProcess int64 `json:"bytesToProcess,omitempty"`
TotalSecondsRemaining int64 `json:"totalSecondsRemaining,omitempty"`
StartedAt *time.Time `json:"startedAt,omitempty"`
EndedAt *time.Time `json:"endedAt,omitempty"`
}
// ZFSDevice represents a device in a ZFS pool
type ZFSDevice struct {
Name string `json:"name"`
Type string `json:"type"` // disk, mirror, raidz, raidz2, raidz3, spare, log, cache, special, special-group
Type string `json:"type"` // disk, mirror, raidz, raidz2, raidz3, spare, log, cache, special, special-group
Role string `json:"role,omitempty"`
Parent string `json:"parent,omitempty"`
GUID string `json:"guid,omitempty"`
Disk string `json:"disk,omitempty"`
Path string `json:"path,omitempty"`
State string `json:"state"` // ONLINE, DEGRADED, FAULTED, OFFLINE, REMOVED, UNAVAIL
ReadErrors int64 `json:"readErrors"`
WriteErrors int64 `json:"writeErrors"`
ChecksumErrors int64 `json:"checksumErrors"`
Missing bool `json:"missing,omitempty"`
Message string `json:"message,omitempty"` // Additional message provided by Proxmox (if any)
}
@@ -2364,6 +2386,7 @@ type CephCluster struct {
FSID string `json:"fsid,omitempty"`
Health string `json:"health"`
HealthMessage string `json:"healthMessage,omitempty"`
HealthChecks []CephHealthCheck `json:"healthChecks,omitempty"`
TotalBytes int64 `json:"totalBytes"`
UsedBytes int64 `json:"usedBytes"`
AvailableBytes int64 `json:"availableBytes"`
@@ -2386,12 +2409,23 @@ func (c CephCluster) NormalizeCollections() CephCluster {
if c.Pools == nil {
c.Pools = []CephPool{}
}
if c.HealthChecks == nil {
c.HealthChecks = []CephHealthCheck{}
}
if c.Services == nil {
c.Services = []CephServiceStatus{}
}
return c
}
// CephHealthCheck preserves one native Ceph health-check code and its
// operator-facing evidence without inferring a ZFS device or pool fault.
type CephHealthCheck struct {
Code string `json:"code"`
Severity string `json:"severity,omitempty"`
Summary string `json:"summary,omitempty"`
}
// CephPool represents usage statistics for a Ceph pool
type CephPool struct {
ID int `json:"id"`
+33
View File
@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"fmt"
"sort"
"strings"
"time"
@@ -157,6 +158,7 @@ func buildCephClusterModel(instanceName string, status *proxmox.CephStatus, df *
}
healthMsg := summarizeCephHealth(status)
healthChecks := cephHealthChecks(status)
numMons := countCephMonitorDaemons(status)
numMgrs := countCephManagerDaemons(status)
@@ -168,6 +170,7 @@ func buildCephClusterModel(instanceName string, status *proxmox.CephStatus, df *
FSID: status.FSID,
Health: status.Health.Status,
HealthMessage: healthMsg,
HealthChecks: healthChecks,
TotalBytes: totalBytes,
UsedBytes: usedBytes,
AvailableBytes: availBytes,
@@ -186,6 +189,36 @@ func buildCephClusterModel(instanceName string, status *proxmox.CephStatus, df *
return cluster
}
func cephHealthChecks(status *proxmox.CephStatus) []models.CephHealthCheck {
if status == nil || len(status.Health.Checks) == 0 {
return nil
}
codes := make([]string, 0, len(status.Health.Checks))
for code := range status.Health.Checks {
codes = append(codes, code)
}
sort.Strings(codes)
checks := make([]models.CephHealthCheck, 0, len(codes))
for _, code := range codes {
check := status.Health.Checks[code]
summary := extractCephCheckSummary(check.Summary)
if summary == "" {
for _, detail := range check.Detail {
if strings.TrimSpace(detail.Message) != "" {
summary = strings.TrimSpace(detail.Message)
break
}
}
}
checks = append(checks, models.CephHealthCheck{
Code: strings.TrimSpace(code),
Severity: strings.TrimSpace(check.Severity),
Summary: summary,
})
}
return checks
}
func countCephMonitorDaemons(status *proxmox.CephStatus) int {
if status == nil {
return 0
+30
View File
@@ -220,6 +220,36 @@ func TestIsCephStorageType(t *testing.T) {
}
}
func TestBuildCephClusterModelPreservesNativeHealthCheckCodes(t *testing.T) {
status := &proxmox.CephStatus{
FSID: "fsid-native",
Health: proxmox.CephHealth{
Status: "HEALTH_WARN",
Checks: map[string]proxmox.CephHealthCheckRaw{
"OSD_DOWN": {
Severity: "HEALTH_WARN",
Summary: json.RawMessage(`{"message":"1 osd down"}`),
},
"PG_DEGRADED": {
Severity: "HEALTH_WARN",
Detail: []proxmox.CephCheckDetail{{Message: "42 pgs degraded"}},
},
},
},
}
cluster := buildCephClusterModel("pve-a", status, nil)
if len(cluster.HealthChecks) != 2 {
t.Fatalf("health checks = %+v", cluster.HealthChecks)
}
if cluster.HealthChecks[0].Code != "OSD_DOWN" || cluster.HealthChecks[0].Summary != "1 osd down" {
t.Fatalf("sorted native check 0 = %+v", cluster.HealthChecks[0])
}
if cluster.HealthChecks[1].Code != "PG_DEGRADED" || cluster.HealthChecks[1].Summary != "42 pgs degraded" {
t.Fatalf("sorted native check 1 = %+v", cluster.HealthChecks[1])
}
}
func TestCountServiceDaemons(t *testing.T) {
t.Parallel()
+13 -1
View File
@@ -4,6 +4,7 @@ import (
"fmt"
"math"
"os"
"sort"
"strconv"
"strings"
"time"
@@ -887,10 +888,21 @@ func convertAgentCephToGlobalCluster(ceph *agentshost.CephCluster, hostname, hos
// Build health message from checks
var healthMessages []string
for _, check := range ceph.Health.Checks {
checkCodes := make([]string, 0, len(ceph.Health.Checks))
for code := range ceph.Health.Checks {
checkCodes = append(checkCodes, code)
}
sort.Strings(checkCodes)
for _, code := range checkCodes {
check := ceph.Health.Checks[code]
if check.Message != "" {
healthMessages = append(healthMessages, check.Message)
}
cluster.HealthChecks = append(cluster.HealthChecks, models.CephHealthCheck{
Code: strings.TrimSpace(code),
Severity: strings.TrimSpace(check.Severity),
Summary: strings.TrimSpace(check.Message),
})
}
if len(healthMessages) > 0 {
cluster.HealthMessage = strings.Join(healthMessages, "; ")
+84 -1
View File
@@ -1007,12 +1007,95 @@ func TestTrueNASPollerHandlesConnectionAddRemove(t *testing.T) {
}
}
func TestTrueNASPollerKeepsPoolHealthConnectionLocalAcrossMatchingAppliances(t *testing.T) {
observedAt := time.Date(2026, 7, 24, 10, 0, 0, 0, time.UTC)
fixture := func(state string) truenas.FixtureSnapshot {
return truenas.FixtureSnapshot{
CollectedAt: observedAt,
System: truenas.SystemInfo{Hostname: "truenas", Healthy: true},
Pools: []truenas.Pool{{
ID: "7",
GUID: "shared-restored-guid",
Name: "tank",
Status: state,
}},
}
}
provider := func(connectionID, state string) *truenas.Provider {
p := truenas.NewLiveProviderForConnection(
&truenas.FixtureFetcher{Snapshot: fixture(state)},
connectionID,
)
if err := p.Refresh(context.Background()); err != nil {
t.Fatalf("refresh %s: %v", connectionID, err)
}
return p
}
poller := NewTrueNASPoller(nil, 0, nil)
poller.providersByOrg["default"] = map[string]*truenas.Provider{
"conn-primary": provider("conn-primary", "DEGRADED"),
"conn-dr": provider("conn-dr", "ONLINE"),
}
poller.configsByOrg["default"] = map[string]config.TrueNASInstance{
"conn-primary": {ID: "conn-primary", Host: "192.0.2.10", Enabled: true},
"conn-dr": {ID: "conn-dr", Host: "192.0.2.11", Enabled: true},
}
poller.rebuildCachedRecordsForOrg("default")
records := poller.GetCurrentRecordsForOrg("default")
poolSources := make(map[string]string)
registry := unifiedresources.NewRegistry(nil)
registry.IngestRecords(unifiedresources.SourceTrueNAS, records)
for _, record := range records {
if record.Resource.Type != unifiedresources.ResourceTypeStorage ||
record.Resource.Storage == nil ||
record.Resource.Storage.Topology != "pool" {
continue
}
health := record.Resource.Storage.PoolHealth
if health == nil {
t.Fatalf("pool health missing from %s: %+v", record.SourceID, record.Resource.Storage)
}
poolSources[record.SourceID] = health.CanonicalState
}
if len(poolSources) != 2 {
t.Fatalf("connection-local pool records = %+v", poolSources)
}
if poolSources["system:conn-primary/pool:tank"] != "DEGRADED" ||
poolSources["system:conn-dr/pool:tank"] != "ONLINE" {
t.Fatalf("connection-local pool health = %+v", poolSources)
}
storageResources := 0
ids := make(map[string]struct{})
for _, resource := range registry.List() {
if resource.Type != unifiedresources.ResourceTypeStorage || resource.Name != "tank" {
continue
}
storageResources++
ids[resource.ID] = struct{}{}
}
if storageResources != 2 || len(ids) != 2 {
t.Fatalf("matching appliances merged pool identity: resources=%d ids=%+v", storageResources, ids)
}
}
func copyTrueNASSnapshot(snapshot *truenas.FixtureSnapshot) *truenas.FixtureSnapshot {
if snapshot == nil {
return nil
}
cloned := *snapshot
cloned.Pools = append([]truenas.Pool(nil), snapshot.Pools...)
cloned.Pools = make([]truenas.Pool, len(snapshot.Pools))
for i := range snapshot.Pools {
cloned.Pools[i] = snapshot.Pools[i]
if snapshot.Pools[i].Scan != nil {
scan := *snapshot.Pools[i].Scan
cloned.Pools[i].Scan = &scan
}
cloned.Pools[i].VDevs = append([]truenas.PoolVDev(nil), snapshot.Pools[i].VDevs...)
cloned.Pools[i].DiskMembers = append([]truenas.PoolDiskMember(nil), snapshot.Pools[i].DiskMembers...)
}
cloned.Datasets = append([]truenas.Dataset(nil), snapshot.Datasets...)
cloned.Disks = append([]truenas.Disk(nil), snapshot.Disks...)
cloned.Alerts = append([]truenas.Alert(nil), snapshot.Alerts...)
+80 -3
View File
@@ -160,14 +160,73 @@ func AssessZFSPool(pool models.ZFSPool) Assessment {
)
}
if scan := pool.ScanDetails; scan != nil {
function := strings.ToUpper(strings.TrimSpace(scan.Function))
state := strings.ToUpper(strings.TrimSpace(scan.State))
switch {
case scan.Errors > 0:
addReason(
"zfs_scan_errors",
RiskCritical,
fmt.Sprintf("ZFS pool %s %s reports %d error(s)", pool.Name, strings.ToLower(firstNonEmpty(function, "scan")), scan.Errors),
)
case state == "FAILED":
addReason("zfs_scan_failed", RiskCritical, fmt.Sprintf("ZFS pool %s %s failed", pool.Name, strings.ToLower(firstNonEmpty(function, "scan"))))
case function == "RESILVER" && zfsScanActive(state):
summary := fmt.Sprintf("ZFS pool %s is resilvering", pool.Name)
if scan.Percentage > 0 {
summary = fmt.Sprintf("ZFS pool %s is resilvering (%.1f%%)", pool.Name, scan.Percentage)
}
addReason("zfs_resilver_active", RiskWarning, summary)
case function == "SCRUB" && zfsScanActive(state):
summary := fmt.Sprintf("ZFS pool %s scrub is running", pool.Name)
if scan.Percentage > 0 {
summary = fmt.Sprintf("ZFS pool %s scrub is running (%.1f%%)", pool.Name, scan.Percentage)
}
addReason("zfs_scrub_active", RiskMonitor, summary)
}
} else {
scan := strings.ToLower(strings.TrimSpace(pool.Scan))
switch {
case strings.Contains(scan, "resilver") && (strings.Contains(scan, "progress") || strings.Contains(scan, "scanning")):
addReason("zfs_resilver_active", RiskWarning, fmt.Sprintf("ZFS pool %s is resilvering", pool.Name))
case strings.Contains(scan, "scrub") && (strings.Contains(scan, "progress") || strings.Contains(scan, "scanning")):
addReason("zfs_scrub_active", RiskMonitor, fmt.Sprintf("ZFS pool %s scrub is running", pool.Name))
}
}
for _, device := range pool.Devices {
deviceState := strings.ToUpper(strings.TrimSpace(device.State))
deviceName := strings.TrimSpace(device.Name)
if deviceName == "" {
deviceName = strings.TrimSpace(device.Path)
}
if deviceName == "" {
deviceName = "unknown"
}
if device.Missing {
addReason("zfs_device_missing", RiskCritical, fmt.Sprintf("ZFS device %s is reported missing by pool topology", deviceName))
}
if device.ReadErrors > 0 || device.WriteErrors > 0 || device.ChecksumErrors > 0 {
addReason(
"zfs_device_errors",
RiskWarning,
fmt.Sprintf("ZFS device %s reports read=%d write=%d checksum=%d errors", deviceName, device.ReadErrors, device.WriteErrors, device.ChecksumErrors),
)
}
if device.Missing {
continue
}
switch deviceState {
case "", "ONLINE":
case "", "ONLINE", "AVAIL", "INUSE":
case "DEGRADED":
addReason("zfs_device_state", RiskWarning, fmt.Sprintf("ZFS device %s is DEGRADED", device.Name))
addReason("zfs_device_state", RiskWarning, fmt.Sprintf("ZFS device %s is DEGRADED", deviceName))
case "FAULTED", "FAILED", "OFFLINE", "REMOVED", "UNAVAIL":
addReason("zfs_device_state", RiskCritical, fmt.Sprintf("ZFS device %s is %s", deviceName, deviceState))
default:
addReason("zfs_device_state", RiskCritical, fmt.Sprintf("ZFS device %s is %s", device.Name, deviceState))
// New or provider-specific states remain visible in the full ZFS
// report but do not become failure claims without a canonical
// mapping.
}
}
@@ -175,6 +234,24 @@ func AssessZFSPool(pool models.ZFSPool) Assessment {
return assessment
}
func zfsScanActive(state string) bool {
switch strings.ToUpper(strings.TrimSpace(state)) {
case "SCANNING", "RUNNING", "IN_PROGRESS", "INPROGRESS":
return true
default:
return false
}
}
func firstNonEmpty(values ...string) string {
for _, value := range values {
if trimmed := strings.TrimSpace(value); trimmed != "" {
return trimmed
}
}
return ""
}
func AssessUnraidStorage(storage models.HostUnraidStorage) Assessment {
assessment := Assessment{Level: RiskHealthy}
addReason := func(code string, severity RiskLevel, summary string) {
@@ -0,0 +1,141 @@
package storagehealth
import (
"testing"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
)
func TestAssessZFSPoolCanonicalStateAndOperationContract(t *testing.T) {
tests := []struct {
name string
pool models.ZFSPool
wantLevel RiskLevel
wantCode string
}{
{
name: "online",
pool: models.ZFSPool{Name: "tank", State: "ONLINE"},
wantLevel: RiskHealthy,
},
{
name: "degraded",
pool: models.ZFSPool{Name: "tank", State: "DEGRADED"},
wantLevel: RiskWarning,
wantCode: "zfs_pool_state",
},
{
name: "faulted",
pool: models.ZFSPool{Name: "tank", State: "FAULTED"},
wantLevel: RiskCritical,
wantCode: "zfs_pool_state",
},
{
name: "offline",
pool: models.ZFSPool{Name: "tank", State: "OFFLINE"},
wantLevel: RiskCritical,
wantCode: "zfs_pool_state",
},
{
name: "unavailable",
pool: models.ZFSPool{Name: "tank", State: "UNAVAIL"},
wantLevel: RiskCritical,
wantCode: "zfs_pool_state",
},
{
name: "resilver",
pool: models.ZFSPool{
Name: "tank",
State: "ONLINE",
ScanDetails: &models.ZFSScan{
Function: "RESILVER",
State: "SCANNING",
Percentage: 25,
},
},
wantLevel: RiskWarning,
wantCode: "zfs_resilver_active",
},
{
name: "scrub monitor only",
pool: models.ZFSPool{
Name: "tank",
State: "ONLINE",
ScanDetails: &models.ZFSScan{
Function: "SCRUB",
State: "SCANNING",
},
},
wantLevel: RiskMonitor,
wantCode: "zfs_scrub_active",
},
{
name: "scrub errors",
pool: models.ZFSPool{
Name: "tank",
State: "ONLINE",
ScanDetails: &models.ZFSScan{
Function: "SCRUB",
State: "FINISHED",
Errors: 2,
},
},
wantLevel: RiskCritical,
wantCode: "zfs_scan_errors",
},
{
name: "spare available is healthy",
pool: models.ZFSPool{
Name: "tank",
State: "ONLINE",
Devices: []models.ZFSDevice{{Name: "sdc", Role: "spare", State: "AVAIL"}},
},
wantLevel: RiskHealthy,
},
{
name: "native missing member",
pool: models.ZFSPool{
Name: "tank",
State: "DEGRADED",
Devices: []models.ZFSDevice{{Name: "sdb", State: "UNAVAIL", Missing: true}},
},
wantLevel: RiskCritical,
wantCode: "zfs_device_missing",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
assessment := AssessZFSPool(tt.pool)
if assessment.Level != tt.wantLevel {
t.Fatalf("level = %q, want %q (%+v)", assessment.Level, tt.wantLevel, assessment.Reasons)
}
if tt.wantCode == "" {
if len(assessment.Reasons) != 0 {
t.Fatalf("unexpected reasons: %+v", assessment.Reasons)
}
return
}
found := false
for _, reason := range assessment.Reasons {
if reason.Code == tt.wantCode {
found = true
}
}
if !found {
t.Fatalf("missing %q in %+v", tt.wantCode, assessment.Reasons)
}
})
}
}
func TestAssessZFSPoolUnknownDeviceStateDoesNotInventFailure(t *testing.T) {
assessment := AssessZFSPool(models.ZFSPool{
Name: "tank",
State: "ONLINE",
Devices: []models.ZFSDevice{{Name: "sda", State: "NOT_REPORTED"}},
})
if assessment.Level != RiskHealthy || len(assessment.Reasons) != 0 {
t.Fatalf("unknown native state must remain evidence-only, got %+v", assessment)
}
}
+278 -62
View File
@@ -14,6 +14,7 @@ import (
"net"
"net/http"
"net/url"
"sort"
"strconv"
"strings"
"sync"
@@ -290,20 +291,10 @@ func (c *Client) getPoolsRPC(ctx context.Context) ([]Pool, error) {
pools := make([]Pool, 0, len(response))
for _, item := range response {
id := strings.TrimSpace(readStringAny(item, "id"))
name := strings.TrimSpace(readStringAny(item, "name"))
if id == "" || id == "0" {
id = name
pool, ok := parsePoolState(item, false)
if ok {
pools = append(pools, pool)
}
pools = append(pools, Pool{
ID: id,
Name: name,
Status: strings.TrimSpace(readStringAny(item, "status")),
TotalBytes: readInt64Any(item, "size", "total", "total_bytes", "totalBytes"),
UsedBytes: readInt64Any(item, "allocated", "used", "used_bytes", "usedBytes"),
FreeBytes: readInt64Any(item, "free", "free_bytes", "freeBytes", "available"),
DiskMembers: poolDiskMembersFromTopology(item["topology"]),
})
}
return pools, nil
}
@@ -337,6 +328,10 @@ func (c *Client) GetBootPool(ctx context.Context) (*Pool, error) {
}
func parseBootPoolState(item map[string]any) (Pool, bool) {
return parsePoolState(item, true)
}
func parsePoolState(item map[string]any, isBoot bool) (Pool, bool) {
if len(item) == 0 {
return Pool{}, false
}
@@ -361,16 +356,26 @@ func parseBootPoolState(item map[string]any) (Pool, bool) {
if topology == nil {
topology = item["groups"]
}
vdevs, diskMembers := poolTopologyFromTopology(topology)
readErrors, writeErrors, checksumErrors := poolErrorTotals(topology)
pool := Pool{
ID: id,
Name: name,
Status: status,
TotalBytes: readInt64Any(item, "size", "total", "total_bytes", "totalBytes"),
UsedBytes: readInt64Any(item, "allocated", "used", "used_bytes", "usedBytes"),
FreeBytes: readInt64Any(item, "free", "free_bytes", "freeBytes", "available"),
IsBoot: true,
DiskMembers: poolDiskMembersFromTopology(topology),
ID: id,
GUID: strings.TrimSpace(readStringAny(item, "guid")),
Name: name,
Status: status,
StatusCode: strings.TrimSpace(readStringAny(item, "status_code", "statusCode")),
StatusDetail: strings.TrimSpace(readStringAny(item, "status_detail", "statusDetail")),
TotalBytes: readInt64Any(item, "size", "total", "total_bytes", "totalBytes"),
UsedBytes: readInt64Any(item, "allocated", "used", "used_bytes", "usedBytes"),
FreeBytes: readInt64Any(item, "free", "free_bytes", "freeBytes", "available"),
ReadErrors: readErrors,
WriteErrors: writeErrors,
ChecksumErrors: checksumErrors,
Scan: poolScanFromAny(item["scan"]),
VDevs: vdevs,
IsBoot: isBoot,
DiskMembers: diskMembers,
}
if pool.TotalBytes == 0 {
pool.TotalBytes = readInt64Any(properties, "size", "total", "total_bytes", "totalBytes")
@@ -413,6 +418,25 @@ func mergeBootPool(pools []Pool, boot Pool) []Pool {
if len(boot.DiskMembers) > 0 {
pools[i].DiskMembers = append([]PoolDiskMember(nil), boot.DiskMembers...)
}
if len(boot.VDevs) > 0 {
pools[i].VDevs = append([]PoolVDev(nil), boot.VDevs...)
}
if boot.GUID != "" {
pools[i].GUID = boot.GUID
}
if boot.StatusCode != "" {
pools[i].StatusCode = boot.StatusCode
}
if boot.StatusDetail != "" {
pools[i].StatusDetail = boot.StatusDetail
}
if boot.Scan != nil {
scan := *boot.Scan
pools[i].Scan = &scan
}
pools[i].ReadErrors = boot.ReadErrors
pools[i].WriteErrors = boot.WriteErrors
pools[i].ChecksumErrors = boot.ChecksumErrors
return pools
}
return append(pools, boot)
@@ -422,42 +446,113 @@ func mergeBootPool(pools []Pool, boot Pool) []Pool {
// topology object across every vdev group, including detached/unavailable
// members that only appear through their unavail_disk datastore row.
func poolDiskMembersFromTopology(topology any) []PoolDiskMember {
_, members := poolTopologyFromTopology(topology)
return members
}
// poolTopologyFromTopology flattens native vdev topology without discarding
// group roles or parentage. It separately returns leaf membership for the
// existing disk-enrichment path.
func poolTopologyFromTopology(topology any) ([]PoolVDev, []PoolDiskMember) {
groups, ok := topology.(map[string]any)
if !ok {
return nil
return nil, nil
}
groupNames := make([]string, 0, len(groups))
for groupName := range groups {
groupNames = append(groupNames, groupName)
}
sort.Strings(groupNames)
var vdevs []PoolVDev
var members []PoolDiskMember
var walk func(node any)
walk = func(node any) {
var walk func(node any, role, parentID, position string)
walk = func(node any, role, parentID, position string) {
vdev, ok := node.(map[string]any)
if !ok {
return
}
if children, ok := vdev["children"].([]any); ok && len(children) > 0 {
for _, child := range children {
walk(child)
stats := readMapAny(vdev, "stats")
device := strings.TrimSpace(readStringAny(vdev, "device"))
path := strings.TrimSpace(readStringAny(vdev, "path"))
if device == "" {
device = strings.TrimPrefix(path, "/dev/")
}
disk := strings.TrimSpace(readStringAny(vdev, "disk"))
unavailableDisk, hasUnavailableDisk := vdev["unavail_disk"].(map[string]any)
if disk == "" && hasUnavailableDisk {
disk = strings.TrimSpace(readStringAny(unavailableDisk, "devname", "name"))
}
if disk == "" {
disk = wholeDiskFromDevice(device)
}
vdevType := strings.TrimSpace(readStringAny(vdev, "type"))
state := strings.TrimSpace(readStringAny(vdev, "status", "state"))
guid := strings.TrimSpace(readStringAny(vdev, "guid"))
name := strings.TrimSpace(readStringAny(vdev, "name"))
if name == "" {
name = firstNonEmptyString(disk, device, path, vdevType, role)
}
id := guid
if id == "" {
id = role + ":" + position + ":" + name
}
readErrors := readInt64Any(vdev, "read_errors", "readErrors")
writeErrors := readInt64Any(vdev, "write_errors", "writeErrors")
checksumErrors := readInt64Any(vdev, "checksum_errors", "checksumErrors")
if readErrors == 0 {
readErrors = readInt64Any(stats, "read_errors", "readErrors")
}
if writeErrors == 0 {
writeErrors = readInt64Any(stats, "write_errors", "writeErrors")
}
if checksumErrors == 0 {
checksumErrors = readInt64Any(stats, "checksum_errors", "checksumErrors")
}
message := strings.TrimSpace(readStringAny(vdev, "status_detail", "statusDetail", "message"))
missing := hasUnavailableDisk || strings.EqualFold(vdevType, "unavail_disk")
vdevs = append(vdevs, PoolVDev{
ID: id,
ParentID: parentID,
GUID: guid,
Name: name,
Type: vdevType,
Role: role,
Disk: disk,
Device: device,
Path: path,
Status: state,
ReadErrors: readErrors,
WriteErrors: writeErrors,
ChecksumErrors: checksumErrors,
Missing: missing,
Message: message,
})
children, hasChildren := vdev["children"].([]any)
if hasChildren && len(children) > 0 {
for index, child := range children {
walk(child, role, id, position+"."+strconv.Itoa(index))
}
return
}
member := PoolDiskMember{
Disk: strings.TrimSpace(readStringAny(vdev, "disk")),
Device: strings.TrimSpace(readStringAny(vdev, "device")),
Status: strings.TrimSpace(readStringAny(vdev, "status")),
}
if member.Device == "" {
member.Device = strings.TrimPrefix(strings.TrimSpace(readStringAny(vdev, "path")), "/dev/")
}
if member.Disk == "" {
// A missing/faulted member's device path no longer resolves;
// middleware then attaches the disk's datastore row instead.
if unavail, ok := vdev["unavail_disk"].(map[string]any); ok {
member.Disk = strings.TrimSpace(readStringAny(unavail, "devname", "name"))
}
}
if member.Disk == "" {
member.Disk = wholeDiskFromDevice(member.Device)
Disk: disk,
Device: device,
Path: path,
GUID: guid,
Type: vdevType,
Role: role,
Status: state,
Missing: missing,
ReadErrors: readErrors,
WriteErrors: writeErrors,
ChecksumErrors: checksumErrors,
Message: message,
}
if member.Disk == "" && member.Device == "" {
return
@@ -465,16 +560,87 @@ func poolDiskMembersFromTopology(topology any) []PoolDiskMember {
members = append(members, member)
}
for _, groupName := range groupNames {
group := groups[groupName]
nodes, ok := group.([]any)
if !ok {
continue
}
for index, node := range nodes {
walk(node, groupName, "", strconv.Itoa(index))
}
}
return vdevs, members
}
func poolErrorTotals(topology any) (int64, int64, int64) {
groups, ok := topology.(map[string]any)
if !ok {
return 0, 0, 0
}
var readErrors, writeErrors, checksumErrors int64
for _, group := range groups {
nodes, ok := group.([]any)
if !ok {
continue
}
for _, node := range nodes {
walk(node)
vdev, ok := node.(map[string]any)
if !ok {
continue
}
stats := readMapAny(vdev, "stats")
readValue := readInt64Any(vdev, "read_errors", "readErrors")
writeValue := readInt64Any(vdev, "write_errors", "writeErrors")
checksumValue := readInt64Any(vdev, "checksum_errors", "checksumErrors")
if readValue == 0 {
readValue = readInt64Any(stats, "read_errors", "readErrors")
}
if writeValue == 0 {
writeValue = readInt64Any(stats, "write_errors", "writeErrors")
}
if checksumValue == 0 {
checksumValue = readInt64Any(stats, "checksum_errors", "checksumErrors")
}
readErrors += readValue
writeErrors += writeValue
checksumErrors += checksumValue
}
}
return members
return readErrors, writeErrors, checksumErrors
}
func poolScanFromAny(raw any) *PoolScan {
scan, ok := raw.(map[string]any)
if !ok || len(scan) == 0 {
return nil
}
function := strings.TrimSpace(readStringAny(scan, "function"))
state := strings.TrimSpace(readStringAny(scan, "state"))
if function == "" && state == "" {
return nil
}
result := &PoolScan{
Function: function,
State: state,
Percentage: readFloatAny(scan, "percentage", "percent"),
Errors: readInt64Any(scan, "errors"),
BytesExamined: readInt64Any(scan, "bytes_examined", "bytesExamined", "processed_bytes", "processedBytes"),
BytesToProcess: readInt64Any(scan, "bytes_to_process", "bytesToProcess", "total_bytes", "totalBytes"),
TotalSecondsRemaining: readInt64Any(scan, "total_secs_left", "totalSecondsRemaining"),
StartedAt: readTimeAny(scan, "start_time", "startTime", "started_at", "startedAt"),
EndedAt: readTimeAny(scan, "end_time", "endTime", "ended_at", "endedAt"),
}
return result
}
func firstNonEmptyString(values ...string) string {
for _, value := range values {
if trimmed := strings.TrimSpace(value); trimmed != "" {
return trimmed
}
}
return ""
}
func wholeDiskFromDevice(device string) string {
@@ -538,15 +704,31 @@ func (c *Client) getPoolsREST(ctx context.Context) ([]Pool, error) {
topology = nil
}
}
var scan any
if len(item.Scan) > 0 {
if err := json.Unmarshal(item.Scan, &scan); err != nil {
scan = nil
}
}
vdevs, diskMembers := poolTopologyFromTopology(topology)
readErrors, writeErrors, checksumErrors := poolErrorTotals(topology)
pools = append(pools, Pool{
ID: id,
Name: strings.TrimSpace(item.Name),
Status: strings.TrimSpace(item.Status),
TotalBytes: item.Size,
UsedBytes: item.Allocated,
FreeBytes: item.Free,
DiskMembers: poolDiskMembersFromTopology(topology),
ID: id,
GUID: strings.TrimSpace(item.GUID),
Name: strings.TrimSpace(item.Name),
Status: strings.TrimSpace(item.Status),
StatusCode: strings.TrimSpace(item.StatusCode),
StatusDetail: strings.TrimSpace(item.StatusDetail),
TotalBytes: item.Size,
UsedBytes: item.Allocated,
FreeBytes: item.Free,
ReadErrors: readErrors,
WriteErrors: writeErrors,
ChecksumErrors: checksumErrors,
Scan: poolScanFromAny(scan),
VDevs: vdevs,
DiskMembers: diskMembers,
})
}
@@ -804,12 +986,20 @@ func (c *Client) disksFromMaps(ctx context.Context, response []map[string]any) (
// health/status field on any TrueNAS version and only names the pool behind
// an extra option the REST bridge cannot pass, so without this every disk
// rendered as an unparented "Unknown" (#1573).
func enrichDisksFromPoolTopology(pools []Pool, disks []Disk) {
func enrichDisksFromPoolTopology(pools []Pool, disks []Disk) []Disk {
type membership struct {
pool string
status string
}
byDevice := make(map[string]membership)
knownDevices := make(map[string]struct{}, len(disks)*2)
for _, disk := range disks {
for _, identity := range []string{disk.Name, disk.ID} {
if identity = strings.TrimSpace(identity); identity != "" {
knownDevices[identity] = struct{}{}
}
}
}
for _, pool := range pools {
for _, member := range pool.DiskMembers {
for _, device := range []string{member.Disk, member.Device} {
@@ -821,10 +1011,31 @@ func enrichDisksFromPoolTopology(pools []Pool, disks []Disk) {
byDevice[device] = membership{pool: pool.Name, status: member.Status}
}
}
if !member.Missing {
continue
}
name := firstNonEmptyString(member.Disk, wholeDiskFromDevice(member.Device), member.Device)
if name == "" {
continue
}
if _, exists := knownDevices[name]; exists {
continue
}
// A synthetic disk resource is emitted only from explicit
// unavail_disk topology evidence. Missing disk.query rows alone are
// not enough to assert that a disk is absent.
disks = append(disks, Disk{
ID: firstNonEmptyString(member.GUID, name),
Name: name,
Pool: pool.Name,
Status: member.Status,
Health: "UNKNOWN",
})
knownDevices[name] = struct{}{}
}
}
if len(byDevice) == 0 {
return
return disks
}
for i := range disks {
@@ -843,6 +1054,7 @@ func enrichDisksFromPoolTopology(pools []Pool, disks []Disk) {
disk.Status = member.status
}
}
return disks
}
// GetDiskTemperatures returns the current temperature by TrueNAS disk name.
@@ -1597,7 +1809,7 @@ func (c *Client) FetchSnapshot(ctx context.Context) (*FixtureSnapshot, error) {
if err != nil {
return nil, fmt.Errorf("fetch truenas disks: %w", err)
}
enrichDisksFromPoolTopology(pools, disks)
disks = enrichDisksFromPoolTopology(pools, disks)
alerts, err := c.GetAlerts(ctx)
if err != nil {
@@ -4143,13 +4355,17 @@ func (f *int64ResponseField) UnmarshalJSON(data []byte) error {
}
type poolResponse struct {
ID int64 `json:"id"`
Name string `json:"name"`
Status string `json:"status"`
Size int64 `json:"size"`
Allocated int64 `json:"allocated"`
Free int64 `json:"free"`
Topology json.RawMessage `json:"topology"`
ID int64 `json:"id"`
GUID string `json:"guid"`
Name string `json:"name"`
Status string `json:"status"`
StatusCode string `json:"status_code"`
StatusDetail string `json:"status_detail"`
Size int64 `json:"size"`
Allocated int64 `json:"allocated"`
Free int64 `json:"free"`
Topology json.RawMessage `json:"topology"`
Scan json.RawMessage `json:"scan"`
}
type datasetResponse struct {
@@ -312,3 +312,102 @@ func TestEnrichDisksFromPoolTopology(t *testing.T) {
t.Fatalf("explicit disk fields must not be overwritten, got %+v", disks[3])
}
}
func TestParsePoolStatePreservesScanTopologyErrorsAndNativeMissingEvidence(t *testing.T) {
item := map[string]any{
"id": float64(7),
"guid": "pool-guid-7",
"name": "tank",
"status": "DEGRADED",
"status_code": "FEAT_DISABLED",
"status_detail": "One or more devices is unavailable",
"scan": map[string]any{
"function": "RESILVER",
"state": "SCANNING",
"percentage": 42.5,
"errors": float64(0),
"bytes_examined": float64(1024),
"bytes_to_process": float64(4096),
"total_secs_left": float64(120),
"start_time": "2026-07-24T09:00:00Z",
},
"topology": map[string]any{
"data": []any{
map[string]any{
"guid": "mirror-guid",
"name": "mirror-0",
"type": "MIRROR",
"status": "DEGRADED",
"stats": map[string]any{
"read_errors": float64(1),
"write_errors": float64(2),
"checksum_errors": float64(3),
},
"children": []any{
map[string]any{
"guid": "disk-guid-0",
"type": "DISK",
"disk": "sda",
"path": "/dev/sda2",
"status": "ONLINE",
},
map[string]any{
"guid": "disk-guid-1",
"type": "UNAVAIL_DISK",
"path": "/dev/disk/by-partuuid/missing-member",
"status": "UNAVAIL",
"unavail_disk": map[string]any{
"devname": "sdb",
},
},
},
},
},
"spare": []any{
map[string]any{
"guid": "spare-guid",
"type": "DISK",
"disk": "sdc",
"path": "/dev/sdc",
"status": "AVAIL",
},
},
},
}
pool, ok := parsePoolState(item, false)
if !ok {
t.Fatal("expected pool")
}
if pool.GUID != "pool-guid-7" || pool.StatusCode != "FEAT_DISABLED" || pool.StatusDetail == "" {
t.Fatalf("native pool identity/status evidence lost: %+v", pool)
}
if pool.Scan == nil || pool.Scan.Function != "RESILVER" || pool.Scan.State != "SCANNING" || pool.Scan.Percentage != 42.5 || pool.Scan.TotalSecondsRemaining != 120 {
t.Fatalf("structured scan evidence = %+v", pool.Scan)
}
if pool.ReadErrors != 1 || pool.WriteErrors != 2 || pool.ChecksumErrors != 3 {
t.Fatalf("pool error totals = read %d write %d checksum %d", pool.ReadErrors, pool.WriteErrors, pool.ChecksumErrors)
}
if len(pool.VDevs) != 4 {
t.Fatalf("vdevs = %+v", pool.VDevs)
}
if len(pool.DiskMembers) != 3 {
t.Fatalf("disk members = %+v", pool.DiskMembers)
}
missing := pool.DiskMembers[1]
if missing.Disk != "sdb" || !missing.Missing || missing.Role != "data" || missing.Path != "/dev/disk/by-partuuid/missing-member" {
t.Fatalf("missing member evidence = %+v", missing)
}
spare := pool.DiskMembers[2]
if spare.Disk != "sdc" || spare.Status != "AVAIL" || spare.Role != "spare" {
t.Fatalf("spare member evidence = %+v", spare)
}
disks := enrichDisksFromPoolTopology([]Pool{pool}, []Disk{{ID: "sda", Name: "sda"}})
if len(disks) != 2 {
t.Fatalf("expected only explicit unavailable member to synthesize, got %+v", disks)
}
if disks[1].Name != "sdb" || disks[1].Pool != "tank" || disks[1].Status != "UNAVAIL" {
t.Fatalf("synthetic unavailable disk = %+v", disks[1])
}
}
+529 -69
View File
@@ -534,7 +534,7 @@ func truenasRecordsFromSnapshot(snapshot *FixtureSnapshot, connectionID string,
systemAssessment := assessSystemStorage(snapshot)
systemRisk := unifiedresources.StorageRiskFromAssessment(systemAssessment)
_, protectionReduced, rebuildInProgress, protectionSummary, rebuildSummary := unifiedresources.StorageRiskSemantics(systemRisk)
systemIncidents, poolIncidents, diskIncidents := buildIncidentAssignments(snapshot, collectedAt)
incidentAssignments := buildIncidentAssignments(snapshot, collectedAt)
records := make([]unifiedresources.IngestRecord, 0, 1+len(snapshot.Pools)+len(snapshot.Datasets)+len(snapshot.Apps)+len(snapshot.VMs)+len(snapshot.Shares)+len(snapshot.Disks))
totalCapacity, totalUsed := aggregatePoolUsage(snapshot.Pools)
@@ -545,7 +545,7 @@ func truenasRecordsFromSnapshot(snapshot *FixtureSnapshot, connectionID string,
Resource: unifiedresources.Resource{
Type: unifiedresources.ResourceTypeAgent,
Name: strings.TrimSpace(snapshot.System.Hostname),
Status: systemStatus(snapshot.System, systemRisk, systemIncidents),
Status: systemStatus(snapshot.System, systemRisk, incidentAssignments.System),
LastSeen: collectedAt,
UpdatedAt: collectedAt,
Metrics: metricsFromTrueNASSystem(snapshot.System, totalCapacity, totalUsed),
@@ -570,7 +570,7 @@ func truenasRecordsFromSnapshot(snapshot *FixtureSnapshot, connectionID string,
snapshot.System.Version,
"zfs",
},
Incidents: systemIncidents,
Incidents: incidentAssignments.System,
},
// The system's identity deliberately carries no machine key: the
// TrueNAS DMI serial is shared by DR clones and can be vendor
@@ -585,7 +585,8 @@ func truenasRecordsFromSnapshot(snapshot *FixtureSnapshot, connectionID string,
for _, pool := range snapshot.Pools {
assessment := assessPool(pool)
risk := unifiedresources.StorageRiskFromAssessment(assessment)
incidents := poolIncidents[strings.TrimSpace(pool.Name)]
incidents := incidentAssignments.Pools[strings.TrimSpace(pool.Name)]
zfsPool := zfsPoolFromPool(pool)
poolSourceID := scopedPoolSourceID(systemSourceID, pool.Name)
records = append(records, unifiedresources.IngestRecord{
SourceID: poolSourceID,
@@ -601,13 +602,19 @@ func truenasRecordsFromSnapshot(snapshot *FixtureSnapshot, connectionID string,
Disk: diskMetric(pool.TotalBytes, pool.UsedBytes),
},
Storage: &unifiedresources.StorageMeta{
Type: "zfs-pool",
IsZFS: true,
Platform: "truenas",
Topology: "pool",
Protection: "zfs",
Risk: risk,
ZFSPoolState: strings.ToUpper(strings.TrimSpace(pool.Status)),
Type: "zfs-pool",
IsZFS: true,
Platform: "truenas",
Topology: poolTopologyLabel(pool),
Protection: "zfs",
Risk: risk,
RiskSummary: unifiedresources.StorageRiskSummary(risk),
PoolHealth: poolHealthFromTrueNASPool(pool, assessment, collectedAt),
ZFSPool: &zfsPool,
ZFSPoolState: strings.ToUpper(strings.TrimSpace(pool.Status)),
ZFSReadErrors: pool.ReadErrors,
ZFSWriteErrors: pool.WriteErrors,
ZFSChecksumErrors: pool.ChecksumErrors,
},
Tags: poolTags(pool),
Incidents: incidents,
@@ -627,6 +634,7 @@ func truenasRecordsFromSnapshot(snapshot *FixtureSnapshot, connectionID string,
parentPool = parentPoolFromDataset(dataset.Name)
}
totalBytes := dataset.UsedBytes + dataset.AvailBytes
incidents := incidentAssignments.Datasets[strings.TrimSpace(dataset.Name)]
records = append(records, unifiedresources.IngestRecord{
SourceID: scopedDatasetSourceID(systemSourceID, dataset.Name),
ParentSourceID: scopedPoolSourceID(systemSourceID, parentPool),
@@ -634,7 +642,7 @@ func truenasRecordsFromSnapshot(snapshot *FixtureSnapshot, connectionID string,
Resource: unifiedresources.Resource{
Type: unifiedresources.ResourceTypeStorage,
Name: dataset.Name,
Status: statusFromDataset(dataset),
Status: unifiedresources.IncidentsStatus(statusFromDataset(dataset), incidents),
LastSeen: collectedAt,
UpdatedAt: collectedAt,
Metrics: &unifiedresources.ResourceMetrics{
@@ -653,6 +661,7 @@ func truenasRecordsFromSnapshot(snapshot *FixtureSnapshot, connectionID string,
"zfs",
datasetStateTag(dataset),
},
Incidents: incidents,
},
Identity: unifiedresources.ResourceIdentity{
Hostnames: []string{
@@ -665,6 +674,7 @@ func truenasRecordsFromSnapshot(snapshot *FixtureSnapshot, connectionID string,
for _, app := range snapshot.Apps {
metrics := metricsFromTrueNASApp(app, snapshot.System.MemoryTotalBytes)
incidents := incidentAssignments.Apps[appIncidentKey(app)]
dockerMeta := &unifiedresources.DockerData{
ContainerID: appCanonicalID(app),
Hostname: strings.TrimSpace(snapshot.System.Hostname),
@@ -703,7 +713,7 @@ func truenasRecordsFromSnapshot(snapshot *FixtureSnapshot, connectionID string,
Type: unifiedresources.ResourceTypeAppContainer,
Technology: "docker",
Name: appDisplayName(app),
Status: statusFromApp(app),
Status: unifiedresources.IncidentsStatus(statusFromApp(app), incidents),
LastSeen: collectedAt,
UpdatedAt: collectedAt,
Metrics: metrics,
@@ -714,6 +724,7 @@ func truenasRecordsFromSnapshot(snapshot *FixtureSnapshot, connectionID string,
},
Capabilities: truenasAppCapabilities(),
Tags: appTags(app),
Incidents: incidents,
},
Identity: unifiedresources.ResourceIdentity{
Hostnames: dedupeStrings([]string{appDisplayName(app)}),
@@ -782,7 +793,7 @@ func truenasRecordsFromSnapshot(snapshot *FixtureSnapshot, connectionID string,
for _, disk := range snapshot.Disks {
assessment := assessDisk(disk)
incidents := diskIncidents[strings.TrimSpace(disk.Name)]
incidents := incidentAssignments.Disks[strings.TrimSpace(disk.Name)]
diskIdentity := unifiedresources.ResourceIdentity{
Hostnames: []string{snapshot.System.Hostname},
}
@@ -821,6 +832,8 @@ func truenasRecordsFromSnapshot(snapshot *FixtureSnapshot, connectionID string,
TemperatureAggregate: temperatureAggregateMetaFromTrueNASDisk(disk),
Wearout: -1,
RPM: rpmFromDisk(disk),
StorageGroup: strings.TrimSpace(disk.Pool),
StorageState: normalizedDiskStatus(disk),
Risk: unifiedresources.PhysicalDiskRiskFromAssessmentAndIncidents(assessment, incidents),
},
Tags: []string{"truenas", "disk", disk.Transport},
@@ -1117,12 +1130,28 @@ func enrichAppStatsFromPreviousSnapshot(current *FixtureSnapshot, previous *Fixt
}
}
func buildIncidentAssignments(snapshot *FixtureSnapshot, observedAt time.Time) ([]unifiedresources.ResourceIncident, map[string][]unifiedresources.ResourceIncident, map[string][]unifiedresources.ResourceIncident) {
systemIncidents := make([]unifiedresources.ResourceIncident, 0)
poolIncidents := make(map[string][]unifiedresources.ResourceIncident)
diskIncidents := make(map[string][]unifiedresources.ResourceIncident)
type trueNASIncidentAssignments struct {
System []unifiedresources.ResourceIncident
Pools map[string][]unifiedresources.ResourceIncident
Datasets map[string][]unifiedresources.ResourceIncident
Disks map[string][]unifiedresources.ResourceIncident
Apps map[string][]unifiedresources.ResourceIncident
}
type poolIncidentProjection struct {
Incident unifiedresources.ResourceIncident
Disk string
}
func buildIncidentAssignments(snapshot *FixtureSnapshot, observedAt time.Time) trueNASIncidentAssignments {
assignments := trueNASIncidentAssignments{
Pools: make(map[string][]unifiedresources.ResourceIncident),
Datasets: make(map[string][]unifiedresources.ResourceIncident),
Disks: make(map[string][]unifiedresources.ResourceIncident),
Apps: make(map[string][]unifiedresources.ResourceIncident),
}
if snapshot == nil {
return systemIncidents, poolIncidents, diskIncidents
return assignments
}
diskPools := make(map[string]string, len(snapshot.Disks))
@@ -1135,7 +1164,7 @@ func buildIncidentAssignments(snapshot *FixtureSnapshot, observedAt time.Time) (
diskPools[diskName] = poolName
}
nativePoolAlertSeen := make(map[string]struct{}, len(snapshot.Alerts))
nativePoolSignals := make(map[string]map[string]struct{}, len(snapshot.Alerts))
for _, alert := range snapshot.Alerts {
if alert.Dismissed {
continue
@@ -1144,19 +1173,22 @@ func buildIncidentAssignments(snapshot *FixtureSnapshot, observedAt time.Time) (
if !ok {
continue
}
systemIncidents = append(systemIncidents, incident)
assignments.System = append(assignments.System, incident)
if poolName := poolNameFromAlert(alert); poolName != "" {
poolIncidents[poolName] = append(poolIncidents[poolName], incident)
if incident.Severity == storagehealth.RiskWarning || incident.Severity == storagehealth.RiskCritical {
nativePoolAlertSeen[poolName] = struct{}{}
assignments.Pools[poolName] = append(assignments.Pools[poolName], incident)
if nativePoolSignals[poolName] == nil {
nativePoolSignals[poolName] = make(map[string]struct{})
}
for _, code := range canonicalSignalsCoveredByNativeAlert(incident.Code) {
nativePoolSignals[poolName][code] = struct{}{}
}
}
if diskName := diskNameFromAlert(alert); diskName != "" {
diskIncidents[diskName] = append(diskIncidents[diskName], incident)
assignments.Disks[diskName] = append(assignments.Disks[diskName], incident)
if poolName := diskPools[diskName]; poolName != "" {
poolIncidents[poolName] = append(poolIncidents[poolName], incident)
assignments.Pools[poolName] = append(assignments.Pools[poolName], incident)
}
}
}
@@ -1166,18 +1198,28 @@ func buildIncidentAssignments(snapshot *FixtureSnapshot, observedAt time.Time) (
if poolName == "" {
continue
}
if _, exists := nativePoolAlertSeen[poolName]; exists {
continue
for _, projection := range incidentsFromPoolHealth(pool, observedAt) {
if _, covered := nativePoolSignals[poolName][projection.Incident.Code]; covered {
continue
}
assignments.System = append(assignments.System, projection.Incident)
assignments.Pools[poolName] = append(assignments.Pools[poolName], projection.Incident)
if projection.Disk != "" {
assignments.Disks[projection.Disk] = append(assignments.Disks[projection.Disk], projection.Incident)
}
}
incident, ok := incidentFromPoolStatus(pool, observedAt)
if !ok {
continue
}
systemIncidents = append(systemIncidents, incident)
poolIncidents[poolName] = append(poolIncidents[poolName], incident)
}
return systemIncidents, poolIncidents, diskIncidents
for _, dataset := range snapshot.Datasets {
if incident, ok := incidentFromDatasetState(dataset, observedAt); ok {
assignments.Datasets[strings.TrimSpace(dataset.Name)] = append(assignments.Datasets[strings.TrimSpace(dataset.Name)], incident)
}
}
for _, app := range snapshot.Apps {
assignments.Apps[appIncidentKey(app)] = append(assignments.Apps[appIncidentKey(app)], incidentsFromAppState(app, observedAt)...)
}
return assignments
}
func assessSystemStorage(snapshot *FixtureSnapshot) storagehealth.Assessment {
@@ -1201,52 +1243,250 @@ func incidentFromAlert(alert Alert) (unifiedresources.ResourceIncident, bool) {
return unifiedresources.ResourceIncident{}, false
}
return unifiedresources.ResourceIncident{
Provider: "truenas",
NativeID: strings.TrimSpace(alert.ID),
Code: incidentCodeFromAlert(alert),
Severity: severity,
Source: strings.TrimSpace(alert.Source),
Summary: strings.TrimSpace(alert.Message),
StartedAt: alert.Datetime,
Provider: "truenas",
NativeID: strings.TrimSpace(alert.ID),
Code: incidentCodeFromAlert(alert),
Severity: severity,
Source: strings.TrimSpace(alert.Source),
Summary: strings.TrimSpace(alert.Message),
StartedAt: alert.Datetime,
RecoveryConfirmationsRequired: 2,
}, true
}
func incidentFromPoolStatus(pool Pool, observedAt time.Time) (unifiedresources.ResourceIncident, bool) {
assessment := assessPool(pool)
if assessment.Level != storagehealth.RiskWarning && assessment.Level != storagehealth.RiskCritical {
return unifiedresources.ResourceIncident{}, false
}
reason, ok := primaryIncidentReason(assessment, "zfs_pool_state")
if !ok {
return unifiedresources.ResourceIncident{}, false
for _, projection := range incidentsFromPoolHealth(pool, observedAt) {
if projection.Incident.Code == "zfs_pool_state" {
return projection.Incident, true
}
}
return unifiedresources.ResourceIncident{}, false
}
func incidentsFromPoolHealth(pool Pool, observedAt time.Time) []poolIncidentProjection {
poolName := strings.TrimSpace(pool.Name)
if poolName == "" {
return nil
}
poolIdentity := firstNonEmptyString(pool.GUID, pool.ID, poolName)
source := "pool.query"
if pool.IsBoot {
source = "boot.get_state"
}
makeIncident := func(nativeID, code string, severity storagehealth.RiskLevel, summary string) unifiedresources.ResourceIncident {
return unifiedresources.ResourceIncident{
Provider: "truenas",
NativeID: nativeID,
Code: code,
Severity: severity,
Source: source,
Summary: summary,
StartedAt: observedAt,
ConfirmationsRequired: 2,
RecoveryConfirmationsRequired: 2,
}
}
var out []poolIncidentProjection
state := strings.ToUpper(strings.TrimSpace(pool.Status))
switch state {
case "DEGRADED":
out = append(out, poolIncidentProjection{Incident: makeIncident(
"pool:"+poolName+":state",
"zfs_pool_state",
storagehealth.RiskWarning,
fmt.Sprintf("ZFS pool %s is DEGRADED", poolName),
)})
case "FAULTED", "OFFLINE", "REMOVED", "UNAVAIL", "SUSPENDED":
out = append(out, poolIncidentProjection{Incident: makeIncident(
"pool:"+poolName+":state",
"zfs_pool_state",
storagehealth.RiskCritical,
fmt.Sprintf("ZFS pool %s is %s", poolName, state),
)})
}
hasVDevErrors := false
for _, vdev := range pool.VDevs {
if vdev.ReadErrors > 0 || vdev.WriteErrors > 0 || vdev.ChecksumErrors > 0 {
hasVDevErrors = true
break
}
}
if !hasVDevErrors && (pool.ReadErrors > 0 || pool.WriteErrors > 0 || pool.ChecksumErrors > 0) {
out = append(out, poolIncidentProjection{Incident: makeIncident(
"pool:"+poolIdentity,
"zfs_pool_errors",
storagehealth.RiskWarning,
fmt.Sprintf("ZFS pool %s reports read=%d write=%d checksum=%d errors", poolName, pool.ReadErrors, pool.WriteErrors, pool.ChecksumErrors),
)})
}
if pool.Scan != nil {
function := strings.ToUpper(strings.TrimSpace(pool.Scan.Function))
scanState := strings.ToUpper(strings.TrimSpace(pool.Scan.State))
switch {
case pool.Scan.Errors > 0:
out = append(out, poolIncidentProjection{Incident: makeIncident(
"pool:"+poolIdentity+":scan",
"zfs_scan_errors",
storagehealth.RiskCritical,
fmt.Sprintf("ZFS pool %s %s reports %d error(s)", poolName, strings.ToLower(firstNonEmptyString(function, "scan")), pool.Scan.Errors),
)})
case scanState == "FAILED":
out = append(out, poolIncidentProjection{Incident: makeIncident(
"pool:"+poolIdentity+":scan",
"zfs_scan_failed",
storagehealth.RiskCritical,
fmt.Sprintf("ZFS pool %s %s failed", poolName, strings.ToLower(firstNonEmptyString(function, "scan"))),
)})
case function == "RESILVER" && poolScanIsActive(scanState):
out = append(out, poolIncidentProjection{Incident: makeIncident(
"pool:"+poolIdentity+":scan",
"zfs_resilver_active",
storagehealth.RiskWarning,
poolScanSummary(pool),
)})
case function == "SCRUB" && poolScanIsActive(scanState):
out = append(out, poolIncidentProjection{Incident: makeIncident(
"pool:"+poolIdentity+":scan",
"zfs_scrub_active",
storagehealth.RiskMonitor,
poolScanSummary(pool),
)})
}
}
for _, vdev := range pool.VDevs {
state := strings.ToUpper(strings.TrimSpace(vdev.Status))
deviceIdentity := firstNonEmptyString(vdev.GUID, vdev.Disk, vdev.Device, vdev.Path, vdev.ID, vdev.Name)
if deviceIdentity == "" {
continue
}
displayName := firstNonEmptyString(vdev.Disk, vdev.Device, vdev.Path, vdev.Name, deviceIdentity)
nativeID := "pool:" + poolIdentity + ":vdev:" + deviceIdentity
diskName := strings.TrimSpace(vdev.Disk)
if vdev.Missing {
out = append(out, poolIncidentProjection{
Incident: makeIncident(nativeID, "zfs_device_missing", storagehealth.RiskCritical, fmt.Sprintf("ZFS device %s is reported missing by pool %s topology", displayName, poolName)),
Disk: diskName,
})
} else {
switch state {
case "DEGRADED":
out = append(out, poolIncidentProjection{
Incident: makeIncident(nativeID, "zfs_device_state", storagehealth.RiskWarning, fmt.Sprintf("ZFS device %s in pool %s is DEGRADED", displayName, poolName)),
Disk: diskName,
})
case "FAULTED", "FAILED", "OFFLINE", "REMOVED", "UNAVAIL":
out = append(out, poolIncidentProjection{
Incident: makeIncident(nativeID, "zfs_device_state", storagehealth.RiskCritical, fmt.Sprintf("ZFS device %s in pool %s is %s", displayName, poolName, state)),
Disk: diskName,
})
}
}
if vdev.ReadErrors > 0 || vdev.WriteErrors > 0 || vdev.ChecksumErrors > 0 {
out = append(out, poolIncidentProjection{
Incident: makeIncident(nativeID, "zfs_device_errors", storagehealth.RiskWarning, fmt.Sprintf("ZFS device %s reports read=%d write=%d checksum=%d errors", displayName, vdev.ReadErrors, vdev.WriteErrors, vdev.ChecksumErrors)),
Disk: diskName,
})
}
}
return out
}
func canonicalSignalsCoveredByNativeAlert(code string) []string {
switch strings.TrimSpace(code) {
case "truenas_volume_status":
return []string{"zfs_pool_state", "zfs_device_state", "zfs_device_missing"}
case "truenas_scrub":
return []string{"zfs_scan_errors", "zfs_scan_failed"}
default:
return nil
}
}
func incidentFromDatasetState(dataset Dataset, observedAt time.Time) (unifiedresources.ResourceIncident, bool) {
name := strings.TrimSpace(dataset.Name)
if name == "" {
return unifiedresources.ResourceIncident{}, false
}
code := ""
summary := ""
switch {
case dataset.Locked:
code = "zfs_dataset_locked"
summary = fmt.Sprintf("ZFS dataset %s is locked and unavailable", name)
case !dataset.Mounted:
code = "zfs_dataset_unmounted"
summary = fmt.Sprintf("ZFS dataset %s is not mounted", name)
default:
return unifiedresources.ResourceIncident{}, false
}
return unifiedresources.ResourceIncident{
Provider: "truenas",
NativeID: "pool:" + poolName + ":state",
Code: reason.Code,
Severity: reason.Severity,
Source: "pool.query",
Summary: reason.Summary,
StartedAt: observedAt,
Provider: "truenas",
NativeID: "dataset:" + firstNonEmptyString(dataset.ID, name),
Code: code,
Severity: storagehealth.RiskWarning,
Source: "pool.dataset.query",
Summary: summary,
StartedAt: observedAt,
ConfirmationsRequired: 2,
RecoveryConfirmationsRequired: 2,
}, true
}
func primaryIncidentReason(assessment storagehealth.Assessment, code string) (storagehealth.Reason, bool) {
code = strings.TrimSpace(code)
for _, reason := range assessment.Reasons {
if strings.TrimSpace(reason.Code) == code {
return reason, true
func incidentsFromAppState(app App, observedAt time.Time) []unifiedresources.ResourceIncident {
appID := appIncidentKey(app)
if appID == "" {
return nil
}
makeIncident := func(nativeID, code string, severity storagehealth.RiskLevel, summary string) unifiedresources.ResourceIncident {
return unifiedresources.ResourceIncident{
Provider: "truenas",
NativeID: nativeID,
Code: code,
Severity: severity,
Source: "app.query",
Summary: summary,
StartedAt: observedAt,
ConfirmationsRequired: 2,
RecoveryConfirmationsRequired: 2,
}
}
if len(assessment.Reasons) == 0 {
return storagehealth.Reason{}, false
name := appDisplayName(app)
var out []unifiedresources.ResourceIncident
switch strings.ToUpper(strings.TrimSpace(app.State)) {
case "CRASHED":
out = append(out, makeIncident("app:"+appID, "truenas_app_crashed", storagehealth.RiskCritical, fmt.Sprintf("TrueNAS app %s is crashed", name)))
case "STOPPED":
out = append(out, makeIncident("app:"+appID, "truenas_app_stopped", storagehealth.RiskWarning, fmt.Sprintf("TrueNAS app %s is stopped", name)))
}
return assessment.Reasons[0], true
if !strings.EqualFold(strings.TrimSpace(app.State), "RUNNING") {
return out
}
for _, container := range app.Containers {
state := strings.ToUpper(strings.TrimSpace(container.State))
if state != "CRASHED" && state != "EXITED" {
continue
}
containerID := firstNonEmptyString(container.ID, container.ServiceName, container.Image)
if containerID == "" {
continue
}
containerName := firstNonEmptyString(container.ServiceName, container.ID, container.Image)
out = append(out, makeIncident(
"app:"+appID+":container:"+containerID,
"truenas_app_container_failed",
storagehealth.RiskCritical,
fmt.Sprintf("Container %s in TrueNAS app %s is %s", containerName, name, state),
))
}
return out
}
func appIncidentKey(app App) string {
return firstNonEmptyString(strings.TrimSpace(app.ID), strings.TrimSpace(app.Name))
}
func severityFromAlertLevel(level string) (storagehealth.RiskLevel, bool) {
@@ -1320,10 +1560,213 @@ func diskNameFromAlert(alert Alert) string {
}
func assessPool(pool Pool) storagehealth.Assessment {
return storagehealth.AssessZFSPool(models.ZFSPool{
Name: strings.TrimSpace(pool.Name),
State: strings.ToUpper(strings.TrimSpace(pool.Status)),
})
return storagehealth.AssessZFSPool(zfsPoolFromPool(pool))
}
func zfsPoolFromPool(pool Pool) models.ZFSPool {
devices := make([]models.ZFSDevice, 0, len(pool.VDevs))
for _, vdev := range pool.VDevs {
devices = append(devices, models.ZFSDevice{
Name: firstNonEmptyString(vdev.Name, vdev.Disk, vdev.Device, vdev.Path, vdev.ID),
Type: strings.ToLower(strings.TrimSpace(vdev.Type)),
Role: strings.ToLower(strings.TrimSpace(vdev.Role)),
Parent: strings.TrimSpace(vdev.ParentID),
GUID: strings.TrimSpace(vdev.GUID),
Disk: strings.TrimSpace(vdev.Disk),
Path: firstNonEmptyString(strings.TrimSpace(vdev.Path), devicePath(vdev.Device)),
State: strings.ToUpper(strings.TrimSpace(vdev.Status)),
ReadErrors: vdev.ReadErrors,
WriteErrors: vdev.WriteErrors,
ChecksumErrors: vdev.ChecksumErrors,
Missing: vdev.Missing,
Message: strings.TrimSpace(vdev.Message),
})
}
var scanDetails *models.ZFSScan
scanSummary := ""
if pool.Scan != nil {
scanDetails = &models.ZFSScan{
Function: strings.ToUpper(strings.TrimSpace(pool.Scan.Function)),
State: strings.ToUpper(strings.TrimSpace(pool.Scan.State)),
Percentage: pool.Scan.Percentage,
Errors: pool.Scan.Errors,
BytesExamined: pool.Scan.BytesExamined,
BytesToProcess: pool.Scan.BytesToProcess,
TotalSecondsRemaining: pool.Scan.TotalSecondsRemaining,
StartedAt: pool.Scan.StartedAt,
EndedAt: pool.Scan.EndedAt,
}
scanSummary = poolScanSummary(pool)
}
return models.ZFSPool{
Name: strings.TrimSpace(pool.Name),
State: strings.ToUpper(strings.TrimSpace(pool.Status)),
Status: firstNonEmptyString(strings.TrimSpace(pool.StatusDetail), strings.TrimSpace(pool.StatusCode)),
Scan: scanSummary,
ScanDetails: scanDetails,
ReadErrors: pool.ReadErrors,
WriteErrors: pool.WriteErrors,
ChecksumErrors: pool.ChecksumErrors,
Devices: devices,
}
}
func devicePath(device string) string {
device = strings.TrimSpace(device)
if device == "" {
return ""
}
if strings.HasPrefix(device, "/") {
return device
}
return "/dev/" + device
}
func poolScanSummary(pool Pool) string {
if pool.Scan == nil {
return ""
}
function := strings.ToLower(strings.TrimSpace(pool.Scan.Function))
state := strings.ToLower(strings.TrimSpace(pool.Scan.State))
if function == "" {
function = "scan"
}
summary := strings.TrimSpace(function + " " + state)
if pool.Scan.Percentage > 0 && poolScanIsActive(state) {
summary = fmt.Sprintf("%s (%.1f%%)", summary, pool.Scan.Percentage)
}
if pool.Scan.Errors > 0 {
summary = fmt.Sprintf("%s, %d error(s)", summary, pool.Scan.Errors)
}
return summary
}
func poolScanIsActive(state string) bool {
switch strings.ToUpper(strings.TrimSpace(state)) {
case "SCANNING", "RUNNING", "IN_PROGRESS", "INPROGRESS":
return true
default:
return false
}
}
func poolTopologyLabel(pool Pool) string {
var dataTypes []string
for _, vdev := range pool.VDevs {
if vdev.ParentID != "" || !strings.EqualFold(strings.TrimSpace(vdev.Role), "data") {
continue
}
vdevType := strings.ToLower(strings.TrimSpace(vdev.Type))
if vdevType == "" || containsExactString(dataTypes, vdevType) {
continue
}
dataTypes = append(dataTypes, vdevType)
}
if len(dataTypes) == 0 {
return "pool"
}
if len(dataTypes) == 1 && dataTypes[0] == "disk" && len(pool.DiskMembers) > 1 {
return "stripe"
}
return strings.Join(dataTypes, "+")
}
func poolHealthFromTrueNASPool(pool Pool, assessment storagehealth.Assessment, observedAt time.Time) *unifiedresources.PoolHealth {
canonicalState := canonicalPoolState(pool.Status)
severity := assessment.Level
if severity == "" {
severity = storagehealth.RiskHealthy
}
summary := ""
if len(assessment.Reasons) > 0 {
summary = strings.TrimSpace(assessment.Reasons[0].Summary)
} else if canonicalState != "UNKNOWN" {
summary = fmt.Sprintf("ZFS pool %s is %s", strings.TrimSpace(pool.Name), canonicalState)
}
evidenceCodes := make([]string, 0, len(assessment.Reasons))
for _, reason := range assessment.Reasons {
if code := strings.TrimSpace(reason.Code); code != "" && !containsExactString(evidenceCodes, code) {
evidenceCodes = append(evidenceCodes, code)
}
}
source := "pool.query"
if pool.IsBoot {
source = "boot.get_state"
}
return &unifiedresources.PoolHealth{
Scope: "pool",
Provider: "truenas",
NativeID: firstNonEmptyString(pool.GUID, pool.ID, pool.Name),
CanonicalState: canonicalState,
NativeState: strings.ToUpper(strings.TrimSpace(pool.Status)),
Severity: severity,
Summary: summary,
Recommendation: zfsPoolRecommendation(assessment),
Source: source,
EvidenceCodes: evidenceCodes,
ObservedAt: observedAt,
}
}
func containsExactString(values []string, target string) bool {
for _, value := range values {
if value == target {
return true
}
}
return false
}
func canonicalPoolState(state string) string {
switch strings.ToUpper(strings.TrimSpace(state)) {
case "ONLINE":
return "ONLINE"
case "DEGRADED":
return "DEGRADED"
case "FAULTED":
return "FAULTED"
case "OFFLINE":
return "OFFLINE"
case "UNAVAIL":
return "UNAVAIL"
case "REMOVED":
return "UNAVAIL"
case "SUSPENDED":
return "FAULTED"
default:
return "UNKNOWN"
}
}
func zfsPoolRecommendation(assessment storagehealth.Assessment) string {
hasCode := func(codes ...string) bool {
for _, reason := range assessment.Reasons {
for _, code := range codes {
if strings.TrimSpace(reason.Code) == code {
return true
}
}
}
return false
}
switch {
case hasCode("zfs_device_missing", "zfs_device_state"):
return "Identify the affected vdev member, replace or reconnect it if native evidence confirms failure, then verify redundancy and resilver completion."
case hasCode("zfs_scan_errors", "zfs_scan_failed"):
return "Review the native scrub or resilver result, inspect affected devices and cabling, and restore a clean scan before making further storage changes."
case hasCode("zfs_pool_state"):
return "Inspect native pool and vdev status, preserve the remaining redundancy, and restore the pool to ONLINE."
case hasCode("zfs_pool_errors", "zfs_device_errors"):
return "Inspect the affected device and path, review SMART and cabling evidence, then run a scrub; replace hardware only when native evidence supports it."
case hasCode("zfs_resilver_active"):
return "Monitor resilver progress and avoid avoidable pool changes until protection returns to ONLINE."
case hasCode("zfs_scrub_active"):
return "Monitor the scrub to completion and review any reported errors."
default:
return ""
}
}
func assessDisk(disk Disk) storagehealth.Assessment {
@@ -2336,7 +2779,7 @@ func copyFixtureSnapshot(snapshot *FixtureSnapshot) *FixtureSnapshot {
copied := *snapshot
copied.System = cloneSystemInfo(snapshot.System)
copied.Pools = append([]Pool(nil), snapshot.Pools...)
copied.Pools = clonePools(snapshot.Pools)
copied.Datasets = append([]Dataset(nil), snapshot.Datasets...)
copied.Disks = append([]Disk(nil), snapshot.Disks...)
copied.Alerts = append([]Alert(nil), snapshot.Alerts...)
@@ -2349,6 +2792,23 @@ func copyFixtureSnapshot(snapshot *FixtureSnapshot) *FixtureSnapshot {
return &copied
}
func clonePools(pools []Pool) []Pool {
if len(pools) == 0 {
return nil
}
out := make([]Pool, len(pools))
for i := range pools {
out[i] = pools[i]
if pools[i].Scan != nil {
scan := *pools[i].Scan
out[i].Scan = &scan
}
out[i].VDevs = append([]PoolVDev(nil), pools[i].VDevs...)
out[i].DiskMembers = append([]PoolDiskMember(nil), pools[i].DiskMembers...)
}
return out
}
func cloneSystemInfo(system SystemInfo) SystemInfo {
cloned := system
if len(system.TemperatureCelsius) > 0 {
@@ -0,0 +1,154 @@
package truenas
import (
"strings"
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/storagehealth"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
)
func TestProviderProjectsFullZFSHealthAndActionableDatasetAppIncidents(t *testing.T) {
observedAt := time.Date(2026, 7, 24, 10, 0, 0, 0, time.UTC)
records := FixtureRecords(FixtureSnapshot{
CollectedAt: observedAt,
System: SystemInfo{Hostname: "nas-a", Healthy: true},
Pools: []Pool{{
ID: "7",
GUID: "pool-guid",
Name: "tank",
Status: "DEGRADED",
StatusDetail: "One or more devices is unavailable",
ReadErrors: 1,
ChecksumErrors: 2,
Scan: &PoolScan{
Function: "RESILVER",
State: "SCANNING",
Percentage: 33.3,
},
VDevs: []PoolVDev{
{ID: "mirror-guid", GUID: "mirror-guid", Name: "mirror-0", Type: "MIRROR", Role: "data", Status: "DEGRADED"},
{ID: "disk-a", ParentID: "mirror-guid", GUID: "disk-a", Name: "sda", Disk: "sda", Path: "/dev/sda2", Type: "DISK", Role: "data", Status: "ONLINE"},
{ID: "disk-b", ParentID: "mirror-guid", GUID: "disk-b", Name: "sdb", Disk: "sdb", Path: "/dev/disk/by-partuuid/missing", Type: "UNAVAIL_DISK", Role: "data", Status: "UNAVAIL", Missing: true},
{ID: "spare", GUID: "spare", Name: "sdc", Disk: "sdc", Path: "/dev/sdc", Type: "DISK", Role: "spare", Status: "AVAIL"},
},
DiskMembers: []PoolDiskMember{
{Disk: "sda", Status: "ONLINE", Role: "data"},
{Disk: "sdb", Status: "UNAVAIL", Role: "data", Missing: true},
{Disk: "sdc", Status: "AVAIL", Role: "spare"},
},
}},
Datasets: []Dataset{
{ID: "tank/locked", Name: "tank/locked", Pool: "tank", Locked: true},
{ID: "tank/receive", Name: "tank/receive", Pool: "tank", Mounted: true, ReadOnly: true, ReadOnlyReason: DatasetReadOnlyReplicationTarget},
},
Disks: []Disk{
{ID: "disk-a", Name: "sda", Pool: "tank", Status: "ONLINE"},
{ID: "disk-b", Name: "sdb", Pool: "tank", Status: "UNAVAIL"},
{ID: "spare", Name: "sdc", Pool: "tank", Status: "AVAIL"},
},
Apps: []App{
{ID: "crashed", Name: "crashed", State: "CRASHED"},
{ID: "stopped", Name: "stopped", State: "STOPPED"},
{ID: "partial", Name: "partial", State: "RUNNING", Containers: []AppContainer{{ID: "worker", ServiceName: "worker", State: "exited"}}},
},
})
pool := requirePoolRecord(t, records, "tank").Resource
if pool.Storage == nil || pool.Storage.ZFSPool == nil || pool.Storage.PoolHealth == nil {
t.Fatalf("full pool health contract missing: %+v", pool.Storage)
}
if pool.Storage.Topology != "mirror" {
t.Fatalf("topology = %q", pool.Storage.Topology)
}
if pool.Storage.ZFSReadErrors != 1 || pool.Storage.ZFSChecksumErrors != 2 {
t.Fatalf("flattened errors = %+v", pool.Storage)
}
if pool.Storage.ZFSPool.ScanDetails == nil || pool.Storage.ZFSPool.ScanDetails.Function != "RESILVER" || len(pool.Storage.ZFSPool.Devices) != 4 {
t.Fatalf("full ZFS report = %+v", pool.Storage.ZFSPool)
}
if pool.Storage.PoolHealth.CanonicalState != "DEGRADED" ||
pool.Storage.PoolHealth.Severity != storagehealth.RiskCritical ||
!strings.Contains(pool.Storage.PoolHealth.Recommendation, "affected vdev member") {
t.Fatalf("canonical pool health = %+v", pool.Storage.PoolHealth)
}
if !poolHealthHasIncidentCode(pool.Incidents, "zfs_pool_state") ||
!poolHealthHasIncidentCode(pool.Incidents, "zfs_resilver_active") ||
!poolHealthHasIncidentCode(pool.Incidents, "zfs_device_missing") {
t.Fatalf("pool incidents = %+v", pool.Incidents)
}
for _, incident := range pool.Incidents {
if incident.Provider == "truenas" && incident.ConfirmationsRequired != 2 {
t.Fatalf("synthetic incident confirmation contract = %+v", incident)
}
}
locked := requireRecordByNameAndType(t, records, "tank/locked", unifiedresources.ResourceTypeStorage)
if !poolHealthHasIncidentCode(locked.Resource.Incidents, "zfs_dataset_locked") {
t.Fatalf("locked dataset incidents = %+v", locked.Resource.Incidents)
}
receive := requireRecordByNameAndType(t, records, "tank/receive", unifiedresources.ResourceTypeStorage)
if receive.Resource.Status != unifiedresources.StatusOnline || len(receive.Resource.Incidents) != 0 {
t.Fatalf("replication target readonly must remain healthy: %+v", receive.Resource)
}
for name, code := range map[string]string{
"crashed": "truenas_app_crashed",
"stopped": "truenas_app_stopped",
"partial": "truenas_app_container_failed",
} {
record := requireRecordByNameAndType(t, records, name, unifiedresources.ResourceTypeAppContainer)
if !poolHealthHasIncidentCode(record.Resource.Incidents, code) {
t.Fatalf("%s incidents = %+v", name, record.Resource.Incidents)
}
}
}
func TestNativeTrueNASPoolAlertSuppressesEquivalentSyntheticSignalsOnly(t *testing.T) {
records := FixtureRecords(FixtureSnapshot{
CollectedAt: time.Date(2026, 7, 24, 10, 0, 0, 0, time.UTC),
System: SystemInfo{Hostname: "nas-a", Healthy: true},
Pools: []Pool{{
ID: "7",
Name: "tank",
Status: "DEGRADED",
ReadErrors: 3,
}},
Alerts: []Alert{{
ID: "native-volume",
Level: "CRITICAL",
Source: "VolumeStatus",
Message: "Pool tank is DEGRADED",
Datetime: time.Date(2026, 7, 24, 9, 0, 0, 0, time.UTC),
}},
})
pool := requirePoolRecord(t, records, "tank").Resource
if poolHealthHasIncidentCode(pool.Incidents, "zfs_pool_state") {
t.Fatalf("native volume alert must suppress equivalent state alert: %+v", pool.Incidents)
}
if !poolHealthHasIncidentCode(pool.Incidents, "truenas_volume_status") || !poolHealthHasIncidentCode(pool.Incidents, "zfs_pool_errors") {
t.Fatalf("native alert and distinct error evidence must both survive: %+v", pool.Incidents)
}
}
func poolHealthHasIncidentCode(incidents []unifiedresources.ResourceIncident, code string) bool {
for _, incident := range incidents {
if incident.Code == code {
return true
}
}
return false
}
func requireRecordByNameAndType(t *testing.T, records []unifiedresources.IngestRecord, name string, resourceType unifiedresources.ResourceType) unifiedresources.IngestRecord {
t.Helper()
for _, record := range records {
if record.Resource.Name == name && record.Resource.Type == resourceType {
return record
}
}
t.Fatalf("missing %s %q in %+v", resourceType, name, records)
return unifiedresources.IngestRecord{}
}
+59 -6
View File
@@ -41,12 +41,20 @@ type SystemInfo struct {
// Pool mirrors the subset of TrueNAS pool fields needed for unified mapping.
type Pool struct {
ID string
Name string
Status string
TotalBytes int64
UsedBytes int64
FreeBytes int64
ID string
GUID string
Name string
Status string
StatusCode string
StatusDetail string
TotalBytes int64
UsedBytes int64
FreeBytes int64
ReadErrors int64
WriteErrors int64
ChecksumErrors int64
Scan *PoolScan
VDevs []PoolVDev
// IsBoot distinguishes the boot pool collected from boot.get_state from
// data pools. Identity remains connection-scoped at projection time.
IsBoot bool
@@ -66,9 +74,54 @@ type PoolDiskMember struct {
// every path-bearing leaf; Device is the partition/label device (da0p2).
Disk string
Device string
Path string
GUID string
Type string
Role string
// Status is the per-member ZFS state: ONLINE, DEGRADED, FAULTED,
// OFFLINE, UNAVAIL, REMOVED (spares report AVAIL/INUSE).
Status string
// Missing is true only when the native topology reports an unavailable
// datastore row or an unavail_disk vdev. Absence from disk.query alone is
// never treated as evidence that a disk is missing.
Missing bool
ReadErrors int64
WriteErrors int64
ChecksumErrors int64
Message string
}
// PoolVDev preserves the flattened native topology for mirrors, RAIDZ groups,
// spares, special/log/cache vdevs, and their leaf disks.
type PoolVDev struct {
ID string
ParentID string
GUID string
Name string
Type string
Role string
Disk string
Device string
Path string
Status string
ReadErrors int64
WriteErrors int64
ChecksumErrors int64
Missing bool
Message string
}
// PoolScan is the native pool.query/boot.get_state scrub or resilver state.
type PoolScan struct {
Function string
State string
Percentage float64
Errors int64
BytesExamined int64
BytesToProcess int64
TotalSecondsRemaining int64
StartedAt *time.Time
EndedAt *time.Time
}
// DatasetReadOnlyReason describes why a dataset is read-only when the API
+83
View File
@@ -1968,10 +1968,25 @@ func resourceFromCephCluster(cluster models.CephCluster) (Resource, ResourceIden
name = cluster.ID
}
healthChecks := make([]CephHealthCheckMeta, 0, len(cluster.HealthChecks))
evidenceCodes := make([]string, 0, len(cluster.HealthChecks))
for _, check := range cluster.HealthChecks {
healthChecks = append(healthChecks, CephHealthCheckMeta{
Code: strings.TrimSpace(check.Code),
Severity: strings.TrimSpace(check.Severity),
Summary: strings.TrimSpace(check.Summary),
})
if code := strings.TrimSpace(check.Code); code != "" {
evidenceCodes = append(evidenceCodes, code)
}
}
poolHealth := cephPoolHealth(cluster, evidenceCodes)
cephMeta := &CephMeta{
FSID: cluster.FSID,
HealthStatus: cluster.Health,
HealthMessage: cluster.HealthMessage,
HealthChecks: healthChecks,
PoolHealth: poolHealth,
NumMons: cluster.NumMons,
NumMgrs: cluster.NumMgrs,
NumOSDs: cluster.NumOSDs,
@@ -1992,6 +2007,9 @@ func resourceFromCephCluster(cluster models.CephCluster) (Resource, ResourceIden
Ceph: cephMeta,
Tags: cephClusterTags(cluster),
}
if incident, ok := cephHealthIncident(cluster); ok {
resource.Incidents = []ResourceIncident{incident}
}
identity := ResourceIdentity{}
if cluster.FSID != "" {
@@ -2002,6 +2020,71 @@ func resourceFromCephCluster(cluster models.CephCluster) (Resource, ResourceIden
return resource, identity
}
func cephPoolHealth(cluster models.CephCluster, evidenceCodes []string) *PoolHealth {
nativeState := strings.ToUpper(strings.TrimSpace(cluster.Health))
normalized := strings.TrimPrefix(nativeState, "HEALTH_")
canonicalState := "UNKNOWN"
severity := storagehealth.RiskHealthy
recommendation := ""
switch normalized {
case "OK":
canonicalState = "ONLINE"
case "WARN":
canonicalState = "DEGRADED"
severity = storagehealth.RiskWarning
recommendation = "Review the native Ceph health checks, avoid high-risk cluster changes, and restore HEALTH_OK."
case "ERR":
canonicalState = "FAULTED"
severity = storagehealth.RiskCritical
recommendation = "Review the native Ceph health checks immediately and restore HEALTH_OK; Pulse does not infer a failed OSD or disk without native evidence."
}
summary := strings.TrimSpace(cluster.HealthMessage)
if summary == "" && normalized != "" {
summary = "Ceph cluster health is HEALTH_" + normalized
}
return &PoolHealth{
Scope: "cluster",
Provider: "ceph",
NativeID: firstNonEmpty(cluster.FSID, cluster.ID),
CanonicalState: canonicalState,
NativeState: nativeState,
Severity: severity,
Summary: summary,
Recommendation: recommendation,
Source: strings.TrimSpace(cluster.Source),
EvidenceCodes: uniqueStrings(evidenceCodes),
ObservedAt: cluster.LastUpdated,
}
}
func cephHealthIncident(cluster models.CephCluster) (ResourceIncident, bool) {
nativeState := strings.TrimPrefix(strings.ToUpper(strings.TrimSpace(cluster.Health)), "HEALTH_")
severity := storagehealth.RiskHealthy
switch nativeState {
case "WARN":
severity = storagehealth.RiskWarning
case "ERR":
severity = storagehealth.RiskCritical
default:
return ResourceIncident{}, false
}
summary := strings.TrimSpace(cluster.HealthMessage)
if summary == "" {
summary = "Ceph cluster health is HEALTH_" + nativeState
}
return ResourceIncident{
Provider: "ceph",
NativeID: firstNonEmpty(cluster.FSID, cluster.ID),
Code: "ceph_cluster_health",
Severity: severity,
Source: strings.TrimSpace(cluster.Source),
Summary: summary,
StartedAt: cluster.LastUpdated,
ConfirmationsRequired: 2,
RecoveryConfirmationsRequired: 2,
}, true
}
func convertCephPools(pools []models.CephPool) []CephPoolMeta {
if len(pools) == 0 {
return nil
@@ -0,0 +1,65 @@
package unifiedresources
import (
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
"github.com/rcourtman/pulse-go-rewrite/internal/storagehealth"
)
func TestCephUsesNativeClusterEvidenceThroughPoolHealthContract(t *testing.T) {
observedAt := time.Date(2026, 7, 24, 10, 0, 0, 0, time.UTC)
resource, identity := resourceFromCephCluster(models.CephCluster{
ID: "pve-a-fsid",
Instance: "pve-a",
Source: models.CephClusterSourceProxmoxAPI,
Name: "Ceph",
FSID: "fsid-1",
Health: "HEALTH_WARN",
HealthMessage: "1 osd down; 42 pgs degraded",
HealthChecks: []models.CephHealthCheck{
{Code: "OSD_DOWN", Severity: "HEALTH_WARN", Summary: "1 osd down"},
{Code: "PG_DEGRADED", Severity: "HEALTH_WARN", Summary: "42 pgs degraded"},
},
LastUpdated: observedAt,
})
if identity.MachineID != "fsid-1" {
t.Fatalf("identity = %+v", identity)
}
if resource.Type != ResourceTypeCeph || resource.Status != StatusWarning || resource.Ceph == nil || resource.Ceph.PoolHealth == nil {
t.Fatalf("ceph resource = %+v", resource)
}
health := resource.Ceph.PoolHealth
if health.Scope != "cluster" || health.CanonicalState != "DEGRADED" || health.NativeState != "HEALTH_WARN" || health.Severity != storagehealth.RiskWarning {
t.Fatalf("pool health = %+v", health)
}
if len(health.EvidenceCodes) != 2 || len(resource.Ceph.HealthChecks) != 2 {
t.Fatalf("native evidence was not preserved: health=%+v ceph=%+v", health, resource.Ceph)
}
if len(resource.Incidents) != 1 || resource.Incidents[0].Code != "ceph_cluster_health" || resource.Incidents[0].NativeID != "fsid-1" {
t.Fatalf("cluster incident = %+v", resource.Incidents)
}
if resource.Incidents[0].ConfirmationsRequired != 2 || resource.Incidents[0].RecoveryConfirmationsRequired != 2 {
t.Fatalf("cluster lifecycle = %+v", resource.Incidents[0])
}
if resource.PhysicalDisk != nil || resource.Storage != nil {
t.Fatal("cluster health must not invent a failed disk or ZFS storage resource")
}
}
func TestCephUnknownHealthDoesNotInventIncident(t *testing.T) {
resource, _ := resourceFromCephCluster(models.CephCluster{
ID: "cluster-without-health",
Instance: "pve-a",
Health: "NOT_REPORTED",
LastUpdated: time.Date(2026, 7, 24, 10, 0, 0, 0, time.UTC),
})
if resource.Status != StatusUnknown || len(resource.Incidents) != 0 {
t.Fatalf("unknown health must remain unknown: %+v", resource)
}
if resource.Ceph == nil || resource.Ceph.PoolHealth == nil || resource.Ceph.PoolHealth.CanonicalState != "UNKNOWN" {
t.Fatalf("unknown pool-health envelope missing: %+v", resource.Ceph)
}
}
+28
View File
@@ -169,6 +169,32 @@ func cloneStorageMeta(in *StorageMeta) *StorageMeta {
out.ConsumerTypes = cloneStringSlice(in.ConsumerTypes)
out.TopConsumers = cloneStorageConsumerMetaSlice(in.TopConsumers)
out.Risk = cloneStorageRisk(in.Risk)
out.PoolHealth = clonePoolHealth(in.PoolHealth)
out.ZFSPool = cloneZFSPool(in.ZFSPool)
return &out
}
func clonePoolHealth(in *PoolHealth) *PoolHealth {
if in == nil {
return nil
}
out := *in
out.EvidenceCodes = cloneStringSlice(in.EvidenceCodes)
return &out
}
func cloneZFSPool(in *models.ZFSPool) *models.ZFSPool {
if in == nil {
return nil
}
out := *in
out.Devices = append([]models.ZFSDevice(nil), in.Devices...)
if in.ScanDetails != nil {
scan := *in.ScanDetails
scan.StartedAt = cloneTimePtr(in.ScanDetails.StartedAt)
scan.EndedAt = cloneTimePtr(in.ScanDetails.EndedAt)
out.ScanDetails = &scan
}
return &out
}
@@ -611,6 +637,8 @@ func cloneCephMeta(in *CephMeta) *CephMeta {
out := *in
out.Pools = cloneCephPoolMetaSlice(in.Pools)
out.Services = cloneCephServiceMetaSlice(in.Services)
out.HealthChecks = append([]CephHealthCheckMeta(nil), in.HealthChecks...)
out.PoolHealth = clonePoolHealth(in.PoolHealth)
return &out
}
@@ -11,6 +11,23 @@ const (
func IncidentActionForResource(resource *Resource, incident ResourceIncident, category string) (string, string) {
baseType := resourceBaseType(resource)
switch incident.Code {
case "zfs_pool_state":
if incident.Severity == storagehealth.RiskCritical {
return IncidentUrgencyNow, "Inspect native pool and vdev status, preserve remaining redundancy, and restore the pool; replace hardware only when native evidence supports it"
}
return IncidentUrgencyToday, "Inspect native pool and vdev status and plan maintenance to restore ONLINE state"
case "zfs_device_state", "zfs_device_missing":
if incident.Severity == storagehealth.RiskCritical {
return IncidentUrgencyNow, "Confirm the affected member in native topology, reconnect or replace it when supported by device evidence, and verify resilver completion"
}
return IncidentUrgencyToday, "Inspect the affected member and path, preserve redundancy, and plan evidence-backed maintenance"
case "zfs_pool_errors", "zfs_device_errors", "zfs_scan_errors", "zfs_scan_failed":
if incident.Severity == storagehealth.RiskCritical {
return IncidentUrgencyNow, "Review the native scan and device evidence, inspect SMART and cabling, and replace hardware only when the evidence identifies it"
}
return IncidentUrgencyToday, "Inspect native error counters, SMART, and cabling, then run or review a scrub before deciding on replacement"
}
switch category {
case IncidentCategoryProtection:
if incident.Severity == storagehealth.RiskCritical {
@@ -48,6 +65,11 @@ func IncidentActionForResource(resource *Resource, incident ResourceIncident, ca
return IncidentUrgencyNow, "Restore storage availability immediately"
}
return IncidentUrgencyToday, "Investigate degraded storage availability"
case IncidentCategoryWorkloadHealth:
if incident.Severity == storagehealth.RiskCritical {
return IncidentUrgencyNow, "Review the app and container logs, restore the failed workload, and confirm it remains running"
}
return IncidentUrgencyToday, "Confirm whether the app should be running, then start it or suppress the alert for intentional downtime"
default:
if baseType == ResourceTypeAgent || baseType == ResourceTypeVM {
if incident.Severity == storagehealth.RiskCritical {
@@ -9,6 +9,7 @@ const (
IncidentCategoryRecoverability = "recoverability"
IncidentCategoryAvailability = "availability"
IncidentCategoryDiskHealth = "disk-health"
IncidentCategoryWorkloadHealth = "workload-health"
IncidentCategoryHealth = "health"
)
@@ -19,9 +20,9 @@ func IncidentCategoryForResource(resource *Resource, incident ResourceIncident)
code := strings.TrimSpace(incident.Code)
switch code {
case "raid_degraded", "raid_unavailable", "unraid_invalid_disks", "unraid_disabled_disks", "unraid_missing_disks", "unraid_parity_unavailable", "unraid_no_parity", "zfs_pool_state":
case "raid_degraded", "raid_unavailable", "unraid_invalid_disks", "unraid_disabled_disks", "unraid_missing_disks", "unraid_parity_unavailable", "unraid_no_parity", "zfs_pool_state", "zfs_device_state", "zfs_device_missing":
return IncidentCategoryProtection
case "raid_rebuilding", "unraid_sync_active":
case "raid_rebuilding", "unraid_sync_active", "zfs_resilver_active", "zfs_scrub_active":
return IncidentCategoryRebuild
case "capacity_runway_low":
if resource.Type == ResourceTypePBS || (resource.Storage != nil && IsBackupStorageResource(resource.Storage)) {
@@ -30,10 +31,12 @@ func IncidentCategoryForResource(resource *Resource, incident ResourceIncident)
return IncidentCategoryCapacity
case "pbs_datastore_state", "pbs_datastore_error", "backup_target_degraded":
return IncidentCategoryRecoverability
case "disk_failed", "disk_unavailable", "disk_smart_failed", "disk_wearout", "disk_health":
case "disk_failed", "disk_unavailable", "disk_smart_failed", "disk_wearout", "disk_health", "zfs_pool_errors", "zfs_device_errors", "zfs_scan_errors", "zfs_scan_failed":
return IncidentCategoryDiskHealth
case "availability_unreachable":
case "availability_unreachable", "zfs_dataset_locked", "zfs_dataset_unmounted":
return IncidentCategoryAvailability
case "truenas_app_crashed", "truenas_app_stopped", "truenas_app_container_failed":
return IncidentCategoryWorkloadHealth
}
if resource.Type == ResourceTypeNetworkEndpoint || len(AvailabilityChecksForResource(*resource)) > 0 {
@@ -1,6 +1,7 @@
package unifiedresources
import (
"strings"
"testing"
"github.com/rcourtman/pulse-go-rewrite/internal/storagehealth"
@@ -170,6 +171,24 @@ func TestIncidentCategoryForResource_TrimmedCodeMatches(t *testing.T) {
}
}
func TestZFSIncidentActionsRequireNativeEvidenceBeforeReplacement(t *testing.T) {
resource := &Resource{ID: "pool:tank", Type: ResourceTypeStorage}
for _, incident := range []ResourceIncident{
{Code: "zfs_pool_state", Severity: storagehealth.RiskCritical},
{Code: "zfs_scan_errors", Severity: storagehealth.RiskCritical},
{Code: "zfs_pool_errors", Severity: storagehealth.RiskWarning},
} {
category := IncidentCategoryForResource(resource, incident)
_, action := IncidentActionForResource(resource, incident, category)
if !strings.Contains(action, "native") && !strings.Contains(action, "SMART") {
t.Fatalf("%s action lacks native evidence gate: %q", incident.Code, action)
}
if strings.Contains(action, "Replace the affected disk") {
t.Fatalf("%s action inferred a failed disk: %q", incident.Code, action)
}
}
}
// --- incidentSummaryPreference ---
func TestIncidentSummaryPreference_EmptySummary(t *testing.T) {
@@ -12,6 +12,8 @@ func IncidentLabelForResource(resource *Resource, incident ResourceIncident, cat
return "Backup Coverage At Risk"
case IncidentCategoryDiskHealth:
return "Disk Health Risk"
case IncidentCategoryWorkloadHealth:
return "Workload Health Issue"
case IncidentCategoryAvailability:
switch resourceBaseType(resource) {
case ResourceTypeAgent:
@@ -16,6 +16,8 @@ func IncidentPriorityForResource(resource *Resource, incident ResourceIncident,
priority += 150
case IncidentCategoryDiskHealth:
priority += 100
case IncidentCategoryWorkloadHealth:
priority += 125
default:
priority += 50
}
+1 -1
View File
@@ -2991,7 +2991,7 @@ func TestResourceRegistry_IngestResourcesDerivesPrimaryIncidentRollups(t *testin
if !ok {
t.Fatal("expected storage resource")
}
if storage.IncidentCount != 2 || storage.IncidentCode != "zfs_pool_state" || storage.IncidentSeverity != storagehealth.RiskWarning || storage.IncidentSummary != "ZFS pool tank is DEGRADED" || storage.IncidentCategory != IncidentCategoryProtection || storage.IncidentLabel != "Protection Reduced" || storage.IncidentPriority != 3400 || storage.IncidentImpactSummary != "" || storage.IncidentUrgency != IncidentUrgencyToday || storage.IncidentAction != "Investigate degraded protection and schedule maintenance to restore redundancy" {
if storage.IncidentCount != 2 || storage.IncidentCode != "zfs_pool_state" || storage.IncidentSeverity != storagehealth.RiskWarning || storage.IncidentSummary != "ZFS pool tank is DEGRADED" || storage.IncidentCategory != IncidentCategoryProtection || storage.IncidentLabel != "Protection Reduced" || storage.IncidentPriority != 3400 || storage.IncidentImpactSummary != "" || storage.IncidentUrgency != IncidentUrgencyToday || storage.IncidentAction != "Inspect native pool and vdev status and plan maintenance to restore ONLINE state" {
t.Fatalf("unexpected storage incident rollup %+v", storage)
}
+3 -3
View File
@@ -63,11 +63,11 @@ func statusFromPhysicalDisk(health string) ResourceStatus {
func statusFromCephHealth(health string) ResourceStatus {
switch strings.ToUpper(strings.TrimSpace(health)) {
case "HEALTH_OK":
case "HEALTH_OK", "OK":
return StatusOnline
case "HEALTH_WARN":
case "HEALTH_WARN", "WARN":
return StatusWarning
case "HEALTH_ERR":
case "HEALTH_ERR", "ERR":
return StatusOffline
default:
return StatusUnknown
+2 -2
View File
@@ -117,13 +117,13 @@ func StorageRiskSemantics(risk *StorageRisk) ([]string, bool, bool, string, stri
codes = append(codes, code)
}
switch code {
case "raid_degraded", "raid_unavailable", "unraid_invalid_disks", "unraid_disabled_disks", "unraid_missing_disks", "unraid_parity_unavailable", "unraid_no_parity", "zfs_pool_state":
case "raid_degraded", "raid_unavailable", "unraid_invalid_disks", "unraid_disabled_disks", "unraid_missing_disks", "unraid_parity_unavailable", "unraid_no_parity", "zfs_pool_state", "zfs_device_state", "zfs_device_missing":
protectionReduced = true
if priority := protectionSummaryPriority(code); protectionSummary == "" || priority > protectionPriority {
protectionSummary = strings.TrimSpace(reason.Summary)
protectionPriority = priority
}
case "raid_rebuilding", "unraid_sync_active":
case "raid_rebuilding", "unraid_sync_active", "zfs_resilver_active":
rebuildInProgress = true
if rebuildSummary == "" {
rebuildSummary = strings.TrimSpace(reason.Summary)
+46 -18
View File
@@ -403,6 +403,7 @@ type StorageMeta struct {
ProtectionSummary string `json:"protectionSummary,omitempty"`
RebuildInProgress bool `json:"rebuildInProgress,omitempty"`
RebuildSummary string `json:"rebuildSummary,omitempty"`
PoolHealth *PoolHealth `json:"poolHealth,omitempty"`
// Accessibility metadata.
Nodes []string `json:"nodes,omitempty"` // PVE nodes where this storage is accessible
@@ -437,13 +438,32 @@ type StorageConsumerMeta struct {
}
type ResourceIncident struct {
Provider string `json:"provider,omitempty"`
NativeID string `json:"nativeId,omitempty"`
Code string `json:"code"`
Severity storagehealth.RiskLevel `json:"severity"`
Source string `json:"source,omitempty"`
Summary string `json:"summary"`
StartedAt time.Time `json:"startedAt,omitempty"`
Provider string `json:"provider,omitempty"`
NativeID string `json:"nativeId,omitempty"`
Code string `json:"code"`
Severity storagehealth.RiskLevel `json:"severity"`
Source string `json:"source,omitempty"`
Summary string `json:"summary"`
StartedAt time.Time `json:"startedAt,omitempty"`
ConfirmationsRequired int `json:"confirmationsRequired,omitempty"`
RecoveryConfirmationsRequired int `json:"recoveryConfirmationsRequired,omitempty"`
}
// PoolHealth is the provider-neutral health envelope shared by storage pools
// and pool-like cluster authorities. Native state and evidence are preserved;
// CanonicalState is the only normalized cross-provider claim.
type PoolHealth struct {
Scope string `json:"scope"`
Provider string `json:"provider"`
NativeID string `json:"nativeId,omitempty"`
CanonicalState string `json:"canonicalState"`
NativeState string `json:"nativeState,omitempty"`
Severity storagehealth.RiskLevel `json:"severity"`
Summary string `json:"summary,omitempty"`
Recommendation string `json:"recommendation,omitempty"`
Source string `json:"source,omitempty"`
EvidenceCodes []string `json:"evidenceCodes,omitempty"`
ObservedAt time.Time `json:"observedAt,omitempty"`
}
// PhysicalDiskMeta contains physical disk-specific metadata.
@@ -522,17 +542,25 @@ type PhysicalDiskRiskReason struct {
// CephMeta contains Ceph cluster-specific metadata.
type CephMeta struct {
FSID string `json:"fsid,omitempty"`
HealthStatus string `json:"healthStatus"`
HealthMessage string `json:"healthMessage,omitempty"`
NumMons int `json:"numMons"`
NumMgrs int `json:"numMgrs"`
NumOSDs int `json:"numOsds"`
NumOSDsUp int `json:"numOsdsUp"`
NumOSDsIn int `json:"numOsdsIn"`
NumPGs int `json:"numPGs"`
Pools []CephPoolMeta `json:"pools,omitempty"`
Services []CephServiceMeta `json:"services,omitempty"`
FSID string `json:"fsid,omitempty"`
HealthStatus string `json:"healthStatus"`
HealthMessage string `json:"healthMessage,omitempty"`
HealthChecks []CephHealthCheckMeta `json:"healthChecks,omitempty"`
PoolHealth *PoolHealth `json:"poolHealth,omitempty"`
NumMons int `json:"numMons"`
NumMgrs int `json:"numMgrs"`
NumOSDs int `json:"numOsds"`
NumOSDsUp int `json:"numOsdsUp"`
NumOSDsIn int `json:"numOsdsIn"`
NumPGs int `json:"numPGs"`
Pools []CephPoolMeta `json:"pools,omitempty"`
Services []CephServiceMeta `json:"services,omitempty"`
}
type CephHealthCheckMeta struct {
Code string `json:"code"`
Severity string `json:"severity,omitempty"`
Summary string `json:"summary,omitempty"`
}
// CephPoolMeta describes a Ceph storage pool.