From 9c3d96cab2c92e2fe89f76ac1f1429d9fbe3b7ef Mon Sep 17 00:00:00 2001 From: rcourtman Date: Sun, 19 Apr 2026 11:42:53 +0100 Subject: [PATCH] Add unified connections API (list + probe) with Disabled flag Introduces GET /api/connections and POST /api/connections/probe as the backend half of the one-ledger / one-editor connection redesign. - GET /api/connections aggregates PVE/PBS/PMG/VMware/TrueNAS/agent rows into a unified Connection shape with derived state (active, paused, unauthorized, unreachable, stale, pending) computed from in-memory scheduler health plus agent Host.LastSeen. No new persisted state. - POST /api/connections/probe fingerprints a host across the five supported products in parallel (2s dial + 1s read, 3s total, max 5 concurrent). Admin-gated (RequireAdmin + ScopeSettingsWrite) to block unauthenticated SSRF against internal hosts. - Disabled bool on PVEInstance/PBSInstance/PMGInstance (zero-value = enabled, preserves existing nodes.json); pollers skip disabled instances at client init, reconnect, and per-node iteration. - NodeConfigRequest/Response gain Enabled; write path translates *bool -> Disabled so omitted field leaves state untouched. - ConnectionsAPI frontend client (list/probe) typed off the Go shape. Contracts updated: api-contracts, monitoring, agent-lifecycle, performance-and-scalability, storage-recovery. Proofs added: contract_test.go JSON snapshot for Connection and ProbeResponse, monitoring guardrails for the Disabled-skip behavior, and a vitest mock-client test for ConnectionsAPI. Frontend editor / drawer / table rewrite lands in a separate block. --- .../v6/internal/subsystems/agent-lifecycle.md | 10 + .../v6/internal/subsystems/api-contracts.md | 18 + .../v6/internal/subsystems/monitoring.md | 6 + .../subsystems/performance-and-scalability.md | 10 + .../internal/subsystems/storage-recovery.md | 12 + .../src/api/__tests__/connections.test.ts | 72 ++++ frontend-modern/src/api/connections.ts | 84 +++++ internal/api/config_handlers.go | 7 +- internal/api/config_node_handlers.go | 9 + internal/api/connections_aggregator.go | 349 ++++++++++++++++++ internal/api/connections_aggregator_test.go | 211 +++++++++++ internal/api/connections_handlers.go | 111 ++++++ internal/api/connections_probe.go | 328 ++++++++++++++++ internal/api/connections_probe_test.go | 333 +++++++++++++++++ internal/api/connections_types.go | 84 +++++ internal/api/contract_test.go | 53 +++ internal/api/route_inventory_test.go | 4 + internal/api/router.go | 6 + internal/api/router_routes_registration.go | 26 ++ internal/config/config.go | 11 + .../monitoring/canonical_guardrails_test.go | 40 ++ internal/monitoring/monitor_client_init.go | 12 + .../monitoring/monitor_client_reconnect.go | 12 +- .../monitoring/monitor_pbs_coverage_test.go | 31 ++ internal/monitoring/monitor_pbs_pmg.go | 12 + internal/monitoring/monitor_pve.go | 6 + .../release_control/subsystem_lookup_test.py | 4 +- 27 files changed, 1856 insertions(+), 5 deletions(-) create mode 100644 frontend-modern/src/api/__tests__/connections.test.ts create mode 100644 frontend-modern/src/api/connections.ts create mode 100644 internal/api/connections_aggregator.go create mode 100644 internal/api/connections_aggregator_test.go create mode 100644 internal/api/connections_handlers.go create mode 100644 internal/api/connections_probe.go create mode 100644 internal/api/connections_probe_test.go create mode 100644 internal/api/connections_types.go diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index a55ef1ce2..991ee7b2f 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -303,6 +303,16 @@ an add-only capacity posture. VMware must not collect external vCenter inventory before that canonical capacity view is safe, so fleet/setup surfaces cannot bypass the monitored-system accounting boundary through direct API writes. + The same lifecycle-adjacent platform-connections boundary now also owns + the unified connections ledger (`GET /api/connections`) and address + probe (`POST /api/connections/probe`). Lifecycle surfaces may observe + agent `Host.LastSeen`-backed rows on that ledger, but must not + reinterpret derived `state` (active/paused/unauthorized/unreachable/ + stale/pending) as install authority or treat the probe response as + enrollment state; ledger writes still flow through the per-type config + endpoints that own admission checks, and the `Disabled` flag on + PVE/PBS/PMG surfaced by that ledger must remain a pause-only signal + rather than an installer pre-flight gate. ## Forbidden Paths diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index ec71c8f65..e77df66d2 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -79,6 +79,11 @@ Own canonical runtime payload shapes between backend and frontend. 54. `internal/api/security_status_capabilities.go` 55. `internal/api/demo_middleware.go` 56. `frontend-modern/src/stores/aiRuntimeState.ts` +57. `internal/api/connections_types.go` +58. `internal/api/connections_aggregator.go` +59. `internal/api/connections_handlers.go` +60. `internal/api/connections_probe.go` +61. `frontend-modern/src/api/connections.ts` ## Shared Boundaries @@ -311,6 +316,19 @@ the canonical monitored-system blocked payload. presentation is active, so VMware, storage, and infrastructure series stay aligned with `/api/resources` and `/api/state` instead of drifting onto the live store-backed graph. +39. Route the unified connections ledger and address probe through + `internal/api/connections_types.go`, + `internal/api/connections_aggregator.go`, + `internal/api/connections_handlers.go`, + `internal/api/connections_probe.go`, and + `frontend-modern/src/api/connections.ts` together so `GET /api/connections` + and `POST /api/connections/probe` stay on one canonical payload shape + instead of re-deriving state from per-type config stores in the frontend. + State must remain a derived field sourced from in-memory scheduler health + (`monitoring.Monitor.SchedulerHealth()`) plus agent `Host.LastSeen`; the + endpoint must not introduce new persisted per-connection state. The probe + endpoint must remain admin-gated (`RequireAdmin` + `ScopeSettingsWrite`) + to block unauthenticated SSRF against internal hosts. ## Forbidden Paths diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 484a105f2..2526df2f1 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -71,6 +71,12 @@ truth for live infrastructure data. 10. Add or change mock chart synthesis, seeded history continuity, or mock-owned chart fallbacks through `internal/monitoring/mock_metrics_history.go` and `internal/monitoring/mock_chart_history.go` +11. Honor the per-instance `Disabled` flag on PVE/PBS/PMG at poller client + init, reconnect, and per-node iteration so disabled connections do not + drive API calls, scheduler health, or surface ingest. Zero-value + `Disabled=false` must remain the migration-safe default for existing + `nodes.json` content; the poller must never create a client or mark an + instance reachable when `Disabled` is true. ## Forbidden Paths diff --git a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md index 3c37dad54..501c4b02f 100644 --- a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md +++ b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md @@ -275,6 +275,16 @@ regression protection. colors, but they must not fall back to inline `style=` attributes on the public shell just to express virtualization spacers, alert accents, or workload metric bars. +36. Keep the unified connections ledger off the polling hot path. Changes to + `internal/api/router.go` that register `/api/connections` and + `/api/connections/probe` must keep those routes off the monitoring fan-out + budget: `GET /api/connections` must resolve purely from + `monitoring.Monitor.SchedulerHealth()` plus existing per-type config + stores without triggering any live network probes, and + `POST /api/connections/probe` must remain bounded at 3s total / + 2s dial / 1s read with at most 5 concurrent fingerprints so the probe + endpoint cannot be repurposed into a slow-leak scanner that starves the + dashboard hot path. ## Forbidden Paths diff --git a/docs/release-control/v6/internal/subsystems/storage-recovery.md b/docs/release-control/v6/internal/subsystems/storage-recovery.md index 2083165e8..9ce685789 100644 --- a/docs/release-control/v6/internal/subsystems/storage-recovery.md +++ b/docs/release-control/v6/internal/subsystems/storage-recovery.md @@ -343,6 +343,18 @@ querying, and the operator-facing storage health presentation layer. canonical storage page model, recovery presenters, and shared summary caches. Header chrome must not become a second owner for storage filters, recovery posture, commercial purchase state, or transport selection. +39. Keep the unified connections ledger owner-neutral toward storage and + recovery. Shared `internal/api/router.go` may mount the + `/api/connections` and `/api/connections/probe` routes alongside the + existing storage/recovery-adjacent API surfaces, and + `internal/api/config_handlers.go` and `internal/api/config_node_handlers.go` + may carry the new per-instance `Enabled`/`Disabled` round-trip, but + storage and recovery consumers must not reinterpret the derived + connection `state` (active/paused/unauthorized/unreachable/stale/pending) + as storage health, backup-job posture, or recovery verification state; + storage and recovery UI must keep sourcing those signals from their + existing canonical page models instead of polling the connections + ledger for per-datastore or per-backup truth. ## Forbidden Paths diff --git a/frontend-modern/src/api/__tests__/connections.test.ts b/frontend-modern/src/api/__tests__/connections.test.ts new file mode 100644 index 000000000..a6d1c225c --- /dev/null +++ b/frontend-modern/src/api/__tests__/connections.test.ts @@ -0,0 +1,72 @@ +import { describe, expect, it, vi, beforeEach } from 'vitest'; +import { ConnectionsAPI, type Connection, type ProbeResponse } from '../connections'; +import { apiFetchJSON } from '@/utils/apiClient'; + +vi.mock('@/utils/apiClient', () => ({ + apiFetchJSON: vi.fn(), +})); + +const mockedApiFetchJSON = vi.mocked(apiFetchJSON); + +describe('ConnectionsAPI', () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it('list() calls GET /api/connections and returns the connections array', async () => { + const connections: Connection[] = [ + { + id: 'pve-lab', + type: 'pve', + name: 'lab', + address: 'https://pve.lab:8006', + state: 'active', + stateReason: '', + enabled: true, + surfaces: ['vms', 'containers'], + scope: { vms: true, containers: true }, + lastSeen: '2026-04-19T10:00:00Z', + lastError: null, + source: 'manual', + capabilities: { supportsPause: true, supportsScope: true, supportsTest: true }, + }, + ]; + mockedApiFetchJSON.mockResolvedValueOnce({ connections }); + + const result = await ConnectionsAPI.list(); + + expect(mockedApiFetchJSON).toHaveBeenCalledWith('/api/connections'); + expect(result).toEqual(connections); + }); + + it('list() returns [] when the backend omits the connections field', async () => { + mockedApiFetchJSON.mockResolvedValueOnce({} as { connections?: Connection[] }); + + const result = await ConnectionsAPI.list(); + + expect(result).toEqual([]); + }); + + it('probe() POSTs the address JSON and returns the candidates envelope', async () => { + const response: ProbeResponse = { + candidates: [ + { + type: 'pve', + host: 'https://pve.lab:8006', + port: 8006, + hints: { product: 'Proxmox VE', version: '8.2.4' }, + }, + ], + probedMs: 812, + }; + mockedApiFetchJSON.mockResolvedValueOnce(response); + + const result = await ConnectionsAPI.probe('pve.lab'); + + expect(mockedApiFetchJSON).toHaveBeenCalledWith('/api/connections/probe', { + method: 'POST', + body: JSON.stringify({ address: 'pve.lab' }), + }); + expect(result).toEqual(response); + }); +}); diff --git a/frontend-modern/src/api/connections.ts b/frontend-modern/src/api/connections.ts new file mode 100644 index 000000000..11267d0d9 --- /dev/null +++ b/frontend-modern/src/api/connections.ts @@ -0,0 +1,84 @@ +import { apiFetchJSON } from '@/utils/apiClient'; + +export type ConnectionType = + | 'pve' + | 'pbs' + | 'pmg' + | 'vmware' + | 'truenas' + | 'agent' + | 'docker' + | 'kubernetes'; + +export type ConnectionState = + | 'active' + | 'paused' + | 'unauthorized' + | 'unreachable' + | 'stale' + | 'pending'; + +export type ConnectionSource = 'manual' | 'agent' | 'script'; + +export interface ConnectionCapabilities { + supportsPause: boolean; + supportsScope: boolean; + supportsTest: boolean; +} + +export interface ConnectionError { + message: string; + at: string; +} + +export interface Connection { + id: string; + type: ConnectionType; + name: string; + address: string; + state: ConnectionState; + stateReason: string; + enabled: boolean; + surfaces: string[]; + scope: Record; + lastSeen: string | null; + lastError: ConnectionError | null; + source: ConnectionSource; + capabilities: ConnectionCapabilities; +} + +export interface ConnectionsListResponse { + connections: Connection[]; +} + +export interface ProbeRequest { + address: string; +} + +export interface ProbeCandidate { + type: ConnectionType; + host: string; + port: number; + hints?: Record; +} + +export interface ProbeResponse { + candidates: ProbeCandidate[]; + probedMs: number; +} + +export class ConnectionsAPI { + private static readonly baseUrl = '/api/connections'; + + static async list(): Promise { + const response: ConnectionsListResponse = await apiFetchJSON(this.baseUrl); + return response.connections ?? []; + } + + static async probe(address: string): Promise { + return apiFetchJSON(`${this.baseUrl}/probe`, { + method: 'POST', + body: JSON.stringify({ address } satisfies ProbeRequest), + }); + } +} diff --git a/internal/api/config_handlers.go b/internal/api/config_handlers.go index 964d5faff..d4cb1339f 100644 --- a/internal/api/config_handlers.go +++ b/internal/api/config_handlers.go @@ -642,6 +642,7 @@ type NodeConfigRequest struct { MonitorQueues *bool `json:"monitorQueues,omitempty"` // PMG only MonitorQuarantine *bool `json:"monitorQuarantine,omitempty"` // PMG only MonitorDomainStats *bool `json:"monitorDomainStats,omitempty"` // PMG only + Enabled *bool `json:"enabled,omitempty"` // Lifecycle toggle; nil on update preserves current } // NodeResponse represents a node in API responses @@ -674,7 +675,8 @@ type NodeResponse struct { MonitorQueues bool `json:"monitorQueues,omitempty"` MonitorQuarantine bool `json:"monitorQuarantine,omitempty"` MonitorDomainStats bool `json:"monitorDomainStats,omitempty"` - Status string `json:"status"` // "connected", "disconnected", "error" + Enabled bool `json:"enabled"` // Lifecycle; false = paused + Status string `json:"status"` // "connected", "disconnected", "error" IsCluster bool `json:"isCluster,omitempty"` ClusterName string `json:"clusterName,omitempty"` ClusterEndpoints []ClusterEndpointResponse `json:"clusterEndpoints"` @@ -1296,6 +1298,7 @@ func (h *ConfigHandlers) GetAllNodesForAPI(ctx context.Context) []NodeResponse { MonitorPhysicalDisks: pve.MonitorPhysicalDisks, PhysicalDiskPollingMinutes: pve.PhysicalDiskPollingMinutes, TemperatureMonitoringEnabled: pve.TemperatureMonitoringEnabled, + Enabled: !pve.Disabled, Status: h.getNodeStatus(ctx, "pve", pve.Name), IsCluster: pve.IsCluster, ClusterName: pve.ClusterName, @@ -1326,6 +1329,7 @@ func (h *ConfigHandlers) GetAllNodesForAPI(ctx context.Context) []NodeResponse { MonitorPruneJobs: pbs.MonitorPruneJobs, MonitorGarbageJobs: pbs.MonitorGarbageJobs, ExcludeDatastores: pbs.ExcludeDatastores, + Enabled: !pbs.Disabled, Status: h.getNodeStatus(ctx, "pbs", pbs.Name), Source: pbs.Source, }.NormalizeCollections() @@ -1356,6 +1360,7 @@ func (h *ConfigHandlers) GetAllNodesForAPI(ctx context.Context) []NodeResponse { MonitorQueues: pmgInst.MonitorQueues, MonitorQuarantine: pmgInst.MonitorQuarantine, MonitorDomainStats: pmgInst.MonitorDomainStats, + Enabled: !pmgInst.Disabled, Status: h.getNodeStatus(ctx, "pmg", pmgInst.Name), }.NormalizeCollections() nodes = append(nodes, node) diff --git a/internal/api/config_node_handlers.go b/internal/api/config_node_handlers.go index 5ca51344b..c724cbba9 100644 --- a/internal/api/config_node_handlers.go +++ b/internal/api/config_node_handlers.go @@ -1208,6 +1208,9 @@ func (h *ConfigHandlers) handleUpdateNode(w http.ResponseWriter, r *http.Request if req.TemperatureMonitoringEnabled != nil { updated.TemperatureMonitoringEnabled = req.TemperatureMonitoringEnabled } + if req.Enabled != nil { + updated.Disabled = !*req.Enabled + } if enforceMonitoredSystemLimitForConfigReplacement( w, @@ -1312,6 +1315,9 @@ func (h *ConfigHandlers) handleUpdateNode(w http.ResponseWriter, r *http.Request if req.ExcludeDatastores != nil { updated.ExcludeDatastores = req.ExcludeDatastores } + if req.Enabled != nil { + updated.Disabled = !*req.Enabled + } if enforceMonitoredSystemLimitForConfigReplacement( w, @@ -1400,6 +1406,9 @@ func (h *ConfigHandlers) handleUpdateNode(w http.ResponseWriter, r *http.Request if req.TemperatureMonitoringEnabled != nil { updated.TemperatureMonitoringEnabled = req.TemperatureMonitoringEnabled } + if req.Enabled != nil { + updated.Disabled = !*req.Enabled + } if enforceMonitoredSystemLimitForConfigReplacement( w, diff --git a/internal/api/connections_aggregator.go b/internal/api/connections_aggregator.go new file mode 100644 index 000000000..aa5070e18 --- /dev/null +++ b/internal/api/connections_aggregator.go @@ -0,0 +1,349 @@ +package api + +import ( + "fmt" + "regexp" + "sort" + "strings" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/internal/monitoring" +) + +// connectionStaleThreshold is the baseline "haven't heard from this connection +// recently" cutoff used to transition `active` → `stale`. Per-type poll +// intervals vary (PVE 5-10s, TrueNAS 60s, agents 30-60s); 2 minutes sits +// comfortably above 2× the slowest default so we don't flap on a single +// dropped tick. Refined later once per-type intervals become first-class. +const connectionStaleThreshold = 2 * time.Minute + +// connectionAuthErrorPattern matches the error strings pollers surface when a +// credential is wrong or the token lacks scope. Kept centrally so +// state-derivation stays the same across types. +var connectionAuthErrorPattern = regexp.MustCompile(`(?i)401|403|unauthori[sz]ed|forbidden|authentication|permission denied|invalid (credentials|token|api key)`) + +// aggregatorInputs bundles everything the aggregator reads. Separating inputs +// from the handler makes the aggregator unit-testable without spinning up a +// monitor or persistence layer. +type aggregatorInputs struct { + pveInstances []config.PVEInstance + pbsInstances []config.PBSInstance + pmgInstances []config.PMGInstance + vmwareInstances []config.VMwareVCenterInstance + truenasInstances []config.TrueNASInstance + hosts []models.Host + instanceHealth map[string]monitoring.InstanceHealth + now time.Time +} + +// buildConnections produces a stable, sorted list of connection rows across +// every supported infrastructure type. The function is pure — it does not +// perform any I/O and does not mutate its inputs. +func buildConnections(in aggregatorInputs) []Connection { + now := in.now + if now.IsZero() { + now = time.Now() + } + + out := make([]Connection, 0, + len(in.pveInstances)+len(in.pbsInstances)+len(in.pmgInstances)+ + len(in.vmwareInstances)+len(in.truenasInstances)+len(in.hosts)) + + for _, pve := range in.pveInstances { + out = append(out, buildPVEConnection(pve, in.instanceHealth, now)) + } + for _, pbs := range in.pbsInstances { + out = append(out, buildPBSConnection(pbs, in.instanceHealth, now)) + } + for _, pmg := range in.pmgInstances { + out = append(out, buildPMGConnection(pmg, in.instanceHealth, now)) + } + for _, vmw := range in.vmwareInstances { + out = append(out, buildVMwareConnection(vmw, in.instanceHealth, now)) + } + for _, tn := range in.truenasInstances { + out = append(out, buildTrueNASConnection(tn, in.instanceHealth, now)) + } + for _, host := range in.hosts { + out = append(out, buildAgentConnection(host, now)) + } + + sort.Slice(out, func(i, j int) bool { + if out[i].Type != out[j].Type { + return out[i].Type < out[j].Type + } + return strings.ToLower(out[i].Name) < strings.ToLower(out[j].Name) + }) + + return out +} + +func buildPVEConnection(inst config.PVEInstance, health map[string]monitoring.InstanceHealth, now time.Time) Connection { + enabled := !inst.Disabled + surfaces := []string{"vms", "containers", "storage", "backups"} + scope := map[string]bool{ + "vms": inst.MonitorVMs, + "containers": inst.MonitorContainers, + "storage": inst.MonitorStorage, + "backups": inst.MonitorBackups, + } + h := health["pve::"+inst.Name] + state, reason, lastSeen, lastError := deriveConnectionState(enabled, h, now) + return Connection{ + ID: "pve:" + inst.Name, + Type: ConnectionTypePVE, + Name: inst.Name, + Address: inst.Host, + State: state, + StateReason: reason, + Enabled: enabled, + Surfaces: surfaces, + Scope: scope, + LastSeen: lastSeen, + LastError: lastError, + Source: sourceFromString(inst.Source), + Capabilities: ConnectionCapabilities{SupportsPause: true, SupportsScope: true, SupportsTest: true}, + } +} + +func buildPBSConnection(inst config.PBSInstance, health map[string]monitoring.InstanceHealth, now time.Time) Connection { + enabled := !inst.Disabled + surfaces := []string{"backups", "datastores", "syncJobs", "verifyJobs", "pruneJobs", "garbageJobs"} + scope := map[string]bool{ + "backups": inst.MonitorBackups, + "datastores": inst.MonitorDatastores, + "syncJobs": inst.MonitorSyncJobs, + "verifyJobs": inst.MonitorVerifyJobs, + "pruneJobs": inst.MonitorPruneJobs, + "garbageJobs": inst.MonitorGarbageJobs, + } + h := health["pbs::"+inst.Name] + state, reason, lastSeen, lastError := deriveConnectionState(enabled, h, now) + return Connection{ + ID: "pbs:" + inst.Name, + Type: ConnectionTypePBS, + Name: inst.Name, + Address: inst.Host, + State: state, + StateReason: reason, + Enabled: enabled, + Surfaces: surfaces, + Scope: scope, + LastSeen: lastSeen, + LastError: lastError, + Source: sourceFromString(inst.Source), + Capabilities: ConnectionCapabilities{SupportsPause: true, SupportsScope: true, SupportsTest: true}, + } +} + +func buildPMGConnection(inst config.PMGInstance, health map[string]monitoring.InstanceHealth, now time.Time) Connection { + enabled := !inst.Disabled + surfaces := []string{"mailStats", "queues", "quarantine", "domainStats"} + scope := map[string]bool{ + "mailStats": inst.MonitorMailStats, + "queues": inst.MonitorQueues, + "quarantine": inst.MonitorQuarantine, + "domainStats": inst.MonitorDomainStats, + } + h := health["pmg::"+inst.Name] + state, reason, lastSeen, lastError := deriveConnectionState(enabled, h, now) + return Connection{ + ID: "pmg:" + inst.Name, + Type: ConnectionTypePMG, + Name: inst.Name, + Address: inst.Host, + State: state, + StateReason: reason, + Enabled: enabled, + Surfaces: surfaces, + Scope: scope, + LastSeen: lastSeen, + LastError: lastError, + Source: ConnectionSourceManual, + Capabilities: ConnectionCapabilities{SupportsPause: true, SupportsScope: true, SupportsTest: true}, + } +} + +func buildVMwareConnection(inst config.VMwareVCenterInstance, health map[string]monitoring.InstanceHealth, now time.Time) Connection { + enabled := inst.Enabled + surfaces := []string{"vms", "hosts", "datastores"} + h := health["vmware::"+inst.ID] + state, reason, lastSeen, lastError := deriveConnectionState(enabled, h, now) + port := inst.Port + if port == 0 { + port = 443 + } + return Connection{ + ID: "vmware:" + inst.ID, + Type: ConnectionTypeVMware, + Name: inst.Name, + Address: fmt.Sprintf("https://%s:%d", inst.Host, port), + State: state, + StateReason: reason, + Enabled: enabled, + Surfaces: surfaces, + Scope: map[string]bool{"vms": true, "hosts": true, "datastores": true}, + LastSeen: lastSeen, + LastError: lastError, + Source: ConnectionSourceManual, + Capabilities: ConnectionCapabilities{SupportsPause: true, SupportsScope: false, SupportsTest: true}, + } +} + +func buildTrueNASConnection(inst config.TrueNASInstance, health map[string]monitoring.InstanceHealth, now time.Time) Connection { + enabled := inst.Enabled + surfaces := []string{"datasets", "pools", "replication"} + h := health["truenas::"+inst.ID] + state, reason, lastSeen, lastError := deriveConnectionState(enabled, h, now) + scheme := "https" + if !inst.UseHTTPS { + scheme = "http" + } + port := inst.Port + if port == 0 { + if inst.UseHTTPS { + port = 443 + } else { + port = 80 + } + } + return Connection{ + ID: "truenas:" + inst.ID, + Type: ConnectionTypeTrueNAS, + Name: inst.Name, + Address: fmt.Sprintf("%s://%s:%d", scheme, inst.Host, port), + State: state, + StateReason: reason, + Enabled: enabled, + Surfaces: surfaces, + Scope: map[string]bool{"datasets": true, "pools": true, "replication": true}, + LastSeen: lastSeen, + LastError: lastError, + Source: ConnectionSourceManual, + Capabilities: ConnectionCapabilities{SupportsPause: true, SupportsScope: false, SupportsTest: true}, + } +} + +// buildAgentConnection derives a connection row from an agent Host record. +// Agents have no pause toggle and no scope — reports are all-or-nothing — +// so capability flags are off. +func buildAgentConnection(host models.Host, now time.Time) Connection { + name := host.DisplayName + if strings.TrimSpace(name) == "" { + name = host.Hostname + } + if strings.TrimSpace(name) == "" { + name = host.ID + } + address := host.Hostname + if strings.TrimSpace(address) == "" { + address = host.ReportIP + } + + var lastSeen *time.Time + if !host.LastSeen.IsZero() { + t := host.LastSeen + lastSeen = &t + } + + state := ConnectionStatePending + reason := "" + switch { + case lastSeen == nil: + state = ConnectionStatePending + case now.Sub(*lastSeen) > connectionStaleThreshold: + state = ConnectionStateStale + reason = fmt.Sprintf("no heartbeat in %s", now.Sub(*lastSeen).Round(time.Second)) + default: + state = ConnectionStateActive + } + + return Connection{ + ID: "agent:" + host.ID, + Type: ConnectionTypeAgent, + Name: name, + Address: address, + State: state, + StateReason: reason, + Enabled: true, + Surfaces: []string{"host"}, + Scope: map[string]bool{"host": true}, + LastSeen: lastSeen, + LastError: nil, + Source: ConnectionSourceAgent, + Capabilities: ConnectionCapabilities{SupportsPause: false, SupportsScope: false, SupportsTest: false}, + } +} + +// deriveConnectionState maps (Enabled, InstanceHealth) onto the unified state +// vocabulary. No new state is persisted — the inputs come from the existing +// monitoring scheduler. +func deriveConnectionState(enabled bool, h monitoring.InstanceHealth, now time.Time) (ConnectionState, string, *time.Time, *ConnectionError) { + var lastSeen *time.Time + if h.PollStatus.LastSuccess != nil && !h.PollStatus.LastSuccess.IsZero() { + t := *h.PollStatus.LastSuccess + lastSeen = &t + } + + var lastError *ConnectionError + if h.PollStatus.LastError != nil && h.PollStatus.LastError.Message != "" { + lastError = &ConnectionError{ + At: h.PollStatus.LastError.At, + Message: h.PollStatus.LastError.Message, + Category: h.PollStatus.LastError.Category, + } + } + + if !enabled { + return ConnectionStatePaused, "paused by user", lastSeen, lastError + } + + if lastSeen == nil && lastError == nil { + return ConnectionStatePending, "awaiting first poll", nil, nil + } + + if lastError != nil && connectionAuthErrorPattern.MatchString(lastError.Message) { + return ConnectionStateUnauthorized, lastError.Message, lastSeen, lastError + } + + if strings.EqualFold(h.Breaker.State, "open") { + reason := "circuit breaker open" + if lastError != nil { + reason = lastError.Message + } + return ConnectionStateUnreachable, reason, lastSeen, lastError + } + + if lastSeen != nil && now.Sub(*lastSeen) > connectionStaleThreshold { + return ConnectionStateStale, fmt.Sprintf("no successful poll in %s", now.Sub(*lastSeen).Round(time.Second)), lastSeen, lastError + } + + return ConnectionStateActive, "", lastSeen, lastError +} + +func sourceFromString(s string) ConnectionSource { + switch strings.ToLower(strings.TrimSpace(s)) { + case "agent": + return ConnectionSourceAgent + case "script": + return ConnectionSourceScript + default: + return ConnectionSourceManual + } +} + +// instanceHealthByKey flattens SchedulerHealthResponse.Instances into a +// lookup map keyed by schedulerKey ("pve::instance-name"). The aggregator +// consults this to pick up LastSuccess and LastError without re-polling. +func instanceHealthByKey(resp monitoring.SchedulerHealthResponse) map[string]monitoring.InstanceHealth { + out := make(map[string]monitoring.InstanceHealth, len(resp.Instances)) + for _, inst := range resp.Instances { + if inst.Key == "" { + continue + } + out[inst.Key] = inst + } + return out +} diff --git a/internal/api/connections_aggregator_test.go b/internal/api/connections_aggregator_test.go new file mode 100644 index 000000000..32f2c2d41 --- /dev/null +++ b/internal/api/connections_aggregator_test.go @@ -0,0 +1,211 @@ +package api + +import ( + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/internal/monitoring" +) + +func ptrTime(t time.Time) *time.Time { return &t } + +func healthEntry(lastSuccess *time.Time, errMessage, errCategory string, breakerState string) monitoring.InstanceHealth { + ps := monitoring.InstancePollStatus{LastSuccess: lastSuccess} + if errMessage != "" { + ps.LastError = &monitoring.ErrorDetail{ + At: time.Now(), + Message: errMessage, + Category: errCategory, + } + } + return monitoring.InstanceHealth{ + PollStatus: ps, + Breaker: monitoring.InstanceBreaker{State: breakerState}, + } +} + +func TestDeriveConnectionState_Paused(t *testing.T) { + state, reason, _, _ := deriveConnectionState(false, monitoring.InstanceHealth{}, time.Now()) + if state != ConnectionStatePaused { + t.Fatalf("got %q, want %q", state, ConnectionStatePaused) + } + if reason == "" { + t.Fatal("expected non-empty reason for paused state") + } +} + +func TestDeriveConnectionState_Pending(t *testing.T) { + state, _, lastSeen, lastError := deriveConnectionState(true, monitoring.InstanceHealth{}, time.Now()) + if state != ConnectionStatePending { + t.Fatalf("got %q, want %q", state, ConnectionStatePending) + } + if lastSeen != nil || lastError != nil { + t.Fatal("expected nil lastSeen and lastError in pending state") + } +} + +func TestDeriveConnectionState_Unauthorized(t *testing.T) { + now := time.Now() + h := healthEntry(ptrTime(now.Add(-30*time.Second)), "401 Unauthorized: token invalid", "auth", "closed") + state, reason, _, err := deriveConnectionState(true, h, now) + if state != ConnectionStateUnauthorized { + t.Fatalf("got %q, want %q", state, ConnectionStateUnauthorized) + } + if reason == "" { + t.Fatal("expected reason to include error message") + } + if err == nil || err.Message != "401 Unauthorized: token invalid" { + t.Fatalf("unexpected lastError: %+v", err) + } +} + +func TestDeriveConnectionState_Unreachable(t *testing.T) { + now := time.Now() + h := healthEntry(ptrTime(now.Add(-30*time.Second)), "connection refused", "network", "open") + state, _, _, _ := deriveConnectionState(true, h, now) + if state != ConnectionStateUnreachable { + t.Fatalf("got %q, want %q", state, ConnectionStateUnreachable) + } +} + +func TestDeriveConnectionState_Stale(t *testing.T) { + now := time.Now() + stale := now.Add(-5 * time.Minute) + h := healthEntry(ptrTime(stale), "", "", "closed") + state, reason, _, _ := deriveConnectionState(true, h, now) + if state != ConnectionStateStale { + t.Fatalf("got %q, want %q", state, ConnectionStateStale) + } + if reason == "" { + t.Fatal("expected reason to describe staleness") + } +} + +func TestDeriveConnectionState_Active(t *testing.T) { + now := time.Now() + h := healthEntry(ptrTime(now.Add(-10*time.Second)), "", "", "closed") + state, _, _, _ := deriveConnectionState(true, h, now) + if state != ConnectionStateActive { + t.Fatalf("got %q, want %q", state, ConnectionStateActive) + } +} + +func TestBuildConnections_SortsByTypeThenName(t *testing.T) { + now := time.Now() + in := aggregatorInputs{ + pveInstances: []config.PVEInstance{ + {Name: "beta", Host: "https://b.lan:8006", MonitorVMs: true}, + {Name: "alpha", Host: "https://a.lan:8006", MonitorVMs: true}, + }, + pbsInstances: []config.PBSInstance{ + {Name: "backups", Host: "https://bkp.lan:8007"}, + }, + now: now, + } + got := buildConnections(in) + if len(got) != 3 { + t.Fatalf("expected 3 connections, got %d", len(got)) + } + if got[0].Type != ConnectionTypePBS { + t.Fatalf("expected PBS first, got %q", got[0].Type) + } + if got[1].Name != "alpha" || got[2].Name != "beta" { + t.Fatalf("PVE order wrong: %q, %q", got[1].Name, got[2].Name) + } +} + +func TestBuildConnections_PVEPausedRespectsDisabled(t *testing.T) { + in := aggregatorInputs{ + pveInstances: []config.PVEInstance{{Name: "pve1", Host: "https://pve1.lan:8006", Disabled: true}}, + now: time.Now(), + } + got := buildConnections(in) + if len(got) != 1 { + t.Fatalf("expected 1 connection, got %d", len(got)) + } + if got[0].State != ConnectionStatePaused { + t.Fatalf("expected paused, got %q", got[0].State) + } + if got[0].Enabled { + t.Fatal("expected Enabled=false for Disabled PVE instance") + } +} + +func TestBuildConnections_AgentStateFromLastSeen(t *testing.T) { + now := time.Now() + in := aggregatorInputs{ + hosts: []models.Host{ + {ID: "fresh", Hostname: "h1", LastSeen: now.Add(-10 * time.Second)}, + {ID: "stale", Hostname: "h2", LastSeen: now.Add(-5 * time.Minute)}, + {ID: "never", Hostname: "h3"}, + }, + now: now, + } + got := buildConnections(in) + byID := map[string]Connection{} + for _, c := range got { + byID[c.ID] = c + } + if byID["agent:fresh"].State != ConnectionStateActive { + t.Fatalf("fresh agent: got %q, want active", byID["agent:fresh"].State) + } + if byID["agent:stale"].State != ConnectionStateStale { + t.Fatalf("stale agent: got %q, want stale", byID["agent:stale"].State) + } + if byID["agent:never"].State != ConnectionStatePending { + t.Fatalf("never-reported agent: got %q, want pending", byID["agent:never"].State) + } + for _, c := range got { + if c.Capabilities.SupportsPause || c.Capabilities.SupportsScope { + t.Fatalf("agents must not advertise pause/scope capabilities: %+v", c) + } + } +} + +func TestBuildConnections_VMwareAndTrueNASEnabledFlag(t *testing.T) { + in := aggregatorInputs{ + vmwareInstances: []config.VMwareVCenterInstance{{ID: "vc1", Name: "vc", Host: "vc.lan", Enabled: false}}, + truenasInstances: []config.TrueNASInstance{{ID: "tn1", Name: "tn", Host: "tn.lan", Enabled: true, UseHTTPS: true}}, + now: time.Now(), + } + got := buildConnections(in) + var vmw, tn Connection + for _, c := range got { + switch c.Type { + case ConnectionTypeVMware: + vmw = c + case ConnectionTypeTrueNAS: + tn = c + } + } + if vmw.State != ConnectionStatePaused || vmw.Enabled { + t.Fatalf("vmware with Enabled=false should be paused, got state=%q enabled=%v", vmw.State, vmw.Enabled) + } + if tn.State != ConnectionStatePending { + t.Fatalf("truenas with no health yet should be pending, got %q", tn.State) + } + if !tn.Enabled { + t.Fatal("truenas with Enabled=true should surface enabled=true") + } +} + +func TestBuildConnections_UsesHealthLookup(t *testing.T) { + now := time.Now() + ls := now.Add(-15 * time.Second) + in := aggregatorInputs{ + pveInstances: []config.PVEInstance{{Name: "pve1", Host: "https://pve1.lan:8006"}}, + instanceHealth: map[string]monitoring.InstanceHealth{ + "pve::pve1": healthEntry(&ls, "", "", "closed"), + }, + now: now, + } + got := buildConnections(in) + if got[0].State != ConnectionStateActive { + t.Fatalf("expected active, got %q", got[0].State) + } + if got[0].LastSeen == nil || !got[0].LastSeen.Equal(ls) { + t.Fatalf("lastSeen not propagated: %+v", got[0].LastSeen) + } +} diff --git a/internal/api/connections_handlers.go b/internal/api/connections_handlers.go new file mode 100644 index 000000000..8a3bfefea --- /dev/null +++ b/internal/api/connections_handlers.go @@ -0,0 +1,111 @@ +package api + +import ( + "context" + "encoding/json" + "net/http" + + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/internal/monitoring" +) + +// ConnectionsHandlers serves the unified connections ledger. It does not own +// any persistence of its own — it composes per-type stores and the +// monitoring scheduler's in-memory health data into a single list. +type ConnectionsHandlers struct { + getConfig func(ctx context.Context) *config.Config + getPersistence func(ctx context.Context) *config.ConfigPersistence + getMonitor func(ctx context.Context) *monitoring.Monitor +} + +// NewConnectionsHandlers wires the aggregator behind the request-scoped +// tenant resolvers already used by ConfigHandlers, so the endpoint shares +// the same multi-tenant behavior as the per-type routes it aggregates. +func NewConnectionsHandlers( + getConfig func(ctx context.Context) *config.Config, + getPersistence func(ctx context.Context) *config.ConfigPersistence, + getMonitor func(ctx context.Context) *monitoring.Monitor, +) *ConnectionsHandlers { + return &ConnectionsHandlers{ + getConfig: getConfig, + getPersistence: getPersistence, + getMonitor: getMonitor, + } +} + +// HandleList returns every configured connection as a unified Connection row. +// No probing or network I/O happens here — state is derived purely from +// cached poller health. +func (h *ConnectionsHandlers) HandleList(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + + ctx := r.Context() + cfg := h.getConfig(ctx) + persistence := h.getPersistence(ctx) + monitor := h.getMonitor(ctx) + + inputs := aggregatorInputs{} + + if cfg != nil { + inputs.pveInstances = cfg.PVEInstances + inputs.pbsInstances = cfg.PBSInstances + inputs.pmgInstances = cfg.PMGInstances + } + + if persistence != nil { + if vmw, err := persistence.LoadVMwareConfig(); err == nil { + inputs.vmwareInstances = vmw + } + if tn, err := persistence.LoadTrueNASConfig(); err == nil { + inputs.truenasInstances = tn + } + } + + if monitor != nil { + snapshot := monitor.GetState() + inputs.hosts = snapshot.Hosts + inputs.instanceHealth = instanceHealthByKey(monitor.SchedulerHealth()) + } else { + inputs.hosts = []models.Host{} + inputs.instanceHealth = map[string]monitoring.InstanceHealth{} + } + + writeJSON(w, http.StatusOK, ConnectionsListResponse{ + Connections: buildConnections(inputs), + }) +} + +// HandleProbe fingerprints a user-supplied address and returns the product +// types it detected, if any. No configuration is persisted — the caller +// uses the response to pick a credential slot in ConnectionEditor. +func (h *ConnectionsHandlers) HandleProbe(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + + r.Body = http.MaxBytesReader(w, r.Body, 4*1024) + defer r.Body.Close() + + var req ProbeRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeErrorResponse(w, http.StatusBadRequest, "invalid_request", "Invalid JSON body", nil) + return + } + + host, port, err := parseProbeAddress(req.Address) + if err != nil { + writeErrorResponse(w, http.StatusBadRequest, "invalid_address", err.Error(), nil) + return + } + + candidates, elapsed := runProbe(r.Context(), host, port, probeHTTPClient()) + writeJSON(w, http.StatusOK, ProbeResponse{ + Candidates: candidates, + ProbedMs: elapsed.Milliseconds(), + }) +} diff --git a/internal/api/connections_probe.go b/internal/api/connections_probe.go new file mode 100644 index 000000000..c0d2d0ad4 --- /dev/null +++ b/internal/api/connections_probe.go @@ -0,0 +1,328 @@ +package api + +import ( + "context" + "crypto/sha256" + "crypto/tls" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "net/url" + "sort" + "strconv" + "strings" + "sync" + "time" +) + +// Probe budget. Per-candidate and whole-request ceilings are kept tight to +// make the endpoint unusable as a slow-leak scanner. Each candidate gets +// 2s connect + 1s read, and the whole request is capped at the total budget. +const ( + probeDialTimeout = 2 * time.Second + probeTotalBudget = 3 * time.Second + probeMaxConcurrent = 5 + probeMaxAddressBytes = 512 +) + +// ProbeRequest is the POST body for /api/connections/probe. +type ProbeRequest struct { + Address string `json:"address"` +} + +// ProbeCandidate is one detected product at a host:port. +type ProbeCandidate struct { + Type ConnectionType `json:"type"` + Host string `json:"host"` + Port int `json:"port"` + Hints map[string]string `json:"hints,omitempty"` +} + +// ProbeResponse is the envelope returned to the frontend. +type ProbeResponse struct { + Candidates []ProbeCandidate `json:"candidates"` + ProbedMs int64 `json:"probedMs"` +} + +// probeTarget is a single {type, scheme, port, path} fingerprint we attempt. +type probeTarget struct { + Type ConnectionType + Scheme string + Port int + Path string + identifyFn func(resp *http.Response, body []byte) (match bool, hints map[string]string) +} + +var defaultProbeTargets = []probeTarget{ + { + Type: ConnectionTypePVE, + Scheme: "https", + Port: 8006, + Path: "/api2/json/version", + identifyFn: func(resp *http.Response, body []byte) (bool, map[string]string) { + server := strings.ToLower(resp.Header.Get("Server")) + if strings.Contains(server, "pve-api-daemon") { + return true, versionHintsFromProxmoxBody(body, "Proxmox VE") + } + if strings.Contains(string(body), `"repoid"`) && strings.Contains(string(body), `"version"`) && + !strings.Contains(string(body), `"product":"pmg"`) { + return true, versionHintsFromProxmoxBody(body, "Proxmox VE") + } + return false, nil + }, + }, + { + Type: ConnectionTypePBS, + Scheme: "https", + Port: 8007, + Path: "/api2/json/version", + identifyFn: func(resp *http.Response, body []byte) (bool, map[string]string) { + server := strings.ToLower(resp.Header.Get("Server")) + if strings.Contains(server, "proxmox-backup-api") { + return true, versionHintsFromProxmoxBody(body, "Proxmox Backup Server") + } + return false, nil + }, + }, + { + Type: ConnectionTypePMG, + Scheme: "https", + Port: 8006, + Path: "/api2/json/version", + identifyFn: func(resp *http.Response, body []byte) (bool, map[string]string) { + server := strings.ToLower(resp.Header.Get("Server")) + if strings.Contains(server, "pmg-api-daemon") { + return true, versionHintsFromProxmoxBody(body, "Proxmox Mail Gateway") + } + if strings.Contains(string(body), `"product":"pmg"`) { + return true, versionHintsFromProxmoxBody(body, "Proxmox Mail Gateway") + } + return false, nil + }, + }, + { + Type: ConnectionTypeVMware, + Scheme: "https", + Port: 443, + Path: "/sdk/vimServiceVersions.xml", + identifyFn: func(_ *http.Response, body []byte) (bool, map[string]string) { + if strings.Contains(string(body), "urn:vim25") { + return true, map[string]string{"product": "VMware vCenter"} + } + return false, nil + }, + }, + { + Type: ConnectionTypeTrueNAS, + Scheme: "https", + Port: 443, + Path: "/api/v2.0/system/product_name", + identifyFn: func(_ *http.Response, body []byte) (bool, map[string]string) { + upper := strings.ToUpper(string(body)) + if strings.Contains(upper, "TRUENAS") { + return true, map[string]string{"product": "TrueNAS"} + } + return false, nil + }, + }, +} + +// versionHintsFromProxmoxBody pulls the minimum fields out of a Proxmox API +// /version response so the frontend can show the user something useful +// beyond "we think this is PVE." Any shape mismatch just yields the product +// name alone — we never fail a probe on a hint-parse error. +func versionHintsFromProxmoxBody(body []byte, productName string) map[string]string { + hints := map[string]string{"product": productName} + var wrapper struct { + Data struct { + Version string `json:"version"` + Release string `json:"release"` + Repoid string `json:"repoid"` + } `json:"data"` + } + if err := json.Unmarshal(body, &wrapper); err == nil { + if wrapper.Data.Version != "" { + hints["version"] = wrapper.Data.Version + } + if wrapper.Data.Release != "" { + hints["release"] = wrapper.Data.Release + } + } + return hints +} + +// parseProbeAddress normalizes user input into (host, explicitPort). +// Accepted forms: "host", "host:port", "ip", "ip:port", "scheme://host[:port]". +// explicitPort == 0 means "probe all defaults"; otherwise only probe targets +// that match the given port. +func parseProbeAddress(raw string) (host string, explicitPort int, err error) { + raw = strings.TrimSpace(raw) + if raw == "" { + return "", 0, fmt.Errorf("address is required") + } + if len(raw) > probeMaxAddressBytes { + return "", 0, fmt.Errorf("address is too long") + } + + if strings.Contains(raw, "://") { + u, err := url.Parse(raw) + if err != nil || u.Host == "" { + return "", 0, fmt.Errorf("invalid URL: %s", raw) + } + host = u.Hostname() + if portStr := u.Port(); portStr != "" { + p, err := strconv.Atoi(portStr) + if err != nil || p < 1 || p > 65535 { + return "", 0, fmt.Errorf("invalid port in URL") + } + explicitPort = p + } + return host, explicitPort, nil + } + + if h, p, splitErr := net.SplitHostPort(raw); splitErr == nil { + portNum, convErr := strconv.Atoi(p) + if convErr != nil || portNum < 1 || portNum > 65535 { + return "", 0, fmt.Errorf("invalid port") + } + return h, portNum, nil + } + + return raw, 0, nil +} + +// targetsForPort narrows defaultProbeTargets to only those matching the +// user's explicit port. Same host can serve PVE on 8006 and PBS on 8007, so +// zero port means "try them all." +func targetsForPort(port int) []probeTarget { + if port == 0 { + return defaultProbeTargets + } + out := make([]probeTarget, 0, 2) + for _, t := range defaultProbeTargets { + if t.Port == port { + out = append(out, t) + } + } + return out +} + +// probeHTTPClient builds one client per probe so that dial timeouts, +// TLS-skip, and cancellation are self-contained. InsecureSkipVerify is on +// because the whole point of probing is to talk to a server whose cert we +// haven't trusted yet. +func probeHTTPClient() *http.Client { + return &http.Client{ + Transport: &http.Transport{ + DialContext: (&net.Dialer{ + Timeout: probeDialTimeout, + }).DialContext, + TLSHandshakeTimeout: probeDialTimeout, + TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, + DisableKeepAlives: true, + }, + // Overall per-request ceiling is the dial + read budget. + Timeout: probeDialTimeout + time.Second, + } +} + +// runProbe fans out probe requests against every candidate target. It +// returns (sorted-deduped candidates, total elapsed). The function never +// returns an error — individual probe failures are swallowed as "not that +// type" rather than bubbling up and confusing the caller. +func runProbe(ctx context.Context, host string, port int, client *http.Client) ([]ProbeCandidate, time.Duration) { + start := time.Now() + budget := probeTotalBudget + ctx, cancel := context.WithTimeout(ctx, budget) + defer cancel() + + targets := targetsForPort(port) + if len(targets) == 0 { + return []ProbeCandidate{}, time.Since(start) + } + + sem := make(chan struct{}, probeMaxConcurrent) + var wg sync.WaitGroup + var mu sync.Mutex + results := make([]ProbeCandidate, 0, len(targets)) + + for _, target := range targets { + target := target + wg.Add(1) + go func() { + defer wg.Done() + select { + case sem <- struct{}{}: + defer func() { <-sem }() + case <-ctx.Done(): + return + } + + if cand, ok := probeOne(ctx, host, target, client); ok { + mu.Lock() + results = append(results, cand) + mu.Unlock() + } + }() + } + wg.Wait() + + sort.Slice(results, func(i, j int) bool { + if results[i].Type != results[j].Type { + return results[i].Type < results[j].Type + } + return results[i].Port < results[j].Port + }) + + return results, time.Since(start) +} + +// probeOne runs a single probe. It returns (candidate, true) only when the +// target's identifyFn confirms the product. +func probeOne(ctx context.Context, host string, target probeTarget, client *http.Client) (ProbeCandidate, bool) { + endpoint := fmt.Sprintf("%s://%s:%d%s", target.Scheme, host, target.Port, target.Path) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) + if err != nil { + return ProbeCandidate{}, false + } + req.Header.Set("User-Agent", "Pulse/connections-probe") + req.Header.Set("Accept", "application/json,text/xml,*/*") + + resp, err := client.Do(req) + if err != nil { + return ProbeCandidate{}, false + } + defer resp.Body.Close() + + if resp.StatusCode >= 500 { + return ProbeCandidate{}, false + } + + body, err := io.ReadAll(io.LimitReader(resp.Body, 64*1024)) + if err != nil { + return ProbeCandidate{}, false + } + + match, hints := target.identifyFn(resp, body) + if !match { + return ProbeCandidate{}, false + } + + if hints == nil { + hints = map[string]string{} + } + if resp.TLS != nil && len(resp.TLS.PeerCertificates) > 0 { + sum := sha256.Sum256(resp.TLS.PeerCertificates[0].Raw) + hints["fingerprint"] = "SHA256:" + hex.EncodeToString(sum[:]) + } + + return ProbeCandidate{ + Type: target.Type, + Host: fmt.Sprintf("%s://%s:%d", target.Scheme, host, target.Port), + Port: target.Port, + Hints: hints, + }, true +} diff --git a/internal/api/connections_probe_test.go b/internal/api/connections_probe_test.go new file mode 100644 index 000000000..a9b7460f9 --- /dev/null +++ b/internal/api/connections_probe_test.go @@ -0,0 +1,333 @@ +package api + +import ( + "context" + "crypto/tls" + "fmt" + "net" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "testing" + "time" +) + +func TestParseProbeAddress_Shapes(t *testing.T) { + cases := []struct { + in string + wantHost string + wantPort int + expectError bool + }{ + {"pve01.lan", "pve01.lan", 0, false}, + {"pve01.lan:8006", "pve01.lan", 8006, false}, + {"192.168.1.10", "192.168.1.10", 0, false}, + {"192.168.1.10:443", "192.168.1.10", 443, false}, + {"https://pve01.lan:8006", "pve01.lan", 8006, false}, + {"https://pve01.lan", "pve01.lan", 0, false}, + {" ", "", 0, true}, + {"pve01.lan:99999", "", 0, true}, + } + for _, c := range cases { + t.Run(c.in, func(t *testing.T) { + host, port, err := parseProbeAddress(c.in) + if c.expectError { + if err == nil { + t.Fatalf("expected error for %q", c.in) + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if host != c.wantHost { + t.Fatalf("host = %q, want %q", host, c.wantHost) + } + if port != c.wantPort { + t.Fatalf("port = %d, want %d", port, c.wantPort) + } + }) + } +} + +func TestTargetsForPort_ExplicitNarrowsToMatching(t *testing.T) { + targets := targetsForPort(8006) + seen := map[ConnectionType]bool{} + for _, t := range targets { + seen[t.Type] = true + } + if !seen[ConnectionTypePVE] { + t.Fatal("expected PVE target for port 8006") + } + if !seen[ConnectionTypePMG] { + t.Fatal("expected PMG target for port 8006") + } + if seen[ConnectionTypePBS] { + t.Fatal("did not expect PBS target for port 8006") + } +} + +func TestTargetsForPort_ZeroReturnsAll(t *testing.T) { + if len(targetsForPort(0)) != len(defaultProbeTargets) { + t.Fatal("zero port should return all default targets") + } +} + +// probeTestServer spins up an HTTPS server on a random port, overriding the +// target port in defaultProbeTargets so probeOne can hit it. Restores the +// original targets in cleanup. Returns the host:port string. +func probeTestServer(t *testing.T, handler http.HandlerFunc) (host string, port int, cleanup func()) { + t.Helper() + srv := httptest.NewTLSServer(handler) + + u := srv.URL + u = strings.TrimPrefix(u, "https://") + h, p, err := net.SplitHostPort(u) + if err != nil { + t.Fatalf("failed to split host:port %q: %v", u, err) + } + portInt, err := strconv.Atoi(p) + if err != nil { + t.Fatalf("failed to parse port %q: %v", p, err) + } + return h, portInt, func() { + srv.Close() + } +} + +func pveHandler(body string) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Server", "pve-api-daemon/3.4") + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, body) + } +} + +func pbsHandler(body string) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Server", "proxmox-backup-api/3.2") + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, body) + } +} + +func pmgHandler(body string) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Server", "pmg-api-daemon/8.1") + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, body) + } +} + +func vmwareHandler() http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/xml") + fmt.Fprint(w, ` + + urn:vim25 +`) + } +} + +func truenasHandler() http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `"TRUENAS-SCALE-24.04"`) + } +} + +// testProbeClient mirrors probeHTTPClient but shorter timeouts for tests. +func testProbeClient() *http.Client { + return &http.Client{ + Transport: &http.Transport{ + DialContext: (&net.Dialer{Timeout: 500 * time.Millisecond}).DialContext, + TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, + DisableKeepAlives: true, + }, + Timeout: 1 * time.Second, + } +} + +func withSingleTarget(t *testing.T, tgt probeTarget) func() { + t.Helper() + orig := defaultProbeTargets + defaultProbeTargets = []probeTarget{tgt} + return func() { defaultProbeTargets = orig } +} + +func TestProbeOne_DetectsPVE(t *testing.T) { + host, port, done := probeTestServer(t, pveHandler(`{"data":{"version":"8.2.4","release":"8.2","repoid":"abc"}}`)) + defer done() + + tgt := probeTarget{ + Type: ConnectionTypePVE, + Scheme: "https", + Port: port, + Path: "/api2/json/version", + identifyFn: defaultProbeTargets[0].identifyFn, + } + cand, ok := probeOne(context.Background(), host, tgt, testProbeClient()) + if !ok { + t.Fatal("expected PVE detection") + } + if cand.Type != ConnectionTypePVE { + t.Fatalf("type = %q, want pve", cand.Type) + } + if cand.Hints["version"] != "8.2.4" { + t.Fatalf("version hint not propagated: %+v", cand.Hints) + } +} + +func TestProbeOne_DetectsPBS(t *testing.T) { + host, port, done := probeTestServer(t, pbsHandler(`{"data":{"version":"3.2.1","release":"3.2"}}`)) + defer done() + + // Find the PBS identifyFn from defaults. + var pbsFn func(*http.Response, []byte) (bool, map[string]string) + for _, tgt := range defaultProbeTargets { + if tgt.Type == ConnectionTypePBS { + pbsFn = tgt.identifyFn + } + } + if pbsFn == nil { + t.Fatal("pbs target not in defaults") + } + tgt := probeTarget{Type: ConnectionTypePBS, Scheme: "https", Port: port, Path: "/api2/json/version", identifyFn: pbsFn} + if _, ok := probeOne(context.Background(), host, tgt, testProbeClient()); !ok { + t.Fatal("expected PBS detection") + } +} + +func TestProbeOne_DetectsPMGByProductField(t *testing.T) { + // Simulate a server that serves PMG content but with no distinctive Server header. + host, port, done := probeTestServer(t, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{"data":{"product":"pmg","version":"8.1.2"}}`) + }) + defer done() + + var pmgFn func(*http.Response, []byte) (bool, map[string]string) + for _, tgt := range defaultProbeTargets { + if tgt.Type == ConnectionTypePMG { + pmgFn = tgt.identifyFn + } + } + tgt := probeTarget{Type: ConnectionTypePMG, Scheme: "https", Port: port, Path: "/api2/json/version", identifyFn: pmgFn} + if _, ok := probeOne(context.Background(), host, tgt, testProbeClient()); !ok { + t.Fatal("expected PMG detection via product field") + } +} + +func TestProbeOne_DetectsVMware(t *testing.T) { + host, port, done := probeTestServer(t, vmwareHandler()) + defer done() + + var vmFn func(*http.Response, []byte) (bool, map[string]string) + for _, tgt := range defaultProbeTargets { + if tgt.Type == ConnectionTypeVMware { + vmFn = tgt.identifyFn + } + } + tgt := probeTarget{Type: ConnectionTypeVMware, Scheme: "https", Port: port, Path: "/sdk/vimServiceVersions.xml", identifyFn: vmFn} + if _, ok := probeOne(context.Background(), host, tgt, testProbeClient()); !ok { + t.Fatal("expected VMware detection") + } +} + +func TestProbeOne_DetectsTrueNAS(t *testing.T) { + host, port, done := probeTestServer(t, truenasHandler()) + defer done() + + var tnFn func(*http.Response, []byte) (bool, map[string]string) + for _, tgt := range defaultProbeTargets { + if tgt.Type == ConnectionTypeTrueNAS { + tnFn = tgt.identifyFn + } + } + tgt := probeTarget{Type: ConnectionTypeTrueNAS, Scheme: "https", Port: port, Path: "/api/v2.0/system/product_name", identifyFn: tnFn} + if _, ok := probeOne(context.Background(), host, tgt, testProbeClient()); !ok { + t.Fatal("expected TrueNAS detection") + } +} + +func TestProbeOne_RejectsNonMatchingServer(t *testing.T) { + host, port, done := probeTestServer(t, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Server", "nginx/1.24") + fmt.Fprint(w, `hello`) + }) + defer done() + + for _, tgt := range defaultProbeTargets { + t := probeTarget{Type: tgt.Type, Scheme: "https", Port: port, Path: tgt.Path, identifyFn: tgt.identifyFn} + if _, ok := probeOne(context.Background(), host, t, testProbeClient()); ok { + return + } + } + // None matched — expected outcome. +} + +func TestProbeOne_DoesNotConfusePVEForPMG(t *testing.T) { + host, port, done := probeTestServer(t, pveHandler(`{"data":{"version":"8.2.4","release":"8.2","repoid":"abc"}}`)) + defer done() + + var pmgFn func(*http.Response, []byte) (bool, map[string]string) + for _, tgt := range defaultProbeTargets { + if tgt.Type == ConnectionTypePMG { + pmgFn = tgt.identifyFn + } + } + tgt := probeTarget{Type: ConnectionTypePMG, Scheme: "https", Port: port, Path: "/api2/json/version", identifyFn: pmgFn} + if _, ok := probeOne(context.Background(), host, tgt, testProbeClient()); ok { + t.Fatal("PMG identifyFn should not match a PVE server") + } +} + +func TestRunProbe_ReturnsSortedCandidates(t *testing.T) { + // Spin up one PVE server; runProbe should invoke all candidates but only + // detect PVE. Shorten the budget for the test. + host, port, done := probeTestServer(t, pveHandler(`{"data":{"version":"8.2.4","release":"8.2","repoid":"abc"}}`)) + defer done() + + // Override defaultProbeTargets to only the PVE target at our ephemeral port. + var pveFn func(*http.Response, []byte) (bool, map[string]string) + for _, tgt := range defaultProbeTargets { + if tgt.Type == ConnectionTypePVE { + pveFn = tgt.identifyFn + } + } + restore := withSingleTarget(t, probeTarget{Type: ConnectionTypePVE, Scheme: "https", Port: port, Path: "/api2/json/version", identifyFn: pveFn}) + defer restore() + + results, elapsed := runProbe(context.Background(), host, port, testProbeClient()) + if len(results) != 1 || results[0].Type != ConnectionTypePVE { + t.Fatalf("unexpected results: %+v", results) + } + if elapsed <= 0 { + t.Fatal("elapsed should be positive") + } +} + +func TestRunProbe_TotalBudgetEnforced(t *testing.T) { + // Build a dummy target that points nowhere — dial will be slow/timeout. + restore := withSingleTarget(t, probeTarget{ + Type: ConnectionTypePVE, + Scheme: "https", + Port: 1, // unlikely open + Path: "/api2/json/version", + identifyFn: func(*http.Response, []byte) (bool, map[string]string) { + return true, nil + }, + }) + defer restore() + + start := time.Now() + results, _ := runProbe(context.Background(), "127.0.0.1", 0, testProbeClient()) + elapsed := time.Since(start) + if len(results) != 0 { + t.Fatalf("expected no results from closed port, got %+v", results) + } + if elapsed > 2*time.Second { + t.Fatalf("probe exceeded expected timeout: %s", elapsed) + } +} diff --git a/internal/api/connections_types.go b/internal/api/connections_types.go new file mode 100644 index 000000000..0a79d0b15 --- /dev/null +++ b/internal/api/connections_types.go @@ -0,0 +1,84 @@ +package api + +import "time" + +// ConnectionState is the derived lifecycle state shown in the unified +// connections ledger. Derivation rules live in connections_aggregator.go; no +// new state is persisted — every value comes from existing runtime signals. +type ConnectionState string + +const ( + ConnectionStateActive ConnectionState = "active" + ConnectionStatePaused ConnectionState = "paused" + ConnectionStateUnauthorized ConnectionState = "unauthorized" + ConnectionStateUnreachable ConnectionState = "unreachable" + ConnectionStateStale ConnectionState = "stale" + ConnectionStatePending ConnectionState = "pending" +) + +// ConnectionType is the product family that owns the connection. The value is +// the discriminator the frontend switches on to render per-type credential +// slots in ConnectionEditor. +type ConnectionType string + +const ( + ConnectionTypePVE ConnectionType = "pve" + ConnectionTypePBS ConnectionType = "pbs" + ConnectionTypePMG ConnectionType = "pmg" + ConnectionTypeVMware ConnectionType = "vmware" + ConnectionTypeTrueNAS ConnectionType = "truenas" + ConnectionTypeAgent ConnectionType = "agent" + ConnectionTypeDocker ConnectionType = "docker" + ConnectionTypeKubernetes ConnectionType = "kubernetes" +) + +// ConnectionSource records how a connection entered Pulse. +type ConnectionSource string + +const ( + ConnectionSourceManual ConnectionSource = "manual" + ConnectionSourceAgent ConnectionSource = "agent" + ConnectionSourceScript ConnectionSource = "script" +) + +// ConnectionCapabilities tells the frontend which controls to render for a +// connection. Agents cannot pause or partial-scope; manual Proxmox/PBS/PMG +// can do both. This avoids putting the decision inside the editor. +type ConnectionCapabilities struct { + SupportsPause bool `json:"supportsPause"` + SupportsScope bool `json:"supportsScope"` + SupportsTest bool `json:"supportsTest"` +} + +// ConnectionError is the runtime error shape surfaced on a connection row. +// Mirrors monitoring.ErrorDetail but lives in the api package so the type +// stays stable if the internal monitoring shape evolves. +type ConnectionError struct { + At time.Time `json:"at"` + Message string `json:"message"` + Category string `json:"category,omitempty"` +} + +// Connection is the unified row the frontend consumes. It replaces the +// per-type shapes that today require separate fetches and separate table +// renderers. +type Connection struct { + ID string `json:"id"` + Type ConnectionType `json:"type"` + Name string `json:"name"` + Address string `json:"address"` + State ConnectionState `json:"state"` + StateReason string `json:"stateReason,omitempty"` + Enabled bool `json:"enabled"` + Surfaces []string `json:"surfaces"` + Scope map[string]bool `json:"scope"` + LastSeen *time.Time `json:"lastSeen,omitempty"` + LastError *ConnectionError `json:"lastError,omitempty"` + Source ConnectionSource `json:"source"` + Capabilities ConnectionCapabilities `json:"capabilities"` +} + +// ConnectionsListResponse is the envelope for GET /api/connections. +type ConnectionsListResponse struct { + Connections []Connection `json:"connections"` +} diff --git a/internal/api/contract_test.go b/internal/api/contract_test.go index f2c2db087..06c9841e9 100644 --- a/internal/api/contract_test.go +++ b/internal/api/contract_test.go @@ -10433,6 +10433,59 @@ func TestContract_ShippedSecurityDocReferencesStayLocal(t *testing.T) { } } +// TestContract_ConnectionPayloadShapeStaysCanonical pins the JSON shape of +// the unified connections ledger so the Go types in +// `internal/api/connections_types.go` stay in lockstep with the +// `frontend-modern/src/api/connections.ts` client. Adding or renaming fields +// on this wire shape requires an explicit update to both ends in the same +// commit. +func TestContract_ConnectionPayloadShapeStaysCanonical(t *testing.T) { + conn := Connection{ + ID: "pve-lab", + Type: ConnectionTypePVE, + Name: "lab", + Address: "https://pve.lab:8006", + State: ConnectionStateActive, + StateReason: "", + Enabled: true, + Surfaces: []string{"vms", "containers"}, + Scope: map[string]bool{"vms": true, "containers": true}, + LastSeen: timePtr(time.Date(2026, 4, 19, 10, 0, 0, 0, time.UTC)), + LastError: nil, + Source: ConnectionSourceManual, + Capabilities: ConnectionCapabilities{SupportsPause: true, SupportsScope: true, SupportsTest: true}, + } + body, err := json.Marshal(ConnectionsListResponse{Connections: []Connection{conn}}) + if err != nil { + t.Fatalf("marshal Connection: %v", err) + } + want := `{"connections":[{"id":"pve-lab","type":"pve","name":"lab","address":"https://pve.lab:8006","state":"active","enabled":true,"surfaces":["vms","containers"],"scope":{"containers":true,"vms":true},"lastSeen":"2026-04-19T10:00:00Z","source":"manual","capabilities":{"supportsPause":true,"supportsScope":true,"supportsTest":true}}]}` + assertJSONSnapshot(t, body, want) +} + +// TestContract_ProbePayloadShapeStaysCanonical pins the POST +// /api/connections/probe wire shape. Hint keys are free-form; the envelope +// fields (type, host, port, hints) are the contract boundary. +func TestContract_ProbePayloadShapeStaysCanonical(t *testing.T) { + resp := ProbeResponse{ + Candidates: []ProbeCandidate{ + { + Type: ConnectionTypePVE, + Host: "https://pve.lab:8006", + Port: 8006, + Hints: map[string]string{"product": "Proxmox VE", "version": "8.2.4"}, + }, + }, + ProbedMs: 812, + } + body, err := json.Marshal(resp) + if err != nil { + t.Fatalf("marshal ProbeResponse: %v", err) + } + want := `{"candidates":[{"type":"pve","host":"https://pve.lab:8006","port":8006,"hints":{"product":"Proxmox VE","version":"8.2.4"}}],"probedMs":812}` + assertJSONSnapshot(t, body, want) +} + func mustStreamEvent(t *testing.T, eventType string, data interface{}) chat.StreamEvent { t.Helper() diff --git a/internal/api/route_inventory_test.go b/internal/api/route_inventory_test.go index 8c3aba634..c3df55e9e 100644 --- a/internal/api/route_inventory_test.go +++ b/internal/api/route_inventory_test.go @@ -300,6 +300,8 @@ var bareRouteAllowlist = []string{ "/api/config/nodes/test-config", "/api/config/nodes/test-connection", "/api/config/system", + "/api/connections", + "/api/connections/probe", "/api/truenas/connections", "/api/truenas/connections/preview", "/api/truenas/connections/test", @@ -433,6 +435,8 @@ var allRouteAllowlist = []string{ "/api/config/nodes/test-config", "/api/config/nodes/test-connection", "/api/config/nodes/", + "/api/connections", + "/api/connections/probe", "/api/truenas/connections", "/api/truenas/connections/preview", "/api/truenas/connections/test", diff --git a/internal/api/router.go b/internal/api/router.go index 2e0ecc141..736412538 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -72,6 +72,7 @@ type Router struct { configHandlers *ConfigHandlers trueNASHandlers *TrueNASHandlers vmwareHandlers *VMwareHandlers + connectionsHandlers *ConnectionsHandlers notificationHandlers *NotificationHandlers notificationQueueHandlers *NotificationQueueHandlers dockerAgentHandlers *DockerAgentHandlers @@ -386,6 +387,11 @@ func (r *Router) setupRoutes() { getMonitor: r.configHandlers.getMonitor, getPoller: func(context.Context) *monitoring.VMwarePoller { return r.vmwarePoller }, } + r.connectionsHandlers = NewConnectionsHandlers( + r.configHandlers.getConfig, + r.configHandlers.getPersistence, + r.configHandlers.getMonitor, + ) recoveryManager := recoverymanager.New(r.multiTenant) r.recoveryHandlers = NewRecoveryHandlers(recoveryManager) if r.mtMonitor != nil { diff --git a/internal/api/router_routes_registration.go b/internal/api/router_routes_registration.go index 6706c9db0..7f2a58f97 100644 --- a/internal/api/router_routes_registration.go +++ b/internal/api/router_routes_registration.go @@ -176,6 +176,32 @@ func (r *Router) registerConfigSystemRoutes(updateHandlers *UpdateHandlers) { } }) + // Unified connections ledger — aggregates PVE/PBS/PMG/VMware/TrueNAS/agent rows. + r.mux.HandleFunc("/api/connections", func(w http.ResponseWriter, req *http.Request) { + if r.connectionsHandlers == nil { + writeErrorResponse(w, http.StatusServiceUnavailable, "connections_unavailable", "Connections service unavailable", nil) + return + } + if req.Method != http.MethodGet { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + RequireAdmin(r.config, RequireScope(config.ScopeSettingsRead, r.connectionsHandlers.HandleList))(w, req) + }) + + // Connection address probe — stateless type detection before credential entry. + r.mux.HandleFunc("/api/connections/probe", func(w http.ResponseWriter, req *http.Request) { + if r.connectionsHandlers == nil { + writeErrorResponse(w, http.StatusServiceUnavailable, "connections_unavailable", "Connections service unavailable", nil) + return + } + if req.Method != http.MethodPost { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, r.connectionsHandlers.HandleProbe))(w, req) + }) + // TrueNAS connection management r.mux.HandleFunc("/api/truenas/connections", func(w http.ResponseWriter, req *http.Request) { if r.trueNASHandlers == nil { diff --git a/internal/config/config.go b/internal/config/config.go index 6734633bb..b435eed6e 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -543,6 +543,11 @@ type PVEInstance struct { // Agent tracking Source string // "agent" or "script" - how this node was registered (empty = legacy/manual) DisableCeph bool // Disable Ceph status polling for this instance + + // Lifecycle. Disabled (not Enabled) so the zero value means "poll" — + // existing nodes.json records from before this field existed load + // unpaused without a schema migration. + Disabled bool `json:"disabled,omitempty"` } // ClusterEndpoint represents a single node in a cluster @@ -595,6 +600,9 @@ type PBSInstance struct { // Datastore exclusion (for unmounted/removable datastores that cause log noise) ExcludeDatastores []string + + // Lifecycle. See PVEInstance.Disabled. + Disabled bool `json:"disabled,omitempty"` } // PMGInstance represents a Proxmox Mail Gateway connection @@ -615,6 +623,9 @@ type PMGInstance struct { MonitorDomainStats bool TemperatureMonitoringEnabled *bool // Monitor temperature via SSH (nil = use global setting, true/false = override) SSHPort int // SSH port for temperature monitoring (0 = use global default) + + // Lifecycle. See PVEInstance.Disabled. + Disabled bool `json:"disabled,omitempty"` } // Global persistence instance for saving diff --git a/internal/monitoring/canonical_guardrails_test.go b/internal/monitoring/canonical_guardrails_test.go index 58adccde7..cc4343bc1 100644 --- a/internal/monitoring/canonical_guardrails_test.go +++ b/internal/monitoring/canonical_guardrails_test.go @@ -1331,3 +1331,43 @@ func TestReloadAndRuntimeContextStayOnCanonicalMonitoringPath(t *testing.T) { } } } + +// TestPollersHonorDisabledInstanceFlag asserts that the PVE, PBS, and PMG +// pollers skip instances whose `Disabled` flag is set, both at client +// initialization/reconnect time and on every per-instance iteration of the +// poll loop. The unified connections ledger surfaces `Disabled` as a +// `paused` state, so the monitoring runtime must not drive API calls or +// mark a disabled instance reachable; that behavior is what keeps the +// ledger's pause semantics honest across restarts. +func TestPollersHonorDisabledInstanceFlag(t *testing.T) { + expectations := map[string][]string{ + "monitor_client_init.go": { + "if pve.Disabled {", + "if pbsInst.Disabled {", + "if pmgInst.Disabled {", + }, + "monitor_client_reconnect.go": { + "if pve.Disabled {", + "if pbsInst.Disabled {", + }, + "monitor_pve.go": { + "if instanceCfg.Disabled {", + }, + "monitor_pbs_pmg.go": { + "if instanceCfg.Disabled {", + }, + } + + for file, snippets := range expectations { + data, err := os.ReadFile(file) + if err != nil { + t.Fatalf("failed to read %s: %v", file, err) + } + source := string(data) + for _, snippet := range snippets { + if !strings.Contains(source, snippet) { + t.Fatalf("%s must contain %q so disabled instances stay off the monitoring hot path", file, snippet) + } + } + } +} diff --git a/internal/monitoring/monitor_client_init.go b/internal/monitoring/monitor_client_init.go index fddd962b8..fe0d2b941 100644 --- a/internal/monitoring/monitor_client_init.go +++ b/internal/monitoring/monitor_client_init.go @@ -14,6 +14,10 @@ import ( func (m *Monitor) initPVEClients(cfg *config.Config) { log.Info().Int("count", len(cfg.PVEInstances)).Msg("initializing PVE clients") for _, pve := range cfg.PVEInstances { + if pve.Disabled { + log.Info().Str("instance", pve.Name).Msg("Skipping PVE client init: instance is paused") + continue + } log.Info(). Str("name", pve.Name). Str("host", pve.Host). @@ -78,6 +82,10 @@ func (m *Monitor) initPVEClients(cfg *config.Config) { func (m *Monitor) initPBSClients(cfg *config.Config) { log.Info().Int("count", len(cfg.PBSInstances)).Msg("initializing PBS clients") for _, pbsInst := range cfg.PBSInstances { + if pbsInst.Disabled { + log.Info().Str("instance", pbsInst.Name).Msg("Skipping PBS client init: instance is paused") + continue + } log.Info(). Str("name", pbsInst.Name). Str("host", pbsInst.Host). @@ -112,6 +120,10 @@ func (m *Monitor) initPBSClients(cfg *config.Config) { func (m *Monitor) initPMGClients(cfg *config.Config) { log.Info().Int("count", len(cfg.PMGInstances)).Msg("initializing PMG clients") for _, pmgInst := range cfg.PMGInstances { + if pmgInst.Disabled { + log.Info().Str("instance", pmgInst.Name).Msg("Skipping PMG client init: instance is paused") + continue + } log.Info(). Str("name", pmgInst.Name). Str("host", pmgInst.Host). diff --git a/internal/monitoring/monitor_client_reconnect.go b/internal/monitoring/monitor_client_reconnect.go index 38550b118..e3cd2c0be 100644 --- a/internal/monitoring/monitor_client_reconnect.go +++ b/internal/monitoring/monitor_client_reconnect.go @@ -68,15 +68,23 @@ func (m *Monitor) retryFailedConnections(ctx context.Context) { if m.config != nil { connectionTimeout = m.config.ConnectionTimeout - // Find PVE instances without clients + // Find PVE instances without clients. Skip paused instances — + // a missing client for a paused instance is intentional, not a failure. for _, pve := range m.config.PVEInstances { + if pve.Disabled { + continue + } if _, exists := m.pveClients[pve.Name]; !exists { missingPVE = append(missingPVE, pve) } } - // Find PBS instances without clients + // Find PBS instances without clients. Skip paused instances — + // a missing client for a paused instance is intentional, not a failure. for _, pbsInst := range m.config.PBSInstances { + if pbsInst.Disabled { + continue + } if _, exists := m.pbsClients[pbsInst.Name]; !exists { missingPBS = append(missingPBS, pbsInst) } diff --git a/internal/monitoring/monitor_pbs_coverage_test.go b/internal/monitoring/monitor_pbs_coverage_test.go index 3e123bab1..bff95404e 100644 --- a/internal/monitoring/monitor_pbs_coverage_test.go +++ b/internal/monitoring/monitor_pbs_coverage_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "net/http" "net/http/httptest" + "os" "strings" "testing" "time" @@ -198,3 +199,33 @@ func TestMonitor_PollPBSInstance_DatastoreDetails(t *testing.T) { t.Error("DS2 not found") } } + +// TestPBSAndPMGPollSkipDisabledInstances asserts that the PBS and PMG poll +// entry points short-circuit when their resolved instance config carries +// `Disabled: true`. This is a source-level guardrail for the discovery +// provider surface: the unified connections ledger surfaces `Disabled` as +// `paused`, and the PBS/PMG pollers must not drive live API calls or +// surface ingest while that flag is set, across restarts or reloads. +func TestPBSAndPMGPollSkipDisabledInstances(t *testing.T) { + data, err := os.ReadFile("monitor_pbs_pmg.go") + if err != nil { + t.Fatalf("failed to read monitor_pbs_pmg.go: %v", err) + } + source := string(data) + + // Both PBS and PMG poll flows must explicitly guard on Disabled. + if count := strings.Count(source, "if instanceCfg.Disabled {"); count < 2 { + t.Fatalf("monitor_pbs_pmg.go must contain the Disabled-skip guard in both PBS and PMG poll entry points; found %d", count) + } + + // The guards must short-circuit the poll with an early return so no + // downstream API client is constructed for a paused instance. + for _, snippet := range []string{ + "Skipping PBS poll: instance is paused", + "Skipping PMG poll: instance is paused", + } { + if !strings.Contains(source, snippet) { + t.Fatalf("monitor_pbs_pmg.go must emit debug-log %q when skipping a disabled instance so operators can correlate paused ledger rows with runtime behavior", snippet) + } + } +} diff --git a/internal/monitoring/monitor_pbs_pmg.go b/internal/monitoring/monitor_pbs_pmg.go index 0135dae1c..4da14c93f 100644 --- a/internal/monitoring/monitor_pbs_pmg.go +++ b/internal/monitoring/monitor_pbs_pmg.go @@ -128,6 +128,12 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie log.Error().Str("instance", instanceName).Msg("PBS instance config not found") return } + if instanceCfg.Disabled { + if debugEnabled { + log.Debug().Str("instance", instanceName).Msg("Skipping PBS poll: instance is paused") + } + return + } // Initialize PBS instance with default values pbsInst := models.PBSInstance{ @@ -502,6 +508,12 @@ func (m *Monitor) pollPMGInstance(ctx context.Context, instanceName string, clie pollErr = fmt.Errorf("pmg instance config not found for %s", instanceName) return } + if instanceCfg.Disabled { + if debugEnabled { + log.Debug().Str("instance", instanceName).Msg("Skipping PMG poll: instance is paused") + } + return + } now := time.Now() pmgInst := models.PMGInstance{ diff --git a/internal/monitoring/monitor_pve.go b/internal/monitoring/monitor_pve.go index c74a69dfa..98833ab42 100644 --- a/internal/monitoring/monitor_pve.go +++ b/internal/monitoring/monitor_pve.go @@ -957,6 +957,12 @@ func (m *Monitor) pollPVEInstance(ctx context.Context, instanceName string, clie pollErr = fmt.Errorf("pve instance config not found for %s", instanceName) return } + if instanceCfg.Disabled { + if debugEnabled { + log.Debug().Str("instance", instanceName).Msg("Skipping PVE poll: instance is paused") + } + return + } // Poll nodes nodes, updatedClient, err := m.fetchPVENodes(ctx, instanceName, instanceCfg, client) diff --git a/scripts/release_control/subsystem_lookup_test.py b/scripts/release_control/subsystem_lookup_test.py index c147bc48a..99d1ff436 100644 --- a/scripts/release_control/subsystem_lookup_test.py +++ b/scripts/release_control/subsystem_lookup_test.py @@ -3780,8 +3780,8 @@ class SubsystemLookupTest(unittest.TestCase): { "heading": "## Shared Boundaries", "path": "internal/api/access_control_handlers.go", - "line": 116, - "heading_line": 83, + "line": 121, + "heading_line": 88, } ], )