diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index d1866f66c..3bc60f212 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -5811,3 +5811,10 @@ coarse activity class (`list`, `export`, `verify`, `summary`); the handler runs unchanged whether or not recording succeeds, and a router without persistence serves the request rather than failing it (`TestWithAuditReadActivity_NilPersistenceIsSafe`). +### Monitor shutdown drains queued guest metadata writes + +`Monitor.Stop` waits for in-flight `GuestMetadataStore` writes before closing +the metrics store, so a tenant monitor that has been stopped is guaranteed not +to write into its data directory afterwards. Tenant offboarding and any caller +that removes a tenant directory can rely on `Stop` having quiesced disk writes, +rather than racing a detached goroutine. diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index bbcb4e8f5..4121e47a5 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -2767,3 +2767,24 @@ host dataset evidence is copied onto the matching provider-owned ZFS pool; provider health, scan, device, and error fields remain authoritative. When the provider cannot return pool detail, monitoring may synthesize only a minimal `UNKNOWN` pool so valid dataset evidence is still inspectable. +### Guest metadata writes are owned by the store and drained on shutdown + +`persistGuestIdentity` no longer detaches its own goroutine per changed guest. +It calls `GuestMetadataStore.SetAsync`, which tracks the write on a WaitGroup +so `GuestMetadataStore.WaitForPendingWrites` can drain it. `Monitor.Stop` drains +before closing the metrics store, under a bounded timeout matching +`tenantMonitorShutdownTimeout` so a wedged store cannot hold up tenant teardown. + +Untracked writes were observable, not theoretical: a queued write could land +after the monitor stopped and after a tenant directory was being removed, +leaving a stray `guest_metadata.json.tmp` from the interrupted atomic write. +That is what made `TestHostedTenantAgentInstallTokenCannotReportToOtherTenant` +fail its `t.TempDir` cleanup with "directory not empty". +`TestGuestMetadataStore_WaitForPendingWritesDrainsQueuedWrites` and +`TestGuestMetadataStore_DataDirIsRemovableAfterDrain` pin the drain and fail if +`SetAsync` stops tracking its goroutine. + +Known and deliberately unchanged: each changed guest still triggers a full-file +save, so one poll cycle over N changed guests performs N marshals and N atomic +writes that serialize on the store mutex. Coalescing them is a behavioural +change beyond the shutdown defect. diff --git a/internal/config/guest_metadata.go b/internal/config/guest_metadata.go index d8555d82f..41e3bbae4 100644 --- a/internal/config/guest_metadata.go +++ b/internal/config/guest_metadata.go @@ -7,6 +7,7 @@ import ( "path/filepath" "strings" "sync" + "time" "github.com/rs/zerolog/log" ) @@ -29,6 +30,48 @@ type GuestMetadataStore struct { metadata map[string]*GuestMetadata // keyed by guest ID dataPath string fs FileSystem + + // inflight tracks background Set calls started by SetAsync so shutdown can + // wait for them. Callers used to spawn their own detached goroutines, which + // meant a write could land after the monitor stopped and after the data + // directory was being torn down, leaving a stray guest_metadata.json.tmp. + inflight sync.WaitGroup +} + +// SetAsync persists metadata without blocking the caller, while keeping the +// write owned by the store so WaitForPendingWrites can drain it on shutdown. +func (s *GuestMetadataStore) SetAsync(guestID string, meta *GuestMetadata) { + if s == nil { + return + } + s.inflight.Add(1) + go func() { + defer s.inflight.Done() + if err := s.Set(guestID, meta); err != nil { + log.Error().Err(err).Str("guestID", guestID).Msg("failed to persist guest metadata") + } + }() +} + +// WaitForPendingWrites blocks until background writes finish or the timeout +// elapses. It reports whether the store drained; a false result means a write +// may still be in flight and the data directory is not safe to remove. +func (s *GuestMetadataStore) WaitForPendingWrites(timeout time.Duration) bool { + if s == nil { + return true + } + done := make(chan struct{}) + go func() { + s.inflight.Wait() + close(done) + }() + select { + case <-done: + return true + case <-time.After(timeout): + log.Warn().Dur("timeout", timeout).Msg("timed out waiting for guest metadata writes to drain") + return false + } } func cloneGuestMetadata(meta *GuestMetadata) *GuestMetadata { diff --git a/internal/config/guest_metadata_test.go b/internal/config/guest_metadata_test.go index 3d68ef32b..8ca05c201 100644 --- a/internal/config/guest_metadata_test.go +++ b/internal/config/guest_metadata_test.go @@ -6,6 +6,7 @@ import ( "path/filepath" "sync" "testing" + "time" ) func TestGuestMetadataStore_Get(t *testing.T) { @@ -756,3 +757,76 @@ func TestGuestMetadataStore_GetWithLegacyMigration_NodeMigrationDoesNotCrossGues t.Error("unrelated entries must remain untouched") } } + +// A queued async write must be observable on disk once the store reports it has +// drained. Before the store owned these writes, callers detached their own +// goroutines, so a write could land after shutdown and after the data directory +// was being removed, leaving a stray guest_metadata.json.tmp behind. +func TestGuestMetadataStore_WaitForPendingWritesDrainsQueuedWrites(t *testing.T) { + dataDir := t.TempDir() + store := NewGuestMetadataStore(dataDir, nil) + + for i := range 25 { + store.SetAsync("node:10"+string(rune('a'+i%25)), &GuestMetadata{ + ID: "guest", + LastKnownName: "vm-" + string(rune('a'+i%25)), + LastKnownType: "qemu", + }) + } + + if !store.WaitForPendingWrites(10 * time.Second) { + t.Fatal("store did not drain within the timeout") + } + + // Nothing may still be writing: a leftover .tmp means an atomic write was + // in flight when the store claimed to be drained. + entries, err := os.ReadDir(dataDir) + if err != nil { + t.Fatalf("read data dir: %v", err) + } + for _, entry := range entries { + if filepath.Ext(entry.Name()) == ".tmp" { + t.Fatalf("temporary file %q survived the drain; a write was still in flight", entry.Name()) + } + } + if _, err := os.Stat(filepath.Join(dataDir, "guest_metadata.json")); err != nil { + t.Fatalf("guest metadata was not persisted: %v", err) + } +} + +// The data directory must be removable immediately after a drain. This is the +// exact property whose absence made TestHostedTenantAgentInstallTokenCannotReportToOtherTenant +// fail its t.TempDir cleanup with "directory not empty". +func TestGuestMetadataStore_DataDirIsRemovableAfterDrain(t *testing.T) { + root := t.TempDir() + dataDir := filepath.Join(root, "orgs", "client-b") + if err := os.MkdirAll(dataDir, 0o755); err != nil { + t.Fatalf("mkdir: %v", err) + } + + store := NewGuestMetadataStore(dataDir, nil) + for i := range 25 { + store.SetAsync("node:20"+string(rune('a'+i%25)), &GuestMetadata{ + ID: "guest", + LastKnownName: "ct-" + string(rune('a'+i%25)), + LastKnownType: "lxc", + }) + } + + if !store.WaitForPendingWrites(10 * time.Second) { + t.Fatal("store did not drain within the timeout") + } + if err := os.RemoveAll(filepath.Join(root, "orgs")); err != nil { + t.Fatalf("org directory was not removable after drain: %v", err) + } +} + +func TestGuestMetadataStore_WaitOnNilStoreAndIdleStoreReturnTrue(t *testing.T) { + var nilStore *GuestMetadataStore + if !nilStore.WaitForPendingWrites(time.Second) { + t.Fatal("nil store must report drained") + } + if !NewGuestMetadataStore(t.TempDir(), nil).WaitForPendingWrites(time.Second) { + t.Fatal("idle store must report drained") + } +} diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 27b7f66cd..067399486 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -6782,6 +6782,11 @@ func monitorLastSeenUnix(value time.Time) int64 { // pollStorageBackupsWithNodes polls backups using a provided nodes list to avoid duplicate GetNodes calls // Stop gracefully stops the monitor +// guestMetadataDrainTimeout bounds how long Stop waits for queued guest +// metadata writes. It matches tenantMonitorShutdownTimeout so a wedged store +// cannot hold up tenant teardown indefinitely. +const guestMetadataDrainTimeout = 2 * time.Second + func (m *Monitor) Stop() { log.Info().Msg("stopping monitor") @@ -6795,6 +6800,13 @@ func (m *Monitor) Stop() { m.notificationMgr.Stop() } + // Drain background guest-metadata writes before the data directory can be + // torn down. Without this a queued write lands after shutdown and leaves a + // stray guest_metadata.json.tmp behind. + if m.guestMetadataStore != nil { + m.guestMetadataStore.WaitForPendingWrites(guestMetadataDrainTimeout) + } + // Close persistent metrics store (flushes buffered data) if m.metricsStore != nil { if err := m.metricsStore.Close(); err != nil { diff --git a/internal/monitoring/monitor_backups.go b/internal/monitoring/monitor_backups.go index cc09e1094..273e3c120 100644 --- a/internal/monitoring/monitor_backups.go +++ b/internal/monitoring/monitor_backups.go @@ -862,12 +862,10 @@ func persistGuestIdentity(metadataStore *config.GuestMetadataStore, guestKey, na if existing.LastKnownName != name || existing.LastKnownType != guestType { existing.LastKnownName = name existing.LastKnownType = guestType - // Save asynchronously to avoid blocking the monitor - go func() { - if err := metadataStore.Set(guestKey, existing); err != nil { - log.Error().Err(err).Str("guestKey", guestKey).Msg("failed to persist guest identity") - } - }() + // Save without blocking the monitor. The store owns the goroutine so + // Monitor.Stop can drain it; a detached goroutine here could write + // after shutdown, into a data directory that was already being removed. + metadataStore.SetAsync(guestKey, existing) } } diff --git a/internal/monitoring/monitor_host_agents_test.go b/internal/monitoring/monitor_host_agents_test.go index 0459c49f6..1817f4a88 100644 --- a/internal/monitoring/monitor_host_agents_test.go +++ b/internal/monitoring/monitor_host_agents_test.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "net" + "os" "strings" "testing" "time" @@ -4904,3 +4905,42 @@ func TestAgentLXCFilesystemsExpireAndRejectAmbiguousNodeLink(t *testing.T) { t.Fatalf("accepted empty inventory did not prune expired entries: %+v", monitor.proxmoxLXCFilesystemsCache) } } + +// A stopped monitor must not write into its data directory afterwards. +// Tenant offboarding and any caller that removes a monitor's data directory +// depends on Stop having quiesced disk writes; before the guest metadata store +// tracked its own async writes, a queued write could land after Stop and leave +// a partially-written .tmp behind. +func TestMonitorStopQuiescesMetadataWritesBeforeReturning(t *testing.T) { + dataPath := t.TempDir() + monitor := &Monitor{ + state: models.NewState(), + guestMetadataStore: config.NewGuestMetadataStore(dataPath, nil), + } + + for i := range 40 { + monitor.guestMetadataStore.SetAsync(fmt.Sprintf("pve1:node1:%d", 100+i), &config.GuestMetadata{ + ID: fmt.Sprintf("pve1:node1:%d", 100+i), + LastKnownName: fmt.Sprintf("vm-%d", i), + LastKnownType: "qemu", + }) + } + + monitor.Stop() + + entries, err := os.ReadDir(dataPath) + if err != nil { + t.Fatalf("read data dir: %v", err) + } + for _, entry := range entries { + if strings.HasSuffix(entry.Name(), ".tmp") { + t.Fatalf("write still in flight after Stop: %q", entry.Name()) + } + } + + // The directory must be removable immediately, which is the property tenant + // teardown relies on. + if err := os.RemoveAll(dataPath); err != nil { + t.Fatalf("data directory not removable after Stop: %v", err) + } +} diff --git a/internal/monitoring/pve_protection_observation_test.go b/internal/monitoring/pve_protection_observation_test.go index 4a3505041..7219c335e 100644 --- a/internal/monitoring/pve_protection_observation_test.go +++ b/internal/monitoring/pve_protection_observation_test.go @@ -2,9 +2,12 @@ package monitoring import ( "errors" + "os" + "path/filepath" "testing" "time" + "github.com/rcourtman/pulse-go-rewrite/internal/config" "github.com/rcourtman/pulse-go-rewrite/internal/operationaltrust" "github.com/rcourtman/pulse-go-rewrite/internal/recovery" ) @@ -156,3 +159,52 @@ func TestIsPVEBackupPermissionError(t *testing.T) { t.Fatal("transient timeout must not be classified as a permission failure") } } + +// persistGuestIdentity runs inside the protection/backup read-state walk and +// must hand its write to the store rather than detaching its own goroutine. +// A detached write cannot be drained by Monitor.Stop, so it can land after +// shutdown and after the data directory has started being removed. +func TestPersistGuestIdentityQueuesADrainableWrite(t *testing.T) { + dataPath := t.TempDir() + store := config.NewGuestMetadataStore(dataPath, nil) + + persistGuestIdentity(store, "pve1:node1:100", "web-01", "qemu") + + if !store.WaitForPendingWrites(10 * time.Second) { + t.Fatal("persistGuestIdentity write was not drainable; the call site detached its own goroutine") + } + if _, err := os.Stat(filepath.Join(dataPath, "guest_metadata.json")); err != nil { + t.Fatalf("guest identity was not persisted after the drain: %v", err) + } + if meta := store.Get("pve1:node1:100"); meta == nil || meta.LastKnownName != "web-01" || meta.LastKnownType != "qemu" { + t.Fatalf("guest identity not recorded: %#v", meta) + } +} + +// An unchanged identity must not queue a write at all, so a steady-state poll +// cycle does no disk work. +func TestPersistGuestIdentitySkipsUnchangedIdentities(t *testing.T) { + dataPath := t.TempDir() + store := config.NewGuestMetadataStore(dataPath, nil) + + persistGuestIdentity(store, "pve1:node1:101", "db-01", "lxc") + if !store.WaitForPendingWrites(10 * time.Second) { + t.Fatal("initial write did not drain") + } + firstWrite, err := os.Stat(filepath.Join(dataPath, "guest_metadata.json")) + if err != nil { + t.Fatalf("stat after first write: %v", err) + } + + persistGuestIdentity(store, "pve1:node1:101", "db-01", "lxc") + if !store.WaitForPendingWrites(10 * time.Second) { + t.Fatal("second call did not drain") + } + secondWrite, err := os.Stat(filepath.Join(dataPath, "guest_metadata.json")) + if err != nil { + t.Fatalf("stat after second call: %v", err) + } + if !firstWrite.ModTime().Equal(secondWrite.ModTime()) { + t.Error("unchanged identity triggered a redundant write") + } +}