wulf-pulse/lib/services/pipeline-steps/enrich-vspc.ts

195 lines
7.8 KiB
TypeScript

/**
* Step Executor: enrich_vspc
* Queries Veeam VSPC (API + local DB) for backup status of a device.
*
* Config:
* lookup_by: 'device_name' | 'organization_uid'
* source_field: template string for the lookup value
*
* Outputs to context:
* vspc_found, vspc_agent_jobs, vspc_server_jobs, vspc_workloads,
* vspc_last_job_status, vspc_last_success, vspc_failure_message,
* vspc_restore_points, vspc_backed_up_size, vspc_free_space,
* vspc_alarms, vspc_summary
*/
import { registerStepExecutor } from '../pipeline-engine';
import { postgresClient } from '../postgres-client';
import { StepExecutorResult, PipelineStep, PipelineContext } from '../../types/pipeline';
registerStepExecutor('enrich_vspc', async (
step: PipelineStep,
context: PipelineContext,
executionId: number
): Promise<StepExecutorResult> => {
const config = step.config as {
lookup_by?: string;
source_field?: string;
};
const lookupBy = config.lookup_by || 'device_name';
const sourceValue = config.source_field || '';
if (!sourceValue) {
return { success: false, output: { vspc_found: false }, error: 'source_field is required' };
}
try {
// ---- 1. Query local DB for backup agent jobs matching this device ----
let agentJobs: any[] = [];
let serverJobs: any[] = [];
let workloads: any[] = [];
let alarms: any[] = [];
if (lookupBy === 'device_name') {
// Match by agent name (agent name often = hostname)
const agentResult = await postgresClient.query(
`SELECT baj.*, ba.name as agent_name, ba.status as agent_status,
ba.agent_platform, ba.version as agent_version,
vo.name as org_name, vo.company_id
FROM veeam_backup_agent_jobs baj
LEFT JOIN veeam_backup_agents ba ON ba.instance_uid = baj.backup_agent_uid
LEFT JOIN veeam_organizations vo ON vo.instance_uid = baj.organization_uid
WHERE LOWER(ba.name) LIKE LOWER($1)
OR LOWER(baj.name) LIKE LOWER($1)
ORDER BY baj.last_run DESC NULLS LAST`,
[`%${sourceValue}%`]
);
agentJobs = agentResult.rows;
// Check protected workloads (VM-level)
const workloadResult = await postgresClient.query(
`SELECT pw.*, bj.name as job_name, bj.status as job_status,
bj.last_run as job_last_run, bj.failure_message as job_failure_message
FROM veeam_protected_workloads pw
LEFT JOIN veeam_backup_jobs bj ON bj.instance_uid = pw.job_uid
WHERE LOWER(pw.name) LIKE LOWER($1)
ORDER BY pw.latest_restore_point_date DESC NULLS LAST`,
[`%${sourceValue}%`]
);
workloads = workloadResult.rows;
// Check backup server jobs that might reference this device
const serverJobResult = await postgresClient.query(
`SELECT bj.*, vo.name as org_name, vo.company_id
FROM veeam_backup_jobs bj
LEFT JOIN veeam_organizations vo ON vo.instance_uid = bj.organization_uid
WHERE LOWER(bj.name) LIKE LOWER($1)
OR LOWER(bj.destination) LIKE LOWER($1)
ORDER BY bj.last_run DESC NULLS LAST`,
[`%${sourceValue}%`]
);
serverJobs = serverJobResult.rows;
// Check active alarms for this device
const alarmResult = await postgresClient.query(
`SELECT * FROM veeam_alarms
WHERE LOWER(object_name) LIKE LOWER($1)
OR LOWER(object_computer_name) LIKE LOWER($1)
ORDER BY last_activation_time DESC NULLS LAST`,
[`%${sourceValue}%`]
);
alarms = alarmResult.rows;
}
// ---- 2. Build summary ----
const allJobs = [...agentJobs, ...serverJobs];
const latestJob = allJobs[0] || null;
const failedJobs = allJobs.filter(j => j.status === 'Failed');
const warningJobs = allJobs.filter(j => j.status === 'Warning');
const successJobs = allJobs.filter(j => j.status === 'Success');
// Find last successful backup across all job types
const lastSuccess = allJobs.find(j => j.status === 'Success');
const lastSuccessDate = lastSuccess?.last_run || lastSuccess?.last_end_time || null;
// Calculate hours since last success
let hoursSinceSuccess: number | null = null;
if (lastSuccessDate) {
hoursSinceSuccess = Math.round((Date.now() - new Date(lastSuccessDate).getTime()) / 3600000);
}
// Aggregate restore points and sizes
const totalRestorePoints = agentJobs.reduce((sum, j) => sum + (j.restore_points || 0), 0)
+ workloads.reduce((sum, w) => sum + (w.restore_points || 0), 0);
const totalBackedUpSize = agentJobs.reduce((sum, j) => sum + (j.backed_up_size || 0), 0);
const summary = [
`Device: ${sourceValue}`,
`Agent Jobs: ${agentJobs.length} (${successJobs.length} success, ${failedJobs.length} failed, ${warningJobs.length} warning)`,
`Server Jobs: ${serverJobs.length}`,
`Protected Workloads: ${workloads.length}`,
`Active Alarms: ${alarms.length}`,
`Total Restore Points: ${totalRestorePoints}`,
latestJob ? `Latest Job: "${latestJob.name}" — ${latestJob.status} at ${latestJob.last_run || 'never'}` : 'No jobs found',
latestJob?.failure_message ? `Failure: ${latestJob.failure_message}` : null,
lastSuccessDate ? `Last Success: ${lastSuccessDate} (${hoursSinceSuccess}h ago)` : 'No successful backup found',
alarms.length > 0 ? `Alarms: ${alarms.map((a: any) => `${a.alarm_area}: ${a.last_activation_message}`).join('; ')}` : null,
].filter(Boolean).join('\n');
const found = agentJobs.length > 0 || serverJobs.length > 0 || workloads.length > 0;
return {
success: true,
output: {
vspc_found: found,
vspc_agent_jobs: agentJobs.map(j => ({
name: j.name,
status: j.status,
last_run: j.last_run,
last_end_time: j.last_end_time,
last_duration: j.last_duration,
failure_message: j.failure_message,
backup_mode: j.backup_mode,
destination: j.destination,
restore_points: j.restore_points,
backed_up_size: j.backed_up_size,
free_space: j.free_space,
is_enabled: j.is_enabled,
agent_name: j.agent_name,
agent_status: j.agent_status,
org_name: j.org_name,
})),
vspc_server_jobs: serverJobs.map(j => ({
name: j.name,
status: j.status,
last_run: j.last_run,
failure_message: j.failure_message,
type: j.type,
destination: j.destination,
bottleneck: j.bottleneck,
backup_chain_size: j.backup_chain_size,
})),
vspc_workloads: workloads.map(w => ({
name: w.name,
restore_points: w.restore_points,
latest_restore_point_date: w.latest_restore_point_date,
latest_restore_point_size: w.latest_restore_point_size,
malware_state: w.malware_state,
job_name: w.job_name,
job_status: w.job_status,
})),
vspc_alarms: alarms.map(a => ({
area: a.alarm_area,
message: a.last_activation_message,
status: a.last_activation_status,
time: a.last_activation_time,
object_name: a.object_name,
})),
vspc_last_job_status: latestJob?.status || null,
vspc_last_success: lastSuccessDate,
vspc_hours_since_success: hoursSinceSuccess,
vspc_failure_message: latestJob?.failure_message || null,
vspc_restore_points: totalRestorePoints,
vspc_backed_up_size: totalBackedUpSize,
vspc_alarm_count: alarms.length,
vspc_failed_job_count: failedJobs.length,
vspc_summary: summary,
},
};
} catch (error) {
const msg = error instanceof Error ? error.message : String(error);
return { success: false, output: { vspc_found: false }, error: `VSPC enrichment failed: ${msg}` };
}
});