diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 46910f9b59..1c23a58a5e 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -750,6 +750,29 @@ must never prevent the remaining configured hosts from being suppressed. `TestGracePeriodGuestContributesNoMemorySample` in `internal/monitoring/memory_trust_characterization_test.go`. +13. Multi-guest vzdump job runs must surface per-guest task status. A + scheduled backup job executes under one UPID whose VMID slot is empty, + so the task listing alone cannot say which guests it covered and the + guest-centric coverage surfaces would show status only for guests + backed up individually. `pollBackupTasks` therefore expands each + job-run task by fetching its task log (`GetTaskLog` on the PVE client, + with cluster failover) and parsing the per-guest markers into + synthetic `BackupTask` entries whose IDs embed the parent UPID. The + task listing queries `source=all` so running jobs are visible: a guest + an in-progress job is currently backing up carries a `running` + synthetic task, which is valid backup-intent evidence for + `resolveBackupIntentContext`. Finished job logs are immutable and are + fetched at most once per instance|UPID, with a per-cycle fetch cap so + a historical backlog cannot stall the shared backup poll budget. + Regression coverage: `TestParseVzdumpJobLogFinishedJob`, + `TestPollBackupTasksSynthesizesJobGuestTasks`, and + `TestPollBackupTasksRunningJobRefetchesAndSuppressesAlerts` in + `internal/monitoring/monitor_backup_job_tasks_test.go`, + `TestResolveBackupIntentContextAcceptsSynthesizedJobGuestTask` in + `internal/monitoring/monitor_alert_intent_test.go`, and + `TestClusterClient_GetTaskLog` in + `pkg/proxmox/cluster_client_api_test.go`. + ## Current State diff --git a/internal/monitoring/harness_integration.go b/internal/monitoring/harness_integration.go index 62012837a9..179f05a661 100644 --- a/internal/monitoring/harness_integration.go +++ b/internal/monitoring/harness_integration.go @@ -398,6 +398,9 @@ func (noopPVEClient) GetStorage(ctx context.Context, node string) ([]proxmox.Sto } func (noopPVEClient) GetAllStorage(ctx context.Context) ([]proxmox.Storage, error) { return nil, nil } func (noopPVEClient) GetBackupTasks(ctx context.Context) ([]proxmox.Task, error) { return nil, nil } +func (noopPVEClient) GetTaskLog(ctx context.Context, node, upid string) ([]proxmox.TaskLogLine, error) { + return nil, nil +} func (noopPVEClient) GetReplicationStatus(ctx context.Context) ([]proxmox.ReplicationJob, error) { return nil, nil } diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 2d3e7bca8d..b039c0f663 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -77,6 +77,7 @@ type PVEClientInterface interface { GetStorage(ctx context.Context, node string) ([]proxmox.Storage, error) GetAllStorage(ctx context.Context) ([]proxmox.Storage, error) GetBackupTasks(ctx context.Context) ([]proxmox.Task, error) + GetTaskLog(ctx context.Context, node, upid string) ([]proxmox.TaskLogLine, error) GetReplicationStatus(ctx context.Context) ([]proxmox.ReplicationJob, error) GetStorageContent(ctx context.Context, node, storage string) ([]proxmox.StorageContent, error) GetVMSnapshots(ctx context.Context, node string, vmid int) ([]proxmox.Snapshot, error) @@ -1099,6 +1100,7 @@ type Monitor struct { lastClusterCheck map[string]time.Time // Track last cluster check for standalone nodes lastPhysicalDiskPoll map[string]time.Time // Track last physical disk poll time per instance lastPVEBackupPoll map[string]time.Time // Track last PVE backup poll per instance + vzdumpJobTaskCache map[string]vzdumpJobTaskCacheEntry // Cache synthesized per-guest tasks for multi-guest vzdump job runs, keyed by instance|UPID lastPBSBackupPoll map[string]time.Time // Track last PBS backup poll per instance pveBackupInventoryReady map[string]map[string]bool // Track PVE guest inventory readiness for backup orphan detection pveBackupTemplateSubjects map[string]map[string]struct{} // Track template VMIDs excluded from runtime workloads but valid for backups @@ -1740,6 +1742,7 @@ func New(cfg *config.Config) (*Monitor, error) { lastClusterCheck: make(map[string]time.Time), lastPhysicalDiskPoll: make(map[string]time.Time), lastPVEBackupPoll: make(map[string]time.Time), + vzdumpJobTaskCache: make(map[string]vzdumpJobTaskCacheEntry), lastPBSBackupPoll: make(map[string]time.Time), pveBackupInventoryReady: make(map[string]map[string]bool), pveBackupTemplateSubjects: make(map[string]map[string]struct{}), diff --git a/internal/monitoring/monitor_alert_intent_test.go b/internal/monitoring/monitor_alert_intent_test.go index 6d4b820524..080f233e2f 100644 --- a/internal/monitoring/monitor_alert_intent_test.go +++ b/internal/monitoring/monitor_alert_intent_test.go @@ -5,6 +5,7 @@ import ( "time" "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/pkg/proxmox" ) func TestResolveBackupIntentContextRequiresFreshActiveMatchingEvidence(t *testing.T) { @@ -68,3 +69,41 @@ func TestResolveBackupIntentContextRequiresFreshActiveMatchingEvidence(t *testin }) } } + +// A guest covered by a running multi-guest vzdump job has no task of its own; +// pollBackupTasks synthesizes one from the job log. That synthetic task must +// count as backup-intent evidence so the guest's alerts are suppressed while +// the job is backing it up, and must stop counting once its section finishes. +func TestResolveBackupIntentContextAcceptsSynthesizedJobGuestTask(t *testing.T) { + now := time.Date(2026, 7, 25, 2, 5, 0, 0, time.UTC) + jobTask := models.BackupTask{ + ID: "pve-a-UPID:node-a:000E9F2C:0AC734B2:68A1B2C3:vzdump::root@pam:", + Node: "node-a", + Instance: "pve-a", + Type: "vzdump", + StartTime: now.Add(-4 * time.Minute), + // no EndTime: the job is still running + } + + synthesized := parseVzdumpJobLog(jobTask, "UPID:node-a:000E9F2C:0AC734B2:68A1B2C3:vzdump::root@pam:", []proxmox.TaskLogLine{ + {LineNumber: 1, Text: "INFO: Starting Backup of VM 101 (qemu)"}, + {LineNumber: 2, Text: "INFO: Finished Backup of VM 101 (00:01:30)"}, + {LineNumber: 3, Text: "INFO: Starting Backup of VM 102 (lxc)"}, + }) + for i := range synthesized { + synthesized[i].ObservedAt = now.Add(-30 * time.Second) + } + + state := models.NewState() + state.UpdateBackupTasksForInstance("pve-a", append([]models.BackupTask{jobTask}, synthesized...)) + monitor := &Monitor{state: state} + + context, found := monitor.resolveBackupIntentContext("", "pve-a", "node-a", 102, now) + if !found || !context.Active { + t.Fatalf("guest being backed up by running job did not resolve: found=%v context=%+v", found, context) + } + + if got, ok := monitor.resolveBackupIntentContext("", "pve-a", "node-a", 101, now); ok { + t.Fatalf("guest whose job section already finished should not carry intent: %+v", got) + } +} diff --git a/internal/monitoring/monitor_backup_job_tasks_test.go b/internal/monitoring/monitor_backup_job_tasks_test.go new file mode 100644 index 0000000000..b475e623e1 --- /dev/null +++ b/internal/monitoring/monitor_backup_job_tasks_test.go @@ -0,0 +1,283 @@ +package monitoring + +import ( + "context" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/pkg/proxmox" +) + +const testJobUPID = "UPID:pve1:000E9F2C:0AC734B2:68A1B2C3:vzdump::root@pam:" + +func testLogLines(lines ...string) []proxmox.TaskLogLine { + out := make([]proxmox.TaskLogLine, 0, len(lines)) + for i, line := range lines { + out = append(out, proxmox.TaskLogLine{LineNumber: int64(i + 1), Text: line}) + } + return out +} + +func TestParseVzdumpJobLogFinishedJob(t *testing.T) { + start := time.Date(2026, 7, 25, 2, 0, 0, 0, time.UTC) + jobTask := models.BackupTask{ + ID: "pve-cluster-" + testJobUPID, + Node: "pve1", + Instance: "pve-cluster", + Type: "vzdump", + Status: "job errors", + StartTime: start, + EndTime: start.Add(10 * time.Minute), + } + + tasks := parseVzdumpJobLog(jobTask, testJobUPID, testLogLines( + "INFO: starting new backup job: vzdump 101 102 103 --mode snapshot --storage pbs", + "INFO: Starting Backup of VM 101 (qemu)", + "INFO: creating Proxmox Backup Server archive 'vm/101/2026-07-25T02:00:01Z'", + "INFO: Finished Backup of VM 101 (00:02:30)", + "INFO: Starting Backup of VM 102 (lxc)", + "ERROR: Backup of VM 102 failed - command 'rsync' failed: exit code 23", + "INFO: Starting Backup of VM 103 (qemu)", + "INFO: Finished Backup of VM 103 (00:01:00)", + "INFO: Backup job finished with errors", + )) + + if len(tasks) != 3 { + t.Fatalf("expected 3 synthetic guest tasks, got %d", len(tasks)) + } + + vm101 := tasks[0] + if vm101.VMID != 101 || vm101.Status != "OK" || vm101.Error != "" { + t.Errorf("vm101 = %+v, want VMID 101 status OK with no error", vm101) + } + if vm101.ID != "pve-cluster-"+testJobUPID+"-vm101" { + t.Errorf("vm101 ID = %q, want parent UPID embedded", vm101.ID) + } + if !vm101.StartTime.Equal(start) || !vm101.EndTime.Equal(start.Add(2*time.Minute+30*time.Second)) { + t.Errorf("vm101 times = %v..%v, want job start plus printed duration", vm101.StartTime, vm101.EndTime) + } + + vm102 := tasks[1] + if vm102.VMID != 102 || vm102.Status != "error" { + t.Errorf("vm102 = %+v, want VMID 102 status error", vm102) + } + if vm102.Error != "command 'rsync' failed: exit code 23" { + t.Errorf("vm102 error = %q, want the reason from the log line", vm102.Error) + } + if !vm102.EndTime.Equal(jobTask.EndTime) { + t.Errorf("vm102 end = %v, want parent job end for failed guest", vm102.EndTime) + } + + vm103 := tasks[2] + if vm103.VMID != 103 || vm103.Status != "OK" { + t.Errorf("vm103 = %+v, want VMID 103 status OK", vm103) + } + // vm102 failed without a duration line, so vm103 starts from vm101's end. + wantStart := start.Add(2*time.Minute + 30*time.Second) + if !vm103.StartTime.Equal(wantStart) || !vm103.EndTime.Equal(wantStart.Add(time.Minute)) { + t.Errorf("vm103 times = %v..%v, want %v..%v", vm103.StartTime, vm103.EndTime, wantStart, wantStart.Add(time.Minute)) + } + + for _, task := range tasks { + if task.Node != "pve1" || task.Instance != "pve-cluster" || task.Type != "vzdump" { + t.Errorf("task %d inherits wrong parent metadata: %+v", task.VMID, task) + } + } +} + +func TestParseVzdumpJobLogRunningJob(t *testing.T) { + start := time.Date(2026, 7, 25, 2, 0, 0, 0, time.UTC) + jobTask := models.BackupTask{ + ID: "pve-cluster-" + testJobUPID, + Node: "pve1", + Instance: "pve-cluster", + Type: "vzdump", + StartTime: start, + } + + tasks := parseVzdumpJobLog(jobTask, testJobUPID, testLogLines( + "INFO: Starting Backup of VM 101 (qemu)", + "INFO: Finished Backup of VM 101 (00:00:45)", + "INFO: Starting Backup of VM 102 (lxc)", + )) + + if len(tasks) != 2 { + t.Fatalf("expected 2 synthetic guest tasks, got %d", len(tasks)) + } + if tasks[0].Status != "OK" || tasks[0].EndTime.IsZero() { + t.Errorf("finished guest in running job = %+v, want status OK with end time", tasks[0]) + } + if tasks[1].Status != "running" || !tasks[1].EndTime.IsZero() { + t.Errorf("in-progress guest = %+v, want status running with zero end time", tasks[1]) + } +} + +func TestParseVzdumpJobLogJobDiedMidGuest(t *testing.T) { + start := time.Date(2026, 7, 25, 2, 0, 0, 0, time.UTC) + jobTask := models.BackupTask{ + ID: "pve-cluster-" + testJobUPID, + Node: "pve1", + Instance: "pve-cluster", + Type: "vzdump", + Status: "unexpected status", + StartTime: start, + EndTime: start.Add(time.Minute), + } + + tasks := parseVzdumpJobLog(jobTask, testJobUPID, testLogLines( + "INFO: Starting Backup of VM 101 (qemu)", + )) + + if len(tasks) != 1 { + t.Fatalf("expected 1 synthetic guest task, got %d", len(tasks)) + } + if tasks[0].Status != "error" || tasks[0].Error == "" { + t.Errorf("guest in dead job = %+v, want status error with message", tasks[0]) + } + if !tasks[0].EndTime.Equal(jobTask.EndTime) { + t.Errorf("guest end = %v, want parent end time", tasks[0].EndTime) + } +} + +func TestParseVzdumpJobLogNoGuestMarkers(t *testing.T) { + jobTask := models.BackupTask{ + ID: "pve-cluster-" + testJobUPID, + Instance: "pve-cluster", + Type: "vzdump", + StartTime: time.Now(), + } + if tasks := parseVzdumpJobLog(jobTask, testJobUPID, testLogLines("INFO: nothing to do")); tasks != nil { + t.Errorf("expected no synthetic tasks, got %+v", tasks) + } +} + +// jobBackupTaskClient serves a multi-guest vzdump job task plus its log. +// Only the two methods pollBackupTasks uses are implemented; the embedded +// interface satisfies the rest. +type jobBackupTaskClient struct { + PVEClientInterface + task proxmox.Task + logLines []proxmox.TaskLogLine + logRequests int +} + +func (c *jobBackupTaskClient) GetBackupTasks(ctx context.Context) ([]proxmox.Task, error) { + return []proxmox.Task{c.task}, nil +} + +func (c *jobBackupTaskClient) GetTaskLog(ctx context.Context, node, upid string) ([]proxmox.TaskLogLine, error) { + c.logRequests++ + return c.logLines, nil +} + +func TestPollBackupTasksSynthesizesJobGuestTasks(t *testing.T) { + start := time.Now().Add(-time.Hour) + client := &jobBackupTaskClient{ + task: proxmox.Task{ + UPID: testJobUPID, + Node: "pve1", + Type: "vzdump", + ID: "", // job-run UPIDs carry no VMID + Status: "OK", + StartTime: start.Unix(), + EndTime: start.Add(5 * time.Minute).Unix(), + }, + logLines: testLogLines( + "INFO: Starting Backup of VM 101 (qemu)", + "INFO: Finished Backup of VM 101 (00:02:00)", + "INFO: Starting Backup of VM 102 (lxc)", + "INFO: Finished Backup of VM 102 (00:01:00)", + ), + } + + m := &Monitor{state: models.NewState()} + m.pollBackupTasks(context.Background(), "pve-test", client) + + state := m.state.GetSnapshot() + if len(state.PVEBackups.BackupTasks) != 3 { + t.Fatalf("expected job task plus 2 synthetic guest tasks, got %d: %+v", len(state.PVEBackups.BackupTasks), state.PVEBackups.BackupTasks) + } + + byVMID := make(map[int]models.BackupTask) + for _, task := range state.PVEBackups.BackupTasks { + byVMID[task.VMID] = task + } + for _, vmid := range []int{101, 102} { + task, ok := byVMID[vmid] + if !ok { + t.Fatalf("no synthetic task for VM %d", vmid) + } + if task.Status != "OK" { + t.Errorf("VM %d status = %q, want OK", vmid, task.Status) + } + if task.ObservedAt.IsZero() { + t.Errorf("VM %d has no observation time", vmid) + } + } + if _, ok := byVMID[0]; !ok { + t.Error("parent job task should remain in state") + } + + if client.logRequests != 1 { + t.Fatalf("expected 1 log fetch, got %d", client.logRequests) + } + + // A finished job's log is immutable: the second poll must serve the + // synthetic tasks from cache without refetching, refreshing ObservedAt. + m.pollBackupTasks(context.Background(), "pve-test", client) + if client.logRequests != 1 { + t.Errorf("expected cached synthesis on second poll, got %d log fetches", client.logRequests) + } + state = m.state.GetSnapshot() + if len(state.PVEBackups.BackupTasks) != 3 { + t.Errorf("expected stable task count after second poll, got %d", len(state.PVEBackups.BackupTasks)) + } +} + +func TestPollBackupTasksRunningJobRefetchesAndSuppressesAlerts(t *testing.T) { + start := time.Now().Add(-2 * time.Minute) + client := &jobBackupTaskClient{ + task: proxmox.Task{ + UPID: testJobUPID, + Node: "pve1", + Type: "vzdump", + ID: "", + StartTime: start.Unix(), + // running: no EndTime, no Status + }, + logLines: testLogLines( + "INFO: Starting Backup of VM 101 (qemu)", + ), + } + + m := &Monitor{state: models.NewState()} + m.pollBackupTasks(context.Background(), "pve-test", client) + m.pollBackupTasks(context.Background(), "pve-test", client) + + if client.logRequests != 2 { + t.Errorf("running job should be re-parsed every poll, got %d log fetches", client.logRequests) + } + + state := m.state.GetSnapshot() + var running *models.BackupTask + for i := range state.PVEBackups.BackupTasks { + if state.PVEBackups.BackupTasks[i].VMID == 101 { + running = &state.PVEBackups.BackupTasks[i] + } + } + if running == nil { + t.Fatal("no synthetic task for VM 101") + } + if running.Status != "running" { + t.Errorf("status = %q, want running", running.Status) + } + + intent, ok := m.resolveBackupIntentContext("", "pve-test", "pve1", 101, time.Now()) + if !ok || !intent.Active { + t.Errorf("expected active backup intent for guest covered by running job, got ok=%v intent=%+v", ok, intent) + } + if _, ok := m.resolveBackupIntentContext("", "pve-test", "pve1", 999, time.Now()); ok { + t.Error("guest not covered by the job should have no backup intent") + } +} diff --git a/internal/monitoring/monitor_backups.go b/internal/monitoring/monitor_backups.go index 21da106c0a..69e97ed302 100644 --- a/internal/monitoring/monitor_backups.go +++ b/internal/monitoring/monitor_backups.go @@ -3,6 +3,7 @@ package monitoring import ( "context" "fmt" + "regexp" "sort" "strconv" "strings" @@ -1981,6 +1982,8 @@ func (m *Monitor) pollBackupTasks(ctx context.Context, instanceName string, clie } var backupTasks []models.BackupTask + var jobTasks []models.BackupTask + jobUPIDs := make(map[string]string) for _, task := range tasks { // Extract VMID from task ID (format: "UPID:node:pid:starttime:type:vmid:user@realm:") vmid := 0 @@ -2008,8 +2011,18 @@ func (m *Monitor) pollBackupTasks(ctx context.Context, instanceName string, clie } backupTasks = append(backupTasks, backupTask) + + // Scheduled multi-guest backup jobs run under a single UPID with an + // empty VMID slot, so the task alone says nothing about which guests + // it covered (#1452 area, regressed with the guest-centric redesign). + if vmid == 0 && task.Type == "vzdump" && task.UPID != "" { + jobUPIDs[backupTask.ID] = task.UPID + jobTasks = append(jobTasks, backupTask) + } } + backupTasks = append(backupTasks, m.synthesizeVzdumpJobGuestTasks(ctx, instanceName, client, jobTasks, jobUPIDs)...) + // Update state with new backup tasks for this instance m.state.UpdateBackupTasksForInstance(instanceName, backupTasks) @@ -2018,6 +2031,226 @@ func (m *Monitor) pollBackupTasks(ctx context.Context, instanceName string, clie m.ingestRecoveryPointsAsync(proxmoxrecoverymapper.FromPVEBackupTasks(backupTasks, guestInfo)) } +// vzdumpJobTaskCacheEntry caches the per-guest tasks synthesized from one +// vzdump job run's log so finished jobs are fetched at most once. +type vzdumpJobTaskCacheEntry struct { + tasks []models.BackupTask + final bool +} + +// maxVzdumpJobLogFetchesPerPoll bounds task-log requests per poll cycle so a +// large backlog of historical job runs (e.g. right after startup) trickles in +// over a few cycles instead of stalling the shared backup-poll budget. +const maxVzdumpJobLogFetchesPerPoll = 25 + +// synthesizeVzdumpJobGuestTasks expands multi-guest vzdump job runs into one +// synthetic per-guest task each by parsing the job's task log. Results are +// cached by instance|UPID; logs of finished jobs are immutable, so those are +// fetched only once. Running jobs are re-parsed every cycle to pick up guests +// as the job progresses. +func (m *Monitor) synthesizeVzdumpJobGuestTasks(ctx context.Context, instanceName string, client PVEClientInterface, jobTasks []models.BackupTask, jobUPIDs map[string]string) []models.BackupTask { + if len(jobTasks) == 0 { + m.pruneVzdumpJobTaskCache(instanceName, nil) + return nil + } + + now := time.Now().UTC() + seen := make(map[string]struct{}, len(jobTasks)) + var synthesized []models.BackupTask + fetches := 0 + + for _, jobTask := range jobTasks { + upid := jobUPIDs[jobTask.ID] + key := instanceName + "|" + upid + seen[key] = struct{}{} + + m.mu.RLock() + entry, cached := m.vzdumpJobTaskCache[key] + m.mu.RUnlock() + + if cached && entry.final { + synthesized = append(synthesized, refreshObservedAt(entry.tasks, now)...) + continue + } + + if fetches >= maxVzdumpJobLogFetchesPerPoll || ctx.Err() != nil { + // Serve a stale non-final entry if we have one; the next cycle + // will catch up. + if cached { + synthesized = append(synthesized, refreshObservedAt(entry.tasks, now)...) + } + continue + } + fetches++ + + logLines, err := client.GetTaskLog(ctx, jobTask.Node, upid) + if err != nil { + log.Debug().Err(err). + Str("instance", instanceName). + Str("upid", upid). + Msg("failed to fetch vzdump job task log; per-guest status unavailable this cycle") + if cached { + synthesized = append(synthesized, refreshObservedAt(entry.tasks, now)...) + } + continue + } + + guestTasks := parseVzdumpJobLog(jobTask, upid, logLines) + m.mu.Lock() + if m.vzdumpJobTaskCache == nil { + m.vzdumpJobTaskCache = make(map[string]vzdumpJobTaskCacheEntry) + } + m.vzdumpJobTaskCache[key] = vzdumpJobTaskCacheEntry{ + tasks: guestTasks, + final: !jobTask.EndTime.IsZero(), + } + m.mu.Unlock() + synthesized = append(synthesized, refreshObservedAt(guestTasks, now)...) + } + + m.pruneVzdumpJobTaskCache(instanceName, seen) + return synthesized +} + +// pruneVzdumpJobTaskCache drops cached job entries for this instance whose +// UPIDs are no longer present in the task list (fell out of the window). +func (m *Monitor) pruneVzdumpJobTaskCache(instanceName string, seen map[string]struct{}) { + prefix := instanceName + "|" + m.mu.Lock() + for key := range m.vzdumpJobTaskCache { + if !strings.HasPrefix(key, prefix) { + continue + } + if _, ok := seen[key]; !ok { + delete(m.vzdumpJobTaskCache, key) + } + } + m.mu.Unlock() +} + +func refreshObservedAt(tasks []models.BackupTask, now time.Time) []models.BackupTask { + if len(tasks) == 0 { + return nil + } + out := make([]models.BackupTask, len(tasks)) + copy(out, tasks) + for i := range out { + out[i].ObservedAt = now + } + return out +} + +// vzdump job logs mark each guest's section with these lines. +var ( + vzdumpLogGuestStartRe = regexp.MustCompile(`^INFO: Starting Backup of VM (\d+)`) + vzdumpLogGuestFinishRe = regexp.MustCompile(`^INFO: Finished Backup of VM (\d+) \((\d+):(\d{2}):(\d{2})\)$`) + vzdumpLogGuestFailRe = regexp.MustCompile(`^ERROR: Backup of VM (\d+) failed -\s*(.*)$`) +) + +// parseVzdumpJobLog turns a multi-guest vzdump job run into synthetic +// per-guest BackupTask entries. Log lines carry no timestamps, so per-guest +// times are reconstructed from the job start plus the durations vzdump prints +// on each "Finished Backup" line. Task IDs embed the parent UPID, keeping them +// stable across polls and distinct from individually-run guest backups. +func parseVzdumpJobLog(jobTask models.BackupTask, upid string, logLines []proxmox.TaskLogLine) []models.BackupTask { + type guestSection struct { + vmid int + start time.Time + end time.Time + status string + errMsg string + } + + var order []int + sections := make(map[int]*guestSection) + cursor := jobTask.StartTime + + ensureSection := func(vmid int) *guestSection { + if sec, ok := sections[vmid]; ok { + return sec + } + sec := &guestSection{vmid: vmid, start: cursor, status: "running"} + sections[vmid] = sec + order = append(order, vmid) + return sec + } + + for _, line := range logLines { + text := strings.TrimSpace(line.Text) + if match := vzdumpLogGuestStartRe.FindStringSubmatch(text); match != nil { + vmid, err := strconv.Atoi(match[1]) + if err != nil || vmid <= 0 { + continue + } + sec := ensureSection(vmid) + sec.start = cursor + sec.status = "running" + } else if match := vzdumpLogGuestFinishRe.FindStringSubmatch(text); match != nil { + vmid, err := strconv.Atoi(match[1]) + if err != nil || vmid <= 0 { + continue + } + sec := ensureSection(vmid) + if sec.status == "running" { + sec.status = "OK" + } + hours, _ := strconv.Atoi(match[2]) + minutes, _ := strconv.Atoi(match[3]) + seconds, _ := strconv.Atoi(match[4]) + duration := time.Duration(hours)*time.Hour + time.Duration(minutes)*time.Minute + time.Duration(seconds)*time.Second + sec.end = sec.start.Add(duration) + cursor = sec.end + } else if match := vzdumpLogGuestFailRe.FindStringSubmatch(text); match != nil { + vmid, err := strconv.Atoi(match[1]) + if err != nil || vmid <= 0 { + continue + } + sec := ensureSection(vmid) + sec.status = "error" + sec.errMsg = strings.TrimSpace(match[2]) + if sec.errMsg == "" { + sec.errMsg = "backup failed" + } + } + } + + if len(order) == 0 { + return nil + } + + jobFinished := !jobTask.EndTime.IsZero() + tasks := make([]models.BackupTask, 0, len(order)) + for _, vmid := range order { + sec := sections[vmid] + if jobFinished { + if sec.status == "running" { + // The job ended without a per-guest verdict; treat as failed + // so the guest doesn't look covered by a backup that never + // completed. + sec.status = "error" + sec.errMsg = "backup job ended before guest backup completed" + } + if sec.end.IsZero() { + sec.end = jobTask.EndTime + } + } + + tasks = append(tasks, models.BackupTask{ + ID: fmt.Sprintf("%s-%s-vm%d", jobTask.Instance, upid, vmid), + Node: jobTask.Node, + Instance: jobTask.Instance, + Type: jobTask.Type, + VMID: vmid, + Status: sec.status, + Error: sec.errMsg, + StartTime: sec.start, + EndTime: sec.end, + }) + } + + return tasks +} + func (m *Monitor) pollPVEBackupsAsync( ctx context.Context, instanceName string, diff --git a/internal/monitoring/monitor_full_coverage_test.go b/internal/monitoring/monitor_full_coverage_test.go index 8cc83e00ef..25b28c9ac9 100644 --- a/internal/monitoring/monitor_full_coverage_test.go +++ b/internal/monitoring/monitor_full_coverage_test.go @@ -739,6 +739,10 @@ func (m *mockPVEClientExtended) GetNodePendingUpdates(ctx context.Context, node return nil, nil } +func (m *mockPVEClientExtended) GetTaskLog(ctx context.Context, node, upid string) ([]proxmox.TaskLogLine, error) { + return nil, nil +} + func (m *mockPVEClientExtended) GetBackupTasks(ctx context.Context) ([]proxmox.Task, error) { return []proxmox.Task{ {UPID: "UPID:node1:00001D1A:00000000:65E1E1E1:vzdump:101:root@pam:", Node: "node1", Status: "OK", StartTime: time.Now().Unix(), ID: "101"}, diff --git a/internal/monitoring/monitor_memory_test.go b/internal/monitoring/monitor_memory_test.go index 8f31687b69..df2a659a1b 100644 --- a/internal/monitoring/monitor_memory_test.go +++ b/internal/monitoring/monitor_memory_test.go @@ -62,6 +62,10 @@ func (s *stubPVEClient) GetAllStorage(ctx context.Context) ([]proxmox.Storage, e return s.allStorage, nil } +func (s *stubPVEClient) GetTaskLog(ctx context.Context, node, upid string) ([]proxmox.TaskLogLine, error) { + return nil, nil +} + func (s *stubPVEClient) GetBackupTasks(ctx context.Context) ([]proxmox.Task, error) { return nil, nil } diff --git a/internal/monitoring/monitor_snapshot_test.go b/internal/monitoring/monitor_snapshot_test.go index fdc7f2bab9..fa88b0c16f 100644 --- a/internal/monitoring/monitor_snapshot_test.go +++ b/internal/monitoring/monitor_snapshot_test.go @@ -62,6 +62,10 @@ type backupStorageTimeoutSnapshotClient struct { storageCalls int } +func (m *backupStorageTimeoutSnapshotClient) GetTaskLog(ctx context.Context, node, upid string) ([]proxmox.TaskLogLine, error) { + return nil, nil +} + func (m *backupStorageTimeoutSnapshotClient) GetBackupTasks(ctx context.Context) ([]proxmox.Task, error) { return nil, nil } diff --git a/internal/monitoring/monitor_snapshots_test.go b/internal/monitoring/monitor_snapshots_test.go index 637c73c925..02fe012790 100644 --- a/internal/monitoring/monitor_snapshots_test.go +++ b/internal/monitoring/monitor_snapshots_test.go @@ -38,6 +38,10 @@ func (f fakeSnapshotClient) GetStorage(ctx context.Context, node string) ([]prox func (f fakeSnapshotClient) GetAllStorage(ctx context.Context) ([]proxmox.Storage, error) { return nil, nil } +func (f fakeSnapshotClient) GetTaskLog(ctx context.Context, node, upid string) ([]proxmox.TaskLogLine, error) { + return nil, nil +} + func (f fakeSnapshotClient) GetBackupTasks(ctx context.Context) ([]proxmox.Task, error) { return nil, nil } diff --git a/internal/monitoring/monitor_storage_test.go b/internal/monitoring/monitor_storage_test.go index a25fbc1880..1a67858389 100644 --- a/internal/monitoring/monitor_storage_test.go +++ b/internal/monitoring/monitor_storage_test.go @@ -59,6 +59,10 @@ func (f *fakeStorageClient) GetAllStorage(ctx context.Context) ([]proxmox.Storag return f.allStorage, nil } +func (f *fakeStorageClient) GetTaskLog(ctx context.Context, node, upid string) ([]proxmox.TaskLogLine, error) { + return nil, nil +} + func (f *fakeStorageClient) GetBackupTasks(ctx context.Context) ([]proxmox.Task, error) { return nil, nil } diff --git a/pkg/proxmox/client.go b/pkg/proxmox/client.go index 876fac5f5b..1948649516 100644 --- a/pkg/proxmox/client.go +++ b/pkg/proxmox/client.go @@ -1274,7 +1274,13 @@ func (c *Client) GetNodeTasks(ctx context.Context, node string) ([]Task, error) return result.Data, nil } -// GetBackupTasks gets all backup tasks across all nodes +// backupTaskListLimit caps how many vzdump tasks are listed per node. The PVE +// default of 50 covers all task types, so vzdump entries get crowded out on +// busy nodes; filtering server-side keeps the window useful. +const backupTaskListLimit = 200 + +// GetBackupTasks gets all backup tasks across all nodes, including currently +// running ones (source=all) so in-progress backup jobs are visible. func (c *Client) GetBackupTasks(ctx context.Context) ([]Task, error) { // First get all nodes nodes, err := c.GetNodes(ctx) @@ -1288,14 +1294,23 @@ func (c *Client) GetBackupTasks(ctx context.Context) ([]Task, error) { continue } - tasks, err := c.GetNodeTasks(ctx, node.Node) + resp, err := c.get(ctx, fmt.Sprintf("/nodes/%s/tasks?typefilter=vzdump&source=all&limit=%d", node.Node, backupTaskListLimit)) if err != nil { // Log error but continue with other nodes continue } + var result struct { + Data []Task `json:"data"` + } + decodeErr := json.NewDecoder(resp.Body).Decode(&result) + resp.Body.Close() + if decodeErr != nil { + continue + } + // Filter for backup tasks - for _, task := range tasks { + for _, task := range result.Data { if task.Type == "vzdump" { allTasks = append(allTasks, task) } @@ -1305,6 +1320,44 @@ func (c *Client) GetBackupTasks(ctx context.Context) ([]Task, error) { return allTasks, nil } +// TaskLogLine is a single line of a task's log output. +type TaskLogLine struct { + LineNumber int64 `json:"n"` + Text string `json:"t"` +} + +// taskLogLineLimit caps how many log lines are fetched per task. The PVE +// default is 50 lines, far too few for multi-guest vzdump job logs; 20k lines +// covers jobs with hundreds of guests while bounding response size. +const taskLogLineLimit = 20000 + +// GetTaskLog fetches the log output of a task. +func (c *Client) GetTaskLog(ctx context.Context, node, upid string) ([]TaskLogLine, error) { + resp, err := c.get(ctx, fmt.Sprintf("/nodes/%s/tasks/%s/log?start=0&limit=%d", node, url.PathEscape(upid), taskLogLineLimit)) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + body, err := readResponseBodyLimited(resp.Body) + if err != nil { + return nil, err + } + return nil, fmt.Errorf("failed to get task log (status %d): %s", resp.StatusCode, strings.TrimSpace(string(body))) + } + + var result struct { + Data []TaskLogLine `json:"data"` + } + + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, err + } + + return result.Data, nil +} + // GetContainerInterfaces returns the network interfaces (with IPs) for a container. func (c *Client) GetContainerInterfaces(ctx context.Context, node string, vmid int) ([]ContainerInterface, error) { resp, err := c.get(ctx, fmt.Sprintf("/nodes/%s/lxc/%d/interfaces", node, vmid)) diff --git a/pkg/proxmox/cluster_client.go b/pkg/proxmox/cluster_client.go index 59950f825d..4ad7b5d5ec 100644 --- a/pkg/proxmox/cluster_client.go +++ b/pkg/proxmox/cluster_client.go @@ -1202,6 +1202,19 @@ func (cc *ClusterClient) GetBackupTasks(ctx context.Context) ([]Task, error) { return result, err } +func (cc *ClusterClient) GetTaskLog(ctx context.Context, node, upid string) ([]TaskLogLine, error) { + var result []TaskLogLine + err := cc.executeWithFailover(ctx, func(client *Client) error { + lines, err := client.GetTaskLog(ctx, node, upid) + if err != nil { + return err + } + result = lines + return nil + }) + return result, err +} + func (cc *ClusterClient) GetReplicationStatus(ctx context.Context) ([]ReplicationJob, error) { var result []ReplicationJob err := cc.executeWithFailover(ctx, func(client *Client) error { diff --git a/pkg/proxmox/cluster_client_api_test.go b/pkg/proxmox/cluster_client_api_test.go index 7bfe14f286..bc3b8af1f6 100644 --- a/pkg/proxmox/cluster_client_api_test.go +++ b/pkg/proxmox/cluster_client_api_test.go @@ -521,3 +521,33 @@ func TestClusterClient_GetContainerConfig(t *testing.T) { t.Errorf("expected hostname ct1, got %v", config["hostname"]) } } + +func TestClusterClient_GetTaskLog(t *testing.T) { + const upid = "UPID:node1:000E9F2C:0AC734B2:68A1B2C3:vzdump::root@pam:" + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + if r.URL.Path == "/api2/json/nodes" { + fmt.Fprint(w, `{"data":[{"node":"node1","status":"online"}]}`) + return + } + if r.URL.Path != "/api2/json/nodes/node1/tasks/"+upid+"/log" { + t.Errorf("unexpected path %q", r.URL.Path) + } + fmt.Fprint(w, `{"data":[{"n":1,"t":"INFO: Starting Backup of VM 101 (qemu)"},{"n":2,"t":"INFO: Finished Backup of VM 101 (00:01:30)"}]}`) + })) + defer server.Close() + + cfg := ClientConfig{Host: server.URL, TokenName: "u@p!t", TokenValue: "v"} + cc := NewClusterClient("test", cfg, []string{server.URL}, nil) + + lines, err := cc.GetTaskLog(context.Background(), "node1", upid) + if err != nil { + t.Fatalf("GetTaskLog failed: %v", err) + } + if len(lines) != 2 { + t.Fatalf("expected 2 log lines, got %d", len(lines)) + } + if lines[0].Text != "INFO: Starting Backup of VM 101 (qemu)" || lines[0].LineNumber != 1 { + t.Errorf("unexpected first line: %+v", lines[0]) + } +}