From 6340cd36f25586ec07a18a95df38198e4a2fa81f Mon Sep 17 00:00:00 2001 From: rcourtman Date: Wed, 10 Jun 2026 10:52:05 +0100 Subject: [PATCH] Dedupe internal/api handler families behind shared flows and generics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Clears the sixteen dupl pair groups in internal/api plus the pkg/pulsecli pair: - router.go: privileged settings endpoints share the serveSetupTokenOrSettingsWrite gate; patrol findings convert via one unifiedFindingFromAI; the five infrastructure-summary chart loops share collectGuestChartData / fillChartSeriesFromBatch; the VM/LXC workloads summary loops share appendGuestWorkloadSummaries. - deploy_handlers.go: preflight and job status/SSE handlers share handleDeployJobStatus / handleDeployJobEvents. - recovery_handlers.go: points series/facets share parseRecoveryListPointsOptions. - truenas_handlers.go / vmware_handlers.go / router_routes_registration.go: the connection update flow (locate, decode-with-fallback, normalize, preserve masked secrets, validate, save, redact) moves to the new platform_connection_shared.go (updatePlatformConnection + decodeOptionalInstanceRequest + the admin-gated item-route builder); per-platform wrappers carry nolint'd declarative wiring only. - docker_metadata.go / guest_metadata.go: GET/PUT payload semantics move to metadata_handlers_shared.go; the zero-record-instead-of-404 contract is pinned by TestContract_MetadataGetPayloadsUseZeroRecordsInsteadOf404. - kubernetes_agents.go / docker_agents.go: lifecycle PUTs share per-handler action helpers. - config_node_handlers.go: PBS/PMG probes share testProxmoxPlatformConnection. - cloud_handoff_handlers.go / purchase_return_redemptions.go: secrets sqlite stores open through openHardenedSecretsDB so permission hardening stays single-sourced. - ai_handlers.go / chat_service_adapter.go: the GetMessages adapters are deliberate contract mirrors — suppressed with nolint and enforced by TestOrchestratorAndChatAdaptersMapTheSameMessageFields. - pkg/pulsecli/actions.go: action subcommands seed env defaults via actionAPIDefaults (tested); the audit/events cobra registration pair is nolint'd parallel wiring. - .golangci.yml: exclude gitignored tmp/ from ./... typechecking. - subsystem_lookup_test.py: refresh the pinned api-contracts.md line numbers shifted by the contract additions. golangci-lint run ./... is now fully green. Full internal/api and pkg/pulsecli test suites pass. --- .golangci.yml | 4 + .../v6/internal/subsystems/ai-runtime.md | 8 +- .../v6/internal/subsystems/api-contracts.md | 26 + internal/api/ai_handlers.go | 1 + internal/api/ai_handlers_test.go | 46 ++ internal/api/chat_service_adapter.go | 1 + internal/api/cloud_handoff_handlers.go | 96 +-- internal/api/config_node_handlers.go | 95 ++- internal/api/deploy_handlers.go | 213 ++---- internal/api/docker_agents.go | 70 +- internal/api/docker_metadata.go | 93 +-- internal/api/guest_metadata.go | 94 +-- internal/api/kubernetes_agents.go | 71 +- internal/api/metadata_handlers_shared.go | 113 +++ internal/api/platform_connection_shared.go | 128 ++++ internal/api/purchase_return_redemptions.go | 50 +- internal/api/recovery_handlers.go | 66 +- internal/api/router.go | 658 +++++++----------- internal/api/router_routes_registration.go | 46 +- internal/api/truenas_handlers.go | 69 +- internal/api/vmware_handlers.go | 69 +- pkg/pulsecli/actions.go | 60 +- pkg/pulsecli/actions_test.go | 28 + .../release_control/subsystem_lookup_test.py | 4 +- 24 files changed, 977 insertions(+), 1132 deletions(-) create mode 100644 internal/api/metadata_handlers_shared.go create mode 100644 internal/api/platform_connection_shared.go diff --git a/.golangci.yml b/.golangci.yml index 02b15b63e..70421e5a9 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -12,6 +12,10 @@ linters: issues: max-same-issues: 0 max-issues-per-linter: 0 + # tmp/ is gitignored scratch (dev-mode module/toolchain caches live there); + # ./... must not typecheck it. + exclude-dirs: + - tmp exclude-rules: # Exclude dupl findings in test files and mock generators — test boilerplate # is inherently repetitive and mock data generators share patterns by design. diff --git a/docs/release-control/v6/internal/subsystems/ai-runtime.md b/docs/release-control/v6/internal/subsystems/ai-runtime.md index 71919822c..ce987a9f2 100644 --- a/docs/release-control/v6/internal/subsystems/ai-runtime.md +++ b/docs/release-control/v6/internal/subsystems/ai-runtime.md @@ -79,7 +79,13 @@ memory history files load through the generic `loadMemoryHistory` `UnifiedFinding`/`unifiedFindingJSON` are deliberate marshal-mirror twins — do not merge them; the mirror invariants are enforced by `TestFindingJSONMirrorStaysInSync` and -`TestUnifiedFindingJSONMirrorStaysInSync`. +`TestUnifiedFindingJSONMirrorStaysInSync`. The same mirror posture applies on +the transport side: `orchestratorChatAdapter.GetMessages` +(`internal/api/ai_handlers.go`) and `chatServiceAdapter.GetMessages` +(`internal/api/chat_service_adapter.go`) convert the same chat-service +messages onto deliberately separate output contracts; do not merge them, and +keep their field mapping aligned — +`TestOrchestratorAndChatAdaptersMapTheSameMessageFields` enforces it. Assistant frontend presentation changes under `frontend-modern/src/components/AI/Chat/` must keep live tool activity aligned diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index 355eb3c1b..070395271 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -26,6 +26,8 @@ product API routes free of maintainer commercial analytics. ## Canonical Files 1. `internal/api/contract_test.go` +1a. `internal/api/platform_connection_shared.go` +1b. `internal/api/metadata_handlers_shared.go` 2. `internal/api/resources.go` 3. `internal/api/discovery_handlers.go` 3. `internal/api/alerts.go` @@ -814,6 +816,30 @@ the canonical monitored-system blocked payload. ## Extension Points +0. Parallel platform/resource handler families share their flow logic instead + of per-platform copies. `internal/api/platform_connection_shared.go` owns + the platform connection update flow (locate by trimmed ID, decode with + stored-record fallback via the generic `decodeOptionalInstanceRequest`, + normalize, preserve masked secrets, validate, persist, respond redacted — + `updatePlatformConnection` + `platformConnectionUpdateSpec`) and the + admin+settings:write item-route mux builder + (`platformConnectionItemRoute`); a new platform connection surface wires a + spec instead of re-rolling the flow, so masked-secret preservation cannot + be forgotten. `internal/api/metadata_handlers_shared.go` owns the + guest/docker metadata GET/PUT payload semantics (empty object instead of + null, zero record instead of 404 — pinned by + `TestContract_MetadataGetPayloadsUseZeroRecordsInsteadOf404`). Deploy + preflights and jobs share `handleDeployJobStatus` / + `handleDeployJobEvents` in `internal/api/deploy_handlers.go`; recovery + points series/facets share `parseRecoveryListPointsOptions`; docker and + kubernetes agent lifecycle PUTs share per-handler + `handleHostLifecycleAction` / `handleClusterLifecycleAction`; PBS/PMG node + connection probes share `testProxmoxPlatformConnection`; secrets-backed + sqlite stores (handoff JTI, purchase-return redemptions) open through + `openHardenedSecretsDB` so the private-permission hardening policy stays + single-sourced; privileged settings endpoints gate through + `serveSetupTokenOrSettingsWrite` in `internal/api/router.go`; patrol + findings convert to the unified store through one `unifiedFindingFromAI`. 1. Add or change payload fields through handler + contract tests together 2. Update frontend API types in lockstep with backend contract changes. Websocket-backed API consumers such as `frontend-modern/src/components/Settings/useAPITokenManagerState.ts` and `frontend-modern/src/components/Settings/useInfrastructureOperationsState.tsx` may read runtime context only through `frontend-modern/src/contexts/appRuntime.ts`; they must not import `frontend-modern/src/App.tsx`, because payload ownership remains in the API contract rather than the root shell. diff --git a/internal/api/ai_handlers.go b/internal/api/ai_handlers.go index 3ead012ba..78010e2cf 100644 --- a/internal/api/ai_handlers.go +++ b/internal/api/ai_handlers.go @@ -2029,6 +2029,7 @@ func (a *orchestratorChatAdapter) ExecuteStream(ctx context.Context, req aicontr }) } +//nolint:dupl // mirrors chatServiceAdapter.GetMessages: same source messages mapped onto a deliberately separate output contract that may diverge func (a *orchestratorChatAdapter) GetMessages(ctx context.Context, sessionID string) ([]aicontracts.OrchestratorMessage, error) { chatMessages, err := a.svc.GetMessages(ctx, sessionID) if err != nil { diff --git a/internal/api/ai_handlers_test.go b/internal/api/ai_handlers_test.go index c6fb6c6da..1398348db 100644 --- a/internal/api/ai_handlers_test.go +++ b/internal/api/ai_handlers_test.go @@ -3316,3 +3316,49 @@ func TestAISettingsHandler_PatrolPreflight_RejectsInvalidProviderName(t *testing assert.Equal(t, http.StatusBadRequest, rec.Code) } + +// TestOrchestratorAndChatAdaptersMapTheSameMessageFields keeps the deliberate +// GetMessages mirror between orchestratorChatAdapter (ai_handlers.go) and +// chatServiceAdapter (chat_service_adapter.go) honest: both convert the same +// chat-service messages onto separate output contracts, and a field mapped by +// one must be mapped by the other. +func TestOrchestratorAndChatAdaptersMapTheSameMessageFields(t *testing.T) { + requiredMappings := []string{ + "ID:", + "Role:", + "Content:", + "ReasoningContent:", + "Timestamp:", + "ToolUseID:", + "IsError:", + ".NormalizeCollections()", + } + + for _, tc := range []struct { + file string + fn string + }{ + {"ai_handlers.go", "func (a *orchestratorChatAdapter) GetMessages("}, + {"chat_service_adapter.go", "func (a *chatServiceAdapter) GetMessages("}, + } { + source, err := os.ReadFile(tc.file) + if err != nil { + t.Fatalf("read %s: %v", tc.file, err) + } + text := string(source) + start := strings.Index(text, tc.fn) + if start < 0 { + t.Fatalf("%s must define %q", tc.file, tc.fn) + } + end := strings.Index(text[start:], "\nfunc ") + if end < 0 { + end = len(text) - start + } + body := text[start : start+end] + for _, required := range requiredMappings { + if !strings.Contains(body, required) { + t.Errorf("%s GetMessages must map %q — keep the adapter mirror in sync", tc.file, required) + } + } + } +} diff --git a/internal/api/chat_service_adapter.go b/internal/api/chat_service_adapter.go index 13a3a6b65..43fc8fd10 100644 --- a/internal/api/chat_service_adapter.go +++ b/internal/api/chat_service_adapter.go @@ -59,6 +59,7 @@ func (a *chatServiceAdapter) ExecutePatrolStream(ctx context.Context, req ai.Pat }, nil } +//nolint:dupl // mirrors orchestratorChatAdapter.GetMessages: same source messages mapped onto a deliberately separate output contract that may diverge func (a *chatServiceAdapter) GetMessages(ctx context.Context, sessionID string) ([]ai.ChatMessage, error) { messages, err := a.svc.GetMessages(ctx, sessionID) if err != nil { diff --git a/internal/api/cloud_handoff_handlers.go b/internal/api/cloud_handoff_handlers.go index 9026b20f0..68e3f35b2 100644 --- a/internal/api/cloud_handoff_handlers.go +++ b/internal/api/cloud_handoff_handlers.go @@ -56,41 +56,57 @@ const deleteExpiredHandoffJTIQuery = ` DELETE FROM handoff_jti INDEXED BY idx_handoff_jti_expires_at WHERE expires_at <= ?` +// openHardenedSecretsDB opens (creating if needed) a single-connection WAL +// sqlite database under /secrets with private dir/file +// permissions, applying schema. dirLabel and label feed error wrapping +// ("handoff" / "handoff jti", "purchase return" / "purchase return +// redemption"). Single-sourcing keeps the permission-hardening policy shared +// by every secrets-backed store. +func openHardenedSecretsDB(configDir, fileName, dirLabel, label, schema string) (*sql.DB, error) { + dir := filepath.Clean(configDir) + if strings.TrimSpace(dir) == "" { + return nil, fmt.Errorf("configDir is required") + } + secretsDir := filepath.Join(dir, "secrets") + if err := os.MkdirAll(secretsDir, handoffPrivateDirPerm); err != nil { + return nil, fmt.Errorf("create %s secrets dir: %w", dirLabel, err) + } + if err := os.Chmod(secretsDir, handoffPrivateDirPerm); err != nil { + return nil, fmt.Errorf("chmod %s secrets dir: %w", dirLabel, err) + } + + dbPath := filepath.Join(secretsDir, fileName) + dsn := dbPath + "?" + url.Values{ + "_pragma": []string{ + "busy_timeout(30000)", + "journal_mode(WAL)", + "synchronous(NORMAL)", + }, + }.Encode() + + db, err := sql.Open("sqlite", dsn) + if err != nil { + return nil, fmt.Errorf("open %s db: %w", label, err) + } + db.SetMaxOpenConns(1) + db.SetMaxIdleConns(1) + db.SetConnMaxLifetime(0) + + if _, err := db.Exec(schema); err != nil { + _ = db.Close() + return nil, fmt.Errorf("init %s schema: %w", label, err) + } + for _, path := range []string{dbPath, dbPath + "-wal", dbPath + "-shm"} { + if err := hardenPrivateFile(path, handoffPrivateFilePerm); err != nil { + _ = db.Close() + return nil, fmt.Errorf("harden %s file permissions: %w", label, err) + } + } + return db, nil +} + func (s *jtiReplayStore) init() { s.once.Do(func() { - dir := filepath.Clean(s.configDir) - if strings.TrimSpace(dir) == "" { - s.initErr = fmt.Errorf("configDir is required") - return - } - secretsDir := filepath.Join(dir, "secrets") - if err := os.MkdirAll(secretsDir, handoffPrivateDirPerm); err != nil { - s.initErr = fmt.Errorf("create handoff secrets dir: %w", err) - return - } - if err := os.Chmod(secretsDir, handoffPrivateDirPerm); err != nil { - s.initErr = fmt.Errorf("chmod handoff secrets dir: %w", err) - return - } - - dbPath := filepath.Join(secretsDir, "handoff_jti.db") - dsn := dbPath + "?" + url.Values{ - "_pragma": []string{ - "busy_timeout(30000)", - "journal_mode(WAL)", - "synchronous(NORMAL)", - }, - }.Encode() - - db, err := sql.Open("sqlite", dsn) - if err != nil { - s.initErr = fmt.Errorf("open handoff jti db: %w", err) - return - } - db.SetMaxOpenConns(1) - db.SetMaxIdleConns(1) - db.SetConnMaxLifetime(0) - schema := ` CREATE TABLE IF NOT EXISTS handoff_jti ( jti TEXT PRIMARY KEY, @@ -98,19 +114,11 @@ func (s *jtiReplayStore) init() { ); CREATE INDEX IF NOT EXISTS idx_handoff_jti_expires_at ON handoff_jti(expires_at); ` - if _, err := db.Exec(schema); err != nil { - _ = db.Close() - s.initErr = fmt.Errorf("init handoff jti schema: %w", err) + db, err := openHardenedSecretsDB(s.configDir, "handoff_jti.db", "handoff", "handoff jti", schema) + if err != nil { + s.initErr = err return } - for _, path := range []string{dbPath, dbPath + "-wal", dbPath + "-shm"} { - if err := hardenPrivateFile(path, handoffPrivateFilePerm); err != nil { - _ = db.Close() - s.initErr = fmt.Errorf("harden handoff jti file permissions: %w", err) - return - } - } - s.db = db }) } diff --git a/internal/api/config_node_handlers.go b/internal/api/config_node_handlers.go index 63309d244..62a71ac91 100644 --- a/internal/api/config_node_handlers.go +++ b/internal/api/config_node_handlers.go @@ -1948,21 +1948,16 @@ func (h *ConfigHandlers) handleTestNode(w http.ResponseWriter, r *http.Request) json.NewEncoder(w).Encode(testResult) } -func testProxmoxBackupConnection(req NodeConfigRequest) map[string]interface{} { +// testProxmoxPlatformConnection runs the shared PBS/PMG node connection +// probe: build a client from the request credentials, exercise one read +// endpoint with a 10s timeout, and report latency. connect returns the +// platform-specific probe call. +func testProxmoxPlatformConnection(req NodeConfigRequest, successMsg string, connect func(verifySSL bool) (func(context.Context) error, error)) map[string]interface{} { verifySSL := false if req.VerifySSL != nil { verifySSL = *req.VerifySSL } - clientConfig := pbs.ClientConfig{ - Host: req.Host, - User: req.User, - Password: req.Password, - TokenName: req.TokenName, - TokenValue: req.TokenValue, - VerifySSL: verifySSL, - Fingerprint: req.Fingerprint, - } - client, err := pbs.NewClient(clientConfig) + probe, err := connect(verifySSL) if err != nil { return map[string]interface{}{ "status": "error", @@ -1974,7 +1969,7 @@ func testProxmoxBackupConnection(req NodeConfigRequest) map[string]interface{} { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() - if _, err := client.GetDatastores(ctx); err != nil { + if err := probe(ctx); err != nil { return map[string]interface{}{ "status": "error", "message": sanitizeErrorMessage(err, "connection"), @@ -1983,47 +1978,49 @@ func testProxmoxBackupConnection(req NodeConfigRequest) map[string]interface{} { latency := time.Since(startTime).Milliseconds() return map[string]interface{}{ "status": "success", - "message": "Connected to PBS instance", + "message": successMsg, "latency": latency, } } +func testProxmoxBackupConnection(req NodeConfigRequest) map[string]interface{} { + return testProxmoxPlatformConnection(req, "Connected to PBS instance", func(verifySSL bool) (func(context.Context) error, error) { + client, err := pbs.NewClient(pbs.ClientConfig{ + Host: req.Host, + User: req.User, + Password: req.Password, + TokenName: req.TokenName, + TokenValue: req.TokenValue, + VerifySSL: verifySSL, + Fingerprint: req.Fingerprint, + }) + if err != nil { + return nil, err + } + return func(ctx context.Context) error { + _, err := client.GetDatastores(ctx) + return err + }, nil + }) +} + func testProxmoxMailGatewayConnection(req NodeConfigRequest) map[string]interface{} { - verifySSL := false - if req.VerifySSL != nil { - verifySSL = *req.VerifySSL - } - clientConfig := pmg.ClientConfig{ - Host: req.Host, - User: req.User, - Password: req.Password, - TokenName: req.TokenName, - TokenValue: req.TokenValue, - VerifySSL: verifySSL, - Fingerprint: req.Fingerprint, - } - client, err := pmg.NewClient(clientConfig) - if err != nil { - return map[string]interface{}{ - "status": "error", - "message": sanitizeErrorMessage(err, "create_client"), + return testProxmoxPlatformConnection(req, "Connected to PMG instance", func(verifySSL bool) (func(context.Context) error, error) { + client, err := pmg.NewClient(pmg.ClientConfig{ + Host: req.Host, + User: req.User, + Password: req.Password, + TokenName: req.TokenName, + TokenValue: req.TokenValue, + VerifySSL: verifySSL, + Fingerprint: req.Fingerprint, + }) + if err != nil { + return nil, err } - } - - startTime := time.Now() - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() - - if _, err := client.GetVersion(ctx); err != nil { - return map[string]interface{}{ - "status": "error", - "message": sanitizeErrorMessage(err, "connection"), - } - } - latency := time.Since(startTime).Milliseconds() - return map[string]interface{}{ - "status": "success", - "message": "Connected to PMG instance", - "latency": latency, - } + return func(ctx context.Context) error { + _, err := client.GetVersion(ctx) + return err + }, nil + }) } diff --git a/internal/api/deploy_handlers.go b/internal/api/deploy_handlers.go index e7f639eab..5e80a88da 100644 --- a/internal/api/deploy_handlers.go +++ b/internal/api/deploy_handlers.go @@ -393,43 +393,45 @@ func (h *DeployHandlers) HandleCreatePreflight(w http.ResponseWriter, r *http.Re json.NewEncoder(w).Encode(resp) } -// HandleGetPreflight returns the current status of a preflight job. -// GET /api/agent-deploy/preflights/{preflightId} -func (h *DeployHandlers) HandleGetPreflight(w http.ResponseWriter, r *http.Request) { +// handleDeployJobStatus serves the shared GET status handler for preflight +// and deploy jobs (both are deploy.Job records in the same store). pathPrefix +// is the route prefix to strip, label the user-facing noun ("Preflight", +// "Job"), logNoun the log vocabulary ("preflight", "deploy"). +func (h *DeployHandlers) handleDeployJobStatus(w http.ResponseWriter, r *http.Request, pathPrefix, label, logNoun string) { if r.Method != http.MethodGet { http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) return } - preflightID := extractPathSuffix(r.URL.Path, "/api/agent-deploy/preflights/") - if preflightID == "" { - writeErrorResponse(w, http.StatusBadRequest, "missing_id", "Preflight ID is required", nil) + jobID := extractPathSuffix(r.URL.Path, pathPrefix) + if jobID == "" { + writeErrorResponse(w, http.StatusBadRequest, "missing_id", label+" ID is required", nil) return } // Strip /events suffix if present (shouldn't happen via routing, but be safe). - preflightID = strings.TrimSuffix(preflightID, "/events") + jobID = strings.TrimSuffix(jobID, "/events") - job, err := h.store.GetJob(r.Context(), preflightID) + job, err := h.store.GetJob(r.Context(), jobID) if err != nil { - log.Error().Err(err).Str("id", preflightID).Msg("Failed to get preflight job") - writeErrorResponse(w, http.StatusInternalServerError, "store_error", "Failed to get preflight", nil) + log.Error().Err(err).Str("id", jobID).Msgf("Failed to get %s job", logNoun) + writeErrorResponse(w, http.StatusInternalServerError, "store_error", "Failed to get "+strings.ToLower(label), nil) return } if job == nil { - writeErrorResponse(w, http.StatusNotFound, "not_found", "Preflight not found", nil) + writeErrorResponse(w, http.StatusNotFound, "not_found", label+" not found", nil) return } // Tenant isolation: verify the job belongs to the caller's org. orgID := resolveTenantOrgID(r) if job.OrgID != orgID { - writeErrorResponse(w, http.StatusNotFound, "not_found", "Preflight not found", nil) + writeErrorResponse(w, http.StatusNotFound, "not_found", label+" not found", nil) return } - targets, err := h.store.GetTargetsForJob(r.Context(), preflightID) + targets, err := h.store.GetTargetsForJob(r.Context(), jobID) if err != nil { - log.Error().Err(err).Str("id", preflightID).Msg("Failed to get preflight targets") + log.Error().Err(err).Str("id", jobID).Msgf("Failed to get %s targets", strings.ToLower(label)) writeErrorResponse(w, http.StatusInternalServerError, "store_error", "Failed to get targets", nil) return } @@ -446,38 +448,39 @@ func (h *DeployHandlers) HandleGetPreflight(w http.ResponseWriter, r *http.Reque json.NewEncoder(w).Encode(resp) } -// HandlePreflightEvents streams SSE events for a preflight job. -// GET /api/agent-deploy/preflights/{preflightId}/events -func (h *DeployHandlers) HandlePreflightEvents(w http.ResponseWriter, r *http.Request) { +// handleDeployJobEvents streams SSE events for a preflight or deploy job: +// replay persisted events, then live events with heartbeats until the client +// disconnects or the job completes. +func (h *DeployHandlers) handleDeployJobEvents(w http.ResponseWriter, r *http.Request, pathPrefix, label, logNoun string) { if r.Method != http.MethodGet { http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) return } - // Extract preflight ID: /api/agent-deploy/preflights/{id}/events - path := strings.TrimPrefix(r.URL.Path, "/api/agent-deploy/preflights/") - preflightID := strings.TrimSuffix(path, "/events") - if preflightID == "" || preflightID == path { - writeErrorResponse(w, http.StatusBadRequest, "missing_id", "Preflight ID is required", nil) + // Extract the job ID: {id}/events + path := strings.TrimPrefix(r.URL.Path, pathPrefix) + jobID := strings.TrimSuffix(path, "/events") + if jobID == "" || jobID == path { + writeErrorResponse(w, http.StatusBadRequest, "missing_id", label+" ID is required", nil) return } - // Verify the preflight exists. - job, err := h.store.GetJob(r.Context(), preflightID) + // Verify the job exists. + job, err := h.store.GetJob(r.Context(), jobID) if err != nil { - log.Error().Err(err).Str("id", preflightID).Msg("Failed to get preflight job for SSE") - writeErrorResponse(w, http.StatusInternalServerError, "store_error", "Failed to get preflight", nil) + log.Error().Err(err).Str("id", jobID).Msgf("Failed to get %s job for SSE", logNoun) + writeErrorResponse(w, http.StatusInternalServerError, "store_error", "Failed to get "+strings.ToLower(label), nil) return } if job == nil { - writeErrorResponse(w, http.StatusNotFound, "not_found", "Preflight not found", nil) + writeErrorResponse(w, http.StatusNotFound, "not_found", label+" not found", nil) return } // Tenant isolation: verify the job belongs to the caller's org. orgID := resolveTenantOrgID(r) if job.OrgID != orgID { - writeErrorResponse(w, http.StatusNotFound, "not_found", "Preflight not found", nil) + writeErrorResponse(w, http.StatusNotFound, "not_found", label+" not found", nil) return } @@ -493,13 +496,13 @@ func (h *DeployHandlers) HandlePreflightEvents(w http.ResponseWriter, r *http.Re // Register SSE client. clientID := generateID("sse") - eventCh := h.addSSEClient(preflightID, clientID) - defer h.removeSSEClient(preflightID, clientID) + eventCh := h.addSSEClient(jobID, clientID) + defer h.removeSSEClient(jobID, clientID) // Send existing events first (replay). - events, replayErr := h.store.GetEventsForJob(r.Context(), preflightID) + events, replayErr := h.store.GetEventsForJob(r.Context(), jobID) if replayErr != nil { - log.Error().Err(replayErr).Str("id", preflightID).Msg("Failed to load events for SSE replay") + log.Error().Err(replayErr).Str("id", jobID).Msg("Failed to load events for SSE replay") // Send an error event so the client knows replay is incomplete. fmt.Fprintf(w, "event: error\ndata: {\"message\":\"failed to load event history\"}\n\n") } @@ -541,6 +544,18 @@ func (h *DeployHandlers) HandlePreflightEvents(w http.ResponseWriter, r *http.Re } } +// HandleGetPreflight returns the current status of a preflight job. +// GET /api/agent-deploy/preflights/{preflightId} +func (h *DeployHandlers) HandleGetPreflight(w http.ResponseWriter, r *http.Request) { + h.handleDeployJobStatus(w, r, "/api/agent-deploy/preflights/", "Preflight", "preflight") +} + +// HandlePreflightEvents streams SSE events for a preflight job. +// GET /api/agent-deploy/preflights/{preflightId}/events +func (h *DeployHandlers) HandlePreflightEvents(w http.ResponseWriter, r *http.Request) { + h.handleDeployJobEvents(w, r, "/api/agent-deploy/preflights/", "Preflight", "preflight") +} + // --- Progress processing --- // processPreflightProgress reads deploy progress events from the agent and @@ -1255,143 +1270,13 @@ func (h *DeployHandlers) HandleCreateJob(w http.ResponseWriter, r *http.Request) // HandleGetJob returns the current status of a deploy job. // GET /api/agent-deploy/jobs/{jobId} func (h *DeployHandlers) HandleGetJob(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodGet { - http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) - return - } - - jobID := extractPathSuffix(r.URL.Path, "/api/agent-deploy/jobs/") - if jobID == "" { - writeErrorResponse(w, http.StatusBadRequest, "missing_id", "Job ID is required", nil) - return - } - jobID = strings.TrimSuffix(jobID, "/events") - - job, err := h.store.GetJob(r.Context(), jobID) - if err != nil { - log.Error().Err(err).Str("id", jobID).Msg("Failed to get deploy job") - writeErrorResponse(w, http.StatusInternalServerError, "store_error", "Failed to get job", nil) - return - } - if job == nil { - writeErrorResponse(w, http.StatusNotFound, "not_found", "Job not found", nil) - return - } - - orgID := resolveTenantOrgID(r) - if job.OrgID != orgID { - writeErrorResponse(w, http.StatusNotFound, "not_found", "Job not found", nil) - return - } - - targets, err := h.store.GetTargetsForJob(r.Context(), jobID) - if err != nil { - log.Error().Err(err).Str("id", jobID).Msg("Failed to get job targets") - writeErrorResponse(w, http.StatusInternalServerError, "store_error", "Failed to get targets", nil) - return - } - - resp := struct { - *deploy.Job - Targets []deploy.Target `json:"targets"` - }{ - Job: job, - Targets: targets, - } - - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(resp) + h.handleDeployJobStatus(w, r, "/api/agent-deploy/jobs/", "Job", "deploy") } // HandleJobEvents streams SSE events for a deploy job. // GET /api/agent-deploy/jobs/{jobId}/events func (h *DeployHandlers) HandleJobEvents(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodGet { - http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) - return - } - - // Extract job ID: /api/agent-deploy/jobs/{id}/events - path := strings.TrimPrefix(r.URL.Path, "/api/agent-deploy/jobs/") - jobID := strings.TrimSuffix(path, "/events") - if jobID == "" || jobID == path { - writeErrorResponse(w, http.StatusBadRequest, "missing_id", "Job ID is required", nil) - return - } - - job, err := h.store.GetJob(r.Context(), jobID) - if err != nil { - log.Error().Err(err).Str("id", jobID).Msg("Failed to get deploy job for SSE") - writeErrorResponse(w, http.StatusInternalServerError, "store_error", "Failed to get job", nil) - return - } - if job == nil { - writeErrorResponse(w, http.StatusNotFound, "not_found", "Job not found", nil) - return - } - - orgID := resolveTenantOrgID(r) - if job.OrgID != orgID { - writeErrorResponse(w, http.StatusNotFound, "not_found", "Job not found", nil) - return - } - - flusher, ok := w.(http.Flusher) - if !ok { - writeErrorResponse(w, http.StatusInternalServerError, "streaming_unsupported", "Streaming not supported", nil) - return - } - - w.Header().Set("Content-Type", "text/event-stream") - w.Header().Set("Cache-Control", "no-cache") - w.Header().Set("Connection", "keep-alive") - - clientID := generateID("sse") - eventCh := h.addSSEClient(jobID, clientID) - defer h.removeSSEClient(jobID, clientID) - - // Replay existing events. - events, replayErr := h.store.GetEventsForJob(r.Context(), jobID) - if replayErr != nil { - log.Error().Err(replayErr).Str("id", jobID).Msg("Failed to load events for SSE replay") - fmt.Fprintf(w, "event: error\ndata: {\"message\":\"failed to load event history\"}\n\n") - } - for _, evt := range events { - data, _ := json.Marshal(evt) - fmt.Fprintf(w, "data: %s\n\n", data) - } - flusher.Flush() - - // If job is terminal, send final and close. - if isDeployJobTerminal(job.Status) { - data, _ := json.Marshal(map[string]string{ - "type": "job_complete", - "status": string(job.Status), - }) - fmt.Fprintf(w, "data: %s\n\n", data) - flusher.Flush() - return - } - - // Stream new events. - heartbeat := time.NewTicker(15 * time.Second) - defer heartbeat.Stop() - - for { - select { - case <-r.Context().Done(): - return - case eventData, ok := <-eventCh: - if !ok { - return - } - fmt.Fprintf(w, "data: %s\n\n", eventData) - flusher.Flush() - case <-heartbeat.C: - fmt.Fprint(w, ": heartbeat\n\n") - flusher.Flush() - } - } + h.handleDeployJobEvents(w, r, "/api/agent-deploy/jobs/", "Job", "deploy") } // HandleCancelJob cancels a running deploy job. diff --git a/internal/api/docker_agents.go b/internal/api/docker_agents.go index e33b057b0..b428f39d7 100644 --- a/internal/api/docker_agents.go +++ b/internal/api/docker_agents.go @@ -1,6 +1,7 @@ package api import ( + "context" "encoding/json" "errors" "net/http" @@ -407,20 +408,29 @@ func (h *DockerAgentHandlers) HandleAllowReenroll(w http.ResponseWriter, r *http } } -// HandleUnhideHost unhides a previously hidden Docker / Podman host. -func (h *DockerAgentHandlers) HandleUnhideHost(w http.ResponseWriter, r *http.Request) { +// handleHostLifecycleAction runs the shared PUT docker host lifecycle flow: +// resolve the agent ID from the suffixed route, apply the monitor action, +// broadcast state, and respond. action returns the canonical host ID. +func (h *DockerAgentHandlers) handleHostLifecycleAction( + w http.ResponseWriter, + r *http.Request, + pathSuffix string, + successMsg string, + logMsg string, + action func(ctx context.Context, agentID string) (string, error), +) { if r.Method != http.MethodPut { writeErrorResponse(w, http.StatusMethodNotAllowed, "method_not_allowed", "Only PUT is allowed", nil) return } - agentID := dockerRuntimeAgentIDFromPath(r.URL.Path, "/unhide") + agentID := dockerRuntimeAgentIDFromPath(r.URL.Path, pathSuffix) if agentID == "" { writeErrorResponse(w, http.StatusBadRequest, "missing_agent_id", "Agent ID is required", nil) return } - host, err := h.getMonitor(r.Context()).UnhideDockerHost(agentID) + hostID, err := action(r.Context(), agentID) if err != nil { writeErrorResponse(w, http.StatusNotFound, "docker_agent_not_found", err.Error(), nil) return @@ -430,41 +440,35 @@ func (h *DockerAgentHandlers) HandleUnhideHost(w http.ResponseWriter, r *http.Re if err := utils.WriteJSONResponse(w, map[string]any{ "success": true, - "agentId": host.ID, - "message": "Docker / Podman module unhidden", + "agentId": hostID, + "message": successMsg, }); err != nil { - log.Error().Err(err).Msg("Failed to serialize docker host unhide response") + log.Error().Err(err).Msg(logMsg) } } +// HandleUnhideHost unhides a previously hidden Docker / Podman host. +func (h *DockerAgentHandlers) HandleUnhideHost(w http.ResponseWriter, r *http.Request) { + h.handleHostLifecycleAction(w, r, "/unhide", "Docker / Podman module unhidden", "Failed to serialize docker host unhide response", + func(ctx context.Context, agentID string) (string, error) { + host, err := h.getMonitor(ctx).UnhideDockerHost(agentID) + if err != nil { + return "", err + } + return host.ID, nil + }) +} + // HandleMarkPendingUninstall marks a Docker / Podman host as pending uninstall. func (h *DockerAgentHandlers) HandleMarkPendingUninstall(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodPut { - writeErrorResponse(w, http.StatusMethodNotAllowed, "method_not_allowed", "Only PUT is allowed", nil) - return - } - - agentID := dockerRuntimeAgentIDFromPath(r.URL.Path, "/pending-uninstall") - if agentID == "" { - writeErrorResponse(w, http.StatusBadRequest, "missing_agent_id", "Agent ID is required", nil) - return - } - - host, err := h.getMonitor(r.Context()).MarkDockerHostPendingUninstall(agentID) - if err != nil { - writeErrorResponse(w, http.StatusNotFound, "docker_agent_not_found", err.Error(), nil) - return - } - - h.broadcastState(r.Context()) - - if err := utils.WriteJSONResponse(w, map[string]any{ - "success": true, - "agentId": host.ID, - "message": "Docker / Podman module marked as pending uninstall", - }); err != nil { - log.Error().Err(err).Msg("Failed to serialize docker host pending uninstall response") - } + h.handleHostLifecycleAction(w, r, "/pending-uninstall", "Docker / Podman module marked as pending uninstall", "Failed to serialize docker host pending uninstall response", + func(ctx context.Context, agentID string) (string, error) { + host, err := h.getMonitor(ctx).MarkDockerHostPendingUninstall(agentID) + if err != nil { + return "", err + } + return host.ID, nil + }) } // HandleSetCustomDisplayName updates the custom display name for a Docker / Podman host. diff --git a/internal/api/docker_metadata.go b/internal/api/docker_metadata.go index 7fa5b8207..d100f036e 100644 --- a/internal/api/docker_metadata.go +++ b/internal/api/docker_metadata.go @@ -45,88 +45,25 @@ func (h *DockerMetadataHandler) Store() *config.DockerMetadataStore { // HandleGetMetadata retrieves metadata for a specific Docker resource or all resources func (h *DockerMetadataHandler) HandleGetMetadata(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodGet { - http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) - return - } - - // Check if requesting specific resource - path := r.URL.Path - // Handle both /api/docker/metadata and /api/docker/metadata/ - if path == "/api/docker/metadata" || path == "/api/docker/metadata/" { - // Get all metadata - w.Header().Set("Content-Type", "application/json") - store := h.getStore(r.Context()) - allMeta := store.GetAll() - if allMeta == nil { - // Return empty object instead of null - json.NewEncoder(w).Encode(make(map[string]*config.DockerMetadata)) - } else { - json.NewEncoder(w).Encode(allMeta) - } - return - } - - // Get specific resource ID from path - resourceID := strings.TrimPrefix(path, "/api/docker/metadata/") - - w.Header().Set("Content-Type", "application/json") - - if resourceID != "" { - // Get specific Docker resource metadata - store := h.getStore(r.Context()) - meta := store.Get(resourceID) - if meta == nil { - // Return empty metadata instead of 404 - json.NewEncoder(w).Encode(&config.DockerMetadata{ID: resourceID}) - } else { - json.NewEncoder(w).Encode(meta) - } - } else { - // This shouldn't happen with current routing, but handle it anyway - http.Error(w, "Invalid request path", http.StatusBadRequest) - } + handleMetadataGetRequest(w, r, "/api/docker/metadata", + func(ctx context.Context) map[string]*config.DockerMetadata { return h.getStore(ctx).GetAll() }, + func(ctx context.Context, id string) *config.DockerMetadata { return h.getStore(ctx).Get(id) }, + func(id string) *config.DockerMetadata { return &config.DockerMetadata{ID: id} }, + ) } // HandleUpdateMetadata updates metadata for a Docker resource func (h *DockerMetadataHandler) HandleUpdateMetadata(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodPut && r.Method != http.MethodPost { - http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) - return - } - - resourceID := strings.TrimPrefix(r.URL.Path, "/api/docker/metadata/") - if resourceID == "" || resourceID == "metadata" { - http.Error(w, "Resource ID required", http.StatusBadRequest) - return - } - - // Limit request body to 16KB to prevent memory exhaustion - r.Body = http.MaxBytesReader(w, r.Body, 16*1024) - - var meta config.DockerMetadata - if err := json.NewDecoder(r.Body).Decode(&meta); err != nil { - http.Error(w, "Invalid request body", http.StatusBadRequest) - return - } - - // Validate URL if provided - if errMsg := validateCustomURL(meta.CustomURL); errMsg != "" { - http.Error(w, errMsg, http.StatusBadRequest) - return - } - - store := h.getStore(r.Context()) - if err := store.Set(resourceID, &meta); err != nil { - log.Error().Err(err).Str("resourceID", resourceID).Msg("Failed to save Docker metadata") - http.Error(w, metadataSaveErrorMessage(err), http.StatusInternalServerError) - return - } - - log.Info().Str("resourceID", resourceID).Str("url", meta.CustomURL).Msg("Updated Docker metadata") - - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(&meta) + handleMetadataUpdateRequest(w, r, "/api/docker/metadata", + "Resource ID required", + "resourceID", + "Failed to save Docker metadata", + "Updated Docker metadata", + func(meta *config.DockerMetadata) string { return meta.CustomURL }, + func(ctx context.Context, id string, meta *config.DockerMetadata) error { + return h.getStore(ctx).Set(id, meta) + }, + ) } // HandleDeleteMetadata removes metadata for a Docker resource diff --git a/internal/api/guest_metadata.go b/internal/api/guest_metadata.go index 6ff975b78..c59824cd1 100644 --- a/internal/api/guest_metadata.go +++ b/internal/api/guest_metadata.go @@ -2,7 +2,6 @@ package api import ( "context" - "encoding/json" "net/http" "strings" @@ -52,88 +51,25 @@ func (h *GuestMetadataHandler) Store() *config.GuestMetadataStore { // HandleGetMetadata retrieves metadata for a specific guest or all guests func (h *GuestMetadataHandler) HandleGetMetadata(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodGet { - http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) - return - } - - // Check if requesting specific guest - path := r.URL.Path - // Handle both /api/guests/metadata and /api/guests/metadata/ - if path == "/api/guests/metadata" || path == "/api/guests/metadata/" { - // Get all metadata - w.Header().Set("Content-Type", "application/json") - store := h.getStore(r.Context()) - allMeta := store.GetAll() - if allMeta == nil { - // Return empty object instead of null - json.NewEncoder(w).Encode(make(map[string]*config.GuestMetadata)) - } else { - json.NewEncoder(w).Encode(allMeta) - } - return - } - - // Get specific guest ID from path - guestID := strings.TrimPrefix(path, "/api/guests/metadata/") - - w.Header().Set("Content-Type", "application/json") - - if guestID != "" { - // Get specific guest metadata - store := h.getStore(r.Context()) - meta := store.Get(guestID) - if meta == nil { - // Return empty metadata instead of 404 - json.NewEncoder(w).Encode(&config.GuestMetadata{ID: guestID}) - } else { - json.NewEncoder(w).Encode(meta) - } - } else { - // This shouldn't happen with current routing, but handle it anyway - http.Error(w, "Invalid request path", http.StatusBadRequest) - } + handleMetadataGetRequest(w, r, "/api/guests/metadata", + func(ctx context.Context) map[string]*config.GuestMetadata { return h.getStore(ctx).GetAll() }, + func(ctx context.Context, id string) *config.GuestMetadata { return h.getStore(ctx).Get(id) }, + func(id string) *config.GuestMetadata { return &config.GuestMetadata{ID: id} }, + ) } // HandleUpdateMetadata updates metadata for a guest func (h *GuestMetadataHandler) HandleUpdateMetadata(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodPut && r.Method != http.MethodPost { - http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) - return - } - - guestID := strings.TrimPrefix(r.URL.Path, "/api/guests/metadata/") - if guestID == "" || guestID == "metadata" { - http.Error(w, "Guest ID required", http.StatusBadRequest) - return - } - - // Limit request body to 16KB to prevent memory exhaustion - r.Body = http.MaxBytesReader(w, r.Body, 16*1024) - - var meta config.GuestMetadata - if err := json.NewDecoder(r.Body).Decode(&meta); err != nil { - http.Error(w, "Invalid request body", http.StatusBadRequest) - return - } - - // Validate URL if provided - if errMsg := validateCustomURL(meta.CustomURL); errMsg != "" { - http.Error(w, errMsg, http.StatusBadRequest) - return - } - - store := h.getStore(r.Context()) - if err := store.Set(guestID, &meta); err != nil { - log.Error().Err(err).Str("guestID", guestID).Msg("Failed to save guest metadata") - http.Error(w, metadataSaveErrorMessage(err), http.StatusInternalServerError) - return - } - - log.Info().Str("guestID", guestID).Str("url", meta.CustomURL).Msg("Updated guest metadata") - - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(&meta) + handleMetadataUpdateRequest(w, r, "/api/guests/metadata", + "Guest ID required", + "guestID", + "Failed to save guest metadata", + "Updated guest metadata", + func(meta *config.GuestMetadata) string { return meta.CustomURL }, + func(ctx context.Context, id string, meta *config.GuestMetadata) error { + return h.getStore(ctx).Set(id, meta) + }, + ) } // HandleDeleteMetadata removes metadata for a guest diff --git a/internal/api/kubernetes_agents.go b/internal/api/kubernetes_agents.go index 738b50008..ed54406f6 100644 --- a/internal/api/kubernetes_agents.go +++ b/internal/api/kubernetes_agents.go @@ -194,22 +194,31 @@ func (h *KubernetesAgentHandlers) HandleAllowReenroll(w http.ResponseWriter, r * } } -// HandleUnhideCluster unhides a previously hidden kubernetes cluster. -func (h *KubernetesAgentHandlers) HandleUnhideCluster(w http.ResponseWriter, r *http.Request) { +// handleClusterLifecycleAction runs the shared PUT cluster lifecycle flow: +// resolve the cluster ID from the suffixed route, apply the monitor action, +// broadcast state, and respond. action returns the canonical cluster ID. +func (h *KubernetesAgentHandlers) handleClusterLifecycleAction( + w http.ResponseWriter, + r *http.Request, + pathSuffix string, + successMsg string, + logMsg string, + action func(ctx context.Context, clusterID string) (string, error), +) { if r.Method != http.MethodPut { writeErrorResponse(w, http.StatusMethodNotAllowed, "method_not_allowed", "Only PUT is allowed", nil) return } trimmedPath := strings.TrimPrefix(r.URL.Path, "/api/agents/kubernetes/clusters/") - trimmedPath = strings.TrimSuffix(trimmedPath, "/unhide") + trimmedPath = strings.TrimSuffix(trimmedPath, pathSuffix) clusterID := strings.TrimSpace(trimmedPath) if clusterID == "" { writeErrorResponse(w, http.StatusBadRequest, "missing_cluster_id", "Kubernetes cluster ID is required", nil) return } - cluster, err := h.getMonitor(r.Context()).UnhideKubernetesCluster(clusterID) + canonicalID, err := action(r.Context(), clusterID) if err != nil { writeErrorResponse(w, http.StatusNotFound, "k8s_cluster_not_found", err.Error(), nil) return @@ -219,43 +228,35 @@ func (h *KubernetesAgentHandlers) HandleUnhideCluster(w http.ResponseWriter, r * if err := utils.WriteJSONResponse(w, map[string]any{ "success": true, - "clusterId": cluster.ID, - "message": "Kubernetes cluster unhidden", + "clusterId": canonicalID, + "message": successMsg, }); err != nil { - log.Error().Err(err).Msg("Failed to serialize kubernetes cluster unhide response") + log.Error().Err(err).Msg(logMsg) } } +// HandleUnhideCluster unhides a previously hidden kubernetes cluster. +func (h *KubernetesAgentHandlers) HandleUnhideCluster(w http.ResponseWriter, r *http.Request) { + h.handleClusterLifecycleAction(w, r, "/unhide", "Kubernetes cluster unhidden", "Failed to serialize kubernetes cluster unhide response", + func(ctx context.Context, clusterID string) (string, error) { + cluster, err := h.getMonitor(ctx).UnhideKubernetesCluster(clusterID) + if err != nil { + return "", err + } + return cluster.ID, nil + }) +} + // HandleMarkPendingUninstall marks a cluster as pending uninstall. func (h *KubernetesAgentHandlers) HandleMarkPendingUninstall(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodPut { - writeErrorResponse(w, http.StatusMethodNotAllowed, "method_not_allowed", "Only PUT is allowed", nil) - return - } - - trimmedPath := strings.TrimPrefix(r.URL.Path, "/api/agents/kubernetes/clusters/") - trimmedPath = strings.TrimSuffix(trimmedPath, "/pending-uninstall") - clusterID := strings.TrimSpace(trimmedPath) - if clusterID == "" { - writeErrorResponse(w, http.StatusBadRequest, "missing_cluster_id", "Kubernetes cluster ID is required", nil) - return - } - - cluster, err := h.getMonitor(r.Context()).MarkKubernetesClusterPendingUninstall(clusterID) - if err != nil { - writeErrorResponse(w, http.StatusNotFound, "k8s_cluster_not_found", err.Error(), nil) - return - } - - h.broadcastState(r.Context()) - - if err := utils.WriteJSONResponse(w, map[string]any{ - "success": true, - "clusterId": cluster.ID, - "message": "Kubernetes cluster marked as pending uninstall", - }); err != nil { - log.Error().Err(err).Msg("Failed to serialize kubernetes cluster pending uninstall response") - } + h.handleClusterLifecycleAction(w, r, "/pending-uninstall", "Kubernetes cluster marked as pending uninstall", "Failed to serialize kubernetes cluster pending uninstall response", + func(ctx context.Context, clusterID string) (string, error) { + cluster, err := h.getMonitor(ctx).MarkKubernetesClusterPendingUninstall(clusterID) + if err != nil { + return "", err + } + return cluster.ID, nil + }) } // HandleSetCustomDisplayName updates the custom display name for a kubernetes cluster. diff --git a/internal/api/metadata_handlers_shared.go b/internal/api/metadata_handlers_shared.go new file mode 100644 index 000000000..152e3cff3 --- /dev/null +++ b/internal/api/metadata_handlers_shared.go @@ -0,0 +1,113 @@ +package api + +import ( + "context" + "encoding/json" + "net/http" + "strings" + + "github.com/rs/zerolog/log" +) + +// handleMetadataGetRequest serves the GET flow shared by the guest and docker +// metadata handlers: the bare route returns the full metadata map (an empty +// object instead of null), a suffixed route returns that resource's metadata +// (a zero record instead of a 404). +func handleMetadataGetRequest[M any]( + w http.ResponseWriter, + r *http.Request, + routePrefix string, + getAll func(ctx context.Context) map[string]*M, + get func(ctx context.Context, id string) *M, + emptyRecord func(id string) *M, +) { + if r.Method != http.MethodGet { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + + path := r.URL.Path + // Handle both the bare route and its trailing-slash form. + if path == routePrefix || path == routePrefix+"/" { + // Get all metadata + w.Header().Set("Content-Type", "application/json") + allMeta := getAll(r.Context()) + if allMeta == nil { + // Return empty object instead of null + json.NewEncoder(w).Encode(make(map[string]*M)) + } else { + json.NewEncoder(w).Encode(allMeta) + } + return + } + + // Get specific resource ID from path + resourceID := strings.TrimPrefix(path, routePrefix+"/") + + w.Header().Set("Content-Type", "application/json") + + if resourceID != "" { + meta := get(r.Context(), resourceID) + if meta == nil { + // Return empty metadata instead of 404 + json.NewEncoder(w).Encode(emptyRecord(resourceID)) + } else { + json.NewEncoder(w).Encode(meta) + } + } else { + // This shouldn't happen with current routing, but handle it anyway + http.Error(w, "Invalid request path", http.StatusBadRequest) + } +} + +// handleMetadataUpdateRequest serves the PUT/POST flow shared by the guest +// and docker metadata handlers: decode a bounded body, validate the custom +// URL, persist, and echo the stored record. +func handleMetadataUpdateRequest[M any]( + w http.ResponseWriter, + r *http.Request, + routePrefix string, + idRequiredMsg string, + idLogKey string, + saveErrLogMsg string, + updatedLogMsg string, + customURL func(*M) string, + set func(ctx context.Context, id string, meta *M) error, +) { + if r.Method != http.MethodPut && r.Method != http.MethodPost { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + + resourceID := strings.TrimPrefix(r.URL.Path, routePrefix+"/") + if resourceID == "" || resourceID == "metadata" { + http.Error(w, idRequiredMsg, http.StatusBadRequest) + return + } + + // Limit request body to 16KB to prevent memory exhaustion + r.Body = http.MaxBytesReader(w, r.Body, 16*1024) + + var meta M + if err := json.NewDecoder(r.Body).Decode(&meta); err != nil { + http.Error(w, "Invalid request body", http.StatusBadRequest) + return + } + + // Validate URL if provided + if errMsg := validateCustomURL(customURL(&meta)); errMsg != "" { + http.Error(w, errMsg, http.StatusBadRequest) + return + } + + if err := set(r.Context(), resourceID, &meta); err != nil { + log.Error().Err(err).Str(idLogKey, resourceID).Msg(saveErrLogMsg) + http.Error(w, metadataSaveErrorMessage(err), http.StatusInternalServerError) + return + } + + log.Info().Str(idLogKey, resourceID).Str("url", customURL(&meta)).Msg(updatedLogMsg) + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(&meta) +} diff --git a/internal/api/platform_connection_shared.go b/internal/api/platform_connection_shared.go new file mode 100644 index 000000000..0caeb3c3a --- /dev/null +++ b/internal/api/platform_connection_shared.go @@ -0,0 +1,128 @@ +package api + +import ( + "bytes" + "encoding/json" + "io" + "net/http" + "strings" + + "github.com/rcourtman/pulse-go-rewrite/internal/config" +) + +// decodeOptionalInstanceRequest decodes an optional JSON body onto a copy of +// base. The results are (instance, decoded, ok): an empty body returns base +// with decoded=false, a malformed body writes a 400 response and returns +// ok=false. Shared by the TrueNAS and VMware connection handlers. +func decodeOptionalInstanceRequest[T any](w http.ResponseWriter, r *http.Request, base T) (T, bool, bool) { + var zero T + r.Body = http.MaxBytesReader(w, r.Body, 32*1024) + defer r.Body.Close() + + body, err := io.ReadAll(r.Body) + if err != nil { + writeErrorResponse(w, http.StatusBadRequest, "invalid_request", "Invalid request body", map[string]string{"error": err.Error()}) + return zero, false, false + } + if len(bytes.TrimSpace(body)) == 0 { + return base, false, true + } + + instance := base + if err := json.Unmarshal(body, &instance); err != nil { + writeErrorResponse(w, http.StatusBadRequest, "invalid_request", "Invalid request body", map[string]string{"error": err.Error()}) + return zero, false, false + } + + return instance, true, true +} + +// platformConnectionUpdateSpec wires the per-platform pieces of the shared +// connection-update flow. The flow itself — locate by trimmed ID, decode with +// the stored record as fallback, normalize, preserve masked secrets, +// validate, persist, respond redacted — is platform-invariant policy; +// single-sourcing it means a new platform cannot forget masked-secret +// preservation. +type platformConnectionUpdateSpec[T any] struct { + notFoundCode string + saveFailCode string + saveFailMsg string + id func(T) string + decode func(http.ResponseWriter, *http.Request, T) (T, bool) + setID func(*T, string) + normalize func(*T) + preserve func(updated *T, stored T) + validate func(T) error + redacted func(T) any + save func([]T) error +} + +// updatePlatformConnection runs the shared platform connection update flow +// over the loaded instances for the resolved connection ID. +func updatePlatformConnection[T any](w http.ResponseWriter, r *http.Request, connectionID string, instances []T, spec platformConnectionUpdateSpec[T]) { + index := -1 + for i := range instances { + if strings.TrimSpace(spec.id(instances[i])) == connectionID { + index = i + break + } + } + if index < 0 { + writeErrorResponse(w, http.StatusNotFound, spec.notFoundCode, "Connection not found", nil) + return + } + + instance, ok := spec.decode(w, r, instances[index]) + if !ok { + return + } + spec.setID(&instance, connectionID) + spec.normalize(&instance) + spec.preserve(&instance, instances[index]) + if err := spec.validate(instance); err != nil { + writeErrorResponse(w, http.StatusBadRequest, "validation_error", err.Error(), nil) + return + } + instances[index] = instance + if err := spec.save(instances); err != nil { + writeErrorResponse(w, http.StatusInternalServerError, spec.saveFailCode, spec.saveFailMsg, map[string]string{"error": err.Error()}) + return + } + + writeJSON(w, http.StatusOK, spec.redacted(instance)) +} + +// platformConnectionItemHandlers carries the per-platform handlers for the +// "/api//connections/" item routes. +type platformConnectionItemHandlers struct { + test http.HandlerFunc + preview http.HandlerFunc + delete http.HandlerFunc + update http.HandlerFunc +} + +// platformConnectionItemRoute builds the shared mux handler for platform +// connection item routes: POST .../test, POST .../preview, DELETE, PUT — all +// admin + settings:write gated. handlers is resolved per request so late +// handler wiring and nil-service checks keep working. +func (r *Router) platformConnectionItemRoute(unavailableCode, unavailableMsg string, handlers func() *platformConnectionItemHandlers) http.HandlerFunc { + return func(w http.ResponseWriter, req *http.Request) { + h := handlers() + if h == nil { + writeErrorResponse(w, http.StatusServiceUnavailable, unavailableCode, unavailableMsg, nil) + return + } + + if req.Method == http.MethodPost && strings.HasSuffix(strings.Trim(req.URL.Path, "/"), "/test") { + RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, h.test))(w, req) + } else if req.Method == http.MethodPost && strings.HasSuffix(strings.Trim(req.URL.Path, "/"), "/preview") { + RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, h.preview))(w, req) + } else if req.Method == http.MethodDelete { + RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, h.delete))(w, req) + } else if req.Method == http.MethodPut { + RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, h.update))(w, req) + } else { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + } + } +} diff --git a/internal/api/purchase_return_redemptions.go b/internal/api/purchase_return_redemptions.go index 7b0ec3ade..6775a5da0 100644 --- a/internal/api/purchase_return_redemptions.go +++ b/internal/api/purchase_return_redemptions.go @@ -3,9 +3,6 @@ package api import ( "database/sql" "fmt" - "net/url" - "os" - "path/filepath" "strings" "sync" "time" @@ -59,39 +56,6 @@ type purchaseReturnRedemptionRecord struct { func (s *purchaseReturnRedemptionStore) init() { s.once.Do(func() { - dir := filepath.Clean(s.configDir) - if strings.TrimSpace(dir) == "" { - s.initErr = fmt.Errorf("configDir is required") - return - } - secretsDir := filepath.Join(dir, "secrets") - if err := os.MkdirAll(secretsDir, handoffPrivateDirPerm); err != nil { - s.initErr = fmt.Errorf("create purchase return secrets dir: %w", err) - return - } - if err := os.Chmod(secretsDir, handoffPrivateDirPerm); err != nil { - s.initErr = fmt.Errorf("chmod purchase return secrets dir: %w", err) - return - } - - dbPath := filepath.Join(secretsDir, "purchase_return_redemptions.db") - dsn := dbPath + "?" + url.Values{ - "_pragma": []string{ - "busy_timeout(30000)", - "journal_mode(WAL)", - "synchronous(NORMAL)", - }, - }.Encode() - - db, err := sql.Open("sqlite", dsn) - if err != nil { - s.initErr = fmt.Errorf("open purchase return redemption db: %w", err) - return - } - db.SetMaxOpenConns(1) - db.SetMaxIdleConns(1) - db.SetConnMaxLifetime(0) - schema := ` CREATE TABLE IF NOT EXISTS purchase_return_redemptions ( portal_handoff_id TEXT NOT NULL, @@ -115,19 +79,11 @@ func (s *purchaseReturnRedemptionStore) init() { CREATE INDEX IF NOT EXISTS idx_purchase_return_redemptions_status ON purchase_return_redemptions(status); ` - if _, err := db.Exec(schema); err != nil { - _ = db.Close() - s.initErr = fmt.Errorf("init purchase return redemption schema: %w", err) + db, err := openHardenedSecretsDB(s.configDir, "purchase_return_redemptions.db", "purchase return", "purchase return redemption", schema) + if err != nil { + s.initErr = err return } - for _, path := range []string{dbPath, dbPath + "-wal", dbPath + "-shm"} { - if err := hardenPrivateFile(path, handoffPrivateFilePerm); err != nil { - _ = db.Close() - s.initErr = fmt.Errorf("harden purchase return redemption file permissions: %w", err) - return - } - } - s.db = db }) } diff --git a/internal/api/recovery_handlers.go b/internal/api/recovery_handlers.go index 539569065..1b0617390 100644 --- a/internal/api/recovery_handlers.go +++ b/internal/api/recovery_handlers.go @@ -292,31 +292,25 @@ func parseBoolQuery(qs map[string][]string, key string) bool { } } -func (h *RecoveryHandlers) HandleListSeries(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodGet { - http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) - return - } - - qs := r.URL.Query() - +// parseRecoveryListPointsOptions parses the recovery points filter options +// shared by the points-series and points-facets handlers. It writes a 400 +// response and returns false when a time bound is not RFC3339. +func parseRecoveryListPointsOptions(w http.ResponseWriter, qs url.Values) (recovery.ListPointsOptions, bool) { var from, to *time.Time if t, err := parseRFC3339QueryTime(qs.Get("from")); err != nil { writeErrorResponse(w, http.StatusBadRequest, "invalid_from", "Invalid from time; must be RFC3339", map[string]string{"error": err.Error()}) - return + return recovery.ListPointsOptions{}, false } else { from = t } if t, err := parseRFC3339QueryTime(qs.Get("to")); err != nil { writeErrorResponse(w, http.StatusBadRequest, "invalid_to", "Invalid to time; must be RFC3339", map[string]string{"error": err.Error()}) - return + return recovery.ListPointsOptions{}, false } else { to = t } - tzOffsetMin := parseIntQuery(qs, "tzOffsetMinutes", 0) - - opts := recovery.ListPointsOptions{ + return recovery.ListPointsOptions{ Provider: parseRecoveryPlatformQuery(qs), Kind: recovery.Kind(strings.TrimSpace(qs.Get("kind"))), Mode: recovery.Mode(strings.TrimSpace(qs.Get("mode"))), @@ -332,8 +326,23 @@ func (h *RecoveryHandlers) HandleListSeries(w http.ResponseWriter, r *http.Reque NamespaceLabel: strings.TrimSpace(firstNonEmpty(qs.Get("namespace"), qs.Get("namespaceLabel"))), WorkloadOnly: strings.TrimSpace(qs.Get("scope")) == "workload" || parseBoolQuery(qs, "workloadOnly"), Verification: strings.TrimSpace(firstNonEmpty(qs.Get("verification"), qs.Get("verified"))), + }, true +} + +func (h *RecoveryHandlers) HandleListSeries(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return } + qs := r.URL.Query() + + opts, ok := parseRecoveryListPointsOptions(w, qs) + if !ok { + return + } + tzOffsetMin := parseIntQuery(qs, "tzOffsetMinutes", 0) + var series []recovery.PointsSeriesBucket if mock.IsMockEnabled() { all := mock.CurrentFixtureGraph().RecoveryPoints() @@ -368,36 +377,9 @@ func (h *RecoveryHandlers) HandleListFacets(w http.ResponseWriter, r *http.Reque qs := r.URL.Query() - var from, to *time.Time - if t, err := parseRFC3339QueryTime(qs.Get("from")); err != nil { - writeErrorResponse(w, http.StatusBadRequest, "invalid_from", "Invalid from time; must be RFC3339", map[string]string{"error": err.Error()}) + opts, ok := parseRecoveryListPointsOptions(w, qs) + if !ok { return - } else { - from = t - } - if t, err := parseRFC3339QueryTime(qs.Get("to")); err != nil { - writeErrorResponse(w, http.StatusBadRequest, "invalid_to", "Invalid to time; must be RFC3339", map[string]string{"error": err.Error()}) - return - } else { - to = t - } - - opts := recovery.ListPointsOptions{ - Provider: parseRecoveryPlatformQuery(qs), - Kind: recovery.Kind(strings.TrimSpace(qs.Get("kind"))), - Mode: recovery.Mode(strings.TrimSpace(qs.Get("mode"))), - Outcome: recovery.Outcome(strings.TrimSpace(qs.Get("outcome"))), - ItemType: recovery.NormalizeRecoveryItemType(firstNonEmpty(qs.Get("itemType"), qs.Get("type"))), - SubjectResourceID: parseRecoveryItemResourceIDQuery(qs), - RollupID: strings.TrimSpace(qs.Get("rollupId")), - From: from, - To: to, - Query: strings.TrimSpace(firstNonEmpty(qs.Get("q"), qs.Get("query"))), - ClusterLabel: strings.TrimSpace(firstNonEmpty(qs.Get("cluster"), qs.Get("clusterLabel"))), - NodeHostLabel: strings.TrimSpace(firstNonEmpty(qs.Get("node"), qs.Get("nodeHost"), qs.Get("nodeHostLabel"))), - NamespaceLabel: strings.TrimSpace(firstNonEmpty(qs.Get("namespace"), qs.Get("namespaceLabel"))), - WorkloadOnly: strings.TrimSpace(qs.Get("scope")) == "workload" || parseBoolQuery(qs, "workloadOnly"), - Verification: strings.TrimSpace(firstNonEmpty(qs.Get("verification"), qs.Get("verified"))), } var facets recovery.PointsFacets diff --git a/internal/api/router.go b/internal/api/router.go index a2efe9c53..099d3031f 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -1075,15 +1075,14 @@ func (r *Router) handleAgentWebSocket(w http.ResponseWriter, req *http.Request) r.agentExecServer.HandleWebSocket(w, req) } -func (r *Router) handleVerifyTemperatureSSH(w http.ResponseWriter, req *http.Request) { - if r.configHandlers == nil { - http.Error(w, "Service unavailable", http.StatusServiceUnavailable) - return - } - +// serveSetupTokenOrSettingsWrite gates a privileged settings endpoint behind +// either a valid setup token (setup scripts) or full authentication plus +// proxy-auth admin and settings:write scope. logLabel tags the unauthorized +// log line; nonAdminMsg is the proxy-auth denial log message. +func (r *Router) serveSetupTokenOrSettingsWrite(w http.ResponseWriter, req *http.Request, logLabel, nonAdminMsg string, delegate http.HandlerFunc) { // Check setup token first (for setup scripts) if r.isValidSetupTokenForRequest(req) { - r.configHandlers.HandleVerifyTemperatureSSH(w, req) + delegate(w, req) return } @@ -1094,7 +1093,7 @@ func (r *Router) handleVerifyTemperatureSSH(w http.ResponseWriter, req *http.Req Str("ip", req.RemoteAddr). Str("path", req.URL.Path). Str("method", req.Method). - Msg("Unauthorized access attempt (verify-temperature-ssh)") + Msg("Unauthorized access attempt (" + logLabel + ")") if !authWriter.wrote { writeAuthenticationRequired(w, req) @@ -1108,18 +1107,26 @@ func (r *Router) handleVerifyTemperatureSSH(w http.ResponseWriter, req *http.Req log.Warn(). Str("ip", GetClientIP(req)). Str("username", username). - Msg("Non-admin user attempted verify-temperature-ssh") + Msg(nonAdminMsg) http.Error(w, "Admin privileges required", http.StatusForbidden) return } } - // Require admin session identity or settings:write token for privileged SSH probes. + // Require admin session identity or settings:write token for privileged writes. if !ensureSettingsWriteScope(r.config, w, req) { return } - r.configHandlers.HandleVerifyTemperatureSSH(w, req) + delegate(w, req) +} + +func (r *Router) handleVerifyTemperatureSSH(w http.ResponseWriter, req *http.Request) { + if r.configHandlers == nil { + http.Error(w, "Service unavailable", http.StatusServiceUnavailable) + return + } + r.serveSetupTokenOrSettingsWrite(w, req, "verify-temperature-ssh", "Non-admin user attempted verify-temperature-ssh", r.configHandlers.HandleVerifyTemperatureSSH) } // handleSSHConfig handles SSH config writes with setup token or API auth @@ -1128,46 +1135,7 @@ func (r *Router) handleSSHConfig(w http.ResponseWriter, req *http.Request) { http.Error(w, "Service unavailable", http.StatusServiceUnavailable) return } - - // Check setup token first (for setup scripts) - if r.isValidSetupTokenForRequest(req) { - r.systemSettingsHandler.HandleSSHConfig(w, req) - return - } - - // Require authentication - authWriter := &responseCapture{ResponseWriter: w} - if !checkAuth(r.config, authWriter, req, false) { - log.Warn(). - Str("ip", req.RemoteAddr). - Str("path", req.URL.Path). - Str("method", req.Method). - Msg("Unauthorized access attempt (ssh-config)") - - if !authWriter.wrote { - writeAuthenticationRequired(w, req) - } - return - } - - // Check admin privileges for proxy auth users - if r.config.ProxyAuthSecret != "" { - if valid, username, isAdmin := CheckProxyAuth(r.config, req); valid && !isAdmin { - log.Warn(). - Str("ip", GetClientIP(req)). - Str("username", username). - Msg("Non-admin user attempted ssh-config update") - http.Error(w, "Admin privileges required", http.StatusForbidden) - return - } - } - - // Require admin session identity or settings:write token for privileged SSH config writes. - if !ensureSettingsWriteScope(r.config, w, req) { - return - } - - r.systemSettingsHandler.HandleSSHConfig(w, req) + r.serveSetupTokenOrSettingsWrite(w, req, "ssh-config", "Non-admin user attempted ssh-config update", r.systemSettingsHandler.HandleSSHConfig) } func extractSetupToken(req *http.Request) string { @@ -1693,6 +1661,69 @@ func (r *Router) StartPatrolForOrg(ctx context.Context, orgID string) { } } +// unifiedLifecycleFromAI converts patrol finding lifecycle events into their +// unified-store representation. +func unifiedLifecycleFromAI(events []ai.FindingLifecycleEvent) []unified.UnifiedFindingLifecycleEvent { + if len(events) == 0 { + return nil + } + out := make([]unified.UnifiedFindingLifecycleEvent, 0, len(events)) + for _, e := range events { + out = append(out, unified.UnifiedFindingLifecycleEvent{ + At: e.At, + Type: e.Type, + Message: e.Message, + From: e.From, + To: e.To, + Metadata: e.Metadata, + }) + } + return out +} + +// unifiedFindingFromAI converts a patrol ai.Finding into its unified-store +// representation. Shared by the live patrol callback and the startup sync of +// persisted findings. +func unifiedFindingFromAI(f *ai.Finding) *unified.UnifiedFinding { + return &unified.UnifiedFinding{ + ID: f.ID, + Source: unified.SourceAIPatrol, + Severity: unified.UnifiedSeverity(f.Severity), + Category: unified.UnifiedCategory(f.Category), + ResourceID: f.ResourceID, + ResourceName: f.ResourceName, + ResourceType: f.ResourceType, + Node: f.Node, + Title: f.Title, + Description: f.Description, + Impact: f.Impact, + PreviousResolvedFixSummary: f.PreviousResolvedFixSummary, + Recommendation: f.Recommendation, + Evidence: f.Evidence, + DetectedAt: f.DetectedAt, + LastSeenAt: f.LastSeenAt, + ResolvedAt: f.ResolvedAt, + AutoResolved: f.AutoResolved, + InvestigationSessionID: f.InvestigationSessionID, + InvestigationStatus: f.InvestigationStatus, + InvestigationOutcome: f.InvestigationOutcome, + LastInvestigatedAt: f.LastInvestigatedAt, + InvestigationAttempts: f.InvestigationAttempts, + InvestigationRecord: f.InvestigationRecord, + LoopState: f.LoopState, + Lifecycle: unifiedLifecycleFromAI(f.Lifecycle), + RegressionCount: f.RegressionCount, + LastRegressionAt: f.LastRegressionAt, + AcknowledgedAt: f.AcknowledgedAt, + SnoozedUntil: f.SnoozedUntil, + DismissedReason: f.DismissedReason, + UserNote: f.UserNote, + Suppressed: f.Suppressed, + TimesRaised: f.TimesRaised, + RemindAt: f.RemindAt, + } +} + func (r *Router) startPatrolForContext(ctx context.Context, orgID string) bool { if r == nil || r.aiSettingsHandler == nil { return false @@ -1865,62 +1896,8 @@ func (r *Router) startPatrolForContext(ctx context.Context, orgID string) bool { patrol := aiService.GetPatrolService() if patrol != nil { if unifiedStore := r.aiSettingsHandler.GetUnifiedStoreForOrg(orgID); unifiedStore != nil { - toUnifiedLifecycle := func(events []ai.FindingLifecycleEvent) []unified.UnifiedFindingLifecycleEvent { - if len(events) == 0 { - return nil - } - out := make([]unified.UnifiedFindingLifecycleEvent, 0, len(events)) - for _, e := range events { - out = append(out, unified.UnifiedFindingLifecycleEvent{ - At: e.At, - Type: e.Type, - Message: e.Message, - From: e.From, - To: e.To, - Metadata: e.Metadata, - }) - } - return out - } patrol.SetUnifiedFindingCallback(func(f *ai.Finding) bool { - // Convert ai.Finding to unified.UnifiedFinding - uf := &unified.UnifiedFinding{ - ID: f.ID, - Source: unified.SourceAIPatrol, - Severity: unified.UnifiedSeverity(f.Severity), - Category: unified.UnifiedCategory(f.Category), - ResourceID: f.ResourceID, - ResourceName: f.ResourceName, - ResourceType: f.ResourceType, - Node: f.Node, - Title: f.Title, - Description: f.Description, - Impact: f.Impact, - PreviousResolvedFixSummary: f.PreviousResolvedFixSummary, - Recommendation: f.Recommendation, - Evidence: f.Evidence, - DetectedAt: f.DetectedAt, - LastSeenAt: f.LastSeenAt, - ResolvedAt: f.ResolvedAt, - AutoResolved: f.AutoResolved, - InvestigationSessionID: f.InvestigationSessionID, - InvestigationStatus: f.InvestigationStatus, - InvestigationOutcome: f.InvestigationOutcome, - LastInvestigatedAt: f.LastInvestigatedAt, - InvestigationAttempts: f.InvestigationAttempts, - InvestigationRecord: f.InvestigationRecord, - LoopState: f.LoopState, - Lifecycle: toUnifiedLifecycle(f.Lifecycle), - RegressionCount: f.RegressionCount, - LastRegressionAt: f.LastRegressionAt, - AcknowledgedAt: f.AcknowledgedAt, - SnoozedUntil: f.SnoozedUntil, - DismissedReason: f.DismissedReason, - UserNote: f.UserNote, - Suppressed: f.Suppressed, - TimesRaised: f.TimesRaised, - RemindAt: f.RemindAt, - } + uf := unifiedFindingFromAI(f) _, isNew := unifiedStore.AddFromAI(uf) // Publish a finding.created event to the agent SSE // stream when the finding is new and not auto-dismissed @@ -2051,43 +2028,7 @@ func (r *Router) startPatrolForContext(ctx context.Context, orgID string) bool { if f == nil { continue } - uf := &unified.UnifiedFinding{ - ID: f.ID, - Source: unified.SourceAIPatrol, - Severity: unified.UnifiedSeverity(f.Severity), - Category: unified.UnifiedCategory(f.Category), - ResourceID: f.ResourceID, - ResourceName: f.ResourceName, - ResourceType: f.ResourceType, - Node: f.Node, - Title: f.Title, - Description: f.Description, - Impact: f.Impact, - PreviousResolvedFixSummary: f.PreviousResolvedFixSummary, - Recommendation: f.Recommendation, - Evidence: f.Evidence, - DetectedAt: f.DetectedAt, - LastSeenAt: f.LastSeenAt, - ResolvedAt: f.ResolvedAt, - AutoResolved: f.AutoResolved, - InvestigationSessionID: f.InvestigationSessionID, - InvestigationStatus: f.InvestigationStatus, - InvestigationOutcome: f.InvestigationOutcome, - LastInvestigatedAt: f.LastInvestigatedAt, - InvestigationAttempts: f.InvestigationAttempts, - InvestigationRecord: f.InvestigationRecord, - LoopState: f.LoopState, - Lifecycle: toUnifiedLifecycle(f.Lifecycle), - RegressionCount: f.RegressionCount, - LastRegressionAt: f.LastRegressionAt, - AcknowledgedAt: f.AcknowledgedAt, - SnoozedUntil: f.SnoozedUntil, - DismissedReason: f.DismissedReason, - UserNote: f.UserNote, - Suppressed: f.Suppressed, - TimesRaised: f.TimesRaised, - RemindAt: f.RemindAt, - } + uf := unifiedFindingFromAI(f) // Copy resolution timestamp if resolved if f.ResolvedAt != nil || f.AutoResolved { now := time.Now() @@ -5471,101 +5412,10 @@ func (r *Router) handleCharts(w http.ResponseWriter, req *http.Request) { currentTime := time.Now().UnixMilli() // JavaScript timestamp format oldestTimestamp := currentTime - // Process VMs - batch-load historical data (1-2 SQL calls instead of N). - vmList := readState.VMs() - vmRequests := make([]monitoring.GuestChartRequest, 0, len(vmList)) - for _, vm := range vmList { - if vm == nil { - continue - } - if vid := vm.SourceID(); vid != "" { - vmRequests = append(vmRequests, monitoring.GuestChartRequest{InMemoryKey: vid, SQLResourceID: vid}) - } - } - vmBatchMetrics := monitor.GetGuestMetricsForChartBatch("vm", vmRequests, duration, infrastructureSummaryMetricOrder...) - for _, vm := range vmList { - if vm == nil { - continue - } - vid := vm.SourceID() - if vid == "" { - continue - } - chartData[vid] = make(VMChartData) - if batchMetrics, ok := vmBatchMetrics[vid]; ok { - for metricType, points := range batchMetrics { - if !sparklineMetrics[metricType] { - continue - } - chartData[vid][metricType] = make([]MetricPoint, len(points)) - for i, point := range points { - ts := point.Timestamp.UnixMilli() - if ts < oldestTimestamp { - oldestTimestamp = ts - } - chartData[vid][metricType][i] = MetricPoint{ - Timestamp: ts, - Value: point.Value, - } - } - } - } - if len(chartData[vid]["cpu"]) == 0 { - chartData[vid]["cpu"] = []MetricPoint{{Timestamp: currentTime, Value: vm.CPUPercent()}} - chartData[vid]["memory"] = []MetricPoint{{Timestamp: currentTime, Value: vm.MemoryPercent()}} - chartData[vid]["disk"] = []MetricPoint{{Timestamp: currentTime, Value: vm.DiskPercent()}} - chartData[vid]["netin"] = []MetricPoint{{Timestamp: currentTime, Value: vm.NetIn()}} - chartData[vid]["netout"] = []MetricPoint{{Timestamp: currentTime, Value: vm.NetOut()}} - } - } - - // Process Containers - batch-load historical data (1-2 SQL calls instead of N). - ctList := readState.Containers() - ctRequests := make([]monitoring.GuestChartRequest, 0, len(ctList)) - for _, ct := range ctList { - if ct == nil { - continue - } - if cid := ct.SourceID(); cid != "" { - ctRequests = append(ctRequests, monitoring.GuestChartRequest{InMemoryKey: cid, SQLResourceID: cid}) - } - } - ctBatchMetrics := monitor.GetGuestMetricsForChartBatch("container", ctRequests, duration, infrastructureSummaryMetricOrder...) - for _, ct := range ctList { - if ct == nil { - continue - } - cid := ct.SourceID() - if cid == "" { - continue - } - chartData[cid] = make(VMChartData) - if batchMetrics, ok := ctBatchMetrics[cid]; ok { - for metricType, points := range batchMetrics { - if !sparklineMetrics[metricType] { - continue - } - chartData[cid][metricType] = make([]MetricPoint, len(points)) - for i, point := range points { - ts := point.Timestamp.UnixMilli() - if ts < oldestTimestamp { - oldestTimestamp = ts - } - chartData[cid][metricType][i] = MetricPoint{ - Timestamp: ts, - Value: point.Value, - } - } - } - } - if len(chartData[cid]["cpu"]) == 0 { - chartData[cid]["cpu"] = []MetricPoint{{Timestamp: currentTime, Value: ct.CPUPercent()}} - chartData[cid]["memory"] = []MetricPoint{{Timestamp: currentTime, Value: ct.MemoryPercent()}} - chartData[cid]["disk"] = []MetricPoint{{Timestamp: currentTime, Value: ct.DiskPercent()}} - chartData[cid]["netin"] = []MetricPoint{{Timestamp: currentTime, Value: ct.NetIn()}} - chartData[cid]["netout"] = []MetricPoint{{Timestamp: currentTime, Value: ct.NetOut()}} - } - } + // Process VMs and Containers - batch-load historical data (1-2 SQL calls + // per family instead of N). + oldestTimestamp = collectGuestChartData(monitor, "vm", readState.VMs(), duration, chartData, currentTime, oldestTimestamp) + oldestTimestamp = collectGuestChartData(monitor, "container", readState.Containers(), duration, chartData, currentTime, oldestTimestamp) // Process Storage - batch-load historical data (1-2 SQL calls instead of N). storageData := make(map[string]StorageChartData) @@ -5721,22 +5571,7 @@ func (r *Router) handleCharts(w http.ResponseWriter, req *http.Request) { } dockerData[responseKey] = make(VMChartData) if batchMetrics, ok := dcBatchMetrics[request.SQLResourceID]; ok { - for metricType, points := range batchMetrics { - if !sparklineMetrics[metricType] { - continue - } - dockerData[responseKey][metricType] = make([]MetricPoint, len(points)) - for i, point := range points { - ts := point.Timestamp.UnixMilli() - if ts < oldestTimestamp { - oldestTimestamp = ts - } - dockerData[responseKey][metricType][i] = MetricPoint{ - Timestamp: ts, - Value: point.Value, - } - } - } + oldestTimestamp = fillChartSeriesFromBatch(dockerData[responseKey], batchMetrics, oldestTimestamp) } if len(dockerData[responseKey]["cpu"]) == 0 { dockerData[responseKey]["cpu"] = []MetricPoint{{Timestamp: currentTime, Value: dc.CPUPercent()}} @@ -5771,22 +5606,7 @@ func (r *Router) handleCharts(w http.ResponseWriter, req *http.Request) { } dockerHostData[dhID] = make(VMChartData) if batchMetrics, ok := dhBatchMetrics[dhID]; ok { - for metricType, points := range batchMetrics { - if !sparklineMetrics[metricType] { - continue - } - dockerHostData[dhID][metricType] = make([]MetricPoint, len(points)) - for i, point := range points { - ts := point.Timestamp.UnixMilli() - if ts < oldestTimestamp { - oldestTimestamp = ts - } - dockerHostData[dhID][metricType][i] = MetricPoint{ - Timestamp: ts, - Value: point.Value, - } - } - } + oldestTimestamp = fillChartSeriesFromBatch(dockerHostData[dhID], batchMetrics, oldestTimestamp) } if len(dockerHostData[dhID]["cpu"]) == 0 { dockerHostData[dhID]["cpu"] = []MetricPoint{{Timestamp: currentTime, Value: dh.CPUPercent()}} @@ -5818,22 +5638,7 @@ func (r *Router) handleCharts(w http.ResponseWriter, req *http.Request) { } agentData[hID] = make(VMChartData) if batchMetrics, ok := agentBatchMetrics[request.SQLResourceID]; ok { - for metricType, points := range batchMetrics { - if !sparklineMetrics[metricType] { - continue - } - agentData[hID][metricType] = make([]MetricPoint, len(points)) - for i, point := range points { - ts := point.Timestamp.UnixMilli() - if ts < oldestTimestamp { - oldestTimestamp = ts - } - agentData[hID][metricType][i] = MetricPoint{ - Timestamp: ts, - Value: point.Value, - } - } - } + oldestTimestamp = fillChartSeriesFromBatch(agentData[hID], batchMetrics, oldestTimestamp) } if len(agentData[hID]["cpu"]) == 0 { agentData[hID]["cpu"] = []MetricPoint{{Timestamp: currentTime, Value: h.CPUPercent()}} @@ -6449,6 +6254,87 @@ func normalizeInfrastructureSummaryChartSeries( // sparklineMetrics lists the metric types consumed by summary sparklines // and density maps. Metrics not in this set are omitted to keep payloads small. +// guestChartSourceView is the guest view subset the infrastructure summary +// chart builder consumes from VMs and LXC containers. +type guestChartSourceView interface { + comparable + SourceID() string + CPUPercent() float64 + MemoryPercent() float64 + DiskPercent() float64 + NetIn() float64 + NetOut() float64 +} + +// collectGuestChartData batch-loads sparkline history for one proxmox guest +// family into chartData (1-2 SQL calls instead of N) and returns the updated +// oldest chart timestamp. Guests without history fall back to a single +// current-value point per metric. +func collectGuestChartData[V guestChartSourceView]( + monitor *monitoring.Monitor, + storeType string, + guests []V, + duration time.Duration, + chartData map[string]VMChartData, + currentTime, oldestTimestamp int64, +) int64 { + var zero V + requests := make([]monitoring.GuestChartRequest, 0, len(guests)) + for _, g := range guests { + if g == zero { + continue + } + if id := g.SourceID(); id != "" { + requests = append(requests, monitoring.GuestChartRequest{InMemoryKey: id, SQLResourceID: id}) + } + } + batch := monitor.GetGuestMetricsForChartBatch(storeType, requests, duration, infrastructureSummaryMetricOrder...) + for _, g := range guests { + if g == zero { + continue + } + id := g.SourceID() + if id == "" { + continue + } + chartData[id] = make(VMChartData) + if batchMetrics, ok := batch[id]; ok { + oldestTimestamp = fillChartSeriesFromBatch(chartData[id], batchMetrics, oldestTimestamp) + } + if len(chartData[id]["cpu"]) == 0 { + chartData[id]["cpu"] = []MetricPoint{{Timestamp: currentTime, Value: g.CPUPercent()}} + chartData[id]["memory"] = []MetricPoint{{Timestamp: currentTime, Value: g.MemoryPercent()}} + chartData[id]["disk"] = []MetricPoint{{Timestamp: currentTime, Value: g.DiskPercent()}} + chartData[id]["netin"] = []MetricPoint{{Timestamp: currentTime, Value: g.NetIn()}} + chartData[id]["netout"] = []MetricPoint{{Timestamp: currentTime, Value: g.NetOut()}} + } + } + return oldestTimestamp +} + +// fillChartSeriesFromBatch copies sparkline-eligible batch metric points +// into dst and returns the updated oldest chart timestamp. Shared by the +// per-family infrastructure summary chart loops. +func fillChartSeriesFromBatch(dst VMChartData, batchMetrics map[string][]monitoring.MetricPoint, oldestTimestamp int64) int64 { + for metricType, points := range batchMetrics { + if !sparklineMetrics[metricType] { + continue + } + dst[metricType] = make([]MetricPoint, len(points)) + for i, point := range points { + ts := point.Timestamp.UnixMilli() + if ts < oldestTimestamp { + oldestTimestamp = ts + } + dst[metricType][i] = MetricPoint{ + Timestamp: ts, + Value: point.Value, + } + } + } + return oldestTimestamp +} + var sparklineMetrics = map[string]bool{ "cpu": true, "memory": true, @@ -8007,117 +7893,12 @@ func (r *Router) handleWorkloadsSummaryCharts(w http.ResponseWriter, req *http.R }() workloadsSummaryBatchWG.Wait() - for idx, vm := range vmList { - responseKey := vmResponseKeys[idx] - metricID := vmRequests[idx].SQLResourceID - guestCounts.Total++ - if workloadSummaryStatusIsRunning("", vm.Status()) { - guestCounts.Running++ - } else { - guestCounts.Stopped++ - } + var guestSummaryPoints int + snapshots, guestSummaryPoints = appendGuestWorkloadSummaries(vmList, vmResponseKeys, vmRequests, vmBatchMetrics, currentTimeTime, &guestCounts, buckets, snapshots, &oldestTimestamp) + guestPointCount += guestSummaryPoints - snapshot := workloadsSummarySnapshot{ - id: responseKey, - name: strings.TrimSpace(vm.Name()), - cpu: clampWorkloadPercent(vm.CPUPercent()), - memory: clampWorkloadPercent(vm.MemoryPercent()), - disk: clampWorkloadPercent(vm.DiskPercent()), - network: clampNonNegativeWorkloadValue(vm.NetIn() + vm.NetOut()), - } - if snapshot.name == "" { - snapshot.name = responseKey - } - - metrics := vmBatchMetrics[metricID] - cpuPoints := metrics["cpu"] - if len(cpuPoints) == 0 { - cpuPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: vm.CPUPercent()}} - } - memoryPoints := metrics["memory"] - if len(memoryPoints) == 0 { - memoryPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: vm.MemoryPercent()}} - } - diskPoints := metrics["disk"] - if len(diskPoints) == 0 { - diskPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: vm.DiskPercent()}} - } - netInPoints := metrics["netin"] - netOutPoints := metrics["netout"] - if len(netInPoints) == 0 && len(netOutPoints) == 0 { - netInPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: vm.NetIn()}} - netOutPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: vm.NetOut()}} - } - - networkPoints := mergeWorkloadNetworkPoints(netInPoints, netOutPoints) - - snapshot.cpu = latestSummaryMetricValue(cpuPoints, snapshot.cpu, clampWorkloadPercent) - snapshot.memory = latestSummaryMetricValue(memoryPoints, snapshot.memory, clampWorkloadPercent) - snapshot.disk = latestSummaryMetricValue(diskPoints, snapshot.disk, clampWorkloadPercent) - snapshot.network = latestSummaryMetricValue(networkPoints, snapshot.network, clampNonNegativeWorkloadValue) - - guestPointCount += appendWorkloadMetricPoints(buckets, cpuPoints, "cpu", &oldestTimestamp) - guestPointCount += appendWorkloadMetricPoints(buckets, memoryPoints, "memory", &oldestTimestamp) - guestPointCount += appendWorkloadMetricPoints(buckets, diskPoints, "disk", &oldestTimestamp) - guestPointCount += appendWorkloadMetricPoints(buckets, networkPoints, "network", &oldestTimestamp) - snapshots = append(snapshots, snapshot) - } - - for idx, ct := range containerList { - responseKey := containerResponseKeys[idx] - metricID := containerRequests[idx].SQLResourceID - guestCounts.Total++ - if workloadSummaryStatusIsRunning("", ct.Status()) { - guestCounts.Running++ - } else { - guestCounts.Stopped++ - } - - snapshot := workloadsSummarySnapshot{ - id: responseKey, - name: strings.TrimSpace(ct.Name()), - cpu: clampWorkloadPercent(ct.CPUPercent()), - memory: clampWorkloadPercent(ct.MemoryPercent()), - disk: clampWorkloadPercent(ct.DiskPercent()), - network: clampNonNegativeWorkloadValue(ct.NetIn() + ct.NetOut()), - } - if snapshot.name == "" { - snapshot.name = responseKey - } - - metrics := containerBatchMetrics[metricID] - cpuPoints := metrics["cpu"] - if len(cpuPoints) == 0 { - cpuPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: ct.CPUPercent()}} - } - memoryPoints := metrics["memory"] - if len(memoryPoints) == 0 { - memoryPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: ct.MemoryPercent()}} - } - diskPoints := metrics["disk"] - if len(diskPoints) == 0 { - diskPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: ct.DiskPercent()}} - } - netInPoints := metrics["netin"] - netOutPoints := metrics["netout"] - if len(netInPoints) == 0 && len(netOutPoints) == 0 { - netInPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: ct.NetIn()}} - netOutPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: ct.NetOut()}} - } - - networkPoints := mergeWorkloadNetworkPoints(netInPoints, netOutPoints) - - snapshot.cpu = latestSummaryMetricValue(cpuPoints, snapshot.cpu, clampWorkloadPercent) - snapshot.memory = latestSummaryMetricValue(memoryPoints, snapshot.memory, clampWorkloadPercent) - snapshot.disk = latestSummaryMetricValue(diskPoints, snapshot.disk, clampWorkloadPercent) - snapshot.network = latestSummaryMetricValue(networkPoints, snapshot.network, clampNonNegativeWorkloadValue) - - guestPointCount += appendWorkloadMetricPoints(buckets, cpuPoints, "cpu", &oldestTimestamp) - guestPointCount += appendWorkloadMetricPoints(buckets, memoryPoints, "memory", &oldestTimestamp) - guestPointCount += appendWorkloadMetricPoints(buckets, diskPoints, "disk", &oldestTimestamp) - guestPointCount += appendWorkloadMetricPoints(buckets, networkPoints, "network", &oldestTimestamp) - snapshots = append(snapshots, snapshot) - } + snapshots, guestSummaryPoints = appendGuestWorkloadSummaries(containerList, containerResponseKeys, containerRequests, containerBatchMetrics, currentTimeTime, &guestCounts, buckets, snapshots, &oldestTimestamp) + guestPointCount += guestSummaryPoints for _, pod := range podList { metricKey := kubernetesPodMetricIDFromView(pod) @@ -8350,6 +8131,91 @@ func (r *Router) handleWorkloadsSummaryCharts(w http.ResponseWriter, req *http.R } } +// guestWorkloadSummaryView is the guest view subset the workloads summary +// loop consumes from VMs and LXC containers. +type guestWorkloadSummaryView interface { + Status() unifiedresources.ResourceStatus + Name() string + CPUPercent() float64 + MemoryPercent() float64 + DiskPercent() float64 + NetIn() float64 + NetOut() float64 +} + +// appendGuestWorkloadSummaries accumulates workload-summary snapshots and +// chart points for one proxmox guest family (VMs or LXC containers), +// returning the extended snapshot slice and the number of points added. +func appendGuestWorkloadSummaries[V guestWorkloadSummaryView]( + guests []V, + responseKeys []string, + requests []monitoring.GuestChartRequest, + batchMetrics map[string]map[string][]monitoring.MetricPoint, + currentTimeTime time.Time, + guestCounts *WorkloadsGuestCounts, + buckets map[int64]*workloadSummaryBuckets, + snapshots []workloadsSummarySnapshot, + oldestTimestamp *int64, +) ([]workloadsSummarySnapshot, int) { + added := 0 + for idx, g := range guests { + responseKey := responseKeys[idx] + metricID := requests[idx].SQLResourceID + guestCounts.Total++ + if workloadSummaryStatusIsRunning("", g.Status()) { + guestCounts.Running++ + } else { + guestCounts.Stopped++ + } + + snapshot := workloadsSummarySnapshot{ + id: responseKey, + name: strings.TrimSpace(g.Name()), + cpu: clampWorkloadPercent(g.CPUPercent()), + memory: clampWorkloadPercent(g.MemoryPercent()), + disk: clampWorkloadPercent(g.DiskPercent()), + network: clampNonNegativeWorkloadValue(g.NetIn() + g.NetOut()), + } + if snapshot.name == "" { + snapshot.name = responseKey + } + + metrics := batchMetrics[metricID] + cpuPoints := metrics["cpu"] + if len(cpuPoints) == 0 { + cpuPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: g.CPUPercent()}} + } + memoryPoints := metrics["memory"] + if len(memoryPoints) == 0 { + memoryPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: g.MemoryPercent()}} + } + diskPoints := metrics["disk"] + if len(diskPoints) == 0 { + diskPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: g.DiskPercent()}} + } + netInPoints := metrics["netin"] + netOutPoints := metrics["netout"] + if len(netInPoints) == 0 && len(netOutPoints) == 0 { + netInPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: g.NetIn()}} + netOutPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: g.NetOut()}} + } + + networkPoints := mergeWorkloadNetworkPoints(netInPoints, netOutPoints) + + snapshot.cpu = latestSummaryMetricValue(cpuPoints, snapshot.cpu, clampWorkloadPercent) + snapshot.memory = latestSummaryMetricValue(memoryPoints, snapshot.memory, clampWorkloadPercent) + snapshot.disk = latestSummaryMetricValue(diskPoints, snapshot.disk, clampWorkloadPercent) + snapshot.network = latestSummaryMetricValue(networkPoints, snapshot.network, clampNonNegativeWorkloadValue) + + added += appendWorkloadMetricPoints(buckets, cpuPoints, "cpu", oldestTimestamp) + added += appendWorkloadMetricPoints(buckets, memoryPoints, "memory", oldestTimestamp) + added += appendWorkloadMetricPoints(buckets, diskPoints, "disk", oldestTimestamp) + added += appendWorkloadMetricPoints(buckets, networkPoints, "network", oldestTimestamp) + snapshots = append(snapshots, snapshot) + } + return snapshots, added +} + func workloadSummaryStatusIsRunning(runtimeState string, status unifiedresources.ResourceStatus) bool { switch strings.ToLower(strings.TrimSpace(runtimeState)) { case "running", "online", "ok": diff --git a/internal/api/router_routes_registration.go b/internal/api/router_routes_registration.go index 373d1e34d..f3614fdbc 100644 --- a/internal/api/router_routes_registration.go +++ b/internal/api/router_routes_registration.go @@ -289,24 +289,17 @@ func (r *Router) registerConfigSystemRoutes(updateHandlers *UpdateHandlers) { } }) - r.mux.HandleFunc("/api/truenas/connections/", func(w http.ResponseWriter, req *http.Request) { + r.mux.HandleFunc("/api/truenas/connections/", r.platformConnectionItemRoute("truenas_unavailable", "TrueNAS service unavailable", func() *platformConnectionItemHandlers { if r.trueNASHandlers == nil { - writeErrorResponse(w, http.StatusServiceUnavailable, "truenas_unavailable", "TrueNAS service unavailable", nil) - return + return nil } - - if req.Method == http.MethodPost && strings.HasSuffix(strings.Trim(req.URL.Path, "/"), "/test") { - RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, r.trueNASHandlers.HandleTestSavedConnection))(w, req) - } else if req.Method == http.MethodPost && strings.HasSuffix(strings.Trim(req.URL.Path, "/"), "/preview") { - RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, r.trueNASHandlers.HandlePreviewSavedConnection))(w, req) - } else if req.Method == http.MethodDelete { - RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, r.trueNASHandlers.HandleDelete))(w, req) - } else if req.Method == http.MethodPut { - RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, r.trueNASHandlers.HandleUpdate))(w, req) - } else { - http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return &platformConnectionItemHandlers{ + test: r.trueNASHandlers.HandleTestSavedConnection, + preview: r.trueNASHandlers.HandlePreviewSavedConnection, + delete: r.trueNASHandlers.HandleDelete, + update: r.trueNASHandlers.HandleUpdate, } - }) + })) // VMware vCenter connection management r.mux.HandleFunc("/api/vmware/connections", func(w http.ResponseWriter, req *http.Request) { @@ -351,24 +344,17 @@ func (r *Router) registerConfigSystemRoutes(updateHandlers *UpdateHandlers) { } }) - r.mux.HandleFunc("/api/vmware/connections/", func(w http.ResponseWriter, req *http.Request) { + r.mux.HandleFunc("/api/vmware/connections/", r.platformConnectionItemRoute("vmware_unavailable", "VMware service unavailable", func() *platformConnectionItemHandlers { if r.vmwareHandlers == nil { - writeErrorResponse(w, http.StatusServiceUnavailable, "vmware_unavailable", "VMware service unavailable", nil) - return + return nil } - - if req.Method == http.MethodPost && strings.HasSuffix(strings.Trim(req.URL.Path, "/"), "/test") { - RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, r.vmwareHandlers.HandleTestSavedConnection))(w, req) - } else if req.Method == http.MethodPost && strings.HasSuffix(strings.Trim(req.URL.Path, "/"), "/preview") { - RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, r.vmwareHandlers.HandlePreviewSavedConnection))(w, req) - } else if req.Method == http.MethodDelete { - RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, r.vmwareHandlers.HandleDelete))(w, req) - } else if req.Method == http.MethodPut { - RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, r.vmwareHandlers.HandleUpdate))(w, req) - } else { - http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return &platformConnectionItemHandlers{ + test: r.vmwareHandlers.HandleTestSavedConnection, + preview: r.vmwareHandlers.HandlePreviewSavedConnection, + delete: r.vmwareHandlers.HandleDelete, + update: r.vmwareHandlers.HandleUpdate, } - }) + })) // Config Profile Routes - Protected by Admin Auth, Settings Scope, and Pro License // SECURITY: Require settings:write scope to prevent low-privilege tokens from modifying agent profiles diff --git a/internal/api/truenas_handlers.go b/internal/api/truenas_handlers.go index 4e3761c90..a812014f0 100644 --- a/internal/api/truenas_handlers.go +++ b/internal/api/truenas_handlers.go @@ -1,10 +1,8 @@ package api import ( - "bytes" "context" "encoding/json" - "io" "net/http" "net/url" "strings" @@ -188,6 +186,8 @@ func (h *TrueNASHandlers) HandleDelete(w http.ResponseWriter, r *http.Request) { // HandleUpdate replaces a configured TrueNAS connection by ID while preserving // unchanged masked secrets from the stored record. +// +//nolint:dupl // parallel platform wiring of the shared updatePlatformConnection flow; only platform symbols differ func (h *TrueNASHandlers) HandleUpdate(w http.ResponseWriter, r *http.Request) { if !h.featureEnabled(w) { return @@ -214,36 +214,21 @@ func (h *TrueNASHandlers) HandleUpdate(w http.ResponseWriter, r *http.Request) { return } - index := -1 - for i := range instances { - if strings.TrimSpace(instances[i].ID) == connectionID { - index = i - break - } - } - if index < 0 { - writeErrorResponse(w, http.StatusNotFound, "truenas_not_found", "Connection not found", nil) - return - } - - instance, ok := decodeTrueNASInstanceRequest(w, r, instances[index]) - if !ok { - return - } - instance.ID = connectionID - normalizeTrueNASInstance(&instance) - instance.PreserveMaskedSecrets(instances[index]) - if err := instance.Validate(); err != nil { - writeErrorResponse(w, http.StatusBadRequest, "validation_error", err.Error(), nil) - return - } - instances[index] = instance - if err := persistence.SaveTrueNASConfig(instances); err != nil { - writeErrorResponse(w, http.StatusInternalServerError, "truenas_save_failed", "Failed to save TrueNAS configuration", map[string]string{"error": err.Error()}) - return - } - - writeJSON(w, http.StatusOK, instance.Redacted()) + updatePlatformConnection(w, r, connectionID, instances, platformConnectionUpdateSpec[config.TrueNASInstance]{ + notFoundCode: "truenas_not_found", + saveFailCode: "truenas_save_failed", + saveFailMsg: "Failed to save TrueNAS configuration", + id: func(inst config.TrueNASInstance) string { return inst.ID }, + decode: decodeTrueNASInstanceRequest, + setID: func(inst *config.TrueNASInstance, id string) { inst.ID = id }, + normalize: normalizeTrueNASInstance, + preserve: func(updated *config.TrueNASInstance, stored config.TrueNASInstance) { + updated.PreserveMaskedSecrets(stored) + }, + validate: func(inst config.TrueNASInstance) error { return inst.Validate() }, + redacted: func(inst config.TrueNASInstance) any { return inst.Redacted() }, + save: persistence.SaveTrueNASConfig, + }) } // HandlePreviewConnection projects the monitored-system impact of a proposed @@ -535,25 +520,7 @@ func decodeOptionalTrueNASInstanceRequest( r *http.Request, base config.TrueNASInstance, ) (config.TrueNASInstance, bool, bool) { - r.Body = http.MaxBytesReader(w, r.Body, 32*1024) - defer r.Body.Close() - - body, err := io.ReadAll(r.Body) - if err != nil { - writeErrorResponse(w, http.StatusBadRequest, "invalid_request", "Invalid request body", map[string]string{"error": err.Error()}) - return config.TrueNASInstance{}, false, false - } - if len(bytes.TrimSpace(body)) == 0 { - return base, false, true - } - - instance := base - if err := json.Unmarshal(body, &instance); err != nil { - writeErrorResponse(w, http.StatusBadRequest, "invalid_request", "Invalid request body", map[string]string{"error": err.Error()}) - return config.TrueNASInstance{}, false, false - } - - return instance, true, true + return decodeOptionalInstanceRequest(w, r, base) } func (h *TrueNASHandlers) featureEnabled(w http.ResponseWriter) bool { diff --git a/internal/api/vmware_handlers.go b/internal/api/vmware_handlers.go index e87629370..6b9da398d 100644 --- a/internal/api/vmware_handlers.go +++ b/internal/api/vmware_handlers.go @@ -1,10 +1,8 @@ package api import ( - "bytes" "context" "encoding/json" - "io" "net/http" "net/url" "strings" @@ -202,6 +200,8 @@ func (h *VMwareHandlers) HandleDelete(w http.ResponseWriter, r *http.Request) { // HandleUpdate replaces a configured VMware vCenter connection by ID while // preserving unchanged masked secrets from the stored record. +// +//nolint:dupl // parallel platform wiring of the shared updatePlatformConnection flow; only platform symbols differ func (h *VMwareHandlers) HandleUpdate(w http.ResponseWriter, r *http.Request) { if !h.featureEnabled(w) { return @@ -228,36 +228,21 @@ func (h *VMwareHandlers) HandleUpdate(w http.ResponseWriter, r *http.Request) { return } - index := -1 - for i := range instances { - if strings.TrimSpace(instances[i].ID) == connectionID { - index = i - break - } - } - if index < 0 { - writeErrorResponse(w, http.StatusNotFound, "vmware_not_found", "Connection not found", nil) - return - } - - instance, ok := decodeVMwareInstanceRequest(w, r, instances[index]) - if !ok { - return - } - instance.ID = connectionID - normalizeVMwareInstance(&instance) - instance.PreserveMaskedSecrets(instances[index]) - if err := instance.Validate(); err != nil { - writeErrorResponse(w, http.StatusBadRequest, "validation_error", err.Error(), nil) - return - } - instances[index] = instance - if err := persistence.SaveVMwareConfig(instances); err != nil { - writeErrorResponse(w, http.StatusInternalServerError, "vmware_save_failed", "Failed to save VMware configuration", map[string]string{"error": err.Error()}) - return - } - - writeJSON(w, http.StatusOK, instance.Redacted()) + updatePlatformConnection(w, r, connectionID, instances, platformConnectionUpdateSpec[config.VMwareVCenterInstance]{ + notFoundCode: "vmware_not_found", + saveFailCode: "vmware_save_failed", + saveFailMsg: "Failed to save VMware configuration", + id: func(inst config.VMwareVCenterInstance) string { return inst.ID }, + decode: decodeVMwareInstanceRequest, + setID: func(inst *config.VMwareVCenterInstance, id string) { inst.ID = id }, + normalize: normalizeVMwareInstance, + preserve: func(updated *config.VMwareVCenterInstance, stored config.VMwareVCenterInstance) { + updated.PreserveMaskedSecrets(stored) + }, + validate: func(inst config.VMwareVCenterInstance) error { return inst.Validate() }, + redacted: func(inst config.VMwareVCenterInstance) any { return inst.Redacted() }, + save: persistence.SaveVMwareConfig, + }) } // HandlePreviewConnection projects the monitored-system impact of a proposed @@ -606,25 +591,7 @@ func decodeOptionalVMwareInstanceRequest( r *http.Request, base config.VMwareVCenterInstance, ) (config.VMwareVCenterInstance, bool, bool) { - r.Body = http.MaxBytesReader(w, r.Body, 32*1024) - defer r.Body.Close() - - body, err := io.ReadAll(r.Body) - if err != nil { - writeErrorResponse(w, http.StatusBadRequest, "invalid_request", "Invalid request body", map[string]string{"error": err.Error()}) - return config.VMwareVCenterInstance{}, false, false - } - if len(bytes.TrimSpace(body)) == 0 { - return base, false, true - } - - instance := base - if err := json.Unmarshal(body, &instance); err != nil { - writeErrorResponse(w, http.StatusBadRequest, "invalid_request", "Invalid request body", map[string]string{"error": err.Error()}) - return config.VMwareVCenterInstance{}, false, false - } - - return instance, true, true + return decodeOptionalInstanceRequest(w, r, base) } func (h *VMwareHandlers) featureEnabled(w http.ResponseWriter) bool { diff --git a/pkg/pulsecli/actions.go b/pkg/pulsecli/actions.go index b18351e8e..f357b59d3 100644 --- a/pkg/pulsecli/actions.go +++ b/pkg/pulsecli/actions.go @@ -109,18 +109,26 @@ type actionExecutionResponse struct { Audit unified.ActionAuditRecord `json:"audit"` } +// actionAPIDefaults seeds the shared Pulse API URL and token defaults for +// action subcommands from the environment. +func actionAPIDefaults(deps *ActionsDeps) (apiURL, token string) { + apiURL = strings.TrimSpace(actionGetenv(deps, "PULSE_API_URL")) + if apiURL == "" { + apiURL = defaultPulseAPIURL + } + return apiURL, strings.TrimSpace(actionGetenv(deps, "PULSE_API_TOKEN")) +} + func newActionsCmd(deps *ActionsDeps) *cobra.Command { actionsCmd := &cobra.Command{ Use: "actions", Short: "Inspect and plan governed Pulse actions", } + apiURL, token := actionAPIDefaults(deps) opts := actionPlanOptions{ - APIURL: strings.TrimSpace(actionGetenv(deps, "PULSE_API_URL")), - Token: strings.TrimSpace(actionGetenv(deps, "PULSE_API_TOKEN")), - } - if opts.APIURL == "" { - opts.APIURL = defaultPulseAPIURL + APIURL: apiURL, + Token: token, } planCmd := &cobra.Command{ @@ -152,12 +160,10 @@ func newActionsCmd(deps *ActionsDeps) *cobra.Command { } func newActionCapabilitiesCmd(deps *ActionsDeps) *cobra.Command { + apiURL, token := actionAPIDefaults(deps) opts := actionCapabilitiesOptions{ - APIURL: strings.TrimSpace(actionGetenv(deps, "PULSE_API_URL")), - Token: strings.TrimSpace(actionGetenv(deps, "PULSE_API_TOKEN")), - } - if opts.APIURL == "" { - opts.APIURL = defaultPulseAPIURL + APIURL: apiURL, + Token: token, } cmd := &cobra.Command{ @@ -174,12 +180,10 @@ func newActionCapabilitiesCmd(deps *ActionsDeps) *cobra.Command { } func newActionDecisionCmd(deps *ActionsDeps) *cobra.Command { + apiURL, token := actionAPIDefaults(deps) opts := actionDecisionOptions{ - APIURL: strings.TrimSpace(actionGetenv(deps, "PULSE_API_URL")), - Token: strings.TrimSpace(actionGetenv(deps, "PULSE_API_TOKEN")), - } - if opts.APIURL == "" { - opts.APIURL = defaultPulseAPIURL + APIURL: apiURL, + Token: token, } cmd := &cobra.Command{ @@ -198,12 +202,10 @@ func newActionDecisionCmd(deps *ActionsDeps) *cobra.Command { } func newActionExecutionCmd(deps *ActionsDeps) *cobra.Command { + apiURL, token := actionAPIDefaults(deps) opts := actionExecutionOptions{ - APIURL: strings.TrimSpace(actionGetenv(deps, "PULSE_API_URL")), - Token: strings.TrimSpace(actionGetenv(deps, "PULSE_API_TOKEN")), - } - if opts.APIURL == "" { - opts.APIURL = defaultPulseAPIURL + APIURL: apiURL, + Token: token, } cmd := &cobra.Command{ @@ -220,15 +222,14 @@ func newActionExecutionCmd(deps *ActionsDeps) *cobra.Command { return cmd } +//nolint:dupl // parallel cobra subcommand registration; flags/help diverge per command func newActionAuditCmd(deps *ActionsDeps) *cobra.Command { + apiURL, token := actionAPIDefaults(deps) opts := actionAuditOptions{ - APIURL: strings.TrimSpace(actionGetenv(deps, "PULSE_API_URL")), - Token: strings.TrimSpace(actionGetenv(deps, "PULSE_API_TOKEN")), + APIURL: apiURL, + Token: token, Limit: 100, } - if opts.APIURL == "" { - opts.APIURL = defaultPulseAPIURL - } cmd := &cobra.Command{ Use: "audit", @@ -245,15 +246,14 @@ func newActionAuditCmd(deps *ActionsDeps) *cobra.Command { return cmd } +//nolint:dupl // parallel cobra subcommand registration; flags/help diverge per command func newActionEventsCmd(deps *ActionsDeps) *cobra.Command { + apiURL, token := actionAPIDefaults(deps) opts := actionEventsOptions{ - APIURL: strings.TrimSpace(actionGetenv(deps, "PULSE_API_URL")), - Token: strings.TrimSpace(actionGetenv(deps, "PULSE_API_TOKEN")), + APIURL: apiURL, + Token: token, Limit: 100, } - if opts.APIURL == "" { - opts.APIURL = defaultPulseAPIURL - } cmd := &cobra.Command{ Use: "events", diff --git a/pkg/pulsecli/actions_test.go b/pkg/pulsecli/actions_test.go index abbf96f4c..ca9dc1c38 100644 --- a/pkg/pulsecli/actions_test.go +++ b/pkg/pulsecli/actions_test.go @@ -651,3 +651,31 @@ func newTestActionsRootCommand(env map[string]string) *cobra.Command { }, ) } + +func TestActionAPIDefaultsSeedsFromEnvironment(t *testing.T) { + deps := &ActionsDeps{Getenv: func(key string) string { + switch key { + case "PULSE_API_URL": + return " https://pulse.example.test " + case "PULSE_API_TOKEN": + return " token-123 " + } + return "" + }} + apiURL, token := actionAPIDefaults(deps) + if apiURL != "https://pulse.example.test" { + t.Fatalf("apiURL = %q, want trimmed env value", apiURL) + } + if token != "token-123" { + t.Fatalf("token = %q, want trimmed env value", token) + } + + empty := &ActionsDeps{Getenv: func(string) string { return "" }} + apiURL, token = actionAPIDefaults(empty) + if apiURL != defaultPulseAPIURL { + t.Fatalf("apiURL = %q, want default %q when env is unset", apiURL, defaultPulseAPIURL) + } + if token != "" { + t.Fatalf("token = %q, want empty when env is unset", token) + } +} diff --git a/scripts/release_control/subsystem_lookup_test.py b/scripts/release_control/subsystem_lookup_test.py index 6e7571232..bb950b7e7 100644 --- a/scripts/release_control/subsystem_lookup_test.py +++ b/scripts/release_control/subsystem_lookup_test.py @@ -2860,8 +2860,8 @@ class SubsystemLookupTest(unittest.TestCase): { "heading": "## Shared Boundaries", "path": "internal/api/access_control_handlers.go", - "line": 450, - "heading_line": 117, + "line": 452, + "heading_line": 119, } ], )