Merge remote-tracking branch 'origin/main' into refactor/drop-unread-patrol-state-members

This commit is contained in:
rcourtman
2026-10-08 00:34:24 +01:00
35 changed files with 1699 additions and 125 deletions
+7 -4
View File
@@ -836,9 +836,12 @@ delivery for all resources.
- `GET /api/admin/reports/catalog` (admin, `settings:read`)
- Returns the canonical reporting catalog for the settings surface, including locked-state teaser copy, enabled-surface guidance copy, performance-report options, canonical single-report filename subject, canonical fallback filename date style, and the nested VM inventory export definition.
- Metadata route: readable without the `advanced_reporting` feature so locked admin surfaces can render the same reporting definition before upsell.
- `GET /api/admin/reports/generate` (admin, `settings:read`)
- Query params: `format` (pdf/csv, default `pdf`), `resourceType`, `resourceId`, `metricType` (optional), `start`/`end` (RFC3339, optional; defaults to last 24h), `title` (optional)
- `POST /api/admin/reports/generate` (admin, `settings:read`)
- Body fields: `format` (pdf/csv, default `pdf`), `resourceType`, `resourceId`, `metricType` (optional), `start`/`end` (RFC3339, optional; defaults to last 24h), `title` (optional)
- If `title` is omitted, the backend applies the canonical default title for that resource report.
- When Pulse Assistant is configured, the configured AI provider narrates a PDF's executive summary; each narration counts against the AI budget and appears in AI usage. If the call fails, times out or the budget is spent, the PDF falls back to the deterministic summary. CSV output is never narrated. The settings UI uses this method.
- `GET /api/admin/reports/generate` (admin, `settings:read`)
- Takes the same fields as query params and returns the same report, except that it never calls an AI provider: a PDF carries the deterministic summary and a note saying AI narration needs `POST`. Other methods return `405` with `Allow: GET, POST`.
- `POST /api/admin/reports/generate-multi` (admin, `settings:read`)
- Body fields: `resources` (1-50 entries of `{resourceType,resourceId}`), `format`, `metricType` (optional), `start`/`end` (RFC3339, optional; defaults to last 24h), `title` (optional)
- If `title` is omitted, the backend applies the canonical default fleet report title.
@@ -851,8 +854,8 @@ Validation and limits:
- Malformed `start`/`end` values, `end` not strictly after `start`, or report windows over 366 days return `400 invalid_time_range`.
- `metricType` must match `[a-zA-Z0-9._:-]+` and be <= 64 chars, otherwise `400 invalid_metric_type`.
- `title` must be <= 256 chars, otherwise `400 invalid_title`.
- Multi-report body max size is 1MB; oversized payloads return `400 body_too_large`.
- Multi-report bodies reject trailing payload and unknown JSON fields with `400 invalid_body`.
- Report body max size is 1MB; oversized payloads return `400 body_too_large`.
- Report bodies reject trailing payload and unknown JSON fields with `400 invalid_body`.
- VM inventory export only accepts `csv`.
Common reporting error codes:
+4 -2
View File
@@ -260,10 +260,12 @@ Each client runtime (or organization) generates its own reports, scoped to
that client's resources:
- **UI**: Settings → Data & Reports.
- **API**: `GET /api/admin/reports/generate` (single resource) and
- **API**: `POST /api/admin/reports/generate` (single resource) and
`POST /api/admin/reports/generate-multi` (up to 50 resources per report),
returning PDF or CSV. In shared-process mode, scope with `X-Pulse-Org-ID`
or an org-bound token.
or an org-bound token. `GET /api/admin/reports/generate` with query
parameters still returns a single-resource report, but never adds the
Pulse Assistant narrative.
- **Schedules**: `GET`/`POST /api/admin/reports/schedules`,
`PUT`/`DELETE /api/admin/reports/schedules/{id}`, and
`POST /api/admin/reports/schedules/{id}/run`. Schedules can target explicit
@@ -7665,6 +7665,16 @@ bounded and classified temperature nodes, and leaves the host report intact
when the helper is absent or incompatible. It does not treat Windows ACPI
thermal zones as hardware sensor evidence, accept remote provider locations,
or grant Pulse hardware-control authority.
### Single-resource report AI narration requires POST
`internal/api/metrics_reporting_handlers.go`, `internal/api/router.go` and
`internal/api/ai_handlers.go` changed only how `/api/admin/reports/generate`
reads its fields (a `POST` JSON body or `GET` query parameters), which of those
may use the tenant's AI narrator (`POST` only), and how a `GET` finds Patrol
findings (`ExistingAIService`, which never constructs a tenant AI service). No
agent registration, enrolment, install, update, removal or report ingest path
changed.
### Agent profile adoption is reported as a count only
The usage telemetry snapshot reads `LoadAgentProfiles` and reports
@@ -7535,6 +7535,18 @@ prompt budget. Reporting is therefore an additive consumer of AI
runtime, not a new ownership boundary, and the narrator/findings
surfaces inherit the same governance the rest of the canonical AI
runtime already enforces.
Report narration is caller-requested spend, so on the single-resource HTTP
route only `POST /api/admin/reports/generate` reaches it (fleet reports, from
`POST .../generate-multi` or a report schedule, use the fleet narrator). The
reporting narrator resolver goes through `AISettingsHandler.GetAIService`,
which constructs a tenant service on first use and can then list provider
models or start background discovery. A GET single-resource report never calls
that resolver:
it narrates deterministically and reads Patrol findings through
`AISettingsHandler.ExistingAIService`, which returns only a service that is
already running and never constructs one
(`TestAISettingsHandler_ExistingAIServiceNeverConstructs`, and through the
router's real wiring `TestContract_SingleReportGETDoesNotConstructTenantAIService`).
The same canonical AI runtime now also owns the fleet-level report
narrative through `report_fleet_narrator.go`. `Service` implements
@@ -9692,6 +9692,32 @@ narrator must fail closed: nil provider, parse failure, timeout, or
empty response causes the engine to fall back to the heuristic
narrative without surfacing the AI failure to the caller, so reporting
is never blocked by AI availability.
Only `POST /api/admin/reports/generate` may hand the tenant's AI narrator to
the engine, because a narrated PDF makes a paid provider call and appends a
`cost.UsageEvent` to the cost ledger. The single-resource handler accepts
`POST` with a JSON body (the `generate-multi` decoding rules: 1MB cap, unknown
fields and trailing payload rejected, through `decodeReportingRequestBody`) and
`GET` with the same fields as query parameters, and answers any other method
with `405` and `Allow: GET, POST`. GET never calls `resolveNarrator`, whose
`GetAIService` can construct a tenant AI service (and with it list provider
models or start background discovery): it narrates through `getReportNarrator`,
the heuristic summary whose `Disclaimer` replaces the "Configure Pulse
Assistant" tip with a note that AI narration needs POST, and reads Patrol
findings through `SetExistingFindingsResolver`, which `router.go` backs with
`AISettingsHandler.ExistingAIService` so only an already running service is
used. GET passes the demo-mode guard and the CSRF check, and a SameSite=Lax
session cookie rides a cross-site top-level GET navigation, so a link must not
be able to spend AI budget. The reporting catalog advertises the transport as
`performanceReport.singleResourceMethod` (`POST`), and the settings UI follows
it. `generate-multi` and the manual schedule run endpoint
(`POST /api/admin/reports/schedules/{id}/run`) were already POST-only; due
schedules run from the server's scheduler, not an HTTP request.
`TestContract_SingleReportAINarrationRequiresPOST` pins the rule on a real
engine: GET, HEAD, PUT and DELETE make no narrator call and no AI service
resolution, while POST makes one of each.
`TestContract_SingleReportGETDoesNotConstructTenantAIService` drives
`Router.wireReportingAIResolvers` with an uncached tenant: GET constructs no
tenant AI service and POST resolves one.
Report alerts keep the alert engine's resolution. The node, VM and
container enrichers in `internal/api/metrics_reporting_handlers.go` build
every row through `reportActiveAlertInfo` and `reportResolvedAlertInfo`,
@@ -6101,6 +6101,14 @@ summaries whose scope, format, and attachments the server fixes. The delivery
surface (`/api/admin/reports/schedules`) and tenant-local storage are
unchanged.
### Provider docs name POST for single-resource reports
`docs/MSP.md` and its shipped copy now give `POST /api/admin/reports/generate`
for per-client single-resource reports and say a `GET` with query parameters
still returns the report without the Pulse Assistant narrative, matching the
server, which lets only `POST` narrate. Install, upgrade and provider bundle
delivery are unchanged.
### Convergence log-reader compatibility (5 September 2026)
Scheduled reconciliation run 33939926237 failed when `gh api` refused terminal
@@ -4849,6 +4849,13 @@ performance-report transport (`/api/reporting` and `/api/reporting/generate-mult
so the reporting panel does not go dead on mixed-version installs, but that
compatibility path is intentionally report-only and must not invent the newer
catalog-owned VM inventory export surface.
`reportingPanelModel.ts` picks the single-resource transport from the catalog's
`performanceReport.singleResourceMethod`: `POST` sends the request fields as a
JSON body, which is the only transport the server lets narrate a PDF with Pulse
Assistant, and `GET` sends the same fields as query parameters.
`reportingCatalogModel.ts` reads a missing field as `GET`, the only method a
backend that predates the field accepts, and the legacy compatibility catalog
names `GET` for `/api/reporting`. Fleet reports stay `POST` on every backend.
`ReportingPanel.tsx` must therefore treat `vmInventoryExport` as optional when
it renders a governed reporting catalog. A legacy compatibility catalog with no
inventory export still owns a valid enabled reporting surface and must continue
@@ -2699,7 +2699,18 @@ truthfulness, not native thaw, containing-release or workload acceptance.
ingestion so canonical host IDs stay stable across restarts (see the
unified-resources contract's durable identity-pin obligation). Rebuild
paths added to the adapter must keep that persistence step; ephemeral
snapshot-bridge adapters stay read-only.
snapshot-bridge adapters stay read-only. The rebuild, the live
supplemental refresh and the read-state overlay ingest records with the
adapter's configured stale thresholds, because record ingest joins
operator links and the freshness gate of every metric merge, a link's or
a source's into an existing row, reads them (unified-resources contract,
"Operator links reach record-ingested resources"). Regression
coverage:
`TestMonitorAdapterJoinsLinkedRecordsWithConfiguredStaleThresholds` and
`TestMonitorAdapterSourceMergesUseConfiguredStaleThresholds` in
`internal/unifiedresources/monitor_adapter_read_state_test.go` and
`TestManualLinkToSupplementalGuestHoldsWithAndWithoutContinuity` in
`internal/monitoring/issue1913_host_continuity_test.go`.
11. The TrueNAS provider projects pools with `Storage.Topology` fixed to
`pool` and the ZFS data vdev layout in `Storage.VDevLayout`. The
@@ -3025,6 +3025,18 @@ Removing the disconnected recorder does not alter alert, action approval or
operator authority. An unrelated organization receives no default archive
fallback.
### Report GETs cannot spend AI budget
`Router.wireReportingAIResolvers` in `internal/api/router.go` wires a second
reporting resolver, `SetExistingFindingsResolver`, backed by
`AISettingsHandler.ExistingAIService`.
A `GET /api/admin/reports/generate` uses only that resolver and the
deterministic narrator, so it makes no AI provider call, writes no cost-ledger
entry and never constructs a tenant AI service. GET passes the demo-mode guard
and the CSRF check, and a SameSite=Lax session cookie rides a cross-site
top-level navigation, so AI narration of a single-resource report now requires
a CSRF-checked `POST`. No authentication, scope or session rule changed.
### Node setup action v7 compatibility
Workflow consumers pin `actions/setup-node` to
@@ -6265,6 +6265,19 @@ is in `applyLicensedFeatureConfigSnapshot`
(`pkg/server/telemetry_licensed_features.go`), pinned by
`TestApplyLicensedFeatureConfigSnapshot_CountsScheduledReportingAndProfiles`.
### Single-resource report AI narration requires POST
`internal/api/metrics_reporting_handlers.go` now accepts `POST` with a JSON
body for `/api/admin/reports/generate` and lets only that method use the
tenant's AI narrator; `GET` keeps working with query parameters and the
deterministic summary. A narrated PDF records the same AI cost-ledger entry as
before, and only a `POST` can now produce one. A `GET` reads Patrol findings
only from an AI service that is already running
(`AISettingsHandler.ExistingAIService`), so it no longer constructs a tenant AI
service, whose construction opens that tenant's SQLite resource export store.
Report content, backup and recovery enrichment, and tenant storage are
otherwise unchanged.
### Audit-read activity is a bounded, content-free local history
`audit_read_activity.json` follows the same shape as the existing external-agent
@@ -1609,7 +1609,12 @@ Source freshness is a separate unified-resource status input, not a fixed
global health timer. Snapshot rebuilds, resource seeding, supplemental record
overlays, and cloned read-state overlays must preserve the monitoring-provided
stale thresholds for Proxmox, PBS, and PMG sources so resources do not flap to
warning/degraded between successful configured poll cycles.
warning/degraded between successful configured poll cycles. A source merge
into an existing row judges metric freshness by the ingest's thresholds and
the stale pass's own threshold rule, so the merge's freshness gate never
calls a source current that the stale pass marks stale, or the reverse. The
gate decides only where no stronger rule does: a hypervisor-managed guest
keeps its platform CPU even while that source is stale.
Service-discovery readiness is a unified-resource payload contract, not a
drawer-local decoration. Resource list/detail payloads that expose a
@@ -2813,7 +2818,9 @@ application resource-provider or WebSocket lifecycle.
ID and drops the superseded pin row (`ApplyCanonicalIDSuccessions`
in `internal/unifiedresources/canonical_id_succession.go`).
Successions never fire while the old ID still belongs to a live
resource, never follow a contradicting machine key (a reinstalled
resource or to one a manual link folded into its primary (see
"Operator links reach record-ingested resources"), never follow a
contradicting machine key (a reinstalled
machine mints fresh, it does not absorb the old host's operator
state), and never rewrite change-journal rows. Known limitation:
machine-keyless hosts whose history already merged under a collapsed
@@ -2894,8 +2901,9 @@ application resource-provider or WebSocket lifecycle.
`ResourceRegistry.IngestRecords` applies the same
`ApplyCanonicalIDSuccessions` semantics as pin-driven successions
(operator state and action audits re-key, the superseded pin row
drops, never while the old ID belongs to a live resource, journal
rows never rewritten). Regression coverage:
drops, never while the old ID belongs to a live resource or one a
manual link folded into its primary, journal rows never rewritten).
Regression coverage:
`TestRegistryIngestRecordsKeepsSameHostnameSystemsDistinct` /
`TestIngestRecordsSucceedLegacyHostnameScopedCanonicalIDs` /
`TestIngestRecordsDoNotCompleteTrueNASIdentityFromPins` in
@@ -6596,6 +6604,75 @@ full heading. This shared drawer rule applies across resource types. The 390px
browser qualification checks heading fit and keyboard expansion; no data model,
authorisation, action dispatch or resource admission contract changes.
### Operator links reach record-ingested resources
Every registry ingest path applies the operator's manual links: snapshot
ingest, resource ingest and record ingest (`IngestRecords`,
`IngestRecordsWithStaleThresholds` and the read-state overlay's record pass).
Either side of a link can arrive only as a supplemental record, such as a
vSphere or TrueNAS VM with a Pulse agent inside, or an agent on a TrueNAS host,
and the monitor's rebuild ingests those records after the snapshot. While only
snapshot and resource ingest applied links, the rebuilt registry behind the
websocket broadcast and `/api/state` kept both rows. The resources API, which
seeds through resource ingest, showed them merged, and so did the published
inventory whenever an unrelated absent machine made continuity hydration
clone the registry. Mock mode builds its unified view from fixtures without
the link store, so links do not reach the mock broadcast; that boundary is
separate.
- The monitor adapter passes its configured stale thresholds to record
ingest, so a link joined there judges each side's metrics the way a
snapshot-time link does. Ordinary source merges into an existing row use
the same thresholds on every ingest path: snapshot and record ingest hold
the caller's thresholds while they run, and `mergeInto` judges each
metric's freshness with them through `sourceStaleThreshold`, the rule the
stale pass reads. Before this, those merges used the defaults, so a
Proxmox node polled every two minutes showed a silent auto-linked agent's
readings while the stale pass called the node's poll current, and a
supplemental refresh of a linked vSphere VM kept the agent's memory over
vSphere's current reading. Outside mock mode the monitor configures only
Proxmox, PBS and PMG; mock mode also configures TrueNAS, vSphere and
availability. Regression coverage:
`TestMonitorAdapterSourceMergesUseConfiguredStaleThresholds` (rebuild,
supplemental refresh, overlay, node with auto-linked agent),
`TestSourceMergesJudgeFreshnessByTheIngestStaleThresholds` and
`TestMetricMergeFreshnessMatchesTheStalePass`.
- Continuity records are the one record ingest that never joins links. A
saved enrollment only fills an absent machine, so it stays its own row
instead of lending its linked guest an offline verdict and an old agent
payload.
- A link merge keeps the TrueNAS and vSphere payloads the primary lacks, so
an agent chosen as primary over a TrueNAS system still carries the
system's facet, and the system's pools re-parent to the merged row.
- The registry remembers each resource a manual link folded into its
primary. A folded resource is still observed, so neither pin-driven nor
record-declared succession may succeed its ID. An agent-type primary
carries the folded resource's machine keys and is pinned with them, and
succeeding the folded ID re-keyed the link onto the primary itself, which
split the pair on the next rebuild. That already broke snapshot links whose
primary is a machine-keyless agent-type row, such as a Proxmox node.
- Fold history is registry-local: resource seeding and overlay clones do not
carry it. A reference to the folded resource's canonical ID, such as an
availability check configured against an agent later linked into its
guest, does not resolve to the primary, so the check loses its target in
the broadcast as it already did in the resources API.
`registry_merge_policy_test.go` pins the record links in both directions
through two rebuilds and the live supplemental refresh, and checks that a
registry seeded from the monitor's listing through resource ingest, as the
resources API seeds, lists the same rows
(`TestManualLinksJoinRecordIngestedResourcesInMonitorRebuild`). The same file
pins the node case (`TestManualLinkKeepsItsEndpointsThroughIdentityPinPersistence`),
record-declared succession (`TestRecordDeclaredSuccessionSparesALinkFoldedResource`)
and both payloads (`TestManualLinkKeepsProviderPayloadsThePrimaryLacks`).
`monitor_adapter_read_state_test.go`
(`TestMonitorAdapterJoinsLinkedRecordsWithConfiguredStaleThresholds`) pins the
thresholds on the rebuild, supplemental and overlay paths, and
`internal/monitoring/issue1913_host_continuity_test.go`
(`TestManualLinkToSupplementalGuestHoldsWithAndWithoutContinuity`) pins the
published inventory with no continuity, an unrelated absent machine and the
linked agent's own saved enrollment.
### Provider link network corroboration
An existing reciprocal host/provider link remains authoritative, including a
@@ -0,0 +1,42 @@
{
"version": 1,
"base_sha": "77d21197927a670076a74e87634e73ea999a566a",
"verified_at": "2026-10-07T19:05:00Z",
"result": "passed",
"changed_paths": [
"frontend-modern/src/components/Settings/reportingCatalogModel.ts",
"frontend-modern/src/components/Settings/reportingPanelModel.ts"
],
"content_sha256": {
"frontend-modern/src/components/Settings/reportingCatalogModel.ts": "fc3ab83f49ac62aaf7ef97b415a0bff1c4e445ca17771efc18304295e94e3df0",
"frontend-modern/src/components/Settings/reportingPanelModel.ts": "e4366a8c50ea897dcb4907b4e488a2c5f758e710d7a9be0d5fe0d3e047e2d9c5"
},
"routes": [
"/settings/support/reporting",
"/api/admin/reports/generate"
],
"viewports": [
{
"width": 1280,
"height": 800
},
{
"width": 390,
"height": 844
}
],
"states": [
"Pro dev licence active, Pulse Assistant enabled on a local stub OpenAI-compatible provider that logs each call, real admin session login (no admin bypass)",
"Data & Reports panel with one mock agent selected, then two",
"Downloaded single-resource PDF opened: POST copy carries the stub AI narrative and provenance footer; GET copy carries the deterministic summary and the new GET note in place of the Configure Pulse Assistant tip",
"AI usage summary after the run: report_narrative recorded only for the two UI POST downloads"
],
"interactions": [
"Generate Report (1 resource) as PDF at 1280x800 and 390x844: one POST /api/admin/reports/generate with a JSON body, 200 application/pdf, download saved, stub provider called once each",
"Generate Report (1 resource) as CSV at both widths: POST, 200 text/csv, stub provider not called",
"Generate Report (2 resources) as PDF at both widths: POST /api/admin/reports/generate-multi unchanged, 200 application/pdf",
"Top-level browser navigation to GET /api/admin/reports/generate?format=pdf with the session cookie: PDF downloaded, stub provider not called",
"HEAD, PUT, DELETE and PATCH return 405 with Allow GET, POST; a session POST without the CSRF header returns 403",
"No console errors on the reporting panel at either width"
]
}
+7 -4
View File
@@ -836,9 +836,12 @@ delivery for all resources.
- `GET /api/admin/reports/catalog` (admin, `settings:read`)
- Returns the canonical reporting catalog for the settings surface, including locked-state teaser copy, enabled-surface guidance copy, performance-report options, canonical single-report filename subject, canonical fallback filename date style, and the nested VM inventory export definition.
- Metadata route: readable without the `advanced_reporting` feature so locked admin surfaces can render the same reporting definition before upsell.
- `GET /api/admin/reports/generate` (admin, `settings:read`)
- Query params: `format` (pdf/csv, default `pdf`), `resourceType`, `resourceId`, `metricType` (optional), `start`/`end` (RFC3339, optional; defaults to last 24h), `title` (optional)
- `POST /api/admin/reports/generate` (admin, `settings:read`)
- Body fields: `format` (pdf/csv, default `pdf`), `resourceType`, `resourceId`, `metricType` (optional), `start`/`end` (RFC3339, optional; defaults to last 24h), `title` (optional)
- If `title` is omitted, the backend applies the canonical default title for that resource report.
- When Pulse Assistant is configured, the configured AI provider narrates a PDF's executive summary; each narration counts against the AI budget and appears in AI usage. If the call fails, times out or the budget is spent, the PDF falls back to the deterministic summary. CSV output is never narrated. The settings UI uses this method.
- `GET /api/admin/reports/generate` (admin, `settings:read`)
- Takes the same fields as query params and returns the same report, except that it never calls an AI provider: a PDF carries the deterministic summary and a note saying AI narration needs `POST`. Other methods return `405` with `Allow: GET, POST`.
- `POST /api/admin/reports/generate-multi` (admin, `settings:read`)
- Body fields: `resources` (1-50 entries of `{resourceType,resourceId}`), `format`, `metricType` (optional), `start`/`end` (RFC3339, optional; defaults to last 24h), `title` (optional)
- If `title` is omitted, the backend applies the canonical default fleet report title.
@@ -851,8 +854,8 @@ Validation and limits:
- Malformed `start`/`end` values, `end` not strictly after `start`, or report windows over 366 days return `400 invalid_time_range`.
- `metricType` must match `[a-zA-Z0-9._:-]+` and be <= 64 chars, otherwise `400 invalid_metric_type`.
- `title` must be <= 256 chars, otherwise `400 invalid_title`.
- Multi-report body max size is 1MB; oversized payloads return `400 body_too_large`.
- Multi-report bodies reject trailing payload and unknown JSON fields with `400 invalid_body`.
- Report body max size is 1MB; oversized payloads return `400 body_too_large`.
- Report bodies reject trailing payload and unknown JSON fields with `400 invalid_body`.
- VM inventory export only accepts `csv`.
Common reporting error codes:
+4 -2
View File
@@ -260,10 +260,12 @@ Each client runtime (or organization) generates its own reports, scoped to
that client's resources:
- **UI**: Settings → Data & Reports.
- **API**: `GET /api/admin/reports/generate` (single resource) and
- **API**: `POST /api/admin/reports/generate` (single resource) and
`POST /api/admin/reports/generate-multi` (up to 50 resources per report),
returning PDF or CSV. In shared-process mode, scope with `X-Pulse-Org-ID`
or an org-bound token.
or an org-bound token. `GET /api/admin/reports/generate` with query
parameters still returns a single-resource report, but never adds the
Pulse Assistant narrative.
- **Schedules**: `GET`/`POST /api/admin/reports/schedules`,
`PUT`/`DELETE /api/admin/reports/schedules/{id}`, and
`POST /api/admin/reports/schedules/{id}/run`. Schedules can target explicit
@@ -35,6 +35,7 @@ const baseCatalog = {
title: 'Performance Reports',
description: 'Historical performance reporting',
singleResourceEndpoint: '/api/admin/reports/generate',
singleResourceMethod: 'POST',
multiResourceEndpoint: '/api/admin/reports/generate-multi',
singleFilenamePrefix: 'report',
singleFilenameSubject: 'resource_id',
@@ -42,6 +42,7 @@ const basePerformanceReport = () => ({
title: 'Performance Reports',
description: 'Historical performance reporting',
singleResourceEndpoint: '/api/admin/reports/generate',
singleResourceMethod: 'POST',
multiResourceEndpoint: '/api/admin/reports/generate-multi',
singleFilenamePrefix: 'report',
singleFilenameSubject: 'resource_id',
@@ -453,6 +454,19 @@ describe('parseReportingPerformanceReportDefinition (via catalog.performanceRepo
});
});
it('throws when singleResourceMethod is neither GET nor POST', () => {
expectInvalid({
...baseCatalog(),
performanceReport: { ...basePerformanceReport(), singleResourceMethod: 'PUT' },
});
});
it('reads a missing singleResourceMethod as GET, the only method older backends accept', () => {
const { singleResourceMethod: _omitted, ...withoutMethod } = basePerformanceReport();
const parsed = parseReportingCatalog({ ...baseCatalog(), performanceReport: withoutMethod });
expect(parsed.performanceReport.singleResourceMethod).toBe('GET');
});
it('throws when multiResourceEndpoint is not a string', () => {
expectInvalid({
...baseCatalog(),
@@ -22,6 +22,7 @@ const baseCatalogPayload = {
title: 'Performance Reports',
description: 'Historical performance reporting',
singleResourceEndpoint: '/api/admin/reports/generate',
singleResourceMethod: 'POST',
multiResourceEndpoint: '/api/admin/reports/generate-multi',
singleFilenamePrefix: 'report',
singleFilenameSubject: 'resource_id',
@@ -74,6 +75,7 @@ describe('reporting catalog model', () => {
const catalog = buildLegacyReportingCatalogFallback();
expect(catalog.performanceReport.singleResourceEndpoint).toBe('/api/reporting');
expect(catalog.performanceReport.singleResourceMethod).toBe('GET');
expect(catalog.performanceReport.multiResourceEndpoint).toBe('/api/reporting/generate-multi');
expect(catalog.vmInventoryExport).toBeNull();
});
@@ -14,6 +14,7 @@ const performanceDefinition: ReportingPerformanceReportDefinition = {
title: 'Performance Reports',
description: 'Historical performance reporting',
singleResourceEndpoint: '/api/admin/reports/generate',
singleResourceMethod: 'POST',
multiResourceEndpoint: '/api/admin/reports/generate-multi',
singleFilenamePrefix: 'report',
singleFilenameSubject: 'resource_id',
@@ -75,12 +76,52 @@ describe('reporting panel model', () => {
);
expect(request.filename).toBe('report-agent-1-20260320.pdf');
// POST, never GET: only POST may narrate the PDF with Pulse Assistant.
expect(request.request.url).toBe('/api/admin/reports/generate');
expect(request.request.init).toMatchObject({
method: 'POST',
headers: { 'Content-Type': 'application/json' },
});
expect(JSON.parse(String(request.request.init?.body))).toEqual({
resourceType: 'agent',
resourceId: 'agent-1',
format: 'pdf',
start: '2026-03-19T12:34:56.000Z',
end: now.toISOString(),
metricType: 'cpu',
});
});
it('keeps the query-string GET transport for backends whose catalog names GET', () => {
const now = new Date('2026-03-20T12:34:56.000Z');
const request = buildReportingRequest(
{
end: now.toISOString(),
format: 'csv',
metricType: 'cpu',
now,
resources: [{ id: 'vm-1', type: 'vm', name: 'vm-a' }],
start: '2026-03-19T12:34:56.000Z',
title: '',
},
{
...performanceDefinition,
singleResourceEndpoint: '/api/reporting',
singleResourceMethod: 'GET',
},
);
expect(request.request.init).toBeUndefined();
expect(request.request.url).toContain('/api/admin/reports/generate?');
expect(request.request.url).toContain('resourceType=agent');
expect(request.request.url).toContain('resourceId=agent-1');
expect(request.request.url).toContain('metricType=cpu');
expect(request.request.url).not.toContain('title=');
const url = new URL(request.request.url, 'http://pulse.local');
expect(url.pathname).toBe('/api/reporting');
expect(Object.fromEntries(url.searchParams)).toEqual({
resourceType: 'vm',
resourceId: 'vm-1',
format: 'csv',
start: '2026-03-19T12:34:56.000Z',
end: now.toISOString(),
metricType: 'cpu',
});
});
it('builds a fleet reporting request body and filename', () => {
@@ -158,9 +199,9 @@ describe('reporting panel model', () => {
},
);
expect(request.request.url).not.toContain('title=');
expect(request.request.url).not.toContain('metricType=');
expect(request.request.url).not.toContain('Custom+fleet+title');
const body = JSON.parse(String(request.request.init?.body));
expect(body).not.toHaveProperty('title');
expect(body).not.toHaveProperty('metricType');
});
it('passes through explicit custom titles without inventing defaults', () => {
@@ -186,7 +227,7 @@ describe('reporting panel model', () => {
performanceDefinition,
);
expect(request.request.url).toContain('title=Custom+report+title');
expect(JSON.parse(String(request.request.init?.body)).title).toBe('Custom report title');
});
it('derives canonical range starts from the selected preset', () => {
@@ -28,6 +28,7 @@ const catalogPayload = {
title: 'Performance Reports',
description: 'Historical performance reporting',
singleResourceEndpoint: '/api/admin/reports/generate',
singleResourceMethod: 'POST',
multiResourceEndpoint: '/api/admin/reports/generate-multi',
singleFilenamePrefix: 'report',
singleFilenameSubject: 'resource_id',
@@ -478,6 +479,48 @@ describe('useReportingPanelState', () => {
dispose();
});
it('posts a single-resource report as a JSON body through apiFetch', async () => {
apiFetchMock.mockImplementation((url: string) => {
if (url === '/api/admin/reports/generate') {
return Promise.resolve(new Response('%PDF', { status: 200 }));
}
if (url === '/api/admin/reports/schedules') {
return Promise.resolve(jsonResponse(schedulesPayload));
}
return Promise.resolve(jsonResponse(catalogPayload));
});
const { createObjectURL, revokeObjectURL } = URL;
URL.createObjectURL = vi.fn(() => 'blob:report');
URL.revokeObjectURL = vi.fn();
try {
const { hookState, dispose } = mountHook();
await flushAsync();
hookState.setSelectedResources([{ id: 'vm-1', type: 'vm', name: 'vm-a' }]);
await hookState.handleGenerate();
const generate = apiFetchMock.mock.calls.filter(([url]) =>
String(url).startsWith('/api/admin/reports/generate'),
);
expect(generate).toHaveLength(1);
expect(generate[0][0]).toBe('/api/admin/reports/generate');
expect(generate[0][1]).toMatchObject({
method: 'POST',
headers: { 'Content-Type': 'application/json' },
});
expect(JSON.parse(String(generate[0][1].body))).toMatchObject({
resourceType: 'vm',
resourceId: 'vm-1',
format: catalogPayload.performanceReport.defaultFormat,
});
expect(showWarningMock).not.toHaveBeenCalled();
dispose();
} finally {
URL.createObjectURL = createObjectURL;
URL.revokeObjectURL = revokeObjectURL;
}
});
it('saves a Patrol weekly summary without a resource scope', async () => {
const savedSchedule = {
id: 'digest-1',
@@ -6,6 +6,7 @@ import {
export type ReportingFormat = 'pdf' | 'csv';
export type ReportingFilenameDateStyle = 'utc_yyyymmdd';
export type ReportingFilenameSubject = 'resource_id';
export type ReportingSingleResourceMethod = 'GET' | 'POST';
export interface ReportingFormatDefinition {
value: ReportingFormat;
@@ -24,6 +25,7 @@ export interface ReportingPerformanceReportDefinition {
title: string;
description: string;
singleResourceEndpoint: string;
singleResourceMethod: ReportingSingleResourceMethod;
multiResourceEndpoint: string;
singleFilenamePrefix: string;
singleFilenameSubject: ReportingFilenameSubject;
@@ -85,6 +87,7 @@ export function buildLegacyReportingCatalogFallback(): ReportingCatalog {
description:
'Generate PDF summaries or CSV metric exports from historical monitoring data for one or more selected resources.',
singleResourceEndpoint: '/api/reporting',
singleResourceMethod: 'GET',
multiResourceEndpoint: '/api/reporting/generate-multi',
singleFilenamePrefix: 'report',
singleFilenameSubject: 'resource_id',
@@ -176,6 +179,9 @@ function parseReportingPerformanceReportDefinition(
typeof candidate.title !== 'string' ||
typeof candidate.description !== 'string' ||
typeof candidate.singleResourceEndpoint !== 'string' ||
(candidate.singleResourceMethod !== undefined &&
candidate.singleResourceMethod !== 'GET' &&
candidate.singleResourceMethod !== 'POST') ||
typeof candidate.multiResourceEndpoint !== 'string' ||
typeof candidate.singleFilenamePrefix !== 'string' ||
candidate.singleFilenameSubject !== 'resource_id' ||
@@ -211,6 +217,8 @@ function parseReportingPerformanceReportDefinition(
title: candidate.title,
description: candidate.description,
singleResourceEndpoint: candidate.singleResourceEndpoint,
// Backends that predate the field accept only GET.
singleResourceMethod: candidate.singleResourceMethod ?? 'GET',
multiResourceEndpoint: candidate.multiResourceEndpoint,
singleFilenamePrefix: candidate.singleFilenamePrefix,
singleFilenameSubject: candidate.singleFilenameSubject,
@@ -96,6 +96,7 @@ export function buildReportingRequest(
| 'singleFilenamePrefix'
| 'singleFilenameSubject'
| 'singleResourceEndpoint'
| 'singleResourceMethod'
| 'supportsCustomTitle'
| 'supportsMetricFilter'
>,
@@ -105,23 +106,41 @@ export function buildReportingRequest(
if (context.resources.length === 1) {
const resource = context.resources[0];
const params = new URLSearchParams({
const fields = {
resourceType: toReportingResourceType(resource.type),
resourceId: resource.id,
format: context.format,
start: context.start,
end: context.end,
});
title: customTitle || undefined,
metricType: metricType || undefined,
};
const filename = buildReportingFilename(context.format, resource.id, context.now, definition);
if (metricType) {
params.append('metricType', metricType);
}
if (customTitle) {
params.append('title', customTitle);
// Only POST may narrate a PDF with Pulse Assistant; the catalog names
// GET only for backends that accept nothing else.
if (definition.singleResourceMethod === 'POST') {
return {
filename,
request: {
url: definition.singleResourceEndpoint,
init: {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(fields),
},
},
};
}
const params = new URLSearchParams();
for (const [key, value] of Object.entries(fields)) {
if (value) {
params.append(key, value);
}
}
return {
filename: buildReportingFilename(context.format, resource.id, context.now, definition),
filename,
request: {
url: `${definition.singleResourceEndpoint}?${params.toString()}`,
},
@@ -577,6 +577,27 @@ describe('docsLinks', () => {
expect(apiReference).not.toContain('Mobile Remote Access');
});
it('documents POST as the only report transport that uses Pulse Assistant', () => {
const handler = readFileSync(
path.join(repoRoot, 'internal', 'api', 'metrics_reporting_handlers.go'),
'utf8',
);
expect(handler).toContain(
'if r.Method == http.MethodPost {\n\t\tnarrator, _, findings = h.resolveNarrator(r.Context())',
);
expect(handler).toContain('req.Narrator = getReportNarrator{}');
for (const guide of ['API.md', 'MSP.md']) {
const rootDoc = readFileSync(path.join(repoRoot, 'docs', guide), 'utf8');
expect(readFileSync(path.join(frontendRoot, 'public', 'docs', guide), 'utf8')).toBe(rootDoc);
expect(rootDoc).toContain('`POST /api/admin/reports/generate` (');
}
const apiReference = readFileSync(path.join(repoRoot, 'docs', 'API.md'), 'utf8');
expect(apiReference).toContain('it never calls an AI provider');
expect(apiReference).toContain('`405` with `Allow: GET, POST`');
const msp = readFileSync(path.join(repoRoot, 'docs', 'MSP.md'), 'utf8');
expect(msp).not.toContain('`GET /api/admin/reports/generate` (single resource)');
});
it('ships credential-free Proxmox setup instructions and a separate token prompt', () => {
const apiReference = readFileSync(path.join(repoRoot, 'docs', 'API.md'), 'utf8');
const shipped = readFileSync(path.join(frontendRoot, 'public', 'docs', 'API.md'), 'utf8');
+19
View File
@@ -385,6 +385,25 @@ func (h *AISettingsHandler) loadAIConfig(ctx context.Context) (*config.AIConfig,
}
// GetAIService returns the underlying AI service
// ExistingAIService returns the org's AI service only when one already exists
// and never constructs one. GetAIService builds and configures a tenant service
// on first use, which can list provider models and start background discovery,
// so read paths that must not reach an AI provider use this instead.
func (h *AISettingsHandler) ExistingAIService(ctx context.Context) *ai.Service {
if h == nil {
return nil
}
mtPersistence, mtMonitor, _, _, _, _ := h.stateRefs()
defaultAIService := h.providerSnapshot().defaultAIService
orgID := GetOrgID(ctx)
if orgID == "default" || orgID == "" || (mtPersistence == nil && mtMonitor == nil) {
return defaultAIService
}
h.aiServicesMu.RLock()
defer h.aiServicesMu.RUnlock()
return h.aiServices[orgID]
}
func (h *AISettingsHandler) GetAIService(ctx context.Context) *ai.Service {
mtPersistence, mtMonitor, _, _, _, _ := h.stateRefs()
providers := h.providerSnapshot()
+30
View File
@@ -306,6 +306,36 @@ func TestAISettingsHandler_GetAIService_MultiTenantProviders(t *testing.T) {
}
}
// Report GETs read Patrol findings through ExistingAIService, so it must never
// construct a tenant service: construction can list provider models and start
// background discovery.
func TestAISettingsHandler_ExistingAIServiceNeverConstructs(t *testing.T) {
handler := NewAISettingsHandler(config.NewMultiTenantPersistence(t.TempDir()), nil, nil)
t.Cleanup(handler.StopServices)
tenant := context.WithValue(context.Background(), OrgIDContextKey, "tenant-1")
if svc := handler.ExistingAIService(tenant); svc != nil {
t.Fatalf("expected no service before the tenant's first GetAIService, got %p", svc)
}
handler.aiServicesMu.RLock()
built := len(handler.aiServices)
handler.aiServicesMu.RUnlock()
if built != 0 {
t.Fatalf("ExistingAIService constructed %d tenant services, want 0", built)
}
created := handler.GetAIService(tenant)
if created == nil {
t.Fatal("expected GetAIService to construct the tenant service")
}
if got := handler.ExistingAIService(tenant); got != created {
t.Fatalf("expected the running tenant service %p, got %p", created, got)
}
if got, want := handler.ExistingAIService(context.Background()), handler.GetAIService(context.Background()); got != want {
t.Fatalf("expected the default service %p, got %p", want, got)
}
}
func TestAISettingsHandler_GetAIService_UsesTenantReadState(t *testing.T) {
tmp := t.TempDir()
mtp := config.NewMultiTenantPersistence(tmp)
+164
View File
@@ -5081,6 +5081,7 @@ func TestContract_ReportingCatalogJSONSnapshot(t *testing.T) {
"title":"Performance Reports",
"description":"Generate PDF summaries or CSV metric exports from historical monitoring data for one or more selected resources.",
"singleResourceEndpoint":"/api/admin/reports/generate",
"singleResourceMethod":"POST",
"multiResourceEndpoint":"/api/admin/reports/generate-multi",
"singleFilenamePrefix":"report",
"singleFilenameSubject":"resource_id",
@@ -5227,6 +5228,169 @@ func TestContract_PerformanceReportTransportUsesCatalogDefinition(t *testing.T)
if engine.lastReq.MetricType != "cpu" || engine.lastReq.Title != "Node report" {
t.Fatalf("expected trimmed canonical optional fields, got %+v", engine.lastReq)
}
post := httptest.NewRequest(
http.MethodPost,
"/api/reporting",
strings.NewReader(`{"resourceType":"node","resourceId":"node-1","metricType":" cpu ","title":" Node report "}`),
)
rec = httptest.NewRecorder()
handler.HandleGenerateReport(rec, post)
if rec.Code != http.StatusOK {
t.Fatalf("expected POST 200, got %d body=%s", rec.Code, rec.Body.String())
}
if engine.lastReq.ResourceID != "node-1" || engine.lastReq.Format != definition.DefaultFormat ||
engine.lastReq.MetricType != "cpu" || engine.lastReq.Title != "Node report" {
t.Fatalf("expected POST body to carry the same canonical fields as GET, got %+v", engine.lastReq)
}
}
type recordingReportEngine struct {
*reporting.ReportEngine
requests []reporting.MetricReportRequest
}
func (e *recordingReportEngine) Generate(req reporting.MetricReportRequest) ([]byte, string, error) {
e.requests = append(e.requests, req)
return e.ReportEngine.Generate(req)
}
type countingReportNarrator struct{ calls atomic.Int32 }
func (n *countingReportNarrator) Narrate(context.Context, reporting.NarrativeInput) (reporting.Narrative, error) {
n.calls.Add(1)
return reporting.Narrative{Source: reporting.NarrativeSourceAI, HealthStatus: "HEALTHY", HealthMessage: "narrated"}, nil
}
// A PDF single-resource report may be narrated by the tenant's Pulse
// Assistant, a paid provider call recorded in the cost ledger. GET passes the
// demo-mode guard and the CSRF check, and SameSite=Lax session cookies ride
// cross-site top-level GET navigations, so only POST, the settings UI
// transport, may reach the AI narrator. GET still returns the report, with
// the deterministic summary and a note in place of the "Configure Pulse
// Assistant" tip.
func TestContract_SingleReportAINarrationRequiresPOST(t *testing.T) {
store, err := metrics.NewStore(metrics.DefaultConfig(t.TempDir()))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = store.Close() })
engine := &recordingReportEngine{ReportEngine: reporting.NewReportEngine(reporting.EngineConfig{MetricsStore: store})}
original := reporting.GetEngine()
reporting.SetEngine(engine)
t.Cleanup(func() { reporting.SetEngine(original) })
// The narrator resolver stands for GetAIService, which can construct a
// tenant AI service; GET must use only the existing-service findings path.
narrator := &countingReportNarrator{}
var narratorResolves, findingsResolves atomic.Int32
handler := NewReportingHandlers(nil, nil)
handler.SetNarratorResolver(func(context.Context) (reporting.Narrator, reporting.FleetNarrator, reporting.FindingsProvider) {
narratorResolves.Add(1)
return narrator, nil, nil
})
handler.SetExistingFindingsResolver(func(context.Context) reporting.FindingsProvider {
findingsResolves.Add(1)
return nil
})
generate := func(req *http.Request) *httptest.ResponseRecorder {
rec := httptest.NewRecorder()
handler.HandleGenerateReport(rec, req)
return rec
}
const query = "/api/admin/reports/generate?format=pdf&resourceType=node&resourceId=node-1"
rec := generate(httptest.NewRequest(http.MethodGet, query, nil))
if rec.Code != http.StatusOK || rec.Header().Get("Content-Type") != "application/pdf" {
t.Fatalf("GET report: status %d, content type %q, body=%s", rec.Code, rec.Header().Get("Content-Type"), rec.Body.String())
}
if got := narrator.calls.Load(); got != 0 {
t.Fatalf("GET report called the AI narrator %d times, want 0", got)
}
narrative, err := engine.NarrativeFor(engine.requests[len(engine.requests)-1])
if err != nil {
t.Fatalf("GET report narrative: %v", err)
}
if narrative.Source != reporting.NarrativeSourceHeuristic || narrative.Disclaimer != reportGETNarrativeNote {
t.Fatalf("GET report narrative = source %q disclaimer %q, want the deterministic summary with the GET note", narrative.Source, narrative.Disclaimer)
}
for _, method := range []string{http.MethodHead, http.MethodPut, http.MethodDelete} {
rec := generate(httptest.NewRequest(method, query, nil))
if rec.Code != http.StatusMethodNotAllowed || rec.Header().Get("Allow") != "GET, POST" {
t.Fatalf("%s report: status %d, Allow %q, want 405 with Allow GET, POST", method, rec.Code, rec.Header().Get("Allow"))
}
}
if got := narrator.calls.Load(); got != 0 {
t.Fatalf("GET, HEAD, PUT and DELETE called the AI narrator %d times, want 0", got)
}
if got := narratorResolves.Load(); got != 0 {
t.Fatalf("GET, HEAD, PUT and DELETE resolved the AI service %d times, want 0", got)
}
if got := findingsResolves.Load(); got != 1 {
t.Fatalf("GET looked up existing Patrol findings %d times, want 1", got)
}
rec = generate(httptest.NewRequest(
http.MethodPost,
"/api/admin/reports/generate",
strings.NewReader(`{"format":"pdf","resourceType":"node","resourceId":"node-1"}`),
))
if rec.Code != http.StatusOK || rec.Header().Get("Content-Type") != "application/pdf" {
t.Fatalf("POST report: status %d, content type %q, body=%s", rec.Code, rec.Header().Get("Content-Type"), rec.Body.String())
}
if got := narrator.calls.Load(); got != 1 {
t.Fatalf("POST report called the AI narrator %d times, want 1", got)
}
if got := narratorResolves.Load(); got != 1 {
t.Fatalf("POST report resolved the AI service %d times, want 1", got)
}
}
// The router's real resolvers: a GET report for a tenant whose AI service is
// not running must not construct one, because construction can list provider
// models and start background discovery. POST resolves it as before.
func TestContract_SingleReportGETDoesNotConstructTenantAIService(t *testing.T) {
engine := &stubReportingEngine{data: []byte("report"), contentType: "application/pdf"}
original := reporting.GetEngine()
reporting.SetEngine(engine)
t.Cleanup(func() { reporting.SetEngine(original) })
router := &Router{
aiSettingsHandler: NewAISettingsHandler(config.NewMultiTenantPersistence(t.TempDir()), nil, nil),
reportingHandlers: NewReportingHandlers(nil, nil),
}
router.wireReportingAIResolvers()
t.Cleanup(router.aiSettingsHandler.StopServices)
tenantServices := func() int {
router.aiSettingsHandler.aiServicesMu.RLock()
defer router.aiSettingsHandler.aiServicesMu.RUnlock()
return len(router.aiSettingsHandler.aiServices)
}
generate := func(req *http.Request) *httptest.ResponseRecorder {
rec := httptest.NewRecorder()
router.reportingHandlers.HandleGenerateReport(rec, req.WithContext(context.WithValue(req.Context(), OrgIDContextKey, "tenant-1")))
return rec
}
rec := generate(httptest.NewRequest(http.MethodGet, "/api/admin/reports/generate?format=pdf&resourceType=node&resourceId=node-1", nil))
if rec.Code != http.StatusOK {
t.Fatalf("GET report: status %d, body=%s", rec.Code, rec.Body.String())
}
if got := tenantServices(); got != 0 {
t.Fatalf("GET report constructed %d tenant AI services, want 0", got)
}
if _, ok := engine.lastReq.Narrator.(getReportNarrator); !ok || engine.lastReq.FindingsProvider != nil {
t.Fatalf("GET report narrator %T, findings %v; want the deterministic narrator and no findings without a running service", engine.lastReq.Narrator, engine.lastReq.FindingsProvider)
}
rec = generate(httptest.NewRequest(http.MethodPost, "/api/admin/reports/generate", strings.NewReader(`{"format":"pdf","resourceType":"node","resourceId":"node-1"}`)))
if rec.Code != http.StatusOK {
t.Fatalf("POST report: status %d, body=%s", rec.Code, rec.Body.String())
}
if got := tenantServices(); got != 1 {
t.Fatalf("POST report resolved %d tenant AI services, want 1", got)
}
}
// Normalization keeps a disk's last-known temperature when the current
+118 -27
View File
@@ -7,6 +7,7 @@ import (
"fmt"
"io"
"net/http"
"net/url"
"os"
"regexp"
"strings"
@@ -99,6 +100,7 @@ type ReportingHandlers struct {
mtMonitor *monitoring.MultiTenantMonitor
recoveryManager *recoverymanager.Manager
narratorResolver func(ctx context.Context) (reporting.Narrator, reporting.FleetNarrator, reporting.FindingsProvider)
existingFindingsResolver func(ctx context.Context) reporting.FindingsProvider
settingsStore reportingSystemSettingsStore
scheduleRunMu sync.Mutex
commercialLicenseResolver func(ctx context.Context) *licenseService
@@ -168,6 +170,23 @@ func (h *ReportingHandlers) resolveNarrator(ctx context.Context) (reporting.Narr
return h.narratorResolver(ctx)
}
// SetExistingFindingsResolver wires the Patrol findings lookup that GET reports
// use. Unlike the narrator resolver it must return only an AI service that is
// already running, never construct one.
func (h *ReportingHandlers) SetExistingFindingsResolver(resolver func(ctx context.Context) reporting.FindingsProvider) {
if h == nil {
return
}
h.existingFindingsResolver = resolver
}
func (h *ReportingHandlers) resolveExistingFindings(ctx context.Context) reporting.FindingsProvider {
if h == nil || h.existingFindingsResolver == nil {
return nil
}
return h.existingFindingsResolver(ctx)
}
func (h *ReportingHandlers) resolveReportBranding(ctx context.Context) reporting.ReportBranding {
service := getLicenseServiceForContext(ctx)
entitled := service != nil && service.HasFeature(featureWhiteLabelValue)
@@ -300,28 +319,33 @@ func normalizePerformanceReportTimeRange(
return start, end, nil
}
func decodeMultiReportRequestBody(w http.ResponseWriter, r *http.Request) (multiReportRequestBody, bool) {
// decodeReportingRequestBody strictly decodes one JSON report request body:
// at most 1MB, no unknown fields, no trailing payload.
func decodeReportingRequestBody[T any](w http.ResponseWriter, r *http.Request) (T, bool) {
r.Body = http.MaxBytesReader(w, r.Body, reportingMultiReportBodyMax)
var body multiReportRequestBody
var body, zero T
decoder := json.NewDecoder(r.Body)
decoder.DisallowUnknownFields()
if err := decoder.Decode(&body); err != nil {
err := decoder.Decode(&body)
if err == nil {
// Anything after the value is invalid, and reading past it can also
// hit the size cap.
if trailing := decoder.Decode(&struct{}{}); trailing != io.EOF {
err = errors.New("trailing payload")
if trailing != nil {
err = trailing
}
}
}
if err != nil {
var maxBytesErr *http.MaxBytesError
if errors.As(err, &maxBytesErr) {
writeErrorResponse(w, http.StatusBadRequest, "body_too_large", "Request body must be 1MB or less", nil)
return multiReportRequestBody{}, false
return zero, false
}
writeErrorResponse(w, http.StatusBadRequest, "invalid_body", "Invalid request body", nil)
return multiReportRequestBody{}, false
}
if decoder.More() {
writeErrorResponse(w, http.StatusBadRequest, "invalid_body", "Invalid request body", nil)
return multiReportRequestBody{}, false
}
if err := decoder.Decode(&struct{}{}); err != io.EOF {
writeErrorResponse(w, http.StatusBadRequest, "invalid_body", "Invalid request body", nil)
return multiReportRequestBody{}, false
return zero, false
}
return body, true
@@ -539,9 +563,67 @@ func (h *ReportingHandlers) listBackupsForReport(ctx context.Context, orgID stri
return out
}
// HandleGenerateReport generates a report
// singleReportRequestBody carries the single-resource report fields. POST
// sends them as a JSON body; GET sends the same names as query parameters.
type singleReportRequestBody struct {
ResourceType string `json:"resourceType"`
ResourceID string `json:"resourceId"`
Format string `json:"format"`
Start string `json:"start"`
End string `json:"end"`
Title string `json:"title"`
MetricType string `json:"metricType"`
}
func singleReportRequestFromQuery(q url.Values) singleReportRequestBody {
return singleReportRequestBody{
ResourceType: q.Get("resourceType"),
ResourceID: q.Get("resourceId"),
Format: q.Get("format"),
Start: q.Get("start"),
End: q.Get("end"),
Title: q.Get("title"),
MetricType: q.Get("metricType"),
}
}
// reportGETNarrativeNote takes the place of the PDF's "Configure Pulse
// Assistant" tip on a GET report, which stays deterministic whatever the
// Assistant settings.
const reportGETNarrativeNote = "Tip: This report was requested with GET, which never uses Pulse Assistant. Generate it from Settings, or request it with POST, to include AI narration when Pulse Assistant is configured."
// getReportNarrator narrates GET single-resource reports with the
// deterministic summary and never calls an AI provider.
type getReportNarrator struct{}
func (getReportNarrator) Narrate(ctx context.Context, in reporting.NarrativeInput) (reporting.Narrative, error) {
out, err := reporting.HeuristicNarrator{}.Narrate(ctx, in)
out.Source = reporting.NarrativeSourceHeuristic
out.Disclaimer = reportGETNarrativeNote
return out, err
}
// HandleGenerateReport generates a single-resource report. POST with a JSON
// body is the canonical transport, used by the settings UI, and the only one
// that may narrate a PDF with the tenant's Pulse Assistant, which spends AI
// budget and records usage in the cost ledger. GET takes the same fields as
// query parameters and always narrates deterministically: GET passes the
// demo-mode guard and the CSRF check, and SameSite=Lax session cookies ride
// cross-site top-level GET navigations, so a link must not be able to spend
// that budget.
func (h *ReportingHandlers) HandleGenerateReport(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
var params singleReportRequestBody
switch r.Method {
case http.MethodGet:
params = singleReportRequestFromQuery(r.URL.Query())
case http.MethodPost:
body, ok := decodeReportingRequestBody[singleReportRequestBody](w, r)
if !ok {
return
}
params = body
default:
w.Header().Set("Allow", "GET, POST")
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
@@ -553,15 +635,14 @@ func (h *ReportingHandlers) HandleGenerateReport(w http.ResponseWriter, r *http.
}
definition := performanceReportDefinition()
q := r.URL.Query()
format, err := normalizePerformanceReportFormat(q.Get("format"), definition)
format, err := normalizePerformanceReportFormat(params.Format, definition)
if err != nil {
writeErrorResponse(w, http.StatusBadRequest, "invalid_format", err.Error(), nil)
return
}
resourceTypeRaw := q.Get("resourceType")
resourceID := q.Get("resourceId")
resourceTypeRaw := params.ResourceType
resourceID := params.ResourceID
if resourceTypeRaw == "" || resourceID == "" {
writeErrorResponse(w, http.StatusBadRequest, "missing_params", "resourceType and resourceId are required", nil)
return
@@ -584,8 +665,8 @@ func (h *ReportingHandlers) HandleGenerateReport(w http.ResponseWriter, r *http.
metricType, title, err := normalizePerformanceReportOptionalFields(
definition,
q.Get("metricType"),
q.Get("title"),
params.MetricType,
params.Title,
)
if err != nil {
code := "invalid_metric_type"
@@ -599,8 +680,8 @@ func (h *ReportingHandlers) HandleGenerateReport(w http.ResponseWriter, r *http.
start, end, err := normalizePerformanceReportTimeRange(
definition,
q.Get("start"),
q.Get("end"),
params.Start,
params.End,
time.Now(),
)
if err != nil {
@@ -628,9 +709,18 @@ func (h *ReportingHandlers) HandleGenerateReport(w http.ResponseWriter, r *http.
// Wire the per-tenant AI narrator and Patrol findings provider when
// configured. Both are nil-safe at the engine layer; absence falls
// back to the heuristic narrator with no findings section. Single-
// resource reports do not use the FleetNarrator.
narrator, _, findings := h.resolveNarrator(r.Context())
req.Narrator = narrator
// resource reports do not use the FleetNarrator. Only POST resolves the
// AI service; GET narrates deterministically and reads findings only
// from an AI service that is already running.
var narrator reporting.Narrator
var findings reporting.FindingsProvider
if r.Method == http.MethodPost {
narrator, _, findings = h.resolveNarrator(r.Context())
req.Narrator = narrator
} else {
findings = h.resolveExistingFindings(r.Context())
req.Narrator = getReportNarrator{}
}
req.FindingsProvider = findings
data, contentType, err := engine.Generate(req)
@@ -647,6 +737,7 @@ func (h *ReportingHandlers) HandleGenerateReport(w http.ResponseWriter, r *http.
log.Info().
Str("event", "reporting.single.generated").
Str("org_id", orgID).
Str("method", r.Method).
Str("resource_type", resourceType).
Str("metric_type", metricType).
Str("format", string(format)).
@@ -1187,7 +1278,7 @@ func (h *ReportingHandlers) HandleGenerateMultiReport(w http.ResponseWriter, r *
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
body, ok := decodeMultiReportRequestBody(w, r)
body, ok := decodeReportingRequestBody[multiReportRequestBody](w, r)
if !ok {
return
}
+42 -1
View File
@@ -75,7 +75,7 @@ func (p reportBrandLicenseProvider) Service(context.Context) *pkglicensing.Servi
func TestReportingHandlers_MethodNotAllowed(t *testing.T) {
handler := NewReportingHandlers(nil, nil)
req := httptest.NewRequest(http.MethodPost, "/api/reporting", nil)
req := httptest.NewRequest(http.MethodPut, "/api/reporting", nil)
rr := httptest.NewRecorder()
handler.HandleGenerateReport(rr, req)
@@ -83,6 +83,47 @@ func TestReportingHandlers_MethodNotAllowed(t *testing.T) {
if rr.Code != http.StatusMethodNotAllowed {
t.Fatalf("expected status %d, got %d", http.StatusMethodNotAllowed, rr.Code)
}
if got := rr.Header().Get("Allow"); got != "GET, POST" {
t.Fatalf("expected Allow GET, POST, got %q", got)
}
}
func TestReportingHandlers_GenerateReportRejectsInvalidPOSTBody(t *testing.T) {
engine := &stubReportingEngine{data: []byte("report"), contentType: "application/pdf"}
original := reporting.GetEngine()
reporting.SetEngine(engine)
t.Cleanup(func() { reporting.SetEngine(original) })
handler := NewReportingHandlers(nil, nil)
for name, body := range map[string]string{
"empty": ``,
"unknown field": `{"resourceType":"node","resourceId":"node-1","narrate":true}`,
"trailing value": `{"resourceType":"node","resourceId":"node-1"} {}`,
"query shape": `resourceType=node&resourceId=node-1`,
} {
req := httptest.NewRequest(http.MethodPost, "/api/reporting", strings.NewReader(body))
rr := httptest.NewRecorder()
handler.HandleGenerateReport(rr, req)
if rr.Code != http.StatusBadRequest || !strings.Contains(rr.Body.String(), "invalid_body") {
t.Fatalf("%s: expected 400 invalid_body, got %d body=%s", name, rr.Code, rr.Body.String())
}
}
req := httptest.NewRequest(http.MethodPost, "/api/reporting?resourceType=node&resourceId=node-1", strings.NewReader(`{}`))
rr := httptest.NewRecorder()
handler.HandleGenerateReport(rr, req)
if rr.Code != http.StatusBadRequest || !strings.Contains(rr.Body.String(), "missing_params") {
t.Fatalf("POST must read fields from the body, not the query: got %d body=%s", rr.Code, rr.Body.String())
}
// A valid value followed by padding past the cap is too large, not malformed.
padded := `{"resourceType":"node","resourceId":"node-1"}` + strings.Repeat(" ", reportingMultiReportBodyMax)
req = httptest.NewRequest(http.MethodPost, "/api/reporting", strings.NewReader(padded))
rr = httptest.NewRecorder()
handler.HandleGenerateReport(rr, req)
if rr.Code != http.StatusBadRequest || !strings.Contains(rr.Body.String(), "body_too_large") {
t.Fatalf("expected 400 body_too_large for an oversized body, got %d body=%s", rr.Code, rr.Body.String())
}
}
func TestReportingHandlers_EngineUnavailable(t *testing.T) {
+45 -27
View File
@@ -805,33 +805,7 @@ func (r *Router) setupRoutes() {
// tenant services are refreshed when Router.SetMonitor replaces a monitor.
r.configureMetadataProviderFactory()
// Wire the per-tenant AI narrator, fleet narrator, and Patrol
// findings provider into reporting. The AI service implements all
// three interfaces; when not configured for the tenant the engine
// falls back to the heuristic narrators with no findings section.
if r.reportingHandlers != nil {
settings := r.aiSettingsHandler
r.reportingHandlers.SetPatrolDigestResolver(func(ctx context.Context, days int) (ai.PatrolDigest, bool) {
if settings == nil {
return ai.PatrolDigest{}, false
}
return settings.BuildPatrolDigest(ctx, days)
})
r.reportingHandlers.SetNarratorResolver(func(ctx context.Context) (reporting.Narrator, reporting.FleetNarrator, reporting.FindingsProvider) {
if settings == nil {
return nil, nil, nil
}
svc := settings.GetAIService(ctx)
if svc == nil {
return nil, nil, nil
}
cfg := svc.GetAIConfig()
if cfg == nil || !cfg.Enabled {
return nil, nil, nil
}
return svc, svc, svc
})
}
r.wireReportingAIResolvers()
// AI chat handler
r.aiHandler = NewAIHandler(r.multiTenant, r.mtMonitor, r.agentExecServer)
@@ -1031,6 +1005,50 @@ func (r *Router) setupRoutes() {
// See issue #334 - ServeMux redirects empty path to "./" which breaks reverse proxies
}
// wireReportingAIResolvers connects reporting to the per-tenant AI service.
func (r *Router) wireReportingAIResolvers() {
// Wire the per-tenant AI narrator, fleet narrator, and Patrol
// findings provider into reporting. The AI service implements all
// three interfaces; when not configured for the tenant the engine
// falls back to the heuristic narrators with no findings section.
if r.reportingHandlers != nil {
settings := r.aiSettingsHandler
r.reportingHandlers.SetPatrolDigestResolver(func(ctx context.Context, days int) (ai.PatrolDigest, bool) {
if settings == nil {
return ai.PatrolDigest{}, false
}
return settings.BuildPatrolDigest(ctx, days)
})
r.reportingHandlers.SetNarratorResolver(func(ctx context.Context) (reporting.Narrator, reporting.FleetNarrator, reporting.FindingsProvider) {
if settings == nil {
return nil, nil, nil
}
svc := settings.GetAIService(ctx)
if svc == nil {
return nil, nil, nil
}
cfg := svc.GetAIConfig()
if cfg == nil || !cfg.Enabled {
return nil, nil, nil
}
return svc, svc, svc
})
r.reportingHandlers.SetExistingFindingsResolver(func(ctx context.Context) reporting.FindingsProvider {
if settings == nil {
return nil
}
svc := settings.ExistingAIService(ctx)
if svc == nil {
return nil
}
if cfg := svc.GetAIConfig(); cfg == nil || !cfg.Enabled {
return nil
}
return svc
})
}
}
// CleanupTenant removes all per-tenant resources (RBAC, AI, License) for a deleted org.
func (r *Router) CleanupTenant(ctx context.Context, orgID string) error {
orgID = strings.TrimSpace(orgID)
@@ -89,3 +89,120 @@ func TestHostContinuityPreservesLiveAgentAfterReenrollment(t *testing.T) {
})
}
}
// Continuity hydration clones the registry through resource ingest, which
// applies operator links, while the rebuild ingests supplemental records after
// the snapshot. A link to a record-sourced guest (an agent inside a vSphere VM)
// must hold in the published inventory either way, or the state the websocket
// broadcasts and the resources API seeds from flips with an unrelated absent
// machine. A saved enrollment only fills an absent machine: when the linked
// agent itself is gone, its continuity row stays separate and the guest keeps
// what vSphere reports.
func TestManualLinkToSupplementalGuestHoldsWithAndWithoutContinuity(t *testing.T) {
now := time.Now().UTC()
guestAgent := models.Host{
ID: "host-app-guest", MachineID: "machine-app-guest", Hostname: "app-guest",
Platform: "linux", Status: "online", LastSeen: now, IntervalSeconds: 30,
}
vmRecords := []unifiedresources.IngestRecord{{
SourceID: "vc-1:vm:vm-42",
Resource: unifiedresources.Resource{
Type: unifiedresources.ResourceTypeVM,
Technology: "vmware",
Name: "app-guest",
Status: unifiedresources.StatusOnline,
LastSeen: now,
VMware: &unifiedresources.VMwareData{ConnectionID: "vc-1", ManagedObjectID: "vm-42", EntityType: "vm"},
},
}}
unlinked := unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(nil))
unlinked.PopulateSnapshotAndSupplemental(
models.StateSnapshot{Hosts: []models.Host{guestAgent}, LastUpdate: now},
map[unifiedresources.DataSource][]unifiedresources.IngestRecord{unifiedresources.SourceVMware: vmRecords},
)
if len(unlinked.VMs()) != 1 || len(unlinked.Hosts()) != 1 {
t.Fatalf("unlinked estate = %d VMs, %d agents, want one of each", len(unlinked.VMs()), len(unlinked.Hosts()))
}
vmID, agentID := unlinked.VMs()[0].ID(), unlinked.Hosts()[0].ID()
for _, tc := range []struct {
name string
reporting bool
continuity []config.HostContinuityEntry
}{
{name: "no-continuity", reporting: true},
{
name: "absent-machine-continuity",
reporting: true,
continuity: []config.HostContinuityEntry{{
HostID: "absent-machine", MachineID: "other-machine", Hostname: "other-host",
Platform: "linux", LastSeen: now.Add(-time.Hour),
}},
},
{
name: "linked-agent-continuity-only",
continuity: []config.HostContinuityEntry{{
HostID: guestAgent.ID, MachineID: guestAgent.MachineID, Hostname: guestAgent.Hostname,
Platform: "linux", LastSeen: now.Add(-10 * time.Minute),
}},
},
} {
t.Run(tc.name, func(t *testing.T) {
links := unifiedresources.NewMemoryStore()
if err := links.AddLink(unifiedresources.ResourceLink{ResourceA: vmID, ResourceB: agentID, PrimaryID: vmID}); err != nil {
t.Fatalf("add link: %v", err)
}
state := models.NewState()
if tc.reporting {
state.Hosts = []models.Host{guestAgent}
}
m := &Monitor{state: state}
if len(tc.continuity) > 0 {
continuity := config.NewHostContinuityStore(t.TempDir(), nil)
for _, entry := range tc.continuity {
if err := continuity.Upsert(entry); err != nil {
t.Fatal(err)
}
}
m.hostContinuityStore = continuity
}
m.SetResourceStore(unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(links)))
m.SetSupplementalRecordsProvider(unifiedresources.SourceVMware, &testSupplementalInventoryReadinessProvider{
source: unifiedresources.SourceVMware,
records: vmRecords,
settled: true,
readyAt: now,
})
resources, _ := m.UnifiedResourceSnapshot()
var guest, agent *unifiedresources.Resource
for i := range resources {
switch resources[i].ID {
case agentID:
agent = &resources[i]
case vmID:
guest = &resources[i]
}
}
if guest == nil {
t.Fatalf("linked VM %s missing from published inventory", vmID)
}
if tc.reporting {
if agent != nil {
t.Fatalf("linked agent %s still published standalone", agentID)
}
if guest.VMware == nil || guest.Agent == nil {
t.Fatalf("published VM vmware=%t agent=%t, want the VM carrying its agent", guest.VMware != nil, guest.Agent != nil)
}
return
}
if agent == nil || agent.Status != unifiedresources.StatusOffline {
t.Fatalf("saved enrollment row = %+v, want its own offline row", agent)
}
if guest.Agent != nil || guest.Status != unifiedresources.StatusOnline {
t.Fatalf("published VM agent=%t status=%s, want vSphere's online VM untouched by the saved enrollment", guest.Agent != nil, guest.Status)
}
})
}
}
@@ -337,8 +337,9 @@ func (index *identityPinIndex) successionsFor(pin ResourceIdentityPin) []Canonic
// the new canonical ID before the pin write, so operator intent
// (never-auto-remediate, maintenance windows) and action-audit history
// survive the era change. Successions are skipped while the old canonical ID
// still belongs to a live resource: a genuinely short-named host must not
// have its rows stolen by an FQDN sibling. Change-journal rows are never
// still belongs to a live resource, or to one a manual link folded into
// another: a genuinely short-named host must not have its rows stolen by an
// FQDN sibling. Change-journal rows are never
// rewritten; EraIDs merges those at read time.
func (rr *ResourceRegistry) PersistIdentityPins() {
if rr.store == nil {
@@ -358,7 +359,11 @@ func (rr *ResourceRegistry) PersistIdentityPins() {
}
pins = append(pins, pin)
for _, succession := range rr.identityPins.successionsFor(pin) {
if _, live := rr.resources[succession.OldCanonicalID]; live {
// A primary that absorbed a linked resource also carries its
// machine keys. That resource is still observed, and succeeding
// it would re-key the link onto the primary itself, splitting
// the pair again on the next rebuild.
if rr.canonicalIDObservedLocked(succession.OldCanonicalID) {
continue
}
successions = append(successions, succession)
+3 -3
View File
@@ -135,7 +135,7 @@ func readStateWithRecords(readState ReadState, source DataSource, records []Inge
cloned := NewRegistry(registry.store)
thresholds := adapter.currentStaleThresholds()
cloned.IngestResourcesWithStaleThresholds(registry.List(), thresholds)
cloned.ingestRecords(source, records, onlyMissing)
cloned.ingestRecords(source, records, onlyMissing, thresholds)
overlay := NewMonitorAdapterWithStaleThresholds(cloned, thresholds)
if keyed {
adapter.overlays.store(registry, rebuiltAt, key, now, overlay)
@@ -450,7 +450,7 @@ func (a *MonitorAdapter) replaceRegistryLocked(snapshot models.StateSnapshot, re
if len(records) == 0 || strings.TrimSpace(string(source)) == "" {
continue
}
rebuilt.IngestRecords(source, records)
rebuilt.IngestRecordsWithStaleThresholds(source, records, staleThresholds)
}
// IngestSnapshot runs its stale pass before the record sources above are
// ingested, so record-sourced resources would otherwise keep their
@@ -665,7 +665,7 @@ func (a *MonitorAdapter) PopulateSupplementalRecords(source DataSource, records
return
}
before := registry.List()
registry.IngestRecords(source, records)
registry.IngestRecordsWithStaleThresholds(source, records, a.currentStaleThresholds())
rebuiltAt := time.Now().UTC()
recordRegistryChanges(registry.store, before, registry.List(), rebuiltAt, nil, SourcePlatformEvent, changeSourceAdapterForDataSource(source))
registry.PersistIdentityPins()
@@ -400,6 +400,234 @@ func TestReadStateWithRecordsPreservesConfiguredStaleThresholds(t *testing.T) {
}
}
// A link whose side arrives as a record joins during record ingest, so it
// must judge freshness by the adapter's thresholds as a snapshot-time link
// does. Under a five-minute vSphere threshold a VM polled three minutes ago
// is current, and its own memory reading outranks its in-guest agent's.
func TestMonitorAdapterJoinsLinkedRecordsWithConfiguredStaleThresholds(t *testing.T) {
now := time.Now().UTC()
snapshot := models.StateSnapshot{
Hosts: []models.Host{{
ID: "host-app-guest",
Hostname: "app-guest",
MachineID: "machine-app-guest",
Status: "online",
LastSeen: now,
Memory: models.Memory{Total: 8 << 30, Used: 6 << 30, Free: 2 << 30, Usage: 75},
}},
LastUpdate: now,
}
vmRecords := []IngestRecord{{
SourceID: "vc-1:vm:vm-42",
Resource: Resource{
Type: ResourceTypeVM,
Technology: "vmware",
Name: "app-guest",
Status: StatusOnline,
LastSeen: now.Add(-3 * time.Minute),
Metrics: &ResourceMetrics{Memory: &MetricValue{Percent: 40, Source: SourceVMware}},
VMware: &VMwareData{ConnectionID: "vc-1", ManagedObjectID: "vm-42", EntityType: "vm"},
},
}}
thresholds := map[DataSource]time.Duration{SourceVMware: 5 * time.Minute}
unlinked := NewMonitorAdapter(NewRegistry(nil))
unlinked.PopulateSnapshotAndSupplemental(snapshot, map[DataSource][]IngestRecord{SourceVMware: vmRecords})
vms, hosts := unlinked.VMs(), unlinked.Hosts()
if len(vms) != 1 || len(hosts) != 1 {
t.Fatalf("unlinked estate = %d VMs, %d agents, want one of each", len(vms), len(hosts))
}
link := ResourceLink{ResourceA: vms[0].ID(), ResourceB: hosts[0].ID(), PrimaryID: hosts[0].ID()}
for _, path := range []struct {
name string
read func(adapter *MonitorAdapter) ReadState
}{
{"rebuild", func(adapter *MonitorAdapter) ReadState {
adapter.PopulateSnapshotAndSupplemental(snapshot, map[DataSource][]IngestRecord{SourceVMware: vmRecords})
return adapter
}},
{"supplemental", func(adapter *MonitorAdapter) ReadState {
adapter.PopulateFromSnapshot(snapshot)
adapter.PopulateSupplementalRecords(SourceVMware, vmRecords)
return adapter
}},
{"overlay", func(adapter *MonitorAdapter) ReadState {
adapter.PopulateFromSnapshot(snapshot)
return ReadStateWithRecords(adapter, SourceVMware, vmRecords)
}},
} {
t.Run(path.name, func(t *testing.T) {
store := NewMemoryStore()
if err := store.AddLink(link); err != nil {
t.Fatalf("add link: %v", err)
}
readState := path.read(NewMonitorAdapterWithStaleThresholds(NewRegistry(store), thresholds))
if hosts := readState.Hosts(); len(hosts) != 0 {
t.Fatalf("linked agent still listed standalone: %s", hosts[0].ID())
}
vms := readState.VMs()
if len(vms) != 1 || vms[0].ID() != link.ResourceA {
t.Fatalf("VMs = %d, want the linked VM %s", len(vms), link.ResourceA)
}
if vms[0].r.Metrics == nil || vms[0].r.Metrics.Memory == nil {
t.Fatal("linked VM lost its memory reading")
}
if memory := vms[0].r.Metrics.Memory; memory.Source != SourceVMware || memory.Percent != 40 {
t.Fatalf("linked VM memory = %.0f%% from %s, want vSphere's current 40%%", memory.Percent, memory.Source)
}
})
}
}
// Every source merge into an existing row judges metric freshness by the
// adapter's configured thresholds, as the rebuild's links and stale pass do.
// A linked vSphere VM first reports no memory, so its in-guest agent supplies
// it; a refresh then brings vSphere memory observed three minutes ago, current
// under a five-minute threshold. The live supplemental refresh and the
// read-state overlay merge that record into the existing row, and must show
// vSphere's memory as a rebuild from the same records does. A Proxmox node
// polled every two minutes is likewise current at ninety seconds, so the
// rebuild keeps its readings over those of its auto-linked agent, silent for
// seventy-five seconds.
func TestMonitorAdapterSourceMergesUseConfiguredStaleThresholds(t *testing.T) {
now := time.Now().UTC()
thresholds := map[DataSource]time.Duration{SourceProxmox: 4 * time.Minute, SourceVMware: 5 * time.Minute}
t.Run("linked vSphere VM refresh", func(t *testing.T) {
snapshot := models.StateSnapshot{
Hosts: []models.Host{{
ID: "host-app-guest",
Hostname: "app-guest",
MachineID: "machine-app-guest",
Status: "online",
LastSeen: now,
Memory: models.Memory{Total: 8 << 30, Used: 6 << 30, Free: 2 << 30, Usage: 75},
}},
LastUpdate: now,
}
vmRecord := func(lastSeen time.Time, metrics *ResourceMetrics) []IngestRecord {
return []IngestRecord{{
SourceID: "vc-1:vm:vm-42",
Resource: Resource{
Type: ResourceTypeVM,
Technology: "vmware",
Name: "app-guest",
Status: StatusOnline,
LastSeen: lastSeen,
Metrics: metrics,
VMware: &VMwareData{ConnectionID: "vc-1", ManagedObjectID: "vm-42", EntityType: "vm"},
},
}}
}
initial := vmRecord(now.Add(-4*time.Minute), &ResourceMetrics{CPU: &MetricValue{Percent: 10, Source: SourceVMware}})
refreshedAt := now.Add(-3 * time.Minute)
refreshed := vmRecord(refreshedAt, &ResourceMetrics{
CPU: &MetricValue{Percent: 10, Source: SourceVMware},
Memory: &MetricValue{Percent: 40, Source: SourceVMware},
})
unlinked := NewMonitorAdapter(NewRegistry(nil))
unlinked.PopulateSnapshotAndSupplemental(snapshot, map[DataSource][]IngestRecord{SourceVMware: initial})
vms, hosts := unlinked.VMs(), unlinked.Hosts()
if len(vms) != 1 || len(hosts) != 1 {
t.Fatalf("unlinked estate = %d VMs, %d agents, want one of each", len(vms), len(hosts))
}
link := ResourceLink{ResourceA: vms[0].ID(), ResourceB: hosts[0].ID(), PrimaryID: hosts[0].ID()}
for _, path := range []struct {
name string
read func(adapter *MonitorAdapter) ReadState
}{
{"rebuild", func(adapter *MonitorAdapter) ReadState {
adapter.PopulateSnapshotAndSupplemental(snapshot, map[DataSource][]IngestRecord{SourceVMware: refreshed})
return adapter
}},
{"supplemental", func(adapter *MonitorAdapter) ReadState {
adapter.PopulateSnapshotAndSupplemental(snapshot, map[DataSource][]IngestRecord{SourceVMware: initial})
adapter.PopulateSupplementalRecords(SourceVMware, refreshed)
return adapter
}},
{"overlay", func(adapter *MonitorAdapter) ReadState {
adapter.PopulateSnapshotAndSupplemental(snapshot, map[DataSource][]IngestRecord{SourceVMware: initial})
return ReadStateWithRecords(adapter, SourceVMware, refreshed)
}},
} {
t.Run(path.name, func(t *testing.T) {
store := NewMemoryStore()
if err := store.AddLink(link); err != nil {
t.Fatalf("add link: %v", err)
}
readState := path.read(NewMonitorAdapterWithStaleThresholds(NewRegistry(store), thresholds))
vms := readState.VMs()
if len(vms) != 1 || vms[0].ID() != link.ResourceA || len(readState.Hosts()) != 0 {
t.Fatalf("estate = %d VMs, %d standalone agents, want only the linked VM %s", len(vms), len(readState.Hosts()), link.ResourceA)
}
if got := vms[0].r.SourceStatus[SourceVMware]; got.Status != "online" || !got.LastSeen.Equal(refreshedAt) {
t.Fatalf("vSphere sighting = %+v, want the refresh, current under the configured threshold", got)
}
if vms[0].r.Metrics == nil || vms[0].r.Metrics.Memory == nil {
t.Fatal("linked VM lost its memory reading")
}
if memory := vms[0].r.Metrics.Memory; memory.Source != SourceVMware || memory.Percent != 40 {
t.Fatalf("linked VM memory = %.0f%% from %s, want vSphere's current 40%%", memory.Percent, memory.Source)
}
})
}
})
t.Run("node with auto-linked agent", func(t *testing.T) {
adapter := NewMonitorAdapterWithStaleThresholds(NewRegistry(nil), thresholds)
adapter.PopulateFromSnapshot(models.StateSnapshot{
Nodes: []models.Node{{
ID: "homelab-pve1",
Name: "pve1",
Instance: "homelab",
Status: "online",
LinkedAgentID: "host-pve1",
CPU: 0.12,
Memory: models.Memory{Total: 64 << 30, Used: 16 << 30, Free: 48 << 30, Usage: 25},
LastSeen: now.Add(-90 * time.Second),
}},
Hosts: []models.Host{{
ID: "host-pve1",
MachineID: "machine-pve1",
Hostname: "pve1",
LinkedNodeID: "homelab-pve1",
Status: "online",
CPUUsage: 80,
Memory: models.Memory{Total: 64 << 30, Used: 48 << 30, Free: 16 << 30, Usage: 75},
LastSeen: now.Add(-75 * time.Second),
IntervalSeconds: 30,
}},
LastUpdate: now,
})
nodes := adapter.Nodes()
if len(nodes) != 1 {
t.Fatalf("nodes = %d, want the node merged with its agent", len(nodes))
}
node := nodes[0].r
if poll := node.SourceStatus[SourceProxmox]; poll.Status != "online" || !poll.LastSeen.Equal(now.Add(-90*time.Second)) {
t.Fatalf("Proxmox sighting = %+v, want the current poll from ninety seconds ago", poll)
}
if got := node.SourceStatus[SourceAgent].Status; got != "stale" {
t.Fatalf("agent sighting = %q, want stale", got)
}
if node.Metrics == nil || node.Metrics.CPU == nil || node.Metrics.Memory == nil {
t.Fatalf("merged node lost its readings: %+v", node.Metrics)
}
if cpu := node.Metrics.CPU; cpu.Source != SourceProxmox || cpu.Percent != 12 {
t.Fatalf("node CPU = %.0f%% from %s, want the current poll's 12%%", cpu.Percent, cpu.Source)
}
if memory := node.Metrics.Memory; memory.Source != SourceProxmox || memory.Percent != 25 {
t.Fatalf("node memory = %.0f%% from %s, want the current poll's 25%%", memory.Percent, memory.Source)
}
})
}
func TestHostContinuityCannotOverwriteCurrentResource(t *testing.T) {
now := time.Now().UTC()
for _, mode := range []string{"machine-identity", "source-identity", "future-timestamp", "provider-only"} {
+90 -30
View File
@@ -95,6 +95,10 @@ type ResourceRegistry struct {
exclusions map[string]struct{}
identityPins *identityPinIndex
pbsBackups []models.PBSBackup
// linkMergedIDs holds the resources a manual link folded into their
// primary. They are still observed, so identity succession treats them
// as live.
linkMergedIDs map[string]struct{}
// supersededIndex maps record-declared retired canonical IDs to the live
// resource that superseded them, so references persisted under a retired
// ID (availability links, API reads) keep resolving. An empty value marks
@@ -148,6 +152,13 @@ type ResourceRegistry struct {
// what the full scan would see; scores stay live through the shared
// pointers.
agentNodeScanIndex map[string][]agentNodeCandidate
// ingestStaleThresholds holds the caller-owned freshness thresholds of
// the snapshot or record ingest in progress, so each source merge into an
// existing row judges metric freshness as that ingest's manual links and
// stale pass do. nil outside an ingest and for callers without
// thresholds, where all three use the defaults.
ingestStaleThresholds map[DataSource]time.Duration
}
// agentNodeCandidate pairs a resources-map key with its entry so the indexed
@@ -243,6 +254,10 @@ func (rr *ResourceRegistry) IngestSnapshotWithStaleThresholds(snapshot models.St
}
func (rr *ResourceRegistry) ingestSnapshot(snapshot models.StateSnapshot, thresholds map[DataSource]time.Duration) {
rr.mu.Lock()
rr.ingestStaleThresholds = thresholds
rr.mu.Unlock()
hostByID := make(map[string]*models.Host, len(snapshot.Hosts))
for i := range snapshot.Hosts {
host := snapshot.Hosts[i]
@@ -606,15 +621,26 @@ func (rr *ResourceRegistry) ingestSnapshot(snapshot models.StateSnapshot, thresh
rr.refreshCanonicalIdentitiesLocked()
rr.markStaleLocked(time.Now().UTC(), thresholds)
rr.invalidateViewsLocked()
rr.ingestStaleThresholds = nil
rr.mu.Unlock()
}
// IngestRecords ingests normalized records for a single source.
func (rr *ResourceRegistry) IngestRecords(source DataSource, records []IngestRecord) {
rr.ingestRecords(source, records, false)
rr.ingestRecords(source, records, false, nil)
}
func (rr *ResourceRegistry) ingestRecords(source DataSource, records []IngestRecord, onlyMissing bool) {
// IngestRecordsWithStaleThresholds ingests normalized records for a single
// source and merges operator-linked resources with caller-owned thresholds.
func (rr *ResourceRegistry) IngestRecordsWithStaleThresholds(source DataSource, records []IngestRecord, thresholds map[DataSource]time.Duration) {
rr.ingestRecords(source, records, false, thresholds)
}
func (rr *ResourceRegistry) ingestRecords(source DataSource, records []IngestRecord, onlyMissing bool, thresholds map[DataSource]time.Duration) {
rr.mu.Lock()
rr.ingestStaleThresholds = thresholds
rr.mu.Unlock()
var successions []CanonicalIDSuccession
supersededSeen := make(map[string]struct{})
for _, record := range records {
@@ -649,6 +675,16 @@ func (rr *ResourceRegistry) ingestRecords(source DataSource, records []IngestRec
rr.applyRecordSuccessions(successions)
rr.mu.Lock()
// Either side of an operator link can arrive as a record: a vSphere or
// TrueNAS VM with a Pulse agent inside, or an agent on a TrueNAS host.
// Snapshot and resource ingest apply the links, so record ingest must too,
// or the monitor's rebuilt registry keeps both rows while REST, which
// seeds from that registry, shows them merged. Continuity records only
// fill absent machines: a saved enrollment joined to its linked guest
// would lend the guest its offline verdict and old agent payload.
if !onlyMissing {
rr.applyManualLinks(thresholds)
}
rr.refreshStorageConsumersLocked()
rr.refreshPBSRollupsLocked()
rr.refreshStoragePostureLocked()
@@ -657,6 +693,7 @@ func (rr *ResourceRegistry) ingestRecords(source DataSource, records []IngestRec
rr.refreshProxmoxSensorSetupLocked()
rr.refreshCanonicalIdentitiesLocked()
rr.invalidateViewsLocked()
rr.ingestStaleThresholds = nil
rr.mu.Unlock()
}
@@ -746,12 +783,25 @@ func (rr *ResourceRegistry) proxmoxGuestResourceIDForSourceRefLocked(ref string)
return uniqueID
}
// canonicalIDObservedLocked reports whether a canonical ID still names an
// observed resource: a live row, or one a manual link folded into its primary.
// Succeeding an observed ID would hand its operator rows, and the link that
// folded it, to another resource.
func (rr *ResourceRegistry) canonicalIDObservedLocked(id string) bool {
if _, live := rr.resources[id]; live {
return true
}
_, folded := rr.linkMergedIDs[id]
return folded
}
// applyRecordSuccessions re-keys operator-owned store rows from canonical IDs
// that ingested records declared superseded (IngestRecord.SupersededCanonicalIDs)
// onto the records' current canonical IDs. Mirrors the guards of pin-driven
// successions in PersistIdentityPins: a superseded ID still held by a live
// resource is skipped so a genuinely distinct sibling never has its rows
// stolen. Re-runs are cheap no-ops once the old rows are gone.
// successions in PersistIdentityPins: a superseded ID still observed, live or
// folded into another resource by a manual link, is skipped so a genuinely
// distinct sibling never has its rows stolen. Re-runs are cheap no-ops once
// the old rows are gone.
func (rr *ResourceRegistry) applyRecordSuccessions(successions []CanonicalIDSuccession) {
if len(successions) == 0 || rr.store == nil {
return
@@ -764,7 +814,7 @@ func (rr *ResourceRegistry) applyRecordSuccessions(successions []CanonicalIDSucc
rr.mu.RLock()
kept := make([]CanonicalIDSuccession, 0, len(successions))
for _, succession := range successions {
if _, live := rr.resources[succession.OldCanonicalID]; live {
if rr.canonicalIDObservedLocked(succession.OldCanonicalID) {
continue
}
kept = append(kept, succession)
@@ -1772,26 +1822,15 @@ func sourceVerdictsRecorded(sightings map[DataSource]SourceStatus) bool {
}
func (rr *ResourceRegistry) markStaleLocked(now time.Time, thresholds map[DataSource]time.Duration) {
thresholds = effectiveStaleThresholds(thresholds)
changed := false
for _, resource := range rr.resources {
previousStatus := resource.Status
staleFound := false
for source, status := range resource.SourceStatus {
threshold, ok := thresholds[source]
if !ok {
threshold = 120 * time.Second
}
if status.ExpectedUpdateIntervalSeconds > 0 {
// Slow inventory polls have their own cadence. A source must miss
// two expected intervals before its retained observation is stale.
threshold = max(threshold, 2*time.Duration(status.ExpectedUpdateIntervalSeconds)*time.Second)
}
if status.LastSeen.IsZero() {
continue
}
if now.Sub(status.LastSeen) > threshold {
if now.Sub(status.LastSeen) > sourceStaleThreshold(source, status, thresholds) {
changed = changed || status.Status != "stale"
status.Status = "stale"
resource.SourceStatus[source] = status
@@ -3940,7 +3979,7 @@ func (rr *ResourceRegistry) mergeInto(existing *Resource, incoming Resource, sou
existing.ParentID = rr.resolveCanonicalParentID(existing)
existing.Status = chooseStatus(existing.Status, incoming.Status, source, existing.Sources)
existing.Metrics = mergeMetrics(existing, existing.Metrics, incoming.Metrics, source, now, existing.SourceStatus, nil)
existing.Metrics = mergeMetrics(existing, existing.Metrics, incoming.Metrics, source, now, existing.SourceStatus, rr.ingestStaleThresholds)
existing.Metrics = clearUnavailableSourceMemoryMetric(existing.Metrics, &incoming, source)
// Prefer agent naming when available
@@ -4811,6 +4850,10 @@ func (rr *ResourceRegistry) applyManualLinks(thresholds map[DataSource]time.Dura
rr.mergeResourceData(primary, other, thresholds)
delete(rr.resources, otherID)
if rr.linkMergedIDs == nil {
rr.linkMergedIDs = make(map[string]struct{})
}
rr.linkMergedIDs[otherID] = struct{}{}
rr.updateSourceMappings(otherID, primaryID)
}
}
@@ -4876,6 +4919,12 @@ func (rr *ResourceRegistry) mergeResourceData(primary *Resource, other *Resource
if primary.Ceph == nil {
primary.Ceph = other.Ceph
}
if primary.TrueNAS == nil {
primary.TrueNAS = other.TrueNAS
}
if primary.VMware == nil {
primary.VMware = other.VMware
}
// Manual links combine already-normalized resources. Preserve each metric's
// recorded source instead of flattening the linked resource to SourceAgent.
@@ -6284,6 +6333,27 @@ func mergeMetrics(
return &merged
}
// sourceStaleThreshold is how long a source may go without delivering before
// its sighting is stale: the caller's threshold for the source, else the
// default. The stale pass and the metric merge's freshness gate both read it,
// so the gate never calls a source current that the pass marks stale, or the
// reverse.
func sourceStaleThreshold(source DataSource, sighting SourceStatus, thresholds map[DataSource]time.Duration) time.Duration {
threshold := thresholds[source]
if threshold <= 0 {
threshold = defaultStaleThresholds[source]
}
if threshold <= 0 {
threshold = 120 * time.Second
}
if sighting.ExpectedUpdateIntervalSeconds > 0 {
// Slow inventory polls have their own cadence. A source must miss
// two expected intervals before its retained observation is stale.
threshold = max(threshold, 2*time.Duration(sighting.ExpectedUpdateIntervalSeconds)*time.Second)
}
return threshold
}
// metricSourceStale reports whether a source's most recent report is older than
// its stale threshold. A zero/unknown last-seen is treated as NOT stale so the
// merge never demotes a source on missing information.
@@ -6300,17 +6370,7 @@ func metricSourceStale(
if !ok || st.LastSeen.IsZero() {
return false
}
threshold := time.Duration(0)
if configured := thresholds[source]; configured > 0 {
threshold = configured
}
if threshold <= 0 {
threshold = defaultStaleThresholds[source]
}
if threshold <= 0 {
threshold = 60 * time.Second
}
return now.Sub(st.LastSeen) > threshold
return now.Sub(st.LastSeen) > sourceStaleThreshold(source, st, thresholds)
}
func mergeMetric(
@@ -247,6 +247,114 @@ func TestLeaseExpiredAgentOnLiveProxmoxNodeStaysOnline(t *testing.T) {
}
}
// A source merge into an existing row judges metric freshness by the ingest's
// stale thresholds, as the stale pass that ends the ingest does. With PVE
// polled every two minutes the monitor allows Proxmox four minutes, so a node
// polled ninety seconds ago is current while its agent, silent for seventy-five
// seconds, is stale, and the row shows the node's readings. Without thresholds
// both sources are stale by the defaults and the agent keeps its precedence.
func TestSourceMergesJudgeFreshnessByTheIngestStaleThresholds(t *testing.T) {
now := time.Now().UTC()
snapshot := models.StateSnapshot{
Nodes: []models.Node{{
ID: "homelab-pve1",
Name: "pve1",
Instance: "homelab",
Status: "online",
LinkedAgentID: "host-pve1",
CPU: 0.12,
Memory: models.Memory{Total: 64 << 30, Used: 16 << 30, Free: 48 << 30, Usage: 25},
LastSeen: now.Add(-90 * time.Second),
}},
Hosts: []models.Host{{
ID: "host-pve1",
MachineID: "machine-pve1",
Hostname: "pve1",
LinkedNodeID: "homelab-pve1",
Status: "online",
CPUUsage: 80,
Memory: models.Memory{Total: 64 << 30, Used: 48 << 30, Free: 16 << 30, Usage: 75},
LastSeen: now.Add(-75 * time.Second),
IntervalSeconds: 30,
}},
}
for _, tc := range []struct {
name string
thresholds map[DataSource]time.Duration
proxmoxPoll string
want DataSource
wantCPU float64
wantMemory float64
}{
{"configured", map[DataSource]time.Duration{SourceProxmox: 4 * time.Minute}, "online", SourceProxmox, 12, 25},
{"defaults", nil, "stale", SourceAgent, 80, 75},
} {
t.Run(tc.name, func(t *testing.T) {
rr := NewRegistry(nil)
rr.IngestSnapshotWithStaleThresholds(snapshot, tc.thresholds)
resource := onlyResourceOfType(t, rr, ResourceTypeAgent)
if got := resource.SourceStatus[SourceProxmox].Status; got != tc.proxmoxPoll {
t.Fatalf("Proxmox sighting = %q, want %q: %+v", got, tc.proxmoxPoll, resource.SourceStatus)
}
if got := resource.SourceStatus[SourceAgent].Status; got != "stale" {
t.Fatalf("agent sighting = %q, want stale", got)
}
if resource.Metrics == nil || resource.Metrics.CPU == nil || resource.Metrics.Memory == nil {
t.Fatalf("merged row lost its readings: %+v", resource.Metrics)
}
if cpu := resource.Metrics.CPU; cpu.Source != tc.want || cpu.Percent != tc.wantCPU {
t.Fatalf("CPU = %.0f%% from %s, want %.0f%% from %s", cpu.Percent, cpu.Source, tc.wantCPU, tc.want)
}
if memory := resource.Metrics.Memory; memory.Source != tc.want || memory.Percent != tc.wantMemory {
t.Fatalf("memory = %.0f%% from %s, want %.0f%% from %s", memory.Percent, memory.Source, tc.wantMemory, tc.want)
}
})
}
}
// The metric merge and the stale pass read one threshold rule, so a source
// sighting the merge treats as current is never one the pass marks stale.
func TestMetricMergeFreshnessMatchesTheStalePass(t *testing.T) {
now := time.Now().UTC()
configured := map[DataSource]time.Duration{SourceProxmox: 4 * time.Minute, SourceVMware: 5 * time.Minute}
for _, tc := range []struct {
name string
source DataSource
sighting SourceStatus
thresholds map[DataSource]time.Duration
wantStale bool
}{
{"proxmox by default", SourceProxmox, SourceStatus{LastSeen: now.Add(-90 * time.Second)}, nil, true},
{"proxmox by configuration", SourceProxmox, SourceStatus{LastSeen: now.Add(-90 * time.Second)}, configured, false},
{"vsphere by default", SourceVMware, SourceStatus{LastSeen: now.Add(-3 * time.Minute)}, nil, true},
{"vsphere by configuration", SourceVMware, SourceStatus{LastSeen: now.Add(-3 * time.Minute)}, configured, false},
{"agent without its own threshold", SourceAgent, SourceStatus{LastSeen: now.Add(-75 * time.Second)}, configured, true},
{"slow disk inventory", SourceProxmox, SourceStatus{LastSeen: now.Add(-5 * time.Minute), ExpectedUpdateIntervalSeconds: 600}, nil, false},
{"missed two disk inventory polls", SourceProxmox, SourceStatus{LastSeen: now.Add(-21 * time.Minute), ExpectedUpdateIntervalSeconds: 600}, nil, true},
} {
t.Run(tc.name, func(t *testing.T) {
status := map[DataSource]SourceStatus{tc.source: tc.sighting}
if got := metricSourceStale(now, status, tc.source, tc.thresholds); got != tc.wantStale {
t.Fatalf("metric merge stale = %v, want %v", got, tc.wantStale)
}
rr := NewRegistry(nil)
rr.resources["resource"] = &Resource{
ID: "resource",
Status: StatusOnline,
Sources: []DataSource{tc.source},
SourceStatus: map[DataSource]SourceStatus{tc.source: tc.sighting},
}
rr.MarkStale(now, tc.thresholds)
if got := rr.resources["resource"].SourceStatus[tc.source].Status == "stale"; got != tc.wantStale {
t.Fatalf("stale pass stale = %v, want %v", got, tc.wantStale)
}
})
}
}
// Docker hosts hold a shorter lease than the registry's Docker stale
// threshold, so a silent Docker host was offline for a while and then
// flipped to warning once its sighting went stale.
@@ -441,6 +549,319 @@ func TestManualLinkKeepsTheLiveSightingOfASharedSource(t *testing.T) {
}
}
// Either side of an operator link can arrive as a supplemental record: a
// vSphere VM with a Pulse agent inside, or an agent on a TrueNAS host. The
// monitor's rebuild ingests records after the snapshot, so it must join them
// there too, keep the record side's provider payload, and list what the
// resources API lists after seeding from it.
func TestManualLinksJoinRecordIngestedResourcesInMonitorRebuild(t *testing.T) {
now := time.Now().UTC()
snapshot := models.StateSnapshot{
Hosts: []models.Host{
{ID: "host-app-guest", Hostname: "app-guest", MachineID: "machine-app-guest", Status: "online", LastSeen: now},
{ID: "host-nas", Hostname: "nas-a-mgmt", MachineID: "machine-nas-a", Status: "online", LastSeen: now},
},
LastUpdate: now,
}
vmRecord := IngestRecord{
SourceID: "vc-1:vm:vm-42",
Resource: Resource{
Type: ResourceTypeVM,
Technology: "vmware",
Name: "app-guest",
Status: StatusOnline,
LastSeen: now,
VMware: &VMwareData{ConnectionID: "vc-1", ManagedObjectID: "vm-42", EntityType: "vm"},
},
Identity: ResourceIdentity{Hostnames: []string{"app-guest"}},
}
records := map[DataSource][]IngestRecord{
SourceVMware: {vmRecord},
SourceTrueNAS: {
{
SourceID: "system:tn-1",
Resource: Resource{
Type: ResourceTypeAgent,
Name: "nas-a",
Status: StatusOnline,
LastSeen: now,
TrueNAS: &TrueNASData{Hostname: "nas-a"},
},
Identity: ResourceIdentity{Hostnames: []string{"nas-a"}},
},
{
SourceID: "system:tn-1:pool:tank",
ParentSourceID: "system:tn-1",
Resource: Resource{
Type: ResourceTypeStorage,
Name: "tank",
Status: StatusOnline,
LastSeen: now,
Storage: &StorageMeta{Type: "zfs-pool", Platform: "truenas", Topology: "pool"},
},
},
},
}
for _, agentPrimary := range []bool{false, true} {
name := "record-primary"
if agentPrimary {
name = "agent-primary"
}
t.Run(name, func(t *testing.T) {
store := NewMemoryStore()
adapter := NewMonitorAdapter(NewRegistry(store))
adapter.PopulateSnapshotAndSupplemental(snapshot, records)
var vmID, guestAgentID, systemID, nasAgentID string
for _, resource := range adapter.GetAll() {
switch {
case resource.VMware != nil:
vmID = resource.ID
case resource.TrueNAS != nil:
systemID = resource.ID
case resource.Name == "app-guest":
guestAgentID = resource.ID
case resource.Name == "nas-a-mgmt":
nasAgentID = resource.ID
}
}
if vmID == "" || guestAgentID == "" || systemID == "" || nasAgentID == "" {
t.Fatalf("unlinked estate = %v, want the VM, both agents and the TrueNAS system", resourceIDs(adapter.GetAll()))
}
guestPrimary, nasPrimary := vmID, systemID
if agentPrimary {
guestPrimary, nasPrimary = guestAgentID, nasAgentID
}
for _, link := range []ResourceLink{
{ResourceA: vmID, ResourceB: guestAgentID, PrimaryID: guestPrimary},
{ResourceA: systemID, ResourceB: nasAgentID, PrimaryID: nasPrimary},
} {
if err := store.AddLink(link); err != nil {
t.Fatalf("add link: %v", err)
}
}
assertJoined := func(stage string) {
t.Helper()
resources := adapter.GetAll()
if len(resources) != 3 {
t.Fatalf("%s: resources = %v, want the linked VM, the linked TrueNAS system and its pool", stage, resourceIDs(resources))
}
// An agent inside a guest supplements the guest, whichever
// side the operator chose as primary.
guest, ok := adapter.currentRegistry().Get(vmID)
if !ok {
t.Fatalf("%s: linked vSphere VM %s missing from %v", stage, vmID, resourceIDs(resources))
}
if guest.Type != ResourceTypeVM || guest.VMware == nil || guest.Agent == nil {
t.Fatalf("%s: linked VM type=%s vmware=%t agent=%t, want the VM carrying its agent", stage, guest.Type, guest.VMware != nil, guest.Agent != nil)
}
if !slices.Contains(guest.Sources, SourceVMware) || !slices.Contains(guest.Sources, SourceAgent) {
t.Fatalf("%s: linked VM sources = %v, want vmware and agent", stage, guest.Sources)
}
system, ok := adapter.currentRegistry().Get(nasPrimary)
if !ok {
t.Fatalf("%s: linked TrueNAS system %s missing from %v", stage, nasPrimary, resourceIDs(resources))
}
if system.TrueNAS == nil || system.Agent == nil {
t.Fatalf("%s: linked system truenas=%t agent=%t, want the primary carrying both payloads", stage, system.TrueNAS != nil, system.Agent != nil)
}
if !slices.Contains(system.Sources, SourceTrueNAS) || !slices.Contains(system.Sources, SourceAgent) {
t.Fatalf("%s: linked system sources = %v, want truenas and agent", stage, system.Sources)
}
pool := onlyResourceOfType(t, adapter.currentRegistry(), ResourceTypeStorage)
if pool.ParentID == nil || *pool.ParentID != nasPrimary {
t.Fatalf("%s: pool parent = %v, want the linked system %s", stage, pool.ParentID, nasPrimary)
}
// Identity pins persisted by the rebuild must not re-key a link
// onto its own primary.
links, err := store.GetLinks()
if err != nil {
t.Fatalf("%s: get links: %v", stage, err)
}
for _, link := range links {
if link.ResourceA == link.ResourceB {
t.Fatalf("%s: link collapsed onto %s", stage, link.ResourceA)
}
}
// The resources API seeds its registry from this listing
// through resource ingest and must agree with it.
rest := NewRegistry(store)
rest.IngestResources(resources)
if got, want := resourceIDs(rest.List()), resourceIDs(resources); !slices.Equal(got, want) {
t.Fatalf("%s: a registry seeded from the monitor lists %v, monitor lists %v", stage, got, want)
}
}
next := snapshot
next.LastUpdate = now.Add(time.Second)
adapter.PopulateSnapshotAndSupplemental(next, records)
assertJoined("rebuild")
adapter.PopulateSupplementalRecords(SourceVMware, records[SourceVMware])
adapter.PopulateSupplementalRecords(SourceTrueNAS, records[SourceTrueNAS])
assertJoined("supplemental refresh")
next.LastUpdate = now.Add(2 * time.Second)
adapter.PopulateSnapshotAndSupplemental(next, records)
assertJoined("second rebuild")
})
}
}
// A Proxmox node that absorbs a linked agent carries the agent's machine key,
// so the rebuild pins the node with it. The agent is still observed: that pin
// must not succeed the agent's ID onto the node, which re-keyed the link onto
// the node itself and split the pair on the next rebuild.
func TestManualLinkKeepsItsEndpointsThroughIdentityPinPersistence(t *testing.T) {
now := time.Now().UTC()
snapshot := models.StateSnapshot{
Nodes: []models.Node{{
ID: "lab-pve1", Name: "pve1", Instance: "lab",
Host: "https://192.0.2.10:8006", Status: "online", LastSeen: now,
}},
Hosts: []models.Host{{
ID: "host-box", Hostname: "box-x", MachineID: "machine-box", Status: "online", LastSeen: now,
}},
LastUpdate: now,
}
store := NewMemoryStore()
adapter := NewMonitorAdapter(NewRegistry(store))
adapter.PopulateSnapshotAndSupplemental(snapshot, nil)
var nodeID, agentID string
for _, resource := range adapter.GetAll() {
if resource.Proxmox != nil {
nodeID = resource.ID
} else if resource.Agent != nil {
agentID = resource.ID
}
}
if nodeID == "" || agentID == "" || nodeID == agentID {
t.Fatalf("unlinked estate = %v, want a node and a separate agent", resourceIDs(adapter.GetAll()))
}
if err := store.AddLink(ResourceLink{ResourceA: nodeID, ResourceB: agentID, PrimaryID: nodeID}); err != nil {
t.Fatalf("add link: %v", err)
}
for rebuild := 1; rebuild <= 3; rebuild++ {
next := snapshot
next.LastUpdate = now.Add(time.Duration(rebuild) * time.Second)
adapter.PopulateSnapshotAndSupplemental(next, nil)
if got := resourceIDs(adapter.GetAll()); !slices.Equal(got, []string{nodeID}) {
t.Fatalf("rebuild %d: resources = %v, want only the linked node %s", rebuild, got, nodeID)
}
links, err := store.GetLinks()
if err != nil {
t.Fatalf("get links: %v", err)
}
if len(links) != 1 || links[0].ResourceA != nodeID || links[0].ResourceB != agentID {
t.Fatalf("rebuild %d: links = %+v, want %s -> %s kept", rebuild, links, nodeID, agentID)
}
}
}
// A record may name an older canonical ID as superseded on every ingest. When
// that ID belongs to a resource a manual link folded into the record's own
// resource, it is still observed: succeeding it on a later refresh would
// re-key the link onto its primary.
func TestRecordDeclaredSuccessionSparesALinkFoldedResource(t *testing.T) {
now := time.Now().UTC()
snapshot := models.StateSnapshot{
Hosts: []models.Host{{ID: "host-nas", Hostname: "nas-a-mgmt", MachineID: "machine-nas-a", Status: "online", LastSeen: now}},
LastUpdate: now,
}
agentID := MachineIdentityCanonicalID(ResourceTypeAgent, "machine-nas-a")
records := []IngestRecord{{
SourceID: "system:tn-1",
SupersededCanonicalIDs: []string{agentID},
Resource: Resource{
Type: ResourceTypeAgent,
Name: "nas-a",
Status: StatusOnline,
LastSeen: now,
TrueNAS: &TrueNASData{Hostname: "nas-a"},
},
Identity: ResourceIdentity{Hostnames: []string{"nas-a"}},
}}
unlinked := NewMonitorAdapter(NewRegistry(nil))
unlinked.PopulateSnapshotAndSupplemental(snapshot, map[DataSource][]IngestRecord{SourceTrueNAS: records})
var systemID string
for _, resource := range unlinked.GetAll() {
if resource.TrueNAS != nil {
systemID = resource.ID
}
}
if systemID == "" || !slices.Contains(resourceIDs(unlinked.GetAll()), agentID) {
t.Fatalf("unlinked estate = %v, want agent %s and a TrueNAS system", resourceIDs(unlinked.GetAll()), agentID)
}
store := NewMemoryStore()
if err := store.AddLink(ResourceLink{ResourceA: systemID, ResourceB: agentID, PrimaryID: systemID}); err != nil {
t.Fatalf("add link: %v", err)
}
adapter := NewMonitorAdapter(NewRegistry(store))
adapter.PopulateSnapshotAndSupplemental(snapshot, map[DataSource][]IngestRecord{SourceTrueNAS: records})
adapter.PopulateSupplementalRecords(SourceTrueNAS, records)
if got := resourceIDs(adapter.GetAll()); !slices.Equal(got, []string{systemID}) {
t.Fatalf("resources = %v, want only the linked system %s", got, systemID)
}
links, err := store.GetLinks()
if err != nil {
t.Fatalf("get links: %v", err)
}
if len(links) != 1 || links[0].ResourceA != systemID || links[0].ResourceB != agentID {
t.Fatalf("links = %+v, want %s -> %s kept", links, systemID, agentID)
}
}
// A link merge folds the other resource's provider payloads into the primary
// where the primary has none, whichever side the operator chose.
func TestManualLinkKeepsProviderPayloadsThePrimaryLacks(t *testing.T) {
now := time.Now().UTC()
store := NewMemoryStore()
for _, link := range []ResourceLink{
{ResourceA: "agent-nas", ResourceB: "truenas-system", PrimaryID: "agent-nas"},
{ResourceA: "agent-esxi", ResourceB: "vmware-host", PrimaryID: "agent-esxi"},
} {
if err := store.AddLink(link); err != nil {
t.Fatalf("add link: %v", err)
}
}
host := func(id, name string, sources ...DataSource) Resource {
status := make(map[DataSource]SourceStatus, len(sources))
for _, source := range sources {
status[source] = SourceStatus{Status: "online", LastSeen: now}
}
return Resource{ID: id, Type: ResourceTypeAgent, Name: name, Status: StatusOnline, LastSeen: now, Sources: sources, SourceStatus: status}
}
agentNAS := host("agent-nas", "nas-a-mgmt", SourceAgent)
agentNAS.Agent = &AgentData{AgentID: "agent-nas", Hostname: "nas-a-mgmt"}
system := host("truenas-system", "nas-a", SourceTrueNAS)
system.TrueNAS = &TrueNASData{Hostname: "nas-a"}
agentESXi := host("agent-esxi", "esxi-01", SourceAgent)
agentESXi.Agent = &AgentData{AgentID: "agent-esxi", Hostname: "esxi-01"}
esxi := host("vmware-host", "esxi-01.lab", SourceVMware)
esxi.VMware = &VMwareData{ConnectionID: "vc-1", ManagedObjectID: "host-101", EntityType: "host"}
rr := NewRegistry(store)
rr.IngestResources([]Resource{agentNAS, system, agentESXi, esxi})
if got := resourceIDs(rr.List()); !slices.Equal(got, []string{"agent-esxi", "agent-nas"}) {
t.Fatalf("resources = %v, want the two linked primaries", got)
}
if nas, _ := rr.Get("agent-nas"); nas.TrueNAS == nil || nas.Agent == nil {
t.Fatalf("linked TrueNAS system truenas=%t agent=%t, want both payloads", nas.TrueNAS != nil, nas.Agent != nil)
}
if host, _ := rr.Get("agent-esxi"); host.VMware == nil || host.Agent == nil {
t.Fatalf("linked vSphere host vmware=%t agent=%t, want both payloads", host.VMware != nil, host.Agent != nil)
}
}
func TestLeaseExpiredKubernetesClusterStaysOffline(t *testing.T) {
rr := NewRegistry(nil)
rr.IngestSnapshot(models.StateSnapshot{
+3
View File
@@ -2,6 +2,7 @@ package reporting
import (
"fmt"
"net/http"
"strings"
"time"
)
@@ -33,6 +34,7 @@ type PerformanceReportDefinition struct {
Title string `json:"title"`
Description string `json:"description"`
SingleResourceEndpoint string `json:"singleResourceEndpoint"`
SingleResourceMethod string `json:"singleResourceMethod"`
MultiResourceEndpoint string `json:"multiResourceEndpoint"`
SingleFilenamePrefix string `json:"singleFilenamePrefix"`
SingleFilenameSubject string `json:"singleFilenameSubject"`
@@ -150,6 +152,7 @@ func DescribePerformanceReport() PerformanceReportDefinition {
Title: "Performance Reports",
Description: "Generate PDF summaries or CSV metric exports from historical monitoring data for one or more selected resources.",
SingleResourceEndpoint: "/api/admin/reports/generate",
SingleResourceMethod: http.MethodPost,
MultiResourceEndpoint: "/api/admin/reports/generate-multi",
SingleFilenamePrefix: "report",
SingleFilenameSubject: FilenameSubjectResourceID,