Restore TrueNAS CORE metrics and keep queued alerts within quiet hours (#2379)

## Additional repair in this publication

Restore native TrueNAS CORE monitoring for #2077 across CPU, memory,
measured temperature, device-bound network/disk I/O, and History. Native
RRD rows use external timing and legends rather than embedded
timestamps; empty/null buckets remain distinct from measured zero. CPU
states are normalized, ARC stays aligned with free RAM, stale readings
are not presented as current, graph selection/splitting is bounded, and
absent device graphs cannot produce misleading partial host totals.
Measured temperature also reaches the existing local and persistent
Thermals paths.

The complete publication range is now
`4317db177e341a0ca52b649c258ac43bc0135d0b..b32a4369e7f55c39a4baed0f122336384185618d`.
It preserves every commit in the preceding proposal and adds native CORE
candidate commits `b50313c59998d0e1506ab5ca42730bbb2e386dcd`,
`c78840e3717b7dd3794ed401529e5f4bdcab2e50`,
`bab5cf77d60f5252311bad05f28aecb4d76f1de1`,
`579eee290759f5518f6ef16d0b9189a2dec7d856`, and
`45eec9a20eac55a356acfb1c17833cd6c8d067ad`, integrated by
`b32a4369e7f55c39a4baed0f122336384185618d`.

The previous head completed its checks. Governance run `36949586647`
failed the remaining subsystem-lookup list assertion, while its reported
build/test, frontend, eight E2E shards/verdict, boundary, docs, security
and benchmark checks succeeded. Additive
`50c24260cab18db1609be0bb1ea91d89c6c99f70` aligns that assertion with
the existing startup verification, without removing coverage. Because a
corrected head is necessary, the independently reviewed CORE repair is
included now; this does not interrupt a still-checking head.

Recorded CORE validation: `source-7of9at5x` passes full TrueNAS
correctness/races/vet and connected monitoring controls; every TrueNAS
blob is unchanged here. `source-6mf6bu65` reproduces the predetermined
parent thermal-write fault and passes final connected/writer
correctness/races/vet, including local/persistent History and the
local-window fast path. `source-d3jaf8_z` passes real embedded frontend
and server/agent builds using verified matching dependency inputs. The
integrated CORE/monitoring/registry source exactly matches that proved
candidate. Earlier ARC/cancellation failures were corrected, not
dismissed. `source-woncpjdu` remains overall failed after 44 frontend
cases/typecheck/assets passed, because it selected the wrong module's Go
cache; it is not a full pass. Native appliance/browser/containing-stable
acceptance remains separate, with compatible next-patch placement owned
rather than adding this work to the frozen v6.4.6 quartet.

Publication provenance passes; both available upstream tips match the
inventory and are retained ancestors. Current-workspace
registry/contract and per-commit completion guards pass. The gate's
confined native-Git executor passes all 163 subsystem lookup fixtures,
including the previously failed assertion. Earlier proposal evidence
remains applicable to its unchanged scope, as retained here:

## What changed and why

Queued alert notifications recheck the current quiet-hours schedule when
they become due, including continuous schedules and late replay. Held
items do not consume a provider attempt; mixed batches deliver only
eligible alerts, preserving destinations, retry budgets,
acknowledgements, cancellation and recovery. Startup binds saved policy
before persisted work starts. Schedule edits do not accelerate
already-future deadlines.

Agent troubleshooting distinguishes permission from an admitted,
connected command channel, addressing the confusion in discussion #2333.
Notification troubleshooting distinguishes a direct test from real alert
delivery, explains retained-failure recovery and duplicate-retry risks,
and replaces an unbounded email pipe that masked reader failures with
bounded local readers. Both Docker streams and exit status matter; raw
provider bodies may expose private information. Source guides and
shipped mirrors agree.

## Complete range and proposal repair

Published main base `4317db177e341a0ca52b649c258ac43bc0135d0b` through
proposal tip `57cd5171c234c5f9626625e2bc0b237de6b4c277`:

- Quiet-hours runtime/tests/contracts
`5213b794b2856fe260ccb3fade0af29efddeda44`, registry-order correction
`c3f627d0d68811cbcc0d7466e0376e83679a0658`, integration
`defaa2344de20d51a681f76b9cf7a57fd215ba7d`.
- Agent-channel guide `c08583dd54a8462566ad34094fcc7e8f577bf854`,
integration `fc4deb6c177183a06f201794a4fd90735ab9fd06`.
- Notification guide/tests `76e1aea061bb60d0922e27fb799a79198bd8de07`.
- Additive governance fixture corrections
`96986379d1d88d81ea56dc872ba3183b11e35419` and
`57cd5171c234c5f9626625e2bc0b237de6b4c277`.

The preceding proposal finished CI: governance run `36940673378` failed
two fixture expectations after the reviewed startup test was added to
the verification registry. The correction aligns all three affected
exact-list expectations, without changing policy, coverage or runtime.
Every previously reviewed identity and author is unchanged. All sixteen
files in the earlier proposal still match it exactly. The newer
independent guide is included now because a corrective proposal head is
needed, not merely to restart an already-running batch.

## Validation

- Recorded `source-5mneylzw` reproduced predetermined predecessor faults
and passed final affected alerts/notifications correctness/races,
bootstrap races, vet and package builds. `source-fmtyrqhz` passed
final-tip equivalence and focused replay/startup races.
- Recorded agent-guide controls and thirteen recipe tests passed in
`source-5y8u5a2z`; `source-059hcqfl` passed 25 documentation/mirror
cases. These are not native appliance recovery.
- This gate's confined native-Git fixture executor passed all 132
governance fixture tests and all 38
notification/request-log/mirror/agent-troubleshooting cases. The first
partial governance correction still failed the later monitoring
assertion; that failed attempt is retained, followed by the complete
correction and pass.
- Public-doc checker passed for 211 Markdown files. Publication
provenance, whitespace, each source commit's completion guard, registry
audit and contract audit passed. Browser guard explicitly skipped: no
user-visible frontend source changed.
- Earlier `source-t_33yywy` had 97 passes and an unchanged agent-profile
timeout; later omitted-check clients did not execute during gateway
restart. Relevant changed-guide checks are now complete; the unrelated
token-profile timeout and frontend-only self-chosen proof remain
separate follow-through, not full-suite passes.
- Optional `source-44iq3ddd` whole-profile VM audit remains failed on
unavailable repository identity/wider layout. Actual-workspace audits
pass, but do not change that receipt. Recorded Go proofs used 1.26.7,
not the declared 1.26.8 environment; no full application build is
claimed.

Exact-head CI must pass before landing. Natural schedule/day/DST,
destination/recipient and installed command-channel acceptance remain
separate. This neither changes the frozen v6.4.6 quartet nor approves
any release; compatible quiet-hours next-patch placement stays owned
separately.


[How Pulse handles
triage](https://github.com/rcourtman/Pulse/blob/main/docs/AI_TRANSPARENCY.md)
This commit is contained in:
pulse-triage[bot]
2026-10-02 02:31:40 +00:00
committed by GitHub
30 changed files with 2786 additions and 139 deletions
+97 -23
View File
@@ -250,32 +250,72 @@ A healthy heartbeat indicates Pulse monitoring-loop progress, not successful
delivery of every resource alert or external reachability of your services.
Continue checking delivery activity for destination failures.
#### Test succeeds but real alerts are missing
A test sends directly to its destination: it skips the persistent delivery queue
and is not listed in **Recent delivery activity**. It does not prove that a real
alert was generated, routed or delivered to the intended recipient.
- Open **Alerts → Notifications**. If **Notifications are paused** is shown,
configured destinations and a successful test do not enable real delivery.
Turn delivery on there only when you intend to send alerts.
- Check the affected alert, the destination's **Enabled** state, minimum alert
severity and tag filters. Review quiet hours and any mute, acknowledgement or
maintenance policy before treating an absent attempt as a transport failure.
- Use **Recent delivery activity** to correlate the original alert, destination
and absolute timestamp, including held-notification reasons. An empty window
is not proof of healthy delivery; an **unavailable** read is not an empty log.
Do not create an outage or repeat a notification storm to populate it.
#### Recover retained delivery failures
Pulse shows a delivery warning for failed or dead-lettered notifications in its
persistent queue, not for every recoverable retry. **Recent delivery activity**
includes safely redacted provider errors; completed attempts remain for 7 days
and dead-letter attempts for 30 days. Start with the failure class and timestamp:
| Failure | Check before retrying |
| --- | --- |
| Authentication | Destination credentials and account permissions, locally; never post them. |
| Rate limited | Provider limits and delivery volume; repeated tests or retries can make this worse. |
| Connectivity | DNS, firewall, proxy and reachability from the Pulse server, not just your browser. |
| TLS | Certificate trust, expiry and hostname matching; do not disable verification to diagnose it. |
| Configuration / rejected | Enabled destination, required fields and the provider's endpoint or payload requirements. |
| Server error / unknown | Destination service status and a relevant, bounded local error excerpt. |
Save the corrected destination settings and send one test; check receipt at the
intended destination. **Retry retained deliveries** gives terminal failures a
fresh retry budget, but a destination that accepted an earlier attempt may
receive a duplicate. Review the confirmation's delivery count and provider
limits before retrying. A successful test does not itself retry retained items.
Use **Dismiss retained failures** only when those deliveries should not be sent.
Dismissal clears the warning without retrying them; delivery history remains.
Neither action deletes the audit trail. Do not delete `notification_queue.db`
or audit data to clear the warning.
#### Emails not sending
- Open **Alerts → Notifications** first. Pulse shows a delivery warning when
failed or dead-lettered notifications remain in the persistent queue; a
missing queue-health read is shown as unavailable rather than healthy.
- **Recent delivery activity** appears directly below that warning. It names
the destination and affected alert, shows an absolute timestamp for timeline
correlation, and includes safely redacted provider errors. Completed attempts
remain for 7 days and dead-letter attempts remain for 30 days.
- After correcting the destination, use **Retry retained deliveries**. Use
**Dismiss retained failures** only when those deliveries should not be sent.
Both actions preserve delivery history; do not delete `notification_queue.db`
to clear the warning.
- Check SMTP settings in **Alerts → Notifications**.
- Check logs: `docker logs pulse | grep email`.
- Ensure your SMTP provider allows the connection (e.g., Gmail App Passwords).
Follow [retained-failure recovery](#recover-retained-delivery-failures) first.
Check SMTP host, port, sender, recipients, authentication and TLS settings in
**Alerts → Notifications** against your provider's requirements (some providers
require an app password). Keep passwords in the settings form, not a diagnostic
command or report. If the delivery error is insufficient, inspect
[bounded notification logs](#inspect-notification-logs) locally.
#### Webhooks failing
- Check the delivery warning in **Alerts → Notifications** and use **Send test**
after correcting the destination. Recoverable retries do not trigger the
warning; retained terminal failures do.
- If the test succeeds, use **Retry retained deliveries** to give the retained
items a fresh retry budget. Dismiss them only when delivery is no longer
wanted; neither action deletes the audit trail.
- Verify the URL is reachable from the Pulse server.
- If targeting private IPs, allow them in **Settings → System → Network → Webhook Security**.
- Check Pulse logs for HTTP status codes and response bodies.
Follow [retained-failure recovery](#recover-retained-delivery-failures) first.
Use the failure class and HTTP status to check the provider's endpoint and
payload requirements. Verify reachability from the Pulse server. For an intended
private destination, review **Settings → System → Network → Webhook Security**;
do not broadly weaken network or TLS controls just to make a test pass.
Prefer the redacted delivery error over raw provider response bodies. A provider
can echo credentials or private content in its response; do not post it wholesale
or enable debug logging just to collect it. If needed, inspect
[bounded notification logs](#inspect-notification-logs) and share only the
consequential, manually redacted error.
### TrueNAS
@@ -309,6 +349,40 @@ Continue checking delivery activity for destination failures.
## 🛠️ Advanced Diagnostics
### Inspect Notification Logs
Prefer **Recent delivery activity** in **Alerts → Notifications**. If a local log
is needed, run only the command for your deployment, on the Pulse host with an
account authorised to read its logs. For Proxmox LXC, run the systemd command
inside the Pulse container, not on the Proxmox host. Adjust the time window to
the original incident and substitute your actual service or container name
(`pulse-backend` on some older systemd installs). These examples read at most
200 records from the last 15 minutes; they do not follow the log or send a test.
```bash
# systemd / Proxmox LXC
journalctl -u pulse --since '15 minutes ago' --lines 200 --no-pager
```
```bash
# Docker
docker logs --since 15m --tail 200 pulse
```
Docker can write application logs to either stdout or stderr; inspect both.
Do not pipe the reader into `grep email`: it can miss SMTP or webhook errors
and hide a failed read behind a matching partial line. A nonzero reader exit,
access error or missing service/container is a failed read, not “no delivery
errors”. Even a successful empty read is inconclusive: the window, retained
logs or selected instance may differ.
These local excerpts are **not sanitised**. Do not post them wholesale. Share
only the relevant timestamp, method, HTTP status or SMTP error code and a
manually redacted error. Remove credentials, cookies, secret URLs, addresses
and private host or personal information, including anything echoed by the
provider. Never upload full environments, configuration, a queue database or
audit data. See [Getting Help](#-getting-help).
### Correlate Logs with Requests
For a failed HTTP API request, inspect its response in your authenticated
+37
View File
@@ -1000,6 +1000,11 @@ LXC guests** in Settings → System → General and any configured
guests and guests already linked to an online guest-local agent. In the latter
case, check that guest agent's Docker monitoring instead.
If the owning node says **Remote control blocked**, first check its
[command channel](#commands-enabled-but-remote-control-blocked). A successful
root `pct exec` check does not establish that Pulse can reach that node's agent;
do not repeat the guest probe when those results are already known.
The host-side path requires the guest's `docker` executable and
`/var/run/docker.sock` in the root `pct exec` context. A working Docker CLI in a
user's login session does not prove that this context can reach the daemon;
@@ -1036,6 +1041,38 @@ Review output before posting and redact any private endpoint in an error. Do not
change LXC privilege, `keyctl`, socket permissions or Docker versions merely to
test a guess.
### Commands enabled but remote control blocked
**Remote control blocked** means the agent reports commands enabled, but Pulse
has no admitted command channel connected for it. Agent Doctor's
`commands command-capable · credential grants exec` line describes the reported
local command ceiling and credential scope, not a connected session. A fresh
monitoring report or **Automatic updates ready** also does not prove that this
separate WebSocket channel is working. Changing Proxmox API permissions cannot
repair a Pulse command-channel connection.
For a systemd agent, inspect the recent journal **locally on the affected node**:
```bash
sudo journalctl -u pulse-agent.service --since '15 minutes ago' -n 200 --no-pager --output=cat
```
Look for **Connected and registered with Pulse command server**, or
**WebSocket connection failed repeatedly, reconnecting** and its error. A
`dial websocket` error is a connection/handshake failure; `registration failed`
means registration was attempted but not accepted. Scope alone does not prove
that the credential's host/agent binding was admitted. No matching entry is
inconclusive: initial retries may be debug-only; do not restart the host or
enable server-wide debug logging to manufacture an error.
If reporting the problem, share only the failure stage and redacted error
reason (or say no command-channel entry is visible), not the full journal or
service configuration. Omit tokens, cookies, URLs, hostnames, addresses and
agent/token IDs. Keep saved identity and credentials intact while distinguishing
connection failures from admission failures; do not delete state, loosen TLS
verification or broaden permissions to force a connection. Guest-local Docker
monitoring with commands disabled remains an alternative to host-side discovery.
### Check Status
```bash
# Linux
@@ -794,6 +794,14 @@ installer download and the agent's subsequent Pulse TLS connection.
## Shared Boundaries
The shared monitor constructor also composes notification bootstrap: it loads
saved alert policy and destinations, binds the alert owner's immutable
quiet-hours policy provider, then activates persistent queue delivery.
This ordering is notifications/alerts-owned and must not change agent enrollment
or transport. `internal/monitoring/monitor_notification_startup_test.go` checks
the actual constructor and reconstruction, including holding due persisted
notifications under a saved continuous schedule without provider attempts.
- Local administrator setup synchronises the router authorizer before returning a browser session, so API Access remains available to manage agent credentials. This changes neither agent token scopes nor command-policy intent, and must not grant an unrelated identity access to credential management.
### Notification recovery reload ownership
@@ -1988,14 +1988,25 @@ quiet-hours policy; nonexistent clock minutes do not shift the configured
start or end. Initial queued replay uses the first real minute outside the
current non-full-day window, including an unselected day reached at midnight.
This does not reinterpret the day toggles as the previous evening's ownership.
Full-day windows retain the existing daily replay boundary; this clock repair
is not current-policy revalidation of already queued work or external receipt.
Full-day windows retain the existing daily revalidation boundary; it is not an
invented open minute or permission to send during a continuous schedule.
`TestQuietHoursDaylightSavingClockAndReplay` in
`internal/alerts/quiet_hours_test.go` checks London, New York and Lord Howe
(one-hour and half-hour changes), inclusive boundaries and exact UTC replay.
`TestQuietHoursOvernightSelectedCalendarDays` checks overnight day selection.
Quiet-hours location fallback is read-only; configuration updates own the
location cache, including when public suppression helpers run concurrently.
`Manager.QuietHoursNotificationPolicy` exposes an immutable read-only snapshot
to the notification owner before it takes queue or per-alert delivery locks.
Initial admission and queued replay share civil-clock/category semantics,
including repeated/skipped minutes and inclusive ends. The queue revalidates
every due attempt against current policy rather than trusting old replay
metadata. Policy snapshots neither mutate alert lifecycle/acknowledgement nor
select destinations; cancellation and actual delivery remain queue-owned.
`TestQuietHoursDeliverySnapshotIsCurrentIndependentAndReadOnly` and
`TestQuietHoursDeliverySnapshotConcurrentConfigUpdates` in
`internal/alerts/quiet_hours_test.go` cover snapshot isolation and concurrent
configuration reads. This establishes source scheduling, not external receipt.
Quiet-hours suppression also applies to alert delivery lifecycle, not only the
initial raised notification. Resolved notifications must not fan out when the
alert was never notified or was already acknowledged, and monitoring-driven
@@ -242,7 +242,8 @@ claim that #2077's missing graph journey is resolved.
Live telemetry and history retain the legacy REST graph/query contract and the
single successful-batch fast path. A batch rejected with HTTP 400, 422 or 500
falls back to one request per graph (at most six calls including the batch).
falls back to one request per selected graph. The native catalogue extension
below bounds the expanded selection and its split loop.
A failed optional graph must not discard successful CPU, memory or ARC graphs.
Authentication, rate-limit, missing-endpoint, service-unavailable, decoding,
transport and cancellation errors are not graph fallback triggers; these also
@@ -275,6 +276,75 @@ timestamps/units, appliance acceptance or the reporter's exact failure. Native
response evidence remains required before claiming that this repair resolves the
reported telemetry journey.
### CORE native reporting rows and device graphs — issue #2077
`internal/truenas/core_reporting.go` owns the native reporting envelope alongside
`client.go`, not a second transport or metrics store. The reporter's five incoming
CORE 13.0-U6.1 results now establish value-only rows with one element per legend,
external `start`/`end`/`step`, FreeBSD memory-class labels, disk-octet labels,
interface RX/TX/overlap and per-core `cputemp` readings. They do not establish that
Pulse's REST bridge, installed browser or complete appliance journey works.
Value-only rows derive time from `start + row-index * step`, with positive step,
valid ordered bounds, exact legend width and overflow-safe end checks. Explicit
row timestamps and supported object aliases retain compatibility. A metric is
never a timestamp, a null is never zero and a window mean is never a replacement
for a missing current row. History retains measured buckets; native current
readings accept only the last two steps before the requested end, preserving
sample time rather than stamping an old value as newly measured. This allowance
never exceeds the live query window, even with a coarse returned step. The five-state
CORE CPU vector is normalized by its sum (RRD state rates need not sum to 100).
All-zero CPU or complete memory-class vectors are empty buckets, not 100% usage;
explicit zero usage, all-idle CPU, zero-free RAM with used pages and zero I/O
remain measured zeros. Free memory comes from `memory-free_value`, not active
pages or the sum of memory classes. ARC subtraction uses the matching free-RAM
bucket when free RAM is present; independently reported ARC remains available
without claiming measured memory usage. Pre-cancelled collection stops before
reading the catalogue or posting a reporting query. Reporting I/O values remain rates in the existing bytes/s contract;
step supplies time, not a second rate division; `overlap` is not extra traffic.
Legacy REST reads `/reporting/graphs` once per live/History collection and sends
its disk/interface identifiers, omitting identifiers on global CPU, memory,
ARC and CPU-temperature graphs. Only missing/method-unsupported catalogue
endpoints retain the older unscoped compatibility request. All other catalogue
failures remain failures; no authentication, TLS, rate-limit, transport, decode
or cancellation retry is added. Selection is deduplicated and capped at 256
entries, refusing oversize rather than publishing a truncated total. Successful
reporting remains one batch; graph-only 400/422/500 failures may split once per
selected graph. Responses are bound to the selection: an omitted device stays
absent. Device rates sum once per identifier only at simultaneous observed
buckets of every selected member, never by carrying a stale member forward or
assuming a missing member was idle. Other graphs remain independently usable.
Native CPU temperature maps into the existing canonical host temperature and
`temperature` History key, using the same package-versus-core outlier safeguard
as current readings. Disk-temperature History uses the same envelope decoder.
The canonical writer also records each positive finite TrueNAS host temperature
in local and persistent `temperature` History, under the existing source/dedup
gates. A full local CPU window must not make the existing Thermals panel depend
on native fallback or lose its observed temperature. Missing temperature is not
written as zero, and no additional sensor writes are added to other sources.
No frontend route, resource identity, alert suppression policy, credential scope
or recognised-legacy transport boundary changes.
Verification: `internal/truenas/core_reporting_test.go` and its anonymised
`testdata/core13_reporting.json` cover five source-bound native excerpts,
row/timing/alias/zero/null/empty/finite controls, device totals, stale buckets,
ARC alignment, current and canonical History projections, request validation,
catalogue limits and failure boundaries. Unique overflow is rejected rather than
truncated, while duplicate identifiers consume only one selection entry. The
excerpt windows are deliberately
shortened controls, not complete native windows. Extended
`TestTrueNASPartialReportingPipeline` in
`internal/monitoring/truenas_poller_test.go` crosses pinned-certificate HTTP,
poller, canonical registry, writer, in-memory/persistent chart readbacks and
native fallback, including temperature and the sufficiently covered local
chart fast path. Existing CORE isolation, recognised
transport/auth/TLS and modern SCALE aggregation tests remain required. Source
proofs do not substitute for containing-release qualification or installed CORE
live/collapsed/expanded/History acceptance.
**Availability backfill preserves concurrent discovery changes (7 September 2026)**
@@ -1799,6 +1869,19 @@ service-history reads plus denial/recovery without fabricated samples.
## Current State
### Saved quiet-hours policy before queue activation
`Monitor.New` binds `alerts.Manager.QuietHoursNotificationPolicy` to the
notification manager after saved policy/destinations load and before
`StartQueueProcessing`. The notification owner revalidates due stored work and
owns all partitioning, delivery and cancellation; monitoring neither duplicates
the schedule nor changes destination selection or operational lifecycle.
`TestNewRevalidatesPersistedQuietHoursBeforeDelivery` in
`internal/monitoring/monitor_notification_startup_test.go` reconstructs the
actual monitor twice with due persisted work under a full-day saved schedule,
checking pending state and no local delivery/audit. This is source bootstrap
proof, not installed restart or provider acceptance.
TrueNAS REST alert arguments may be object, scalar, array, null or absent.
Non-object arguments must not abort snapshot collection or discard alerts.
Only typed object fields supply disk identity and SMART counters; neither scalar
@@ -73,6 +73,7 @@ or displayed a notification.
10. `internal/notifications/delivery_health.go`
11. `internal/notifications/deadman_config.go`
12. `internal/notifications/failure_class.go`
13. `internal/notifications/quiet_hours_queue.go`
## Shared Boundaries
@@ -141,6 +142,44 @@ stable opaque routing identities and must not expose credentials.
## Current State
### Current-policy quiet-hours replay
The persistent queue checks every due provider attempt against an immutable
snapshot from `alerts.Manager.QuietHoursNotificationPolicy`, installed by
`NotificationManager.SetQuietHoursPolicyProvider` before saved work is activated
in the monitor constructor. The snapshot is obtained before queue/database or
per-alert delivery locks; its evaluator does not call back into the alert owner.
Stored replay timestamps are wake conditions, not evidence that the current
schedule permits delivery. Full-day schedules recheck at the retained daily
boundary, and late replay into another quiet interval remains held.
`quiet_hours_queue.go` reloads the pending row after acquiring delivery gates,
preserving cancellations that rewrote a grouped row after worker discovery.
An all-held batch moves only its wake time. A mixed batch atomically retains the
held original and creates one ready child with the admitted destination bytes,
occurrence/transition links, creation time and remaining attempt budget. Child
IDs bind the parent and exact ready payload; held work does not gain an attempt,
audit entry, firing receipt or false success. Prior attempt audits remain on the
original batch, while actual child attempts carry their own linked receipts.
Already-due recovery groups cannot reopen their grouping window while held.
Partition persistence or ambiguous linkage failure leaves the original intact
and makes no provider attempt. Due-time refresh prevents stale worker snapshots
from repeatedly postponing or claiming a now-held row. Actual retries retain
the existing claim/cancellation gates and provider enabled checks.
This does not accelerate a future stored deadline after a schedule edit; the
current rule is applied at the next admitted wake. It does not change initial
target selection, mutable recovery tag admission, acknowledgement semantics,
operator retry budgets or at-least-once delivery after an interrupted send.
`quiet_hours_queue_test.go` proves continuous/late/DST replay holds, mixed
immediate/held delivery, exact admitted destination and operational linkage,
budget retention, transaction rollback, durable reopen, cancellation and stale
concurrent snapshots, grouped recoveries and all six provider/event families.
`internal/monitoring/monitor_notification_startup_test.go` exercises the actual
constructor with due persisted work across two startups. These are local source
and HTTP-receipt proofs, not installed provider or natural-day acceptance.
### Persistent resolved-alert grouping
`SendResolvedAlert` applies the configured grouping window to recoveries as
@@ -1623,6 +1623,7 @@
"internal/monitoring/docker_metric_presence_test.go",
"internal/monitoring/monitor_host_agent_removal_lifecycle_test.go",
"internal/monitoring/monitor_host_agents_test.go",
"internal/monitoring/monitor_notification_startup_test.go",
"internal/monitoring/physical_disk_roundtrip_test.go",
"scripts/installtests/agent_state_dir_lifecycle_test.go",
"scripts/installtests/install_ps1_test.go",
@@ -6156,10 +6157,12 @@
"match_prefixes": [],
"match_files": [
"internal/truenas/client.go",
"internal/truenas/core_reporting.go",
"internal/truenas/disk_health.go",
"internal/truenas/fixtures.go",
"internal/truenas/provider.go",
"internal/truenas/telemetry.go",
"internal/truenas/testdata/core13_reporting.json",
"internal/truenas/transport.go",
"internal/truenas/types.go"
],
@@ -6172,6 +6175,7 @@
"internal/truenas/client_api_shapes_test.go",
"internal/truenas/client_test.go",
"internal/truenas/contract_test.go",
"internal/truenas/core_reporting_test.go",
"internal/truenas/provider_pool_health_contract_test.go",
"internal/truenas/provider_test.go",
"internal/truenas/transport_test.go"
@@ -6566,6 +6570,7 @@
"internal/monitoring/monitor_host_agent_removal_lifecycle_test.go",
"internal/monitoring/monitor_host_agents_test.go",
"internal/monitoring/monitor_mock_alerts_test.go",
"internal/monitoring/monitor_notification_startup_test.go",
"internal/monitoring/monitor_pve_cluster_refresh_test.go",
"internal/monitoring/monitor_pve_guest_lxc_test.go",
"internal/monitoring/physical_disk_roundtrip_test.go",
+97 -23
View File
@@ -250,32 +250,72 @@ A healthy heartbeat indicates Pulse monitoring-loop progress, not successful
delivery of every resource alert or external reachability of your services.
Continue checking delivery activity for destination failures.
#### Test succeeds but real alerts are missing
A test sends directly to its destination: it skips the persistent delivery queue
and is not listed in **Recent delivery activity**. It does not prove that a real
alert was generated, routed or delivered to the intended recipient.
- Open **Alerts → Notifications**. If **Notifications are paused** is shown,
configured destinations and a successful test do not enable real delivery.
Turn delivery on there only when you intend to send alerts.
- Check the affected alert, the destination's **Enabled** state, minimum alert
severity and tag filters. Review quiet hours and any mute, acknowledgement or
maintenance policy before treating an absent attempt as a transport failure.
- Use **Recent delivery activity** to correlate the original alert, destination
and absolute timestamp, including held-notification reasons. An empty window
is not proof of healthy delivery; an **unavailable** read is not an empty log.
Do not create an outage or repeat a notification storm to populate it.
#### Recover retained delivery failures
Pulse shows a delivery warning for failed or dead-lettered notifications in its
persistent queue, not for every recoverable retry. **Recent delivery activity**
includes safely redacted provider errors; completed attempts remain for 7 days
and dead-letter attempts for 30 days. Start with the failure class and timestamp:
| Failure | Check before retrying |
| --- | --- |
| Authentication | Destination credentials and account permissions, locally; never post them. |
| Rate limited | Provider limits and delivery volume; repeated tests or retries can make this worse. |
| Connectivity | DNS, firewall, proxy and reachability from the Pulse server, not just your browser. |
| TLS | Certificate trust, expiry and hostname matching; do not disable verification to diagnose it. |
| Configuration / rejected | Enabled destination, required fields and the provider's endpoint or payload requirements. |
| Server error / unknown | Destination service status and a relevant, bounded local error excerpt. |
Save the corrected destination settings and send one test; check receipt at the
intended destination. **Retry retained deliveries** gives terminal failures a
fresh retry budget, but a destination that accepted an earlier attempt may
receive a duplicate. Review the confirmation's delivery count and provider
limits before retrying. A successful test does not itself retry retained items.
Use **Dismiss retained failures** only when those deliveries should not be sent.
Dismissal clears the warning without retrying them; delivery history remains.
Neither action deletes the audit trail. Do not delete `notification_queue.db`
or audit data to clear the warning.
#### Emails not sending
- Open **Alerts → Notifications** first. Pulse shows a delivery warning when
failed or dead-lettered notifications remain in the persistent queue; a
missing queue-health read is shown as unavailable rather than healthy.
- **Recent delivery activity** appears directly below that warning. It names
the destination and affected alert, shows an absolute timestamp for timeline
correlation, and includes safely redacted provider errors. Completed attempts
remain for 7 days and dead-letter attempts remain for 30 days.
- After correcting the destination, use **Retry retained deliveries**. Use
**Dismiss retained failures** only when those deliveries should not be sent.
Both actions preserve delivery history; do not delete `notification_queue.db`
to clear the warning.
- Check SMTP settings in **Alerts → Notifications**.
- Check logs: `docker logs pulse | grep email`.
- Ensure your SMTP provider allows the connection (e.g., Gmail App Passwords).
Follow [retained-failure recovery](#recover-retained-delivery-failures) first.
Check SMTP host, port, sender, recipients, authentication and TLS settings in
**Alerts → Notifications** against your provider's requirements (some providers
require an app password). Keep passwords in the settings form, not a diagnostic
command or report. If the delivery error is insufficient, inspect
[bounded notification logs](#inspect-notification-logs) locally.
#### Webhooks failing
- Check the delivery warning in **Alerts → Notifications** and use **Send test**
after correcting the destination. Recoverable retries do not trigger the
warning; retained terminal failures do.
- If the test succeeds, use **Retry retained deliveries** to give the retained
items a fresh retry budget. Dismiss them only when delivery is no longer
wanted; neither action deletes the audit trail.
- Verify the URL is reachable from the Pulse server.
- If targeting private IPs, allow them in **Settings → System → Network → Webhook Security**.
- Check Pulse logs for HTTP status codes and response bodies.
Follow [retained-failure recovery](#recover-retained-delivery-failures) first.
Use the failure class and HTTP status to check the provider's endpoint and
payload requirements. Verify reachability from the Pulse server. For an intended
private destination, review **Settings → System → Network → Webhook Security**;
do not broadly weaken network or TLS controls just to make a test pass.
Prefer the redacted delivery error over raw provider response bodies. A provider
can echo credentials or private content in its response; do not post it wholesale
or enable debug logging just to collect it. If needed, inspect
[bounded notification logs](#inspect-notification-logs) and share only the
consequential, manually redacted error.
### TrueNAS
@@ -309,6 +349,40 @@ Continue checking delivery activity for destination failures.
## 🛠️ Advanced Diagnostics
### Inspect Notification Logs
Prefer **Recent delivery activity** in **Alerts → Notifications**. If a local log
is needed, run only the command for your deployment, on the Pulse host with an
account authorised to read its logs. For Proxmox LXC, run the systemd command
inside the Pulse container, not on the Proxmox host. Adjust the time window to
the original incident and substitute your actual service or container name
(`pulse-backend` on some older systemd installs). These examples read at most
200 records from the last 15 minutes; they do not follow the log or send a test.
```bash
# systemd / Proxmox LXC
journalctl -u pulse --since '15 minutes ago' --lines 200 --no-pager
```
```bash
# Docker
docker logs --since 15m --tail 200 pulse
```
Docker can write application logs to either stdout or stderr; inspect both.
Do not pipe the reader into `grep email`: it can miss SMTP or webhook errors
and hide a failed read behind a matching partial line. A nonzero reader exit,
access error or missing service/container is a failed read, not “no delivery
errors”. Even a successful empty read is inconclusive: the window, retained
logs or selected instance may differ.
These local excerpts are **not sanitised**. Do not post them wholesale. Share
only the relevant timestamp, method, HTTP status or SMTP error code and a
manually redacted error. Remove credentials, cookies, secret URLs, addresses
and private host or personal information, including anything echoed by the
provider. Never upload full environments, configuration, a queue database or
audit data. See [Getting Help](#-getting-help).
### Correlate Logs with Requests
For a failed HTTP API request, inspect its response in your authenticated
@@ -1000,6 +1000,11 @@ LXC guests** in Settings → System → General and any configured
guests and guests already linked to an online guest-local agent. In the latter
case, check that guest agent's Docker monitoring instead.
If the owning node says **Remote control blocked**, first check its
[command channel](#commands-enabled-but-remote-control-blocked). A successful
root `pct exec` check does not establish that Pulse can reach that node's agent;
do not repeat the guest probe when those results are already known.
The host-side path requires the guest's `docker` executable and
`/var/run/docker.sock` in the root `pct exec` context. A working Docker CLI in a
user's login session does not prove that this context can reach the daemon;
@@ -1036,6 +1041,38 @@ Review output before posting and redact any private endpoint in an error. Do not
change LXC privilege, `keyctl`, socket permissions or Docker versions merely to
test a guess.
### Commands enabled but remote control blocked
**Remote control blocked** means the agent reports commands enabled, but Pulse
has no admitted command channel connected for it. Agent Doctor's
`commands command-capable · credential grants exec` line describes the reported
local command ceiling and credential scope, not a connected session. A fresh
monitoring report or **Automatic updates ready** also does not prove that this
separate WebSocket channel is working. Changing Proxmox API permissions cannot
repair a Pulse command-channel connection.
For a systemd agent, inspect the recent journal **locally on the affected node**:
```bash
sudo journalctl -u pulse-agent.service --since '15 minutes ago' -n 200 --no-pager --output=cat
```
Look for **Connected and registered with Pulse command server**, or
**WebSocket connection failed repeatedly, reconnecting** and its error. A
`dial websocket` error is a connection/handshake failure; `registration failed`
means registration was attempted but not accepted. Scope alone does not prove
that the credential's host/agent binding was admitted. No matching entry is
inconclusive: initial retries may be debug-only; do not restart the host or
enable server-wide debug logging to manufacture an error.
If reporting the problem, share only the failure stage and redacted error
reason (or say no command-channel entry is visible), not the full journal or
service configuration. Omit tokens, cookies, URLs, hostnames, addresses and
agent/token IDs. Keep saved identity and credentials intact while distinguishing
connection failures from admission failures; do not delete state, loosen TLS
verification or broaden permissions to force a connection. Guest-local Docker
monitoring with commands disabled remains an alternative to host-side discovery.
### Check Status
```bash
# Linux
+42 -9
View File
@@ -481,14 +481,18 @@ func quietHoursCategoryForAlert(alert *Alert) string {
}
func (m *Manager) shouldSuppressNotification(alert *Alert) (bool, string) {
if alert == nil {
return false, ""
}
// Only quiet-hours matches are replayable. Operator policy is checked by
// each caller before this helper, because it must drop rather than defer.
if !m.isInQuietHours() {
return false, ""
}
return quietHoursSuppressionForAlert(alert, m.config.Schedule.QuietHours.Suppress)
}
func quietHoursSuppressionForAlert(alert *Alert, suppress QuietHoursSuppression) (bool, string) {
if alert == nil {
return false, ""
}
if alert.Level != AlertLevelCritical {
return true, "non-critical"
@@ -497,15 +501,15 @@ func (m *Manager) shouldSuppressNotification(alert *Alert) (bool, string) {
category := quietHoursCategoryForAlert(alert)
switch category {
case "performance":
if m.config.Schedule.QuietHours.Suppress.Performance {
if suppress.Performance {
return true, category
}
case "storage":
if m.config.Schedule.QuietHours.Suppress.Storage {
if suppress.Storage {
return true, category
}
case "offline":
if m.config.Schedule.QuietHours.Suppress.Offline {
if suppress.Offline {
return true, category
}
}
@@ -516,13 +520,42 @@ func (m *Manager) shouldSuppressNotification(alert *Alert) (bool, string) {
func (m *Manager) quietHoursReplayAt() time.Time {
now := m.policyNow()
clock, valid := m.quietHoursClock()
if !valid || !clock.contains(now) {
if !valid {
return now.Add(time.Minute).UTC()
}
return clock.replayAt(now)
}
// QuietHoursNotificationPolicy returns an immutable snapshot of the current
// quiet-hours rule. The delivery owner obtains it before taking queue or
// per-alert delivery locks, so evaluation never calls back into this manager
// while cancellation is waiting for those locks. It does not mutate alert
// lifecycle, acknowledgement, metadata or destination selection.
func (m *Manager) QuietHoursNotificationPolicy() func(*Alert, time.Time) *time.Time {
m.mu.RLock()
clock, valid := m.quietHoursClock()
suppress := m.config.Schedule.QuietHours.Suppress
m.mu.RUnlock()
return func(alert *Alert, now time.Time) *time.Time {
if !valid || !clock.contains(now) {
return nil
}
if suppressed, _ := quietHoursSuppressionForAlert(alert, suppress); !suppressed {
return nil
}
replayAt := clock.replayAt(now)
return &replayAt
}
}
func (clock quietHoursClock) replayAt(now time.Time) time.Time {
if !clock.contains(now) {
return now.Add(time.Minute).UTC()
}
// A full-day window has no non-quiet clock minute. Preserve its existing
// daily replay boundary; queued-policy revalidation is a separate delivery
// concern, not permission to invent an end to a continuous schedule.
// daily revalidation boundary. The queue must check policy again there,
// not invent an end to a continuous schedule or attempt a provider send.
if (clock.end-clock.start+24*60)%(24*60) == 24*60-1 {
local := now.In(clock.location)
endExclusive := time.Date(local.Year(), local.Month(), local.Day(), clock.end/60, clock.end%60, 0, 0, clock.location).Add(time.Minute)
+69
View File
@@ -1,6 +1,7 @@
package alerts
import (
"reflect"
"sync"
"testing"
"time"
@@ -8,6 +9,74 @@ import (
alertspecs "github.com/rcourtman/pulse-go-rewrite/internal/alerts/specs"
)
func TestQuietHoursDeliverySnapshotIsCurrentIndependentAndReadOnly(t *testing.T) {
now := time.Date(2026, 10, 25, 1, 45, 0, 0, time.UTC)
quiet := QuietHours{Enabled: true, Start: "01:30", End: "02:00", Timezone: "Europe/London",
Days: map[string]bool{"sunday": true}, Suppress: QuietHoursSuppression{Storage: true}}
m := fixedQuietHoursTestManager(now, quiet)
defer m.Stop()
policy := m.QuietHoursNotificationPolicy()
for _, tc := range []struct {
kind string
level AlertLevel
held bool
}{
{"cpu", AlertLevelWarning, true},
{"cpu", AlertLevelCritical, false},
{"connectivity", AlertLevelCritical, false},
{"disk-health", AlertLevelCritical, true},
} {
alert := &Alert{ID: tc.kind, Type: tc.kind, Level: tc.level, Acknowledged: true,
Metadata: map[string]interface{}{MetadataQuietHoursReplayAt: "old-admission", "unchanged": "value"}}
before := alert.Clone()
at := policy(alert, now)
if (at != nil) != tc.held || (at != nil && !at.Equal(time.Date(2026, 10, 25, 2, 1, 0, 0, time.UTC))) {
t.Fatalf("%s level=%s held=%t replay=%v", tc.kind, tc.level, tc.held, at)
}
if !reflect.DeepEqual(before, alert) {
t.Fatal("delivery policy mutated acknowledgement or original occurrence metadata")
}
}
if at := policy(nil, now); at != nil {
t.Fatalf("nil alert acquired a hold: %v", at)
}
config := m.GetConfig()
config.Schedule.QuietHours = QuietHours{}
m.UpdateConfig(config)
alert := &Alert{Type: "cpu", Level: AlertLevelWarning}
if at := policy(alert, now); at == nil {
t.Fatal("immutable policy snapshot changed after configuration update")
}
if at := m.QuietHoursNotificationPolicy()(alert, now); at != nil {
t.Fatal("fresh delivery snapshot did not pick up disabled quiet hours")
}
}
func TestQuietHoursDeliverySnapshotConcurrentConfigUpdates(t *testing.T) {
now := time.Date(2026, 10, 2, 0, 0, 0, 0, time.UTC)
m := newManagerWithQuietHoursSuppress(QuietHoursSuppression{})
defer m.Stop()
config := m.GetConfig()
var wg sync.WaitGroup
for i := 0; i < 8; i++ {
wg.Add(1)
go func(i int) {
defer wg.Done()
for range 20 {
if i == 0 {
m.UpdateConfig(config)
} else {
at := m.QuietHoursNotificationPolicy()(&Alert{Type: "cpu", Level: AlertLevelWarning}, now)
if at == nil || !at.After(now) {
t.Errorf("continuous schedule acquired an invented open minute: %v", at)
}
}
}
}(i)
}
wg.Wait()
}
func fixedQuietHoursTestManager(now time.Time, quietHours QuietHours) *Manager {
m := NewManager()
m.now = func() time.Time { return now }
+13
View File
@@ -1931,6 +1931,7 @@ func New(cfg *config.Config) (*Monitor, error) {
// The queue worker starts with the manager but has no processor until all
// saved destinations are installed. Activate it only after the last load;
// a slow migration or config read must not cancel due persisted work.
m.notificationMgr.SetQuietHoursPolicyProvider(m.alertManager.QuietHoursNotificationPolicy)
m.notificationMgr.StartQueueProcessing()
// In mock mode the canonical sampler owns demo chart history by default.
@@ -5538,6 +5539,18 @@ func (m *Monitor) syncUnifiedAgentMetrics(store ResourceStoreInterface, sinks ..
metricKey := fmt.Sprintf("agent:%s", targetID)
observedAt := unifiedResourceObservedAt(resource, now)
// Native TrueNAS temperature must also survive the local History path.
// A sufficiently covered ring/store read need not query native fallback.
if monitorHasSource(resource.Sources, unifiedresources.SourceTrueNAS) && resource.Temperature != nil {
value := *resource.Temperature
if value > 0 && !math.IsNaN(value) && !math.IsInf(value, 0) {
if m.metricsHistory != nil {
m.metricsHistory.AddGuestMetric(metricKey, "temperature", value, observedAt)
}
appendStoreWrite("agent", targetID, "temperature", value, observedAt)
}
}
if metric := resource.Metrics.CPU; metric != nil {
value := metric.Percent
if value == 0 {
@@ -1,14 +1,106 @@
package monitoring
import (
"encoding/json"
"net/http"
"net/http/httptest"
"reflect"
"sync/atomic"
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
"github.com/rcourtman/pulse-go-rewrite/internal/config"
"github.com/rcourtman/pulse-go-rewrite/internal/notifications"
)
// A due persisted row must receive current saved quiet-hours policy BEFORE the
// autonomous queue is activated, on initial construction and reconstruction.
// This observes local HTTP acceptance only, not an installed destination.
func TestNewRevalidatesPersistedQuietHoursBeforeDelivery(t *testing.T) {
dir := t.TempDir()
t.Setenv("PULSE_DATA_DIR", dir)
var deliveries atomic.Int32
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
deliveries.Add(1)
w.WriteHeader(http.StatusOK)
}))
defer server.Close()
quiet := alerts.QuietHours{Enabled: true, Start: "00:00", End: "23:59", Timezone: "UTC",
Days: map[string]bool{"sunday": true, "monday": true, "tuesday": true,
"wednesday": true, "thursday": true, "friday": true, "saturday": true}}
persistence := config.NewConfigPersistence(dir)
saved := alerts.AlertConfig{Enabled: true, ActivationState: alerts.ActivationActive,
Schedule: alerts.ScheduleConfig{QuietHours: quiet}}
if err := persistence.SaveAlertConfig(saved); err != nil {
t.Fatal(err)
}
hook := notifications.WebhookConfig{ID: "saved-local-ops", Name: "saved-local-ops", Enabled: true, URL: server.URL, Service: "generic"}
if err := persistence.SaveWebhooks([]notifications.WebhookConfig{hook}); err != nil {
t.Fatal(err)
}
configJSON, err := json.Marshal(hook)
if err != nil {
t.Fatal(err)
}
for startup := 0; startup < 2; startup++ {
seed, err := notifications.NewNotificationQueue(dir)
if err != nil {
t.Fatal(err)
}
due := time.Now().Add(-time.Hour)
id := "saved-quiet-first"
if startup == 1 {
id = "saved-quiet-second"
}
if err := seed.Enqueue(&notifications.QueuedNotification{ID: id, Type: "webhook", Config: configJSON,
Alerts: []*alerts.Alert{{ID: id, Type: "cpu", Level: alerts.AlertLevelWarning, StartTime: due}},
NextRetryAt: &due}); err != nil {
t.Fatal(err)
}
if err := seed.Stop(); err != nil {
t.Fatal(err)
}
m, err := New(&config.Config{DataPath: dir})
if err != nil {
t.Fatal(err)
}
n := m.GetNotificationManager()
if err := n.UpdateAllowedPrivateCIDRs("127.0.0.1/32"); err != nil {
m.Stop()
t.Fatal(err)
}
q := n.GetQueue()
deadline := time.Now().Add(5 * time.Second)
for {
pending, err := q.GetPending(10)
if err != nil {
m.Stop()
t.Fatal(err)
}
if len(pending) == 0 {
break
}
if time.Now().After(deadline) {
m.Stop()
t.Fatal("persisted quiet notification was not revalidated")
}
time.Sleep(10 * time.Millisecond)
}
stats, err := q.GetQueueStats()
if err != nil || stats["pending"] != startup+1 || stats["sent"] != 0 || stats["dlq"] != 0 || deliveries.Load() != 0 {
m.Stop()
t.Fatalf("QUIET_BOOTSTRAP: saved current schedule did not hold every persisted row: stats=%v deliveries=%d err=%v", stats, deliveries.Load(), err)
}
logs, err := q.GetDeliveryLog(time.Now().Add(-time.Hour), 10)
if err != nil || len(logs) != 0 {
m.Stop()
t.Fatalf("quiet bootstrap manufactured provider attempts: %+v, %v", logs, err)
}
m.Stop()
}
}
// Cover the real monitor constructor, not a fixture which reapplies manager
// setters after restart. Saving here uses persistence directly: this is not
// browser/API save acceptance, a process restart, or destination receipt proof.
+85 -8
View File
@@ -2221,13 +2221,23 @@ func TestTrueNASPartialReportingPipeline(t *testing.T) {
status int
want map[string]float64
errorCategory string
native bool
}{
{`[{"name":"memory","legend":["free"],"data":[[1789000060,8]]}]`, 200, map[string]float64{"memory": 50}, ""},
{`[{"name":"cpu","legend":["usage"],"data":[[1789000060,0]]}]`, 200, map[string]float64{"cpu": 0}, ""},
{`provider-private-text`, 401, nil, "authentication"},
{`[{"name":"cpu","legend":["usage"],"data":[[1789000060,0]]},{"name":"memory","legend":["free"],"data":[[1789000060,0]]},{"name":"interface","legend":["received"],"data":[[1789000060,0]]},{"name":"disk","legend":["write"],"data":[[1789000060,0]]}]`, 200, map[string]float64{"cpu": 0, "memory": 100, "netin": 0, "diskwrite": 0}, ""},
{`[{"name":"memory","legend":["free"],"data":[[1789000060,8]]}]`, 200, map[string]float64{"memory": 50}, "", false},
{`[{"name":"cpu","legend":["usage"],"data":[[1789000060,0]]}]`, 200, map[string]float64{"cpu": 0}, "", false},
{`provider-private-text`, 401, nil, "authentication", false},
{`[{"name":"cpu","legend":["usage"],"data":[[1789000060,0]]},{"name":"memory","legend":["free"],"data":[[1789000060,0]]},{"name":"interface","legend":["received"],"data":[[1789000060,0]]},{"name":"disk","legend":["write"],"data":[[1789000060,0]]}]`, 200, map[string]float64{"cpu": 0, "memory": 100, "netin": 0, "diskwrite": 0}, "", false},
{`[
{"name":"cpu","legend":["interrupt","system","user","nice","idle"],"data":[[1,2,3,4,90]]},
{"name":"memory","legend":["memory-active_value","memory-inactive_value","memory-wired_value","memory-laundry_value","memory-free_value"],"data":[[1,1,10,0,4]]},
{"name":"arcsize","legend":["arcsize_value"],"data":[[2]]},
{"name":"cputemp","legend":["cputemp0","cputemp1"],"data":[[41,42]]},
{"name":"interface","identifier":"nic-a","legend":["rx","tx","overlap"],"data":[[8,16,8]]},
{"name":"disk","identifier":"disk-a","legend":["disk_octets_read","disk_octets_write"],"data":[[0,32]]}
]`, 200, map[string]float64{"cpu": 10, "memory": 62.5, "netin": 8, "netout": 16, "diskread": 0, "diskwrite": 32, "temperature": 42}, "", true},
}
var current atomic.Int64
var nativeEnd atomic.Int64
server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
switch r.URL.Path {
@@ -2237,10 +2247,48 @@ func TestTrueNASPartialReportingPipeline(t *testing.T) {
_, _ = w.Write([]byte(`[{"id":1,"name":"tank","status":"ONLINE","size":100,"allocated":50,"free":50}]`))
case "/api/v2.0/pool/dataset", "/api/v2.0/disk", "/api/v2.0/alert/list":
_, _ = w.Write([]byte(`[]`))
case "/api/v2.0/reporting/graphs":
if cycles[current.Load()].native {
_, _ = w.Write([]byte(`[{"name":"disk","identifiers":["disk-a"]},{"name":"interface","identifiers":["nic-a"]}]`))
} else {
http.NotFound(w, r)
}
case "/api/v2.0/reporting/get_data":
cycle := cycles[current.Load()]
w.WriteHeader(cycle.status)
_, _ = w.Write([]byte(cycle.body))
if !cycle.native {
_, _ = w.Write([]byte(cycle.body))
break
}
var query struct {
Query map[string]any `json:"query"`
Graphs []map[string]any `json:"graphs"`
}
if err := json.NewDecoder(r.Body).Decode(&query); err != nil {
t.Error(err)
return
}
var rows []map[string]any
if err := json.Unmarshal([]byte(cycle.body), &rows); err != nil {
t.Error(err)
return
}
end := int64(query.Query["end"].(float64))
nativeEnd.Store(end)
for _, row := range rows {
row["start"] = end
row["end"] = end
row["step"] = 10
}
for _, graph := range query.Graphs {
if name := graph["name"]; (name == "interface" || name == "disk") && graph["identifier"] == nil {
t.Error("native device requested without identifier")
return
}
}
if err := json.NewEncoder(w).Encode(rows); err != nil {
t.Error(err)
}
default:
http.NotFound(w, r)
}
@@ -2318,6 +2366,9 @@ func TestTrueNASPartialReportingPipeline(t *testing.T) {
if host.Agent.Memory.UsageUnavailable != !availability.Memory {
t.Fatalf("cycle %d stale/fabricated memory metadata: %+v", index, host.Agent.Memory)
}
if cycle.native && (host.Temperature == nil || *host.Temperature != 42) {
t.Fatalf("native collapsed/expanded temperature missing: %+v", host)
}
var writes []metrics.WriteMetric
monitor.syncUnifiedAgentMetrics(resourceStore, &writes)
seen := make(map[string]bool)
@@ -2338,7 +2389,7 @@ func TestTrueNASPartialReportingPipeline(t *testing.T) {
persistent.WriteBatchSync(writes)
inMemory := monitor.GetGuestMetrics("agent:"+instance.ID, time.Hour)
chart := monitor.GetGuestMetricsForChart("agent:"+instance.ID, "agent", instance.ID, time.Hour)
for _, key := range []string{"cpu", "memory", "disk", "netin", "netout", "diskread", "diskwrite"} {
for _, key := range []string{"cpu", "memory", "disk", "netin", "netout", "diskread", "diskwrite", "temperature"} {
if len(inMemory[key]) != counts[key] || len(chart[key]) != counts[key] {
t.Errorf("cycle %d %s chart contains fabricated/lost points: memory=%d chart=%d want=%d", index, key, len(inMemory[key]), len(chart[key]), counts[key])
}
@@ -2350,14 +2401,40 @@ func TestTrueNASPartialReportingPipeline(t *testing.T) {
}
continue
}
if err != nil || id != instance.ID || len(native) != len(cycle.want) {
wantNative := len(cycle.want)
if err != nil || id != instance.ID || len(native) != wantNative {
t.Fatalf("cycle %d native History=%+v id=%q err=%v", index, native, id, err)
}
for key, want := range cycle.want {
points := native[key]
if len(points) != 1 || points[0].Value != want || points[0].Timestamp.Unix() != 1789000060 {
if len(points) != 1 || points[0].Value != want || points[0].Timestamp.Unix() != func() int64 {
if cycle.native {
return nativeEnd.Load()
}
return 1789000060
}() {
t.Errorf("cycle %d %s native data/timestamp altered: %+v", index, key, points)
}
}
if cycle.native {
points := native["temperature"]
if len(points) != 1 || points[0].Value != 42 || points[0].Timestamp.Unix() != nativeEnd.Load() {
t.Fatalf("native temperature History missing: %+v", points)
}
shared := poller.GuestMetricHistory(nil, "default", "agent", time.Hour)[instance.ID]
if len(shared) != 7 || len(shared["temperature"]) != 1 {
t.Fatalf("shared History fallback dropped native panels: %+v", shared)
}
}
}
// Once local CPU History covers the window, the chart's fast path must
// still retain Thermals without relying on another appliance request.
now := time.Now()
for i := 0; i <= 60; i++ {
monitor.metricsHistory.AddGuestMetric("agent:"+instance.ID, "cpu", 10, now.Add(time.Duration(i-60)*time.Minute))
}
chart := monitor.GetGuestMetricsForChart("agent:"+instance.ID, "agent", instance.ID, time.Hour)
if points := chart["temperature"]; len(points) != 1 || points[0].Value != 42 {
t.Fatalf("sufficiently covered local History lost Thermals: %+v", points)
}
}
+14
View File
@@ -792,6 +792,20 @@ func (n *NotificationManager) StartQueueProcessing() {
}
}
// SetQuietHoursPolicyProvider binds persistent replay to the alert owner's
// current schedule. Install it before StartQueueProcessing during bootstrap.
// Each provider call must return a lock-independent, read-only evaluator.
func (n *NotificationManager) SetQuietHoursPolicyProvider(provider func() func(*alerts.Alert, time.Time) *time.Time) {
n.mu.RLock()
queue := n.queue
n.mu.RUnlock()
if queue != nil {
queue.mu.Lock()
queue.quietHoursPolicy = provider
queue.mu.Unlock()
}
}
// SetTenantIdentityResolver installs an org-backed resolver for the tenant
// identity stamped into webhook payloads. It overrides the
// PULSE_TENANT_ID / PULSE_TENANT_NAME environment defaults; multi-tenant
+24 -3
View File
@@ -187,8 +187,10 @@ type NotificationQueue struct {
cleanupTicker *time.Ticker
notifyChan chan struct{} // Signal when new notifications are added
processor func(*QueuedNotification) error // Notification processor function
deliveryHealthChanged func() // Reconcile the monitoring-owned delivery alert after health-changing transitions
workerSem chan struct{} // Semaphore for limiting concurrent workers
quietHoursPolicy func() func(*alerts.Alert, time.Time) *time.Time
now func() time.Time // Queue scheduling clock; nil uses wall time.
deliveryHealthChanged func() // Reconcile the monitoring-owned delivery alert after health-changing transitions
workerSem chan struct{} // Semaphore for limiting concurrent workers
}
// notificationDeliveryGate orders delivery and cancellation for a single
@@ -1156,7 +1158,7 @@ func (nq *NotificationQueue) GetPending(limit int) ([]*QueuedNotification, error
}
// scanNotification scans a database row into a QueuedNotification
func (nq *NotificationQueue) scanNotification(rows *sql.Rows) (*QueuedNotification, error) {
func (nq *NotificationQueue) scanNotification(rows interface{ Scan(...any) error }) (*QueuedNotification, error) {
var notif QueuedNotification
var alertsJSON, configJSON, linksJSON string
var lastAttempt, nextRetryAt, completedAt *int64
@@ -1757,6 +1759,16 @@ func (nq *NotificationQueue) processNotification(notif *QueuedNotification) {
Msg("Skipping cancelled notification")
return
}
// Obtain policy before the delivery gates: the policy owner may have
// lifecycle callbacks that cancel queued work. The returned evaluator is
// immutable and must not acquire any owner locks.
nq.mu.RLock()
policyProvider := nq.quietHoursPolicy
nq.mu.RUnlock()
var quietHoursPolicy func(*alerts.Alert, time.Time) *time.Time
if policyProvider != nil {
quietHoursPolicy = policyProvider()
}
releaseDeliveryGates := nq.acquireAlertDeliveryGates(alertIdentifiersFromAlerts(notif.Alerts), false)
healthChanged := false
defer func() {
@@ -1765,6 +1777,15 @@ func (nq *NotificationQueue) processNotification(notif *QueuedNotification) {
nq.notifyDeliveryHealthChanged()
}
}()
if quietHoursPolicy != nil {
ready, err := nq.prepareQuietHoursDelivery(notif, quietHoursPolicy)
if err != nil {
log.Error().Err(err).Str("id", notif.ID).Msg("Failed to revalidate queued quiet-hours delivery")
}
if err != nil || !ready {
return
}
}
// Atomically claim the pending row. A concurrent resolution may have
// cancelled it while it was waiting for its per-alert delivery gate.
+180
View File
@@ -0,0 +1,180 @@
package notifications
import (
"crypto/sha256"
"database/sql"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"strings"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
"github.com/rcourtman/pulse-go-rewrite/internal/operationaltrust"
)
// prepareQuietHoursDelivery runs under the original batch's delivery gates,
// before claiming a provider attempt. Reloading the row preserves cancellation
// rewrites made after GetPending. Held work keeps the original row; a mixed
// batch gets one ready child, committed atomically with the held remainder.
// A hold is scheduling, not an attempt, failure, receipt or lifecycle change.
func (nq *NotificationQueue) prepareQuietHoursDelivery(
notif *QueuedNotification,
policy func(*alerts.Alert, time.Time) *time.Time,
) (bool, error) {
nq.mu.Lock()
defer nq.mu.Unlock()
now := time.Now()
if nq.now != nil {
now = nq.now()
}
current, err := nq.scanNotification(nq.db.QueryRow(`
SELECT id, type, method, status, alerts, config, attempts, max_attempts,
last_attempt, last_error, created_at, next_retry_at, completed_at,
payload_bytes, operational_links
FROM notification_queue WHERE id = ? AND status = 'pending'`, notif.ID))
if errors.Is(err, sql.ErrNoRows) {
return false, nil
}
if err != nil {
return false, fmt.Errorf("reload quiet-hours notification: %w", err)
}
// Another worker may already have postponed this row since GetPending.
if current.NextRetryAt != nil && current.NextRetryAt.After(now) {
return false, nil
}
var ready, held []*alerts.Alert
var replayAt *time.Time
for _, alert := range current.Alerts {
at := policy(alert, now)
if at == nil || !at.After(now) {
ready = append(ready, alert)
continue
}
held = append(held, alert)
if replayAt == nil || at.Before(*replayAt) {
value := at.UTC()
replayAt = &value
}
}
if len(held) == 0 {
*notif = *current
return true, nil
}
// The queue stores second precision. Never round a policy deadline down
// into a provider send (or a busy retry loop) before it becomes eligible.
deadline := replayAt.Truncate(time.Second)
if deadline.Before(*replayAt) {
deadline = deadline.Add(time.Second)
}
method := current.Method
if strings.HasPrefix(method, "resolved-group") {
// This batch has reached its admitted grouping deadline. It must not
// reopen that window and absorb later recoveries while quiet-held.
method = "quiet-hours-replay"
}
if len(ready) == 0 {
_, err := nq.db.Exec(`UPDATE notification_queue
SET next_retry_at = ?, method = ? WHERE id = ? AND status = 'pending'`, deadline.Unix(), method, current.ID)
if err != nil {
return false, fmt.Errorf("postpone quiet-hours notification: %w", err)
}
current.NextRetryAt = &deadline
current.Method = method
*notif = *current
return false, nil
}
readyJSON, err := json.Marshal(ready)
if err != nil {
return false, fmt.Errorf("encode ready quiet-hours alerts: %w", err)
}
heldJSON, err := json.Marshal(held)
if err != nil {
return false, fmt.Errorf("encode held quiet-hours alerts: %w", err)
}
// Include the parent identity and exact ready payload. Restarts and duplicate
// worker snapshots cannot generate a second child for the same partition.
sum := sha256.Sum256(append([]byte(current.ID+"\x00"), readyJSON...))
childID := current.ID + ":quiet:" + hex.EncodeToString(sum[:8])
readyLinks, heldLinks, err := partitionQuietHoursLinks(current.Links, ready, held, childID)
if err != nil {
return false, err
}
readyLinksJSON, err := json.Marshal(readyLinks)
if err != nil {
return false, fmt.Errorf("encode ready quiet-hours links: %w", err)
}
heldLinksJSON, err := json.Marshal(heldLinks)
if err != nil {
return false, fmt.Errorf("encode held quiet-hours links: %w", err)
}
tx, err := nq.db.Begin()
if err != nil {
return false, fmt.Errorf("begin quiet-hours partition: %w", err)
}
defer tx.Rollback()
// Copy the admitted destination bytes and attempt budget, not the current
// destination configuration. Previous attempts remain in the original audit
// trail; splitting never resets the retry budget or fabricates a new audit.
_, err = tx.Exec(`INSERT INTO notification_queue
(id, type, method, status, alerts, operational_links, config, attempts,
max_attempts, last_attempt, last_error, created_at, next_retry_at,
completed_at, payload_bytes)
SELECT ?, type, method, status, ?, ?, config, attempts, max_attempts,
last_attempt, last_error, created_at, NULL, completed_at, payload_bytes
FROM notification_queue WHERE id = ? AND status = 'pending'`, childID, string(readyJSON), string(readyLinksJSON), current.ID)
if err != nil {
return false, fmt.Errorf("persist ready quiet-hours partition: %w", err)
}
_, err = tx.Exec(`UPDATE notification_queue
SET alerts = ?, operational_links = ?, next_retry_at = ?, method = ?
WHERE id = ? AND status = 'pending'`, string(heldJSON), string(heldLinksJSON), deadline.Unix(), method, current.ID)
if err != nil {
return false, fmt.Errorf("persist held quiet-hours partition: %w", err)
}
if err := tx.Commit(); err != nil {
return false, fmt.Errorf("commit quiet-hours partition: %w", err)
}
current.ID = childID
current.Alerts = ready
current.Links = readyLinks
current.NextRetryAt = nil
*notif = *current
return true, nil
}
func partitionQuietHoursLinks(
links []operationaltrust.NotificationLink,
ready, held []*alerts.Alert,
childID string,
) ([]operationaltrust.NotificationLink, []operationaltrust.NotificationLink, error) {
keys := func(batch []*alerts.Alert) map[string]bool {
result := make(map[string]bool, len(batch))
for _, alert := range batch {
if alert != nil && alert.OperationalRecord != nil && alert.LatestTransition != nil {
result[alert.OperationalRecord.ID+"\x00"+alert.LatestTransition.ID] = true
}
}
return result
}
readyKeys, heldKeys := keys(ready), keys(held)
var readyLinks, heldLinks []operationaltrust.NotificationLink
for _, link := range links {
key := link.OperationalRecordID + "\x00" + link.TransitionID
switch {
case readyKeys[key] && !heldKeys[key]:
link = link.Clone()
link.NotificationID = childID
readyLinks = append(readyLinks, link)
case heldKeys[key] && !readyKeys[key]:
heldLinks = append(heldLinks, link.Clone())
default:
// Do not silently discard or misattribute a historical linkage.
return nil, nil, fmt.Errorf("quiet-hours partition has an ambiguous operational link")
}
}
return readyLinks, heldLinks, nil
}
@@ -0,0 +1,512 @@
package notifications
import (
"database/sql"
"encoding/json"
"net/http"
"net/http/httptest"
"path/filepath"
"reflect"
"strings"
"sync"
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
"github.com/rcourtman/pulse-go-rewrite/internal/operationaltrust"
)
// Use the real persistent schema and delivery processor, without a background
// ticker racing these controlled replay instants. The monitor constructor test
// covers installing this policy on the autonomous worker after saved-config load.
func newQuietReplayQueue(t *testing.T, dir string, now *time.Time) *NotificationQueue {
t.Helper()
db, err := sql.Open("sqlite", queueSQLiteDSN(filepath.Join(dir, "replay.db")))
if err != nil {
t.Fatal(err)
}
db.SetMaxOpenConns(1)
q := &NotificationQueue{db: db, deliveryGates: make(map[string]*notificationDeliveryGate),
notifyChan: make(chan struct{}, 100), now: func() time.Time { return *now }}
if err := q.initSchema(); err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = db.Close() })
return q
}
func quietReplaySchedule(start, end, zone string) alerts.QuietHours {
return alerts.QuietHours{Enabled: true, Start: start, End: end, Timezone: zone,
Days: map[string]bool{"sunday": true, "monday": true, "tuesday": true,
"wednesday": true, "thursday": true, "friday": true, "saturday": true}}
}
func quietReplayAlert(id string, level alerts.AlertLevel, start time.Time) *alerts.Alert {
return &alerts.Alert{ID: id, Type: "cpu", Level: level, StartTime: start,
ResourceID: id, ResourceName: id,
OperationalRecord: &operationaltrust.OperationalRecord{ID: "record-" + id},
LatestTransition: &operationaltrust.LifecycleTransition{ID: "transition-" + id,
To: operationaltrust.OperationalOpen, CauseKey: "cause-" + id}}
}
type quietReplayReceipt struct {
Event string `json:"event"`
Alerts []*alerts.Alert `json:"alerts"`
}
type quietReplaySink struct {
mu sync.Mutex
receipts []quietReplayReceipt
}
func (s *quietReplaySink) all() []quietReplayReceipt {
s.mu.Lock()
defer s.mu.Unlock()
return append([]quietReplayReceipt(nil), s.receipts...)
}
func newQuietReplayNotifier(t *testing.T, q *NotificationQueue, policyOwner *alerts.Manager) (*NotificationManager, WebhookConfig, *quietReplaySink) {
t.Helper()
sink := &quietReplaySink{}
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
defer r.Body.Close()
var receipt quietReplayReceipt
if err := json.NewDecoder(r.Body).Decode(&receipt); err != nil {
t.Errorf("decode local delivery: %v", err)
w.WriteHeader(http.StatusBadRequest)
return
}
sink.mu.Lock()
sink.receipts = append(sink.receipts, receipt)
sink.mu.Unlock()
w.WriteHeader(http.StatusOK)
}))
t.Cleanup(server.Close)
hook := WebhookConfig{ID: "admitted-ops", Enabled: true, Service: "generic", URL: server.URL}
n := &NotificationManager{enabled: true, queue: q, webhooks: []WebhookConfig{hook},
webhookClient: server.Client(), webhookRateLimits: make(map[string]*webhookRateLimit),
lastNotified: make(map[string]notificationRecord), deliveryReceipts: make(map[string]struct{})}
if err := n.UpdateAllowedPrivateCIDRs("127.0.0.1/32"); err != nil {
t.Fatal(err)
}
n.SetQuietHoursPolicyProvider(policyOwner.QuietHoursNotificationPolicy)
q.processor = n.ProcessQueuedNotification
return n, hook, sink
}
func loadQuietReplay(t *testing.T, q *NotificationQueue, id string) *QueuedNotification {
t.Helper()
n, err := q.scanNotification(q.db.QueryRow(`SELECT id, type, method, status, alerts, config,
attempts, max_attempts, last_attempt, last_error, created_at, next_retry_at,
completed_at, payload_bytes, operational_links FROM notification_queue WHERE id = ?`, id))
if err != nil {
t.Fatal(err)
}
return n
}
func quietReplayRows(t *testing.T, q *NotificationQueue) []*QueuedNotification {
t.Helper()
rows, err := q.db.Query(`SELECT id, type, method, status, alerts, config,
attempts, max_attempts, last_attempt, last_error, created_at, next_retry_at,
completed_at, payload_bytes, operational_links FROM notification_queue ORDER BY id`)
if err != nil {
t.Fatal(err)
}
defer rows.Close()
var result []*QueuedNotification
for rows.Next() {
n, err := q.scanNotification(rows)
if err != nil {
t.Fatal(err)
}
result = append(result, n)
}
if err := rows.Err(); err != nil {
t.Fatal(err)
}
return result
}
func seedQuietReplay(t *testing.T, q *NotificationQueue, hook WebhookConfig, now time.Time, batch ...*alerts.Alert) *QueuedNotification {
t.Helper()
config, err := json.Marshal(hook)
if err != nil {
t.Fatal(err)
}
due := now.Add(-time.Hour)
n := &QueuedNotification{ID: "admitted-batch", Type: "webhook", DestinationID: "webhook:" + hook.ID,
Alerts: batch, Config: config, MaxAttempts: 3, CreatedAt: now.Add(-24 * time.Hour), NextRetryAt: &due}
if err := q.Enqueue(n); err != nil {
t.Fatal(err)
}
return n
}
func setQuietReplaySchedule(m *alerts.Manager, quiet alerts.QuietHours) {
config := m.GetConfig()
config.Schedule.QuietHours = quiet
m.UpdateConfig(config)
}
func TestQueuedQuietHoursRevalidatesCurrentSchedule(t *testing.T) {
for _, tc := range []struct {
name, now, start, end, zone string
}{
{"continuous_midnight", "2026-10-02T00:00:00Z", "00:00", "23:59", "UTC"},
{"continuous_offset_boundary", "2026-10-02T12:00:00Z", "12:00", "11:59", "UTC"},
{"late_replay_next_night", "2026-10-02T22:30:00Z", "22:00", "06:00", "UTC"},
{"edited_longer_window", "2026-10-02T07:00:00Z", "22:00", "08:00", "UTC"},
{"repeated_end_minute", "2026-10-25T01:45:00Z", "01:30", "02:00", "Europe/London"},
} {
t.Run(tc.name, func(t *testing.T) {
now, err := time.Parse(time.RFC3339, tc.now)
if err != nil {
t.Fatal(err)
}
owner := alerts.NewManagerWithDataDir(t.TempDir())
defer owner.Stop()
setQuietReplaySchedule(owner, quietReplaySchedule(tc.start, tc.end, tc.zone))
q := newQuietReplayQueue(t, t.TempDir(), &now)
_, hook, sink := newQuietReplayNotifier(t, q, owner)
alert := quietReplayAlert("warning", alerts.AlertLevelWarning, now.Add(-2*time.Hour))
alert.Metadata = map[string]interface{}{alerts.MetadataQuietHoursReplayAt: now.Add(-time.Hour).Format(time.RFC3339)}
n := seedQuietReplay(t, q, hook, now, alert)
before := loadQuietReplay(t, q, n.ID)
stale := *before
q.processNotification(n)
after := loadQuietReplay(t, q, before.ID)
if got := sink.all(); len(got) != 0 {
t.Fatalf("QUIET_REPLAY_LEAK: held notification reached provider: %+v", got)
}
if after.Status != QueueStatusPending || after.Attempts != 0 || after.LastAttempt != nil ||
after.NextRetryAt == nil || !after.NextRetryAt.After(now) {
t.Fatalf("hold counted as delivery or did not move replay: %+v", after)
}
if !reflect.DeepEqual(before.Alerts, after.Alerts) || string(before.Config) != string(after.Config) ||
!reflect.DeepEqual(before.Links, after.Links) {
t.Fatal("quiet hold changed occurrence, destination or operational identity")
}
for range 3 {
copy := stale
q.processNotification(&copy)
}
if repeated := loadQuietReplay(t, q, before.ID); !reflect.DeepEqual(after, repeated) {
t.Fatal("stale worker snapshot consumed an attempt or moved a held row again")
}
logs, err := q.GetDeliveryLog(before.CreatedAt, 10)
if err != nil || len(logs) != 0 {
t.Fatalf("quiet hold manufactured delivery audit: %+v, %v", logs, err)
}
// A continuous schedule never gets an invented open minute. Once the
// operator disables it, the next admitted wake reuses the same row.
setQuietReplaySchedule(owner, alerts.QuietHours{})
now = after.NextRetryAt.Add(time.Second)
q.processNotification(&stale)
if got := sink.all(); len(got) != 1 || len(got[0].Alerts) != 1 || got[0].Alerts[0].ID != alert.ID ||
!got[0].Alerts[0].StartTime.Equal(alert.StartTime) {
t.Fatalf("eligible replay did not preserve occurrence: %+v", got)
}
if sent := loadQuietReplay(t, q, before.ID); sent.Status != QueueStatusSent || sent.Attempts != 1 {
t.Fatalf("eligible replay attempt: %+v", sent)
}
})
}
}
func TestQueuedQuietHoursMixedBatchPreservesDeliveryAndBudget(t *testing.T) {
now := time.Date(2026, 10, 2, 1, 0, 0, 0, time.UTC)
owner := alerts.NewManagerWithDataDir(t.TempDir())
defer owner.Stop()
setQuietReplaySchedule(owner, quietReplaySchedule("22:00", "06:00", "UTC"))
q := newQuietReplayQueue(t, t.TempDir(), &now)
notifier, hook, sink := newQuietReplayNotifier(t, q, owner)
warning := quietReplayAlert("warning", alerts.AlertLevelWarning, now.Add(-time.Hour))
critical := quietReplayAlert("critical", alerts.AlertLevelCritical, now.Add(-time.Minute))
n := seedQuietReplay(t, q, hook, now, warning, critical)
for range 2 {
if err := q.IncrementAttemptAndSetStatus(n.ID, QueueStatusSending); err != nil {
t.Fatal(err)
}
}
if err := q.ScheduleRetry(n.ID, 2); err != nil {
t.Fatal(err)
}
if _, err := q.db.Exec(`UPDATE notification_queue SET next_retry_at = ?, last_error = 'destination unavailable' WHERE id = ?`, now.Add(-time.Second).Unix(), n.ID); err != nil {
t.Fatal(err)
}
before := loadQuietReplay(t, q, n.ID)
if err := q.RecordAudit(before, false, "destination unavailable"); err != nil {
t.Fatal(err)
}
originalID := n.ID
q.processNotification(n)
got := sink.all()
if len(got) != 1 || len(got[0].Alerts) != 1 || got[0].Alerts[0].ID != critical.ID {
t.Fatalf("QUIET_MIXED_BATCH: immediately eligible subset was delayed or held member leaked: %+v", got)
}
rows := quietReplayRows(t, q)
if len(rows) != 2 {
t.Fatalf("mixed batch rows=%d, want held original plus ready child", len(rows))
}
held := loadQuietReplay(t, q, originalID)
if len(held.Alerts) != 1 || held.Alerts[0].ID != warning.ID || held.Status != QueueStatusPending ||
held.Attempts != 2 || held.MaxAttempts != 3 || !reflect.DeepEqual(held.LastAttempt, before.LastAttempt) ||
!reflect.DeepEqual(held.LastError, before.LastError) {
t.Fatalf("held partition lost its admission/attempt history: %+v", held)
}
if held.NextRetryAt == nil || !held.NextRetryAt.Equal(time.Date(2026, 10, 2, 6, 1, 0, 0, time.UTC)) {
t.Fatalf("held replay deadline=%v, want inclusive end at 06:01", held.NextRetryAt)
}
child := loadQuietReplay(t, q, n.ID)
if child.ID == originalID || child.Attempts != 3 || child.Status != QueueStatusSent || child.MaxAttempts != 3 {
t.Fatalf("ready partition reset budget or did not complete: %+v", child)
}
for _, row := range []*QueuedNotification{held, child} {
if string(row.Config) != string(before.Config) || row.DestinationID != before.DestinationID || !row.CreatedAt.Equal(before.CreatedAt) || len(row.Links) != 1 {
t.Fatalf("partition lost admitted destination, time or links: %+v", row)
}
link := row.Links[0]
if err := link.Validate(); err != nil || link.NotificationID != row.ID ||
link.OperationalRecordID != row.Alerts[0].OperationalRecord.ID ||
link.TransitionID != row.Alerts[0].LatestTransition.ID || link.DestinationID != before.DestinationID {
t.Fatalf("partition misattributed operational receipt: %+v, %v", link, err)
}
}
if held.Links[0].DeliveryState != operationaltrust.NotificationRetrying || child.Links[0].DeliveryState != operationaltrust.NotificationDelivered {
t.Fatal("held link gained a delivery or lost its existing retry evidence")
}
jobs := buildNotificationDeliveryJobs(EmailConfig{}, []WebhookConfig{hook}, AppriseConfig{}, []*alerts.Alert{warning, critical}, eventResolved, now)
filtered := notifier.filterResolvedJobsByDeliveryReceipt(jobs)
if len(filtered) != 1 || len(filtered[0].Alerts) != 1 || filtered[0].Alerts[0].ID != critical.ID {
t.Fatalf("held member gained a firing receipt: %+v", filtered)
}
// Release at its real schedule boundary: only the held occurrence gets its
// third provider attempt, with no repetition of the already sent child.
now = *held.NextRetryAt
q.processNotification(held)
if got := sink.all(); len(got) != 2 || got[1].Alerts[0].ID != warning.ID || len(got[1].Alerts) != 1 {
t.Fatalf("released partition delivery: %+v", got)
}
if done := loadQuietReplay(t, q, originalID); done.Attempts != 3 || done.Status != QueueStatusSent {
t.Fatalf("held retry budget after release: %+v", done)
}
}
func TestQueuedQuietHoursPartitionRollbackAndRestart(t *testing.T) {
for _, failure := range []string{"ready_insert", "held_rewrite"} {
t.Run(failure, func(t *testing.T) {
now := time.Date(2026, 10, 2, 1, 0, 0, 0, time.UTC)
owner := alerts.NewManagerWithDataDir(t.TempDir())
defer owner.Stop()
setQuietReplaySchedule(owner, quietReplaySchedule("22:00", "06:00", "UTC"))
dir := t.TempDir()
q := newQuietReplayQueue(t, dir, &now)
_, hook, sink := newQuietReplayNotifier(t, q, owner)
n := seedQuietReplay(t, q, hook, now,
quietReplayAlert("warning", alerts.AlertLevelWarning, now.Add(-time.Hour)),
quietReplayAlert("critical", alerts.AlertLevelCritical, now.Add(-time.Minute)))
before := loadQuietReplay(t, q, n.ID)
trigger := `CREATE TRIGGER reject_partition BEFORE INSERT ON notification_queue WHEN NEW.id != 'admitted-batch' BEGIN SELECT RAISE(ABORT, 'fixture partition persistence failed'); END`
if failure == "held_rewrite" {
trigger = `CREATE TRIGGER reject_partition BEFORE UPDATE OF alerts ON notification_queue BEGIN SELECT RAISE(ABORT, 'fixture partition persistence failed'); END`
}
if _, err := q.db.Exec(trigger); err != nil {
t.Fatal(err)
}
q.processNotification(n)
if got := sink.all(); len(got) != 0 {
t.Fatalf("QUIET_PARTITION_FAILURE: persistence failure still attempted delivery: %+v", got)
}
if rows := quietReplayRows(t, q); len(rows) != 1 || !reflect.DeepEqual(rows[0], before) {
t.Fatalf("failed transaction did not retain original complete batch: %+v", rows)
}
if _, err := q.db.Exec("DROP TRIGGER reject_partition"); err != nil {
t.Fatal(err)
}
q.processNotification(n)
if got := sink.all(); len(got) != 1 || len(got[0].Alerts) != 1 || got[0].Alerts[0].ID != "critical" {
t.Fatalf("successful partition: %+v", got)
}
if err := q.db.Close(); err != nil {
t.Fatal(err)
}
q = newQuietReplayQueue(t, dir, &now)
_, _, restartedSink := newQuietReplayNotifier(t, q, owner)
rows := quietReplayRows(t, q)
if len(rows) != 2 {
t.Fatalf("reopen lost partition: %+v", rows)
}
held := loadQuietReplay(t, q, before.ID)
if count, err := q.CancelByAlertIdentifiers([]string{"warning"}); err != nil || count != 1 {
t.Fatalf("cancellation of held member: %d, %v", count, err)
}
now = held.NextRetryAt.Add(time.Second)
q.processNotification(held)
if got := restartedSink.all(); len(got) != 0 {
t.Fatalf("cancelled held member resurrected after reopen: %+v", got)
}
if cancelled := loadQuietReplay(t, q, before.ID); cancelled.Status != QueueStatusCancelled || cancelled.Attempts != 0 {
t.Fatalf("cancelled member acquired attempt: %+v", cancelled)
}
})
}
}
func TestQueuedQuietHoursConcurrentSnapshotsAndCancellation(t *testing.T) {
now := time.Date(2026, 10, 2, 1, 0, 0, 0, time.UTC)
owner := alerts.NewManagerWithDataDir(t.TempDir())
defer owner.Stop()
setQuietReplaySchedule(owner, quietReplaySchedule("22:00", "06:00", "UTC"))
q := newQuietReplayQueue(t, t.TempDir(), &now)
_, hook, sink := newQuietReplayNotifier(t, q, owner)
n := seedQuietReplay(t, q, hook, now,
quietReplayAlert("cancelled", alerts.AlertLevelWarning, now.Add(-time.Hour)),
quietReplayAlert("held", alerts.AlertLevelWarning, now.Add(-time.Hour)),
quietReplayAlert("ready", alerts.AlertLevelCritical, now.Add(-time.Minute)))
snapshotTaken := make(chan struct{})
continueDelivery := make(chan struct{})
q.quietHoursPolicy = func() func(*alerts.Alert, time.Time) *time.Time {
policy := owner.QuietHoursNotificationPolicy()
close(snapshotTaken)
<-continueDelivery
return policy
}
done := make(chan struct{})
go func() { q.processNotification(n); close(done) }()
<-snapshotTaken
if count, err := q.CancelByAlertIdentifiers([]string{"cancelled"}); err != nil || count != 1 {
t.Fatalf("cancellation between snapshot and claim: %d, %v", count, err)
}
close(continueDelivery)
<-done
q.quietHoursPolicy = owner.QuietHoursNotificationPolicy
held := loadQuietReplay(t, q, "admitted-batch")
var wg sync.WaitGroup
for range 12 {
wg.Add(1)
go func() {
defer wg.Done()
copy := *held
q.processNotification(&copy)
}()
}
wg.Wait()
if got := sink.all(); len(got) != 1 || len(got[0].Alerts) != 1 || got[0].Alerts[0].ID != "ready" {
t.Fatalf("QUIET_CANCELLATION: stale snapshots lost cancellation or duplicated delivery: %+v", got)
}
if len(held.Alerts) != 1 || held.Alerts[0].ID != "held" || held.Attempts != 0 || len(held.Links) != 1 || held.Links[0].TransitionID != "transition-held" {
t.Fatalf("held remainder resurrected removed member or consumed budget: %+v", held)
}
if count, err := q.CancelByAlertIdentifiers([]string{"held"}); err != nil || count != 1 {
t.Fatalf("post-partition cancellation: %d, %v", count, err)
}
now = held.NextRetryAt.Add(time.Second)
q.processNotification(held)
if got := sink.all(); len(got) != 1 {
t.Fatalf("post-partition cancellation leaked delivery: %+v", got)
}
}
func TestQueuedQuietHoursResolvedKeepsUndeliveredReceipt(t *testing.T) {
now := time.Date(2026, 10, 2, 1, 0, 0, 0, time.UTC)
owner := alerts.NewManagerWithDataDir(t.TempDir())
defer owner.Stop()
setQuietReplaySchedule(owner, quietReplaySchedule("22:00", "06:00", "UTC"))
q := newQuietReplayQueue(t, t.TempDir(), &now)
notifier, hook, sink := newQuietReplayNotifier(t, q, owner)
warning := quietReplayAlert("warning", alerts.AlertLevelWarning, now.Add(-2*time.Hour))
critical := quietReplayAlert("critical", alerts.AlertLevelCritical, now.Add(-time.Hour))
batch := []*alerts.Alert{warning, critical}
firing := notificationDeliveryJob{Type: "webhook", Event: eventAlert, Alerts: batch, WebhookConfig: &hook}
notifier.recordSuccessfulDelivery(firing, now.Add(-time.Hour))
for _, alert := range batch {
alert.LatestTransition.To = operationaltrust.OperationalResolved
annotateResolvedMetadata(alert, now.Add(-time.Minute))
}
n := seedQuietReplay(t, q, hook, now, batch...)
if _, err := q.db.Exec(`UPDATE notification_queue SET type = 'webhook_resolved', method = 'resolved-group' WHERE id = ?`, n.ID); err != nil {
t.Fatal(err)
}
q.processNotification(n)
if got := sink.all(); len(got) != 1 || got[0].Event != "resolved" || len(got[0].Alerts) != 1 || got[0].Alerts[0].ID != critical.ID {
t.Fatalf("QUIET_RECOVERY: grouped recovery leaked held member or lost ready member: %+v", got)
}
held := loadQuietReplay(t, q, "admitted-batch")
if held.Method != "quiet-hours-replay" || held.Attempts != 0 {
t.Fatalf("postponed recovery reopened grouping window or consumed attempt: %+v", held)
}
resolved := notificationDeliveryJob{Type: "webhook", Event: eventResolved, Alerts: batch, WebhookConfig: &hook}
if eligible := notifier.filterResolvedJobsByDeliveryReceipt([]notificationDeliveryJob{resolved}); len(eligible) != 1 || len(eligible[0].Alerts) != 1 || eligible[0].Alerts[0].ID != warning.ID {
t.Fatalf("held recovery lost the original occurrence receipt: %+v", eligible)
}
if count, err := q.CancelByAlertIdentifiers([]string{warning.ID}); err != nil || count != 0 {
t.Fatalf("firing cancellation consumed admitted recovery: %d, %v", count, err)
}
now = *held.NextRetryAt
q.processNotification(held)
if got := sink.all(); len(got) != 2 || got[1].Event != "resolved" || len(got[1].Alerts) != 1 || got[1].Alerts[0].ID != warning.ID {
t.Fatalf("held recovery did not complete once: %+v", got)
}
if eligible := notifier.filterResolvedJobsByDeliveryReceipt([]notificationDeliveryJob{resolved}); len(eligible) != 0 {
t.Fatalf("delivered recovery did not consume original receipts: %+v", eligible)
}
}
func TestQueuedQuietHoursProviderFamiliesAndLegacyRows(t *testing.T) {
for _, kind := range []string{"email", "webhook", "apprise", "email_resolved", "webhook_resolved", "apprise_resolved"} {
t.Run(kind, func(t *testing.T) {
now := time.Date(2026, 10, 2, 0, 0, 0, 0, time.UTC)
owner := alerts.NewManagerWithDataDir(t.TempDir())
defer owner.Stop()
setQuietReplaySchedule(owner, quietReplaySchedule("00:00", "23:59", "UTC"))
q := newQuietReplayQueue(t, t.TempDir(), &now)
notifier, hook, sink := newQuietReplayNotifier(t, q, owner)
notifier.emailConfig.Enabled = true
notifier.appriseConfig.Enabled = true
n := seedQuietReplay(t, q, hook, now, quietReplayAlert("legacy-warning", alerts.AlertLevelWarning, now.Add(-time.Hour)))
// Legacy payloads have no operational links or quiet replay metadata;
// they are still governed by the current policy, not a schema marker.
if _, err := q.db.Exec(`UPDATE notification_queue SET type = ?, config = '{}', operational_links = '[]' WHERE id = ?`, kind, n.ID); err != nil {
t.Fatal(err)
}
q.processNotification(n)
after := loadQuietReplay(t, q, n.ID)
if after.Status != QueueStatusPending || after.Attempts != 0 || after.NextRetryAt == nil || !after.NextRetryAt.After(now) {
t.Fatalf("QUIET_PROVIDER_FAMILY: %s bypassed quiet revalidation: %+v", kind, after)
}
if got := sink.all(); len(got) != 0 {
t.Fatalf("legacy quiet row leaked: %+v", got)
}
})
}
}
func TestQueuedQuietHoursRejectsAmbiguousLinks(t *testing.T) {
now := time.Date(2026, 10, 2, 1, 0, 0, 0, time.UTC)
owner := alerts.NewManagerWithDataDir(t.TempDir())
defer owner.Stop()
setQuietReplaySchedule(owner, quietReplaySchedule("22:00", "06:00", "UTC"))
q := newQuietReplayQueue(t, t.TempDir(), &now)
_, hook, sink := newQuietReplayNotifier(t, q, owner)
n := seedQuietReplay(t, q, hook, now,
quietReplayAlert("held", alerts.AlertLevelWarning, now.Add(-time.Hour)),
quietReplayAlert("ready", alerts.AlertLevelCritical, now.Add(-time.Minute)))
links := append([]operationaltrust.NotificationLink(nil), n.Links...)
links[0].OperationalRecordID = "orphaned-record"
encoded, err := json.Marshal(links)
if err != nil {
t.Fatal(err)
}
if _, err := q.db.Exec(`UPDATE notification_queue SET operational_links = ? WHERE id = ?`, string(encoded), n.ID); err != nil {
t.Fatal(err)
}
before := loadQuietReplay(t, q, n.ID)
ready, err := q.prepareQuietHoursDelivery(n, owner.QuietHoursNotificationPolicy())
if ready || err == nil || !strings.Contains(err.Error(), "ambiguous operational link") {
t.Fatalf("orphaned operational link was silently discarded: ready=%t err=%v", ready, err)
}
if got := sink.all(); len(got) != 0 || !reflect.DeepEqual(before, loadQuietReplay(t, q, before.ID)) {
t.Fatal("rejected linkage still changed batch or attempted delivery")
}
}
+124 -65
View File
@@ -288,7 +288,7 @@ func (c *Client) getSystemTelemetryREST(ctx context.Context) (*SystemInfo, error
if err != nil {
return nil, err
}
history := parseSystemMetricHistory(response)
history := parseSystemMetricHistory(liveReportingResponses(response, end))
if history == nil {
return nil, errNoSystemTelemetry
}
@@ -302,7 +302,27 @@ func systemInfoFromMetricHistory(history *SystemMetricHistory) *SystemInfo {
if history == nil {
return nil
}
system := &SystemInfo{CollectedAt: time.Now().UTC(), Telemetry: &SystemTelemetryAvailability{}}
system := &SystemInfo{Telemetry: &SystemTelemetryAvailability{}}
for _, series := range [][]TimeSeriesPoint{history.CPUPercent, history.MemoryAvailableBytes, history.NetInRate, history.NetOutRate, history.DiskReadRate, history.DiskWriteRate} {
for _, point := range series {
if point.Timestamp.After(system.CollectedAt) {
system.CollectedAt = point.Timestamp
}
}
}
for key, series := range history.TemperatureCelsius {
if value, ok := latestTimeSeriesValue(series); ok {
if system.TemperatureCelsius == nil {
system.TemperatureCelsius = make(map[string]float64)
}
system.TemperatureCelsius[key] = value
}
for _, point := range series {
if point.Timestamp.After(system.CollectedAt) {
system.CollectedAt = point.Timestamp
}
}
}
if value, ok := latestTimeSeriesValue(history.CPUPercent); ok {
system.CPUPercent = value
system.Telemetry.CPU = true
@@ -314,9 +334,28 @@ func systemInfoFromMetricHistory(history *SystemMetricHistory) *SystemInfo {
system.MemoryAvailableBytes = int64(value)
system.Telemetry.Memory = value >= 0
}
if value, ok := latestTimeSeriesValue(history.ARCSizeBytes); ok {
system.ARCSizeBytes = int64(value)
// Cache and free RAM must describe the same bucket. An old ARC sample
// must not make a newer full-RAM reading look healthy.
var latestFree time.Time
for _, free := range history.MemoryAvailableBytes {
if free.Timestamp.After(latestFree) {
latestFree = free.Timestamp
}
}
// ARC can be observed independently when no free-RAM sample exists;
// retain it without implying known memory usage. With free RAM, require
// cache and free to describe the same bucket before deriving usage.
if latestFree.IsZero() {
if value, ok := latestTimeSeriesValue(history.ARCSizeBytes); ok {
system.ARCSizeBytes = int64(value)
}
}
for _, arc := range history.ARCSizeBytes {
if arc.Timestamp.Equal(latestFree) {
system.ARCSizeBytes = int64(arc.Value)
}
}
if value, ok := latestTimeSeriesValue(history.NetInRate); ok {
system.NetInRate = value
system.Telemetry.NetIn = true
@@ -388,18 +427,26 @@ func legacyRESTReportingGraphs() []map[string]any {
reportingGraph("arcsize", ""),
reportingGraph("interface", ""),
reportingGraph("disk", ""),
reportingGraph("cputemp", ""),
}
}
// getLegacySystemReportingData keeps a rejected optional graph from discarding
// usable CPU/memory readings. Keep the successful batch fast path; only split
// graph-validation/server errors, never authentication, rate-limit, endpoint,
// transport or cancellation failures. The query and graph set remain unchanged.
// transport or cancellation failures. Bind the catalogue-selected graph set once
// for both the batch and split; the requested query window remains unchanged.
func (c *Client) getLegacySystemReportingData(ctx context.Context, query map[string]any) ([]trueNASReportingGetDataResponse, error) {
graphs := legacyRESTReportingGraphs()
graphs, err := c.legacyReportingGraphs(ctx)
if err != nil {
return nil, err
}
response, err := c.getReportingDataREST(ctx, graphs, query)
if err == nil || !isReportingGraphFailure(err) {
return response, err
if err == nil {
return bindLegacyReportingResponses(graphs, response), nil
}
if !isReportingGraphFailure(err) {
return nil, err
}
batchErr := err
var collected []trueNASReportingGetDataResponse
@@ -419,7 +466,7 @@ func (c *Client) getLegacySystemReportingData(ctx context.Context, query map[str
if len(collected) == 0 {
return nil, batchErr
}
return collected, nil
return bindLegacyReportingResponses(graphs, collected), nil
}
func isReportingGraphFailure(err error) bool {
@@ -2378,6 +2425,7 @@ type trueNASReportingGetDataResponse struct {
Aggregations trueNASReportingAggregations `json:"aggregations"`
Start int64 `json:"start"`
End int64 `json:"end"`
Step int64 `json:"step"`
Legend []string `json:"legend"`
}
@@ -3178,34 +3226,25 @@ func parseSystemTelemetry(fields map[string]any, intervalSeconds int, collectedA
}
func parseSystemTemperatures(responses []trueNASReportingGetDataResponse) map[string]float64 {
if len(responses) == 0 {
return nil
}
temperatures := make(map[string]float64)
for _, response := range responses {
if strings.TrimSpace(strings.ToLower(response.Name)) != "cputemp" || len(response.Legend) == 0 {
if !strings.EqualFold(strings.TrimSpace(response.Name), "cputemp") {
continue
}
values := extractReportingLegendFloatValues(response.Aggregations.Mean, response.Legend)
if len(values) == 0 && len(response.Data) > 0 {
values = extractReportingLegendFloatValues(response.Data[len(response.Data)-1], response.Legend)
values := latestReportingValues(response)
// Retain aggregate-only compatibility, never use a window mean to fill a
// missing/null current row or an invalid native timing envelope.
if len(response.Data) == 0 && response.Step == 0 {
values = finiteReportingValues(extractReportingLegendFloatValues(response.Aggregations.Mean, response.Legend))
}
if len(values) == 0 {
continue
}
for index, legend := range response.Legend {
value, ok := values[legend]
if !ok || value <= 0 {
continue
}
key := canonicalSystemTemperatureKey(legend, index, len(response.Legend))
if key == "" {
continue
if key := canonicalSystemTemperatureKey(legend, index, len(response.Legend)); key != "" {
temperatures[key] = value
}
temperatures[key] = value
}
}
if len(temperatures) == 0 {
@@ -3218,75 +3257,93 @@ func parseSystemMetricHistory(responses []trueNASReportingGetDataResponse) *Syst
if len(responses) == 0 {
return nil
}
history := &SystemMetricHistory{}
history := &SystemMetricHistory{TemperatureCelsius: make(map[string][]TimeSeriesPoint)}
io := map[string]map[string][]TimeSeriesPoint{
"netin": {}, "netout": {}, "diskread": {}, "diskwrite": {},
}
seen := make(map[string]bool)
for _, response := range responses {
name := strings.TrimSpace(strings.ToLower(response.Name))
name := strings.ToLower(strings.TrimSpace(response.Name))
switch name {
case "cpu", "memory", "arcsize", "interface", "disk":
case "cpu", "memory", "arcsize", "cputemp", "interface", "disk":
default:
continue
}
for _, raw := range response.Data {
timestamp, values, ok := parseReportingSeriesValues(raw, response.Legend)
if !ok || timestamp.IsZero() || len(values) == 0 {
identifier := readStringAny(map[string]any{"identifier": response.Identifier}, "identifier")
key := name + "\x00" + identifier
if seen[key] {
continue
}
seen[key] = true
if name == "interface" {
io["netin"][key] = nil
io["netout"][key] = nil
}
if name == "disk" {
io["diskread"][key] = nil
io["diskwrite"][key] = nil
}
for index := range response.Data {
timestamp, values, ok := reportingRow(response, index)
if !ok {
continue
}
switch name {
case "cpu":
if value, ok := parseSystemCPUPercent(values); ok {
if value, ok := reportingCPUPercent(values, response.Legend); ok {
history.CPUPercent = appendTimeSeriesPoint(history.CPUPercent, timestamp, value)
}
case "memory":
// FreeBSD active pages are only one used-memory component, not
// total usage. Without an explicit used series, let the provider
// derive usage from system capacity, free memory and ARC.
if !reportingMemoryRowValid(values, response.Legend) {
continue
}
if value, ok := parseSystemMemoryPercent(values); ok {
history.MemoryPercent = appendTimeSeriesPoint(history.MemoryPercent, timestamp, value)
}
if value, ok := pickReportingValue(values, "used", "memory_used", "used_bytes", "memory"); ok {
if value, ok := pickReportingValue(values, "used", "memory_used", "used_bytes", "memory"); ok && value >= 0 {
history.MemoryUsedBytes = appendTimeSeriesPoint(history.MemoryUsedBytes, timestamp, value)
}
if value, ok := pickReportingValue(values, "available", "free", "available_bytes", "free_bytes"); ok {
if value, ok := pickReportingValue(values, "available", "free", "available_bytes", "free_bytes", "memory-free_value"); ok && value >= 0 {
history.MemoryAvailableBytes = appendTimeSeriesPoint(history.MemoryAvailableBytes, timestamp, value)
}
if value, ok := pickReportingValue(values, "total", "memory_total", "total_bytes", "physical_memory_total"); ok {
if value, ok := pickReportingValue(values, "total", "memory_total", "total_bytes", "physical_memory_total"); ok && value > 0 {
history.MemoryTotalBytes = appendTimeSeriesPoint(history.MemoryTotalBytes, timestamp, value)
}
case "arcsize":
if value, ok := pickReportingValue(values, "arc_size", "size", "arcsize"); ok {
if value, ok := pickReportingValue(values, "arc_size", "size", "arcsize", "arcsize_value"); ok && value >= 0 {
history.ARCSizeBytes = appendTimeSeriesPoint(history.ARCSizeBytes, timestamp, value)
}
case "interface":
if value, ok := pickReportingValue(values, "received", "received_bytes", "rx", "rx_bytes", "netin", "in"); ok {
history.NetInRate = appendTimeSeriesPoint(history.NetInRate, timestamp, value)
case "cputemp":
for i, legend := range response.Legend {
if value, ok := values[legend]; ok && value > 0 {
key := canonicalSystemTemperatureKey(legend, i, len(response.Legend))
if key != "" {
history.TemperatureCelsius[key] = appendTimeSeriesPoint(history.TemperatureCelsius[key], timestamp, value)
}
}
}
if value, ok := pickReportingValue(values, "sent", "sent_bytes", "tx", "tx_bytes", "netout", "out"); ok {
history.NetOutRate = appendTimeSeriesPoint(history.NetOutRate, timestamp, value)
case "interface":
if value, ok := pickReportingValue(values, "received", "received_bytes", "rx", "rx_bytes", "netin", "in"); ok && value >= 0 {
io["netin"][key] = appendTimeSeriesPoint(io["netin"][key], timestamp, value)
}
if value, ok := pickReportingValue(values, "sent", "sent_bytes", "tx", "tx_bytes", "netout", "out"); ok && value >= 0 {
io["netout"][key] = appendTimeSeriesPoint(io["netout"][key], timestamp, value)
}
case "disk":
if value, ok := pickReportingValue(values, "read", "read_bytes", "diskread", "bytes_read"); ok {
history.DiskReadRate = appendTimeSeriesPoint(history.DiskReadRate, timestamp, value)
if value, ok := pickReportingValue(values, "read", "read_bytes", "diskread", "bytes_read", "disk_octets_read"); ok && value >= 0 {
io["diskread"][key] = appendTimeSeriesPoint(io["diskread"][key], timestamp, value)
}
if value, ok := pickReportingValue(values, "write", "write_bytes", "diskwrite", "bytes_written"); ok {
history.DiskWriteRate = appendTimeSeriesPoint(history.DiskWriteRate, timestamp, value)
if value, ok := pickReportingValue(values, "write", "write_bytes", "diskwrite", "bytes_written", "disk_octets_write"); ok && value >= 0 {
io["diskwrite"][key] = appendTimeSeriesPoint(io["diskwrite"][key], timestamp, value)
}
}
}
}
if len(history.CPUPercent) == 0 &&
len(history.MemoryPercent) == 0 &&
len(history.MemoryUsedBytes) == 0 &&
len(history.MemoryAvailableBytes) == 0 &&
len(history.MemoryTotalBytes) == 0 &&
len(history.ARCSizeBytes) == 0 &&
len(history.NetInRate) == 0 &&
len(history.NetOutRate) == 0 &&
len(history.DiskReadRate) == 0 &&
len(history.DiskWriteRate) == 0 {
history.NetInRate = sumReportingDevices(io["netin"])
history.NetOutRate = sumReportingDevices(io["netout"])
history.DiskReadRate = sumReportingDevices(io["diskread"])
history.DiskWriteRate = sumReportingDevices(io["diskwrite"])
if len(history.CPUPercent)+len(history.MemoryPercent)+len(history.MemoryUsedBytes)+len(history.MemoryAvailableBytes)+len(history.MemoryTotalBytes)+len(history.ARCSizeBytes)+len(history.NetInRate)+len(history.NetOutRate)+len(history.DiskReadRate)+len(history.DiskWriteRate)+len(history.TemperatureCelsius) == 0 {
return nil
}
return history
@@ -3337,8 +3394,10 @@ func parseReportingDiskTemperatureHistory(responses []trueNASReportingGetDataRes
}
points := make([]TimeSeriesPoint, 0, len(response.Data))
for _, raw := range response.Data {
timestamp, value, ok := parseReportingSeriesPoint(raw, response.Legend)
for index := range response.Data {
timestamp, values, ok := reportingRow(response, index)
value, hasValue := pickReportingValue(values, response.Legend...)
ok = ok && hasValue && value > 0
if !ok || timestamp.IsZero() {
continue
}
+4 -4
View File
@@ -2195,8 +2195,8 @@ func TestRESTReportingGraphFailurePreservesSnapshotTelemetry(t *testing.T) {
t.Fatal("inventory lost")
}
}
if transport.calls != 12 {
t.Fatalf("reporting calls = %d, want bounded batch + five graphs per snapshot", transport.calls)
if transport.calls != 2*(1+len(legacyRESTReportingGraphs())) {
t.Fatalf("reporting calls = %d, want bounded batch + selected graphs per snapshot", transport.calls)
}
history, err := client.GetSystemMetricHistory(context.Background(), time.Hour)
if err != nil {
@@ -2222,7 +2222,7 @@ func TestRESTReportingGraphFailureBoundaries(t *testing.T) {
}
want := 1
if status == 400 || status == 422 || status == 500 {
want = 6
want = 7
}
if transport.calls != want {
t.Fatalf("calls = %d, want %d", transport.calls, want)
@@ -2464,7 +2464,7 @@ func TestRESTSnapshotRetainsSanitizedTelemetryFailure(t *testing.T) {
}
wantCalls := 1
if tc.status == 422 {
wantCalls = 6
wantCalls = 1 + len(legacyRESTReportingGraphs())
}
if transport.calls != wantCalls {
t.Fatalf("reporting calls=%d, want %d", transport.calls, wantCalls)
+281
View File
@@ -0,0 +1,281 @@
package truenas
import (
"context"
"errors"
"fmt"
"math"
"net/http"
"sort"
"strings"
"time"
)
// CORE's RRD graphs are device-scoped. Ask the reporting catalogue for its
// identifiers, not disk serials or configured (possibly renamed) interfaces.
// Older REST bridges without the catalogue retain the unscoped compatibility
// request. Authentication, transport and malformed replies are never retried.
func (c *Client) legacyReportingGraphs(ctx context.Context) ([]map[string]any, error) {
if err := ctx.Err(); err != nil {
return nil, err
}
var catalogue []struct {
Name string `json:"name"`
Identifiers []string `json:"identifiers"`
}
if err := c.getJSON(ctx, http.MethodGet, "/reporting/graphs", &catalogue); err != nil {
if isAPIStatus(err, http.StatusNotFound) || isAPIStatus(err, http.StatusMethodNotAllowed) {
return legacyRESTReportingGraphs(), nil
}
return nil, err
}
graphs := []map[string]any{
reportingGraph("cpu", ""), reportingGraph("memory", ""),
reportingGraph("arcsize", ""), reportingGraph("cputemp", ""),
}
seen := make(map[string]bool)
for _, graph := range catalogue {
name := strings.ToLower(strings.TrimSpace(graph.Name))
if name != "disk" && name != "interface" {
continue
}
for _, identifier := range dedupeStrings(graph.Identifiers) {
identifier = strings.TrimSpace(identifier)
key := name + "\x00" + identifier
if identifier == "" || seen[key] {
continue
}
seen[key] = true
// Refuse an oversized catalogue, rather than silently publishing a
// truncated host total or allowing an unbounded split retry loop.
if len(graphs) >= 256 {
return nil, fmt.Errorf("truenas reporting catalogue exceeds 256 graphs")
}
graphs = append(graphs, reportingGraph(name, identifier))
}
}
return graphs, nil
}
func isAPIStatus(err error, status int) bool {
var apiErr *APIError
return errors.As(err, &apiErr) && apiErr.StatusCode == status
}
func finiteReportingValues(values map[string]float64) map[string]float64 {
for key, value := range values {
if math.IsNaN(value) || math.IsInf(value, 0) {
delete(values, key)
}
}
return values
}
// A native row has exactly one value per legend; its timestamp is external.
// An explicit-timestamp row has one extra element. Never guess an epoch from a
// value-only row with missing/invalid timing or accept a misaligned short row.
func reportingRow(response trueNASReportingGetDataResponse, index int) (time.Time, map[string]float64, bool) {
raw := response.Data[index]
if row, ok := raw.([]any); ok {
if len(response.Legend) == 0 {
return time.Time{}, nil, false
}
if len(row) == len(response.Legend) {
if response.Start <= 0 || response.End < response.Start || response.Step <= 0 ||
int64(index) > (response.End-response.Start)/response.Step {
return time.Time{}, nil, false
}
timestamp := time.Unix(response.Start+int64(index)*response.Step, 0).UTC()
values := finiteReportingValues(extractReportingLegendFloatValues(row, response.Legend))
return timestamp, values, len(values) > 0
}
if len(row) != len(response.Legend)+1 {
return time.Time{}, nil, false
}
}
timestamp, values, ok := parseReportingSeriesValues(raw, response.Legend)
values = finiteReportingValues(values)
return timestamp, values, ok && timestamp.Unix() > 0 && len(values) > 0
}
// Keep native History intact, but do not label an old RRD point as a current
// reading. Two returned steps allow the usual unfinished/null RRD bucket;
// older points stay available only through History. Never exceed the live
// query window even if a coarse/malformed step allows older buckets. Preserve
// row indices.
func liveReportingResponses(responses []trueNASReportingGetDataResponse, end int64) []trueNASReportingGetDataResponse {
live := append([]trueNASReportingGetDataResponse(nil), responses...)
for i, response := range live {
if response.Step <= 0 {
continue // timestamped legacy/modern compatibility payload
}
live[i].Data = append([]any(nil), response.Data...)
for index := range response.Data {
timestamp, _, ok := reportingRow(response, index)
gap := end - timestamp.Unix()
if !ok || timestamp.Unix() > end || gap > legacyRESTTelemetryWindowSeconds || (gap > response.Step && gap-response.Step > response.Step) {
live[i].Data[index] = nil
}
}
}
return live
}
func latestReportingValues(response trueNASReportingGetDataResponse) map[string]float64 {
var latest time.Time
var values map[string]float64
for index := range response.Data {
timestamp, row, ok := reportingRow(response, index)
if ok && (values == nil || timestamp.After(latest)) {
latest, values = timestamp, row
}
}
return values
}
var coreCPUStates = []string{"interrupt", "system", "user", "nice", "idle"}
var coreMemoryClasses = []string{"memory-active_value", "memory-inactive_value", "memory-wired_value", "memory-laundry_value", "memory-free_value"}
func hasReportingLegends(legends, required []string) bool {
present := make(map[string]bool, len(legends))
for _, legend := range legends {
present[normalizeTemperatureLegendLabel(legend)] = true
}
for _, legend := range required {
if !present[normalizeTemperatureLegendLabel(legend)] {
return false
}
}
return true
}
// RRD CPU state rates are not necessarily percentages (the supplied states sum
// above 100). Normalize the complete state vector; an all-zero bucket is not
// 100% busy. Explicit usage=0 and a measured all-idle vector remain real zeros.
func reportingCPUPercent(values map[string]float64, legends []string) (float64, bool) {
if !hasReportingLegends(legends, coreCPUStates) {
return parseSystemCPUPercent(values)
}
var total, idle float64
for _, state := range coreCPUStates {
value, ok := pickReportingValue(values, state)
if !ok || value < 0 {
return 0, false
}
total += value
if state == "idle" {
idle = value
}
}
if total <= 0 || math.IsInf(total, 0) {
return 0, false
}
return (total - idle) / total * 100, true
}
// Five zero memory classes cannot describe physical RAM. Keep this native RRD
// empty-bucket sentinel distinct from a real zero-free reading with used pages.
func reportingMemoryRowValid(values map[string]float64, legends []string) bool {
if !hasReportingLegends(legends, coreMemoryClasses) {
return true
}
var total float64
for _, class := range coreMemoryClasses {
value, ok := pickReportingValue(values, class)
if !ok || value < 0 {
return false
}
total += value
}
return total > 0 && !math.IsInf(total, 0)
}
// Sum only simultaneous observations of every returned device. Null/missing
// members are not zero, duplicate device replies are not additional traffic,
// and values are already rates: step is for time, not another division.
func sumReportingDevices(devices map[string][]TimeSeriesPoint) []TimeSeriesPoint {
if len(devices) == 0 {
return nil
}
var totals map[int64]float64
for _, series := range devices {
points := make(map[int64]float64, len(series))
for _, point := range series {
points[point.Timestamp.Unix()] = point.Value
}
if totals == nil {
totals = points
continue
}
for timestamp, total := range totals {
value, ok := points[timestamp]
if !ok {
delete(totals, timestamp)
} else {
totals[timestamp] = total + value
}
}
}
timestamps := make([]int64, 0, len(totals))
for timestamp := range totals {
timestamps = append(timestamps, timestamp)
}
sort.Slice(timestamps, func(i, j int) bool { return timestamps[i] < timestamps[j] })
points := make([]TimeSeriesPoint, 0, len(timestamps))
for _, timestamp := range timestamps {
if !math.IsInf(totals[timestamp], 0) {
points = appendTimeSeriesPoint(points, time.Unix(timestamp, 0).UTC(), totals[timestamp])
}
}
return points
}
// Bind device responses to the requested catalogue. A failed/omitted device
// graph remains an absent member, not a smaller apparently complete host sum.
func bindLegacyReportingResponses(graphs []map[string]any, responses []trueNASReportingGetDataResponse) []trueNASReportingGetDataResponse {
byKey := make(map[string]trueNASReportingGetDataResponse)
for _, response := range responses {
name := strings.ToLower(strings.TrimSpace(response.Name))
identifier := readStringAny(map[string]any{"identifier": response.Identifier}, "identifier")
key := name + "\x00" + identifier
if _, exists := byKey[key]; !exists {
byKey[key] = response
}
}
bound := make([]trueNASReportingGetDataResponse, 0, len(graphs))
for _, graph := range graphs {
name := readStringAny(graph, "name")
identifier := readStringAny(graph, "identifier")
response, ok := byKey[name+"\x00"+identifier]
if !ok {
response = trueNASReportingGetDataResponse{Name: name, Identifier: identifier}
}
bound = append(bound, response)
}
return bound
}
func systemTemperatureHistory(sensors map[string][]TimeSeriesPoint) []TimeSeriesPoint {
at := make(map[int64]map[string]float64)
for key, series := range sensors {
for _, point := range series {
timestamp := point.Timestamp.Unix()
if at[timestamp] == nil {
at[timestamp] = make(map[string]float64)
}
at[timestamp][key] = point.Value
}
}
timestamps := make([]int64, 0, len(at))
for timestamp := range at {
timestamps = append(timestamps, timestamp)
}
sort.Slice(timestamps, func(i, j int) bool { return timestamps[i] < timestamps[j] })
var points []TimeSeriesPoint
for _, timestamp := range timestamps {
if value := maxTrueNASSystemTemperature(SystemInfo{TemperatureCelsius: at[timestamp]}); value != nil {
points = appendTimeSeriesPoint(points, time.Unix(timestamp, 0).UTC(), *value)
}
}
return points
}
+447
View File
@@ -0,0 +1,447 @@
package truenas
import (
"context"
"encoding/json"
"errors"
"fmt"
"math"
"net/http"
"os"
"reflect"
"strings"
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
)
// Five native result excerpts supplied in #2077 on 1 October 2026. Identifiers
// are anonymised and the SHORT EXCERPT windows made coherent. This is decoder
// evidence, not a complete time window, REST capture or appliance acceptance.
func core13ReportingFixture(t *testing.T) []trueNASReportingGetDataResponse {
t.Helper()
raw, err := os.ReadFile("testdata/core13_reporting.json")
if err != nil {
t.Fatal(err)
}
var replies []trueNASReportingGetDataResponse
if err := json.Unmarshal(raw, &replies); err != nil {
t.Fatal(err)
}
return replies
}
func assertCOREPoint(t *testing.T, points []TimeSeriesPoint, index int, timestamp int64, value float64) {
t.Helper()
if len(points) <= index || points[index].Timestamp.Unix() != timestamp || math.Abs(points[index].Value-value) > 0.000001 {
t.Fatalf("point %d: %+v, want %d / %v", index, points, timestamp, value)
}
}
// These five subtests also compile with the predecessor decoder and are the
// predetermined adverse controls; do not substitute compiler failure for them.
func TestCORE13NativeResultControls(t *testing.T) {
replies := core13ReportingFixture(t)
start := int64(1789000000)
t.Run("CPU", func(t *testing.T) {
history := parseSystemMetricHistory(replies[:1])
if history == nil {
t.Fatal("native CPU rows produced no history")
}
want := (0.033358538468 + 4.4805217137 + 7.8754519386 + 7.8754519386) / (100 + 0.033358538468 + 4.4805217137 + 7.8754519386 + 7.8754519386) * 100
assertCOREPoint(t, history.CPUPercent, 0, start, want)
if len(history.CPUPercent) != 1 {
t.Fatal("all-zero state buckets became CPU measurements")
}
})
t.Run("memory", func(t *testing.T) {
history := parseSystemMetricHistory(replies[2:3])
if history == nil {
t.Fatal("native memory aliases produced no history")
}
assertCOREPoint(t, history.MemoryAvailableBytes, 0, start, 2178945024)
if len(history.MemoryAvailableBytes) != 1 || len(history.MemoryUsedBytes) != 0 {
t.Fatal("empty bucket or active pages became total usage")
}
})
t.Run("disk", func(t *testing.T) {
history := parseSystemMetricHistory(replies[3:4])
if history == nil {
t.Fatal("native disk rows produced no history")
}
assertCOREPoint(t, history.DiskReadRate, 0, start, 484.54070578)
assertCOREPoint(t, history.DiskReadRate, 1, start+10, 0)
assertCOREPoint(t, history.DiskWriteRate, 1, start+10, 1276361.9904)
if len(history.DiskReadRate) != 2 {
t.Fatal("null disk bucket became zero")
}
})
t.Run("network", func(t *testing.T) {
history := parseSystemMetricHistory(replies[4:5])
if history == nil {
t.Fatal("native network rows produced no history")
}
assertCOREPoint(t, history.NetInRate, 0, start, 9728.2270494)
assertCOREPoint(t, history.NetOutRate, 0, start, 45551.708899)
if len(history.NetInRate) != 1 {
t.Fatal("null or overlap became RX")
}
})
t.Run("temperature request", func(t *testing.T) {
found := false
for _, graph := range legacyRESTReportingGraphs() {
found = found || graph["name"] == "cputemp"
}
if !found {
t.Fatal("legacy graph set never requests CPU temperature")
}
})
}
func TestCORE13TemperatureUsesSamplesNotMeans(t *testing.T) {
replies := core13ReportingFixture(t)[1:2]
replies[0].Aggregations.Mean = []any{999., 999., 999., 999., 999., 999., 999., 999.}
temperatures := parseSystemTemperatures(replies)
if len(temperatures) != 8 || temperatures["cputemp2"] != 30.95 {
t.Fatalf("temperature samples lost: %+v", temperatures)
}
history := parseSystemMetricHistory(replies)
if history == nil {
t.Fatal("temperature-only history lost")
}
system := systemInfoFromMetricHistory(history)
if got := maxTrueNASSystemTemperature(*system); got == nil || *got != 30.95 {
t.Fatalf("current temperature = %v", got)
}
points := systemTemperatureHistory(history.TemperatureCelsius)
assertCOREPoint(t, points, 0, 1789000000, 30.95)
replies[0].Data[0] = []any{nil, nil, nil, nil, nil, nil, nil, nil}
if got := parseSystemTemperatures(replies); len(got) != 0 {
t.Fatalf("all-null rows substituted a mean: %v", got)
}
}
func TestCORE13TimingAndMissingValueBoundaries(t *testing.T) {
for _, tc := range []struct {
name, body string
count int
last int64
}{
{"external zero", `{"start":1789000000,"end":1789000020,"step":10,"legend":["usage"],"data":[[0],[null],[12]]}`, 2, 1789000020},
{"missing step", `{"start":1789000000,"end":1789000020,"legend":["usage"],"data":[[42]]}`, 0, 0},
{"zero step", `{"start":1789000000,"end":1789000020,"step":0,"legend":["usage"],"data":[[42]]}`, 0, 0},
{"negative step", `{"start":1789000000,"end":1789000020,"step":-10,"legend":["usage"],"data":[[42]]}`, 0, 0},
{"missing start", `{"end":1789000020,"step":10,"legend":["usage"],"data":[[42]]}`, 0, 0},
{"reversed range", `{"start":1789000020,"end":1789000000,"step":10,"legend":["usage"],"data":[[42]]}`, 0, 0},
{"past end", `{"start":1789000000,"end":1789000000,"step":10,"legend":["usage"],"data":[[42],[43]]}`, 1, 1789000000},
{"overflow", `{"start":9223372036854775797,"end":9223372036854775807,"step":9223372036854775807,"legend":["usage"],"data":[[42],[43]]}`, 1, 9223372036854775797},
{"timestamped", `{"legend":["usage"],"data":[[1789000000,0],[1789000010000,12]]}`, 2, 1789000010},
{"object", `{"legend":["usage"],"data":[{"timestamp":"2026-09-10T01:46:40Z","usage":0}]}`, 1, 1789004800},
{"short row", `{"legend":["rx","tx","overlap"],"data":[[1789000000,12]]}`, 0, 0},
{"nonfinite", `{"start":1789000000,"end":1789000000,"step":10,"legend":["usage"],"data":[["NaN"],["+Inf"]]}`, 0, 0},
} {
t.Run(tc.name, func(t *testing.T) {
var response trueNASReportingGetDataResponse
if err := json.Unmarshal([]byte(tc.body), &response); err != nil {
t.Fatal(err)
}
response.Name = "cpu"
history := parseSystemMetricHistory([]trueNASReportingGetDataResponse{response})
var points []TimeSeriesPoint
if history != nil {
points = history.CPUPercent
}
if len(points) != tc.count || (tc.count > 0 && points[len(points)-1].Timestamp.Unix() != tc.last) {
t.Fatalf("invalid timing/value handling: %+v", points)
}
})
}
}
func TestCORE13CPUAndMemoryEmptyBuckets(t *testing.T) {
for _, tc := range []struct {
row []any
valid bool
want float64
}{
{[]any{0., 0., 0., 0., 100.}, true, 0}, {[]any{0., 0., 0., 0., 0.}, false, 0},
{[]any{nil, 0., 0., 0., 100.}, false, 0}, {[]any{-1., 0., 0., 0., 100.}, false, 0},
{[]any{10., 10., 10., 10., 160.}, true, 20},
} {
response := trueNASReportingGetDataResponse{Name: "cpu", Legend: coreCPUStates, Start: 1789000000, End: 1789000000, Step: 10, Data: []any{tc.row}}
history := parseSystemMetricHistory([]trueNASReportingGetDataResponse{response})
if (history != nil) != tc.valid {
t.Fatalf("CPU row %v became %+v", tc.row, history)
}
if tc.valid {
assertCOREPoint(t, history.CPUPercent, 0, response.Start, tc.want)
}
}
for _, tc := range []struct {
row []any
valid bool
}{
{[]any{0., 0., 0., 16., 0.}, true}, {[]any{0., 0., 0., 0., 0.}, false}, {[]any{1., 1., 1., nil, 0.}, false},
} {
response := trueNASReportingGetDataResponse{Name: "memory", Legend: coreMemoryClasses, Start: 1789000000, End: 1789000000, Step: 10, Data: []any{tc.row}}
history := parseSystemMetricHistory([]trueNASReportingGetDataResponse{response})
if (history != nil) != tc.valid {
t.Fatalf("memory row %v became %+v", tc.row, history)
}
if tc.valid {
assertCOREPoint(t, history.MemoryAvailableBytes, 0, response.Start, 0)
}
}
}
func TestCORE13MultipleDevicesAndFreshness(t *testing.T) {
base := trueNASReportingGetDataResponse{Name: "interface", Identifier: "nic-a", Legend: []string{"rx", "tx", "overlap"}, Start: 1789000000, End: 1789000020, Step: 10, Data: []any{[]any{10., 20., 10.}, []any{0., 5., 0.}, []any{nil, nil, nil}}}
second := base
second.Identifier = "nic-b"
second.Data = []any{[]any{100., 200., 100.}, []any{0., nil, 0.}, []any{4., 8., 4.}}
for _, replies := range [][]trueNASReportingGetDataResponse{{base, second, base}, {second, base, base}} {
history := parseSystemMetricHistory(replies)
assertCOREPoint(t, history.NetInRate, 0, base.Start, 110)
assertCOREPoint(t, history.NetInRate, 1, base.Start+10, 0)
assertCOREPoint(t, history.NetOutRate, 0, base.Start, 220)
if len(history.NetInRate) != 2 || len(history.NetOutRate) != 1 {
t.Fatalf("missing device reading treated as zero: %+v", history)
}
}
if got := parseSystemMetricHistory(liveReportingResponses([]trueNASReportingGetDataResponse{base}, base.End+31)); got != nil {
t.Fatalf("stale sample became current: %+v", got)
}
fresh := parseSystemMetricHistory(liveReportingResponses([]trueNASReportingGetDataResponse{base}, base.End))
assertCOREPoint(t, fresh.NetInRate, 1, base.Start+10, 0)
if at := systemInfoFromMetricHistory(fresh).CollectedAt.Unix(); at != base.Start+10 {
t.Fatalf("sample falsely stamped as current: %d", at)
}
if !reflect.DeepEqual(base.Data[0], []any{10., 20., 10.}) {
t.Fatal("live filtering mutated native History")
}
}
// A request-validating REST bridge models catalogue -> per-device requests ->
// reporting rows -> complete snapshot and canonical host/history. Only the five
// excerpts are native evidence; catalogue, ARC, HTTP and timing are controls.
type core13ReportingTransport struct {
routes alertArgsTransport
replies []trueNASReportingGetDataResponse
requested []map[string]any
catalogueCalls, reportingCalls int
}
func (s *core13ReportingTransport) RoundTrip(r *http.Request) (*http.Response, error) {
if r.URL.Path == "/api/v2.0/reporting/graphs" {
s.catalogueCalls++
return (alertArgsTransport{r.URL.Path: {body: `[{"name":"disk","identifiers":["disk-a","disk-a",""]},{"name":"interface","identifiers":["nic-a"]}]`}}).RoundTrip(r)
}
if r.URL.Path != "/api/v2.0/reporting/get_data" {
return s.routes.RoundTrip(r)
}
s.reportingCalls++
var request struct {
Graphs []map[string]any `json:"graphs"`
Query map[string]any `json:"query"`
}
if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
return nil, err
}
s.requested = append(s.requested, request.Graphs...)
var responses []trueNASReportingGetDataResponse
for _, graph := range request.Graphs {
name := readStringAny(graph, "name")
identifier := readStringAny(graph, "identifier")
if (name == "disk" && identifier != "disk-a") || (name == "interface" && identifier != "nic-a") || (name != "disk" && name != "interface" && graph["identifier"] != nil) {
return (alertArgsTransport{r.URL.Path: {status: 422, body: `{"error":"native graph needs the correct device identifier"}`}}).RoundTrip(r)
}
for _, reply := range s.replies {
if reply.Name != name {
continue
}
reply.End = int64(request.Query["end"].(float64))
reply.Start = reply.End - int64(len(reply.Data)-1)*reply.Step
responses = append(responses, reply)
}
if name == "arcsize" {
end := int64(request.Query["end"].(float64))
responses = append(responses, trueNASReportingGetDataResponse{Name: name, Start: end - 10, End: end, Step: 10, Legend: []string{"arcsize_value"}, Data: []any{[]any{float64(8 << 30)}, []any{nil}}})
}
}
raw, err := json.Marshal(responses)
if err != nil {
return nil, err
}
return (alertArgsTransport{r.URL.Path: {body: string(raw)}}).RoundTrip(r)
}
func TestCORE13SnapshotAndCanonicalHistoryJourney(t *testing.T) {
previous := IsFeatureEnabled()
SetFeatureEnabled(true)
t.Cleanup(func() { SetFeatureEnabled(previous) })
routes := alertArgsTransport(defaultAPIResponses())
routes["/api/v2.0/system/info"] = apiResponse{body: `{"hostname":"core-fixture","version":"TrueNAS-13.0-U6.1","physmem":34359738368,"cores":8}`}
transport := &core13ReportingTransport{routes: routes, replies: core13ReportingFixture(t)}
client := newLegacyRESTReportingClient(t, routes)
client.httpClient.Transport = transport
t.Cleanup(client.Close)
provider := NewLiveProviderForConnection(&APIFetcher{Client: client}, "core-fixture-connection")
for cycle := 0; cycle < 2; cycle++ {
if err := provider.Refresh(context.Background()); err != nil {
t.Fatal(err)
}
snapshot := provider.Snapshot()
if snapshot.System.CPUCount != 8 || snapshot.System.MemoryTotalBytes != 34359738368 || len(snapshot.Disks) != 2 || len(snapshot.Pools) != 1 {
t.Fatal("native telemetry lost inventory/capacity")
}
var host *unifiedresources.Resource
for _, record := range provider.Records() {
if record.Resource.Type == unifiedresources.ResourceTypeAgent {
copy := record.Resource
host = &copy
}
}
if host == nil || host.Metrics.CPU == nil || host.Metrics.Memory == nil || host.Temperature == nil || host.Agent.Memory == nil {
t.Fatalf("collapsed/expanded row lost native readings: %+v", host)
}
wantUsed := int64(34359738368-2178945024) - int64(8<<30)
if *host.Metrics.Memory.Used != wantUsed || host.Agent.Memory.Used != wantUsed || *host.Temperature != 30.95 || host.Agent.Memory.Cache != 8<<30 {
t.Fatalf("ARC-aware current values wrong: %+v %+v", host.Metrics, host.Agent)
}
for key, metric := range map[string]*unifiedresources.MetricValue{"netin": host.Metrics.NetIn, "netout": host.Metrics.NetOut, "diskread": host.Metrics.DiskRead, "diskwrite": host.Metrics.DiskWrite} {
want := map[string]float64{"netin": 9728.2270494, "netout": 45551.708899, "diskread": 0, "diskwrite": 1276361.9904}[key]
if metric == nil || metric.Unit != "bytes/s" || math.Abs(metric.Value-want) > 0.000001 {
t.Fatalf("%s current rate: %+v", key, metric)
}
}
id, history, err := provider.SystemMetricHistory(context.Background(), time.Hour)
if err != nil {
t.Fatal(err)
}
if id != "core-fixture-connection" || len(history) != 7 {
t.Fatalf("canonical History panels: %q %+v", id, history)
}
if len(history["cpu"]) != 1 || len(history["memory"]) != 1 || len(history["temperature"]) != 1 || len(history["diskread"]) != 2 {
t.Fatal("empty buckets filled with means or zeros")
}
if math.Abs(history["memory"][0].Value-host.Metrics.Memory.Percent) > 0.000001 {
t.Fatal("live/History disagree on free RAM/ARC")
}
}
if transport.catalogueCalls != 4 || transport.reportingCalls != 4 {
t.Fatalf("successful catalogue/batch unexpectedly retried: %+v", transport)
}
for _, graph := range transport.requested {
if name := readStringAny(graph, "name"); (name == "disk" || name == "interface") && readStringAny(graph, "identifier") == "" {
t.Fatal("device graph queried without identifier")
}
}
}
func TestCORE13CatalogueAndResponseSafety(t *testing.T) {
for _, tc := range []struct {
status int
body string
fallback bool
}{
{404, `{}`, true}, {405, `{}`, true}, {401, `{}`, false}, {403, `{}`, false}, {429, `{}`, false}, {500, `{}`, false}, {200, `{`, false},
} {
routes := alertArgsTransport{"/api/v2.0/reporting/graphs": {status: tc.status, body: tc.body}}
client := newLegacyRESTReportingClient(t, routes)
graphs, err := client.legacyReportingGraphs(context.Background())
if tc.fallback {
if err != nil || len(graphs) != 6 {
t.Fatalf("missing catalogue fallback: %v %v", graphs, err)
}
} else if err == nil {
t.Fatalf("catalogue failure concealed: %d", tc.status)
}
client.Close()
}
client := newLegacyRESTReportingClient(t, nil)
ctx, cancel := context.WithCancel(context.Background())
cancel()
if _, err := client.GetSystemTelemetry(ctx); !errors.Is(err, context.Canceled) {
t.Fatalf("cancelled catalogue: %v", err)
}
replies := core13ReportingFixture(t)
missing := bindLegacyReportingResponses([]map[string]any{reportingGraph("interface", "nic-a"), reportingGraph("interface", "nic-b")}, replies)
if got := parseSystemMetricHistory(missing); got != nil {
t.Fatalf("omitted device became a complete host rate: %+v", got)
}
oversized := `[{"name":"disk","identifiers":[` + strings.Repeat(`"disk-a",`, 300) + `"disk-b"]}]`
// Duplicate catalogue entries do not consume the graph bound.
client = newLegacyRESTReportingClient(t, alertArgsTransport{"/api/v2.0/reporting/graphs": {body: oversized}})
if graphs, err := client.legacyReportingGraphs(context.Background()); err != nil || len(graphs) != 6 {
t.Fatalf("catalogue deduplication: %v %v", graphs, err)
}
client.Close()
}
func TestCORE13CatalogueRejectsUniqueOverflow(t *testing.T) {
identifiers := make([]string, 253)
for i := range identifiers {
identifiers[i] = fmt.Sprintf("disk-%d", i)
}
raw, err := json.Marshal([]map[string]any{{"name": "disk", "identifiers": identifiers}})
if err != nil {
t.Fatal(err)
}
client := newLegacyRESTReportingClient(t, alertArgsTransport{"/api/v2.0/reporting/graphs": {body: string(raw)}})
defer client.Close()
if graphs, err := client.legacyReportingGraphs(context.Background()); err == nil || graphs != nil {
t.Fatal("oversized unique catalogue became a partial host total")
}
}
func TestCORE13ARCAlignedWithFreeMemory(t *testing.T) {
replies := core13ReportingFixture(t)
history := parseSystemMetricHistory(replies)
history.ARCSizeBytes = []TimeSeriesPoint{{Timestamp: time.Unix(1789000000-10, 0), Value: 8 << 30}}
system := systemInfoFromMetricHistory(history)
if system.ARCSizeBytes != 0 {
t.Fatal("old ARC subtracted from newer memory")
}
history.ARCSizeBytes = append(history.ARCSizeBytes, TimeSeriesPoint{Timestamp: time.Unix(1789000000, 0), Value: 8 << 30})
system = systemInfoFromMetricHistory(history)
system.MemoryTotalBytes = 32 << 30
if system.ARCSizeBytes != 8<<30 {
t.Fatal("same-bucket ARC lost")
}
points := systemMemoryPercentHistory(history, system.MemoryTotalBytes)
want := metricsFromTrueNASSystem(*system, 0, 0).Memory.Percent
assertCOREPoint(t, points, 0, 1789000000, want)
history.ARCSizeBytes[1].Value = 64 << 30
points = systemMemoryPercentHistory(history, system.MemoryTotalBytes)
assertCOREPoint(t, points, 0, 1789000000, 0)
}
func TestCORE13CoarseStepDoesNotMakeOldRowsCurrent(t *testing.T) {
response := trueNASReportingGetDataResponse{Name: "cpu", Legend: []string{"usage"}, Start: 1789000000, End: 1789000000, Step: 3600, Data: []any{[]any{12.}}}
if history := parseSystemMetricHistory(liveReportingResponses([]trueNASReportingGetDataResponse{response}, 1789000400)); history != nil {
t.Fatal("coarse step made an old CPU sample current")
}
if history := parseSystemMetricHistory([]trueNASReportingGetDataResponse{response}); history == nil {
t.Fatal("live age filtering destroyed native History")
}
}
func TestCORE13ARCOnlyDoesNotInventMemoryUsage(t *testing.T) {
history := &SystemMetricHistory{ARCSizeBytes: []TimeSeriesPoint{{Timestamp: time.Unix(1789000000, 0), Value: 8 << 30}}}
system := systemInfoFromMetricHistory(history)
if system.ARCSizeBytes != 8<<30 || system.Telemetry.Memory {
t.Fatal("ARC-only telemetry lost or claimed free RAM")
}
system.MemoryTotalBytes = 32 << 30
if metricsFromTrueNASSystem(*system, 0, 0).Memory != nil {
t.Fatal("ARC-only reading invented usage")
}
}
func TestCORE13DiskTemperatureHistoryExternalTiming(t *testing.T) {
response := trueNASReportingGetDataResponse{Name: "disktemp", Identifier: "disk-a", Legend: []string{"temperature"}, Start: 1789000000, End: 1789000020, Step: 10, Data: []any{[]any{40.}, []any{nil}, []any{42.}}}
points := parseReportingDiskTemperatureHistory([]trueNASReportingGetDataResponse{response})["disk-a"]
assertCOREPoint(t, points, 0, response.Start, 40)
assertCOREPoint(t, points, 1, response.Start+20, 42)
}
+3
View File
@@ -369,6 +369,9 @@ func (p *Provider) SystemMetricHistory(ctx context.Context, duration time.Durati
if len(nativeHistory.DiskWriteRate) > 0 {
metricMap["diskwrite"] = cloneTimeSeriesPoints(nativeHistory.DiskWriteRate)
}
if temperatures := systemTemperatureHistory(nativeHistory.TemperatureCelsius); len(temperatures) > 0 {
metricMap["temperature"] = temperatures
}
if len(metricMap) == 0 {
return resourceID, nil, nil
}
@@ -221,6 +221,9 @@ func TestRESTReportingDefaultAggregationsPreserveSamples(t *testing.T) {
var calls int
client := newLegacyRESTReportingClient(t, nil)
client.httpClient.Transport = reportingAggregationRESTFunc(func(r *http.Request) (*http.Response, error) {
if r.URL.Path == "/api/v2.0/reporting/graphs" {
return (alertArgsTransport{}).RoundTrip(r)
}
calls++
var request struct {
Graphs []map[string]any `json:"graphs"`
+266
View File
@@ -0,0 +1,266 @@
[
{
"name": "cpu",
"identifier": null,
"data": [
[
0.033358538468,
4.4805217137,
7.8754519386,
7.8754519386,
100.0
],
[
0.0,
0.0,
0.0,
0.0,
0.0
],
[
0.0,
0.0,
0.0,
0.0,
0.0
]
],
"start": 1789000000,
"end": 1789000020,
"step": 10,
"legend": [
"interrupt",
"system",
"user",
"nice",
"idle"
],
"aggregations": {
"min": [
0.0,
0.0,
0.0,
0.0,
0.0
],
"mean": [
0.019921323668173407,
1.734965537597784,
2.9715496785945983,
2.9715496785945983,
99.44598337950139
],
"max": [
0.088098918119,
4.7359432381,
8.3218063112,
8.3218063112,
100.0
]
}
},
{
"name": "cputemp",
"identifier": null,
"data": [
[
29.95,
29.95,
30.95,
30.95,
30.95,
30.95,
29.95,
29.95
],
[
null,
null,
null,
null,
null,
null,
null,
null
]
],
"start": 1789000000,
"end": 1789000010,
"step": 10,
"legend": [
"cputemp0",
"cputemp1",
"cputemp2",
"cputemp3",
"cputemp4",
"cputemp5",
"cputemp6",
"cputemp7"
],
"aggregations": {
"min": [
29.95,
29.95,
30.95,
30.95,
30.95,
30.95,
29.95,
29.95
],
"mean": [
29.95,
29.95,
30.95,
30.95,
30.95,
30.95,
29.95,
29.95
],
"max": [
29.95,
29.95,
30.95,
30.95,
30.95,
30.95,
29.95,
29.95
]
}
},
{
"name": "memory",
"identifier": null,
"data": [
[
188961888.74,
1918028677.6,
29016502272.0,
1335296.0,
2178945024.0
],
[
0.0,
0.0,
0.0,
0.0,
0.0
]
],
"start": 1789000000,
"end": 1789000010,
"step": 10,
"legend": [
"memory-active_value",
"memory-inactive_value",
"memory-wired_value",
"memory-laundry_value",
"memory-free_value"
],
"aggregations": {
"min": [
0.0,
0.0,
0.0,
0.0,
0.0
],
"mean": [
189003434.29373962,
1911812503.3603878,
28935538170.930748,
1331597.1191135733,
2179191845.1520777
],
"max": [
193937266.75,
1921061077.8,
29038160280.0,
1335296.0,
2199964529.7
]
}
},
{
"name": "disk",
"identifier": "disk-a",
"data": [
[
484.54070578,
538546.26401
],
[
0.0,
1276361.9904
],
[
null,
null
]
],
"start": 1789000000,
"end": 1789000020,
"step": 10,
"legend": [
"disk_octets_read",
"disk_octets_write"
],
"aggregations": {
"min": [
0.0,
383856.71231
],
"mean": [
336.1788029753333,
1030387.1104943056
],
"max": [
2563.1187967,
3593935.6719
]
}
},
{
"name": "interface",
"identifier": "nic-a",
"data": [
[
9728.2270494,
45551.708899,
9728.2270494
],
[
null,
null,
null
]
],
"start": 1789000000,
"end": 1789000010,
"step": 10,
"legend": [
"rx",
"tx",
"overlap"
],
"aggregations": {
"min": [
3905.0068688,
2298.9182121,
2298.9182121
],
"mean": [
11905.4645962075,
16747.828344191388,
8876.474186066389
],
"max": [
61200.017929,
86660.554326,
61200.017929
]
}
}
]
+1
View File
@@ -204,6 +204,7 @@ type TimeSeriesPoint struct {
// SystemMetricHistory stores provider-native TrueNAS system history before it
// is normalized onto the canonical monitoring guest-chart surface.
type SystemMetricHistory struct {
TemperatureCelsius map[string][]TimeSeriesPoint
CPUPercent []TimeSeriesPoint
MemoryPercent []TimeSeriesPoint
MemoryUsedBytes []TimeSeriesPoint
@@ -229,6 +229,7 @@ class CanonicalCompletionGuardTest(unittest.TestCase):
"internal/monitoring/docker_metric_presence_test.go",
"internal/monitoring/monitor_host_agent_removal_lifecycle_test.go",
"internal/monitoring/monitor_host_agents_test.go",
"internal/monitoring/monitor_notification_startup_test.go",
"internal/monitoring/physical_disk_roundtrip_test.go",
"scripts/installtests/agent_state_dir_lifecycle_test.go",
"scripts/installtests/install_ps1_test.go",
@@ -321,6 +322,7 @@ class CanonicalCompletionGuardTest(unittest.TestCase):
"internal/monitoring/monitor_host_agent_removal_lifecycle_test.go",
"internal/monitoring/monitor_host_agents_test.go",
"internal/monitoring/monitor_mock_alerts_test.go",
"internal/monitoring/monitor_notification_startup_test.go",
"internal/monitoring/monitor_pve_cluster_refresh_test.go",
"internal/monitoring/monitor_pve_guest_lxc_test.go",
"internal/monitoring/physical_disk_roundtrip_test.go",
@@ -466,6 +468,7 @@ class CanonicalCompletionGuardTest(unittest.TestCase):
"internal/monitoring/docker_metric_presence_test.go",
"internal/monitoring/monitor_host_agent_removal_lifecycle_test.go",
"internal/monitoring/monitor_host_agents_test.go",
"internal/monitoring/monitor_notification_startup_test.go",
"internal/monitoring/physical_disk_roundtrip_test.go",
"scripts/installtests/agent_state_dir_lifecycle_test.go",
"scripts/installtests/install_ps1_test.go",
@@ -4337,6 +4337,7 @@ class SubsystemLookupTest(unittest.TestCase):
"internal/monitoring/docker_metric_presence_test.go",
"internal/monitoring/monitor_host_agent_removal_lifecycle_test.go",
"internal/monitoring/monitor_host_agents_test.go",
"internal/monitoring/monitor_notification_startup_test.go",
"internal/monitoring/physical_disk_roundtrip_test.go",
"scripts/installtests/agent_state_dir_lifecycle_test.go",
"scripts/installtests/install_ps1_test.go",
@@ -20,7 +20,10 @@ SECRET = "synthetic-docker-docs-secret"
def section(heading):
guide = (ROOT / "docs/UNIFIED_AGENT.md").read_text()
return guide.split("### " + heading + "\n", 1)[1].split("\n### ", 1)[0]
marker = "### " + heading + "\n"
if marker not in guide:
raise AssertionError(f"missing diagnostic section: {heading}")
return guide.split(marker, 1)[1].split("\n### ", 1)[0]
def command(heading):
@@ -106,6 +109,52 @@ class DockerTroubleshootingDocsTest(unittest.TestCase):
self.assertIn(operation, recipe)
self.assertIn(operation, source)
def test_command_channel_guidance_does_not_confuse_scope_with_admission(self):
text = section("Commands enabled but remote control blocked")
for required in (
"no admitted command channel connected", "not a connected session",
"Automatic updates ready", "Changing Proxmox API permissions cannot",
"locally on the affected node", "`dial websocket`", "`registration failed`",
"No matching entry is\ninconclusive", "not the full journal",
"Omit tokens, cookies, URLs, hostnames, addresses", "Keep saved identity and credentials intact",
"do not delete state, loosen TLS", "commands disabled remains an alternative",
):
with self.subTest(required=required):
self.assertIn(required, text)
for label in ("Connected and registered with Pulse command server",
"WebSocket connection failed repeatedly, reconnecting"):
self.assertIn(label, text)
self.assertIn(label, (ROOT / "internal/hostagent/commands.go").read_text())
policy = (ROOT / "internal/api/connections_aggregator.go").read_text()
self.assertIn("agent reports command execution enabled, but no admitted command channel is connected", policy)
guest = section("Docker visible in one LXC but missing in another")
self.assertIn("#commands-enabled-but-remote-control-blocked", guest)
self.assertIn("do not repeat the guest probe", guest)
self.assertNotRegex(text, r"--property=Environment|systemctl (?:cat|restart)|--insecure")
def test_command_channel_journal_check_is_bounded_and_read_only(self):
recipe = command("Commands enabled but remote control blocked")
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
for name, body in (
("sudo", '#!/bin/sh\nexec "$@"\n'),
("journalctl", '#!/usr/bin/python3\nimport json, sys\nfrom pathlib import Path\n'
'Path("' + str(root / "argv.json") + '").write_text(json.dumps(sys.argv[1:]))\n'
'print("WebSocket connection failed repeatedly, reconnecting")\n'),
):
path = root / name
path.write_text(body)
path.chmod(0o700)
env = dict(os.environ, PATH=f"{root}:{os.environ['PATH']}", SYNTHETIC_SECRET=SECRET)
result = subprocess.run(["bash", "-eu", "-c", recipe], env=env,
capture_output=True, text=True, timeout=5)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(json.loads((root / "argv.json").read_text()), [
"-u", "pulse-agent.service", "--since", "15 minutes ago", "-n", "200",
"--no-pager", "--output=cat",
])
self.assertNotIn(SECRET, result.stdout + result.stderr)
def run_guest(self, *, socket="present", docker="ok", pct="ok", cli=True):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
@@ -0,0 +1,155 @@
#!/usr/bin/env python3
"""Exercise copied notification-log recipes without a service or destination.
Synthetic readers prove bounds, stream visibility and exit handling. They do
not prove native installation, notification delivery or automatic redaction.
"""
from __future__ import annotations
import json
import os
from pathlib import Path
import re
import subprocess
import tempfile
import unittest
ROOT = Path(__file__).resolve().parents[2]
DOC = ROOT / "docs/TROUBLESHOOTING.md"
SMTP_ERROR = '{"level":"error","message":"SMTP authentication failed"}\n'
WEBHOOK_ERROR = '{"level":"error","message":"webhook returned HTTP 429"}\n'
PARTIAL_EMAIL_LOG = '{"message":"attempting to send email"}\n'
SYNTHETIC_SECRET = "synthetic-provider-echo-secret"
def section(heading: str, level: int = 3) -> str:
marker = "#" * level + " " + heading + "\n"
text = DOC.read_text(encoding="utf-8")
if marker not in text:
raise AssertionError(f"missing diagnostic section: {heading}")
# These guide sections have no later H1. Ignore the single-# shell comments
# inside recipes while stopping at the next same/higher-level section.
return re.split(r"\n#{2," + str(level) + r"} ", text.split(marker, 1)[1], 1)[0]
def recipes() -> dict[str, str]:
blocks = re.findall(r"```bash\n(.*?)```", section("Inspect Notification Logs"), re.DOTALL)
if len(blocks) != 2:
raise AssertionError("expected one recipe per log reader")
return {"docker" if "# Docker" in block else "journalctl": block for block in blocks}
READER = '''#!/usr/bin/env python3
import json, os, pathlib, sys
pathlib.Path(os.environ['READER_ARGV']).write_text(json.dumps(sys.argv[1:]))
sys.stdout.write(os.environ.get('LOG_STDOUT', ''))
sys.stderr.write(os.environ.get('LOG_STDERR', ''))
sys.exit(int(os.environ.get('READER_EXIT', '0')))
'''
class NotificationTroubleshootingDocsTest(unittest.TestCase):
def exercise(self, reader: str, *, stdout: str = "", stderr: str = "",
exit_code: int = 0, recipe: str | None = None):
with tempfile.TemporaryDirectory() as temporary:
directory = Path(temporary)
command = directory / reader
command.write_text(READER, encoding="utf-8")
command.chmod(0o700)
argv = directory / "argv.json"
env = dict(os.environ, PATH=f"{directory}:{os.environ['PATH']}",
READER_ARGV=str(argv), LOG_STDOUT=stdout, LOG_STDERR=stderr,
READER_EXIT=str(exit_code))
copied = recipe if recipe is not None else recipes()[reader]
result = subprocess.run(["bash", "-c", copied], env=env,
capture_output=True, text=True, timeout=5)
self.assertTrue(argv.exists(), "copied command did not invoke its log reader")
return result, json.loads(argv.read_text())
def test_shipped_document_matches_the_tested_source(self):
self.assertEqual(DOC.read_bytes(),
(ROOT / "frontend-modern/public/docs/TROUBLESHOOTING.md").read_bytes())
def test_copied_commands_bound_the_read_without_filtering_or_mutation(self):
expected = {
"journalctl": ["-u", "pulse", "--since", "15 minutes ago", "--lines", "200", "--no-pager"],
"docker": ["logs", "--since", "15m", "--tail", "200", "pulse"],
}
self.assertEqual(set(recipes()), set(expected))
for reader, copied in recipes().items():
with self.subTest(reader=reader):
result, argv = self.exercise(reader, stdout=SMTP_ERROR + WEBHOOK_ERROR)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(argv, expected[reader])
self.assertEqual(result.stdout, SMTP_ERROR + WEBHOOK_ERROR)
self.assertNotRegex(copied, r"[|<>]|\b(?:grep|curl|inspect|restart|printenv)\b|--follow|--token")
def test_docker_application_logs_on_both_streams_remain_visible(self):
result, _ = self.exercise("docker", stdout=WEBHOOK_ERROR, stderr=SMTP_ERROR)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(result.stdout, WEBHOOK_ERROR)
self.assertEqual(result.stderr, SMTP_ERROR)
def test_reader_failure_keeps_its_exit_even_after_a_partial_email_log(self):
for reader in recipes():
for output in ("", PARTIAL_EMAIL_LOG):
with self.subTest(reader=reader, partial=bool(output)):
result, _ = self.exercise(reader, stdout=output,
stderr="synthetic reader access failure\n", exit_code=2)
self.assertEqual(result.returncode, 2)
self.assertEqual(result.stdout, output)
self.assertEqual(result.stderr, "synthetic reader access failure\n")
def test_successful_empty_read_is_not_documented_as_healthy_delivery(self):
for reader in recipes():
result, _ = self.exercise(reader)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(result.stdout + result.stderr, "")
self.assertIn("successful empty read is inconclusive", section("Inspect Notification Logs"))
def test_raw_provider_echo_is_not_claimed_to_be_sanitised(self):
# The command does not magically sanitise either stream. The nearby
# disclosure boundary must survive edits to the copied recipe.
for reader in recipes():
result, _ = self.exercise(reader, stderr=SYNTHETIC_SECRET + "\n")
self.assertIn(SYNTHETIC_SECRET, result.stderr)
text = " ".join(section("Inspect Notification Logs").split())
for boundary in ("not sanitised", "Do not post them wholesale", "manually redacted",
"anything echoed by the provider", "Never upload full environments",
"queue database", "inside the Pulse container", "actual service or container name"):
self.assertIn(boundary, text)
def test_legacy_email_pipeline_is_an_adverse_control(self):
legacy = "docker logs pulse | grep email"
result, argv = self.exercise("docker", stderr=SMTP_ERROR, recipe=legacy)
self.assertEqual(argv, ["logs", "pulse"])
self.assertEqual(result.returncode, 1)
self.assertEqual(result.stdout, "")
# A human can see stderr in a terminal, but the old search itself loses
# the match, encouraging collection of more history or debug output.
self.assertEqual(result.stderr, SMTP_ERROR)
result, _ = self.exercise("docker", stdout=PARTIAL_EMAIL_LOG,
stderr="synthetic reader access failure\n", exit_code=2, recipe=legacy)
self.assertEqual(result.returncode, 0, "legacy pipe must demonstrate the masked reader failure")
def test_recovery_guidance_uses_existing_ui_without_promising_delivery(self):
presentation = (ROOT / "frontend-modern/src/utils/alertDestinationsPresentation.ts").read_text()
notifications = " ".join(section("Notifications").split())
for label in ("Recent delivery activity", "Retry retained deliveries", "Dismiss retained failures"):
self.assertIn(label, presentation)
self.assertIn(label, notifications)
for boundary in ("skips the persistent delivery queue", "not listed", "Notifications are paused",
"minimum alert severity and tag filters", "held-notification reasons",
"unavailable", "may receive a duplicate", "successful test does not itself retry",
"do not disable verification", "Do not delete `notification_queue.db`",
"do not post it wholesale", "enable debug logging just to collect it"):
self.assertIn(boundary, notifications)
for heading in ("Emails not sending", "Webhooks failing"):
self.assertIn("#recover-retained-delivery-failures", section(heading, 4))
self.assertIn("#inspect-notification-logs", section(heading, 4))
if __name__ == "__main__":
unittest.main()