Preserve resource URLs across runtime identity changes

Contract-Neutral: Docker URL migration touches shared monitor ingest files without changing the host deletion or re-enrollment contract.
This commit is contained in:
rcourtman
2026-07-23 22:20:47 +01:00
parent 0a95b25f41
commit 15dbabfe9a
43 changed files with 1790 additions and 134 deletions
+56 -2
View File
@@ -69,6 +69,7 @@ type AISettingsHandler struct {
policyMutation func(func() error) error
patrolAutopilotPolicy func() unifiedresources.PatrolAutopilotServerPolicy
metadataProvider ai.MetadataProvider
metadataProviderFactory func(orgID string) ai.MetadataProvider
patrolThresholdProvider ai.ThresholdProvider
metricsHistoryProvider ai.MetricsHistoryProvider
baselineStore *ai.BaselineStore
@@ -237,6 +238,7 @@ type aiSettingsProviderSnapshot struct {
defaultAIService *ai.Service
stateProvider ai.StateProvider
metadataProvider ai.MetadataProvider
metadataProviderFactory func(orgID string) ai.MetadataProvider
patrolThresholdProvider ai.ThresholdProvider
metricsHistoryProvider ai.MetricsHistoryProvider
baselineStore *ai.BaselineStore
@@ -268,6 +270,7 @@ func (h *AISettingsHandler) providerSnapshot() aiSettingsProviderSnapshot {
defaultAIService: h.defaultAIService,
stateProvider: h.stateProvider,
metadataProvider: h.metadataProvider,
metadataProviderFactory: h.metadataProviderFactory,
patrolThresholdProvider: h.patrolThresholdProvider,
metricsHistoryProvider: h.metricsHistoryProvider,
baselineStore: h.baselineStore,
@@ -415,6 +418,10 @@ func (h *AISettingsHandler) GetAIService(ctx context.Context) *ai.Service {
if svc, exists = h.aiServices[orgID]; exists {
return svc
}
// A provider factory can be replaced while this service is waiting on
// aiServicesMu during monitor reload. Refresh after acquiring the creation
// lock so the new service cannot retain the pre-reload metadata store.
providers = h.providerSnapshot()
// Create new service for this tenant
persistence, err := mtPersistence.GetPersistence(orgID)
@@ -449,8 +456,8 @@ func (h *AISettingsHandler) GetAIService(ctx context.Context) *ai.Service {
if provider := h.unifiedResourceProviderForOrg(orgID); provider != nil {
svc.SetUnifiedResourceProvider(provider)
}
if providers.metadataProvider != nil {
svc.SetMetadataProvider(providers.metadataProvider)
if metadataProvider := providers.metadataProviderForOrg(orgID); metadataProvider != nil {
svc.SetMetadataProvider(metadataProvider)
}
if orgID == "default" {
if providers.patrolThresholdProvider != nil {
@@ -752,10 +759,21 @@ func (h *AISettingsHandler) SetUnifiedResourceProvider(urp ai.UnifiedResourcePro
}
}
func (p aiSettingsProviderSnapshot) metadataProviderForOrg(orgID string) ai.MetadataProvider {
if p.metadataProviderFactory != nil {
return p.metadataProviderFactory(orgID)
}
return p.metadataProvider
}
// SetMetadataProvider sets the metadata provider for AI URL discovery
func (h *AISettingsHandler) SetMetadataProvider(mp ai.MetadataProvider) {
if h == nil {
return
}
h.stateMu.Lock()
h.metadataProvider = mp
h.metadataProviderFactory = nil
defaultAIService := h.defaultAIService
h.stateMu.Unlock()
if defaultAIService != nil {
@@ -769,6 +787,42 @@ func (h *AISettingsHandler) SetMetadataProvider(mp ai.MetadataProvider) {
}
}
// SetMetadataProviderFactory scopes URL metadata updates to each AI service's
// organization and refreshes providers already created before a monitor reload.
func (h *AISettingsHandler) SetMetadataProviderFactory(
factory func(orgID string) ai.MetadataProvider,
) {
if h == nil {
return
}
h.stateMu.Lock()
h.metadataProvider = nil
h.metadataProviderFactory = factory
defaultAIService := h.defaultAIService
h.stateMu.Unlock()
if defaultAIService != nil {
var provider ai.MetadataProvider
if factory != nil {
provider = factory("default")
}
defaultAIService.SetMetadataProvider(provider)
}
h.aiServicesMu.Lock()
defer h.aiServicesMu.Unlock()
for orgID, svc := range h.aiServices {
if svc == nil {
continue
}
var provider ai.MetadataProvider
if factory != nil {
provider = factory(orgID)
}
svc.SetMetadataProvider(provider)
}
}
// StartPatrol starts the background AI patrol service
func (h *AISettingsHandler) StartPatrol(ctx context.Context) {
h.GetAIService(ctx).StartPatrol(ctx)
+52
View File
@@ -24,6 +24,58 @@ func (stubMetadataProvider) SetGuestURL(string, string) error { return nil }
func (stubMetadataProvider) SetDockerURL(string, string) error { return nil }
func (stubMetadataProvider) SetHostURL(string, string) error { return nil }
type recordingMetadataProvider struct {
guestURLs map[string]string
}
func (p *recordingMetadataProvider) SetGuestURL(id, url string) error {
p.guestURLs[id] = url
return nil
}
func (*recordingMetadataProvider) SetDockerURL(string, string) error { return nil }
func (*recordingMetadataProvider) SetHostURL(string, string) error { return nil }
func TestAISettingsHandlerMetadataProviderFactoryScopesExistingServices(t *testing.T) {
handler := NewAISettingsHandler(config.NewMultiTenantPersistence(t.TempDir()), nil, nil)
tenantContext := context.WithValue(context.Background(), OrgIDContextKey, "tenant-a")
defaultService := handler.GetAIService(context.Background())
tenantService := handler.GetAIService(tenantContext)
if defaultService == nil || tenantService == nil {
t.Fatal("expected default and tenant AI services")
}
providers := map[string]*recordingMetadataProvider{}
handler.SetMetadataProviderFactory(func(orgID string) ai.MetadataProvider {
provider := &recordingMetadataProvider{guestURLs: make(map[string]string)}
providers[orgID] = provider
return provider
})
if err := defaultService.SetResourceURL("vm", "shared-id", "https://default.internal"); err != nil {
t.Fatalf("set default URL: %v", err)
}
if err := tenantService.SetResourceURL("vm", "shared-id", "https://tenant.internal"); err != nil {
t.Fatalf("set tenant URL: %v", err)
}
if got := providers["default"].guestURLs["shared-id"]; got != "https://default.internal" {
t.Fatalf("default provider URL = %q", got)
}
if got := providers["tenant-a"].guestURLs["shared-id"]; got != "https://tenant.internal" {
t.Fatalf("tenant provider URL = %q", got)
}
newTenantContext := context.WithValue(context.Background(), OrgIDContextKey, "tenant-b")
newTenantService := handler.GetAIService(newTenantContext)
if err := newTenantService.SetResourceURL("vm", "shared-id", "https://new-tenant.internal"); err != nil {
t.Fatalf("set new tenant URL: %v", err)
}
if got := providers["tenant-b"].guestURLs["shared-id"]; got != "https://new-tenant.internal" {
t.Fatalf("new tenant provider URL = %q", got)
}
}
type stubThresholdProvider struct{}
func (stubThresholdProvider) GetNodeCPUThreshold() float64 { return 80 }
@@ -186,7 +186,7 @@ func (h *ConfigHandlers) handleImportConfig(w http.ResponseWriter, r *http.Reque
// Reload guest metadata from disk.
if h.guestMetadataHandler != nil {
if err := h.guestMetadataHandler.Reload(); err != nil {
if err := h.guestMetadataHandler.Reload(r.Context()); err != nil {
log.Warn().Err(err).Msg("Failed to reload guest metadata after import")
} else {
log.Info().Msg("Reloaded guest metadata after import")
+16
View File
@@ -332,6 +332,7 @@ func (h *ConfigHandlers) getContextState(ctx context.Context) (*config.Config, *
// and setup state.
if orgID == "default" &&
(defaultConfig != nil || defaultPersistence != nil || defaultMonitor != nil) {
bindMonitorMetadataStores(defaultPersistence, defaultMonitor)
return defaultConfig, defaultPersistence, defaultMonitor
}
@@ -345,6 +346,7 @@ func (h *ConfigHandlers) getContextState(ctx context.Context) (*config.Config, *
if mtPersistence != nil {
p, _ = mtPersistence.GetPersistence(orgID)
}
bindMonitorMetadataStores(p, m)
return cfg, p, m
} else {
if orgID == "default" {
@@ -368,6 +370,20 @@ func (h *ConfigHandlers) getContextState(ctx context.Context) (*config.Config, *
return nil, nil, nil
}
func bindMonitorMetadataStores(
persistence *config.ConfigPersistence,
monitor *monitoring.Monitor,
) {
if persistence == nil || monitor == nil {
return
}
persistence.SetMetadataStores(
monitor.GuestMetadataStore(),
monitor.DockerMetadataStore(),
monitor.HostMetadataStore(),
)
}
func (h *ConfigHandlers) getConfig(ctx context.Context) *config.Config {
c, _, _ := h.getContextState(ctx)
return c
+14
View File
@@ -18,6 +18,7 @@ const (
// DockerMetadataHandler handles Docker resource metadata operations
type DockerMetadataHandler struct {
mtPersistence *config.MultiTenantPersistence
storeResolver func(context.Context) *config.DockerMetadataStore
}
// NewDockerMetadataHandler creates a new Docker metadata handler
@@ -27,7 +28,20 @@ func NewDockerMetadataHandler(mtPersistence *config.MultiTenantPersistence) *Doc
}
}
// SetStoreResolver makes API reads and writes use the active monitor's store.
// The persistence-backed store remains the initialization/test fallback.
func (h *DockerMetadataHandler) SetStoreResolver(
resolver func(context.Context) *config.DockerMetadataStore,
) {
h.storeResolver = resolver
}
func (h *DockerMetadataHandler) getStore(ctx context.Context) *config.DockerMetadataStore {
if h != nil && h.storeResolver != nil {
if store := h.storeResolver(ctx); store != nil {
return store
}
}
orgID := "default"
if ctx != nil {
if requestOrgID := GetOrgID(ctx); requestOrgID != "" {
+20 -9
View File
@@ -12,6 +12,7 @@ import (
// GuestMetadataHandler handles guest metadata operations
type GuestMetadataHandler struct {
mtPersistence *config.MultiTenantPersistence
storeResolver func(context.Context) *config.GuestMetadataStore
}
// NewGuestMetadataHandler creates a new guest metadata handler
@@ -21,7 +22,20 @@ func NewGuestMetadataHandler(mtPersistence *config.MultiTenantPersistence) *Gues
}
}
// SetStoreResolver makes API reads and writes use the active monitor's store.
// The persistence-backed store remains the initialization/test fallback.
func (h *GuestMetadataHandler) SetStoreResolver(
resolver func(context.Context) *config.GuestMetadataStore,
) {
h.storeResolver = resolver
}
func (h *GuestMetadataHandler) getStore(ctx context.Context) *config.GuestMetadataStore {
if h != nil && h.storeResolver != nil {
if store := h.storeResolver(ctx); store != nil {
return store
}
}
// Default to "default" org if none specified (though middleware should always set it)
orgID := "default"
if ctx != nil {
@@ -33,15 +47,12 @@ func (h *GuestMetadataHandler) getStore(ctx context.Context) *config.GuestMetada
return p.GetGuestMetadataStore()
}
// Reload reloads the guest metadata from disk
func (h *GuestMetadataHandler) Reload() error {
// For multi-tenant, we might need to reload all loaded stores?
// Or we just rely on lazy loading.
// Since stores are cached in ConfigPersistence, we currently don't have an easy way to iterate all.
// But stores load on init. Reload() method on store might be needed if modified on disk externally.
// For now, this is a no-op or TODO for multi-tenant deep reload.
// Actually, we can get "default" store and reload it for legacy compat.
return h.getStore(context.Background()).Load()
// Reload reloads the request tenant's guest metadata from disk.
func (h *GuestMetadataHandler) Reload(ctx context.Context) error {
if ctx == nil {
ctx = context.Background()
}
return h.getStore(ctx).Load()
}
// Store returns the underlying metadata store for the default tenant (Legacy support)
@@ -1,6 +1,7 @@
package api
import (
"context"
"testing"
"github.com/rcourtman/pulse-go-rewrite/internal/config"
@@ -10,7 +11,7 @@ func TestGuestMetadataHandler_Reload(t *testing.T) {
mtp := config.NewMultiTenantPersistence(t.TempDir())
handler := NewGuestMetadataHandler(mtp)
if err := handler.Reload(); err != nil {
if err := handler.Reload(context.Background()); err != nil {
t.Fatalf("Reload error: %v", err)
}
}
+14
View File
@@ -28,6 +28,7 @@ func hostMetadataPathParts(path string) (agentID string, isCollection bool, ok b
// HostMetadataHandler handles agent metadata operations.
type HostMetadataHandler struct {
mtPersistence *config.MultiTenantPersistence
storeResolver func(context.Context) *config.HostMetadataStore
}
// NewHostMetadataHandler creates a new host metadata handler
@@ -37,7 +38,20 @@ func NewHostMetadataHandler(mtPersistence *config.MultiTenantPersistence) *HostM
}
}
// SetStoreResolver makes API reads and writes use the active monitor's store.
// The persistence-backed store remains the initialization/test fallback.
func (h *HostMetadataHandler) SetStoreResolver(
resolver func(context.Context) *config.HostMetadataStore,
) {
h.storeResolver = resolver
}
func (h *HostMetadataHandler) getStore(ctx context.Context) *config.HostMetadataStore {
if h != nil && h.storeResolver != nil {
if store := h.storeResolver(ctx); store != nil {
return store
}
}
orgID := "default"
if ctx != nil {
if requestOrgID := GetOrgID(ctx); requestOrgID != "" {
+110
View File
@@ -1,6 +1,7 @@
package api
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
@@ -146,3 +147,112 @@ func TestHostMetadataHandler(t *testing.T) {
t.Fatalf("unexpected metadata via agent alias: %+v", meta)
}
}
func TestMetadataHandlersUseLiveMonitorStoresForImmediateRoundTrips(t *testing.T) {
fallbackPersistence := config.NewMultiTenantPersistence(t.TempDir())
defaultPersistence, err := fallbackPersistence.GetPersistence("default")
if err != nil {
t.Fatalf("get default persistence: %v", err)
}
t.Run("guest", func(t *testing.T) {
livePath := t.TempDir()
liveStore := config.NewGuestMetadataStore(livePath, nil)
defaultPersistence.SetMetadataStores(liveStore, nil, nil)
handler := NewGuestMetadataHandler(fallbackPersistence)
resolvedOrgID := ""
handler.SetStoreResolver(func(ctx context.Context) *config.GuestMetadataStore {
resolvedOrgID = GetOrgID(ctx)
return liveStore
})
req := httptest.NewRequest(
http.MethodPut,
"/api/guests/metadata/app-container:host:name:app",
strings.NewReader(`{"customUrl":"https://app.internal"}`),
)
resp := httptest.NewRecorder()
handler.HandleUpdateMetadata(resp, req)
if resp.Code != http.StatusOK {
t.Fatalf("update status = %d, body = %s", resp.Code, resp.Body.String())
}
if meta := liveStore.Get("app-container:host:name:app"); meta == nil || meta.CustomURL != "https://app.internal" {
t.Fatalf("live guest store metadata = %#v", meta)
}
if defaultPersistence.GetGuestMetadataStore() != liveStore {
t.Fatal("persistence and monitor guest stores do not share one live instance")
}
// A config import writes the same file out of band, then Reload must
// refresh the monitor-owned in-memory store rather than a parallel
// persistence cache.
importWriter := config.NewGuestMetadataStore(livePath, nil)
if err := importWriter.Set("imported", &config.GuestMetadata{
CustomURL: "https://imported.internal",
}); err != nil {
t.Fatalf("write imported guest metadata: %v", err)
}
tenantContext := context.WithValue(context.Background(), OrgIDContextKey, "tenant-a")
if err := handler.Reload(tenantContext); err != nil {
t.Fatalf("reload live guest metadata: %v", err)
}
if resolvedOrgID != "tenant-a" {
t.Fatalf("reload resolved org %q, want tenant-a", resolvedOrgID)
}
if meta := liveStore.Get("imported"); meta == nil || meta.CustomURL != "https://imported.internal" {
t.Fatalf("reloaded live guest store metadata = %#v", meta)
}
})
t.Run("docker", func(t *testing.T) {
liveStore := config.NewDockerMetadataStore(t.TempDir(), nil)
defaultPersistence.SetMetadataStores(nil, liveStore, nil)
handler := NewDockerMetadataHandler(fallbackPersistence)
handler.SetStoreResolver(func(context.Context) *config.DockerMetadataStore {
return liveStore
})
req := httptest.NewRequest(
http.MethodPut,
"/api/docker/metadata/docker-host:container:runtime-id",
strings.NewReader(`{"customUrl":"https://container.internal"}`),
)
resp := httptest.NewRecorder()
handler.HandleUpdateMetadata(resp, req)
if resp.Code != http.StatusOK {
t.Fatalf("update status = %d, body = %s", resp.Code, resp.Body.String())
}
if meta := liveStore.Get("docker-host:container:runtime-id"); meta == nil || meta.CustomURL != "https://container.internal" {
t.Fatalf("live Docker store metadata = %#v", meta)
}
if defaultPersistence.GetDockerMetadataStore() != liveStore {
t.Fatal("persistence and monitor Docker stores do not share one live instance")
}
})
t.Run("host", func(t *testing.T) {
liveStore := config.NewHostMetadataStore(t.TempDir(), nil)
defaultPersistence.SetMetadataStores(nil, nil, liveStore)
handler := NewHostMetadataHandler(fallbackPersistence)
handler.SetStoreResolver(func(context.Context) *config.HostMetadataStore {
return liveStore
})
req := httptest.NewRequest(
http.MethodPut,
"/api/agents/metadata/agent-live",
strings.NewReader(`{"customUrl":"https://agent.internal"}`),
)
resp := httptest.NewRecorder()
handler.HandleUpdateMetadata(resp, req)
if resp.Code != http.StatusOK {
t.Fatalf("update status = %d, body = %s", resp.Code, resp.Body.String())
}
if meta := liveStore.Get("agent-live"); meta == nil || meta.CustomURL != "https://agent.internal" {
t.Fatalf("live host store metadata = %#v", meta)
}
if defaultPersistence.GetHostMetadataStore() != liveStore {
t.Fatal("persistence and monitor host stores do not share one live instance")
}
})
}
+90 -8
View File
@@ -432,7 +432,26 @@ func (r *Router) setupRoutes() {
r.configHandlers = NewConfigHandlers(r.multiTenant, r.mtMonitor, r.reloadFunc, r.wsHub, guestMetadataHandler, r.reloadSystemSettings)
if r.monitor != nil {
r.configHandlers.SetMonitor(r.monitor)
r.bindDefaultMetadataStores(r.monitor)
}
guestMetadataHandler.SetStoreResolver(func(ctx context.Context) *config.GuestMetadataStore {
if monitor := r.configHandlers.getMonitor(ctx); monitor != nil {
return monitor.GuestMetadataStore()
}
return nil
})
dockerMetadataHandler.SetStoreResolver(func(ctx context.Context) *config.DockerMetadataStore {
if monitor := r.configHandlers.getMonitor(ctx); monitor != nil {
return monitor.DockerMetadataStore()
}
return nil
})
hostMetadataHandler.SetStoreResolver(func(ctx context.Context) *config.HostMetadataStore {
if monitor := r.configHandlers.getMonitor(ctx); monitor != nil {
return monitor.HostMetadataStore()
}
return nil
})
r.configHandlers.SetConfig(r.config)
r.configHandlers.SetMockModeChangeHook(r.syncPlatformSupplementalProviders)
r.trueNASHandlers = &TrueNASHandlers{
@@ -755,14 +774,9 @@ func (r *Router) setupRoutes() {
} else {
log.Warn().Msg("[Router] unified resource provider is nil, cannot inject unified resource provider")
}
// Inject metadata provider for AI URL discovery feature
// This allows AI to set resource URLs when it discovers web services
metadataProvider := NewMetadataProvider(
guestMetadataHandler.Store(),
dockerMetadataHandler.Store(),
hostMetadataHandler.Store(),
)
r.aiSettingsHandler.SetMetadataProvider(metadataProvider)
// Inject tenant-scoped metadata providers for AI URL discovery. Existing
// tenant services are refreshed when Router.SetMonitor replaces a monitor.
r.configureMetadataProviderFactory()
// Wire the per-tenant AI narrator, fleet narrator, and Patrol
// findings provider into reporting. The AI service implements all
@@ -1483,6 +1497,8 @@ func (r *Router) startLifecycleWorker(worker func()) {
// SetMonitor updates the router and associated handlers with a new monitor instance.
func (r *Router) SetMonitor(m *monitoring.Monitor) {
r.monitor = m
r.bindDefaultMetadataStores(m)
r.configureMetadataProviderFactory()
if r.alertHandlers != nil {
r.alertHandlers.SetMonitor(NewAlertMonitorWrapper(m))
}
@@ -1541,6 +1557,72 @@ func (r *Router) SetMonitor(m *monitoring.Monitor) {
}
}
func (r *Router) bindDefaultMetadataStores(monitor *monitoring.Monitor) {
if r == nil || monitor == nil {
return
}
if r.persistence != nil {
r.persistence.SetMetadataStores(
monitor.GuestMetadataStore(),
monitor.DockerMetadataStore(),
monitor.HostMetadataStore(),
)
}
if r.multiTenant != nil {
if persistence, err := r.multiTenant.GetPersistence("default"); err == nil {
persistence.SetMetadataStores(
monitor.GuestMetadataStore(),
monitor.DockerMetadataStore(),
monitor.HostMetadataStore(),
)
}
}
}
func (r *Router) configureMetadataProviderFactory() {
if r == nil || r.aiSettingsHandler == nil {
return
}
r.aiSettingsHandler.SetMetadataProviderFactory(func(orgID string) ai.MetadataProvider {
orgID = strings.TrimSpace(orgID)
if orgID == "" {
orgID = "default"
}
var monitor *monitoring.Monitor
if orgID == "default" {
monitor = r.monitor
} else if r.mtMonitor != nil {
monitor, _ = r.mtMonitor.GetMonitor(orgID)
}
if monitor != nil {
return NewMetadataProvider(
monitor.GuestMetadataStore(),
monitor.DockerMetadataStore(),
monitor.HostMetadataStore(),
)
}
if r.multiTenant != nil {
if persistence, err := r.multiTenant.GetPersistence(orgID); err == nil && persistence != nil {
return NewMetadataProvider(
persistence.GetGuestMetadataStore(),
persistence.GetDockerMetadataStore(),
persistence.GetHostMetadataStore(),
)
}
}
if orgID == "default" && r.persistence != nil {
return NewMetadataProvider(
r.persistence.GetGuestMetadataStore(),
r.persistence.GetDockerMetadataStore(),
r.persistence.GetHostMetadataStore(),
)
}
return nil
})
}
func (r *Router) configureProxmoxGuestDockerDetection(m *monitoring.Monitor) {
if r == nil || m == nil {
return
+2
View File
@@ -115,9 +115,11 @@ func (r *Router) SetMultiTenantMonitor(mtm *monitoring.MultiTenantMonitor) {
if mtm != nil {
if m, err := mtm.GetMonitor("default"); err == nil {
r.monitor = m
r.bindDefaultMetadataStores(m)
}
mtm.SetMonitorInitializer(r.configureMonitorDependencies)
}
r.configureMetadataProviderFactory()
// Wire tenant state provider to resource handlers
if r.resourceHandlers != nil {
@@ -37,13 +37,17 @@ func TestRouterSetMonitor_UpdatesAgentHandlers(t *testing.T) {
// fallback, which is exactly the field SetMonitor must refresh.
kubernetesHandlers := NewKubernetesAgentHandlers(nil, monitor1, nil)
dockerHandlers := NewDockerAgentHandlers(nil, monitor1, nil, cfg)
multiTenantPersistence := config.NewMultiTenantPersistence(tempDir)
// Minimal router, as in TestReloadSystemSettings_AppliesWebhookCIDRsToNewMonitor:
// SetMonitor only touches the handlers that are non-nil.
router := &Router{
config: cfg,
persistence: config.NewConfigPersistence(tempDir),
multiTenant: multiTenantPersistence,
kubernetesAgentHandlers: kubernetesHandlers,
dockerAgentHandlers: dockerHandlers,
aiSettingsHandler: NewAISettingsHandler(multiTenantPersistence, nil, nil),
}
router.SetMonitor(monitor2)
@@ -55,4 +59,26 @@ func TestRouterSetMonitor_UpdatesAgentHandlers(t *testing.T) {
if got := dockerHandlers.getMonitor(ctx); got != monitor2 {
t.Fatal("docker agent handlers still resolve the pre-reload monitor after Router.SetMonitor")
}
if got := router.persistence.GetGuestMetadataStore(); got != monitor2.GuestMetadataStore() {
t.Fatal("config persistence still resolves the pre-reload guest metadata store")
}
if got := router.persistence.GetDockerMetadataStore(); got != monitor2.DockerMetadataStore() {
t.Fatal("config persistence still resolves the pre-reload Docker metadata store")
}
if got := router.persistence.GetHostMetadataStore(); got != monitor2.HostMetadataStore() {
t.Fatal("config persistence still resolves the pre-reload host metadata store")
}
if err := router.aiSettingsHandler.GetAIService(ctx).SetResourceURL(
"vm",
"instance:node:100",
"https://guest.internal",
); err != nil {
t.Fatalf("AI metadata update after monitor reload: %v", err)
}
if meta := monitor2.GuestMetadataStore().Get("instance:node:100"); meta == nil || meta.CustomURL != "https://guest.internal" {
t.Fatalf("replacement monitor guest metadata = %#v", meta)
}
if meta := monitor1.GuestMetadataStore().Get("instance:node:100"); meta != nil {
t.Fatalf("pre-reload monitor received URL update: %#v", meta)
}
}
+7 -1
View File
@@ -8,6 +8,7 @@ import (
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/config"
"github.com/rcourtman/pulse-go-rewrite/internal/monitoring"
)
// EntitlementPayload is the normalized entitlement response for frontend consumption.
@@ -183,8 +184,10 @@ func (h *LicenseHandlers) entitlementUsageSnapshot(ctx context.Context) entitlem
// Count canonical top-level monitored systems from monitor state.
var monitorResolved bool
var resolvedMonitor *monitoring.Monitor
if h.mtMonitor != nil {
if monitor, err := h.mtMonitor.GetMonitor(orgID); err == nil && monitor != nil {
resolvedMonitor = monitor
state := monitor.MonitoredSystemUsage()
if state.Available {
usage.MonitoredSystems = int64(state.Count)
@@ -197,6 +200,7 @@ func (h *LicenseHandlers) entitlementUsageSnapshot(ctx context.Context) entitlem
}
}
if !monitorResolved && orgID == "default" && h.monitor != nil {
resolvedMonitor = h.monitor
state := h.monitor.MonitoredSystemUsage()
if state.Available {
usage.MonitoredSystems = int64(state.Count)
@@ -208,7 +212,9 @@ func (h *LicenseHandlers) entitlementUsageSnapshot(ctx context.Context) entitlem
}
// Guest metadata for guest limit tracking.
if h.mtPersistence != nil {
if resolvedMonitor != nil && resolvedMonitor.GuestMetadataStore() != nil {
usage.Guests = int64(len(resolvedMonitor.GuestMetadataStore().GetAll()))
} else if h.mtPersistence != nil {
if persistence, err := h.mtPersistence.GetPersistence(orgID); err == nil && persistence != nil {
if guestStore := persistence.GetGuestMetadataStore(); guestStore != nil {
usage.Guests = int64(len(guestStore.GetAll()))
+50
View File
@@ -273,3 +273,53 @@ func (s *DockerMetadataStore) ReplaceAll(metadata map[string]*DockerMetadata) er
return s.save()
}
// UpdateAll applies one atomic in-memory container-metadata mutation and
// persists it once while preserving host metadata. Concurrent Set/Delete
// calls cannot be lost between snapshot and replacement.
func (s *DockerMetadataStore) UpdateAll(
update func(map[string]*DockerMetadata) bool,
) error {
if update == nil {
return nil
}
s.mu.Lock()
defer s.mu.Unlock()
working := make(map[string]*DockerMetadata, len(s.metadata))
for id, meta := range s.metadata {
if meta == nil {
continue
}
clone := *meta
if len(meta.Tags) > 0 {
clone.Tags = append([]string(nil), meta.Tags...)
}
if len(meta.Notes) > 0 {
clone.Notes = append([]string(nil), meta.Notes...)
}
working[id] = &clone
}
if !update(working) {
return nil
}
for id, meta := range working {
if meta == nil {
delete(working, id)
continue
}
meta.ID = id
if meta.Tags == nil {
meta.Tags = []string{}
}
}
previous := s.metadata
s.metadata = working
if err := s.save(); err != nil {
s.metadata = previous
return err
}
return nil
}
+29
View File
@@ -145,3 +145,32 @@ func TestDockerMetadataStore_Save_Error(t *testing.T) {
err := store.Set("c1", &DockerMetadata{ID: "c1"})
assert.Error(t, err)
}
func TestDockerMetadataStoreUpdateAllPreservesHostMetadata(t *testing.T) {
store := NewDockerMetadataStore(t.TempDir(), nil)
require.NoError(t, store.Set("old", &DockerMetadata{CustomURL: "https://old.internal"}))
require.NoError(t, store.SetHostMetadata("host", &DockerHostMetadata{
CustomURL: "https://host.internal",
}))
require.NoError(t, store.UpdateAll(func(metadata map[string]*DockerMetadata) bool {
metadata["new"] = metadata["old"]
delete(metadata, "old")
return true
}))
assert.Nil(t, store.Get("old"))
require.NotNil(t, store.Get("new"))
assert.Equal(t, "new", store.Get("new").ID)
assert.Equal(t, "https://old.internal", store.Get("new").CustomURL)
require.NotNil(t, store.GetHostMetadata("host"))
assert.Equal(t, "https://host.internal", store.GetHostMetadata("host").CustomURL)
reloaded := NewDockerMetadataStore(store.dataPath, nil)
assert.Nil(t, reloaded.Get("old"))
require.NotNil(t, reloaded.Get("new"))
assert.Equal(t, "new", reloaded.Get("new").ID)
assert.Equal(t, "https://old.internal", reloaded.Get("new").CustomURL)
require.NotNil(t, reloaded.GetHostMetadata("host"))
assert.Equal(t, "https://host.internal", reloaded.GetHostMetadata("host").CustomURL)
}
+46
View File
@@ -147,6 +147,9 @@ func (s *GuestMetadataStore) GetWithLegacyMigration(guestID, instance, node stri
s.mu.Lock()
defer s.mu.Unlock()
if current := s.metadata[guestID]; current != nil {
return cloneGuestMetadata(current)
}
legacyMeta := s.metadata[legacyID]
if legacyMeta == nil {
return cloneGuestMetadata(s.metadata[guestID])
@@ -164,6 +167,8 @@ func (s *GuestMetadataStore) GetWithLegacyMigration(guestID, instance, node stri
// Persist while holding the store lock. save() assumes locked access.
if err := s.save(); err != nil {
log.Error().Err(err).Msg("Failed to save guest metadata after migration")
s.metadata[legacyID] = legacyMeta
delete(s.metadata, guestID)
}
return cloneGuestMetadata(migrated)
@@ -257,3 +262,44 @@ func (s *GuestMetadataStore) ReplaceAll(metadata map[string]*GuestMetadata) erro
return s.save()
}
// UpdateAll applies one atomic in-memory mutation and persists it once.
// The callback receives a deep-cloned working set and returns whether it
// changed. Concurrent Set/Delete calls cannot be lost between snapshot and
// replacement.
func (s *GuestMetadataStore) UpdateAll(
update func(map[string]*GuestMetadata) bool,
) error {
if update == nil {
return nil
}
s.mu.Lock()
defer s.mu.Unlock()
working := make(map[string]*GuestMetadata, len(s.metadata))
for id, meta := range s.metadata {
working[id] = cloneGuestMetadata(meta)
}
if !update(working) {
return nil
}
for id, meta := range working {
if meta == nil {
delete(working, id)
continue
}
meta.ID = id
if meta.Tags == nil {
meta.Tags = []string{}
}
}
previous := s.metadata
s.metadata = working
if err := s.save(); err != nil {
s.metadata = previous
return err
}
return nil
}
+58
View File
@@ -652,3 +652,61 @@ func TestGuestMetadataStore_GetWithLegacyMigration_ConcurrentMigration(t *testin
t.Error("New ID should exist after migration")
}
}
func TestGuestMetadataStoreUpdateAllAppliesOneWorkingSet(t *testing.T) {
store := NewGuestMetadataStore(t.TempDir(), nil)
if err := store.Set("old", &GuestMetadata{CustomURL: "https://old.internal"}); err != nil {
t.Fatalf("seed metadata: %v", err)
}
if err := store.UpdateAll(func(metadata map[string]*GuestMetadata) bool {
metadata["new"] = metadata["old"]
delete(metadata, "old")
return true
}); err != nil {
t.Fatalf("UpdateAll: %v", err)
}
if store.Get("old") != nil {
t.Fatal("old metadata key still exists")
}
if meta := store.Get("new"); meta == nil || meta.ID != "new" || meta.CustomURL != "https://old.internal" {
t.Fatalf("new metadata = %#v", meta)
}
reloaded := NewGuestMetadataStore(store.dataPath, nil)
if reloaded.Get("old") != nil {
t.Fatal("persisted old metadata key still exists")
}
if meta := reloaded.Get("new"); meta == nil || meta.ID != "new" || meta.CustomURL != "https://old.internal" {
t.Fatalf("persisted new metadata = %#v", meta)
}
}
func TestGuestMetadataStoreUpdateAllRollsBackFailedPersistence(t *testing.T) {
store := NewGuestMetadataStore(t.TempDir(), nil)
if err := store.Set("old", &GuestMetadata{CustomURL: "https://old.internal"}); err != nil {
t.Fatalf("seed metadata: %v", err)
}
badPath := filepath.Join(t.TempDir(), "not-a-directory")
if err := os.WriteFile(badPath, []byte("occupied"), 0o600); err != nil {
t.Fatalf("create invalid data path: %v", err)
}
store.dataPath = badPath
err := store.UpdateAll(func(metadata map[string]*GuestMetadata) bool {
metadata["new"] = metadata["old"]
delete(metadata, "old")
return true
})
if err == nil {
t.Fatal("expected persistence error")
}
if meta := store.Get("old"); meta == nil || meta.CustomURL != "https://old.internal" {
t.Fatalf("rolled-back old metadata = %#v", meta)
}
if meta := store.Get("new"); meta != nil {
t.Fatalf("failed update remained in memory: %#v", meta)
}
}
@@ -1,5 +1,30 @@
package config
// SetMetadataStores makes persistence consumers share the active monitor's
// in-memory metadata stores. The stores already persist to this scope's data
// directory; sharing the instances prevents API, export/import, and monitor
// projection from observing different caches of the same files.
func (c *ConfigPersistence) SetMetadataStores(
guest *GuestMetadataStore,
docker *DockerMetadataStore,
host *HostMetadataStore,
) {
if c == nil {
return
}
c.metadataMu.Lock()
defer c.metadataMu.Unlock()
if guest != nil {
c.guestMetadataStore = guest
}
if docker != nil {
c.dockerMetadataStore = docker
}
if host != nil {
c.hostMetadataStore = host
}
}
// GetGuestMetadataStore returns the guest metadata store, creating it if necessary
func (c *ConfigPersistence) GetGuestMetadataStore() *GuestMetadataStore {
c.metadataMu.Lock()
@@ -31,3 +31,16 @@ func TestConfigPersistence_LoadDockerMetadata(t *testing.T) {
// Ensure we can use the store
assert.Empty(t, store.GetAll())
}
func TestConfigPersistenceSharesActiveMetadataStores(t *testing.T) {
cp := config.NewConfigPersistence(t.TempDir())
guest := config.NewGuestMetadataStore(t.TempDir(), nil)
docker := config.NewDockerMetadataStore(t.TempDir(), nil)
host := config.NewHostMetadataStore(t.TempDir(), nil)
cp.SetMetadataStores(guest, docker, host)
assert.Same(t, guest, cp.GetGuestMetadataStore())
assert.Same(t, docker, cp.GetDockerMetadataStore())
assert.Same(t, host, cp.GetHostMetadataStore())
}
+316 -79
View File
@@ -34,44 +34,48 @@ func (m *Monitor) CopyDockerContainerMetadata(hostID, oldContainerID, newContain
oldKey := dockerContainerRuntimeMetadataKey(hostID, oldContainerID)
newKey := dockerContainerRuntimeMetadataKey(hostID, newContainerID)
oldMeta := m.dockerMetadataStore.Get(oldKey)
if oldMeta == nil {
return nil
}
if oldMeta.CustomURL == "" && oldMeta.Description == "" && len(oldMeta.Tags) == 0 && len(oldMeta.Notes) == 0 {
return nil
}
return m.dockerMetadataStore.UpdateAll(func(metadata map[string]*config.DockerMetadata) bool {
oldMeta := metadata[oldKey]
if oldMeta == nil {
return false
}
if oldMeta.CustomURL == "" && oldMeta.Description == "" && len(oldMeta.Tags) == 0 && len(oldMeta.Notes) == 0 {
return false
}
newMeta := m.dockerMetadataStore.Get(newKey)
var merged config.DockerMetadata
if newMeta != nil {
merged = *newMeta
}
newMeta := metadata[newKey]
var merged config.DockerMetadata
if newMeta != nil {
merged = *newMeta
}
// Merge missing fields from old -> new, so we don't clobber any metadata already present under the new ID.
if merged.CustomURL == "" {
merged.CustomURL = oldMeta.CustomURL
}
if merged.Description == "" {
merged.Description = oldMeta.Description
}
if len(merged.Tags) == 0 && len(oldMeta.Tags) > 0 {
merged.Tags = append([]string(nil), oldMeta.Tags...)
}
if len(merged.Notes) == 0 && len(oldMeta.Notes) > 0 {
merged.Notes = append([]string(nil), oldMeta.Notes...)
}
// Merge missing fields from old -> new, so a concurrent or already
// persisted value under the recreated runtime ID always wins.
if merged.CustomURL == "" {
merged.CustomURL = oldMeta.CustomURL
}
if merged.Description == "" {
merged.Description = oldMeta.Description
}
if len(merged.Tags) == 0 && len(oldMeta.Tags) > 0 {
merged.Tags = append([]string(nil), oldMeta.Tags...)
}
if len(merged.Notes) == 0 && len(oldMeta.Notes) > 0 {
merged.Notes = append([]string(nil), oldMeta.Notes...)
}
// Avoid an unnecessary disk write if nothing changed.
if newMeta != nil &&
merged.CustomURL == newMeta.CustomURL &&
merged.Description == newMeta.Description &&
slices.Equal(merged.Tags, newMeta.Tags) &&
slices.Equal(merged.Notes, newMeta.Notes) {
return nil
}
if newMeta != nil &&
merged.CustomURL == newMeta.CustomURL &&
merged.Description == newMeta.Description &&
slices.Equal(merged.Tags, newMeta.Tags) &&
slices.Equal(merged.Notes, newMeta.Notes) {
return false
}
return m.dockerMetadataStore.Set(newKey, &merged)
merged.ID = newKey
metadata[newKey] = &merged
return true
})
}
func dockerContainerRuntimeMetadataKey(hostID, containerID string) string {
@@ -148,24 +152,28 @@ func copyDockerMetadataAliasIfTargetMissing(store *config.DockerMetadataStore, s
if store == nil || strings.TrimSpace(sourceKey) == "" || strings.TrimSpace(targetKey) == "" {
return nil
}
if store.Get(targetKey) != nil {
return nil
}
source := store.Get(sourceKey)
if source == nil {
return nil
}
clone := &config.DockerMetadata{
CustomURL: source.CustomURL,
Description: source.Description,
}
if len(source.Tags) > 0 {
clone.Tags = append([]string(nil), source.Tags...)
}
if len(source.Notes) > 0 {
clone.Notes = append([]string(nil), source.Notes...)
}
return store.Set(targetKey, clone)
return store.UpdateAll(func(metadata map[string]*config.DockerMetadata) bool {
if metadata[targetKey] != nil {
return false
}
source := metadata[sourceKey]
if source == nil {
return false
}
clone := &config.DockerMetadata{
ID: targetKey,
CustomURL: source.CustomURL,
Description: source.Description,
}
if len(source.Tags) > 0 {
clone.Tags = append([]string(nil), source.Tags...)
}
if len(source.Notes) > 0 {
clone.Notes = append([]string(nil), source.Notes...)
}
metadata[targetKey] = clone
return true
})
}
func copyGuestMetadataAliasIfTargetMissing(
@@ -177,26 +185,248 @@ func copyGuestMetadataAliasIfTargetMissing(
if store == nil || strings.TrimSpace(sourceKey) == "" || strings.TrimSpace(targetKey) == "" {
return nil
}
if store.Get(targetKey) != nil {
return store.UpdateAll(func(metadata map[string]*config.GuestMetadata) bool {
if metadata[targetKey] != nil {
return false
}
source := metadata[sourceKey]
if source == nil {
return false
}
clone := &config.GuestMetadata{
ID: targetKey,
CustomURL: source.CustomURL,
Description: source.Description,
LastKnownName: normalizeDockerContainerMetadataIdentity(containerName),
LastKnownType: "app-container",
}
if len(source.Tags) > 0 {
clone.Tags = append([]string(nil), source.Tags...)
}
if len(source.Notes) > 0 {
clone.Notes = append([]string(nil), source.Notes...)
}
metadata[targetKey] = clone
return true
})
}
type dockerStableMetadataMove struct {
sourceKey string
targetKey string
targetName string
}
func dockerContainerStableMetadataMoves(
hostID string,
previousContainers []models.DockerContainer,
currentContainers []models.DockerContainer,
keyForName func(string, string) string,
) []dockerStableMetadataMove {
previousByID := make(map[string]models.DockerContainer, len(previousContainers))
ambiguousIDs := make(map[string]struct{})
previousNameCounts := make(map[string]int, len(previousContainers))
for _, container := range previousContainers {
containerID := strings.TrimSpace(container.ID)
if name := normalizeDockerContainerMetadataIdentity(container.Name); name != "" {
previousNameCounts[keyForName(hostID, name)]++
}
if containerID == "" {
continue
}
if _, exists := previousByID[containerID]; exists {
ambiguousIDs[containerID] = struct{}{}
delete(previousByID, containerID)
continue
}
if _, ambiguous := ambiguousIDs[containerID]; ambiguous {
continue
}
previousByID[containerID] = container
}
currentNameCounts := make(map[string]int, len(currentContainers))
for _, container := range currentContainers {
if name := normalizeDockerContainerMetadataIdentity(container.Name); name != "" {
currentNameCounts[keyForName(hostID, name)]++
}
}
sourceCounts := make(map[string]int)
targetCounts := make(map[string]int)
candidates := make([]dockerStableMetadataMove, 0)
for _, current := range currentContainers {
containerID := strings.TrimSpace(current.ID)
if containerID == "" {
continue
}
previous, ok := previousByID[containerID]
if !ok {
continue
}
previousName := normalizeDockerContainerMetadataIdentity(previous.Name)
currentName := normalizeDockerContainerMetadataIdentity(current.Name)
if previousName == "" || currentName == "" || previousName == currentName {
continue
}
sourceKey := keyForName(hostID, previousName)
targetKey := keyForName(hostID, currentName)
if sourceKey == "" || targetKey == "" || sourceKey == targetKey {
continue
}
candidates = append(candidates, dockerStableMetadataMove{
sourceKey: sourceKey,
targetKey: targetKey,
targetName: currentName,
})
sourceCounts[sourceKey]++
targetCounts[targetKey]++
}
moves := make([]dockerStableMetadataMove, 0, len(candidates))
for _, move := range candidates {
if sourceCounts[move.sourceKey] == 1 &&
targetCounts[move.targetKey] == 1 &&
previousNameCounts[move.sourceKey] == 1 &&
currentNameCounts[move.targetKey] == 1 {
moves = append(moves, move)
}
}
return moves
}
func applyStableMetadataMoves[T any](
moves []dockerStableMetadataMove,
updateAll func(func(map[string]*T) bool) error,
cloneForTarget func(*T, dockerStableMetadataMove) *T,
) error {
if len(moves) == 0 {
return nil
}
source := store.Get(sourceKey)
if source == nil {
return nil
return updateAll(func(metadata map[string]*T) bool {
sourceSnapshots := make(map[string]*T, len(moves))
sourceKeys := make(map[string]struct{}, len(moves))
targetKeys := make(map[string]struct{}, len(moves))
for _, move := range moves {
sourceKeys[move.sourceKey] = struct{}{}
targetKeys[move.targetKey] = struct{}{}
if _, loaded := sourceSnapshots[move.sourceKey]; !loaded {
sourceSnapshots[move.sourceKey] = metadata[move.sourceKey]
}
}
changed := false
for _, move := range moves {
source := sourceSnapshots[move.sourceKey]
if source == nil {
continue
}
target := metadata[move.targetKey]
_, targetIsMovingSource := sourceKeys[move.targetKey]
if target != nil && !targetIsMovingSource {
// The destination already has intentional name-scoped metadata.
// Preserve it rather than attaching a renamed container's metadata
// to a name that may represent a different logical resource.
continue
}
metadata[move.targetKey] = cloneForTarget(source, move)
changed = true
}
for _, move := range moves {
if sourceSnapshots[move.sourceKey] == nil {
continue
}
if _, sourceIsAnotherTarget := targetKeys[move.sourceKey]; sourceIsAnotherTarget {
continue
}
delete(metadata, move.sourceKey)
changed = true
}
return changed
})
}
func (m *Monitor) migrateDockerContainerMetadataForRenamedContainers(
hostID string,
previousContainers []models.DockerContainer,
currentContainers []models.DockerContainer,
) {
if m == nil {
return
}
clone := &config.GuestMetadata{
CustomURL: source.CustomURL,
Description: source.Description,
LastKnownName: normalizeDockerContainerMetadataIdentity(containerName),
LastKnownType: "app-container",
hostID = strings.TrimSpace(hostID)
if hostID == "" || len(previousContainers) == 0 || len(currentContainers) == 0 {
return
}
if len(source.Tags) > 0 {
clone.Tags = append([]string(nil), source.Tags...)
if m.guestMetadataStore != nil {
moves := dockerContainerStableMetadataMoves(
hostID,
previousContainers,
currentContainers,
dockerAppContainerMetadataKey,
)
err := applyStableMetadataMoves(
moves,
m.guestMetadataStore.UpdateAll,
func(source *config.GuestMetadata, move dockerStableMetadataMove) *config.GuestMetadata {
clone := &config.GuestMetadata{
CustomURL: source.CustomURL,
Description: source.Description,
LastKnownName: move.targetName,
LastKnownType: "app-container",
}
if len(source.Tags) > 0 {
clone.Tags = append([]string(nil), source.Tags...)
}
if len(source.Notes) > 0 {
clone.Notes = append([]string(nil), source.Notes...)
}
return clone
},
)
if err != nil {
log.Warn().
Err(err).
Str("dockerHostID", hostID).
Int("renameCount", len(moves)).
Msg("Failed to move guest metadata after Docker container rename")
}
}
if len(source.Notes) > 0 {
clone.Notes = append([]string(nil), source.Notes...)
if m.dockerMetadataStore != nil {
moves := dockerContainerStableMetadataMoves(
hostID,
previousContainers,
currentContainers,
dockerContainerNameMetadataKey,
)
err := applyStableMetadataMoves(
moves,
m.dockerMetadataStore.UpdateAll,
func(source *config.DockerMetadata, _ dockerStableMetadataMove) *config.DockerMetadata {
clone := &config.DockerMetadata{
CustomURL: source.CustomURL,
Description: source.Description,
}
if len(source.Tags) > 0 {
clone.Tags = append([]string(nil), source.Tags...)
}
if len(source.Notes) > 0 {
clone.Notes = append([]string(nil), source.Notes...)
}
return clone
},
)
if err != nil {
log.Warn().
Err(err).
Str("dockerHostID", hostID).
Int("renameCount", len(moves)).
Msg("Failed to move Docker metadata after container rename")
}
}
return store.Set(targetKey, clone)
}
func (m *Monitor) migrateCurrentDockerContainerMetadataToStableIdentities(
@@ -260,19 +490,26 @@ func (m *Monitor) migrateCurrentDockerContainerMetadataToStableIdentities(
source = m.dockerMetadataStore.Get(dockerContainerNameMetadataKey(hostID, containerName))
}
if source != nil {
clone := &config.GuestMetadata{
CustomURL: source.CustomURL,
Description: source.Description,
LastKnownName: containerName,
LastKnownType: "app-container",
}
if len(source.Tags) > 0 {
clone.Tags = append([]string(nil), source.Tags...)
}
if len(source.Notes) > 0 {
clone.Notes = append([]string(nil), source.Notes...)
}
if err := m.guestMetadataStore.Set(stableKey, clone); err != nil {
if err := m.guestMetadataStore.UpdateAll(func(metadata map[string]*config.GuestMetadata) bool {
if metadata[stableKey] != nil {
return false
}
clone := &config.GuestMetadata{
ID: stableKey,
CustomURL: source.CustomURL,
Description: source.Description,
LastKnownName: containerName,
LastKnownType: "app-container",
}
if len(source.Tags) > 0 {
clone.Tags = append([]string(nil), source.Tags...)
}
if len(source.Notes) > 0 {
clone.Notes = append([]string(nil), source.Notes...)
}
metadata[stableKey] = clone
return true
}); err != nil {
log.Warn().
Err(err).
Str("dockerHostID", hostID).
@@ -0,0 +1,128 @@
package monitoring
import (
"fmt"
"strings"
"github.com/rcourtman/pulse-go-rewrite/internal/config"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
"github.com/rs/zerolog/log"
)
func kubernetesWorkloadMetadataKey(resource unifiedresources.Resource) string {
if resource.Kubernetes == nil {
return ""
}
clusterID := strings.TrimSpace(resource.Kubernetes.ClusterID)
namespace := strings.TrimSpace(resource.Kubernetes.Namespace)
name := strings.TrimSpace(resource.Name)
if clusterID == "" || namespace == "" || name == "" {
return ""
}
var kind string
switch unifiedresources.CanonicalResourceType(resource.Type) {
case unifiedresources.ResourceTypePod:
kind = "pod"
case unifiedresources.ResourceTypeK8sDeployment:
kind = "deployment"
case unifiedresources.ResourceTypeK8sService:
kind = "service"
default:
return ""
}
return fmt.Sprintf("k8s-workload:%s:%s:%s:%s", clusterID, kind, namespace, name)
}
func kubernetesWorkloadLegacyMetadataKeys(resource unifiedresources.Resource) []string {
candidates := []string{strings.TrimSpace(resource.ID)}
if resource.Kubernetes != nil &&
unifiedresources.CanonicalResourceType(resource.Type) == unifiedresources.ResourceTypePod {
clusterID := strings.TrimSpace(resource.Kubernetes.ClusterID)
podUID := strings.TrimSpace(resource.Kubernetes.PodUID)
if clusterID != "" && podUID != "" {
candidates = append(candidates, fmt.Sprintf("k8s:%s:pod:%s", clusterID, podUID))
}
}
out := make([]string, 0, len(candidates))
seen := make(map[string]struct{}, len(candidates))
for _, candidate := range candidates {
if candidate == "" {
continue
}
if _, exists := seen[candidate]; exists {
continue
}
seen[candidate] = struct{}{}
out = append(out, candidate)
}
return out
}
func cloneKubernetesGuestMetadata(
source *config.GuestMetadata,
resource unifiedresources.Resource,
) *config.GuestMetadata {
clone := &config.GuestMetadata{
CustomURL: source.CustomURL,
Description: source.Description,
LastKnownName: strings.TrimSpace(resource.Name),
LastKnownType: string(unifiedresources.CanonicalResourceType(resource.Type)),
}
if len(source.Tags) > 0 {
clone.Tags = append([]string(nil), source.Tags...)
}
if len(source.Notes) > 0 {
clone.Notes = append([]string(nil), source.Notes...)
}
return clone
}
func (m *Monitor) kubernetesWorkloadCustomURL(resource unifiedresources.Resource) (string, bool) {
if m == nil || m.guestMetadataStore == nil {
return "", false
}
stableKey := kubernetesWorkloadMetadataKey(resource)
if stableKey == "" {
return "", false
}
var (
customURL string
found bool
migratedLegacy string
)
err := m.guestMetadataStore.UpdateAll(func(metadata map[string]*config.GuestMetadata) bool {
if meta := metadata[stableKey]; meta != nil {
customURL = strings.TrimSpace(meta.CustomURL)
found = true
return false
}
for _, legacyKey := range kubernetesWorkloadLegacyMetadataKeys(resource) {
meta := metadata[legacyKey]
if meta == nil {
continue
}
customURL = strings.TrimSpace(meta.CustomURL)
found = true
migratedLegacy = legacyKey
clone := cloneKubernetesGuestMetadata(meta, resource)
clone.ID = stableKey
metadata[stableKey] = clone
return true
}
return false
})
if err != nil {
log.Warn().
Err(err).
Str("resourceID", resource.ID).
Str("legacyMetadataKey", migratedLegacy).
Str("stableMetadataKey", stableKey).
Msg("Failed to migrate Kubernetes workload metadata to stable logical identity")
}
return customURL, found
}
+30
View File
@@ -0,0 +1,30 @@
package monitoring
import "github.com/rcourtman/pulse-go-rewrite/internal/config"
// GuestMetadataStore returns the live guest metadata store used by resource
// projection, migration, and monitoring decisions.
func (m *Monitor) GuestMetadataStore() *config.GuestMetadataStore {
if m == nil {
return nil
}
return m.guestMetadataStore
}
// DockerMetadataStore returns the live Docker metadata store used by resource
// projection, migration, and monitoring decisions.
func (m *Monitor) DockerMetadataStore() *config.DockerMetadataStore {
if m == nil {
return nil
}
return m.dockerMetadataStore
}
// HostMetadataStore returns the live host metadata store used by resource
// projection and monitoring decisions.
func (m *Monitor) HostMetadataStore() *config.HostMetadataStore {
if m == nil {
return nil
}
return m.hostMetadataStore
}
+25 -12
View File
@@ -4008,7 +4008,7 @@ func (m *Monitor) buildBroadcastFrontendStateFromSnapshot(snapshot models.StateS
unifiedView := m.currentUnifiedStateView()
metricsTargetResolver := broadcastMetricsTargetResolver(unifiedView.readState)
broadcastResources := unifiedresources.CoalescePresentationHostResources(unifiedView.resources)
broadcastResources = m.applyDockerMetadataToUnifiedResources(broadcastResources)
broadcastResources = m.applyPersistedMetadataToUnifiedResources(broadcastResources)
frontendState.Resources = convertResourcesForBroadcast(broadcastResources, metricsTargetResolver)
frontendState.ConnectedInfrastructure = buildConnectedInfrastructure(broadcastResources, snapshot)
if !unifiedView.freshness.IsZero() {
@@ -4618,7 +4618,7 @@ func (m *Monitor) UnifiedResourceSnapshot() ([]unifiedresources.Resource, time.T
// REST /api/resources seeds its registry from this snapshot. Apply the
// same user-metadata hydration (container customUrl) as the websocket
// broadcast path, so the two payload shapes cannot drift.
return m.applyDockerMetadataToUnifiedResources(view.resources), view.freshness
return m.applyPersistedMetadataToUnifiedResources(view.resources), view.freshness
}
// GetUnifiedReadState returns a typed unified read-state provider when the
@@ -5347,12 +5347,12 @@ func (m *Monitor) getResourcesForBroadcast() []models.ResourceFrontend {
store := m.resourceStore
m.mu.RUnlock()
return convertResourcesForBroadcast(
m.applyDockerMetadataToUnifiedResources(m.getUnifiedResourcesForBroadcast()),
m.applyPersistedMetadataToUnifiedResources(m.getUnifiedResourcesForBroadcast()),
broadcastMetricsTargetResolver(store),
)
}
func (m *Monitor) applyDockerMetadataToUnifiedResources(resources []unifiedresources.Resource) []unifiedresources.Resource {
func (m *Monitor) applyPersistedMetadataToUnifiedResources(resources []unifiedresources.Resource) []unifiedresources.Resource {
if len(resources) == 0 || m == nil {
return resources
}
@@ -5361,16 +5361,29 @@ func (m *Monitor) applyDockerMetadataToUnifiedResources(resources []unifiedresou
copy(out, resources)
for i := range out {
resource := &out[i]
if resource.Type != unifiedresources.ResourceTypeAppContainer || strings.TrimSpace(resource.CustomURL) != "" || resource.Docker == nil {
if strings.TrimSpace(resource.CustomURL) != "" {
continue
}
hostID := strings.TrimSpace(resource.Docker.HostSourceID)
containerID := strings.TrimSpace(resource.Docker.ContainerID)
if hostID == "" {
continue
}
if customURL, ok := m.dockerAppContainerCustomURL(*resource, hostID, containerID); ok {
resource.CustomURL = customURL
switch unifiedresources.CanonicalResourceType(resource.Type) {
case unifiedresources.ResourceTypeAppContainer:
if resource.Docker == nil {
continue
}
hostID := strings.TrimSpace(resource.Docker.HostSourceID)
containerID := strings.TrimSpace(resource.Docker.ContainerID)
if hostID == "" {
continue
}
if customURL, ok := m.dockerAppContainerCustomURL(*resource, hostID, containerID); ok {
resource.CustomURL = customURL
}
case unifiedresources.ResourceTypePod,
unifiedresources.ResourceTypeK8sDeployment,
unifiedresources.ResourceTypeK8sService:
if customURL, ok := m.kubernetesWorkloadCustomURL(*resource); ok {
resource.CustomURL = customURL
}
}
}
return out
+1
View File
@@ -1969,6 +1969,7 @@ func (m *Monitor) ApplyDockerReport(report agentsdocker.Report, tokenRecord *con
}
if hasPrevious {
m.migrateDockerContainerMetadataForRenamedContainers(identifier, previous.Containers(), host.Containers)
m.migrateDockerContainerMetadataForRecreatedContainers(identifier, previous.Containers(), host.Containers)
}
m.migrateCurrentDockerContainerMetadataToStableIdentities(identifier, host.Containers)
+365 -8
View File
@@ -511,12 +511,196 @@ func TestApplyDockerReportMigratesGuestMetadataToStableContainerName(t *testing.
},
}
got := monitor.applyDockerMetadataToUnifiedResources(resources)
got := monitor.applyPersistedMetadataToUnifiedResources(resources)
if got[0].CustomURL != "https://app.internal" {
t.Fatalf("CustomURL after recreate gap = %q, want https://app.internal", got[0].CustomURL)
}
}
func TestApplyDockerReportMovesStableMetadataOnRenameWithoutLeakingToReusedName(t *testing.T) {
monitor := newTestMonitor(t)
baseTimestamp := time.Now().UTC()
report := agentsdocker.Report{
Agent: agentsdocker.AgentInfo{
ID: "agent-rename",
Version: "1.0.0",
IntervalSeconds: 30,
},
Host: agentsdocker.HostInfo{
Hostname: "docker-host-rename",
MachineID: "machine-rename",
},
Containers: []agentsdocker.Container{
{ID: "container-stable", Name: "app"},
},
Timestamp: baseTimestamp,
}
host, err := monitor.ApplyDockerReport(report, nil)
if err != nil {
t.Fatalf("first ApplyDockerReport failed: %v", err)
}
oldKey := dockerAppContainerMetadataKey(host.ID, "app")
if err := monitor.guestMetadataStore.Set(oldKey, &config.GuestMetadata{
CustomURL: "https://app.internal",
LastKnownName: "app",
LastKnownType: "app-container",
}); err != nil {
t.Fatalf("seed stable guest metadata: %v", err)
}
oldDockerKey := dockerContainerNameMetadataKey(host.ID, "app")
if err := monitor.dockerMetadataStore.Set(oldDockerKey, &config.DockerMetadata{
CustomURL: "https://app-drawer.internal",
}); err != nil {
t.Fatalf("seed stable Docker metadata: %v", err)
}
report.Timestamp = baseTimestamp.Add(30 * time.Second)
report.Containers = []agentsdocker.Container{
{ID: "container-stable", Name: "renamed-app"},
}
if _, err := monitor.ApplyDockerReport(report, nil); err != nil {
t.Fatalf("renamed-container ApplyDockerReport failed: %v", err)
}
newKey := dockerAppContainerMetadataKey(host.ID, "renamed-app")
if meta := monitor.guestMetadataStore.Get(newKey); meta == nil || meta.CustomURL != "https://app.internal" {
t.Fatalf("renamed stable metadata = %#v, want preserved URL", meta)
}
if meta := monitor.guestMetadataStore.Get(oldKey); meta != nil {
t.Fatalf("old stable metadata key still exists after rename: %#v", meta)
}
newDockerKey := dockerContainerNameMetadataKey(host.ID, "renamed-app")
if meta := monitor.dockerMetadataStore.Get(newDockerKey); meta == nil || meta.CustomURL != "https://app-drawer.internal" {
t.Fatalf("renamed stable Docker metadata = %#v, want preserved URL", meta)
}
if meta := monitor.dockerMetadataStore.Get(oldDockerKey); meta != nil {
t.Fatalf("old stable Docker metadata key still exists after rename: %#v", meta)
}
report.Timestamp = baseTimestamp.Add(60 * time.Second)
report.Containers = []agentsdocker.Container{
{ID: "container-stable", Name: "renamed-app"},
{ID: "container-unrelated", Name: "app"},
}
if _, err := monitor.ApplyDockerReport(report, nil); err != nil {
t.Fatalf("reused-name ApplyDockerReport failed: %v", err)
}
resources := []unifiedresources.Resource{
{
ID: dockerAppContainerLegacyResourceID(host.ID, "container-stable"),
Type: unifiedresources.ResourceTypeAppContainer,
Name: "renamed-app",
Docker: &unifiedresources.DockerData{
HostSourceID: host.ID,
ContainerID: "container-stable",
},
},
{
ID: dockerAppContainerLegacyResourceID(host.ID, "container-unrelated"),
Type: unifiedresources.ResourceTypeAppContainer,
Name: "app",
Docker: &unifiedresources.DockerData{
HostSourceID: host.ID,
ContainerID: "container-unrelated",
},
},
}
got := monitor.applyPersistedMetadataToUnifiedResources(resources)
if got[0].CustomURL != "https://app.internal" {
t.Fatalf("renamed container CustomURL = %q, want preserved URL", got[0].CustomURL)
}
if got[1].CustomURL != "" {
t.Fatalf("unrelated container reusing old name inherited CustomURL %q", got[1].CustomURL)
}
}
func TestDockerContainerStableMetadataMovesFailClosedOnAmbiguousNames(t *testing.T) {
moves := dockerContainerStableMetadataMoves(
"docker-host",
[]models.DockerContainer{
{ID: "container-a", Name: "/app"},
{ID: "container-b", Name: "app"},
},
[]models.DockerContainer{
{ID: "container-a", Name: "renamed-app"},
{ID: "container-b", Name: "app"},
},
dockerAppContainerMetadataKey,
)
if len(moves) != 0 {
t.Fatalf("ambiguous normalized source names produced moves: %#v", moves)
}
}
func TestMigrateDockerContainerMetadataForRenamedContainersPreservesNameSwapOwnership(t *testing.T) {
monitor := newTestMonitor(t)
hostID := "docker-host-swap"
if err := monitor.guestMetadataStore.Set(
dockerAppContainerMetadataKey(hostID, "blue"),
&config.GuestMetadata{CustomURL: "https://blue.internal"},
); err != nil {
t.Fatalf("seed blue metadata: %v", err)
}
if err := monitor.guestMetadataStore.Set(
dockerAppContainerMetadataKey(hostID, "green"),
&config.GuestMetadata{CustomURL: "https://green.internal"},
); err != nil {
t.Fatalf("seed green metadata: %v", err)
}
monitor.migrateDockerContainerMetadataForRenamedContainers(
hostID,
[]models.DockerContainer{
{ID: "container-blue", Name: "blue"},
{ID: "container-green", Name: "green"},
},
[]models.DockerContainer{
{ID: "container-blue", Name: "green"},
{ID: "container-green", Name: "blue"},
},
)
if meta := monitor.guestMetadataStore.Get(dockerAppContainerMetadataKey(hostID, "green")); meta == nil || meta.CustomURL != "https://blue.internal" {
t.Fatalf("container-blue metadata after swap = %#v", meta)
}
if meta := monitor.guestMetadataStore.Get(dockerAppContainerMetadataKey(hostID, "blue")); meta == nil || meta.CustomURL != "https://green.internal" {
t.Fatalf("container-green metadata after swap = %#v", meta)
}
}
func TestMigrateDockerContainerMetadataForRenamedContainersPreservesExistingDestination(t *testing.T) {
monitor := newTestMonitor(t)
hostID := "docker-host-destination"
oldKey := dockerAppContainerMetadataKey(hostID, "old-name")
newKey := dockerAppContainerMetadataKey(hostID, "reserved-name")
if err := monitor.guestMetadataStore.Set(oldKey, &config.GuestMetadata{
CustomURL: "https://old.internal",
}); err != nil {
t.Fatalf("seed old metadata: %v", err)
}
if err := monitor.guestMetadataStore.Set(newKey, &config.GuestMetadata{
CustomURL: "https://reserved.internal",
}); err != nil {
t.Fatalf("seed destination metadata: %v", err)
}
monitor.migrateDockerContainerMetadataForRenamedContainers(
hostID,
[]models.DockerContainer{{ID: "container", Name: "old-name"}},
[]models.DockerContainer{{ID: "container", Name: "reserved-name"}},
)
if meta := monitor.guestMetadataStore.Get(newKey); meta == nil || meta.CustomURL != "https://reserved.internal" {
t.Fatalf("destination metadata = %#v", meta)
}
if meta := monitor.guestMetadataStore.Get(oldKey); meta != nil {
t.Fatalf("obsolete source could leak onto later name reuse: %#v", meta)
}
}
func TestApplyDockerReportFoldsRuntimeKeyDockerMetadataIntoStableGuestKey(t *testing.T) {
monitor := newTestMonitor(t)
@@ -575,7 +759,7 @@ func TestApplyDockerReportFoldsRuntimeKeyDockerMetadataIntoStableGuestKey(t *tes
},
},
}
got := monitor.applyDockerMetadataToUnifiedResources(resources)
got := monitor.applyPersistedMetadataToUnifiedResources(resources)
if got[0].CustomURL != "https://wud.internal" {
t.Fatalf("projected CustomURL = %q, want https://wud.internal", got[0].CustomURL)
}
@@ -643,7 +827,7 @@ func TestApplyDockerReportKeepsClearedStableGuestKeyOverRuntimeDockerMetadata(t
},
},
}
got := monitor.applyDockerMetadataToUnifiedResources(resources)
got := monitor.applyPersistedMetadataToUnifiedResources(resources)
if got[0].CustomURL != "" {
t.Fatalf("projected CustomURL = %q, want empty (cleared)", got[0].CustomURL)
}
@@ -716,12 +900,12 @@ func TestApplyDockerMetadataToUnifiedResourcesAddsContainerCustomURL(t *testing.
},
}
got := monitor.applyDockerMetadataToUnifiedResources(resources)
got := monitor.applyPersistedMetadataToUnifiedResources(resources)
if got[0].CustomURL != "https://app.internal" {
t.Fatalf("CustomURL = %q, want migrated Docker metadata URL", got[0].CustomURL)
}
if resources[0].CustomURL != "" {
t.Fatalf("applyDockerMetadataToUnifiedResources mutated input CustomURL to %q", resources[0].CustomURL)
t.Fatalf("applyPersistedMetadataToUnifiedResources mutated input CustomURL to %q", resources[0].CustomURL)
}
}
@@ -745,7 +929,7 @@ func TestApplyDockerMetadataToUnifiedResourcesKeepsResourceCustomURL(t *testing.
},
}
got := monitor.applyDockerMetadataToUnifiedResources(resources)
got := monitor.applyPersistedMetadataToUnifiedResources(resources)
if got[0].CustomURL != "https://resource.internal" {
t.Fatalf("CustomURL = %q, want existing resource URL to win", got[0].CustomURL)
}
@@ -771,7 +955,7 @@ func TestApplyDockerMetadataToUnifiedResourcesUsesStableDockerMetadata(t *testin
},
}
got := monitor.applyDockerMetadataToUnifiedResources(resources)
got := monitor.applyPersistedMetadataToUnifiedResources(resources)
if got[0].CustomURL != "https://stable-docker.internal" {
t.Fatalf("CustomURL = %q, want stable Docker metadata URL", got[0].CustomURL)
}
@@ -802,12 +986,185 @@ func TestApplyDockerMetadataToUnifiedResourcesStableGuestMetadataBlocksLegacyFal
},
}
got := monitor.applyDockerMetadataToUnifiedResources(resources)
got := monitor.applyPersistedMetadataToUnifiedResources(resources)
if got[0].CustomURL != "" {
t.Fatalf("CustomURL = %q, want stable empty guest metadata to block legacy fallback", got[0].CustomURL)
}
}
func TestApplyPersistedMetadataToUnifiedResourcesMigratesKubernetesURLToStableLogicalIdentity(t *testing.T) {
monitor := newTestMonitor(t)
legacyPodID := "k8s:cluster-a:pod:pod-uid-old"
if err := monitor.guestMetadataStore.Set(legacyPodID, &config.GuestMetadata{
CustomURL: "https://checkout.internal",
}); err != nil {
t.Fatalf("seed legacy pod metadata: %v", err)
}
oldPod := unifiedresources.Resource{
ID: "resource:pod:old",
Type: unifiedresources.ResourceTypePod,
Name: "checkout",
Kubernetes: &unifiedresources.K8sData{
ClusterID: "cluster-a",
Namespace: "payments",
ResourceKind: "Pod",
PodUID: "pod-uid-old",
},
}
got := monitor.applyPersistedMetadataToUnifiedResources([]unifiedresources.Resource{oldPod})
if got[0].CustomURL != "https://checkout.internal" {
t.Fatalf("legacy pod CustomURL = %q, want migrated URL", got[0].CustomURL)
}
recreatedPod := oldPod
recreatedPod.ID = "resource:pod:new"
recreatedPod.Kubernetes = &unifiedresources.K8sData{
ClusterID: "cluster-a",
Namespace: "payments",
ResourceKind: "Pod",
PodUID: "pod-uid-new",
}
got = monitor.applyPersistedMetadataToUnifiedResources([]unifiedresources.Resource{recreatedPod})
if got[0].CustomURL != "https://checkout.internal" {
t.Fatalf("recreated pod CustomURL = %q, want stable logical URL", got[0].CustomURL)
}
unrelated := []unifiedresources.Resource{
{
ID: "resource:pod:other-cluster",
Type: unifiedresources.ResourceTypePod,
Name: "checkout",
Kubernetes: &unifiedresources.K8sData{
ClusterID: "cluster-b",
Namespace: "payments",
ResourceKind: "Pod",
PodUID: "pod-uid-other-cluster",
},
},
{
ID: "resource:pod:other-namespace",
Type: unifiedresources.ResourceTypePod,
Name: "checkout",
Kubernetes: &unifiedresources.K8sData{
ClusterID: "cluster-a",
Namespace: "staging",
ResourceKind: "Pod",
PodUID: "pod-uid-other-namespace",
},
},
{
ID: "resource:deployment:same-name",
Type: unifiedresources.ResourceTypeK8sDeployment,
Name: "checkout",
Kubernetes: &unifiedresources.K8sData{
ClusterID: "cluster-a",
Namespace: "payments",
ResourceKind: "Deployment",
ResourceUID: "deployment-uid",
},
},
{
ID: "resource:pod:different-name",
Type: unifiedresources.ResourceTypePod,
Name: "checkout-canary",
Kubernetes: &unifiedresources.K8sData{
ClusterID: "cluster-a",
Namespace: "payments",
ResourceKind: "Pod",
PodUID: "pod-uid-different-name",
},
},
}
got = monitor.applyPersistedMetadataToUnifiedResources(unrelated)
for i := range got {
if got[i].CustomURL != "" {
t.Fatalf("unrelated resource %d inherited CustomURL %q", i, got[i].CustomURL)
}
}
}
func TestApplyPersistedMetadataToUnifiedResourcesMigratesKubernetesControllersToStableLogicalIdentity(t *testing.T) {
for _, testCase := range []struct {
name string
resourceType unifiedresources.ResourceType
kind string
}{
{
name: "deployment",
resourceType: unifiedresources.ResourceTypeK8sDeployment,
kind: "Deployment",
},
{
name: "service",
resourceType: unifiedresources.ResourceTypeK8sService,
kind: "Service",
},
} {
t.Run(testCase.name, func(t *testing.T) {
monitor := newTestMonitor(t)
oldResourceID := "resource:" + testCase.name + ":old"
if err := monitor.guestMetadataStore.Set(oldResourceID, &config.GuestMetadata{
CustomURL: "https://" + testCase.name + ".internal",
}); err != nil {
t.Fatalf("seed legacy metadata: %v", err)
}
resource := unifiedresources.Resource{
ID: oldResourceID,
Type: testCase.resourceType,
Name: "checkout",
Kubernetes: &unifiedresources.K8sData{
ClusterID: "cluster-a",
Namespace: "payments",
ResourceKind: testCase.kind,
ResourceUID: testCase.name + "-uid-old",
},
}
got := monitor.applyPersistedMetadataToUnifiedResources([]unifiedresources.Resource{resource})
if got[0].CustomURL != "https://"+testCase.name+".internal" {
t.Fatalf("legacy %s CustomURL = %q", testCase.name, got[0].CustomURL)
}
resource.ID = "resource:" + testCase.name + ":new"
resource.Kubernetes.ResourceUID = testCase.name + "-uid-new"
got = monitor.applyPersistedMetadataToUnifiedResources([]unifiedresources.Resource{resource})
if got[0].CustomURL != "https://"+testCase.name+".internal" {
t.Fatalf("recreated %s CustomURL = %q", testCase.name, got[0].CustomURL)
}
})
}
}
func TestApplyPersistedMetadataToUnifiedResourcesKubernetesStableClearBlocksLegacyFallback(t *testing.T) {
monitor := newTestMonitor(t)
resource := unifiedresources.Resource{
ID: "resource:pod:current",
Type: unifiedresources.ResourceTypePod,
Name: "checkout",
Kubernetes: &unifiedresources.K8sData{
ClusterID: "cluster-a",
Namespace: "payments",
ResourceKind: "Pod",
PodUID: "pod-uid-current",
},
}
stableKey := "k8s-workload:cluster-a:pod:payments:checkout"
if err := monitor.guestMetadataStore.Set(stableKey, &config.GuestMetadata{CustomURL: ""}); err != nil {
t.Fatalf("seed cleared stable pod metadata: %v", err)
}
if err := monitor.guestMetadataStore.Set("k8s:cluster-a:pod:pod-uid-current", &config.GuestMetadata{
CustomURL: "https://stale.internal",
}); err != nil {
t.Fatalf("seed legacy pod metadata: %v", err)
}
got := monitor.applyPersistedMetadataToUnifiedResources([]unifiedresources.Resource{resource})
if got[0].CustomURL != "" {
t.Fatalf("CustomURL = %q, want stable empty Kubernetes metadata to block legacy fallback", got[0].CustomURL)
}
}
func TestApplyDockerReportComputesContainerNetworkAndDiskRates(t *testing.T) {
monitor := newTestMonitor(t)
baseTime := time.Now().UTC()
@@ -225,6 +225,11 @@ func (mtm *MultiTenantMonitor) GetMonitor(orgID string) (*Monitor, error) {
if err != nil {
return nil, fmt.Errorf("failed to create monitor for org %s: %w", orgID, err)
}
tenantPersistence.SetMetadataStores(
monitor.GuestMetadataStore(),
monitor.DockerMetadataStore(),
monitor.HostMetadataStore(),
)
// Set org ID for tenant isolation
// This enables tenant-scoped WebSocket broadcasts