vrr/Pulse
2
0
Fork 0
mirror of https://github.com/rcourtman/Pulse.git synced 2026-10-19 02:28:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 13

fix(backups): synthesize per-guest task status from vzdump job logs

Browse source 
Scheduled multi-guest vzdump jobs run under a single UPID whose VMID slot
is empty, so pollBackupTasks stored them with VMID 0 and the guest-centric
backups coverage view dropped them entirely: only individually backed-up
guests ever showed task status. (Regressed with the v6.0.0 guest-centric
redesign, which removed the flat task table that used to render job runs.)

pollBackupTasks now fetches the job task's log and parses the per-guest
markers ("Starting Backup of VM", "Finished Backup of VM (duration)",
"Backup of VM failed - reason") into synthetic per-guest BackupTask
entries. Their IDs embed the parent UPID, keeping them stable across polls
and distinct from individually-run backups; per-guest times are
reconstructed from the job start plus the printed durations. Finished
jobs' logs are immutable, so results are cached per instance|UPID and each
finished run is fetched at most once, with a per-cycle fetch cap so a
historical backlog trickles in without stalling the backup poll budget.

The task listing now uses source=all + typefilter=vzdump, so running jobs
are visible too: guests covered by an in-progress job get a "running"
synthetic task, which also feeds resolveBackupIntentContext and
suppresses offline/backup alerts for guests the job is actively backing
up. The frontend needs no changes - synthetic tasks carry real VMIDs and
flow through the existing coverage model, recovery mapper, and alert
intent paths.

Contract: monitoring.md completion obligation 13 records the per-guest
synthesis boundary; proofs land in monitor_backup_job_tasks_test.go,
monitor_alert_intent_test.go, and cluster_client_api_test.go.

Reported by Johannes Strasser (support thread "PBS Bug").

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
courtmanr@gmail.com 2026-07-26 20:03:46 +01:00
parent c08da19ae7
commit a402d23503
14 changed files with 703 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

23
docs/release-control/v6/internal/subsystems/monitoring.md
View file

@ -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

3
internal/monitoring/harness_integration.go
View file

@ -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
}

3
internal/monitoring/monitor.go
View file

@ -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{}),

39
internal/monitoring/monitor_alert_intent_test.go
View file

@ -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)
}
}

283
internal/monitoring/monitor_backup_job_tasks_test.go Normal file
View file

@ -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")
}
}

233
internal/monitoring/monitor_backups.go
View file

@ -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,

4
internal/monitoring/monitor_full_coverage_test.go
View file

@ -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"},

4
internal/monitoring/monitor_memory_test.go
View file

@ -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
}

4
internal/monitoring/monitor_snapshot_test.go
View file

@ -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
}

4
internal/monitoring/monitor_snapshots_test.go
View file

@ -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
}

4
internal/monitoring/monitor_storage_test.go
View file

@ -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
}

59
pkg/proxmox/client.go
View file

@ -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))

13
pkg/proxmox/cluster_client.go
View file

@ -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 {

30
pkg/proxmox/cluster_client_api_test.go
View file

@ -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])
}
}