Install cluster member agents in one step from a covered sibling

Adding the Pulse Agent to a Proxmox cluster member meant generating a
token in the UI and pasting it into an interactive prompt on the host,
even though Pulse already knows the cluster and has an agent on its
siblings. The server-side cluster deploy that avoids this was still
shipped, but its wizard was deleted as an orphan in May, and it could
not reach members discovered by hostname: targets were parsed from the
API URL, so a name like "delly2" hit the agent's literal-IP guard and
failed preflight.

Deploy targets now resolve to a literal IP in the member's API URL or
its cluster endpoint's effective IP, matched on the member's own
connection, and members with no known address come back undeployable
with a reason. The Settings > Infrastructure install actions open one
dialog that checks, installs, and confirms each node is reporting,
falling back to the scoped installer when no sibling agent is
connected or a node cannot be reached.

The deploy routes are keyed on the cluster name, so they now refuse a
name shared by two connections instead of merging two sites, and job
creation admits one install per cluster at a time under a lock, which
lets the dialog keep a run going after it is closed.

The agent-lifecycle, api-contracts, frontend-primitives, and
storage-recovery contracts record the new install path, its address
rule, and the persisted, time-bounded admission check.
This commit is contained in:
rcourtman
2026-10-05 11:43:02 +01:00
parent 3af37a8862
commit 76b6f7ea97
19 changed files with 2008 additions and 31 deletions
@@ -6400,25 +6400,41 @@ frontend-local merge of raw unified-resource facets and removed runtime arrays,
and v6 clients no longer treat those removed runtime arrays as a parallel
settings contract, so lifecycle scope and reconnect behavior stay canonical
across host, Docker, and Kubernetes reporting.
deploy results surface: `ResultsStep` must request the canonical backend
install command from `/api/agent-install-command` for failed deploy targets
Cluster members install in one step. On a PVE cluster where at least one
member already runs a connected Pulse Agent, the Settings → Infrastructure
cluster, member, and coverage-summary install actions open
`ClusterAgentDeployDialog`, which drives the server-side cluster deploy API
(`/api/clusters/{cluster}/agent-deploy` candidates, preflights, and jobs). The
server mints a per-node bootstrap credential for every target, so the dialog
never renders, copies, or reuses a token and the operator never pastes one.
The dialog folds target statuses into checking, installing, connecting,
reporting, and failed, and it reports a node as installed only after the
candidates projection shows that node with an agent. A finished deploy job
alone is not success, because targets can still be enrolling. Clusters with no
connected agent, members Pulse has no address for, and failed targets hand off
to the scoped manual installer (`/settings/infrastructure?add=linux-host`)
instead of rebuilding a local shell snippet that can drift from the governed
installer contract. That fallback surface must consume the shared validated
`NodesAPI.getAgentInstallCommand` response, so malformed backend payloads fail
closed and the raw backend install token stays inside the shared client
boundary rather than leaking into deploy UI state.
Deploy wizard target tables are lifecycle-owned presentation surfaces:
`CandidatesStep`, `ConfirmStep`, `PreflightStep`, `DeployingStep`, and
`ResultsStep` must use the shared frontend `Table` primitive for scroll and
table semantics instead of raw table markup or step-local scroll frames.
Deploy selection and retry UI must not consume retired monitored-system
capacity boundaries. Lifecycle UI must avoid workspace-capacity, legacy
license-slot, and plan-upgrade language in deploy confirmation, preflight, or
retry surfaces.
That same deploy wizard boundary must also stay on the direct
`deploy-fallback-install-surface` proof path, rather than relying only on the
shared install helper or downstream deploy tests to catch lifecycle drift in
the infrastructure fallback surface.
installer contract. Deploy targets must resolve to a literal IP through
`DeployHandlers.deployTargetIP`: a literal IP in the member's API URL first,
then the cluster endpoint's `EffectiveIP()` (the operator's connection-address
override, else the address Proxmox reports for the member). Members that Pulse
discovered by hostname must never reach the agent's `validateNodeIP` injection
guard as hostnames, and the candidates response marks a member with no known
address `deployable: false` with reason `no_address` so the UI can explain it
before any install starts. The deploy routes are keyed on the cluster name, so
they must refuse with `ambiguous_cluster` when more than one Proxmox connection
reports that name instead of merging two sites' members, and the address
resolver must match the member's own connection by exact instance name with no
cluster-name fallback. The jobs route must refuse with `deploy_in_progress`
while a recent install job on the same cluster is unfinished, checking and
inserting under one admission lock so concurrent requests cannot both pass, which is what
lets the dialog keep an install running after it is closed without a reopened
dialog or second tab starting an overlapping one. Deploy selection UI must not consume retired
monitored-system capacity boundaries, and lifecycle UI must avoid
workspace-capacity, legacy license-slot, and plan-upgrade language in deploy
surfaces. `frontend-modern/src/components/Settings/__tests__/ClusterAgentDeployDialog.test.tsx`
and `InfrastructureSourceManager.test.tsx` pin the dialog flow and its routing,
and `internal/api/deploy_handlers_test.go` pins the address resolution.
The same Windows install, upgrade, and uninstall copies must also preserve
operator-selected transport and capability toggles: if the settings surface
enables insecure TLS mode or Pulse command execution, the PowerShell path must
@@ -3527,6 +3527,21 @@ a new API state machine, queue contract, or verification-accounting field.
as compatibility input into that same canonical `agent-<hostname>`
binding, and unbound agent-exec tokens must fail closed instead of being
treated as global command authority.
The cluster agent deploy payloads in `internal/api/deploy_handlers.go`
are a typed client contract consumed through
`frontend-modern/src/api/agentDeploy.ts`. Candidate and preflight target
addresses must be literal IPs resolved by
`DeployHandlers.deployTargetIP` (a literal IP in the member's API URL,
else the cluster endpoint's `EffectiveIP()`), never a hostname parsed out
of the API URL, and a member with no known address must come back
`deployable: false` with reason `no_address` rather than failing later in
the source agent's preflight. Candidates and preflights must answer
`409 ambiguous_cluster` when more than one Proxmox connection reports the
cluster name, and job creation must answer `409 deploy_in_progress` while
a recent install job on the cluster is unfinished, admitting atomically so
concurrent requests start at most one install.
`internal/api/deploy_handlers_test.go` and
`frontend-modern/src/api/__tests__/agentDeploy.test.ts` pin that shape.
Proxmox-side LXC Docker detection and inventory wiring in
`internal/api/router.go` shares that agent-exec transport boundary:
router startup may configure the monitoring checker or collector only when
@@ -2608,6 +2608,15 @@ Agent`), with the plain-language source phrase available through accessible
first-run/readiness actions, or the discovery band's explicit review action,
and the direct address-probe utility may appear as first-run setup guidance
instead of a second saved-network-scan command.
That same shared shell boundary owns the cluster-member install action.
`InfrastructureSourceManager.tsx` routes the cluster, member, and
coverage-summary `Install agent` actions for a PVE cluster with a connected
sibling agent into `ClusterAgentDeployDialog.tsx`, mounted once under a
keyed `<Show>` so an install that outlives the dialog never reads a stale
accessor, while every other row keeps the scoped installer hand-off.
`useConnectionsLedger.ts` carries the Proxmox cluster name on cluster rows
as `clusterName`, so the dialog keys the deploy API on the real cluster name
rather than on a display label.
Discovered API-backed candidates stay visible in the same platform-group
table as configured sources, using the existing tree/table hierarchy
instead of spawning a second discovery-only page or card stack.
@@ -25,6 +25,14 @@
"api-contracts"
]
},
{
"path": "frontend-modern/src/api/agentDeploy.ts",
"rationale": "the cluster agent deploy client is both the agent lifecycle one-step cluster member install surface and a canonical API payload contract boundary",
"subsystems": [
"agent-lifecycle",
"api-contracts"
]
},
{
"path": "frontend-modern/src/api/agentProfiles.ts",
"rationale": "the agent profiles frontend client is both an agent lifecycle control surface and a canonical API payload contract boundary",
@@ -1334,10 +1342,12 @@
"cmd/pulse-agent/collector_lifecycle.go",
"cmd/pulse-agent/main.go",
"cmd/pulse-agent/service_windows.go",
"frontend-modern/src/api/agentDeploy.ts",
"frontend-modern/src/api/agentProfiles.ts",
"frontend-modern/src/api/nodes.ts",
"frontend-modern/src/components/Settings/agentProfileSettings.ts",
"frontend-modern/src/components/Settings/AgentProfilesPanel.tsx",
"frontend-modern/src/components/Settings/ClusterAgentDeployDialog.tsx",
"frontend-modern/src/components/Settings/ConnectionEditor/AddressProbeStep.tsx",
"frontend-modern/src/components/Settings/ConnectionEditor/ConnectionEditor.tsx",
"frontend-modern/src/components/Settings/ConnectionEditor/CredentialSlots/NodeCredentialSlot.tsx",
@@ -1819,6 +1829,22 @@
"frontend-modern/src/utils/__tests__/agentProfilesPresentation.test.ts"
]
},
{
"id": "cluster-agent-deploy-surface",
"label": "one-step cluster member agent install proof",
"match_prefixes": [],
"match_files": [
"frontend-modern/src/api/agentDeploy.ts",
"frontend-modern/src/components/Settings/ClusterAgentDeployDialog.tsx"
],
"allow_same_subsystem_tests": false,
"test_prefixes": [],
"exact_files": [
"frontend-modern/src/api/__tests__/agentDeploy.test.ts",
"frontend-modern/src/components/Settings/__tests__/ClusterAgentDeployDialog.test.tsx",
"frontend-modern/src/components/Settings/__tests__/InfrastructureSourceManager.test.tsx"
]
},
{
"id": "frontend-install-command-helper",
"label": "frontend install command helper proof",
@@ -6444,6 +6444,19 @@ successful enrollment response from describing credential state that a restart
would undo. Their success and forced-write-failure proofs live in
`internal/api/deploy_handlers_test.go`.
### Deploy job admission reads persisted state, bounded in time
Install-job admission on a cluster reads the persisted deploy store, not
in-memory state. `HandleCreateJob` refuses a new install while
`Store.UnfinishedJobsForCluster` returns an install job for the same
organization and cluster created within the last 30 minutes, and it checks and
inserts under one admission lock. A job left unfinished by a crash or a
vanished source agent therefore keeps blocking overlapping installs across a
restart, but only until it ages out of that window, so a stale row never locks
a cluster out for good. This adds no backup, restore, retention, snapshot, or
recovery authority. `internal/api/deploy_handlers_test.go` and
`internal/deploy/store_test.go` pin the admission and the query.
### Destination severity policy preserves notification recovery identity
Email, webhook, Apprise, and Relay minimum-severity floors persist through
@@ -0,0 +1,64 @@
import { beforeEach, describe, expect, it, vi } from 'vitest';
import { ClusterAgentDeployAPI, isClusterDeployJobActive } from '../agentDeploy';
import { apiFetchJSON } from '@/utils/apiClient';
vi.mock('@/utils/apiClient', () => ({
apiFetchJSON: vi.fn(),
}));
describe('ClusterAgentDeployAPI', () => {
beforeEach(() => {
vi.mocked(apiFetchJSON).mockReset();
vi.mocked(apiFetchJSON).mockResolvedValue({});
});
it('keys cluster calls on the encoded Proxmox cluster name', async () => {
await ClusterAgentDeployAPI.getCandidates('home lab/1');
expect(apiFetchJSON).toHaveBeenCalledWith(
'/api/clusters/home%20lab%2F1/agent-deploy/candidates',
);
});
it('sends the source agent and targets for a preflight', async () => {
await ClusterAgentDeployAPI.createPreflight('homelab', 'agent-delly', ['n-delly2']);
const [url, options] = vi.mocked(apiFetchJSON).mock.calls[0];
expect(url).toBe('/api/clusters/homelab/agent-deploy/preflights');
expect(options?.method).toBe('POST');
expect(JSON.parse(String(options?.body))).toEqual({
sourceAgentId: 'agent-delly',
targetNodeIds: ['n-delly2'],
maxParallel: 2,
});
});
it('creates a job from a passed preflight', async () => {
await ClusterAgentDeployAPI.createJob('homelab', 'agent-delly', 'pf_1', ['n-delly2']);
const [url, options] = vi.mocked(apiFetchJSON).mock.calls[0];
expect(url).toBe('/api/clusters/homelab/agent-deploy/jobs');
expect(JSON.parse(String(options?.body))).toMatchObject({
sourceAgentId: 'agent-delly',
preflightId: 'pf_1',
targetNodeIds: ['n-delly2'],
});
});
it('reads preflights and jobs from their own status routes', async () => {
await ClusterAgentDeployAPI.getPreflight('pf_1');
await ClusterAgentDeployAPI.getJob('dep_1');
expect(apiFetchJSON).toHaveBeenNthCalledWith(1, '/api/agent-deploy/preflights/pf_1');
expect(apiFetchJSON).toHaveBeenNthCalledWith(2, '/api/agent-deploy/jobs/dep_1');
});
});
describe('isClusterDeployJobActive', () => {
it('treats only in-flight job states as active', () => {
expect(isClusterDeployJobActive('queued')).toBe(true);
expect(isClusterDeployJobActive('waiting_source')).toBe(true);
expect(isClusterDeployJobActive('running')).toBe(true);
expect(isClusterDeployJobActive('canceling')).toBe(true);
expect(isClusterDeployJobActive('succeeded')).toBe(false);
expect(isClusterDeployJobActive('partial_success')).toBe(false);
expect(isClusterDeployJobActive('failed')).toBe(false);
expect(isClusterDeployJobActive('canceled')).toBe(false);
});
});
+126
View File
@@ -0,0 +1,126 @@
import { apiFetchJSON } from '@/utils/apiClient';
/**
* Cluster agent deploy: Pulse installs its agent on uncovered Proxmox cluster
* members by asking a member that already runs the agent to install it on its
* peers over the cluster's own SSH trust. The server mints a per-node bootstrap
* credential for each target, so no token is ever shown to, copied by, or
* reused across operators or hosts.
*/
export type ClusterDeployCandidateReason = 'already_agent' | 'no_address' | (string & {});
export interface ClusterDeployCandidateNode {
nodeId: string;
name: string;
ip?: string;
hasAgent: boolean;
deployable: boolean;
reason?: ClusterDeployCandidateReason;
}
export interface ClusterDeploySourceAgent {
agentId: string;
nodeId: string;
online: boolean;
}
export interface ClusterDeployCandidates {
clusterId: string;
clusterName: string;
sourceAgents: ClusterDeploySourceAgent[] | null;
nodes: ClusterDeployCandidateNode[] | null;
}
export type ClusterDeployJobStatus =
| 'queued'
| 'waiting_source'
| 'running'
| 'succeeded'
| 'partial_success'
| 'failed'
| 'canceling'
| 'canceled';
export type ClusterDeployTargetStatus =
| 'pending'
| 'preflighting'
| 'ready'
| 'installing'
| 'enrolling'
| 'verifying'
| 'succeeded'
| 'failed_retryable'
| 'failed_permanent'
| 'skipped_already_agent'
| 'canceled';
export interface ClusterDeployTarget {
id: string;
nodeId: string;
nodeName: string;
nodeIP: string;
status: ClusterDeployTargetStatus;
errorMessage?: string;
}
export interface ClusterDeployJob {
id: string;
clusterId: string;
status: ClusterDeployJobStatus;
targets?: ClusterDeployTarget[] | null;
}
export interface ClusterDeployPreflightCreated {
preflightId: string;
status: ClusterDeployJobStatus;
}
export interface ClusterDeployJobCreated {
jobId: string;
acceptedTargets: string[] | null;
skippedTargets: { nodeId: string; reason: string }[] | null;
}
const ACTIVE_JOB_STATUSES: ReadonlySet<ClusterDeployJobStatus> = new Set([
'queued',
'waiting_source',
'running',
'canceling',
]);
export const isClusterDeployJobActive = (status: ClusterDeployJobStatus): boolean =>
ACTIVE_JOB_STATUSES.has(status);
const clusterPath = (clusterName: string) =>
`/api/clusters/${encodeURIComponent(clusterName)}/agent-deploy`;
export const ClusterAgentDeployAPI = {
getCandidates: (clusterName: string) =>
apiFetchJSON<ClusterDeployCandidates>(`${clusterPath(clusterName)}/candidates`),
createPreflight: (clusterName: string, sourceAgentId: string, targetNodeIds: string[]) =>
apiFetchJSON<ClusterDeployPreflightCreated>(`${clusterPath(clusterName)}/preflights`, {
method: 'POST',
body: JSON.stringify({ sourceAgentId, targetNodeIds, maxParallel: 2 }),
}),
getPreflight: (preflightId: string) =>
apiFetchJSON<ClusterDeployJob>(
`/api/agent-deploy/preflights/${encodeURIComponent(preflightId)}`,
),
createJob: (
clusterName: string,
sourceAgentId: string,
preflightId: string,
targetNodeIds: string[],
) =>
apiFetchJSON<ClusterDeployJobCreated>(`${clusterPath(clusterName)}/jobs`, {
method: 'POST',
body: JSON.stringify({ sourceAgentId, preflightId, targetNodeIds, maxParallel: 2 }),
}),
getJob: (jobId: string) =>
apiFetchJSON<ClusterDeployJob>(`/api/agent-deploy/jobs/${encodeURIComponent(jobId)}`),
};
@@ -0,0 +1,518 @@
import { For, Show, createEffect, createMemo, createSignal } from 'solid-js';
import type { Component } from 'solid-js';
import { Button } from '@/components/shared/Button';
import { Card } from '@/components/shared/Card';
import { Dialog } from '@/components/shared/Dialog';
import { LoadingSpinner } from '@/components/shared/LoadingSpinner';
import { SectionHeader } from '@/components/shared/SectionHeader';
import {
ClusterAgentDeployAPI,
isClusterDeployJobActive,
type ClusterDeployCandidateNode,
type ClusterDeployJob,
type ClusterDeployTargetStatus,
} from '@/api/agentDeploy';
/**
* One-step install of the Pulse Agent on uncovered Proxmox cluster members.
*
* Pulse already knows the cluster's members and has an agent on at least one
* of them, so it asks that agent to install on its peers over the SSH trust
* the cluster nodes already share. The server mints a per-node bootstrap
* credential, so the operator never copies, pastes, or reuses a token. When
* the cluster has no reporting agent to install from, or a member cannot be
* reached, the dialog hands off to the scoped manual installer.
*/
export type ClusterAgentDeployNodeState =
'idle' | 'checking' | 'installing' | 'connecting' | 'reporting' | 'failed';
export interface ClusterAgentDeployNodeProgress {
state: ClusterAgentDeployNodeState;
detail?: string;
}
/** Maps a server target status onto the four steps the operator cares about. */
export const clusterDeployNodeStateFromTarget = (
status: ClusterDeployTargetStatus,
): ClusterAgentDeployNodeState => {
switch (status) {
case 'pending':
case 'preflighting':
case 'ready':
return 'checking';
case 'installing':
return 'installing';
case 'enrolling':
case 'verifying':
case 'succeeded':
case 'skipped_already_agent':
return 'connecting';
case 'failed_retryable':
case 'failed_permanent':
case 'canceled':
return 'failed';
default:
return 'checking';
}
};
const NODE_STATE_PRESENTATION: Record<
ClusterAgentDeployNodeState,
{ label: string; badgeClass: string }
> = {
idle: { label: 'Not installed', badgeClass: 'bg-surface-alt text-muted' },
checking: {
label: 'Checking',
badgeClass: 'bg-blue-100 text-blue-800 dark:bg-blue-900/25 dark:text-blue-200',
},
installing: {
label: 'Installing',
badgeClass: 'bg-blue-100 text-blue-800 dark:bg-blue-900/25 dark:text-blue-200',
},
connecting: {
label: 'Connecting to Pulse',
badgeClass: 'bg-blue-100 text-blue-800 dark:bg-blue-900/25 dark:text-blue-200',
},
reporting: {
label: 'Reporting',
badgeClass: 'bg-green-100 text-green-800 dark:bg-green-900/25 dark:text-green-300',
},
failed: {
label: 'Failed',
badgeClass: 'bg-red-100 text-red-800 dark:bg-red-900/25 dark:text-red-300',
},
};
export const clusterDeployUnavailableReason = (node: ClusterDeployCandidateNode): string => {
if (node.reason === 'no_address') {
return 'Pulse has no IP address for this node. Set its connection address under Manage.';
}
return 'This node cannot be installed from here.';
};
type Phase = 'loading' | 'select' | 'running' | 'done' | 'unavailable' | 'error';
const POLL_INTERVAL_MS = 2000;
const REPORTING_TIMEOUT_MS = 90_000;
const JOB_TIMEOUT_MS = 20 * 60_000;
const REQUEST_TIMEOUT_MS = 30_000;
interface ClusterAgentDeployDialogProps {
isOpen: boolean;
/** The Proxmox cluster name, which the deploy API is keyed on. */
clusterName: string;
/** Member to preselect when opened from that member's row. */
preselectNodeName?: string;
onClose: () => void;
/** Hands off to the scoped manual installer. */
onUseInstaller: () => void;
/** Called once at least one node is reporting, so the caller can refresh. */
onInstalled?: () => void;
/** Overridable for tests. */
pollIntervalMs?: number;
/** Overridable for tests. */
jobTimeoutMs?: number;
/** Overridable for tests. */
requestTimeoutMs?: number;
}
const errorMessage = (error: unknown, fallback: string): string =>
error instanceof Error && error.message ? error.message : fallback;
export const ClusterAgentDeployDialog: Component<ClusterAgentDeployDialogProps> = (props) => {
const [phase, setPhase] = createSignal<Phase>('loading');
const [nodes, setNodes] = createSignal<ClusterDeployCandidateNode[]>([]);
const [sourceAgentId, setSourceAgentId] = createSignal('');
const [sourceNodeName, setSourceNodeName] = createSignal('');
const [selected, setSelected] = createSignal<ReadonlySet<string>>(new Set());
const [progress, setProgress] = createSignal<Record<string, ClusterAgentDeployNodeProgress>>({});
const [failure, setFailure] = createSignal('');
// A run keeps going after the dialog closes. The install continues on the
// server, and abandoning the browser side mid-run would skip starting the
// install after a passed check. Reopening starts a fresh view of the
// cluster, and the server refuses an overlapping install on it.
let runId = 0;
const wait = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
const uncoveredNodes = createMemo(() => nodes().filter((node) => !node.hasAgent));
const deployableNodes = createMemo(() => uncoveredNodes().filter((node) => node.deployable));
const selectedCount = createMemo(
() => deployableNodes().filter((node) => selected().has(node.nodeId)).length,
);
const setNodeProgress = (nodeId: string, next: ClusterAgentDeployNodeProgress) =>
setProgress((current) => ({ ...current, [nodeId]: next }));
const loadCandidates = async (run: number) => {
setPhase('loading');
setFailure('');
setProgress({});
try {
const candidates = await ClusterAgentDeployAPI.getCandidates(props.clusterName);
if (run !== runId) return;
const candidateNodes = candidates.nodes ?? [];
setNodes(candidateNodes);
const source = (candidates.sourceAgents ?? []).find((agent) => agent.online);
if (!source) {
setPhase('unavailable');
return;
}
setSourceAgentId(source.agentId);
setSourceNodeName(
candidateNodes.find((node) => node.nodeId === source.nodeId)?.name ?? 'a cluster node',
);
const deployable = candidateNodes.filter((node) => !node.hasAgent && node.deployable);
// Opened from one member's row means that member only. If it cannot be
// installed from here, nothing is preselected rather than its siblings.
const preselect = props.preselectNodeName?.trim().toLowerCase();
const chosen = preselect
? deployable.filter((node) => node.name.toLowerCase() === preselect)
: deployable;
setSelected(new Set(chosen.map((node) => node.nodeId)));
setPhase('select');
} catch (error) {
if (run !== runId) return;
setFailure(errorMessage(error, 'Pulse could not load this cluster’s nodes.'));
setPhase('error');
}
};
createEffect(() => {
if (props.isOpen) {
runId += 1;
void loadCandidates(runId);
}
});
// Everything a run needs is captured when it starts. The run outlives the
// dialog when it is closed mid-install, and props can no longer be read once
// the parent has unmounted it.
interface RunConfig {
run: number;
clusterName: string;
sourceAgentId: string;
pollMs: number;
jobTimeoutMs: number;
requestTimeoutMs: number;
onInstalled?: () => void;
}
// Bounds each request so a stalled one cannot hold the run past its
// deadline or skip the final sweep.
const withRequestDeadline = <T,>(cfg: RunConfig, request: Promise<T>): Promise<T> =>
new Promise<T>((resolve, reject) => {
const timer = setTimeout(
() => reject(new Error('Pulse did not answer in time. Check this page again shortly.')),
cfg.requestTimeoutMs,
);
request.then(
(value) => {
clearTimeout(timer);
resolve(value);
},
(error: unknown) => {
clearTimeout(timer);
reject(error);
},
);
});
const pollUntilSettled = async (
cfg: RunConfig,
fetchJob: () => Promise<ClusterDeployJob>,
): Promise<ClusterDeployJob | null> => {
const deadline = Date.now() + cfg.jobTimeoutMs;
for (;;) {
await wait(cfg.pollMs);
if (cfg.run !== runId) return null;
if (Date.now() > deadline) {
throw new Error(
'Pulse stopped waiting after 20 minutes. The install may still finish, so check this page again shortly.',
);
}
const job = await withRequestDeadline(cfg, fetchJob());
if (cfg.run !== runId) return null;
for (const target of job.targets ?? []) {
setNodeProgress(target.nodeId, {
state: clusterDeployNodeStateFromTarget(target.status),
detail: target.errorMessage,
});
}
if (!isClusterDeployJobActive(job.status)) return job;
}
};
// A finished job only means the agent was installed and started enrolling.
// The outcome the operator wants is Pulse counting the node, so confirm it.
const confirmReporting = async (cfg: RunConfig, nodeIds: string[]) => {
const pending = new Set(nodeIds);
const deadline = Date.now() + REPORTING_TIMEOUT_MS;
while (pending.size > 0 && Date.now() < deadline) {
await wait(cfg.pollMs);
if (cfg.run !== runId) return;
const candidates = await withRequestDeadline(
cfg,
ClusterAgentDeployAPI.getCandidates(cfg.clusterName),
);
if (cfg.run !== runId) return;
for (const node of candidates.nodes ?? []) {
if (pending.has(node.nodeId) && node.hasAgent) {
pending.delete(node.nodeId);
setNodeProgress(node.nodeId, { state: 'reporting' });
}
}
}
for (const nodeId of pending) {
setNodeProgress(nodeId, {
state: 'failed',
detail: 'The agent was installed but has not reported to Pulse yet.',
});
}
};
const install = async () => {
const cfg: RunConfig = {
run: runId,
clusterName: props.clusterName,
sourceAgentId: sourceAgentId(),
pollMs: props.pollIntervalMs ?? POLL_INTERVAL_MS,
jobTimeoutMs: props.jobTimeoutMs ?? JOB_TIMEOUT_MS,
requestTimeoutMs: props.requestTimeoutMs ?? REQUEST_TIMEOUT_MS,
onInstalled: props.onInstalled,
};
const targetIds = deployableNodes()
.filter((node) => selected().has(node.nodeId))
.map((node) => node.nodeId);
if (targetIds.length === 0) return;
setPhase('running');
setFailure('');
for (const nodeId of targetIds) setNodeProgress(nodeId, { state: 'checking' });
try {
const preflight = await withRequestDeadline(
cfg,
ClusterAgentDeployAPI.createPreflight(cfg.clusterName, cfg.sourceAgentId, targetIds),
);
const checked = await pollUntilSettled(cfg, () =>
ClusterAgentDeployAPI.getPreflight(preflight.preflightId),
);
if (!checked) return;
const checkedTargets = checked.targets ?? [];
const readyIds = checkedTargets
.filter((target) => target.status === 'ready')
.map((target) => target.nodeId);
// A node that gained an agent since the dialog opened is skipped by the
// check. It still needs confirming, not installing.
const toConfirm = checkedTargets
.filter((target) => target.status === 'skipped_already_agent')
.map((target) => target.nodeId);
if (readyIds.length > 0) {
for (const nodeId of readyIds) setNodeProgress(nodeId, { state: 'installing' });
const job = await withRequestDeadline(
cfg,
ClusterAgentDeployAPI.createJob(
cfg.clusterName,
cfg.sourceAgentId,
preflight.preflightId,
readyIds,
),
);
const finished = await pollUntilSettled(cfg, () => ClusterAgentDeployAPI.getJob(job.jobId));
if (!finished) return;
toConfirm.push(
...(finished.targets ?? [])
.filter((target) => clusterDeployNodeStateFromTarget(target.status) === 'connecting')
.map((target) => target.nodeId),
);
}
if (toConfirm.length > 0) await confirmReporting(cfg, toConfirm);
} catch (error) {
if (cfg.run !== runId) return;
setFailure(errorMessage(error, 'The install could not be started.'));
for (const nodeId of targetIds) {
if (progress()[nodeId]?.state !== 'reporting') {
setNodeProgress(nodeId, { state: 'failed' });
}
}
}
if (cfg.run !== runId) return;
// A node the preflight dropped or the job skipped never gets a terminal
// status from the server, so close it out here rather than leaving it
// looking busy forever.
for (const nodeId of targetIds) {
const state = progress()[nodeId]?.state;
if (state === 'checking' || state === 'installing' || state === 'connecting') {
setNodeProgress(nodeId, {
state: 'failed',
detail: 'Pulse could not finish installing on this node.',
});
}
}
setPhase('done');
if (Object.values(progress()).some((entry) => entry.state === 'reporting')) {
cfg.onInstalled?.();
}
};
const toggle = (nodeId: string, checked: boolean) =>
setSelected((current) => {
const next = new Set(current);
if (checked) next.add(nodeId);
else next.delete(nodeId);
return next;
});
const nodeState = (node: ClusterDeployCandidateNode): ClusterAgentDeployNodeState =>
progress()[node.nodeId]?.state ?? 'idle';
const anyFailed = createMemo(() =>
Object.values(progress()).some((entry) => entry.state === 'failed'),
);
const useInstaller = () => {
props.onClose();
props.onUseInstaller();
};
return (
<Dialog
isOpen={props.isOpen}
onClose={props.onClose}
panelClass="max-w-lg"
ariaLabel={`Install Pulse Agent on ${props.clusterName} nodes`}
>
<Card padding="lg" class="max-w-lg w-full">
<SectionHeader
title={`Install Pulse Agent on ${props.clusterName}`}
size="md"
class="mb-2"
/>
<Show when={phase() === 'loading'}>
<div class="flex items-center gap-2 py-6 text-sm text-muted">
<LoadingSpinner size="sm" tone="muted" />
Looking up the cluster’s nodes…
</div>
</Show>
<Show when={phase() === 'unavailable'}>
<p class="text-sm text-base-content">
Pulse installs from a node that already runs the Pulse Agent, and none of this cluster’s
nodes has one connected right now. Use the installer to add the first one. After that,
the rest can be installed from here.
</p>
</Show>
<Show when={phase() === 'error'}>
<p class="text-sm text-red-700 dark:text-red-300" role="alert">
{failure()}
</p>
</Show>
<Show when={['select', 'running', 'done'].includes(phase())}>
<p class="mb-4 text-sm text-muted">
Pulse will install the agent from {sourceNodeName()} over the SSH access your cluster
nodes already share. There is no token to copy.
</p>
<Show
when={uncoveredNodes().length > 0}
fallback={
<p class="text-sm text-base-content">Every node in this cluster has the agent.</p>
}
>
<ul class="space-y-2" aria-label="Cluster nodes without the Pulse Agent">
<For each={uncoveredNodes()}>
{(node) => (
<li class="rounded-md border border-border px-3 py-2">
<div class="flex items-center justify-between gap-3">
<label class="flex min-w-0 items-center gap-2 text-sm text-base-content">
<input
type="checkbox"
class="rounded"
checked={node.deployable && selected().has(node.nodeId)}
disabled={!node.deployable || phase() !== 'select'}
onChange={(event) => toggle(node.nodeId, event.currentTarget.checked)}
/>
<span class="truncate font-medium">{node.name}</span>
<Show when={node.ip}>
<span class="text-xs text-muted">{node.ip}</span>
</Show>
</label>
<span
class={`shrink-0 rounded px-2 py-0.5 text-xs ${NODE_STATE_PRESENTATION[nodeState(node)].badgeClass}`}
>
<Show
when={['checking', 'installing', 'connecting'].includes(nodeState(node))}
>
<LoadingSpinner size="xs" tone="current" class="mr-1 inline-block" />
</Show>
{NODE_STATE_PRESENTATION[nodeState(node)].label}
</span>
</div>
<Show when={!node.deployable}>
<p class="mt-1 text-xs text-muted">{clusterDeployUnavailableReason(node)}</p>
</Show>
<Show when={progress()[node.nodeId]?.detail}>
<p class="mt-1 text-xs text-red-700 dark:text-red-300">
{progress()[node.nodeId]?.detail}
</p>
</Show>
</li>
)}
</For>
</ul>
</Show>
<Show when={phase() === 'running'}>
<p class="mt-3 text-xs text-muted">
You can close this. The install keeps going, and each node shows up as covered once it
reports to Pulse.
</p>
</Show>
<Show when={failure()}>
<p class="mt-3 text-sm text-red-700 dark:text-red-300" role="alert">
{failure()}
</p>
</Show>
</Show>
<div class="mt-5 flex flex-wrap items-center justify-end gap-2">
<Show
when={
phase() === 'unavailable' ||
phase() === 'error' ||
anyFailed() ||
(phase() === 'select' && deployableNodes().length < uncoveredNodes().length)
}
>
<Button type="button" variant="ghost" onClick={useInstaller}>
Use the installer instead
</Button>
</Show>
<Show when={phase() === 'select'}>
<Button type="button" variant="outline" onClick={props.onClose}>
Cancel
</Button>
<Button
type="button"
variant="primary"
disabled={selectedCount() === 0}
onClick={() => void install()}
>
{selectedCount() === 1 ? 'Install on 1 node' : `Install on ${selectedCount()} nodes`}
</Button>
</Show>
<Show when={phase() !== 'select'}>
<Button type="button" variant="outline" onClick={props.onClose}>
Close
</Button>
</Show>
</div>
</Card>
</Dialog>
);
};
@@ -52,6 +52,7 @@ import {
type InfrastructureSourcePickerRouteStep,
} from '@/utils/infrastructureOnboardingPresentation';
import { getAgentHostProfileFamily } from '@/utils/platformSupportManifest';
import { ClusterAgentDeployDialog } from './ClusterAgentDeployDialog';
interface InfrastructureSourceManagerProps {
rows: Accessor<readonly InfrastructureSystemRow[]>;
@@ -467,6 +468,33 @@ export const InfrastructureSourceManager: Component<InfrastructureSourceManagerP
props.onAddSource?.('agent');
};
// Cluster members are installed in one step when a sibling already runs
// the agent: Pulse asks that agent to install on its peers, so there is no
// per-host token to copy. Clusters without a reporting agent, and standalone
// hosts, keep the scoped manual installer.
const [clusterDeploy, setClusterDeploy] = createSignal<{
clusterName: string;
nodeName?: string;
} | null>(null);
const canInstallFromCluster = (row: InfrastructureSystemRow): boolean =>
!props.readOnly &&
row.isCluster &&
Boolean(row.clusterName) &&
row.members.some((member) => Boolean(member.agentConnection));
const handleInstallAgent = (
row: InfrastructureSystemRow,
member?: InfrastructureSystemMemberRow,
) => {
if (canInstallFromCluster(row)) {
setClusterDeploy({
clusterName: row.clusterName!,
nodeName: member?.nativeName || member?.name,
});
return;
}
handleInstallAgentShortcut();
};
const hasAnyConfigured = createMemo(() => infrastructureRows().length > 0);
const hasAnyDiscovered = createMemo(() => props.discoveredNodes().length > 0);
@@ -571,11 +599,18 @@ export const InfrastructureSourceManager: Component<InfrastructureSourceManagerP
if (uncoveredAgentTargetCount() > 0 && (props.onAddSourceStep || props.onAddSource)) {
const namesText = uncoveredAgentTargetNamesText();
const target = namesText ?? formatCount(uncoveredAgentTargetCount(), 'Proxmox host');
// When every uncovered host sits in one cluster that can install from a
// sibling, the summary action opens that one-step install directly.
const uncoveredRows = infrastructureRows().filter((row) => rowNeedsAgentCoverage(row));
const soleCluster =
uncoveredRows.length === 1 && canInstallFromCluster(uncoveredRows[0])
? uncoveredRows[0]
: undefined;
return {
kind: 'agent',
label: uncoveredAgentTargetCount() === 1 ? 'Install agent' : 'Install agents',
detail: `Install one Pulse Agent on each uncovered Proxmox host (${target}) to add node-local telemetry such as temperatures, SMART data, and host identity.`,
onClick: handleInstallAgentShortcut,
onClick: soleCluster ? () => handleInstallAgent(soleCluster) : handleInstallAgentShortcut,
};
}
@@ -1090,11 +1125,13 @@ export const InfrastructureSourceManager: Component<InfrastructureSourceManagerP
variant="outline"
size="xs"
class="gap-1.5"
onClick={handleInstallAgentShortcut}
onClick={() => handleInstallAgent(row)}
aria-label={installAgentActionLabel(row)}
title={
row.isCluster
? 'Generate an installer command to run once on each uncovered Proxmox node.'
? canInstallFromCluster(row)
? 'Install Pulse Agent on the uncovered nodes from a node that already runs it.'
: 'Generate an installer command to run once on each uncovered Proxmox node.'
: 'Install Pulse Agent on this system to add node-local telemetry (temperatures, SMART, host identity).'
}
>
@@ -1255,9 +1292,15 @@ export const InfrastructureSourceManager: Component<InfrastructureSourceManagerP
variant="outline"
size="xs"
class="gap-1.5"
onClick={handleInstallAgentShortcut}
onClick={() =>
handleInstallAgent(row, member)
}
aria-label={`Install agent on ${member.name}`}
title={`Generate an installer command to run on ${member.name}.`}
title={
canInstallFromCluster(row)
? `Install Pulse Agent on ${member.name} from a node that already runs it.`
: `Generate an installer command to run on ${member.name}.`
}
>
<Cpu class="h-3.5 w-3.5" />
Install agent
@@ -1580,9 +1623,13 @@ export const InfrastructureSourceManager: Component<InfrastructureSourceManagerP
variant="outline"
size="xs"
class="mt-2 min-h-11 gap-1.5 lg:min-h-0"
onClick={handleInstallAgentShortcut}
onClick={() => handleInstallAgent(row, member)}
aria-label={`Install agent on ${member.name}`}
title={`Generate an installer command to run on ${member.name}.`}
title={
canInstallFromCluster(row)
? `Install Pulse Agent on ${member.name} from a node that already runs it.`
: `Generate an installer command to run on ${member.name}.`
}
>
<Cpu class="h-3.5 w-3.5" />
Install agent
@@ -1669,11 +1716,13 @@ export const InfrastructureSourceManager: Component<InfrastructureSourceManagerP
variant="outline"
size="xs"
class="min-h-11 gap-1.5 lg:min-h-0"
onClick={handleInstallAgentShortcut}
onClick={() => handleInstallAgent(row)}
aria-label={installAgentActionLabel(row)}
title={
row.isCluster
? 'Generate an installer command to run once on each uncovered Proxmox node.'
? canInstallFromCluster(row)
? 'Install Pulse Agent on the uncovered nodes from a node that already runs it.'
: 'Generate an installer command to run once on each uncovered Proxmox node.'
: 'Install Pulse Agent on this system to add node-local telemetry (temperatures, SMART, host identity).'
}
>
@@ -1795,6 +1844,19 @@ export const InfrastructureSourceManager: Component<InfrastructureSourceManagerP
</div>
</Show>
{discoveryMonitorBand()}
{/* Keyed so the dialog gets plain values: an install it started keeps
running after close, and must not read a stale <Show> accessor. */}
<Show when={clusterDeploy()} keyed>
{(target) => (
<ClusterAgentDeployDialog
isOpen={true}
clusterName={target.clusterName}
preselectNodeName={target.nodeName}
onClose={() => setClusterDeploy(null)}
onUseInstaller={handleInstallAgentShortcut}
/>
)}
</Show>
</SettingsPanel>
</div>
);
@@ -0,0 +1,399 @@
import { cleanup, fireEvent, render, screen, waitFor, within } from '@solidjs/testing-library';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import type { ClusterDeployCandidates, ClusterDeployJob } from '@/api/agentDeploy';
import {
ClusterAgentDeployDialog,
clusterDeployNodeStateFromTarget,
} from '../ClusterAgentDeployDialog';
const api = vi.hoisted(() => ({
getCandidates: vi.fn(),
createPreflight: vi.fn(),
getPreflight: vi.fn(),
createJob: vi.fn(),
getJob: vi.fn(),
}));
vi.mock('@/api/agentDeploy', async (importOriginal) => {
const actual = await importOriginal<typeof import('@/api/agentDeploy')>();
return { ...actual, ClusterAgentDeployAPI: api };
});
const candidates = (overrides: Partial<ClusterDeployCandidates> = {}): ClusterDeployCandidates => ({
clusterId: 'homelab',
clusterName: 'homelab',
sourceAgents: [{ agentId: 'agent-delly', nodeId: 'n-delly', online: true }],
nodes: [
{
nodeId: 'n-delly',
name: 'delly',
ip: '192.168.0.5',
hasAgent: true,
deployable: false,
reason: 'already_agent',
},
{ nodeId: 'n-delly2', name: 'delly2', ip: '192.168.0.111', hasAgent: false, deployable: true },
{ nodeId: 'n-minipc', name: 'minipc', ip: '192.168.0.134', hasAgent: false, deployable: true },
{ nodeId: 'n-pi', name: 'pi', hasAgent: false, deployable: false, reason: 'no_address' },
],
...overrides,
});
const job = (
id: string,
status: ClusterDeployJob['status'],
targets: ClusterDeployJob['targets'],
): ClusterDeployJob => ({ id, clusterId: 'homelab', status, targets });
const renderDialog = (props: Partial<Parameters<typeof ClusterAgentDeployDialog>[0]> = {}) => {
const onClose = vi.fn();
const onUseInstaller = vi.fn();
const onInstalled = vi.fn();
render(() => (
<ClusterAgentDeployDialog
isOpen={true}
clusterName="homelab"
onClose={onClose}
onUseInstaller={onUseInstaller}
onInstalled={onInstalled}
pollIntervalMs={1}
{...props}
/>
));
return { onClose, onUseInstaller, onInstalled };
};
describe('ClusterAgentDeployDialog', () => {
beforeEach(() => {
Object.values(api).forEach((fn) => fn.mockReset());
});
afterEach(() => cleanup());
it('lists uncovered nodes, preselects the installable ones, and explains the rest', async () => {
api.getCandidates.mockResolvedValue(candidates());
renderDialog();
const list = await screen.findByRole('list', { name: 'Cluster nodes without the Pulse Agent' });
expect(within(list).queryByText('delly')).toBeNull();
expect(within(list).getByRole('checkbox', { name: /delly2/ })).toBeChecked();
expect(within(list).getByRole('checkbox', { name: /minipc/ })).toBeChecked();
const pi = within(list).getByRole('checkbox', { name: /pi/ });
expect(pi).toBeDisabled();
expect(pi).not.toBeChecked();
expect(screen.getByText(/Pulse has no IP address for this node/)).toBeInTheDocument();
expect(screen.getByText(/install the agent from delly/)).toBeInTheDocument();
expect(screen.getByRole('button', { name: 'Install on 2 nodes' })).toBeEnabled();
expect(screen.getByRole('button', { name: 'Use the installer instead' })).toBeInTheDocument();
});
it('selects only the member the operator clicked install on', async () => {
api.getCandidates.mockResolvedValue(candidates());
renderDialog({ preselectNodeName: 'delly2' });
expect(await screen.findByRole('checkbox', { name: /delly2/ })).toBeChecked();
expect(screen.getByRole('checkbox', { name: /minipc/ })).not.toBeChecked();
expect(screen.getByRole('button', { name: 'Install on 1 node' })).toBeEnabled();
});
it('checks, installs, and confirms the node is reporting before calling it done', async () => {
api.getCandidates.mockResolvedValueOnce(candidates()).mockResolvedValue(
candidates({
nodes: candidates().nodes!.map((node) =>
node.nodeId === 'n-delly2' ? { ...node, hasAgent: true, deployable: false } : node,
),
}),
);
api.createPreflight.mockResolvedValue({ preflightId: 'pf_1', status: 'running' });
api.getPreflight.mockResolvedValue(
job('pf_1', 'succeeded', [
{
id: 't1',
nodeId: 'n-delly2',
nodeName: 'delly2',
nodeIP: '192.168.0.111',
status: 'ready',
},
]),
);
api.createJob.mockResolvedValue({
jobId: 'dep_1',
acceptedTargets: ['n-delly2'],
skippedTargets: [],
});
api.getJob.mockResolvedValue(
job('dep_1', 'succeeded', [
{
id: 't2',
nodeId: 'n-delly2',
nodeName: 'delly2',
nodeIP: '192.168.0.111',
status: 'enrolling',
},
]),
);
const { onInstalled } = renderDialog({ preselectNodeName: 'delly2' });
fireEvent.click(await screen.findByRole('button', { name: 'Install on 1 node' }));
await waitFor(() => expect(screen.getByText('Reporting')).toBeInTheDocument());
expect(api.createPreflight).toHaveBeenCalledWith('homelab', 'agent-delly', ['n-delly2']);
expect(api.createJob).toHaveBeenCalledWith('homelab', 'agent-delly', 'pf_1', ['n-delly2']);
expect(onInstalled).toHaveBeenCalledTimes(1);
expect(screen.getByRole('button', { name: 'Close' })).toBeInTheDocument();
});
it('closes out a node the job never reported on instead of leaving it busy', async () => {
const reported = candidates({
nodes: candidates().nodes!.map((node) =>
node.nodeId === 'n-delly2' ? { ...node, hasAgent: true, deployable: false } : node,
),
});
api.getCandidates.mockResolvedValueOnce(candidates()).mockResolvedValue(reported);
api.createPreflight.mockResolvedValue({ preflightId: 'pf_3', status: 'running' });
api.getPreflight.mockResolvedValue(
job('pf_3', 'succeeded', [
{
id: 't1',
nodeId: 'n-delly2',
nodeName: 'delly2',
nodeIP: '192.168.0.111',
status: 'ready',
},
{
id: 't2',
nodeId: 'n-minipc',
nodeName: 'minipc',
nodeIP: '192.168.0.134',
status: 'ready',
},
]),
);
api.createJob.mockResolvedValue({
jobId: 'dep_3',
acceptedTargets: ['n-delly2'],
skippedTargets: [{ nodeId: 'n-minipc', reason: 'preflight_status_failed' }],
});
api.getJob.mockResolvedValue(
job('dep_3', 'succeeded', [
{
id: 't3',
nodeId: 'n-delly2',
nodeName: 'delly2',
nodeIP: '192.168.0.111',
status: 'succeeded',
},
]),
);
renderDialog();
fireEvent.click(await screen.findByRole('button', { name: 'Install on 2 nodes' }));
expect(
await screen.findByText('Pulse could not finish installing on this node.'),
).toBeInTheDocument();
expect(screen.getByText('Reporting')).toBeInTheDocument();
expect(screen.getByText('Failed')).toBeInTheDocument();
expect(screen.getByRole('button', { name: 'Use the installer instead' })).toBeInTheDocument();
});
it('stops before installing when the check fails and offers the installer', async () => {
api.getCandidates.mockResolvedValue(candidates());
api.createPreflight.mockResolvedValue({ preflightId: 'pf_2', status: 'running' });
api.getPreflight.mockResolvedValue(
job('pf_2', 'failed', [
{
id: 't1',
nodeId: 'n-delly2',
nodeName: 'delly2',
nodeIP: '192.168.0.111',
status: 'failed_permanent',
errorMessage: 'SSH to 192.168.0.111 was refused',
},
]),
);
const { onUseInstaller, onClose } = renderDialog({ preselectNodeName: 'delly2' });
fireEvent.click(await screen.findByRole('button', { name: 'Install on 1 node' }));
expect(await screen.findByText('SSH to 192.168.0.111 was refused')).toBeInTheDocument();
expect(screen.getByText('Failed')).toBeInTheDocument();
expect(api.createJob).not.toHaveBeenCalled();
fireEvent.click(screen.getByRole('button', { name: 'Use the installer instead' }));
expect(onClose).toHaveBeenCalled();
expect(onUseInstaller).toHaveBeenCalled();
});
it('hands off to the installer when no cluster node has a connected agent', async () => {
api.getCandidates.mockResolvedValue(candidates({ sourceAgents: [] }));
const { onUseInstaller } = renderDialog();
expect(
await screen.findByText(/none of this cluster’s nodes has one connected/),
).toBeInTheDocument();
expect(screen.queryByRole('button', { name: /Install on/ })).toBeNull();
fireEvent.click(screen.getByRole('button', { name: 'Use the installer instead' }));
expect(onUseInstaller).toHaveBeenCalled();
});
});
describe('ClusterAgentDeployDialog edge cases', () => {
beforeEach(() => {
Object.values(api).forEach((fn) => fn.mockReset());
});
afterEach(() => cleanup());
it('selects nothing when the clicked member cannot be installed from here', async () => {
api.getCandidates.mockResolvedValue(candidates());
renderDialog({ preselectNodeName: 'pi' });
expect(await screen.findByRole('checkbox', { name: /delly2/ })).not.toBeChecked();
expect(screen.getByRole('checkbox', { name: /minipc/ })).not.toBeChecked();
expect(screen.getByRole('button', { name: 'Install on 0 nodes' })).toBeDisabled();
expect(screen.getByText(/Pulse has no IP address for this node/)).toBeInTheDocument();
});
it('confirms a node that gained an agent during the check without installing it', async () => {
const covered = candidates({
nodes: candidates().nodes!.map((node) =>
node.nodeId === 'n-delly2' ? { ...node, hasAgent: true, deployable: false } : node,
),
});
api.getCandidates.mockResolvedValueOnce(candidates()).mockResolvedValue(covered);
api.createPreflight.mockResolvedValue({ preflightId: 'pf_4', status: 'running' });
api.getPreflight.mockResolvedValue(
job('pf_4', 'succeeded', [
{
id: 't1',
nodeId: 'n-delly2',
nodeName: 'delly2',
nodeIP: '192.168.0.111',
status: 'skipped_already_agent',
},
]),
);
const { onInstalled } = renderDialog({ preselectNodeName: 'delly2' });
fireEvent.click(await screen.findByRole('button', { name: 'Install on 1 node' }));
await waitFor(() => expect(screen.getByText('Reporting')).toBeInTheDocument());
expect(api.createJob).not.toHaveBeenCalled();
expect(onInstalled).toHaveBeenCalledTimes(1);
});
it('gives up on a stalled install with an explanation and the installer', async () => {
api.getCandidates.mockResolvedValue(candidates());
api.createPreflight.mockResolvedValue({ preflightId: 'pf_5', status: 'running' });
api.getPreflight.mockResolvedValue(
job('pf_5', 'succeeded', [
{
id: 't1',
nodeId: 'n-delly2',
nodeName: 'delly2',
nodeIP: '192.168.0.111',
status: 'ready',
},
]),
);
api.createJob.mockResolvedValue({
jobId: 'dep_5',
acceptedTargets: ['n-delly2'],
skippedTargets: [],
});
api.getJob.mockResolvedValue(
job('dep_5', 'running', [
{
id: 't2',
nodeId: 'n-delly2',
nodeName: 'delly2',
nodeIP: '192.168.0.111',
status: 'installing',
},
]),
);
renderDialog({ preselectNodeName: 'delly2', jobTimeoutMs: 20 });
fireEvent.click(await screen.findByRole('button', { name: 'Install on 1 node' }));
expect(await screen.findByText(/Pulse stopped waiting after 20 minutes/)).toBeInTheDocument();
expect(screen.getByText('Failed')).toBeInTheDocument();
expect(screen.getByRole('button', { name: 'Use the installer instead' })).toBeInTheDocument();
});
it('gives up on a status request that never answers', async () => {
api.getCandidates.mockResolvedValue(candidates());
api.createPreflight.mockResolvedValue({ preflightId: 'pf_7', status: 'running' });
api.getPreflight.mockReturnValue(new Promise(() => {}));
renderDialog({ preselectNodeName: 'delly2', requestTimeoutMs: 20 });
fireEvent.click(await screen.findByRole('button', { name: 'Install on 1 node' }));
expect(await screen.findByText(/Pulse did not answer in time/)).toBeInTheDocument();
expect(screen.getByText('Failed')).toBeInTheDocument();
expect(screen.getByRole('button', { name: 'Use the installer instead' })).toBeInTheDocument();
});
it('keeps installing after the dialog is closed mid-run', async () => {
api.getCandidates.mockResolvedValue(candidates());
api.createPreflight.mockResolvedValue({ preflightId: 'pf_6', status: 'running' });
api.getPreflight.mockResolvedValue(
job('pf_6', 'succeeded', [
{
id: 't1',
nodeId: 'n-delly2',
nodeName: 'delly2',
nodeIP: '192.168.0.111',
status: 'ready',
},
]),
);
api.createJob.mockResolvedValue({
jobId: 'dep_6',
acceptedTargets: ['n-delly2'],
skippedTargets: [],
});
// The job ends on its first poll so the orphaned run finishes instead of
// polling on in the background after this test.
api.getJob.mockResolvedValue(
job('dep_6', 'failed', [
{
id: 't2',
nodeId: 'n-delly2',
nodeName: 'delly2',
nodeIP: '192.168.0.111',
status: 'failed_permanent',
},
]),
);
renderDialog({ preselectNodeName: 'delly2' });
fireEvent.click(await screen.findByRole('button', { name: 'Install on 1 node' }));
cleanup();
await waitFor(() =>
expect(api.createJob).toHaveBeenCalledWith('homelab', 'agent-delly', 'pf_6', ['n-delly2']),
);
});
it('shows the server refusal for an ambiguous cluster and offers the installer', async () => {
api.getCandidates.mockRejectedValue(
new Error('More than one Proxmox connection reports a cluster named "homelab".'),
);
const { onUseInstaller } = renderDialog();
expect(await screen.findByRole('alert')).toHaveTextContent(/More than one Proxmox connection/);
fireEvent.click(screen.getByRole('button', { name: 'Use the installer instead' }));
expect(onUseInstaller).toHaveBeenCalled();
});
});
describe('clusterDeployNodeStateFromTarget', () => {
it('folds server target statuses into the steps an operator sees', () => {
expect(clusterDeployNodeStateFromTarget('preflighting')).toBe('checking');
expect(clusterDeployNodeStateFromTarget('ready')).toBe('checking');
expect(clusterDeployNodeStateFromTarget('installing')).toBe('installing');
expect(clusterDeployNodeStateFromTarget('enrolling')).toBe('connecting');
expect(clusterDeployNodeStateFromTarget('succeeded')).toBe('connecting');
expect(clusterDeployNodeStateFromTarget('failed_retryable')).toBe('failed');
expect(clusterDeployNodeStateFromTarget('canceled')).toBe('failed');
});
});
@@ -1,4 +1,4 @@
import { cleanup, fireEvent, render, screen, within } from '@solidjs/testing-library';
import { cleanup, fireEvent, render, screen, waitFor, within } from '@solidjs/testing-library';
import { afterEach, describe, expect, it, vi } from 'vitest';
import infrastructureSourceManagerSource from '../InfrastructureSourceManager.tsx?raw';
import {
@@ -13,6 +13,19 @@ import {
} from '../connectionsTableModel';
import type { Connection } from '@/api/connections';
const clusterDeployApi = vi.hoisted(() => ({
getCandidates: vi.fn(),
createPreflight: vi.fn(),
getPreflight: vi.fn(),
createJob: vi.fn(),
getJob: vi.fn(),
}));
vi.mock('@/api/agentDeploy', async (importOriginal) => {
const actual = await importOriginal<typeof import('@/api/agentDeploy')>();
return { ...actual, ClusterAgentDeployAPI: clusterDeployApi };
});
const connectionFixture = (overrides: Partial<Connection> = {}): Connection => ({
id: 'agent:host-1',
type: 'agent',
@@ -569,6 +582,182 @@ describe('InfrastructureSourceManager setup summary', () => {
expect(onAddSourceStep).toHaveBeenCalledWith('linux-host');
});
it('installs uncovered cluster members in one step when a sibling already runs the agent', async () => {
clusterDeployApi.getCandidates.mockResolvedValue({
clusterId: 'remote',
clusterName: 'remote',
sourceAgents: [{ agentId: 'agent-pve-a', nodeId: 'n-a', online: true }],
nodes: [
{ nodeId: 'n-a', name: 'pve-a', ip: '10.15.5.11', hasAgent: true, deployable: false },
{ nodeId: 'n-b', name: 'pve-b', ip: '10.15.5.12', hasAgent: false, deployable: true },
],
});
const onAddSourceStep = vi.fn();
const attachedAgent = connectionFixture({ id: 'agent:pve-a', name: 'pve-a', address: 'pve-a' });
render(() => (
<InfrastructureSourceManager
rows={() => [
row({
id: 'pve:remote-cluster',
ownerType: 'pve',
name: 'remote',
clusterName: 'remote',
subtitle: 'Cluster · 2 nodes',
source: 'both',
isCluster: true,
canEdit: true,
connection: connectionFixture({
id: 'pve:remote-cluster',
type: 'pve',
name: 'remote',
source: 'manual',
capabilities: { supportsPause: true, supportsScope: true, supportsTest: true },
}),
attachedConnections: [attachedAgent],
members: [
member({
id: 'pve-a',
name: 'pve-a',
source: 'both',
agentConnection: attachedAgent,
}),
member({
id: 'pve-b',
name: 'pve-b',
source: 'api',
coverageLabels: [],
agentConnection: undefined,
}),
],
}),
]}
discoveredNodes={() => []}
discoveryEnabled={false}
discoveryScanStatus={() => ({ scanning: false })}
readOnly={false}
onAddSourceStep={onAddSourceStep}
onOpenConnection={vi.fn()}
/>
));
fireEvent.click(screen.getByRole('button', { name: 'Show 2 nodes for remote' }));
fireEvent.click(screen.getByRole('button', { name: 'Install agent on pve-b' }));
const dialog = await screen.findByRole('dialog', {
name: 'Install Pulse Agent on remote nodes',
});
expect(await within(dialog).findByRole('checkbox', { name: /pve-b/ })).toBeChecked();
expect(clusterDeployApi.getCandidates).toHaveBeenCalledWith('remote');
expect(onAddSourceStep).not.toHaveBeenCalled();
});
it('keeps a started cluster install going after the dialog is closed', async () => {
clusterDeployApi.getCandidates.mockResolvedValue({
clusterId: 'remote',
clusterName: 'remote',
sourceAgents: [{ agentId: 'agent-pve-a', nodeId: 'n-a', online: true }],
nodes: [
{ nodeId: 'n-a', name: 'pve-a', ip: '10.15.5.11', hasAgent: true, deployable: false },
{ nodeId: 'n-b', name: 'pve-b', ip: '10.15.5.12', hasAgent: false, deployable: true },
],
});
clusterDeployApi.createPreflight.mockResolvedValue({ preflightId: 'pf_r', status: 'running' });
clusterDeployApi.getPreflight.mockResolvedValue({
id: 'pf_r',
clusterId: 'remote',
status: 'succeeded',
targets: [
{ id: 't1', nodeId: 'n-b', nodeName: 'pve-b', nodeIP: '10.15.5.12', status: 'ready' },
],
});
clusterDeployApi.createJob.mockResolvedValue({
jobId: 'dep_r',
acceptedTargets: ['n-b'],
skippedTargets: [],
});
clusterDeployApi.getJob.mockResolvedValue({
id: 'dep_r',
clusterId: 'remote',
status: 'failed',
targets: [
{
id: 't2',
nodeId: 'n-b',
nodeName: 'pve-b',
nodeIP: '10.15.5.12',
status: 'failed_permanent',
},
],
});
const attachedAgent = connectionFixture({ id: 'agent:pve-a', name: 'pve-a', address: 'pve-a' });
render(() => (
<InfrastructureSourceManager
rows={() => [
row({
id: 'pve:remote-cluster',
ownerType: 'pve',
name: 'remote',
clusterName: 'remote',
subtitle: 'Cluster · 2 nodes',
source: 'both',
isCluster: true,
canEdit: true,
connection: connectionFixture({
id: 'pve:remote-cluster',
type: 'pve',
name: 'remote',
source: 'manual',
capabilities: { supportsPause: true, supportsScope: true, supportsTest: true },
}),
attachedConnections: [attachedAgent],
members: [
member({
id: 'pve-a',
name: 'pve-a',
source: 'both',
agentConnection: attachedAgent,
}),
member({
id: 'pve-b',
name: 'pve-b',
source: 'api',
coverageLabels: [],
agentConnection: undefined,
}),
],
}),
]}
discoveredNodes={() => []}
discoveryEnabled={false}
discoveryScanStatus={() => ({ scanning: false })}
readOnly={false}
onAddSourceStep={vi.fn()}
onOpenConnection={vi.fn()}
/>
));
fireEvent.click(screen.getByRole('button', { name: 'Show 2 nodes for remote' }));
fireEvent.click(screen.getByRole('button', { name: 'Install agent on pve-b' }));
const dialog = await screen.findByRole('dialog', {
name: 'Install Pulse Agent on remote nodes',
});
fireEvent.click(await within(dialog).findByRole('button', { name: 'Install on 1 node' }));
fireEvent.click(await within(dialog).findByRole('button', { name: 'Close' }));
// The parent has unmounted the dialog; the run must still start the install
// with the cluster it captured, rather than reading a stale accessor.
expect(screen.queryByRole('dialog')).toBeNull();
await waitFor(
() =>
expect(clusterDeployApi.createJob).toHaveBeenCalledWith('remote', 'agent-pve-a', 'pf_r', [
'n-b',
]),
{ timeout: 6000 },
);
});
it('offers the same uncovered-node onboarding in the narrow card layout', () => {
Object.defineProperty(window, 'innerWidth', { configurable: true, value: 600 });
const onAddSourceStep = vi.fn();
@@ -1650,6 +1650,13 @@ describe('settings architecture guardrails', () => {
expect(infrastructureSourceManagerSource).toContain('Manage');
expect(infrastructureSourceManagerSource).not.toContain('Detect address');
expect(infrastructureSourceManagerSource).toContain("'Install agent'");
// Cluster members install in one step through the server-side deploy, so
// the install actions route through the canonical dialog and keep the
// scoped installer as the fallback instead of a hand-copied token.
expect(infrastructureSourceManagerSource).toContain("from './ClusterAgentDeployDialog'");
expect(infrastructureSourceManagerSource).toContain('<ClusterAgentDeployDialog');
expect(infrastructureSourceManagerSource).toContain('onUseInstaller={handleInstallAgentShortcut}');
expect(infrastructureSourceManagerSource).toContain('<Show when={clusterDeploy()} keyed>');
expect(infrastructureSourceManagerSource).toContain('Add infrastructure');
expect(infrastructureSourceManagerSource).not.toContain('Monitor endpoint');
expect(infrastructureSourceManagerSource).toContain('getInfrastructureEmptyStateSummary');
@@ -1017,6 +1017,8 @@ export interface InfrastructureSystemRow {
canRemove: boolean;
isAgent: boolean;
isCluster: boolean;
/** Proxmox cluster name for cluster rows; the cluster agent deploy API is keyed on it. */
clusterName?: string;
attachedConnections: Connection[];
members: InfrastructureSystemMemberRow[];
connection: Connection;
@@ -430,6 +430,7 @@ const buildRow = (
canRemove: primaryConnection.type !== 'docker' && primaryConnection.type !== 'kubernetes',
isAgent: primaryConnection.type === 'agent',
isCluster,
...(isCluster ? { clusterName } : {}),
attachedConnections,
members,
connection: primaryConnection,
+131 -3
View File
@@ -6,6 +6,7 @@ import (
"errors"
"fmt"
"io"
"net"
"net/http"
"net/url"
"strings"
@@ -34,6 +35,8 @@ type DeployHandlers struct {
config *config.Config
persistence *config.ConfigPersistence
// jobAdmissionMu serializes install-job admission (see HandleCreateJob).
jobAdmissionMu sync.Mutex
// Active preflight SSE subscriptions keyed by preflightID.
sseMu sync.Mutex
sseSubs map[string]*deploySSESub
@@ -122,6 +125,7 @@ func (h *DeployHandlers) HandleCandidates(w http.ResponseWriter, r *http.Request
nodes []candidateNode
sourceAgents []sourceAgentInfo
)
connections := make(map[string]bool)
for _, node := range readState.Nodes() {
if node == nil {
@@ -134,6 +138,7 @@ func (h *DeployHandlers) HandleCandidates(w http.ResponseWriter, r *http.Request
if node.ClusterName() != clusterID {
continue
}
connections[node.Instance()] = true
if clusterName == "" {
clusterName = node.ClusterName()
}
@@ -142,7 +147,7 @@ func (h *DeployHandlers) HandleCandidates(w http.ResponseWriter, r *http.Request
cn := candidateNode{
NodeID: node.ID(),
Name: nodeName(node),
IP: nodeIP(node.HostURL()),
IP: h.deployTargetIP(node),
HasAgent: hasAgent,
}
@@ -159,6 +164,12 @@ func (h *DeployHandlers) HandleCandidates(w http.ResponseWriter, r *http.Request
Online: true,
})
}
} else if cn.IP == "" {
// Preflight would reject the target anyway; say why up front so
// the UI can point the operator at the member's connection
// address instead of offering an install that cannot start.
cn.Deployable = false
cn.Reason = "no_address"
} else {
cn.Deployable = true
}
@@ -166,6 +177,11 @@ func (h *DeployHandlers) HandleCandidates(w http.ResponseWriter, r *http.Request
nodes = append(nodes, cn)
}
if len(connections) > 1 {
writeAmbiguousClusterError(w, clusterID)
return
}
resp := candidatesResponse{
ClusterID: clusterID,
ClusterName: clusterName,
@@ -241,12 +257,14 @@ func (h *DeployHandlers) HandleCreatePreflight(w http.ResponseWriter, r *http.Re
clusterName := ""
sourceNodeID := ""
nodesByID := make(map[string]*unifiedresources.NodeView)
connections := make(map[string]bool)
for _, node := range readState.Nodes() {
if node == nil {
continue
}
if node.ClusterName() == clusterID && node.IsClusterMember() {
nodesByID[node.ID()] = node
connections[node.Instance()] = true
if clusterName == "" {
clusterName = node.ClusterName()
}
@@ -256,6 +274,11 @@ func (h *DeployHandlers) HandleCreatePreflight(w http.ResponseWriter, r *http.Re
}
}
if len(connections) > 1 {
writeAmbiguousClusterError(w, clusterID)
return
}
if sourceNodeID == "" {
writeErrorResponse(w, http.StatusBadRequest, "source_not_in_cluster",
"Source agent is not linked to a node in this cluster", nil)
@@ -295,9 +318,9 @@ func (h *DeployHandlers) HandleCreatePreflight(w http.ResponseWriter, r *http.Re
if !ok {
continue // skip nodes not in cluster
}
ip := nodeIP(node.HostURL())
ip := h.deployTargetIP(node)
if ip == "" {
continue // skip nodes without IP
continue // skip nodes without a literal IP the agent can reach
}
targetID := generateID("tgt")
@@ -1058,6 +1081,24 @@ func (h *DeployHandlers) HandleCreateJob(w http.ResponseWriter, r *http.Request)
"Preflight source agent does not match request source agent", nil)
return
}
// Hold admission from the active-job check through the new job's insert,
// so two concurrent requests cannot both pass and start overlapping
// installs. The deploy store is this server's own SQLite file, so an
// in-process lock serializes every admission against it.
h.jobAdmissionMu.Lock()
defer h.jobAdmissionMu.Unlock()
activeJobID, err := h.activeDeployJobForCluster(ctx, orgID, clusterID)
if err != nil {
log.Error().Err(err).Str("cluster_id", clusterID).Msg("Failed to check for an active deploy job")
writeErrorResponse(w, http.StatusInternalServerError, "store_error", "Failed to check for an active deploy job", nil)
return
}
if activeJobID != "" {
writeErrorResponse(w, http.StatusConflict, "deploy_in_progress",
"An agent install is already running on this cluster. Wait for it to finish, then try again.",
map[string]string{"jobId": activeJobID})
return
}
// Get preflight targets — filter requested nodeIDs against Ready targets.
pfTargets, err := h.store.GetTargetsForJob(ctx, req.PreflightID)
@@ -1748,6 +1789,93 @@ func extractPathSuffix(path, prefix string) string {
return strings.TrimSpace(s)
}
// deployTargetIP returns the literal IP the source agent should SSH to for a
// cluster member, or "" when Pulse knows of none. The agent only accepts
// literal IPs (validateNodeIP guards the SSH command against injection), but
// Pulse often discovers members by name, so the API URL can carry a hostname
// such as "pve-b" that would fail preflight. Mirror the monitor's endpoint
// precedence: a literal IP in the API URL first, then the cluster endpoint's
// effective IP (the operator override, else the address Proxmox reports).
func (h *DeployHandlers) deployTargetIP(node *unifiedresources.NodeView) string {
if node == nil {
return ""
}
if host := nodeIP(node.HostURL()); net.ParseIP(host) != nil {
return host
}
if endpoint, ok := h.clusterEndpointForNode(node); ok {
if ip := strings.TrimSpace(endpoint.EffectiveIP()); net.ParseIP(ip) != nil {
return ip
}
}
return ""
}
// clusterEndpointForNode finds the configured cluster endpoint for a member,
// matching the PVE connection by exact instance name and the endpoint by
// Proxmox node name. There is deliberately no cluster-name fallback: two
// sites can share a cluster name, and a wrong match would send a source agent
// privileged SSH work against another site's address.
func (h *DeployHandlers) clusterEndpointForNode(node *unifiedresources.NodeView) (config.ClusterEndpoint, bool) {
if h == nil || h.config == nil || node == nil {
return config.ClusterEndpoint{}, false
}
memberName := strings.TrimSpace(node.NodeName())
if memberName == "" {
memberName = strings.TrimSpace(node.Name())
}
instance := node.Instance()
if memberName == "" || instance == "" {
return config.ClusterEndpoint{}, false
}
for _, inst := range h.config.PVEInstances {
if !inst.IsCluster || inst.Name != instance {
continue
}
for _, endpoint := range inst.ClusterEndpoints {
if strings.EqualFold(strings.TrimSpace(endpoint.NodeName), memberName) {
return endpoint, true
}
}
}
return config.ClusterEndpoint{}, false
}
// writeAmbiguousClusterError refuses a deploy request whose cluster name is
// reported by more than one Proxmox connection. The deploy routes are keyed
// on the cluster name, so serving such a request would merge two sites'
// members and could send a source agent to the other site's addresses.
func writeAmbiguousClusterError(w http.ResponseWriter, clusterID string) {
writeErrorResponse(w, http.StatusConflict, "ambiguous_cluster",
fmt.Sprintf("More than one Proxmox connection reports a cluster named %q, so Pulse cannot tell which nodes to install on. Use the installer for these nodes instead.", clusterID),
nil)
}
// deployActiveWindow bounds how long an unfinished install job blocks the next
// one on the same cluster. A job whose source agent vanished mid-run can stay
// "running" in the store and must not lock the cluster out indefinitely.
const deployActiveWindow = 30 * time.Minute
// activeDeployJobForCluster returns the ID of a recent install job on the
// cluster that has not finished, so overlapping installs (a reopened dialog,
// a second tab) are refused instead of racing over the same targets.
// Callers must hold jobAdmissionMu so the check and the new job's insert are
// atomic with respect to a concurrent request for the same cluster.
func (h *DeployHandlers) activeDeployJobForCluster(ctx context.Context, orgID, clusterID string) (string, error) {
jobs, err := h.store.UnfinishedJobsForCluster(ctx, orgID, clusterID)
if err != nil {
return "", err
}
cutoff := time.Now().Add(-deployActiveWindow)
for _, job := range jobs {
// Preflights share the jobs table; only install jobs conflict.
if strings.HasPrefix(job.ID, "dep_") && !job.CreatedAt.Before(cutoff) {
return job.ID, nil
}
}
return "", nil
}
// nodeIP extracts the hostname/IP from a node host URL (e.g. "https://198.51.100.2:8006" -> "198.51.100.2").
func nodeIP(hostURL string) string {
raw := strings.TrimSpace(hostURL)
+330
View File
@@ -9,6 +9,7 @@ import (
"os"
"path/filepath"
"strings"
"sync"
"testing"
"time"
@@ -1808,3 +1809,332 @@ func TestGetTarget(t *testing.T) {
t.Fatalf("got Arch=%q, want amd64", target.Arch)
}
}
// clusterDeployAddressFixture models a cluster whose members Pulse discovered by
// name: the API URLs carry hostnames, while the cluster endpoints record the
// addresses Proxmox reports (and, for pve-c, an operator override).
func clusterDeployAddressFixture(t *testing.T) *DeployHandlers {
t.Helper()
nodes := []models.Node{
{
ID: "node_pve-a", Name: "pve-a", Host: "https://10.0.0.1:8006", Instance: "lab-conn",
IsClusterMember: true, ClusterName: "lab", LinkedAgentID: "host-a",
},
{
ID: "node_pve-b", Name: "pve-b", Host: "https://pve-b:8006", Instance: "lab-conn",
IsClusterMember: true, ClusterName: "lab",
},
{
ID: "node_pve-c", Name: "pve-c", Host: "https://pve-c:8006", Instance: "lab-conn",
IsClusterMember: true, ClusterName: "lab",
},
{
ID: "node_pve-d", Name: "pve-d", Host: "https://pve-d:8006", Instance: "lab-conn",
IsClusterMember: true, ClusterName: "lab",
},
{
ID: "node_pve-e", Name: "pve-e", Host: "https://10.0.0.5:8006", Instance: "lab-conn",
IsClusterMember: true, ClusterName: "lab",
},
}
h := newTestDeployHandlers(t, nodes, nil)
h.config.PVEInstances = []config.PVEInstance{{
Name: "lab-conn",
IsCluster: true,
ClusterName: "lab",
ClusterEndpoints: []config.ClusterEndpoint{
{NodeName: "pve-a", Host: "https://10.0.0.1:8006", IP: "10.0.0.1"},
{NodeName: "pve-b", Host: "https://pve-b:8006", IP: "10.0.0.2"},
{NodeName: "pve-c", Host: "https://pve-c:8006", IP: "10.0.0.3", IPOverride: "10.0.9.3"},
// pve-d has no recorded address at all.
{NodeName: "pve-d", Host: "https://pve-d:8006"},
// pve-e's API URL already holds a literal IP; that wins.
{NodeName: "pve-e", Host: "https://10.0.0.5:8006", IP: "10.0.0.55"},
},
}}
return h
}
func TestHandleCandidatesResolvesHostnameMembersToClusterAddress(t *testing.T) {
h := clusterDeployAddressFixture(t)
req := httptest.NewRequest(http.MethodGet, "/api/clusters/lab/agent-deploy/candidates", nil)
rec := httptest.NewRecorder()
h.HandleCandidates(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("expected 200, got %d. Body: %s", rec.Code, rec.Body.String())
}
var resp candidatesResponse
if err := json.NewDecoder(rec.Body).Decode(&resp); err != nil {
t.Fatalf("decode: %v", err)
}
want := map[string]struct {
ip string
deployable bool
reason string
}{
"pve-a": {ip: "10.0.0.1", deployable: false, reason: "already_agent"},
"pve-b": {ip: "10.0.0.2", deployable: true},
"pve-c": {ip: "10.0.9.3", deployable: true},
"pve-d": {ip: "", deployable: false, reason: "no_address"},
"pve-e": {ip: "10.0.0.5", deployable: true},
}
if len(resp.Nodes) != len(want) {
t.Fatalf("expected %d nodes, got %d: %+v", len(want), len(resp.Nodes), resp.Nodes)
}
for _, n := range resp.Nodes {
w, ok := want[n.Name]
if !ok {
t.Errorf("unexpected node %q", n.Name)
continue
}
if n.IP != w.ip || n.Deployable != w.deployable || n.Reason != w.reason {
t.Errorf("%s: got ip=%q deployable=%v reason=%q, want ip=%q deployable=%v reason=%q",
n.Name, n.IP, n.Deployable, n.Reason, w.ip, w.deployable, w.reason)
}
}
}
// candidateNodeID returns the unified node ID the candidates endpoint reports
// for a member name, which is what the UI sends back as a preflight target.
func candidateNodeID(t *testing.T, h *DeployHandlers, name string) string {
t.Helper()
rec := httptest.NewRecorder()
h.HandleCandidates(rec, httptest.NewRequest(http.MethodGet, "/api/clusters/lab/agent-deploy/candidates", nil))
var resp candidatesResponse
if err := json.NewDecoder(rec.Body).Decode(&resp); err != nil {
t.Fatalf("decode candidates: %v", err)
}
for _, n := range resp.Nodes {
if n.Name == name {
return n.NodeID
}
}
t.Fatalf("candidate %q not found in %+v", name, resp.Nodes)
return ""
}
func TestHandleCreatePreflightSendsClusterAddressForHostnameMember(t *testing.T) {
h := clusterDeployAddressFixture(t)
t.Cleanup(h.execServer.TestRegisterAgent("host-a", "host-a"))
body := `{"sourceAgentId":"host-a","targetNodeIds":["` + candidateNodeID(t, h, "pve-b") + `"],"maxParallel":1}`
req := httptest.NewRequest(http.MethodPost, "/api/clusters/lab/agent-deploy/preflights", strings.NewReader(body))
rec := httptest.NewRecorder()
h.HandleCreatePreflight(rec, req)
if rec.Code != http.StatusAccepted {
t.Fatalf("expected 202, got %d: %s", rec.Code, rec.Body.String())
}
var resp createPreflightResponse
if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil {
t.Fatalf("decode: %v", err)
}
targets, err := h.store.GetTargetsForJob(context.Background(), resp.PreflightID)
if err != nil {
t.Fatalf("get targets: %v", err)
}
if len(targets) != 1 {
t.Fatalf("expected 1 target, got %d", len(targets))
}
if targets[0].NodeIP != "10.0.0.2" {
t.Fatalf("target NodeIP = %q, want the cluster-reported 10.0.0.2 rather than the hostname", targets[0].NodeIP)
}
}
func TestHandleCreatePreflightRejectsMemberWithoutAddress(t *testing.T) {
h := clusterDeployAddressFixture(t)
t.Cleanup(h.execServer.TestRegisterAgent("host-a", "host-a"))
body := `{"sourceAgentId":"host-a","targetNodeIds":["` + candidateNodeID(t, h, "pve-d") + `"],"maxParallel":1}`
req := httptest.NewRequest(http.MethodPost, "/api/clusters/lab/agent-deploy/preflights", strings.NewReader(body))
rec := httptest.NewRecorder()
h.HandleCreatePreflight(rec, req)
if rec.Code != http.StatusBadRequest || !strings.Contains(rec.Body.String(), "no_valid_targets") {
t.Fatalf("expected 400 no_valid_targets, got %d: %s", rec.Code, rec.Body.String())
}
}
func TestDeployRefusesClusterNameSharedByTwoConnections(t *testing.T) {
// Two sites whose clusters happen to share a name must not be merged: the
// deploy routes are keyed on the name, so serving them would let a source
// agent at one site be pointed at the other site's member addresses.
nodes := []models.Node{
{
ID: "node_a1", Name: "pve-a", Host: "https://10.0.0.1:8006", Instance: "site-a",
IsClusterMember: true, ClusterName: "lab", LinkedAgentID: "host-a",
},
{
ID: "node_b1", Name: "pve-b", Host: "https://10.9.0.1:8006", Instance: "site-b",
IsClusterMember: true, ClusterName: "lab",
},
}
h := newTestDeployHandlers(t, nodes, nil)
t.Cleanup(h.execServer.TestRegisterAgent("host-a", "host-a"))
rec := httptest.NewRecorder()
h.HandleCandidates(rec, httptest.NewRequest(http.MethodGet, "/api/clusters/lab/agent-deploy/candidates", nil))
if rec.Code != http.StatusConflict || !strings.Contains(rec.Body.String(), "ambiguous_cluster") {
t.Fatalf("candidates: expected 409 ambiguous_cluster, got %d: %s", rec.Code, rec.Body.String())
}
body := `{"sourceAgentId":"host-a","targetNodeIds":["anything"],"maxParallel":1}`
rec = httptest.NewRecorder()
h.HandleCreatePreflight(rec, httptest.NewRequest(http.MethodPost, "/api/clusters/lab/agent-deploy/preflights", strings.NewReader(body)))
if rec.Code != http.StatusConflict || !strings.Contains(rec.Body.String(), "ambiguous_cluster") {
t.Fatalf("preflight: expected 409 ambiguous_cluster, got %d: %s", rec.Code, rec.Body.String())
}
}
func TestDeployTargetIPNeverBorrowsAnotherConnectionsEndpoint(t *testing.T) {
nodes := []models.Node{
{
ID: "node_pve-a", Name: "pve-a", Host: "https://10.0.0.1:8006", Instance: "site-a",
IsClusterMember: true, ClusterName: "lab", LinkedAgentID: "host-a",
},
{
ID: "node_pve-b", Name: "pve-b", Host: "https://pve-b:8006", Instance: "site-a",
IsClusterMember: true, ClusterName: "lab",
},
}
h := newTestDeployHandlers(t, nodes, nil)
// site-b is listed first and has a same-named cluster and member; its
// address must never be used for site-a's pve-b.
h.config.PVEInstances = []config.PVEInstance{
{
Name: "site-b", IsCluster: true, ClusterName: "lab",
ClusterEndpoints: []config.ClusterEndpoint{{NodeName: "pve-b", IP: "10.9.0.2"}},
},
{
Name: "site-a", IsCluster: true, ClusterName: "lab",
ClusterEndpoints: []config.ClusterEndpoint{{NodeName: "pve-b", IP: "10.0.0.2"}},
},
}
rec := httptest.NewRecorder()
h.HandleCandidates(rec, httptest.NewRequest(http.MethodGet, "/api/clusters/lab/agent-deploy/candidates", nil))
var resp candidatesResponse
if err := json.NewDecoder(rec.Body).Decode(&resp); err != nil {
t.Fatalf("decode: %v (body %s)", err, rec.Body.String())
}
for _, n := range resp.Nodes {
if n.Name == "pve-b" && n.IP != "10.0.0.2" {
t.Fatalf("pve-b resolved to %q, want its own connection's 10.0.0.2", n.IP)
}
}
}
func TestHandleCreateJob_RefusesWhileAnotherInstallIsRunning(t *testing.T) {
nodes := []models.Node{
{
ID: "node_pve-a", Name: "pve-a", Host: "https://10.0.0.1:8006",
IsClusterMember: true, ClusterName: "lab", LinkedAgentID: "host-a",
},
{
ID: "node_pve-b", Name: "pve-b", Host: "https://10.0.0.2:8006",
IsClusterMember: true, ClusterName: "lab",
},
}
h := newTestDeployHandlers(t, nodes, nil)
ctx := context.Background()
t.Cleanup(h.execServer.TestRegisterAgent("host-a", "host-a"))
now := time.Now().UTC()
for _, j := range []*deploy.Job{
{ID: "pf_ok", ClusterID: "lab", ClusterName: "lab", SourceAgentID: "host-a", SourceNodeID: "node_pve-a",
OrgID: "default", Status: deploy.JobSucceeded, MaxParallel: 1, CreatedAt: now, UpdatedAt: now},
// A stale install that never finished must not block anything.
{ID: "dep_stale", ClusterID: "lab", ClusterName: "lab", SourceAgentID: "host-a", SourceNodeID: "node_pve-a",
OrgID: "default", Status: deploy.JobRunning, MaxParallel: 1,
CreatedAt: now.Add(-2 * deployActiveWindow), UpdatedAt: now.Add(-2 * deployActiveWindow)},
} {
if err := h.store.CreateJob(ctx, j); err != nil {
t.Fatalf("create job %s: %v", j.ID, err)
}
}
if err := h.store.CreateTarget(ctx, &deploy.Target{
ID: "tgt_b", JobID: "pf_ok", NodeID: "node_pve-b", NodeName: "pve-b", NodeIP: "10.0.0.2",
Arch: "amd64", Status: deploy.TargetReady, CreatedAt: now, UpdatedAt: now,
}); err != nil {
t.Fatalf("create target: %v", err)
}
body := `{"sourceAgentId":"host-a","preflightId":"pf_ok","targetNodeIds":["node_pve-b"]}`
rec := httptest.NewRecorder()
h.HandleCreateJob(rec, httptest.NewRequest(http.MethodPost, "/api/clusters/lab/agent-deploy/jobs", strings.NewReader(body)))
if rec.Code != http.StatusAccepted {
t.Fatalf("first install: expected 202 despite the stale job, got %d: %s", rec.Code, rec.Body.String())
}
rec = httptest.NewRecorder()
h.HandleCreateJob(rec, httptest.NewRequest(http.MethodPost, "/api/clusters/lab/agent-deploy/jobs", strings.NewReader(body)))
if rec.Code != http.StatusConflict || !strings.Contains(rec.Body.String(), "deploy_in_progress") {
t.Fatalf("overlapping install: expected 409 deploy_in_progress, got %d: %s", rec.Code, rec.Body.String())
}
}
func TestHandleCreateJob_ConcurrentRequestsAdmitOneInstall(t *testing.T) {
nodes := []models.Node{
{
ID: "node_pve-a", Name: "pve-a", Host: "https://10.0.0.1:8006",
IsClusterMember: true, ClusterName: "lab", LinkedAgentID: "host-a",
},
{
ID: "node_pve-b", Name: "pve-b", Host: "https://10.0.0.2:8006",
IsClusterMember: true, ClusterName: "lab",
},
}
h := newTestDeployHandlers(t, nodes, nil)
ctx := context.Background()
t.Cleanup(h.execServer.TestRegisterAgent("host-a", "host-a"))
now := time.Now().UTC()
if err := h.store.CreateJob(ctx, &deploy.Job{
ID: "pf_ok", ClusterID: "lab", ClusterName: "lab", SourceAgentID: "host-a", SourceNodeID: "node_pve-a",
OrgID: "default", Status: deploy.JobSucceeded, MaxParallel: 1, CreatedAt: now, UpdatedAt: now,
}); err != nil {
t.Fatalf("create preflight: %v", err)
}
if err := h.store.CreateTarget(ctx, &deploy.Target{
ID: "tgt_b", JobID: "pf_ok", NodeID: "node_pve-b", NodeName: "pve-b", NodeIP: "10.0.0.2",
Arch: "amd64", Status: deploy.TargetReady, CreatedAt: now, UpdatedAt: now,
}); err != nil {
t.Fatalf("create target: %v", err)
}
body := `{"sourceAgentId":"host-a","preflightId":"pf_ok","targetNodeIds":["node_pve-b"]}`
const requests = 8
codes := make(chan int, requests)
var wg sync.WaitGroup
for i := 0; i < requests; i++ {
wg.Add(1)
go func() {
defer wg.Done()
rec := httptest.NewRecorder()
h.HandleCreateJob(rec, httptest.NewRequest(http.MethodPost, "/api/clusters/lab/agent-deploy/jobs", strings.NewReader(body)))
codes <- rec.Code
}()
}
wg.Wait()
close(codes)
accepted, refused := 0, 0
for code := range codes {
switch code {
case http.StatusAccepted:
accepted++
case http.StatusConflict:
refused++
default:
t.Errorf("unexpected status %d", code)
}
}
if accepted != 1 || refused != requests-1 {
t.Fatalf("accepted=%d refused=%d, want exactly one install admitted", accepted, refused)
}
}
+25
View File
@@ -211,6 +211,31 @@ func (s *Store) ListJobs(ctx context.Context, orgID string, limit int) ([]Job, e
return jobs, rows.Err()
}
// UnfinishedJobsForCluster returns the organization's jobs on one cluster that
// are still queued, waiting for their source, running, or canceling, newest
// first. Unlike ListJobs it has no row limit, so a busy organization's newer
// preflights cannot push an unfinished install out of view.
func (s *Store) UnfinishedJobsForCluster(ctx context.Context, orgID, clusterID string) ([]Job, error) {
rows, err := s.db.QueryContext(ctx,
`SELECT id, cluster_id, cluster_name, source_agent_id, source_node_id, org_id, status, max_parallel, retry_max, created_at, updated_at, completed_at
FROM deploy_jobs WHERE org_id = ? AND cluster_id = ? AND status IN (?, ?, ?, ?) ORDER BY created_at DESC`,
orgID, clusterID, string(JobQueued), string(JobWaitingSource), string(JobRunning), string(JobCanceling))
if err != nil {
return nil, err
}
defer rows.Close()
var jobs []Job
for rows.Next() {
j, err := scanJobRow(rows)
if err != nil {
return nil, err
}
jobs = append(jobs, j)
}
return jobs, rows.Err()
}
// --- Targets ---
// GetTarget retrieves a single deployment target by ID.
+9
View File
@@ -36,6 +36,15 @@ func TestDeployStoreQueryPlansUseIndexes(t *testing.T) {
args: []any{"org-03", 50},
wantIndex: "idx_deploy_jobs_org_created",
},
{
name: "unfinished jobs for cluster uses org created index",
query: `
SELECT id, cluster_id, cluster_name, source_agent_id, source_node_id, org_id, status, max_parallel, retry_max, created_at, updated_at, completed_at
FROM deploy_jobs WHERE org_id = ? AND cluster_id = ? AND status IN (?, ?, ?, ?) ORDER BY created_at DESC
`,
args: []any{"org-03", "cluster-01", "queued", "waiting_source", "running", "canceling"},
wantIndex: "idx_deploy_jobs_org_created",
},
{
name: "get targets for job uses job created index",
query: `
+38
View File
@@ -557,3 +557,41 @@ func TestResetTargetsForRetry_EmptyList(t *testing.T) {
t.Fatalf("expected 0, got %d", count)
}
}
func TestUnfinishedJobsForClusterFiltersByOrgClusterAndStatus(t *testing.T) {
s := testStore(t)
ctx := context.Background()
now := time.Now().UTC().Truncate(time.Second)
for i, j := range []struct {
id, org, cluster string
status JobStatus
}{
{"dep-running", "org-1", "lab", JobRunning},
{"dep-queued", "org-1", "lab", JobQueued},
{"dep-done", "org-1", "lab", JobSucceeded},
{"dep-other-cluster", "org-1", "prod", JobRunning},
{"dep-other-org", "org-2", "lab", JobRunning},
} {
created := now.Add(time.Duration(i) * time.Second)
if err := s.CreateJob(ctx, &Job{
ID: j.id, ClusterID: j.cluster, ClusterName: j.cluster, SourceAgentID: "agent-1",
SourceNodeID: "node-1", OrgID: j.org, Status: j.status, MaxParallel: 1,
CreatedAt: created, UpdatedAt: created,
}); err != nil {
t.Fatalf("CreateJob %s: %v", j.id, err)
}
}
jobs, err := s.UnfinishedJobsForCluster(ctx, "org-1", "lab")
if err != nil {
t.Fatalf("UnfinishedJobsForCluster: %v", err)
}
var ids []string
for _, j := range jobs {
ids = append(ids, j.ID)
}
if len(ids) != 2 || ids[0] != "dep-queued" || ids[1] != "dep-running" {
t.Fatalf("got %v, want [dep-queued dep-running] (newest first, other org/cluster/finished excluded)", ids)
}
}