From c73e78aeea3efee2018138018ee8fa516ce4fc7c Mon Sep 17 00:00:00 2001 From: rcourtman Date: Mon, 20 Jul 2026 07:50:22 +0100 Subject: [PATCH] Stabilize RC runtime ownership Contract-Neutral: RC qualification fixes preserve existing public API, tenant, monitoring, and organization contracts while correcting canonical runtime ownership and test fixtures. --- internal/api/config_handlers.go | 37 +++- .../multi_tenant_setters_additional_test.go | 34 +++ internal/api/org_handlers.go | 3 +- internal/api/router.go | 2 +- internal/api/update_readiness.go | 13 ++ internal/api/update_readiness_test.go | 29 +++ internal/mock/metric_runtime.go | 48 +++- internal/mock/metric_runtime_test.go | 45 ++++ .../monitoring/canonical_guardrails_test.go | 5 +- internal/monitoring/mock_metrics_history.go | 206 +++++++++--------- .../monitoring/mock_metrics_history_test.go | 39 ++++ internal/monitoring/multi_tenant_monitor.go | 81 +++++-- .../multi_tenant_monitor_additional_test.go | 33 +++ .../tests/05-settings-mobile-audit.spec.ts | 30 ++- .../42-vmware-ai-chat-read-recovery.spec.ts | 57 +++-- .../tests/vmware-workload-fixture.ts | 20 +- 16 files changed, 513 insertions(+), 169 deletions(-) diff --git a/internal/api/config_handlers.go b/internal/api/config_handlers.go index 0ff90bc8a..9dd93b49c 100644 --- a/internal/api/config_handlers.go +++ b/internal/api/config_handlers.go @@ -325,7 +325,19 @@ func (h *ConfigHandlers) getContextState(ctx context.Context) (*config.Config, * } } - // Resolve from multi-tenant managers first. + // The default org is the primary runtime owned by Router. Its config and + // monitor are updated in place by setup and reload flows, while the + // multi-tenant manager keeps isolated tenant copies. Never route default-org + // requests back through that tenant-copy path or they can observe stale auth + // and setup state. + if orgID == "default" && + (defaultConfig != nil || defaultPersistence != nil || defaultMonitor != nil) { + return defaultConfig, defaultPersistence, defaultMonitor + } + + // Resolve non-default organizations, plus initialization-only handlers + // that have not yet been given the primary default runtime, from the + // multi-tenant managers. if mtMonitor != nil { if m, err := mtMonitor.GetMonitor(orgID); err == nil && m != nil { cfg := m.GetConfig() @@ -335,18 +347,25 @@ func (h *ConfigHandlers) getContextState(ctx context.Context) (*config.Config, * } return cfg, p, m } else { - if orgID != "default" { - log.Warn().Str("orgID", orgID).Err(err).Msg("Tenant config resolution failed for non-default org") - return nil, nil, nil - } - if err != nil { - log.Warn().Str("orgID", orgID).Err(err).Msg("Falling back to default-org config after default tenant monitor lookup failure") + if orgID == "default" { + log.Warn().Str("orgID", orgID).Err(err).Msg("Default config resolution failed before primary runtime initialization") + return defaultConfig, defaultPersistence, defaultMonitor } + log.Warn().Str("orgID", orgID).Err(err).Msg("Tenant config resolution failed for non-default org") + return nil, nil, nil } } - // Fallback to default-org state (primarily for "default" org or initialization). - return defaultConfig, defaultPersistence, defaultMonitor + if orgID == "default" { + return defaultConfig, defaultPersistence, defaultMonitor + } + // Single-runtime handlers have no tenant manager and therefore no + // alternate state to resolve. Keep their historical default-state + // behavior; multi-tenant handlers above continue to fail closed. + if mtMonitor == nil { + return defaultConfig, defaultPersistence, defaultMonitor + } + return nil, nil, nil } func (h *ConfigHandlers) getConfig(ctx context.Context) *config.Config { diff --git a/internal/api/multi_tenant_setters_additional_test.go b/internal/api/multi_tenant_setters_additional_test.go index 19d1fa0d2..e932f0435 100644 --- a/internal/api/multi_tenant_setters_additional_test.go +++ b/internal/api/multi_tenant_setters_additional_test.go @@ -98,6 +98,40 @@ func TestConfigHandlersNonDefaultMissingTenantMonitorFailsClosed(t *testing.T) { } } +func TestConfigHandlersDefaultContextUsesPrimaryRuntimeState(t *testing.T) { + tenantCopyConfig := &config.Config{DataPath: t.TempDir(), ConfigPath: t.TempDir()} + tenantCopyMonitor, _, _ := newTestMonitor(t) + setUnexportedField(t, tenantCopyMonitor, "config", tenantCopyConfig) + + mtm := &monitoring.MultiTenantMonitor{} + setUnexportedField(t, mtm, "monitors", map[string]*monitoring.Monitor{ + "default": tenantCopyMonitor, + }) + + primaryConfig := &config.Config{DataPath: t.TempDir(), ConfigPath: t.TempDir()} + primaryMonitor, _, _ := newTestMonitor(t) + setUnexportedField(t, primaryMonitor, "config", primaryConfig) + primaryPersistence := config.NewConfigPersistence(primaryConfig.ConfigPath) + + handler := &ConfigHandlers{ + defaultConfig: primaryConfig, + defaultPersistence: primaryPersistence, + defaultMonitor: primaryMonitor, + mtMonitor: mtm, + } + + cfg, persistence, monitor := handler.getContextState(context.Background()) + if cfg != primaryConfig { + t.Fatalf("default config = %#v, want primary config %#v", cfg, primaryConfig) + } + if persistence != primaryPersistence { + t.Fatalf("default persistence = %#v, want primary persistence %#v", persistence, primaryPersistence) + } + if monitor != primaryMonitor { + t.Fatalf("default monitor = %#v, want primary monitor %#v", monitor, primaryMonitor) + } +} + func TestNewRouterWiresTenantResourceStateProvider(t *testing.T) { cfg := &config.Config{ DataPath: t.TempDir(), diff --git a/internal/api/org_handlers.go b/internal/api/org_handlers.go index 63ae0a163..0704db88e 100644 --- a/internal/api/org_handlers.go +++ b/internal/api/org_handlers.go @@ -387,7 +387,8 @@ func (h *OrgHandlers) HandleDeleteOrg(w http.ResponseWriter, r *http.Request) { } if h.mtMonitor != nil { - h.mtMonitor.RemoveTenant(orgID) + h.mtMonitor.BeginTenantDeletion(orgID) + defer h.mtMonitor.FinishTenantDeletion(orgID) } if err := h.persistence.DeleteOrganization(orgID); err != nil { diff --git a/internal/api/router.go b/internal/api/router.go index 9d6f02a49..61b0e0c2a 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -478,7 +478,7 @@ func (r *Router) setupRoutes() { r.vmwarePoller.Start(r.lifecycleCtx) updateHandlers := NewUpdateHandlersWithContext(r.updateManager, r.updateHistory, r.lifecycleCtx) updateHandlers.SetUpdateReadinessSources( - r.configHandlers.getConfig, + r.updateReadinessConfigSnapshot, func(context.Context) []models.Host { if r.monitor == nil { return nil diff --git a/internal/api/update_readiness.go b/internal/api/update_readiness.go index 8e36a0841..1e173b54d 100644 --- a/internal/api/update_readiness.go +++ b/internal/api/update_readiness.go @@ -1,6 +1,7 @@ package api import ( + "context" "fmt" "strconv" "strings" @@ -29,6 +30,18 @@ type updateReadinessInputs struct { now time.Time } +// updateReadinessConfigSnapshot reads the canonical runtime configuration. +// Tenant monitors own isolated infrastructure config copies, while API token +// mutations are system-level state held by Router.config. +func (r *Router) updateReadinessConfigSnapshot(context.Context) *config.Config { + config.Mu.RLock() + defer config.Mu.RUnlock() + if r == nil || r.config == nil { + return nil + } + return r.config.DeepCopy() +} + func buildUpdateReadiness(in updateReadinessInputs) *updates.UpdateReadiness { now := in.now if now.IsZero() { diff --git a/internal/api/update_readiness_test.go b/internal/api/update_readiness_test.go index f3d5653f0..15c5d9cf0 100644 --- a/internal/api/update_readiness_test.go +++ b/internal/api/update_readiness_test.go @@ -1,6 +1,7 @@ package api import ( + "context" "testing" "time" @@ -9,6 +10,34 @@ import ( "github.com/rcourtman/pulse-go-rewrite/internal/updates" ) +func TestRouterUpdateReadinessConfigSnapshotUsesCanonicalRuntimeTokens(t *testing.T) { + token, err := config.NewAPITokenRecord( + "0123456789abcdef0123456789abcdef", + "agent", + []string{config.ScopeAgentReport}, + ) + if err != nil { + t.Fatalf("NewAPITokenRecord() error = %v", err) + } + + r := &Router{ + config: &config.Config{APITokens: []config.APITokenRecord{*token}}, + configHandlers: &ConfigHandlers{ + defaultConfig: &config.Config{}, + }, + } + + snapshot := r.updateReadinessConfigSnapshot(context.Background()) + if snapshot == nil || len(snapshot.APITokens) != 1 { + t.Fatalf("snapshot APITokens = %#v, want canonical runtime token", snapshot) + } + + r.config.APITokens = nil + if len(snapshot.APITokens) != 1 { + t.Fatal("expected update readiness snapshot to be independent of later runtime mutations") + } +} + func TestBuildUpdateReadiness_ActiveV5AgentWarnsForFirstHopTransport(t *testing.T) { now := time.Date(2026, 5, 28, 12, 0, 0, 0, time.UTC) record, err := config.NewAPITokenRecord("abcdef1234567890abcdef1234567890", "agent", []string{config.ScopeAgentReport}) diff --git a/internal/mock/metric_runtime.go b/internal/mock/metric_runtime.go index 3a9e3cc13..0cd58f539 100644 --- a/internal/mock/metric_runtime.go +++ b/internal/mock/metric_runtime.go @@ -140,8 +140,11 @@ func MetricBounds(resourceClass, metric string) (float64, float64) { } func MetricBoundsForResource(resourceClass, resourceID, metric string) (float64, float64) { + return metricBoundsForRole(resourceClass, metric, MetricRole(resourceClass, resourceID)) +} + +func metricBoundsForRole(resourceClass, metric, role string) (float64, float64) { min, max := MetricBounds(resourceClass, metric) - role := MetricRole(resourceClass, resourceID) metricKey := strings.ToLower(strings.TrimSpace(metric)) switch metricKey { @@ -207,6 +210,49 @@ func MetricBoundsForResource(resourceClass, resourceID, metric string) (float64, return min, max } +// MetricSampler binds metric persona resolution to one immutable fixture +// graph snapshot. Historical seeding and its live continuation must not depend +// on the process-global persona registry changing while another monitor or +// fixture update is initializing. +type MetricSampler struct { + roles map[string]string +} + +func NewMetricSampler(graph FixtureGraph) MetricSampler { + return MetricSampler{roles: buildMetricRoleRegistry(graph)} +} + +func (s MetricSampler) role(resourceClass, resourceID string) string { + if role := strings.TrimSpace(s.roles[metricRoleRegistryKey(resourceClass, resourceID)]); role != "" { + return role + } + return inferMetricRole(resourceClass, resourceID) +} + +func (s MetricSampler) SampleMetric(resourceClass, resourceID, metric string, at time.Time) float64 { + role := s.role(resourceClass, resourceID) + min, max := metricBoundsForRole(resourceClass, metric, role) + seed := MetricSeed(resourceClass, resourceID, metric) + return mockmodel.ValueAtMetricWithRole(seed, min, max, metric, metricSpeed(metric), role, at) +} + +func (s MetricSampler) SampleMetricSeries(resourceClass, resourceID, metric string, timestamps []time.Time) []float64 { + if len(timestamps) == 0 { + return nil + } + + role := s.role(resourceClass, resourceID) + min, max := metricBoundsForRole(resourceClass, metric, role) + seed := MetricSeed(resourceClass, resourceID, metric) + speed := metricSpeed(metric) + + values := make([]float64, len(timestamps)) + for i, at := range timestamps { + values[i] = mockmodel.ValueAtMetricWithRole(seed, min, max, metric, speed, role, at) + } + return values +} + func metricSpeed(metric string) float64 { switch strings.ToLower(strings.TrimSpace(metric)) { case "memory": diff --git a/internal/mock/metric_runtime_test.go b/internal/mock/metric_runtime_test.go index 9eb22e20c..775d2ec8b 100644 --- a/internal/mock/metric_runtime_test.go +++ b/internal/mock/metric_runtime_test.go @@ -4,6 +4,8 @@ import ( "math" "testing" "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/models" ) func TestSampleMetricSeriesMatchesCanonicalPointSampler(t *testing.T) { @@ -44,3 +46,46 @@ func TestSampleMetricSeriesEmptyInput(t *testing.T) { t.Fatalf("empty series = %#v, want nil", got) } } + +func TestMetricSamplerRemainsBoundToFixtureGraph(t *testing.T) { + previousRegistry := currentMetricRoleRegistry() + t.Cleanup(func() { + setMetricRoleRegistry(previousRegistry) + }) + + graph := FixtureGraph{ + State: models.StateSnapshot{ + Containers: []models.Container{{ + ID: "neutral-155", + Name: "backup-orchestrator", + Status: "running", + }}, + }, + } + sampler := NewMetricSampler(graph) + if got := sampler.role("container", "neutral-155"); got != metricRoleBackup { + t.Fatalf("sampler role = %q, want %q", got, metricRoleBackup) + } + + at := time.Date(2026, time.July, 19, 12, 0, 0, 0, time.UTC) + want := sampler.SampleMetric("container", "neutral-155", "memory", at) + + setMetricRoleRegistry(map[string]string{ + metricRoleRegistryKey("container", "neutral-155"): metricRoleDatabase, + }) + if got := sampler.SampleMetric("container", "neutral-155", "memory", at); math.Abs(got-want) > 1e-12 { + t.Fatalf("sampler changed after global registry update: got %v, want %v", got, want) + } + if global := SampleMetric("container", "neutral-155", "memory", at); math.Abs(global-want) < 1e-6 { + t.Fatalf("global database sample unexpectedly matched graph-bound backup sample: %v", global) + } + + timestamps := []time.Time{at.Add(-time.Minute), at, at.Add(time.Minute)} + series := sampler.SampleMetricSeries("container", "neutral-155", "memory", timestamps) + for i, timestamp := range timestamps { + point := sampler.SampleMetric("container", "neutral-155", "memory", timestamp) + if math.Abs(series[i]-point) > 1e-12 { + t.Fatalf("series point %d = %v, want %v", i, series[i], point) + } + } +} diff --git a/internal/monitoring/canonical_guardrails_test.go b/internal/monitoring/canonical_guardrails_test.go index b6483c3c3..94ebd0373 100644 --- a/internal/monitoring/canonical_guardrails_test.go +++ b/internal/monitoring/canonical_guardrails_test.go @@ -2009,8 +2009,9 @@ func TestMockNativePollersDeferToCanonicalMockSampler(t *testing.T) { { file: "mock_metrics_history.go", snippets: []string{ - `cpu := mock.SampleMetric("k8s", metricID, "cpu", ts)`, - `memory := mock.SampleMetric("k8s", metricID, "memory", ts)`, + "sampler := mock.NewMetricSampler(graph)", + `cpu := sampler.SampleMetric("k8s", metricID, "cpu", ts)`, + `memory := sampler.SampleMetric("k8s", metricID, "memory", ts)`, `ms.Write("k8s", metricID, "memory", memory, ts)`, "m.prewarmMockDashboardChartCaches()", }, diff --git a/internal/monitoring/mock_metrics_history.go b/internal/monitoring/mock_metrics_history.go index 1c8a4d3e9..a21326b4f 100644 --- a/internal/monitoring/mock_metrics_history.go +++ b/internal/monitoring/mock_metrics_history.go @@ -489,10 +489,20 @@ func normalizeMockMetricTimestamp(at time.Time, interval time.Duration) time.Tim } func canonicalMetricSeries(resourceType, resourceID, metric string, timestamps []time.Time) []float64 { + return canonicalMetricSeriesWithSampler( + mock.NewMetricSampler(mock.CurrentFixtureGraph()), + resourceType, + resourceID, + metric, + timestamps, + ) +} + +func canonicalMetricSeriesWithSampler(sampler mock.MetricSampler, resourceType, resourceID, metric string, timestamps []time.Time) []float64 { if len(timestamps) == 0 || strings.TrimSpace(resourceID) == "" { return nil } - return mock.SampleMetricSeries(resourceType, resourceID, metric, timestamps) + return sampler.SampleMetricSeries(resourceType, resourceID, metric, timestamps) } func seedMockMetricsHistory(mh *MetricsHistory, ms *metrics.Store, graph mock.FixtureGraph, now time.Time, seedDuration, interval time.Duration) { @@ -503,6 +513,7 @@ func seedMockMetricsHistory(mh *MetricsHistory, ms *metrics.Store, graph mock.Fi if seedDuration <= 0 || interval <= 0 { return } + sampler := mock.NewMetricSampler(graph) now = normalizeMockMetricTimestamp(now, interval) // Build one canonical timestamp list so seeded history and subsequent live @@ -584,7 +595,7 @@ func seedMockMetricsHistory(mh *MetricsHistory, ms *metrics.Store, graph mock.Fi for _, plan := range mockStoreSeedPlans { coverageKey := metrics.NormalizedSeriesKey(resourceType, resourceID, metricType) for _, ts := range seedStoreGapTimestamps(plan, coverageKey) { - queueStorePoint(resourceType, resourceID, metricType, mock.SampleMetric(resourceType, resourceID, metricType, ts), ts, plan.tier) + queueStorePoint(resourceType, resourceID, metricType, sampler.SampleMetric(resourceType, resourceID, metricType, ts), ts, plan.tier) } } } @@ -600,7 +611,7 @@ func seedMockMetricsHistory(mh *MetricsHistory, ms *metrics.Store, graph mock.Fi for _, plan := range mockStoreSeedPlans { coverageKey := metrics.NormalizedSeriesKey("storage", storageID, "usage") for _, ts := range seedStoreGapTimestamps(plan, coverageKey) { - usage := clampFloat(mock.SampleMetric("storage", storageID, "usage", ts), 0, 100) + usage := clampFloat(sampler.SampleMetric("storage", storageID, "usage", ts), 0, 100) used := currentTotal * (usage / 100.0) avail := math.Max(0, currentTotal-used) queueStorePoint("storage", storageID, "usage", usage, ts, plan.tier) @@ -615,7 +626,7 @@ func seedMockMetricsHistory(mh *MetricsHistory, ms *metrics.Store, graph mock.Fi return } - usageSeries := canonicalMetricSeries("storage", storageID, "usage", seedTimestamps) + usageSeries := canonicalMetricSeriesWithSampler(sampler, "storage", storageID, "usage", seedTimestamps) usedSeries := make([]float64, numPoints) availSeries := make([]float64, numPoints) totalSeries := make([]float64, numPoints) @@ -639,9 +650,9 @@ func seedMockMetricsHistory(mh *MetricsHistory, ms *metrics.Store, graph mock.Fi return } - cpuSeries := canonicalMetricSeries("node", node.ID, "cpu", seedTimestamps) - memSeries := canonicalMetricSeries("node", node.ID, "memory", seedTimestamps) - diskSeries := canonicalMetricSeries("node", node.ID, "disk", seedTimestamps) + cpuSeries := canonicalMetricSeriesWithSampler(sampler, "node", node.ID, "cpu", seedTimestamps) + memSeries := canonicalMetricSeriesWithSampler(sampler, "node", node.ID, "memory", seedTimestamps) + diskSeries := canonicalMetricSeriesWithSampler(sampler, "node", node.ID, "disk", seedTimestamps) mh.addNodeMetricSeries(node.ID, "cpu", cpuSeries, seedTimestamps) mh.addNodeMetricSeries(node.ID, "memory", memSeries, seedTimestamps) @@ -675,20 +686,20 @@ func seedMockMetricsHistory(mh *MetricsHistory, ms *metrics.Store, graph mock.Fi return } - cpuSeries := canonicalMetricSeries(storeType, storeID, "cpu", seedTimestamps) - memSeries := canonicalMetricSeries(storeType, storeID, "memory", seedTimestamps) + cpuSeries := canonicalMetricSeriesWithSampler(sampler, storeType, storeID, "cpu", seedTimestamps) + memSeries := canonicalMetricSeriesWithSampler(sampler, storeType, storeID, "memory", seedTimestamps) var diskSeries []float64 if includeDisk { - diskSeries = canonicalMetricSeries(storeType, storeID, "disk", seedTimestamps) + diskSeries = canonicalMetricSeriesWithSampler(sampler, storeType, storeID, "disk", seedTimestamps) } var diskReadSeries, diskWriteSeries, netInSeries, netOutSeries []float64 if includeDiskIO { - diskReadSeries = canonicalMetricSeries(storeType, storeID, "diskread", seedTimestamps) - diskWriteSeries = canonicalMetricSeries(storeType, storeID, "diskwrite", seedTimestamps) + diskReadSeries = canonicalMetricSeriesWithSampler(sampler, storeType, storeID, "diskread", seedTimestamps) + diskWriteSeries = canonicalMetricSeriesWithSampler(sampler, storeType, storeID, "diskwrite", seedTimestamps) } if includeNetwork { - netInSeries = canonicalMetricSeries(storeType, storeID, "netin", seedTimestamps) - netOutSeries = canonicalMetricSeries(storeType, storeID, "netout", seedTimestamps) + netInSeries = canonicalMetricSeriesWithSampler(sampler, storeType, storeID, "netin", seedTimestamps) + netOutSeries = canonicalMetricSeriesWithSampler(sampler, storeType, storeID, "netout", seedTimestamps) } for _, metricID := range uniqueMetricIDs { @@ -815,10 +826,10 @@ func seedMockMetricsHistory(mh *MetricsHistory, ms *metrics.Store, graph mock.Fi // series for one physical disk, shared by the native and TrueNAS fixture // disk loops below. seedDiskTelemetry := func(resourceID string) { - tempSeries := canonicalMetricSeries("disk", resourceID, "smart_temp", seedTimestamps) - busySeries := canonicalMetricSeries("disk", resourceID, "disk", seedTimestamps) - diskReadSeries := canonicalMetricSeries("disk", resourceID, "diskread", seedTimestamps) - diskWriteSeries := canonicalMetricSeries("disk", resourceID, "diskwrite", seedTimestamps) + tempSeries := canonicalMetricSeriesWithSampler(sampler, "disk", resourceID, "smart_temp", seedTimestamps) + busySeries := canonicalMetricSeriesWithSampler(sampler, "disk", resourceID, "disk", seedTimestamps) + diskReadSeries := canonicalMetricSeriesWithSampler(sampler, "disk", resourceID, "diskread", seedTimestamps) + diskWriteSeries := canonicalMetricSeriesWithSampler(sampler, "disk", resourceID, "diskwrite", seedTimestamps) mh.addDiskMetricSeries(resourceID, "smart_temp", tempSeries, seedTimestamps) mh.addDiskMetricSeries(resourceID, "disk", busySeries, seedTimestamps) mh.addDiskMetricSeries(resourceID, "diskread", diskReadSeries, seedTimestamps) @@ -920,7 +931,7 @@ func seedMockMetricsHistory(mh *MetricsHistory, ms *metrics.Store, graph mock.Fi for _, pool := range trueNASFixtures.Pools { poolKey := mock.TrueNASPoolMetricID(trueNASFixtures.System.Hostname, pool.Name) - diskSeries := canonicalMetricSeries("storage", poolKey, "usage", seedTimestamps) + diskSeries := canonicalMetricSeriesWithSampler(sampler, "storage", poolKey, "usage", seedTimestamps) mh.addGuestMetricSeries(poolKey, "disk", diskSeries, seedTimestamps) recordStorageTimeline(poolKey, float64(pool.TotalBytes)) } @@ -928,7 +939,7 @@ func seedMockMetricsHistory(mh *MetricsHistory, ms *metrics.Store, graph mock.Fi for _, dataset := range trueNASFixtures.Datasets { dsKey := mock.TrueNASDatasetMetricID(trueNASFixtures.System.Hostname, dataset.Name) totalBytes := dataset.UsedBytes + dataset.AvailBytes - diskSeries := canonicalMetricSeries("storage", dsKey, "usage", seedTimestamps) + diskSeries := canonicalMetricSeriesWithSampler(sampler, "storage", dsKey, "usage", seedTimestamps) mh.addGuestMetricSeries(dsKey, "disk", diskSeries, seedTimestamps) recordStorageTimeline(dsKey, float64(totalBytes)) } @@ -1057,17 +1068,17 @@ func prepareMockMetricsHistory( // recordTrueNASFixturesMetrics records live fixture ticks for TrueNAS host, // storage, and app-container metrics. -func recordTrueNASFixturesMetrics(mh *MetricsHistory, ms *metrics.Store, fixtures mock.PlatformFixtures, ts time.Time) { +func recordTrueNASFixturesMetrics(mh *MetricsHistory, ms *metrics.Store, sampler mock.MetricSampler, fixtures mock.PlatformFixtures, ts time.Time) { snapshot := fixtures.TrueNAS if strings.TrimSpace(snapshot.System.Hostname) != "" { - systemCPU := mock.SampleMetric("agent", snapshot.System.Hostname, "cpu", ts) - systemMemory := mock.SampleMetric("agent", snapshot.System.Hostname, "memory", ts) - systemDisk := mock.SampleMetric("agent", snapshot.System.Hostname, "disk", ts) - systemDiskRead := mock.SampleMetric("agent", snapshot.System.Hostname, "diskread", ts) - systemDiskWrite := mock.SampleMetric("agent", snapshot.System.Hostname, "diskwrite", ts) - systemNetIn := mock.SampleMetric("agent", snapshot.System.Hostname, "netin", ts) - systemNetOut := mock.SampleMetric("agent", snapshot.System.Hostname, "netout", ts) + systemCPU := sampler.SampleMetric("agent", snapshot.System.Hostname, "cpu", ts) + systemMemory := sampler.SampleMetric("agent", snapshot.System.Hostname, "memory", ts) + systemDisk := sampler.SampleMetric("agent", snapshot.System.Hostname, "disk", ts) + systemDiskRead := sampler.SampleMetric("agent", snapshot.System.Hostname, "diskread", ts) + systemDiskWrite := sampler.SampleMetric("agent", snapshot.System.Hostname, "diskwrite", ts) + systemNetIn := sampler.SampleMetric("agent", snapshot.System.Hostname, "netin", ts) + systemNetOut := sampler.SampleMetric("agent", snapshot.System.Hostname, "netout", ts) systemMetricIDs := []string{ "system:" + snapshot.System.Hostname, "agent:" + snapshot.System.Hostname, @@ -1096,7 +1107,7 @@ func recordTrueNASFixturesMetrics(mh *MetricsHistory, ms *metrics.Store, fixture if pool.TotalBytes > 0 { poolKey := mock.TrueNASPoolMetricID(snapshot.System.Hostname, pool.Name) total := float64(pool.TotalBytes) - usage := clampFloat(mock.SampleMetric("storage", poolKey, "usage", ts), 0, 100) + usage := clampFloat(sampler.SampleMetric("storage", poolKey, "usage", ts), 0, 100) used := total * (usage / 100.0) avail := math.Max(0, total-used) mh.AddGuestMetric(poolKey, "disk", usage, ts) @@ -1119,7 +1130,7 @@ func recordTrueNASFixturesMetrics(mh *MetricsHistory, ms *metrics.Store, fixture if totalBytes > 0 { dsKey := mock.TrueNASDatasetMetricID(snapshot.System.Hostname, dataset.Name) total := float64(totalBytes) - usage := clampFloat(mock.SampleMetric("storage", dsKey, "usage", ts), 0, 100) + usage := clampFloat(sampler.SampleMetric("storage", dsKey, "usage", ts), 0, 100) used := total * (usage / 100.0) avail := math.Max(0, total-used) mh.AddGuestMetric(dsKey, "disk", usage, ts) @@ -1145,10 +1156,10 @@ func recordTrueNASFixturesMetrics(mh *MetricsHistory, ms *metrics.Store, fixture if resourceID == "" { continue } - temp := mock.SampleMetric("disk", resourceID, "smart_temp", ts) - busy := mock.SampleMetric("disk", resourceID, "disk", ts) - diskRead := mock.SampleMetric("disk", resourceID, "diskread", ts) - diskWrite := mock.SampleMetric("disk", resourceID, "diskwrite", ts) + temp := sampler.SampleMetric("disk", resourceID, "smart_temp", ts) + busy := sampler.SampleMetric("disk", resourceID, "disk", ts) + diskRead := sampler.SampleMetric("disk", resourceID, "diskread", ts) + diskWrite := sampler.SampleMetric("disk", resourceID, "diskwrite", ts) mh.AddDiskMetric(resourceID, "smart_temp", temp, ts) mh.AddDiskMetric(resourceID, "disk", busy, ts) @@ -1167,13 +1178,13 @@ func recordTrueNASFixturesMetrics(mh *MetricsHistory, ms *metrics.Store, fixture if strings.TrimSpace(appID) == "" { continue } - cpu := mock.SampleMetric("dockerContainer", appID, "cpu", ts) - memPercent := mock.SampleMetric("dockerContainer", appID, "memory", ts) - diskPercent := mock.SampleMetric("dockerContainer", appID, "disk", ts) - diskRead := mock.SampleMetric("dockerContainer", appID, "diskread", ts) - diskWrite := mock.SampleMetric("dockerContainer", appID, "diskwrite", ts) - netIn := mock.SampleMetric("dockerContainer", appID, "netin", ts) - netOut := mock.SampleMetric("dockerContainer", appID, "netout", ts) + cpu := sampler.SampleMetric("dockerContainer", appID, "cpu", ts) + memPercent := sampler.SampleMetric("dockerContainer", appID, "memory", ts) + diskPercent := sampler.SampleMetric("dockerContainer", appID, "disk", ts) + diskRead := sampler.SampleMetric("dockerContainer", appID, "diskread", ts) + diskWrite := sampler.SampleMetric("dockerContainer", appID, "diskwrite", ts) + netIn := sampler.SampleMetric("dockerContainer", appID, "netin", ts) + netOut := sampler.SampleMetric("dockerContainer", appID, "netout", ts) metricKey := "docker:" + appID mh.AddGuestMetric(metricKey, "cpu", cpu, ts) mh.AddGuestMetric(metricKey, "memory", memPercent, ts) @@ -1194,7 +1205,7 @@ func recordTrueNASFixturesMetrics(mh *MetricsHistory, ms *metrics.Store, fixture } } -func recordVMwareFixturesMetrics(mh *MetricsHistory, ms *metrics.Store, fixtures mock.PlatformFixtures, ts time.Time) { +func recordVMwareFixturesMetrics(mh *MetricsHistory, ms *metrics.Store, sampler mock.MetricSampler, fixtures mock.PlatformFixtures, ts time.Time) { snapshot := fixtures.VMware for _, host := range snapshot.Hosts { @@ -1202,13 +1213,13 @@ func recordVMwareFixturesMetrics(mh *MetricsHistory, ms *metrics.Store, fixtures if sourceID == "" { continue } - cpu := mock.SampleMetric("agent", sourceID, "cpu", ts) - memory := mock.SampleMetric("agent", sourceID, "memory", ts) - disk := mock.SampleMetric("agent", sourceID, "disk", ts) - diskRead := mock.SampleMetric("agent", sourceID, "diskread", ts) - diskWrite := mock.SampleMetric("agent", sourceID, "diskwrite", ts) - netIn := mock.SampleMetric("agent", sourceID, "netin", ts) - netOut := mock.SampleMetric("agent", sourceID, "netout", ts) + cpu := sampler.SampleMetric("agent", sourceID, "cpu", ts) + memory := sampler.SampleMetric("agent", sourceID, "memory", ts) + disk := sampler.SampleMetric("agent", sourceID, "disk", ts) + diskRead := sampler.SampleMetric("agent", sourceID, "diskread", ts) + diskWrite := sampler.SampleMetric("agent", sourceID, "diskwrite", ts) + netIn := sampler.SampleMetric("agent", sourceID, "netin", ts) + netOut := sampler.SampleMetric("agent", sourceID, "netout", ts) mh.AddGuestMetric("agent:"+sourceID, "cpu", cpu, ts) mh.AddGuestMetric("agent:"+sourceID, "memory", memory, ts) mh.AddGuestMetric("agent:"+sourceID, "disk", disk, ts) @@ -1232,13 +1243,13 @@ func recordVMwareFixturesMetrics(mh *MetricsHistory, ms *metrics.Store, fixtures if sourceID == "" { continue } - cpu := mock.SampleMetric("vm", sourceID, "cpu", ts) - memory := mock.SampleMetric("vm", sourceID, "memory", ts) - disk := mock.SampleMetric("vm", sourceID, "disk", ts) - diskRead := mock.SampleMetric("vm", sourceID, "diskread", ts) - diskWrite := mock.SampleMetric("vm", sourceID, "diskwrite", ts) - netIn := mock.SampleMetric("vm", sourceID, "netin", ts) - netOut := mock.SampleMetric("vm", sourceID, "netout", ts) + cpu := sampler.SampleMetric("vm", sourceID, "cpu", ts) + memory := sampler.SampleMetric("vm", sourceID, "memory", ts) + disk := sampler.SampleMetric("vm", sourceID, "disk", ts) + diskRead := sampler.SampleMetric("vm", sourceID, "diskread", ts) + diskWrite := sampler.SampleMetric("vm", sourceID, "diskwrite", ts) + netIn := sampler.SampleMetric("vm", sourceID, "netin", ts) + netOut := sampler.SampleMetric("vm", sourceID, "netout", ts) mh.AddGuestMetric(sourceID, "cpu", cpu, ts) mh.AddGuestMetric(sourceID, "memory", memory, ts) mh.AddGuestMetric(sourceID, "disk", disk, ts) @@ -1262,7 +1273,7 @@ func recordVMwareFixturesMetrics(mh *MetricsHistory, ms *metrics.Store, fixtures if sourceID == "" { continue } - usage := clampFloat(mock.SampleMetric("storage", sourceID, "usage", ts), 0, 100) + usage := clampFloat(sampler.SampleMetric("storage", sourceID, "usage", ts), 0, 100) total := datastore.Capacity used := int64(float64(total) * (usage / 100.0)) if used > total { @@ -1300,20 +1311,20 @@ type containerAdapter struct{ *models.Container } func (c containerAdapter) GetID() string { return c.Container.ID } func (c containerAdapter) GetStatus() string { return c.Container.Status } -func recordGuestMetrics[T guestMetricSource](mh *MetricsHistory, ms *metrics.Store, guests []T, prefix string, ts time.Time) { +func recordGuestMetrics[T guestMetricSource](mh *MetricsHistory, ms *metrics.Store, sampler mock.MetricSampler, guests []T, prefix string, ts time.Time) { for _, guest := range guests { if guest.GetID() == "" || guest.GetStatus() != "running" { continue } id := guest.GetID() - cpu := mock.SampleMetric(prefix, id, "cpu", ts) - memory := mock.SampleMetric(prefix, id, "memory", ts) - disk := mock.SampleMetric(prefix, id, "disk", ts) - diskread := mock.SampleMetric(prefix, id, "diskread", ts) - diskwrite := mock.SampleMetric(prefix, id, "diskwrite", ts) - netin := mock.SampleMetric(prefix, id, "netin", ts) - netout := mock.SampleMetric(prefix, id, "netout", ts) + cpu := sampler.SampleMetric(prefix, id, "cpu", ts) + memory := sampler.SampleMetric(prefix, id, "memory", ts) + disk := sampler.SampleMetric(prefix, id, "disk", ts) + diskread := sampler.SampleMetric(prefix, id, "diskread", ts) + diskwrite := sampler.SampleMetric(prefix, id, "diskwrite", ts) + netin := sampler.SampleMetric(prefix, id, "netin", ts) + netout := sampler.SampleMetric(prefix, id, "netout", ts) mh.AddGuestMetric(id, "cpu", cpu, ts) mh.AddGuestMetric(id, "memory", memory, ts) @@ -1356,15 +1367,16 @@ func recordMockStateToMetricsHistory(mh *MetricsHistory, ms *metrics.Store, grap return } state := graph.State + sampler := mock.NewMetricSampler(graph) for _, node := range state.Nodes { if node.ID == "" || node.Status != "online" { continue } - cpu := mock.SampleMetric("node", node.ID, "cpu", ts) - memory := mock.SampleMetric("node", node.ID, "memory", ts) - disk := mock.SampleMetric("node", node.ID, "disk", ts) + cpu := sampler.SampleMetric("node", node.ID, "cpu", ts) + memory := sampler.SampleMetric("node", node.ID, "memory", ts) + disk := sampler.SampleMetric("node", node.ID, "disk", ts) mh.AddNodeMetric(node.ID, "cpu", cpu, ts) mh.AddNodeMetric(node.ID, "memory", memory, ts) mh.AddNodeMetric(node.ID, "disk", disk, ts) @@ -1382,18 +1394,18 @@ func recordMockStateToMetricsHistory(mh *MetricsHistory, ms *metrics.Store, grap } } - recordGuestMetrics(mh, ms, adaptVMs(state.VMs), "vm", ts) - recordGuestMetrics(mh, ms, adaptContainers(state.Containers), "container", ts) + recordGuestMetrics(mh, ms, sampler, adaptVMs(state.VMs), "vm", ts) + recordGuestMetrics(mh, ms, sampler, adaptContainers(state.Containers), "container", ts) recordKubernetesMetric := func(metricID string, includeDiskIO bool) { if metricID == "" { return } - cpu := mock.SampleMetric("k8s", metricID, "cpu", ts) - memory := mock.SampleMetric("k8s", metricID, "memory", ts) - disk := mock.SampleMetric("k8s", metricID, "disk", ts) - netIn := mock.SampleMetric("k8s", metricID, "netin", ts) - netOut := mock.SampleMetric("k8s", metricID, "netout", ts) + cpu := sampler.SampleMetric("k8s", metricID, "cpu", ts) + memory := sampler.SampleMetric("k8s", metricID, "memory", ts) + disk := sampler.SampleMetric("k8s", metricID, "disk", ts) + netIn := sampler.SampleMetric("k8s", metricID, "netin", ts) + netOut := sampler.SampleMetric("k8s", metricID, "netout", ts) mh.AddGuestMetric(metricID, "cpu", cpu, ts) mh.AddGuestMetric(metricID, "memory", memory, ts) @@ -1413,8 +1425,8 @@ func recordMockStateToMetricsHistory(mh *MetricsHistory, ms *metrics.Store, grap return } - diskRead := mock.SampleMetric("k8s", metricID, "diskread", ts) - diskWrite := mock.SampleMetric("k8s", metricID, "diskwrite", ts) + diskRead := sampler.SampleMetric("k8s", metricID, "diskread", ts) + diskWrite := sampler.SampleMetric("k8s", metricID, "diskwrite", ts) mh.AddGuestMetric(metricID, "diskread", diskRead, ts) mh.AddGuestMetric(metricID, "diskwrite", diskWrite, ts) if ms != nil { @@ -1448,7 +1460,7 @@ func recordMockStateToMetricsHistory(mh *MetricsHistory, ms *metrics.Store, grap if total <= 0 { total = currentUsed + currentAvail } - usage := mock.SampleMetric("storage", storage.ID, "usage", ts) + usage := sampler.SampleMetric("storage", storage.ID, "usage", ts) usage = clampFloat(usage, 0, 100) used := total * (usage / 100.0) if used < 0 { @@ -1479,10 +1491,10 @@ func recordMockStateToMetricsHistory(mh *MetricsHistory, ms *metrics.Store, grap if resourceID == "" { continue } - temp := mock.SampleMetric("disk", resourceID, "smart_temp", ts) - busy := mock.SampleMetric("disk", resourceID, "disk", ts) - diskRead := mock.SampleMetric("disk", resourceID, "diskread", ts) - diskWrite := mock.SampleMetric("disk", resourceID, "diskwrite", ts) + temp := sampler.SampleMetric("disk", resourceID, "smart_temp", ts) + busy := sampler.SampleMetric("disk", resourceID, "disk", ts) + diskRead := sampler.SampleMetric("disk", resourceID, "diskread", ts) + diskWrite := sampler.SampleMetric("disk", resourceID, "diskwrite", ts) mh.AddDiskMetric(resourceID, "smart_temp", temp, ts) mh.AddDiskMetric(resourceID, "disk", busy, ts) @@ -1519,9 +1531,9 @@ func recordMockStateToMetricsHistory(mh *MetricsHistory, ms *metrics.Store, grap } hostKey := "dockerHost:" + host.ID - cpu := mock.SampleMetric("dockerHost", host.ID, "cpu", ts) - memory := mock.SampleMetric("dockerHost", host.ID, "memory", ts) - disk := mock.SampleMetric("dockerHost", host.ID, "disk", ts) + cpu := sampler.SampleMetric("dockerHost", host.ID, "cpu", ts) + memory := sampler.SampleMetric("dockerHost", host.ID, "memory", ts) + disk := sampler.SampleMetric("dockerHost", host.ID, "disk", ts) mh.AddGuestMetric(hostKey, "cpu", cpu, ts) mh.AddGuestMetric(hostKey, "memory", memory, ts) mh.AddGuestMetric(hostKey, "disk", disk, ts) @@ -1538,9 +1550,9 @@ func recordMockStateToMetricsHistory(mh *MetricsHistory, ms *metrics.Store, grap } metricKey := "docker:" + container.ID - cpu := mock.SampleMetric("dockerContainer", container.ID, "cpu", ts) - memory := mock.SampleMetric("dockerContainer", container.ID, "memory", ts) - disk := mock.SampleMetric("dockerContainer", container.ID, "disk", ts) + cpu := sampler.SampleMetric("dockerContainer", container.ID, "cpu", ts) + memory := sampler.SampleMetric("dockerContainer", container.ID, "memory", ts) + disk := sampler.SampleMetric("dockerContainer", container.ID, "disk", ts) mh.AddGuestMetric(metricKey, "cpu", cpu, ts) mh.AddGuestMetric(metricKey, "memory", memory, ts) mh.AddGuestMetric(metricKey, "disk", disk, ts) @@ -1559,13 +1571,13 @@ func recordMockStateToMetricsHistory(mh *MetricsHistory, ms *metrics.Store, grap } hostKey := "agent:" + host.ID - cpu := mock.SampleMetric("agent", host.ID, "cpu", ts) - memory := mock.SampleMetric("agent", host.ID, "memory", ts) - disk := mock.SampleMetric("agent", host.ID, "disk", ts) - diskread := mock.SampleMetric("agent", host.ID, "diskread", ts) - diskwrite := mock.SampleMetric("agent", host.ID, "diskwrite", ts) - netin := mock.SampleMetric("agent", host.ID, "netin", ts) - netout := mock.SampleMetric("agent", host.ID, "netout", ts) + cpu := sampler.SampleMetric("agent", host.ID, "cpu", ts) + memory := sampler.SampleMetric("agent", host.ID, "memory", ts) + disk := sampler.SampleMetric("agent", host.ID, "disk", ts) + diskread := sampler.SampleMetric("agent", host.ID, "diskread", ts) + diskwrite := sampler.SampleMetric("agent", host.ID, "diskwrite", ts) + netin := sampler.SampleMetric("agent", host.ID, "netin", ts) + netout := sampler.SampleMetric("agent", host.ID, "netout", ts) temperature := hostPrimaryTemperatureCelsius(host.Sensors) mh.AddGuestMetric(hostKey, "cpu", cpu, ts) mh.AddGuestMetric(hostKey, "memory", memory, ts) @@ -1593,8 +1605,8 @@ func recordMockStateToMetricsHistory(mh *MetricsHistory, ms *metrics.Store, grap } // Record TrueNAS pool/dataset disk-usage live ticks - recordTrueNASFixturesMetrics(mh, ms, graph.PlatformFixtures, ts) - recordVMwareFixturesMetrics(mh, ms, graph.PlatformFixtures, ts) + recordTrueNASFixturesMetrics(mh, ms, sampler, graph.PlatformFixtures, ts) + recordVMwareFixturesMetrics(mh, ms, sampler, graph.PlatformFixtures, ts) } func diskMetricsResourceID(disk models.PhysicalDisk) string { diff --git a/internal/monitoring/mock_metrics_history_test.go b/internal/monitoring/mock_metrics_history_test.go index 14f005414..8de6f8c0e 100644 --- a/internal/monitoring/mock_metrics_history_test.go +++ b/internal/monitoring/mock_metrics_history_test.go @@ -1578,6 +1578,45 @@ func TestRecordMockStateToMetricsHistory_ContinuesCanonicalKubernetesClusterNode } } +func TestMockMetricsHistoryKeepsGraphPersonaAcrossSeedAndLiveBoundary(t *testing.T) { + now := time.Now().UTC().Truncate(time.Minute) + next := now.Add(time.Minute) + resourceID := "persona-boundary-7c9f0b33" + graph := fixtureGraphWithState(models.StateSnapshot{ + Containers: []models.Container{{ + ID: resourceID, + Name: "backup-orchestrator", + Status: "running", + }}, + }) + sampler := mock.NewMetricSampler(graph) + + mh := NewMetricsHistory(5000, boundedMockHistoryProofWindow) + seedMockMetricsHistory(mh, nil, graph, now, boundedMockHistoryProofWindow, time.Minute) + recordMockStateToMetricsHistory(mh, nil, graph, next) + + series := mh.GetGuestMetrics(resourceID, "memory", boundedMockHistoryProofWindow) + if len(series) < 2 { + t.Fatalf("expected seeded and live memory points, got %d", len(series)) + } + for _, expected := range []struct { + index int + at time.Time + }{ + {index: len(series) - 2, at: now}, + {index: len(series) - 1, at: next}, + } { + point := series[expected.index] + if !point.Timestamp.Equal(expected.at) { + t.Fatalf("point %d timestamp = %v, want %v", expected.index, point.Timestamp, expected.at) + } + want := sampler.SampleMetric("container", resourceID, "memory", expected.at) + if diff := math.Abs(point.Value - want); diff > 1e-9 { + t.Fatalf("point %d value = %f, want graph-bound %f", expected.index, point.Value, want) + } + } +} + func TestSeedMockMetricsHistory_UsesCanonicalMetricModelForPlatformFixtures(t *testing.T) { ts := time.Now().UTC().Truncate(time.Minute) graph := mock.FixtureGraph{ diff --git a/internal/monitoring/multi_tenant_monitor.go b/internal/monitoring/multi_tenant_monitor.go index 08b0b678e..9bad36342 100644 --- a/internal/monitoring/multi_tenant_monitor.go +++ b/internal/monitoring/multi_tenant_monitor.go @@ -15,31 +15,33 @@ import ( // MultiTenantMonitor manages a dedicated Monitor instance for each organization. type MultiTenantMonitor struct { - mu sync.RWMutex - monitors map[string]*Monitor - tenantCancel map[string]context.CancelFunc - tenantDone map[string]chan struct{} - persistence *config.MultiTenantPersistence - baseConfig *config.Config - wsHub *websocket.Hub - recoveryMgr *recoverymanager.Manager - initializer func(*Monitor) - globalCtx context.Context - globalCancel context.CancelFunc + mu sync.RWMutex + monitors map[string]*Monitor + tenantCancel map[string]context.CancelFunc + tenantDone map[string]chan struct{} + tenantDeleting map[string]struct{} + persistence *config.MultiTenantPersistence + baseConfig *config.Config + wsHub *websocket.Hub + recoveryMgr *recoverymanager.Manager + initializer func(*Monitor) + globalCtx context.Context + globalCancel context.CancelFunc } // NewMultiTenantMonitor creates a new multi-tenant monitor manager. func NewMultiTenantMonitor(baseCfg *config.Config, persistence *config.MultiTenantPersistence, wsHub *websocket.Hub) *MultiTenantMonitor { ctx, cancel := context.WithCancel(context.Background()) return &MultiTenantMonitor{ - monitors: make(map[string]*Monitor), - tenantCancel: make(map[string]context.CancelFunc), - tenantDone: make(map[string]chan struct{}), - persistence: persistence, - baseConfig: baseCfg, // Used as a template or for global settings - wsHub: wsHub, - globalCtx: ctx, - globalCancel: cancel, + monitors: make(map[string]*Monitor), + tenantCancel: make(map[string]context.CancelFunc), + tenantDone: make(map[string]chan struct{}), + tenantDeleting: make(map[string]struct{}), + persistence: persistence, + baseConfig: baseCfg, // Used as a template or for global settings + wsHub: wsHub, + globalCtx: ctx, + globalCancel: cancel, } } @@ -137,8 +139,12 @@ func (mtm *MultiTenantMonitor) GetMonitor(orgID string) (*Monitor, error) { mtm.mu.RLock() monitor, exists := mtm.monitors[orgID] + _, deleting := mtm.tenantDeleting[orgID] mtm.mu.RUnlock() + if deleting { + return nil, fmt.Errorf("organization %q is being deleted", orgID) + } if exists { return monitor, nil } @@ -155,8 +161,14 @@ func (mtm *MultiTenantMonitor) GetMonitor(orgID string) (*Monitor, error) { if mtm.tenantDone == nil { mtm.tenantDone = make(map[string]chan struct{}) } + if mtm.tenantDeleting == nil { + mtm.tenantDeleting = make(map[string]struct{}) + } // Double-check locking pattern + if _, deleting = mtm.tenantDeleting[orgID]; deleting { + return nil, fmt.Errorf("organization %q is being deleted", orgID) + } if monitor, exists = mtm.monitors[orgID]; exists { return monitor, nil } @@ -305,12 +317,43 @@ func (mtm *MultiTenantMonitor) Stop() { // RemoveTenant stops and removes a specific tenant's monitor. // Useful for offboarding or manual reloading. func (mtm *MultiTenantMonitor) RemoveTenant(orgID string) { + mtm.removeTenant(orgID, false) +} + +// BeginTenantDeletion prevents lazy monitor initialization while a tenant's +// persistence is being removed. Call FinishTenantDeletion after the +// persistence operation completes, whether it succeeds or fails. +func (mtm *MultiTenantMonitor) BeginTenantDeletion(orgID string) { + mtm.removeTenant(orgID, true) +} + +// FinishTenantDeletion releases the temporary lifecycle guard installed by +// BeginTenantDeletion. Once persistence has been removed, GetMonitor remains +// unable to recreate the tenant because the organization no longer exists. +func (mtm *MultiTenantMonitor) FinishTenantDeletion(orgID string) { orgID = strings.TrimSpace(orgID) if orgID == "" { return } mtm.mu.Lock() + delete(mtm.tenantDeleting, orgID) + mtm.mu.Unlock() +} + +func (mtm *MultiTenantMonitor) removeTenant(orgID string, deleting bool) { + orgID = strings.TrimSpace(orgID) + if orgID == "" { + return + } + + mtm.mu.Lock() + if deleting { + if mtm.tenantDeleting == nil { + mtm.tenantDeleting = make(map[string]struct{}) + } + mtm.tenantDeleting[orgID] = struct{}{} + } monitor, exists := mtm.monitors[orgID] cancel := mtm.tenantCancel[orgID] done := mtm.tenantDone[orgID] diff --git a/internal/monitoring/multi_tenant_monitor_additional_test.go b/internal/monitoring/multi_tenant_monitor_additional_test.go index f00176c16..aad06d4ea 100644 --- a/internal/monitoring/multi_tenant_monitor_additional_test.go +++ b/internal/monitoring/multi_tenant_monitor_additional_test.go @@ -75,6 +75,39 @@ func TestMultiTenantMonitorRemoveTenantCancelsTenantRuntime(t *testing.T) { } } +func TestMultiTenantMonitorTenantDeletionBlocksLazyRecreation(t *testing.T) { + mtp, _ := newTestTenantPersistence(t) + baseCfg := &config.Config{DataPath: t.TempDir()} + mtm := NewMultiTenantMonitor(baseCfg, mtp, nil) + t.Cleanup(mtm.Stop) + + if err := mtp.SaveOrganization(&models.Organization{ + ID: "org-delete", + DisplayName: "Org Delete", + }); err != nil { + t.Fatalf("SaveOrganization(org-delete) error = %v", err) + } + + first, err := mtm.GetMonitor("org-delete") + if err != nil { + t.Fatalf("GetMonitor(org-delete) error = %v", err) + } + + mtm.BeginTenantDeletion("org-delete") + if _, err := mtm.GetMonitor("org-delete"); err == nil { + t.Fatal("expected tenant deletion guard to block lazy monitor recreation") + } + + mtm.FinishTenantDeletion("org-delete") + second, err := mtm.GetMonitor("org-delete") + if err != nil { + t.Fatalf("GetMonitor(org-delete) after deletion guard error = %v", err) + } + if second == first { + t.Fatal("expected a fresh monitor after the deletion guard is released") + } +} + func TestMultiTenantMonitorGetMonitor_MetricsIsolationByTenant(t *testing.T) { orgAMonitor := &Monitor{ metricsHistory: NewMetricsHistory(32, time.Hour), diff --git a/tests/integration/tests/05-settings-mobile-audit.spec.ts b/tests/integration/tests/05-settings-mobile-audit.spec.ts index 8fc39e734..bc1979949 100644 --- a/tests/integration/tests/05-settings-mobile-audit.spec.ts +++ b/tests/integration/tests/05-settings-mobile-audit.spec.ts @@ -128,10 +128,17 @@ const prepareOrganizationAuditFixture = async ( { method: "POST", data: { userId: "mobile-audit-viewer", role: "viewer" }, - headers: { "Content-Type": "application/json" }, + headers: { + "Content-Type": "application/json", + "X-Org-ID": source.id, + "X-Pulse-Org-ID": source.id, + }, }, ); - expect(invitation.ok(), "mobile access fixture invitation").toBeTruthy(); + expect( + invitation.ok(), + `mobile access fixture invitation: ${invitation.status()} ${await invitation.text()}`, + ).toBeTruthy(); } if (route === "/settings/organization/sharing") { @@ -147,10 +154,17 @@ const prepareOrganizationAuditFixture = async ( resourceName: "Mobile Audit Shared View", accessRole: "viewer", }, - headers: { "Content-Type": "application/json" }, + headers: { + "Content-Type": "application/json", + "X-Org-ID": source.id, + "X-Pulse-Org-ID": source.id, + }, }, ); - expect(share.ok(), "mobile sharing fixture").toBeTruthy(); + expect( + share.ok(), + `mobile sharing fixture: ${share.status()} ${await share.text()}`, + ).toBeTruthy(); } return createdOrgIDs; @@ -330,10 +344,6 @@ test.describe("Settings mobile optimization audit", () => { await expect(organizationSwitcher).toHaveValue("default", { timeout: 20000, }); - await page.reload({ waitUntil: "domcontentloaded" }); - await expect( - page.getByRole("combobox", { name: "Organization" }), - ).toHaveValue("default", { timeout: 20000 }); } } for (const orgID of [...createdOrgIDs].reverse()) { @@ -348,9 +358,11 @@ test.describe("Settings mobile optimization audit", () => { }, }, ); + const responseBody = + response.status() === 204 ? "" : await response.text(); expect( [204, 404].includes(response.status()), - `cleanup organization ${orgID}`, + `cleanup organization ${orgID}: ${response.status()} ${responseBody}`, ).toBeTruthy(); } } diff --git a/tests/integration/tests/42-vmware-ai-chat-read-recovery.spec.ts b/tests/integration/tests/42-vmware-ai-chat-read-recovery.spec.ts index a2b794759..91515a7d8 100644 --- a/tests/integration/tests/42-vmware-ai-chat-read-recovery.spec.ts +++ b/tests/integration/tests/42-vmware-ai-chat-read-recovery.spec.ts @@ -3,6 +3,7 @@ import path from 'node:path'; import { fileURLToPath } from 'node:url'; import { test as base, expect } from '@playwright/test'; import { createAuthenticatedStorageState } from './helpers'; +import { installVmwareWorkloadResourceRoute } from './vmware-workload-fixture'; const __dirname = path.dirname(fileURLToPath(import.meta.url)); const SCREENSHOT_PATH = '/tmp/vmware-ai-chat-read-recovery.png'; @@ -15,23 +16,26 @@ const test = base.extend<{}, WorkerFixtures>({ storageState: async ({ authStorageStatePath }, use) => { await use(authStorageStatePath); }, - authStorageStatePath: [async ({ browser }, use, workerInfo) => { - const storageStatePath = path.resolve( - __dirname, - '..', - '..', - 'tmp', - 'playwright-auth', - `vmware-ai-chat-read-recovery-${workerInfo.project.name}.json`, - ); - fs.mkdirSync(path.dirname(storageStatePath), { recursive: true }); - await createAuthenticatedStorageState(browser, storageStatePath); - try { - await use(storageStatePath); - } finally { - fs.rmSync(storageStatePath, { force: true }); - } - }, { scope: 'worker' }], + authStorageStatePath: [ + async ({ browser }, use, workerInfo) => { + const storageStatePath = path.resolve( + __dirname, + '..', + '..', + 'tmp', + 'playwright-auth', + `vmware-ai-chat-read-recovery-${workerInfo.project.name}.json`, + ); + fs.mkdirSync(path.dirname(storageStatePath), { recursive: true }); + await createAuthenticatedStorageState(browser, storageStatePath); + try { + await use(storageStatePath); + } finally { + fs.rmSync(storageStatePath, { force: true }); + } + }, + { scope: 'worker' }, + ], }); // The assistant transport moved from the retired /api/ai/chat SSE endpoint @@ -57,6 +61,8 @@ test.describe('VMware AI chat read recovery', () => { await route.abort(); }); + await installVmwareWorkloadResourceRoute(page); + await page.goto('/vmware', { waitUntil: 'domcontentloaded' }); await expect(page.locator('[data-testid="vmware-page"]')).toBeVisible(); @@ -67,7 +73,9 @@ test.describe('VMware AI chat read recovery', () => { // First websocket state frame can lag on a freshly booted backend. await textarea.focus(); await textarea.fill('@warehouse'); - const mentionListbox = page.getByRole('listbox', { name: 'Assistant resources' }); + const mentionListbox = page.getByRole('listbox', { + name: 'Assistant resources', + }); await expect(mentionListbox).toBeVisible({ timeout: 30_000 }); const vmOption = mentionListbox .getByRole('option') @@ -77,12 +85,15 @@ test.describe('VMware AI chat read recovery', () => { await page.keyboard.press('Enter'); await expect(textarea).toHaveValue('@warehouse-api-01 '); - await textarea.fill('@warehouse-api-01 show me recent status'); + await textarea.pressSequentially('show me recent status'); + await expect(textarea).toHaveValue('@warehouse-api-01 show me recent status'); // Submit through the composer control that phone users actually tap. // A page-global keyboard action is not a stable submission primitive in // mobile WebKit and can remain pending after the textarea has accepted the // completed prompt. - await page.getByRole('button', { name: 'Send message' }).click(); + const sendButton = page.getByRole('button', { name: 'Send message' }); + await expect(sendButton).toBeEnabled(); + await sendButton.click(); // Chat turns ride the shared assistant transport (websocket-backed since // the assistant modernization, so REST route interception cannot observe @@ -90,9 +101,9 @@ test.describe('VMware AI chat read recovery', () => { // user message lands in the conversation and the mock assistant answers // the turn, all without the browser touching provider-local vmware // endpoints. - await expect( - page.getByText('@warehouse-api-01 show me recent status').first(), - ).toBeVisible({ timeout: 30_000 }); + await expect(page.getByText('@warehouse-api-01 show me recent status').first()).toBeVisible({ + timeout: 30_000, + }); await expect(page.getByText('Pulse mock Assistant').first()).toBeVisible({ timeout: 30_000, }); diff --git a/tests/integration/tests/vmware-workload-fixture.ts b/tests/integration/tests/vmware-workload-fixture.ts index 62ce92f3b..84224ce83 100644 --- a/tests/integration/tests/vmware-workload-fixture.ts +++ b/tests/integration/tests/vmware-workload-fixture.ts @@ -1,8 +1,5 @@ import type { Page } from '@playwright/test'; -const VMWARE_WORKLOAD_RESOURCE_URL = - /\/api\/resources\?type=vm(?:,|%2C)system-container(?:,|%2C)app-container(?:,|%2C)pod&page=1&limit=200$/i; - const VMWARE_WORKLOAD_RESOURCE = { id: 'vmware:vc-mock-1:vm:vm-201', type: 'vm', @@ -57,14 +54,23 @@ const VMWARE_WORKLOAD_RESOURCE = { } as const; export async function installVmwareWorkloadResourceRoute(page: Page): Promise { - await page.route(VMWARE_WORKLOAD_RESOURCE_URL, async (route) => { + await page.route('**/api/resources?*', async (route) => { + const url = new URL(route.request().url()); + const resourceTypes = (url.searchParams.get('type') ?? '').split(','); + if (!resourceTypes.includes('vm')) { + await route.continue(); + return; + } + + const pageNumber = Number(url.searchParams.get('page') ?? '1'); + const limit = Number(url.searchParams.get('limit') ?? '100'); await route.fulfill({ status: 200, json: { - data: [VMWARE_WORKLOAD_RESOURCE], + data: pageNumber === 1 ? [VMWARE_WORKLOAD_RESOURCE] : [], meta: { - page: 1, - limit: 200, + page: pageNumber, + limit, total: 1, totalPages: 1, },