diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index 7125b2f88e..a463d73638 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -1002,6 +1002,13 @@ completion may proceed only when it either dedupes onto an already-counted top-level monitored system or fits within the deduped monitored-system limit; the lifecycle surface may not preserve a special exemption that keeps API- backed monitored systems outside the self-hosted commercial cap. +That admission decision must come from the same canonical prospective +monitored-system projection the runtime uses for final grouped counting. +Auto-register may preview its own candidate, but it must not keep a +lifecycle-local counter, drift on source priority, or treat missing monitored- +system usage as zero; when an active cap is present and usage cannot be +resolved, the route must fail closed with retryable unavailable guidance +instead of silently admitting a net-new monitored system. That same validation contract must stay coherent across the public `/api/auto-register` route and the direct canonical handler path used by the same runtime surface, so Unified Agent/setup entry points do not inherit divergent diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index e3b194b179..4cf29e3bf0 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -404,6 +404,13 @@ Own canonical runtime payload shapes between backend and frontend. The API layer already uses contract tests in many places, but every major live contract should continue moving toward canonical-only runtime shapes. +Monitored-system commercial admission is now also part of that owned live +contract. Add and update routes must project prospective candidates or +previewed source records through the canonical monitored-system resolver +before persistence, and `/api/license/entitlements` must expose +`current_available` when an active monitored-system cap cannot resolve current +usage so callers can fail closed without misreading unavailable usage as a +real zero. The shared metrics-history contract now also owns physical-disk live I/O windows. `/api/metrics-store/history` must accept `resourceType=disk`, keep `30m` as a valid compact live range, and resolve `disk`, `diskread`, diff --git a/docs/release-control/v6/internal/subsystems/cloud-paid.md b/docs/release-control/v6/internal/subsystems/cloud-paid.md index a10e30dab8..6658a8394f 100644 --- a/docs/release-control/v6/internal/subsystems/cloud-paid.md +++ b/docs/release-control/v6/internal/subsystems/cloud-paid.md @@ -361,6 +361,17 @@ decode-only compatibility inputs at the storage or grant boundary. Runtime enforcement, entitlement payload `current` usage, checkout/activation flows, and upgrade messaging must all treat the cap as deduped top-level monitored systems across agent, API, and Kubernetes views. +That same counted-unit contract now also owns prospective API-backed +admission. Proxmox/PBS/PMG config adds, TrueNAS adds, VMware inventory +previews, and equivalent updates must ask the canonical monitored-system +projection whether they increase counted systems before the runtime persists +them, including replacement-aware projections for source swaps on an existing +grouped host. +Under an active monitored-system cap, inability to resolve current usage is +not a free pass. Runtime enforcement and entitlement payloads must fail closed +for net-new monitored-system admissions and surface usage availability +explicitly through the governed limit payload instead of treating unavailable +usage as `current: 0`. The monitored-system ledger settings surface now also has to explain those count decisions. Commercial usage UI may show grouped monitored systems, but it must render the canonical backend explanation for why one or more diff --git a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md index 46d0c5b6bb..8c7f306f38 100644 --- a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md +++ b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md @@ -221,6 +221,12 @@ sampling work and drift out of sync with the governed history timeline. All governed performance-owned runtime files now require explicit registry path-policy coverage, so new protected hot paths must be mapped to a concrete proof route instead of falling back to subsystem-default verification. +Monitored-system admission and entitlement usage now also sit on a protected +backend hot path. `internal/api/monitored_system_limit_enforcement.go` and +the unified-resource projection helpers must reuse one current monitored- +system snapshot plus prospective candidate/preview projection for add and +update checks instead of rescanning platform inventories per handler or +falling back to zero when monitor state is unavailable. The dashboard workload selector path and the dashboard runtime that consumes it are now part of the protected performance surface rather than proof-only diff --git a/docs/release-control/v6/internal/subsystems/storage-recovery.md b/docs/release-control/v6/internal/subsystems/storage-recovery.md index 8bf46a0542..c636c98684 100644 --- a/docs/release-control/v6/internal/subsystems/storage-recovery.md +++ b/docs/release-control/v6/internal/subsystems/storage-recovery.md @@ -325,6 +325,12 @@ querying, and the operator-facing storage health presentation layer. This subsystem now sits under the dedicated storage and recovery lane so the operator-facing storage page, recovery timeline, and recovery-point persistence engine stop hiding inside broader monitoring and E2E buckets. +That same shared `internal/api/` dependency now also expects replacement-aware +monitored-system admission and fail-closed usage availability. Storage- or +recovery-adjacent setup, deploy, and API-backed update helpers may reuse the +canonical monitored-system cap boundary, but they must preserve remaining +grouped sources on a counted host and must not reinterpret unavailable usage as +an empty estate. That same adjacent `internal/api/` boundary now also governs public-demo commercial redaction for storage and recovery viewers. Shared storage/recovery surfaces may run beside a demo runtime that has real internal entitlements, @@ -1420,6 +1426,12 @@ shared setup, deploy, entitlement, and API-backed monitoring helpers may not preserve an API-only exemption that would let storage- or recovery-adjacent systems consume no commercial slot when the same monitored system is visible through canonical unified-resource roots. +That same shared boundary now also assumes replacement-aware monitored-system +projection. When a storage- or recovery-adjacent update replaces one source on +an already-counted host, the API helper must strip only that source from the +prospective grouped system and preserve any remaining top-level evidence such +as agent or sibling API ownership, rather than briefly freeing a slot or +double-counting the same monitored system. That same shared `internal/api/` dependency also assumes session-carried OIDC refresh tokens stay fail-closed at rest: `session_store.go` may only persist or recover those tokens through encrypted-at-rest session payloads, and any diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index 7141d4829e..283a1930c3 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -261,6 +261,11 @@ canonical resource identity, discovery normalization, and platform-runtime coverage stay governed as a first-class Pulse product surface, including the shared VMware signal-metadata and `resource-incident` timeline vocabulary that canonical resources expose to alerts, AI, and frontend consumers. +That same runtime now also owns prospective monitored-system projection. Add +and update consumers must ask the unified-resource layer whether candidate or +preview records change the deduped top-level monitored-system count, including +replacement-aware projection and VMware host-UUID identity, instead of +rebuilding handler-local counting heuristics. That shared consumer ownership now includes same-tab summary hydration too. `frontend-modern/src/components/Infrastructure/useInfrastructureSummaryState.ts` may keep an in-memory remount cache for canonical resource charts, but the key @@ -1374,6 +1379,15 @@ commercial entitlement and ledger surfaces. `MonitoredSystems(...)` must keep top-level counted-system identity, representative resource selection, and agent/API deduplication on unified-resource ownership instead of letting API handlers or licensing helpers rebuild their own counted-system grouping logic. +That same registry-owned contract now also covers prospective admission and +source replacement. Add/update handlers must project candidates or preview +records through shared monitored-system projection helpers, including the +delta from replacing one existing source-owned surface, instead of guessing +from handler-local priority rules or transport-specific counters. +VMware host previews are part of that same contract: canonical host identity +must honor VMware host UUID plus normalized hostnames so a vCenter add or +update cannot bypass the monitored-system cap by discovering host-backed +systems only after persistence. Canonical source-owned identifiers must also normalize surrounding whitespace before they become by-source map keys or source-specific hash IDs. The same @@ -1513,6 +1527,11 @@ canonical unified-resource roots, so read-state helpers that derive commercial-count groups must union agent, Proxmox, Docker, PBS, PMG, TrueNAS, and Kubernetes cluster views through canonical identity evidence instead of through transport-local counters or child-resource totals. +When one counted group is being updated in place, the prospective projection +must remove only the replaced source from that grouped root and preserve any +remaining canonical source ownership that still keeps the monitored system +counted, so replacement-aware admission stays aligned with final runtime +counting. Canonical unified resources now also own first-class policy metadata for the v6 bridge release. Cloned and API-exported resources must carry diff --git a/internal/api/config_handlers_add_test.go b/internal/api/config_handlers_add_test.go index 6be218fb2d..5996fbf7e4 100644 --- a/internal/api/config_handlers_add_test.go +++ b/internal/api/config_handlers_add_test.go @@ -312,6 +312,85 @@ func TestHandleAddNode_BlocksNewCountedSystemAtLimit(t *testing.T) { } } +func TestHandleAddNode_AllowsCanonicalOverlapAtLimit(t *testing.T) { + stubAutoRegisterNetworkDeps(t) + + setMaxMonitoredSystemsLicenseForTests(t, 1) + + cfg := &config.Config{ + DataPath: t.TempDir(), + } + handler := newTestConfigHandlers(t, cfg) + registry := unifiedresources.NewRegistry(nil) + registry.IngestRecords(unifiedresources.SourceAgent, []unifiedresources.IngestRecord{ + { + SourceID: "host-1", + Resource: unifiedresources.Resource{ + ID: "host-1", + Type: unifiedresources.ResourceTypeAgent, + Name: "tower.local", + Status: unifiedresources.StatusOnline, + Agent: &unifiedresources.AgentData{ + AgentID: "agent-1", + Hostname: "tower.local", + MachineID: "machine-1", + }, + Identity: unifiedresources.ResourceIdentity{ + MachineID: "machine-1", + Hostnames: []string{"tower.local"}, + }, + }, + }, + }) + monitor := &monitoring.Monitor{} + monitor.SetResourceStore(unifiedresources.NewMonitorAdapter(registry)) + handler.defaultMonitor = monitor + + body, _ := json.Marshal(map[string]any{ + "name": "tower", + "type": "pve", + "host": "tower.local", + "tokenName": "root@pam!pulse", + "tokenValue": "secret", + }) + + req := httptest.NewRequest(http.MethodPost, "/api/config/nodes", bytes.NewBuffer(body)) + rec := httptest.NewRecorder() + handler.HandleAddNode(rec, req) + + if rec.Code != http.StatusCreated { + t.Fatalf("expected overlapping Proxmox registration to be allowed at limit, got %d: %s", rec.Code, rec.Body.String()) + } +} + +func TestHandleAddNode_FailsClosedWhenUsageUnavailable(t *testing.T) { + stubAutoRegisterNetworkDeps(t) + + setMaxMonitoredSystemsLicenseForTests(t, 1) + + cfg := &config.Config{ + DataPath: t.TempDir(), + } + handler := newTestConfigHandlers(t, cfg) + handler.defaultMonitor = nil + + body, _ := json.Marshal(map[string]any{ + "name": "new-node", + "type": "pve", + "host": "10.0.0.2", + "tokenName": "root@pam!pulse", + "tokenValue": "secret", + }) + + req := httptest.NewRequest(http.MethodPost, "/api/config/nodes", bytes.NewBuffer(body)) + rec := httptest.NewRecorder() + handler.HandleAddNode(rec, req) + + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("expected 503 when monitored-system usage cannot be resolved, got %d: %s", rec.Code, rec.Body.String()) + } +} + func TestHandleAddNodeConsolidatesStandaloneOverlapIntoClusterInMemory(t *testing.T) { tempDir := t.TempDir() diff --git a/internal/api/config_handlers_auto_register_test.go b/internal/api/config_handlers_auto_register_test.go index 6b3c0e052d..9858be9769 100644 --- a/internal/api/config_handlers_auto_register_test.go +++ b/internal/api/config_handlers_auto_register_test.go @@ -9,6 +9,7 @@ import ( "testing" "time" + "github.com/rcourtman/pulse-go-rewrite/internal/alerts" "github.com/rcourtman/pulse-go-rewrite/internal/config" "github.com/rcourtman/pulse-go-rewrite/internal/monitoring" "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" @@ -28,6 +29,9 @@ func newTestConfigHandlers(t *testing.T, cfg *config.Config) *ConfigHandlers { h := NewConfigHandlers(nil, nil, func() error { return nil }, nil, nil, func() {}) h.defaultConfig = cfg h.defaultPersistence = config.NewConfigPersistence(cfg.DataPath) + monitor, _, _ := newTestMonitor(t) + setUnexportedField(t, monitor, "alertManager", alerts.NewManager()) + h.defaultMonitor = monitor return h } diff --git a/internal/api/config_handlers_update_test.go b/internal/api/config_handlers_update_test.go index 5bc7ce53cc..bb69488ef4 100644 --- a/internal/api/config_handlers_update_test.go +++ b/internal/api/config_handlers_update_test.go @@ -9,6 +9,8 @@ import ( "testing" "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/internal/monitoring" + "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" ) func TestHandleUpdateNode(t *testing.T) { @@ -142,3 +144,76 @@ func TestHandleUpdateNode(t *testing.T) { }) } } + +func TestHandleUpdateNode_BlocksProjectedNetNewSystemAtLimit(t *testing.T) { + setMaxMonitoredSystemsLicenseForTests(t, 1) + + cfg := &config.Config{ + DataPath: t.TempDir(), + PVEInstances: []config.PVEInstance{{ + Name: "tower", + Host: "https://tower.local:8006", + TokenName: "root@pam!pulse", + TokenValue: "secret", + }}, + } + handler := newTestConfigHandlers(t, cfg) + + registry := unifiedresources.NewRegistry(nil) + registry.IngestRecords(unifiedresources.SourceAgent, []unifiedresources.IngestRecord{ + { + SourceID: "host-1", + Resource: unifiedresources.Resource{ + ID: "host-1", + Type: unifiedresources.ResourceTypeAgent, + Name: "tower.local", + Status: unifiedresources.StatusOnline, + Agent: &unifiedresources.AgentData{ + AgentID: "agent-1", + Hostname: "tower.local", + MachineID: "machine-1", + }, + Identity: unifiedresources.ResourceIdentity{ + MachineID: "machine-1", + Hostnames: []string{"tower.local"}, + }, + }, + }, + }) + registry.IngestRecords(unifiedresources.SourceProxmox, []unifiedresources.IngestRecord{ + { + SourceID: "pve-1", + Resource: unifiedresources.Resource{ + ID: "tower-resource", + Type: unifiedresources.ResourceTypeAgent, + Name: "tower", + Status: unifiedresources.StatusOnline, + Proxmox: &unifiedresources.ProxmoxData{ + Instance: "tower", + NodeName: "tower", + HostURL: "https://tower.local:8006", + }, + }, + Identity: unifiedresources.ResourceIdentity{ + MachineID: "machine-1", + Hostnames: []string{"tower.local"}, + }, + }, + }) + setUnexportedField(t, handler.defaultMonitor, "resourceStore", monitoring.ResourceStoreInterface(unifiedresources.NewMonitorAdapter(registry))) + + body, _ := json.Marshal(map[string]any{ + "host": "https://backup.local:8006", + }) + req := httptest.NewRequest(http.MethodPut, "/api/config/nodes/pve-0", bytes.NewBuffer(body)) + rec := httptest.NewRecorder() + + handler.HandleUpdateNode(rec, req) + + if rec.Code != http.StatusPaymentRequired { + t.Fatalf("expected 402 when update would add a new monitored system, got %d: %s", rec.Code, rec.Body.String()) + } + if got := cfg.PVEInstances[0].Host; got != "https://tower.local:8006" { + t.Fatalf("expected blocked update to preserve original host, got %q", got) + } +} diff --git a/internal/api/config_node_handlers.go b/internal/api/config_node_handlers.go index bf2ea26042..ec3dc85794 100644 --- a/internal/api/config_node_handlers.go +++ b/internal/api/config_node_handlers.go @@ -454,9 +454,10 @@ func (h *ConfigHandlers) handleAddNode(w http.ResponseWriter, r *http.Request) { } if enforceMonitoredSystemLimitForConfigRegistration(w, r.Context(), h.getConfig(r.Context()), h.getMonitor(r.Context()), unifiedresources.MonitoredSystemCandidate{ + Source: unifiedresources.SourceProxmox, Type: unifiedresources.ResourceTypeAgent, Name: displayName, - Hostname: displayName, + Hostname: pulseTokenHostCandidate(host), HostURL: host, }) { return @@ -586,9 +587,10 @@ func (h *ConfigHandlers) handleAddNode(w http.ResponseWriter, r *http.Request) { TemperatureMonitoringEnabled: req.TemperatureMonitoringEnabled, } if enforceMonitoredSystemLimitForConfigRegistration(w, r.Context(), h.getConfig(r.Context()), h.getMonitor(r.Context()), unifiedresources.MonitoredSystemCandidate{ + Source: unifiedresources.SourcePBS, Type: unifiedresources.ResourceTypePBS, Name: pbsDisplayName, - Hostname: pbsDisplayName, + Hostname: pulseTokenHostCandidate(host), HostURL: host, }) { return @@ -676,9 +678,10 @@ func (h *ConfigHandlers) handleAddNode(w http.ResponseWriter, r *http.Request) { TemperatureMonitoringEnabled: req.TemperatureMonitoringEnabled, } if enforceMonitoredSystemLimitForConfigRegistration(w, r.Context(), h.getConfig(r.Context()), h.getMonitor(r.Context()), unifiedresources.MonitoredSystemCandidate{ + Source: unifiedresources.SourcePMG, Type: unifiedresources.ResourceTypePMG, Name: pmgDisplayName, - Hostname: pmgDisplayName, + Hostname: pulseTokenHostCandidate(host), HostURL: host, }) { return @@ -710,6 +713,98 @@ func (h *ConfigHandlers) handleAddNode(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(map[string]string{"status": "success"}) } +func proxmoxMonitoredSystemCandidate(instance config.PVEInstance) unifiedresources.MonitoredSystemCandidate { + return unifiedresources.MonitoredSystemCandidate{ + Source: unifiedresources.SourceProxmox, + Type: unifiedresources.ResourceTypeAgent, + Name: instance.Name, + Hostname: pulseTokenHostCandidate(instance.Host), + HostURL: instance.Host, + } +} + +func pbsMonitoredSystemCandidate(instance config.PBSInstance) unifiedresources.MonitoredSystemCandidate { + return unifiedresources.MonitoredSystemCandidate{ + Source: unifiedresources.SourcePBS, + Type: unifiedresources.ResourceTypePBS, + Name: instance.Name, + Hostname: pulseTokenHostCandidate(instance.Host), + HostURL: instance.Host, + } +} + +func pmgMonitoredSystemCandidate(instance config.PMGInstance) unifiedresources.MonitoredSystemCandidate { + return unifiedresources.MonitoredSystemCandidate{ + Source: unifiedresources.SourcePMG, + Type: unifiedresources.ResourceTypePMG, + Name: instance.Name, + Hostname: pulseTokenHostCandidate(instance.Host), + HostURL: instance.Host, + } +} + +func proxmoxMonitoredSystemReplacement(instance config.PVEInstance) unifiedresources.MonitoredSystemReplacement { + name := strings.TrimSpace(instance.Name) + host := strings.TrimSpace(instance.Host) + hostname := pulseTokenHostCandidate(instance.Host) + return unifiedresources.MonitoredSystemReplacement{ + Source: unifiedresources.SourceProxmox, + Matches: func(resource unifiedresources.Resource) bool { + if resource.Proxmox == nil { + return false + } + if name != "" && strings.EqualFold(strings.TrimSpace(resource.Proxmox.Instance), name) { + return true + } + if name != "" && strings.EqualFold(strings.TrimSpace(resource.Proxmox.NodeName), name) { + return true + } + if host != "" && strings.EqualFold(strings.TrimSpace(resource.Proxmox.HostURL), host) { + return true + } + return hostname != "" && strings.EqualFold(pulseTokenHostCandidate(resource.Proxmox.HostURL), hostname) + }, + } +} + +func pbsMonitoredSystemReplacement(instance config.PBSInstance) unifiedresources.MonitoredSystemReplacement { + name := strings.TrimSpace(instance.Name) + host := strings.TrimSpace(instance.Host) + hostname := pulseTokenHostCandidate(instance.Host) + return unifiedresources.MonitoredSystemReplacement{ + Source: unifiedresources.SourcePBS, + Matches: func(resource unifiedresources.Resource) bool { + if resource.PBS == nil { + return false + } + if name != "" && strings.EqualFold(strings.TrimSpace(resource.PBS.InstanceID), name) { + return true + } + if host != "" && strings.EqualFold(strings.TrimSpace(resource.PBS.HostURL), host) { + return true + } + return hostname != "" && strings.EqualFold(strings.TrimSpace(resource.PBS.Hostname), hostname) + }, + } +} + +func pmgMonitoredSystemReplacement(instance config.PMGInstance) unifiedresources.MonitoredSystemReplacement { + name := strings.TrimSpace(instance.Name) + hostname := pulseTokenHostCandidate(instance.Host) + return unifiedresources.MonitoredSystemReplacement{ + Source: unifiedresources.SourcePMG, + Matches: func(resource unifiedresources.Resource) bool { + if resource.PMG == nil { + return false + } + if name != "" && strings.EqualFold(strings.TrimSpace(resource.PMG.InstanceID), name) { + return true + } + return hostname != "" && strings.EqualFold(strings.TrimSpace(resource.PMG.Hostname), hostname) + }, + } +} + // HandleTestConnection tests a node connection without saving func (h *ConfigHandlers) handleTestConnection(w http.ResponseWriter, r *http.Request) { @@ -1057,10 +1152,12 @@ func (h *ConfigHandlers) handleUpdateNode(w http.ResponseWriter, r *http.Request // Update the node if nodeType == "pve" && index < len(h.getConfig(r.Context()).PVEInstances) { pve := &h.getConfig(r.Context()).PVEInstances[index] + current := *pve + updated := current // Only update name if provided if req.Name != "" { - pve.Name = req.Name + updated.Name = req.Name } if req.Host != "" { @@ -1069,76 +1166,89 @@ func (h *ConfigHandlers) handleUpdateNode(w http.ResponseWriter, r *http.Request http.Error(w, err.Error(), http.StatusBadRequest) return } - pve.Host = host + updated.Host = host } // Update GuestURL if provided - pve.GuestURL = req.GuestURL + updated.GuestURL = req.GuestURL // Handle authentication updates - only switch auth method if explicitly provided if req.TokenName != "" || req.TokenValue != "" { // Switching to or updating token authentication if req.TokenName != "" { - pve.TokenName = req.TokenName + updated.TokenName = req.TokenName } if req.TokenValue != "" { - pve.TokenValue = req.TokenValue + updated.TokenValue = req.TokenValue } // Clear password to avoid conflicts - pve.Password = "" + updated.Password = "" if req.User != "" { - pve.User = req.User + updated.User = req.User } } else if req.Password != "" { // Explicitly switching to password authentication if req.User != "" { - pve.User = normalizePVEUser(req.User) - } else if pve.User != "" { - pve.User = normalizePVEUser(pve.User) + updated.User = normalizePVEUser(req.User) + } else if updated.User != "" { + updated.User = normalizePVEUser(updated.User) } - pve.Password = req.Password + updated.Password = req.Password // Clear token fields when switching to password auth - pve.TokenName = "" - pve.TokenValue = "" + updated.TokenName = "" + updated.TokenValue = "" } else { // No authentication changes - preserve existing auth fields // Only normalize user if it exists - if pve.User != "" { - pve.User = normalizePVEUser(pve.User) + if updated.User != "" { + updated.User = normalizePVEUser(updated.User) } } - pve.Fingerprint = req.Fingerprint + updated.Fingerprint = req.Fingerprint if req.VerifySSL != nil { - pve.VerifySSL = *req.VerifySSL + updated.VerifySSL = *req.VerifySSL } if req.MonitorVMs != nil { - pve.MonitorVMs = *req.MonitorVMs + updated.MonitorVMs = *req.MonitorVMs } if req.MonitorContainers != nil { - pve.MonitorContainers = *req.MonitorContainers + updated.MonitorContainers = *req.MonitorContainers } if req.MonitorStorage != nil { - pve.MonitorStorage = *req.MonitorStorage + updated.MonitorStorage = *req.MonitorStorage } if req.MonitorBackups != nil { - pve.MonitorBackups = *req.MonitorBackups + updated.MonitorBackups = *req.MonitorBackups } if req.MonitorPhysicalDisks != nil { - pve.MonitorPhysicalDisks = req.MonitorPhysicalDisks + updated.MonitorPhysicalDisks = req.MonitorPhysicalDisks } if req.PhysicalDiskPollingMinutes != nil { - pve.PhysicalDiskPollingMinutes = *req.PhysicalDiskPollingMinutes + updated.PhysicalDiskPollingMinutes = *req.PhysicalDiskPollingMinutes } if req.TemperatureMonitoringEnabled != nil { - pve.TemperatureMonitoringEnabled = req.TemperatureMonitoringEnabled + updated.TemperatureMonitoringEnabled = req.TemperatureMonitoringEnabled } + + if enforceMonitoredSystemLimitForConfigReplacement( + w, + r.Context(), + h.getMonitor(r.Context()), + proxmoxMonitoredSystemReplacement(current), + proxmoxMonitoredSystemCandidate(updated), + ) { + return + } + *pve = updated } else if nodeType == "pbs" && index < len(h.getConfig(r.Context()).PBSInstances) { pbs := &h.getConfig(r.Context()).PBSInstances[index] + current := *pbs + updated := current // Only update name if provided if req.Name != "" { - pbs.Name = req.Name + updated.Name = req.Name } if req.Host != "" { @@ -1147,38 +1257,38 @@ func (h *ConfigHandlers) handleUpdateNode(w http.ResponseWriter, r *http.Request http.Error(w, err.Error(), http.StatusBadRequest) return } - pbs.Host = host + updated.Host = host } // Update GuestURL if provided - pbs.GuestURL = req.GuestURL + updated.GuestURL = req.GuestURL // Handle authentication updates - only switch auth method if explicitly provided if req.TokenName != "" && req.TokenValue != "" { // Switching to token authentication - pbs.TokenName = req.TokenName - pbs.TokenValue = req.TokenValue + updated.TokenName = req.TokenName + updated.TokenValue = req.TokenValue // Clear user/password when switching to token auth - pbs.User = "" - pbs.Password = "" + updated.User = "" + updated.Password = "" } else if req.TokenName != "" { // Token name provided without new value - keep existing token value - pbs.TokenName = req.TokenName + updated.TokenName = req.TokenName // Clear user/password when using token auth - pbs.User = "" - pbs.Password = "" + updated.User = "" + updated.Password = "" } else if req.Password != "" { // Switching to password authentication - pbs.Password = req.Password + updated.Password = req.Password // Ensure user has realm for PBS pbsUser := req.User if req.User != "" && !strings.Contains(req.User, "@") { pbsUser = req.User + "@pbs" // Default to @pbs realm if not specified } - pbs.User = pbsUser + updated.User = pbsUser // Clear token fields when switching to password auth - pbs.TokenName = "" - pbs.TokenValue = "" + updated.TokenName = "" + updated.TokenValue = "" } else if req.User != "" { // User provided - assume password auth but keep existing password // Ensure user has realm for PBS @@ -1186,47 +1296,60 @@ func (h *ConfigHandlers) handleUpdateNode(w http.ResponseWriter, r *http.Request if !strings.Contains(req.User, "@") { pbsUser = req.User + "@pbs" // Default to @pbs realm if not specified } - pbs.User = pbsUser + updated.User = pbsUser // Clear token fields when using password auth - pbs.TokenName = "" - pbs.TokenValue = "" + updated.TokenName = "" + updated.TokenValue = "" } // else: No authentication changes - preserve existing auth fields - pbs.Fingerprint = req.Fingerprint + updated.Fingerprint = req.Fingerprint if req.VerifySSL != nil { - pbs.VerifySSL = *req.VerifySSL + updated.VerifySSL = *req.VerifySSL } if req.MonitorBackups != nil { - pbs.MonitorBackups = *req.MonitorBackups + updated.MonitorBackups = *req.MonitorBackups } else { - pbs.MonitorBackups = true // Enable by default for PBS + updated.MonitorBackups = true // Enable by default for PBS } if req.MonitorDatastores != nil { - pbs.MonitorDatastores = *req.MonitorDatastores + updated.MonitorDatastores = *req.MonitorDatastores } if req.MonitorSyncJobs != nil { - pbs.MonitorSyncJobs = *req.MonitorSyncJobs + updated.MonitorSyncJobs = *req.MonitorSyncJobs } if req.MonitorVerifyJobs != nil { - pbs.MonitorVerifyJobs = *req.MonitorVerifyJobs + updated.MonitorVerifyJobs = *req.MonitorVerifyJobs } if req.MonitorPruneJobs != nil { - pbs.MonitorPruneJobs = *req.MonitorPruneJobs + updated.MonitorPruneJobs = *req.MonitorPruneJobs } if req.MonitorGarbageJobs != nil { - pbs.MonitorGarbageJobs = *req.MonitorGarbageJobs + updated.MonitorGarbageJobs = *req.MonitorGarbageJobs } if req.TemperatureMonitoringEnabled != nil { - pbs.TemperatureMonitoringEnabled = req.TemperatureMonitoringEnabled + updated.TemperatureMonitoringEnabled = req.TemperatureMonitoringEnabled } // Update datastore exclusion list if req.ExcludeDatastores != nil { - pbs.ExcludeDatastores = req.ExcludeDatastores + updated.ExcludeDatastores = req.ExcludeDatastores } + + if enforceMonitoredSystemLimitForConfigReplacement( + w, + r.Context(), + h.getMonitor(r.Context()), + pbsMonitoredSystemReplacement(current), + pbsMonitoredSystemCandidate(updated), + ) { + return + } + *pbs = updated } else if nodeType == "pmg" && index < len(h.getConfig(r.Context()).PMGInstances) { pmgInst := &h.getConfig(r.Context()).PMGInstances[index] - pmgInst.Name = req.Name + current := *pmgInst + updated := current + updated.Name = req.Name if req.Host != "" { host, err := normalizeNodeHost(req.Host, nodeType) @@ -1234,20 +1357,20 @@ func (h *ConfigHandlers) handleUpdateNode(w http.ResponseWriter, r *http.Request http.Error(w, err.Error(), http.StatusBadRequest) return } - pmgInst.Host = host + updated.Host = host } // Update GuestURL if provided - pmgInst.GuestURL = req.GuestURL + updated.GuestURL = req.GuestURL // Handle authentication updates - only switch auth method if explicitly provided if req.TokenName != "" && req.TokenValue != "" { // Switching to token authentication - pmgInst.TokenName = req.TokenName - pmgInst.TokenValue = req.TokenValue + updated.TokenName = req.TokenName + updated.TokenValue = req.TokenValue // Clear user/password when switching to token auth - pmgInst.User = "" - pmgInst.Password = "" + updated.User = "" + updated.Password = "" } else if req.Password != "" { // Switching to password authentication if req.User != "" { @@ -1255,50 +1378,61 @@ func (h *ConfigHandlers) handleUpdateNode(w http.ResponseWriter, r *http.Request if !strings.Contains(user, "@") { user = user + "@pmg" } - pmgInst.User = user + updated.User = user } - pmgInst.Password = req.Password + updated.Password = req.Password // Clear token fields when switching to password auth - pmgInst.TokenName = "" - pmgInst.TokenValue = "" + updated.TokenName = "" + updated.TokenValue = "" } else if req.User != "" { // User provided - assume password auth but keep existing password user := req.User if !strings.Contains(user, "@") { user = user + "@pmg" } - pmgInst.User = user + updated.User = user // Clear token fields when using password auth - pmgInst.TokenName = "" - pmgInst.TokenValue = "" + updated.TokenName = "" + updated.TokenValue = "" } // else: No authentication changes - preserve existing auth fields - pmgInst.Fingerprint = req.Fingerprint + updated.Fingerprint = req.Fingerprint if req.VerifySSL != nil { - pmgInst.VerifySSL = *req.VerifySSL + updated.VerifySSL = *req.VerifySSL } // Special logic for MonitorMailStats: default to true if all monitor flags are false/unset if req.MonitorMailStats != nil { - pmgInst.MonitorMailStats = *req.MonitorMailStats + updated.MonitorMailStats = *req.MonitorMailStats } else if (req.MonitorMailStats == nil || !*req.MonitorMailStats) && (req.MonitorQueues == nil || !*req.MonitorQueues) && (req.MonitorQuarantine == nil || !*req.MonitorQuarantine) && (req.MonitorDomainStats == nil || !*req.MonitorDomainStats) { - pmgInst.MonitorMailStats = true + updated.MonitorMailStats = true } if req.MonitorQueues != nil { - pmgInst.MonitorQueues = *req.MonitorQueues + updated.MonitorQueues = *req.MonitorQueues } if req.MonitorQuarantine != nil { - pmgInst.MonitorQuarantine = *req.MonitorQuarantine + updated.MonitorQuarantine = *req.MonitorQuarantine } if req.MonitorDomainStats != nil { - pmgInst.MonitorDomainStats = *req.MonitorDomainStats + updated.MonitorDomainStats = *req.MonitorDomainStats } if req.TemperatureMonitoringEnabled != nil { - pmgInst.TemperatureMonitoringEnabled = req.TemperatureMonitoringEnabled + updated.TemperatureMonitoringEnabled = req.TemperatureMonitoringEnabled } + + if enforceMonitoredSystemLimitForConfigReplacement( + w, + r.Context(), + h.getMonitor(r.Context()), + pmgMonitoredSystemReplacement(current), + pmgMonitoredSystemCandidate(updated), + ) { + return + } + *pmgInst = updated } else { http.Error(w, "Node not found", http.StatusNotFound) return diff --git a/internal/api/config_setup_handlers.go b/internal/api/config_setup_handlers.go index 98b9aabe19..eb92f03a33 100644 --- a/internal/api/config_setup_handlers.go +++ b/internal/api/config_setup_handlers.go @@ -860,9 +860,10 @@ func (h *ConfigHandlers) handleCanonicalAutoRegister(w http.ResponseWriter, r *h log.Info().Str("host", host).Str("type", "pve").Msg(canonicalAutoRegisterMatchMessage("host; updated token in-place")) } else { if enforceMonitoredSystemLimitForConfigRegistration(w, r.Context(), h.getConfig(r.Context()), h.getMonitor(r.Context()), unifiedresources.MonitoredSystemCandidate{ + Source: unifiedresources.SourceProxmox, Type: unifiedresources.ResourceTypeAgent, Name: serverName, - Hostname: serverName, + Hostname: pulseTokenHostCandidate(host), HostURL: host, }) { return @@ -934,9 +935,10 @@ func (h *ConfigHandlers) handleCanonicalAutoRegister(w http.ResponseWriter, r *h log.Info().Str("host", host).Str("type", "pbs").Msg(canonicalAutoRegisterMatchMessage("host; updated token in-place")) } else { if enforceMonitoredSystemLimitForConfigRegistration(w, r.Context(), h.getConfig(r.Context()), h.getMonitor(r.Context()), unifiedresources.MonitoredSystemCandidate{ + Source: unifiedresources.SourcePBS, Type: unifiedresources.ResourceTypePBS, Name: serverName, - Hostname: serverName, + Hostname: pulseTokenHostCandidate(host), HostURL: host, }) { return diff --git a/internal/api/contract_test.go b/internal/api/contract_test.go index 474ce49b8d..1873233e86 100644 --- a/internal/api/contract_test.go +++ b/internal/api/contract_test.go @@ -4771,6 +4771,38 @@ func TestContract_EntitlementPayloadMonitoredSystemUsageJSONSnapshot(t *testing. assertJSONSnapshot(t, got, want) } +func TestContract_EntitlementPayloadMonitoredSystemUsageUnavailableJSONSnapshot(t *testing.T) { + payload := buildEntitlementPayloadWithUsage(&licenseStatus{ + Valid: true, + Tier: pkglicensing.TierPro, + Features: append([]string(nil), pkglicensing.TierFeatures[pkglicensing.TierPro]...), + MaxMonitoredSystems: 15, + }, string(pkglicensing.SubStateActive), entitlementUsageSnapshot{}, nil) + + got, err := json.Marshal(payload) + if err != nil { + t.Fatalf("marshal entitlement payload: %v", err) + } + + const want = `{ + "capabilities":["update_alerts","sso","ai_patrol","relay","mobile_app","push_notifications","long_term_metrics","ai_alerts","ai_autofix","kubernetes_ai","agent_profiles","advanced_sso","rbac","audit_logging","advanced_reporting"], + "limits":[{"key":"max_monitored_systems","limit":15,"current":0,"current_available":false,"state":"ok"}], + "subscription_state":"active", + "upgrade_reasons":[], + "tier":"pro", + "hosted_mode":false, + "valid":true, + "is_lifetime":false, + "days_remaining":0, + "trial_eligible":false, + "max_history_days":90, + "legacy_connections":{"proxmox_nodes":0,"docker_hosts":0,"kubernetes_clusters":0}, + "has_migration_gap":false + }` + + assertJSONSnapshot(t, got, want) +} + func TestContract_HostedBillingStateFallbackJSONSnapshot(t *testing.T) { baseDir := t.TempDir() store := config.NewFileBillingStore(baseDir) diff --git a/internal/api/deploy_handlers.go b/internal/api/deploy_handlers.go index 86f61f4caa..fa1ee8bcfc 100644 --- a/internal/api/deploy_handlers.go +++ b/internal/api/deploy_handlers.go @@ -1092,9 +1092,12 @@ func (h *DeployHandlers) HandleCreateJob(w http.ResponseWriter, r *http.Request) // License slot check. maxLimit := maxMonitoredSystemsLimitForContext(ctx) if maxLimit > 0 { - currentCount := monitoredSystemCount(h.monitor) - reservedCount := h.reservation.ReservedForOrg(orgID) - available := maxLimit - currentCount - reservedCount + decision := monitoredSystemLimitDecisionForAdditionalSlots(ctx, h.monitor, 0) + if !decision.usageAvailable { + writeMonitoredSystemUsageUnavailable(w) + return + } + available := decision.limit - decision.current if available < 0 { available = 0 } @@ -1589,9 +1592,12 @@ func (h *DeployHandlers) HandleRetryJob(w http.ResponseWriter, r *http.Request) // License slot re-check. maxLimit := maxMonitoredSystemsLimitForContext(ctx) if maxLimit > 0 { - currentCount := monitoredSystemCount(h.monitor) - reservedCount := h.reservation.ReservedForOrg(orgID) - available := maxLimit - currentCount - reservedCount + decision := monitoredSystemLimitDecisionForAdditionalSlots(ctx, h.monitor, 0) + if !decision.usageAvailable { + writeMonitoredSystemUsageUnavailable(w) + return + } + available := decision.limit - decision.current if available < 0 { available = 0 } diff --git a/internal/api/entitlement_handlers_test.go b/internal/api/entitlement_handlers_test.go index a87e6cfea1..c13d138767 100644 --- a/internal/api/entitlement_handlers_test.go +++ b/internal/api/entitlement_handlers_test.go @@ -120,6 +120,9 @@ func TestBuildEntitlementPayloadWithUsage_CurrentValues(t *testing.T) { if agentLimit.Current != 12 { t.Fatalf("expected agent current 12, got %d", agentLimit.Current) } + if agentLimit.CurrentAvailable == nil || !*agentLimit.CurrentAvailable { + t.Fatalf("expected agent current availability true, got %+v", agentLimit.CurrentAvailable) + } if guestLimit.Current != 44 { t.Fatalf("expected guest current 44, got %d", guestLimit.Current) } @@ -131,6 +134,26 @@ func TestBuildEntitlementPayloadWithUsage_CurrentValues(t *testing.T) { } } +func TestBuildEntitlementPayloadWithUsage_MonitoredSystemUsageUnavailable(t *testing.T) { + status := &license.LicenseStatus{ + Valid: true, + Tier: license.TierPro, + Features: append([]string(nil), license.TierFeatures[license.TierPro]...), + MaxMonitoredSystems: 50, + } + + payload := buildEntitlementPayloadWithUsage(status, "", entitlementUsageSnapshot{}, nil) + if len(payload.Limits) != 1 { + t.Fatalf("expected one monitored-system limit, got %d", len(payload.Limits)) + } + if payload.Limits[0].Current != 0 { + t.Fatalf("expected unresolved current to remain 0, got %d", payload.Limits[0].Current) + } + if payload.Limits[0].CurrentAvailable == nil || *payload.Limits[0].CurrentAvailable { + t.Fatalf("expected unresolved current availability false, got %+v", payload.Limits[0].CurrentAvailable) + } +} + func TestBuildEntitlementPayload_Expired(t *testing.T) { status := &license.LicenseStatus{ Valid: false, diff --git a/internal/api/license_limits_test_helpers_test.go b/internal/api/license_limits_test_helpers_test.go index b017426762..16095fc1c7 100644 --- a/internal/api/license_limits_test_helpers_test.go +++ b/internal/api/license_limits_test_helpers_test.go @@ -2,6 +2,7 @@ package api import ( "context" + "sync" "testing" "time" @@ -12,6 +13,8 @@ type staticLicenseProvider struct { service *pkglicensing.Service } +var testLicenseProviderMu sync.Mutex + func (p *staticLicenseProvider) Service(context.Context) *pkglicensing.Service { return p.service } @@ -19,6 +22,8 @@ func (p *staticLicenseProvider) Service(context.Context) *pkglicensing.Service { func setMaxMonitoredSystemsLicenseForTests(t *testing.T, maxMonitoredSystems int) { t.Helper() + testLicenseProviderMu.Lock() + t.Setenv("PULSE_LICENSE_DEV_MODE", "true") service := pkglicensing.NewService() @@ -36,5 +41,6 @@ func setMaxMonitoredSystemsLicenseForTests(t *testing.T, maxMonitoredSystems int SetLicenseServiceProvider(&staticLicenseProvider{service: service}) t.Cleanup(func() { SetLicenseServiceProvider(nil) + testLicenseProviderMu.Unlock() }) } diff --git a/internal/api/monitored_system_limit_enforcement.go b/internal/api/monitored_system_limit_enforcement.go index a73abfd290..1f226c9b91 100644 --- a/internal/api/monitored_system_limit_enforcement.go +++ b/internal/api/monitored_system_limit_enforcement.go @@ -120,13 +120,38 @@ func maxMonitoredSystemsLimitForContext(ctx context.Context) int { // monitoredSystemCount returns the canonical number of counted top-level // monitored systems. Agent-backed and API-backed views share the same cap. func monitoredSystemCount(monitor *monitoring.Monitor) int { + return monitoredSystemUsage(monitor).count +} + +type monitoredSystemUsageSnapshot struct { + count int + readState unifiedresources.ReadState + available bool +} + +func monitoredSystemUsage(monitor *monitoring.Monitor) monitoredSystemUsageSnapshot { if monitor == nil { - return 0 + return monitoredSystemUsageSnapshot{} } - if rs := monitor.GetUnifiedReadStateOrSnapshot(); rs != nil { - return unifiedresources.MonitoredSystemCount(rs) + + readState := monitor.GetUnifiedReadStateOrSnapshot() + if readState == nil { + return monitoredSystemUsageSnapshot{} } - return 0 + + return monitoredSystemUsageSnapshot{ + count: unifiedresources.MonitoredSystemCount(readState), + readState: readState, + available: true, + } +} + +type monitoredSystemLimitDecision struct { + current int + limit int + additional int + usageAvailable bool + exceeded bool } func legacyConnectionCounts(monitor *monitoring.Monitor) legacyConnectionCountsModel { @@ -145,25 +170,160 @@ func writeMaxMonitoredSystemsLimitExceeded(w http.ResponseWriter, current, limit ) } -func monitoredSystemLimitState( +func writeMonitoredSystemUsageUnavailable(w http.ResponseWriter) { + writeErrorResponse( + w, + http.StatusServiceUnavailable, + "monitored_system_usage_unavailable", + "Unable to verify monitored-system capacity right now", + nil, + ) +} + +func monitoredSystemLimitDecisionFromAdditional( + ctx context.Context, + limit int, + current int, + additional int, +) monitoredSystemLimitDecision { + if limit <= 0 { + return monitoredSystemLimitDecision{ + limit: limit, + additional: additional, + usageAvailable: true, + } + } + + effectiveCurrent := current + deployReservedCount(ctx) + return monitoredSystemLimitDecision{ + current: effectiveCurrent, + limit: limit, + additional: additional, + usageAvailable: true, + exceeded: additional > 0 && effectiveCurrent+additional > limit, + } +} + +func monitoredSystemLimitDecisionForCandidate( ctx context.Context, monitor *monitoring.Monitor, candidate unifiedresources.MonitoredSystemCandidate, -) (current, limit int, exceeded bool) { - limit = maxMonitoredSystemsLimitForContext(ctx) - if limit <= 0 || monitor == nil { - return 0, limit, false +) monitoredSystemLimitDecision { + limit := maxMonitoredSystemsLimitForContext(ctx) + if limit <= 0 { + return monitoredSystemLimitDecision{ + limit: limit, + usageAvailable: true, + } } - rs := monitor.GetUnifiedReadStateOrSnapshot() - if rs != nil && unifiedresources.HasMatchingMonitoredSystem(rs, candidate) { - return 0, limit, false + usage := monitoredSystemUsage(monitor) + if !usage.available { + return monitoredSystemLimitDecision{ + limit: limit, + } } - current = monitoredSystemCount(monitor) - reserved := deployReservedCount(ctx) - effective := current + reserved - return effective, limit, effective+1 > limit + projection := unifiedresources.ProjectMonitoredSystemCandidate(usage.readState, candidate) + return monitoredSystemLimitDecisionFromAdditional(ctx, limit, projection.CurrentCount, projection.AdditionalCount) +} + +func monitoredSystemLimitDecisionForCandidateReplacement( + ctx context.Context, + monitor *monitoring.Monitor, + replacement unifiedresources.MonitoredSystemReplacement, + candidate unifiedresources.MonitoredSystemCandidate, +) monitoredSystemLimitDecision { + limit := maxMonitoredSystemsLimitForContext(ctx) + if limit <= 0 { + return monitoredSystemLimitDecision{ + limit: limit, + usageAvailable: true, + } + } + + usage := monitoredSystemUsage(monitor) + if !usage.available { + return monitoredSystemLimitDecision{ + limit: limit, + } + } + + projection := unifiedresources.ProjectMonitoredSystemCandidateReplacement(usage.readState, replacement, candidate) + return monitoredSystemLimitDecisionFromAdditional(ctx, limit, projection.CurrentCount, projection.AdditionalCount) +} + +func monitoredSystemLimitDecisionForRecords( + ctx context.Context, + monitor *monitoring.Monitor, + recordsBySource map[unifiedresources.DataSource][]unifiedresources.IngestRecord, +) monitoredSystemLimitDecision { + limit := maxMonitoredSystemsLimitForContext(ctx) + if limit <= 0 { + return monitoredSystemLimitDecision{ + limit: limit, + usageAvailable: true, + } + } + + usage := monitoredSystemUsage(monitor) + if !usage.available { + return monitoredSystemLimitDecision{ + limit: limit, + } + } + + projection := unifiedresources.ProjectMonitoredSystemRecords(usage.readState, recordsBySource) + return monitoredSystemLimitDecisionFromAdditional(ctx, limit, projection.CurrentCount, projection.AdditionalCount) +} + +func monitoredSystemLimitDecisionForRecordsReplacement( + ctx context.Context, + monitor *monitoring.Monitor, + replacement unifiedresources.MonitoredSystemReplacement, + recordsBySource map[unifiedresources.DataSource][]unifiedresources.IngestRecord, +) monitoredSystemLimitDecision { + limit := maxMonitoredSystemsLimitForContext(ctx) + if limit <= 0 { + return monitoredSystemLimitDecision{ + limit: limit, + usageAvailable: true, + } + } + + usage := monitoredSystemUsage(monitor) + if !usage.available { + return monitoredSystemLimitDecision{ + limit: limit, + } + } + + projection := unifiedresources.ProjectMonitoredSystemRecordsReplacement(usage.readState, replacement, recordsBySource) + return monitoredSystemLimitDecisionFromAdditional(ctx, limit, projection.CurrentCount, projection.AdditionalCount) +} + +func monitoredSystemLimitDecisionForAdditionalSlots( + ctx context.Context, + monitor *monitoring.Monitor, + additional int, +) monitoredSystemLimitDecision { + limit := maxMonitoredSystemsLimitForContext(ctx) + if limit <= 0 { + return monitoredSystemLimitDecision{ + limit: limit, + additional: additional, + usageAvailable: true, + } + } + + usage := monitoredSystemUsage(monitor) + if !usage.available { + return monitoredSystemLimitDecision{ + limit: limit, + } + } + + return monitoredSystemLimitDecisionFromAdditional(ctx, limit, usage.count, additional) } func enforceMonitoredSystemLimitForConfigRegistration( @@ -173,13 +333,38 @@ func enforceMonitoredSystemLimitForConfigRegistration( monitor *monitoring.Monitor, candidate unifiedresources.MonitoredSystemCandidate, ) bool { - current, limit, exceeded := monitoredSystemLimitState(ctx, monitor, candidate) - if !exceeded { + decision := monitoredSystemLimitDecisionForCandidate(ctx, monitor, candidate) + if !decision.usageAvailable { + writeMonitoredSystemUsageUnavailable(w) + return true + } + if !decision.exceeded { return false } - emitLimitBlockedEvent(ctx, current, limit) - writeMaxMonitoredSystemsLimitExceeded(w, current, limit) + emitLimitBlockedEvent(ctx, decision.current, decision.limit) + writeMaxMonitoredSystemsLimitExceeded(w, decision.current, decision.limit) + return true +} + +func enforceMonitoredSystemLimitForConfigReplacement( + w http.ResponseWriter, + ctx context.Context, + monitor *monitoring.Monitor, + replacement unifiedresources.MonitoredSystemReplacement, + candidate unifiedresources.MonitoredSystemCandidate, +) bool { + decision := monitoredSystemLimitDecisionForCandidateReplacement(ctx, monitor, replacement, candidate) + if !decision.usageAvailable { + writeMonitoredSystemUsageUnavailable(w) + return true + } + if !decision.exceeded { + return false + } + + emitLimitBlockedEvent(ctx, decision.current, decision.limit) + writeMaxMonitoredSystemsLimitExceeded(w, decision.current, decision.limit) return true } @@ -192,17 +377,12 @@ func enforceMonitoredSystemLimitForHostReport( report agentshost.Report, tokenRecord *config.APITokenRecord, ) bool { - if monitor == nil { - return false - } - - hosts := monitor.GetLiveHostsSnapshot() - // Updates to an already-known system are always permitted. - if hostReportTargetsExistingHost(hosts, report, tokenRecord) { + if monitor != nil && hostReportTargetsExistingHost(monitor.GetLiveHostsSnapshot(), report, tokenRecord) { return false } candidate := unifiedresources.MonitoredSystemCandidate{ + Source: unifiedresources.SourceAgent, Type: unifiedresources.ResourceTypeAgent, Name: report.Host.DisplayName, Hostname: report.Host.Hostname, @@ -211,13 +391,17 @@ func enforceMonitoredSystemLimitForHostReport( MachineID: report.Host.MachineID, ResourceID: report.Host.ID, } - current, limit, exceeded := monitoredSystemLimitState(ctx, monitor, candidate) - if !exceeded { + decision := monitoredSystemLimitDecisionForCandidate(ctx, monitor, candidate) + if !decision.usageAvailable { + writeMonitoredSystemUsageUnavailable(w) + return true + } + if !decision.exceeded { return false } - emitLimitBlockedEvent(ctx, current, limit) - writeMaxMonitoredSystemsLimitExceeded(w, current, limit) + emitLimitBlockedEvent(ctx, decision.current, decision.limit) + writeMaxMonitoredSystemsLimitExceeded(w, decision.current, decision.limit) return true } @@ -270,19 +454,16 @@ func enforceMonitoredSystemLimitForDockerReport( report agentsdocker.Report, tokenRecord *config.APITokenRecord, ) bool { - if monitor == nil { - return false - } - tokenID := "" if tokenRecord != nil { tokenID = tokenRecord.ID } - if dockerReportTargetsExistingHostFromLicensing(monitor.ReadSnapshot(), report, tokenID) { + if monitor != nil && dockerReportTargetsExistingHostFromLicensing(monitor.ReadSnapshot(), report, tokenID) { return false } candidate := unifiedresources.MonitoredSystemCandidate{ + Source: unifiedresources.SourceDocker, Type: unifiedresources.ResourceTypeAgent, Name: report.Host.Name, Hostname: report.Host.Hostname, @@ -290,13 +471,17 @@ func enforceMonitoredSystemLimitForDockerReport( MachineID: report.Host.MachineID, ResourceID: report.AgentKey(), } - current, limit, exceeded := monitoredSystemLimitState(ctx, monitor, candidate) - if !exceeded { + decision := monitoredSystemLimitDecisionForCandidate(ctx, monitor, candidate) + if !decision.usageAvailable { + writeMonitoredSystemUsageUnavailable(w) + return true + } + if !decision.exceeded { return false } - emitLimitBlockedEvent(ctx, current, limit) - writeMaxMonitoredSystemsLimitExceeded(w, current, limit) + emitLimitBlockedEvent(ctx, decision.current, decision.limit) + writeMaxMonitoredSystemsLimitExceeded(w, decision.current, decision.limit) return true } @@ -309,19 +494,16 @@ func enforceMonitoredSystemLimitForKubernetesReport( report agentsk8s.Report, tokenRecord *config.APITokenRecord, ) bool { - if monitor == nil { - return false - } - tokenID := "" if tokenRecord != nil { tokenID = tokenRecord.ID } - if kubernetesReportTargetsExistingClusterFromLicensing(monitor.ReadSnapshot(), report, tokenID) { + if monitor != nil && kubernetesReportTargetsExistingClusterFromLicensing(monitor.ReadSnapshot(), report, tokenID) { return false } candidate := unifiedresources.MonitoredSystemCandidate{ + Source: unifiedresources.SourceK8s, Type: unifiedresources.ResourceTypeK8sCluster, Name: report.Cluster.Name, Hostname: report.Cluster.Name, @@ -329,13 +511,17 @@ func enforceMonitoredSystemLimitForKubernetesReport( AgentID: report.Agent.ID, ResourceID: kubernetesReportIdentifierFromLicensing(report), } - current, limit, exceeded := monitoredSystemLimitState(ctx, monitor, candidate) - if !exceeded { + decision := monitoredSystemLimitDecisionForCandidate(ctx, monitor, candidate) + if !decision.usageAvailable { + writeMonitoredSystemUsageUnavailable(w) + return true + } + if !decision.exceeded { return false } - emitLimitBlockedEvent(ctx, current, limit) - writeMaxMonitoredSystemsLimitExceeded(w, current, limit) + emitLimitBlockedEvent(ctx, decision.current, decision.limit) + writeMaxMonitoredSystemsLimitExceeded(w, decision.current, decision.limit) return true } diff --git a/internal/api/router.go b/internal/api/router.go index c2f34b7d6c..558d057048 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -380,6 +380,7 @@ func (r *Router) setupRoutes() { } r.vmwareHandlers = &VMwareHandlers{ getPersistence: r.configHandlers.getPersistence, + getMonitor: r.configHandlers.getMonitor, getPoller: func(context.Context) *monitoring.VMwarePoller { return r.vmwarePoller }, } recoveryManager := recoverymanager.New(r.multiTenant) diff --git a/internal/api/subscription_entitlements.go b/internal/api/subscription_entitlements.go index b2007a575c..6d97cfc06b 100644 --- a/internal/api/subscription_entitlements.go +++ b/internal/api/subscription_entitlements.go @@ -168,14 +168,22 @@ func (h *LicenseHandlers) entitlementUsageSnapshot(ctx context.Context) entitlem var monitorResolved bool if h.mtMonitor != nil { if monitor, err := h.mtMonitor.GetMonitor(orgID); err == nil && monitor != nil { - usage.MonitoredSystems = int64(monitoredSystemCount(monitor)) - usage.LegacyConnections = legacyConnectionCounts(monitor) - monitorResolved = true + state := monitoredSystemUsage(monitor) + if state.available { + usage.MonitoredSystems = int64(state.count) + usage.MonitoredSystemsAvailable = true + usage.LegacyConnections = legacyConnectionCounts(monitor) + monitorResolved = true + } } } if !monitorResolved && orgID == "default" && h.monitor != nil { - usage.MonitoredSystems = int64(monitoredSystemCount(h.monitor)) - usage.LegacyConnections = legacyConnectionCounts(h.monitor) + state := monitoredSystemUsage(h.monitor) + if state.available { + usage.MonitoredSystems = int64(state.count) + usage.MonitoredSystemsAvailable = true + usage.LegacyConnections = legacyConnectionCounts(h.monitor) + } } // Guest metadata for guest limit tracking. diff --git a/internal/api/truenas_handlers.go b/internal/api/truenas_handlers.go index 5a2e784c3e..e6674db840 100644 --- a/internal/api/truenas_handlers.go +++ b/internal/api/truenas_handlers.go @@ -248,6 +248,9 @@ func (h *TrueNASHandlers) HandleUpdate(w http.ResponseWriter, r *http.Request) { writeErrorResponse(w, http.StatusBadRequest, "validation_error", err.Error(), nil) return } + if h.enforceMonitoredSystemLimitReplacement(w, r, instances[index], instance) { + return + } instances[index] = instance if err := persistence.SaveTrueNASConfig(instances); err != nil { @@ -467,39 +470,66 @@ func (h *TrueNASHandlers) enforceMonitoredSystemLimit( r *http.Request, instance config.TrueNASInstance, ) bool { - if h == nil || h.getMonitor == nil { + var monitor *monitoring.Monitor + if h != nil && h.getMonitor != nil { + monitor = h.getMonitor(r.Context()) + } + + decision := monitoredSystemLimitDecisionForCandidate(r.Context(), monitor, trueNASMonitoredSystemCandidate(instance)) + if !decision.usageAvailable { + writeMonitoredSystemUsageUnavailable(w) + return true + } + if !decision.exceeded { return false } - monitor := h.getMonitor(r.Context()) - limit := maxMonitoredSystemsLimitForContext(r.Context()) - if monitor == nil || limit <= 0 { - return false - } - - if rs := monitor.GetUnifiedReadStateOrSnapshot(); rs != nil { - candidate := unifiedresources.MonitoredSystemCandidate{ - Type: unifiedresources.ResourceTypeAgent, - Name: instance.Name, - Hostname: instance.Host, - HostURL: instance.Host, - } - if unifiedresources.HasMatchingMonitoredSystem(rs, candidate) { - return false - } - } - - current := monitoredSystemCount(monitor) - effective := current + deployReservedCount(r.Context()) - if effective+1 <= limit { - return false - } - - emitLimitBlockedEvent(r.Context(), effective, limit) - writeMaxMonitoredSystemsLimitExceeded(w, effective, limit) + emitLimitBlockedEvent(r.Context(), decision.current, decision.limit) + writeMaxMonitoredSystemsLimitExceeded(w, decision.current, decision.limit) return true } +func (h *TrueNASHandlers) enforceMonitoredSystemLimitReplacement( + w http.ResponseWriter, + r *http.Request, + current config.TrueNASInstance, + next config.TrueNASInstance, +) bool { + var monitor *monitoring.Monitor + if h != nil && h.getMonitor != nil { + monitor = h.getMonitor(r.Context()) + } + + replacementHost := pulseTokenHostCandidate(current.Host) + decision := monitoredSystemLimitDecisionForCandidateReplacement(r.Context(), monitor, unifiedresources.MonitoredSystemReplacement{ + Source: unifiedresources.SourceTrueNAS, + Matches: func(resource unifiedresources.Resource) bool { + return resource.TrueNAS != nil && strings.EqualFold(strings.TrimSpace(resource.TrueNAS.Hostname), replacementHost) + }, + }, trueNASMonitoredSystemCandidate(next)) + if !decision.usageAvailable { + writeMonitoredSystemUsageUnavailable(w) + return true + } + if !decision.exceeded { + return false + } + + emitLimitBlockedEvent(r.Context(), decision.current, decision.limit) + writeMaxMonitoredSystemsLimitExceeded(w, decision.current, decision.limit) + return true +} + +func trueNASMonitoredSystemCandidate(instance config.TrueNASInstance) unifiedresources.MonitoredSystemCandidate { + return unifiedresources.MonitoredSystemCandidate{ + Source: unifiedresources.SourceTrueNAS, + Type: unifiedresources.ResourceTypeAgent, + Name: instance.Name, + Hostname: pulseTokenHostCandidate(instance.Host), + HostURL: instance.Host, + } +} + func trueNASConnectionIDFromPath(path string) (string, bool) { if !strings.HasPrefix(path, trueNASConnectionsPathPrefix) { return "", false diff --git a/internal/api/truenas_handlers_test.go b/internal/api/truenas_handlers_test.go index 55b9dbe0b8..4e676ecd94 100644 --- a/internal/api/truenas_handlers_test.go +++ b/internal/api/truenas_handlers_test.go @@ -143,7 +143,7 @@ func TestTrueNASHandlers_HandleAdd_BlocksNewCountedSystemAtLimit(t *testing.T) { }, }, }) - monitor.SetResourceStore(unifiedresources.NewMonitorAdapter(registry)) + setUnexportedField(t, monitor, "resourceStore", monitoring.ResourceStoreInterface(unifiedresources.NewMonitorAdapter(registry))) if err := persistence.SaveTrueNASConfig([]config.TrueNASInstance{ {ID: "existing", Host: "nas-a.local", APIKey: "a"}, }); err != nil { @@ -172,6 +172,58 @@ func TestTrueNASHandlers_HandleAdd_BlocksNewCountedSystemAtLimit(t *testing.T) { } } +func TestTrueNASHandlers_HandleAdd_AllowsCanonicalOverlapAtLimit(t *testing.T) { + setTrueNASFeatureForTest(t, true) + setMockModeForTrueNASTest(t, false) + setMaxMonitoredSystemsLicenseForTests(t, 1) + + handler, persistence, monitor := newTrueNASHandlersForTest(t, nil) + registry := unifiedresources.NewRegistry(nil) + registry.IngestRecords(unifiedresources.SourceAgent, []unifiedresources.IngestRecord{ + { + SourceID: "host-1", + Resource: unifiedresources.Resource{ + ID: "host-1", + Type: unifiedresources.ResourceTypeAgent, + Name: "archive.local", + Status: unifiedresources.StatusOnline, + Agent: &unifiedresources.AgentData{ + AgentID: "agent-1", + Hostname: "archive.local", + MachineID: "machine-1", + }, + Identity: unifiedresources.ResourceIdentity{ + MachineID: "machine-1", + Hostnames: []string{"archive.local"}, + }, + }, + }, + }) + setUnexportedField(t, monitor, "resourceStore", monitoring.ResourceStoreInterface(unifiedresources.NewMonitorAdapter(registry))) + + body := marshalTrueNASRequest(t, map[string]any{ + "name": "archive", + "host": "archive.local", + "apiKey": "super-secret", + }) + + req := httptest.NewRequest(http.MethodPost, "/api/truenas/connections", bytes.NewReader(body)) + rec := httptest.NewRecorder() + handler.HandleAdd(rec, req) + + if rec.Code != http.StatusCreated { + t.Fatalf("expected overlapping TrueNAS add to be allowed at limit, got %d: %s", rec.Code, rec.Body.String()) + } + + stored, err := persistence.LoadTrueNASConfig() + if err != nil { + t.Fatalf("load persisted config: %v", err) + } + if len(stored) != 1 { + t.Fatalf("expected overlapping TrueNAS add to persist, got %d instances", len(stored)) + } +} + func TestTrueNASHandlers_HandleList_RedactsSensitiveFields(t *testing.T) { setTrueNASFeatureForTest(t, true) @@ -439,6 +491,90 @@ func TestTrueNASHandlers_HandleUpdate_PreservesMaskedSecretsAndReplacesFields(t } } +func TestTrueNASHandlers_HandleUpdate_BlocksProjectedNetNewSystemAtLimit(t *testing.T) { + setTrueNASFeatureForTest(t, true) + setMockModeForTrueNASTest(t, false) + setMaxMonitoredSystemsLicenseForTests(t, 1) + + handler, persistence, monitor := newTrueNASHandlersForTest(t, nil) + if err := persistence.SaveTrueNASConfig([]config.TrueNASInstance{ + { + ID: "alpha", + Name: "archive", + Host: "archive.local", + APIKey: "super-secret", + UseHTTPS: true, + Enabled: true, + }, + }); err != nil { + t.Fatalf("seed truenas config: %v", err) + } + + registry := unifiedresources.NewRegistry(nil) + registry.IngestRecords(unifiedresources.SourceAgent, []unifiedresources.IngestRecord{ + { + SourceID: "host-1", + Resource: unifiedresources.Resource{ + ID: "host-1", + Type: unifiedresources.ResourceTypeAgent, + Name: "archive.local", + Status: unifiedresources.StatusOnline, + Agent: &unifiedresources.AgentData{ + AgentID: "agent-1", + Hostname: "archive.local", + MachineID: "machine-1", + }, + Identity: unifiedresources.ResourceIdentity{ + MachineID: "machine-1", + Hostnames: []string{"archive.local"}, + }, + }, + }, + }) + registry.IngestRecords(unifiedresources.SourceTrueNAS, []unifiedresources.IngestRecord{ + { + SourceID: "system:archive.local", + Resource: unifiedresources.Resource{ + ID: "truenas-1", + Type: unifiedresources.ResourceTypeAgent, + Name: "archive", + Status: unifiedresources.StatusOnline, + TrueNAS: &unifiedresources.TrueNASData{ + Hostname: "archive.local", + }, + }, + Identity: unifiedresources.ResourceIdentity{ + MachineID: "machine-1", + Hostnames: []string{"archive.local"}, + }, + }, + }) + setUnexportedField(t, monitor, "resourceStore", monitoring.ResourceStoreInterface(unifiedresources.NewMonitorAdapter(registry))) + + body := marshalTrueNASRequest(t, map[string]any{ + "name": "archive", + "host": "backup.local", + "apiKey": "********", + "useHttps": true, + "enabled": true, + }) + req := httptest.NewRequest(http.MethodPut, "/api/truenas/connections/alpha", bytes.NewReader(body)) + rec := httptest.NewRecorder() + handler.HandleUpdate(rec, req) + + if rec.Code != http.StatusPaymentRequired { + t.Fatalf("expected 402 when update would add a new monitored system, got %d: %s", rec.Code, rec.Body.String()) + } + + stored, err := persistence.LoadTrueNASConfig() + if err != nil { + t.Fatalf("load persisted config: %v", err) + } + if stored[0].Host != "archive.local" { + t.Fatalf("expected blocked update to preserve original host, got %+v", stored[0]) + } +} + func TestTrueNASHandlers_HandleUpdate_UnknownID(t *testing.T) { setTrueNASFeatureForTest(t, true) setMockModeForTrueNASTest(t, false) @@ -686,7 +822,7 @@ func newTrueNASHandlersForTest(t *testing.T, cfg *config.Config) (*TrueNASHandle } persistence := config.NewConfigPersistence(cfg.DataPath) - mon := &monitoring.Monitor{} + mon, _, _ := newTestMonitor(t) handler := &TrueNASHandlers{ getPersistence: func(context.Context) *config.ConfigPersistence { return persistence }, diff --git a/internal/api/vmware_handlers.go b/internal/api/vmware_handlers.go index 72a7c28397..4ab5d7181f 100644 --- a/internal/api/vmware_handlers.go +++ b/internal/api/vmware_handlers.go @@ -14,6 +14,7 @@ import ( "github.com/rcourtman/pulse-go-rewrite/internal/config" "github.com/rcourtman/pulse-go-rewrite/internal/mock" "github.com/rcourtman/pulse-go-rewrite/internal/monitoring" + "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" "github.com/rcourtman/pulse-go-rewrite/internal/vmware" ) @@ -22,8 +23,10 @@ const vmwareConnectionsPathPrefix = "/api/vmware/connections/" // VMwareHandlers manages VMware vCenter connection CRUD and connectivity checks. type VMwareHandlers struct { getPersistence func(ctx context.Context) *config.ConfigPersistence + getMonitor func(ctx context.Context) *monitoring.Monitor getPoller func(ctx context.Context) *monitoring.VMwarePoller newClient func(vmware.ClientConfig) (vmwareClient, error) + previewRecords func(ctx context.Context, instance config.VMwareVCenterInstance) ([]unifiedresources.IngestRecord, error) statusMu sync.RWMutex statuses map[string]vmwareConnectionRuntimeStatus @@ -79,6 +82,9 @@ func (h *VMwareHandlers) HandleAdd(w http.ResponseWriter, r *http.Request) { if persistence == nil { return } + if h.enforceMonitoredSystemLimit(w, r, instance) { + return + } instances, err := persistence.LoadVMwareConfig() if err != nil { @@ -256,6 +262,9 @@ func (h *VMwareHandlers) HandleUpdate(w http.ResponseWriter, r *http.Request) { writeErrorResponse(w, http.StatusBadRequest, "validation_error", err.Error(), nil) return } + if h.enforceMonitoredSystemLimitReplacement(w, r, instances[index], instance) { + return + } instances[index] = instance if err := persistence.SaveVMwareConfig(instances); err != nil { @@ -502,6 +511,132 @@ func (h *VMwareHandlers) pollerForRequest(ctx context.Context) *monitoring.VMwar return h.getPoller(ctx) } +func (h *VMwareHandlers) monitorForRequest(ctx context.Context) *monitoring.Monitor { + if h == nil || h.getMonitor == nil { + return nil + } + return h.getMonitor(ctx) +} + +func (h *VMwareHandlers) enforceMonitoredSystemLimit( + w http.ResponseWriter, + r *http.Request, + instance config.VMwareVCenterInstance, +) bool { + if maxMonitoredSystemsLimitForContext(r.Context()) <= 0 { + return false + } + + monitor := h.monitorForRequest(r.Context()) + if monitor == nil { + writeMonitoredSystemUsageUnavailable(w) + return true + } + + records, invalidConfig, err := h.previewMonitoredSystemRecords(r.Context(), instance) + if err != nil { + h.writeConnectionFailure(w, invalidConfig, err) + return true + } + + decision := monitoredSystemLimitDecisionForRecords(r.Context(), monitor, map[unifiedresources.DataSource][]unifiedresources.IngestRecord{ + unifiedresources.SourceVMware: records, + }) + if !decision.usageAvailable { + writeMonitoredSystemUsageUnavailable(w) + return true + } + if !decision.exceeded { + return false + } + + emitLimitBlockedEvent(r.Context(), decision.current, decision.limit) + writeMaxMonitoredSystemsLimitExceeded(w, decision.current, decision.limit) + return true +} + +func (h *VMwareHandlers) enforceMonitoredSystemLimitReplacement( + w http.ResponseWriter, + r *http.Request, + current config.VMwareVCenterInstance, + next config.VMwareVCenterInstance, +) bool { + limit := maxMonitoredSystemsLimitForContext(r.Context()) + if limit <= 0 { + return false + } + + monitor := h.monitorForRequest(r.Context()) + if monitor == nil { + writeMonitoredSystemUsageUnavailable(w) + return true + } + + records, invalidConfig, err := h.previewMonitoredSystemRecords(r.Context(), next) + if err != nil { + h.writeConnectionFailure(w, invalidConfig, err) + return true + } + + replacementID := strings.TrimSpace(current.ID) + decision := monitoredSystemLimitDecisionForRecordsReplacement(r.Context(), monitor, unifiedresources.MonitoredSystemReplacement{ + Source: unifiedresources.SourceVMware, + Matches: func(resource unifiedresources.Resource) bool { + return resource.VMware != nil && strings.TrimSpace(resource.VMware.ConnectionID) == replacementID + }, + }, map[unifiedresources.DataSource][]unifiedresources.IngestRecord{ + unifiedresources.SourceVMware: records, + }) + if !decision.usageAvailable { + writeMonitoredSystemUsageUnavailable(w) + return true + } + if !decision.exceeded { + return false + } + + emitLimitBlockedEvent(r.Context(), decision.current, decision.limit) + writeMaxMonitoredSystemsLimitExceeded(w, decision.current, decision.limit) + return true +} + +func (h *VMwareHandlers) previewMonitoredSystemRecords( + ctx context.Context, + instance config.VMwareVCenterInstance, +) ([]unifiedresources.IngestRecord, bool, error) { + if h != nil && h.previewRecords != nil { + records, err := h.previewRecords(ctx, instance) + return records, false, err + } + + client, err := vmware.NewClient(vmware.ClientConfig{ + Host: instance.Host, + Port: instance.Port, + Username: instance.Username, + Password: instance.Password, + InsecureSkipVerify: instance.InsecureSkipVerify, + Timeout: 20 * time.Second, + }) + if err != nil { + return nil, true, err + } + + provider := vmware.NewAPIProvider(vmware.ProviderMetadata{ + ConnectionID: strings.TrimSpace(instance.ID), + ConnectionName: strings.TrimSpace(instance.Name), + VCenterHost: strings.TrimSpace(instance.Host), + }, client) + defer provider.Close() + + refreshCtx, cancel := context.WithTimeout(ctx, 20*time.Second) + defer cancel() + + if err := provider.Refresh(refreshCtx); err != nil { + return nil, false, err + } + return provider.Records(), false, nil +} + func (h *VMwareHandlers) runtimeStatus(connectionID string) vmwareConnectionRuntimeStatus { if h == nil { return vmwareConnectionRuntimeStatus{} diff --git a/internal/api/vmware_handlers_test.go b/internal/api/vmware_handlers_test.go index 2e3d95ff04..4ea2256623 100644 --- a/internal/api/vmware_handlers_test.go +++ b/internal/api/vmware_handlers_test.go @@ -14,6 +14,7 @@ import ( "github.com/rcourtman/pulse-go-rewrite/internal/config" "github.com/rcourtman/pulse-go-rewrite/internal/mock" "github.com/rcourtman/pulse-go-rewrite/internal/monitoring" + "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" "github.com/rcourtman/pulse-go-rewrite/internal/vmware" ) @@ -118,6 +119,166 @@ func TestVMwareHandlers_HandleAdd_ValidationAndFeatureGate(t *testing.T) { }) } +func TestVMwareHandlers_HandleAdd_BlocksProjectedNetNewSystemsAtLimit(t *testing.T) { + setVMwareFeatureForTest(t, true) + setMockModeForVMwareTest(t, false) + setMaxMonitoredSystemsLicenseForTests(t, 1) + + handler, persistence := newVMwareHandlersForTest(t) + registry := unifiedresources.NewRegistry(nil) + registry.IngestRecords(unifiedresources.SourceAgent, []unifiedresources.IngestRecord{ + { + SourceID: "host-1", + Resource: unifiedresources.Resource{ + ID: "host-1", + Type: unifiedresources.ResourceTypeAgent, + Name: "existing.local", + Status: unifiedresources.StatusOnline, + Agent: &unifiedresources.AgentData{ + AgentID: "agent-1", + Hostname: "existing.local", + MachineID: "machine-1", + }, + Identity: unifiedresources.ResourceIdentity{ + MachineID: "machine-1", + Hostnames: []string{"existing.local"}, + }, + }, + }, + }) + monitor := &monitoring.Monitor{} + monitor.SetResourceStore(unifiedresources.NewMonitorAdapter(registry)) + handler.getMonitor = func(context.Context) *monitoring.Monitor { return monitor } + handler.previewRecords = func(context.Context, config.VMwareVCenterInstance) ([]unifiedresources.IngestRecord, error) { + return []unifiedresources.IngestRecord{ + { + SourceID: "vc-1-host-1", + Resource: unifiedresources.Resource{ + Type: unifiedresources.ResourceTypeAgent, + Name: "esxi-02.lab.local", + Status: unifiedresources.StatusOnline, + VMware: &unifiedresources.VMwareData{ + ConnectionID: "vc-1", + ManagedObjectID: "host-1", + EntityType: "host", + HostUUID: "vmware-host-2", + }, + }, + Identity: unifiedresources.ResourceIdentity{ + DMIUUID: "vmware-host-2", + Hostnames: []string{"esxi-02.lab.local"}, + }, + }, + }, nil + } + + body := marshalVMwareRequest(t, map[string]any{ + "id": "vc-1", + "name": "lab-vcenter", + "host": "vcsa.lab.local", + "port": 443, + "username": "administrator@vsphere.local", + "password": "super-secret", + "enabled": true, + }) + + req := httptest.NewRequest(http.MethodPost, "/api/vmware/connections", bytes.NewReader(body)) + rec := httptest.NewRecorder() + handler.HandleAdd(rec, req) + + if rec.Code != http.StatusPaymentRequired { + t.Fatalf("expected 402 once projected VMware inventory exceeds the cap, got %d: %s", rec.Code, rec.Body.String()) + } + + stored, err := persistence.LoadVMwareConfig() + if err != nil { + t.Fatalf("load vmware config: %v", err) + } + if len(stored) != 0 { + t.Fatalf("expected blocked VMware add not to persist, got %d connections", len(stored)) + } +} + +func TestVMwareHandlers_HandleAdd_AllowsCanonicalOverlapAtLimit(t *testing.T) { + setVMwareFeatureForTest(t, true) + setMockModeForVMwareTest(t, false) + setMaxMonitoredSystemsLicenseForTests(t, 1) + + handler, persistence := newVMwareHandlersForTest(t) + registry := unifiedresources.NewRegistry(nil) + registry.IngestRecords(unifiedresources.SourceAgent, []unifiedresources.IngestRecord{ + { + SourceID: "host-1", + Resource: unifiedresources.Resource{ + ID: "host-1", + Type: unifiedresources.ResourceTypeAgent, + Name: "esxi-01.lab.local", + Status: unifiedresources.StatusOnline, + Agent: &unifiedresources.AgentData{ + AgentID: "agent-1", + Hostname: "esxi-01.lab.local", + MachineID: "machine-1", + }, + Identity: unifiedresources.ResourceIdentity{ + MachineID: "machine-1", + Hostnames: []string{"esxi-01.lab.local"}, + }, + }, + }, + }) + monitor := &monitoring.Monitor{} + monitor.SetResourceStore(unifiedresources.NewMonitorAdapter(registry)) + handler.getMonitor = func(context.Context) *monitoring.Monitor { return monitor } + handler.previewRecords = func(context.Context, config.VMwareVCenterInstance) ([]unifiedresources.IngestRecord, error) { + return []unifiedresources.IngestRecord{ + { + SourceID: "vc-1-host-1", + Resource: unifiedresources.Resource{ + Type: unifiedresources.ResourceTypeAgent, + Name: "esxi-01.lab.local", + Status: unifiedresources.StatusOnline, + VMware: &unifiedresources.VMwareData{ + ConnectionID: "vc-1", + ManagedObjectID: "host-1", + EntityType: "host", + HostUUID: "vmware-host-1", + }, + }, + Identity: unifiedresources.ResourceIdentity{ + DMIUUID: "vmware-host-1", + Hostnames: []string{"esxi-01.lab.local"}, + }, + }, + }, nil + } + + body := marshalVMwareRequest(t, map[string]any{ + "id": "vc-1", + "name": "lab-vcenter", + "host": "vcsa.lab.local", + "port": 443, + "username": "administrator@vsphere.local", + "password": "super-secret", + "enabled": true, + }) + + req := httptest.NewRequest(http.MethodPost, "/api/vmware/connections", bytes.NewReader(body)) + rec := httptest.NewRecorder() + handler.HandleAdd(rec, req) + + if rec.Code != http.StatusCreated { + t.Fatalf("expected overlapping VMware add to be allowed at limit, got %d: %s", rec.Code, rec.Body.String()) + } + + stored, err := persistence.LoadVMwareConfig() + if err != nil { + t.Fatalf("load vmware config: %v", err) + } + if len(stored) != 1 { + t.Fatalf("expected allowed VMware add to persist, got %d connections", len(stored)) + } +} + func TestVMwareHandlers_HandleList_RedactsSensitiveFieldsAndIncludesRuntimeSummary(t *testing.T) { setVMwareFeatureForTest(t, true) @@ -380,6 +541,123 @@ func TestVMwareHandlers_HandleUpdate_PreservesMaskedSecretsAndReplacesFields(t * } } +func TestVMwareHandlers_HandleUpdate_BlocksProjectedNetNewSystemsAtLimit(t *testing.T) { + setVMwareFeatureForTest(t, true) + setMockModeForVMwareTest(t, false) + setMaxMonitoredSystemsLicenseForTests(t, 1) + + handler, persistence := newVMwareHandlersForTest(t) + if err := persistence.SaveVMwareConfig([]config.VMwareVCenterInstance{{ + ID: "alpha", + Name: "vc-a", + Host: "vc-a.lab.local", + Port: 443, + Username: "administrator@vsphere.local", + Password: "super-secret", + InsecureSkipVerify: true, + Enabled: true, + }}); err != nil { + t.Fatalf("seed vmware config: %v", err) + } + + monitor := &monitoring.Monitor{} + registry := unifiedresources.NewRegistry(nil) + registry.IngestRecords(unifiedresources.SourceAgent, []unifiedresources.IngestRecord{ + { + SourceID: "host-1", + Resource: unifiedresources.Resource{ + ID: "host-1", + Type: unifiedresources.ResourceTypeAgent, + Name: "archive.local", + Status: unifiedresources.StatusOnline, + Agent: &unifiedresources.AgentData{ + AgentID: "agent-1", + Hostname: "archive.local", + MachineID: "machine-1", + }, + Identity: unifiedresources.ResourceIdentity{ + MachineID: "machine-1", + Hostnames: []string{"archive.local"}, + }, + }, + }, + }) + registry.IngestRecords(unifiedresources.SourceVMware, []unifiedresources.IngestRecord{ + { + SourceID: "vmware:alpha:host-1", + Resource: unifiedresources.Resource{ + ID: "vmware-host-1", + Type: unifiedresources.ResourceTypeAgent, + Name: "archive.local", + Status: unifiedresources.StatusOnline, + VMware: &unifiedresources.VMwareData{ + ConnectionID: "alpha", + ConnectionName: "vc-a", + VCenterHost: "vc-a.lab.local", + ManagedObjectID: "host-1", + EntityType: "host", + HostUUID: "vmware-host-1", + }, + }, + Identity: unifiedresources.ResourceIdentity{ + MachineID: "machine-1", + Hostnames: []string{"archive.local"}, + }, + }, + }) + monitor.SetResourceStore(unifiedresources.NewMonitorAdapter(registry)) + handler.getMonitor = func(context.Context) *monitoring.Monitor { return monitor } + handler.previewRecords = func(context.Context, config.VMwareVCenterInstance) ([]unifiedresources.IngestRecord, error) { + return []unifiedresources.IngestRecord{ + { + SourceID: "vmware:alpha:host-2", + Resource: unifiedresources.Resource{ + ID: "vmware-host-2", + Type: unifiedresources.ResourceTypeAgent, + Name: "backup.local", + Status: unifiedresources.StatusOnline, + VMware: &unifiedresources.VMwareData{ + ConnectionID: "alpha", + ConnectionName: "vc-a", + VCenterHost: "vc-b.lab.local", + ManagedObjectID: "host-2", + EntityType: "host", + HostUUID: "vmware-host-2", + }, + }, + Identity: unifiedresources.ResourceIdentity{ + Hostnames: []string{"backup.local"}, + }, + }, + }, nil + } + + body := marshalVMwareRequest(t, map[string]any{ + "name": "vc-b", + "host": "vc-b.lab.local", + "port": 443, + "username": "administrator@vsphere.local", + "password": "********", + "insecureSkipVerify": true, + "enabled": true, + }) + req := httptest.NewRequest(http.MethodPut, "/api/vmware/connections/alpha", bytes.NewReader(body)) + rec := httptest.NewRecorder() + handler.HandleUpdate(rec, req) + + if rec.Code != http.StatusPaymentRequired { + t.Fatalf("expected 402 when update would add a new monitored system, got %d: %s", rec.Code, rec.Body.String()) + } + + stored, err := persistence.LoadVMwareConfig() + if err != nil { + t.Fatalf("load persisted config: %v", err) + } + if stored[0].Host != "vc-a.lab.local" { + t.Fatalf("expected blocked update to preserve original host, got %+v", stored[0]) + } +} + func TestVMwareHandlers_HandleTestConnection_SuccessAndFailure(t *testing.T) { setVMwareFeatureForTest(t, true) @@ -674,8 +952,13 @@ func newVMwareHandlersForTest(t *testing.T) (*VMwareHandlers, *config.ConfigPers t.Helper() persistence := config.NewConfigPersistence(t.TempDir()) + monitor, _, _ := newTestMonitor(t) handler := &VMwareHandlers{ getPersistence: func(context.Context) *config.ConfigPersistence { return persistence }, + getMonitor: func(context.Context) *monitoring.Monitor { return monitor }, + previewRecords: func(context.Context, config.VMwareVCenterInstance) ([]unifiedresources.IngestRecord, error) { + return nil, nil + }, } return handler, persistence diff --git a/internal/unifiedresources/monitored_system_projection.go b/internal/unifiedresources/monitored_system_projection.go new file mode 100644 index 0000000000..03fd743357 --- /dev/null +++ b/internal/unifiedresources/monitored_system_projection.go @@ -0,0 +1,341 @@ +package unifiedresources + +import "strings" + +// MonitoredSystemProjection describes how a prospective change affects the +// canonical monitored-system count after shared top-level deduplication. +type MonitoredSystemProjection struct { + CurrentCount int + ProjectedCount int + AdditionalCount int +} + +// MonitoredSystemReplacement describes one existing monitored-system surface to +// replace while preserving any other source ownership already attached to the +// same top-level resource. +type MonitoredSystemReplacement struct { + Source DataSource + Matches func(Resource) bool +} + +// ProjectMonitoredSystemCandidate projects one prospective monitored-system +// candidate through the canonical top-level resolver. +func ProjectMonitoredSystemCandidate( + rs ReadState, + candidate MonitoredSystemCandidate, +) MonitoredSystemProjection { + resource := monitoredSystemCandidateResource(candidate) + if resource == nil { + return projectMonitoredSystemResources(rs, nil, nil, nil) + } + return projectMonitoredSystemResources(rs, []Resource{*resource}, nil, nil) +} + +// ProjectMonitoredSystemCandidateReplacement projects a prospective monitored-system +// candidate while replacing one existing source-owned surface in the canonical +// top-level resolver. +func ProjectMonitoredSystemCandidateReplacement( + rs ReadState, + replacement MonitoredSystemReplacement, + candidate MonitoredSystemCandidate, +) MonitoredSystemProjection { + resource := monitoredSystemCandidateResource(candidate) + if resource == nil { + return projectMonitoredSystemResources(rs, nil, nil, &replacement) + } + return projectMonitoredSystemResources(rs, []Resource{*resource}, nil, &replacement) +} + +// ProjectMonitoredSystemRecords projects source-native records through the +// canonical top-level resolver before they are admitted into runtime state. +func ProjectMonitoredSystemRecords( + rs ReadState, + recordsBySource map[DataSource][]IngestRecord, +) MonitoredSystemProjection { + return projectMonitoredSystemResources(rs, nil, recordsBySource, nil) +} + +// ProjectMonitoredSystemRecordsReplacement projects source-native records while +// replacing one existing source-owned surface in the canonical top-level +// resolver. +func ProjectMonitoredSystemRecordsReplacement( + rs ReadState, + replacement MonitoredSystemReplacement, + recordsBySource map[DataSource][]IngestRecord, +) MonitoredSystemProjection { + return projectMonitoredSystemResources(rs, nil, recordsBySource, &replacement) +} + +func projectMonitoredSystemResources( + rs ReadState, + additionalResources []Resource, + recordsBySource map[DataSource][]IngestRecord, + replacement *MonitoredSystemReplacement, +) MonitoredSystemProjection { + currentRoots := monitoredSystemRootResources(rs) + currentCount := ResolveTopLevelSystems(currentRoots).Count() + + projectedRoots := monitoredSystemRootsWithReplacement(currentRoots, replacement) + projectedRoots = append(projectedRoots, additionalResources...) + projectedRoots = append(projectedRoots, monitoredSystemRootsFromRecords(recordsBySource)...) + + projectedCount := ResolveTopLevelSystems(projectedRoots).Count() + additionalCount := projectedCount - currentCount + if additionalCount < 0 { + additionalCount = 0 + } + + return MonitoredSystemProjection{ + CurrentCount: currentCount, + ProjectedCount: projectedCount, + AdditionalCount: additionalCount, + } +} + +func monitoredSystemRootsWithReplacement( + currentRoots []Resource, + replacement *MonitoredSystemReplacement, +) []Resource { + if replacement == nil || replacement.Matches == nil { + return append([]Resource(nil), currentRoots...) + } + + projected := make([]Resource, 0, len(currentRoots)) + for _, root := range currentRoots { + if !replacement.Matches(root) { + projected = append(projected, root) + continue + } + + stripped, ok := stripMonitoredSystemSource(root, replacement.Source) + if ok { + projected = append(projected, stripped) + } + } + return projected +} + +func stripMonitoredSystemSource(resource Resource, source DataSource) (Resource, bool) { + stripped := cloneResource(&resource) + stripped.Sources = removeMonitoredSystemSource(stripped.Sources, source) + + switch source { + case SourceAgent: + stripped.Agent = nil + case SourceDocker: + stripped.Docker = nil + case SourceProxmox: + stripped.Proxmox = nil + case SourceTrueNAS: + stripped.TrueNAS = nil + case SourcePBS: + stripped.PBS = nil + case SourcePMG: + stripped.PMG = nil + case SourceK8s: + stripped.Kubernetes = nil + case SourceVMware: + stripped.VMware = nil + } + + if !isMonitoredSystemRootResource(stripped) { + return Resource{}, false + } + return stripped, true +} + +func isMonitoredSystemRootResource(resource Resource) bool { + switch { + case resource.Agent != nil: + return true + case resource.Docker != nil: + return true + case resource.Proxmox != nil: + return true + case resource.TrueNAS != nil: + return true + case resource.PBS != nil: + return true + case resource.PMG != nil: + return true + case resource.Kubernetes != nil: + return true + case resource.VMware != nil: + return true + default: + return false + } +} + +func removeMonitoredSystemSource(sources []DataSource, target DataSource) []DataSource { + if len(sources) == 0 { + return nil + } + filtered := make([]DataSource, 0, len(sources)) + for _, source := range sources { + if source == target { + continue + } + filtered = append(filtered, source) + } + return filtered +} + +func monitoredSystemRootResources(rs ReadState) []Resource { + roots := monitoredSystemRoots(rs) + resources := make([]Resource, 0, len(roots)) + for _, resource := range roots { + if resource == nil { + continue + } + resources = append(resources, cloneResource(resource)) + } + return resources +} + +func monitoredSystemRootsFromRecords(recordsBySource map[DataSource][]IngestRecord) []Resource { + if len(recordsBySource) == 0 { + return nil + } + + registry := NewRegistry(nil) + for source, records := range recordsBySource { + if strings.TrimSpace(string(source)) == "" || len(records) == 0 { + continue + } + registry.IngestRecords(source, records) + } + return monitoredSystemRootResources(registry) +} + +func monitoredSystemCandidateResource(candidate MonitoredSystemCandidate) *Resource { + source := normalizeMonitoredSystemCandidateSource(candidate) + + host := firstTrimmed(candidate.Hostname, extractHostname(candidate.HostURL)) + name := firstTrimmed(candidate.Name, host, candidate.ResourceID, candidate.AgentID) + if name == "" { + return nil + } + + resourceID := monitoredSystemCandidateResourceID(candidate, source, host, name) + if resourceID == "" { + return nil + } + + resource := &Resource{ + ID: resourceID, + Name: name, + Status: StatusOnline, + } + + identity := ResourceIdentity{ + MachineID: strings.TrimSpace(candidate.MachineID), + Hostnames: uniqueTrimmed(host, extractHostname(candidate.HostURL)), + } + + switch source { + case SourceDocker: + resource.Type = ResourceTypeAgent + resource.Docker = &DockerData{ + HostSourceID: strings.TrimSpace(candidate.ResourceID), + AgentID: strings.TrimSpace(candidate.AgentID), + Hostname: host, + MachineID: strings.TrimSpace(candidate.MachineID), + } + case SourceProxmox: + resource.Type = ResourceTypeAgent + resource.Proxmox = &ProxmoxData{ + SourceID: strings.TrimSpace(candidate.ResourceID), + NodeName: name, + HostURL: strings.TrimSpace(candidate.HostURL), + } + case SourceTrueNAS: + resource.Type = ResourceTypeAgent + resource.TrueNAS = &TrueNASData{ + Hostname: host, + } + case SourcePBS: + resource.Type = ResourceTypePBS + resource.PBS = &PBSData{ + InstanceID: strings.TrimSpace(candidate.ResourceID), + Hostname: host, + HostURL: strings.TrimSpace(candidate.HostURL), + } + case SourcePMG: + resource.Type = ResourceTypePMG + resource.PMG = &PMGData{ + InstanceID: strings.TrimSpace(candidate.ResourceID), + Hostname: host, + } + case SourceK8s: + clusterName := firstTrimmed(candidate.Name, host, candidate.ResourceID) + resource.Type = ResourceTypeK8sCluster + resource.Name = clusterName + resource.Kubernetes = &K8sData{ + ClusterID: firstTrimmed(candidate.ResourceID, clusterName), + ClusterName: clusterName, + AgentID: strings.TrimSpace(candidate.AgentID), + Server: strings.TrimSpace(candidate.HostURL), + } + identity.Hostnames = uniqueTrimmed(clusterName, host, extractHostname(candidate.HostURL)) + default: + resource.Type = ResourceTypeAgent + resource.Agent = &AgentData{ + AgentID: strings.TrimSpace(candidate.AgentID), + Hostname: host, + MachineID: strings.TrimSpace(candidate.MachineID), + ReportIP: strings.TrimSpace(candidate.HostURL), + } + } + + resource.Identity = identity + return resource +} + +func normalizeMonitoredSystemCandidateSource(candidate MonitoredSystemCandidate) DataSource { + if candidate.Source != "" { + return candidate.Source + } + + switch CanonicalResourceType(candidate.Type) { + case ResourceTypePBS: + return SourcePBS + case ResourceTypePMG: + return SourcePMG + case ResourceTypeK8sCluster: + return SourceK8s + default: + return SourceAgent + } +} + +func monitoredSystemCandidateResourceID( + candidate MonitoredSystemCandidate, + source DataSource, + host string, + name string, +) string { + if sourceID := strings.TrimSpace(candidate.ResourceID); sourceID != "" { + switch source { + case SourcePBS: + return SourceSpecificID(ResourceTypePBS, source, sourceID) + case SourcePMG: + return SourceSpecificID(ResourceTypePMG, source, sourceID) + case SourceK8s: + return SourceSpecificID(ResourceTypeK8sCluster, source, sourceID) + default: + return CanonicalResourceID(sourceID) + } + } + + if machineID := strings.TrimSpace(candidate.MachineID); machineID != "" { + return CanonicalResourceID("candidate:" + string(source) + ":" + machineID) + } + if agentID := strings.TrimSpace(candidate.AgentID); agentID != "" { + return CanonicalResourceID("candidate:" + string(source) + ":" + agentID) + } + if host != "" { + return CanonicalResourceID("candidate:" + string(source) + ":" + strings.ToLower(host)) + } + return CanonicalResourceID("candidate:" + string(source) + ":" + strings.ToLower(strings.TrimSpace(name))) +} diff --git a/internal/unifiedresources/monitored_systems.go b/internal/unifiedresources/monitored_systems.go index 46d60b356b..97877ecb7c 100644 --- a/internal/unifiedresources/monitored_systems.go +++ b/internal/unifiedresources/monitored_systems.go @@ -9,6 +9,7 @@ import ( // MonitoredSystemCandidate describes a prospective top-level monitored system // that may be added through an agent report or API-backed registration. type MonitoredSystemCandidate struct { + Source DataSource Type ResourceType Name string Hostname string @@ -112,7 +113,7 @@ func MonitoredSystems(rs ReadState) []MonitoredSystemRecord { // HasMatchingMonitoredSystem reports whether a prospective monitored system // would dedupe onto an already-counted top-level monitored system. func HasMatchingMonitoredSystem(rs ReadState, candidate MonitoredSystemCandidate) bool { - return resolveMonitoredSystemTopLevelSystems(rs).HasMatchingCandidate(candidate) + return ProjectMonitoredSystemCandidate(rs, candidate).AdditionalCount == 0 } type monitoredSystemGroup struct { diff --git a/internal/unifiedresources/monitored_systems_test.go b/internal/unifiedresources/monitored_systems_test.go index 705cb4963c..f64f1f25a2 100644 --- a/internal/unifiedresources/monitored_systems_test.go +++ b/internal/unifiedresources/monitored_systems_test.go @@ -212,3 +212,176 @@ func TestMonitoredSystemCountExactHostnameFallbackDoesNotMergeEqualPriorityHosts t.Fatalf("MonitoredSystemCount() = %d, want 2", got) } } + +func TestProjectMonitoredSystemCandidateUsesCanonicalPlatformProjection(t *testing.T) { + registry := NewRegistry(nil) + registry.IngestRecords(SourceAgent, []IngestRecord{ + { + SourceID: "host-1", + Resource: Resource{ + ID: "host-1", + Type: ResourceTypeAgent, + Name: "tower.local", + Status: StatusOnline, + Agent: &AgentData{ + AgentID: "agent-1", + Hostname: "tower.local", + MachineID: "machine-1", + }, + Identity: ResourceIdentity{ + MachineID: "machine-1", + Hostnames: []string{"tower.local"}, + }, + }, + }, + }) + + testCases := []struct { + name string + candidate MonitoredSystemCandidate + }{ + { + name: "proxmox node overlapping agent host", + candidate: MonitoredSystemCandidate{ + Source: SourceProxmox, + Type: ResourceTypeAgent, + Name: "tower", + Hostname: "tower.local", + HostURL: "https://tower.local:8006", + }, + }, + { + name: "truenas appliance overlapping agent host", + candidate: MonitoredSystemCandidate{ + Source: SourceTrueNAS, + Type: ResourceTypeAgent, + Name: "tower storage", + Hostname: "tower.local", + HostURL: "https://tower.local", + }, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + projection := ProjectMonitoredSystemCandidate(registry, tc.candidate) + if projection.CurrentCount != 1 { + t.Fatalf("CurrentCount = %d, want 1", projection.CurrentCount) + } + if projection.ProjectedCount != 1 { + t.Fatalf("ProjectedCount = %d, want 1", projection.ProjectedCount) + } + if projection.AdditionalCount != 0 { + t.Fatalf("AdditionalCount = %d, want 0", projection.AdditionalCount) + } + }) + } +} + +func TestProjectMonitoredSystemCandidateReplacementPreservesOverlappingSources(t *testing.T) { + registry := NewRegistry(nil) + registry.IngestRecords(SourceAgent, []IngestRecord{ + { + SourceID: "host-1", + Resource: Resource{ + ID: "host-1", + Type: ResourceTypeAgent, + Name: "archive.local", + Status: StatusOnline, + Agent: &AgentData{ + AgentID: "agent-1", + Hostname: "archive.local", + MachineID: "machine-1", + }, + Identity: ResourceIdentity{ + MachineID: "machine-1", + Hostnames: []string{"archive.local"}, + }, + }, + }, + }) + registry.IngestRecords(SourceTrueNAS, []IngestRecord{ + { + SourceID: "system:archive.local", + Resource: Resource{ + ID: "truenas-1", + Type: ResourceTypeAgent, + Name: "archive", + Status: StatusOnline, + TrueNAS: &TrueNASData{ + Hostname: "archive.local", + }, + }, + Identity: ResourceIdentity{ + MachineID: "machine-1", + Hostnames: []string{"archive.local"}, + }, + }, + }) + + projection := ProjectMonitoredSystemCandidateReplacement(registry, MonitoredSystemReplacement{ + Source: SourceTrueNAS, + Matches: func(resource Resource) bool { + return resource.TrueNAS != nil && resource.TrueNAS.Hostname == "archive.local" + }, + }, MonitoredSystemCandidate{ + Source: SourceTrueNAS, + Type: ResourceTypeAgent, + Name: "backup", + Hostname: "backup.local", + HostURL: "https://backup.local", + }) + + if projection.CurrentCount != 1 { + t.Fatalf("CurrentCount = %d, want 1", projection.CurrentCount) + } + if projection.ProjectedCount != 2 { + t.Fatalf("ProjectedCount = %d, want 2", projection.ProjectedCount) + } + if projection.AdditionalCount != 1 { + t.Fatalf("AdditionalCount = %d, want 1", projection.AdditionalCount) + } +} + +func TestProjectMonitoredSystemCandidateReplacementRemovesStandaloneSource(t *testing.T) { + registry := NewRegistry(nil) + registry.IngestRecords(SourcePBS, []IngestRecord{ + { + SourceID: "pbs-1", + Resource: Resource{ + ID: "pbs-1", + Type: ResourceTypePBS, + Name: "pbs-a", + Status: StatusOnline, + PBS: &PBSData{ + InstanceID: "pbs-a", + Hostname: "pbs-a.local", + HostURL: "https://pbs-a.local:8007", + }, + }, + }, + }) + + projection := ProjectMonitoredSystemCandidateReplacement(registry, MonitoredSystemReplacement{ + Source: SourcePBS, + Matches: func(resource Resource) bool { + return resource.PBS != nil && resource.PBS.InstanceID == "pbs-a" + }, + }, MonitoredSystemCandidate{ + Source: SourcePBS, + Type: ResourceTypePBS, + Name: "pbs-b", + Hostname: "pbs-b.local", + HostURL: "https://pbs-b.local:8007", + }) + + if projection.CurrentCount != 1 { + t.Fatalf("CurrentCount = %d, want 1", projection.CurrentCount) + } + if projection.ProjectedCount != 1 { + t.Fatalf("ProjectedCount = %d, want 1", projection.ProjectedCount) + } + if projection.AdditionalCount != 0 { + t.Fatalf("AdditionalCount = %d, want 0", projection.AdditionalCount) + } +} diff --git a/internal/unifiedresources/registry_test.go b/internal/unifiedresources/registry_test.go index 43c6e9ee18..49f4feecc8 100644 --- a/internal/unifiedresources/registry_test.go +++ b/internal/unifiedresources/registry_test.go @@ -2059,6 +2059,62 @@ func TestRegistryHasMatchingMonitoredSystemUsesCanonicalHostIdentity(t *testing. } } +func TestProjectMonitoredSystemRecords_DedupesVMwareHostAgainstExistingAgentCount(t *testing.T) { + rr := NewRegistry(nil) + now := time.Date(2026, 4, 7, 12, 0, 0, 0, time.UTC) + + rr.IngestRecords(SourceAgent, []IngestRecord{ + { + SourceID: "agent-host-1", + Resource: Resource{ + Type: ResourceTypeAgent, + Name: "esxi-01.lab.local", + Status: StatusOnline, + LastSeen: now, + }, + Identity: ResourceIdentity{ + DMIUUID: "uuid-host-1", + Hostnames: []string{"esxi-01.lab.local"}, + }, + }, + }) + + projection := ProjectMonitoredSystemRecords(rr, map[DataSource][]IngestRecord{ + SourceVMware: { + { + SourceID: "vc-1:host:host-101", + Resource: Resource{ + Type: ResourceTypeAgent, + Name: "esxi-01.lab.local", + Status: StatusOnline, + LastSeen: now, + VMware: &VMwareData{ + ConnectionID: "vc-1", + ConnectionName: "Lab VC", + ManagedObjectID: "host-101", + EntityType: "host", + HostUUID: "uuid-host-1", + }, + }, + Identity: ResourceIdentity{ + DMIUUID: "uuid-host-1", + Hostnames: []string{"esxi-01.lab.local"}, + }, + }, + }, + }) + + if projection.CurrentCount != 1 { + t.Fatalf("CurrentCount = %d, want 1", projection.CurrentCount) + } + if projection.ProjectedCount != 1 { + t.Fatalf("ProjectedCount = %d, want 1", projection.ProjectedCount) + } + if projection.AdditionalCount != 0 { + t.Fatalf("AdditionalCount = %d, want 0", projection.AdditionalCount) + } +} + func TestResourceRegistry_WorkloadsIncludeCanonicalAppContainers(t *testing.T) { rr := NewRegistry(nil) now := time.Date(2026, 3, 29, 12, 0, 0, 0, time.UTC) diff --git a/internal/unifiedresources/top_level_systems.go b/internal/unifiedresources/top_level_systems.go index fcdff1766a..212d2fe9a0 100644 --- a/internal/unifiedresources/top_level_systems.go +++ b/internal/unifiedresources/top_level_systems.go @@ -215,32 +215,31 @@ func (r TopLevelSystemResolver) records() []MonitoredSystemRecord { } func (r TopLevelSystemResolver) HasMatchingCandidate(candidate MonitoredSystemCandidate) bool { - strongIDs := monitoredSystemCandidateStrongIDs(candidate) - for _, group := range r.groups { - if topLevelSystemSetsOverlap(group.strongIDs, strongIDs) { - return true - } + resource := monitoredSystemCandidateResource(candidate) + if resource == nil { + return false } - - if !monitoredSystemCandidateAllowsHostAttachment(candidate) { + if !monitoredSystemCandidateAllowsHostAttachment(candidate) && + len(monitoredSystemCandidateStrongIDs(candidate)) == 0 { return false } - groupByID := make(map[string]topLevelSystemResolvedGroup, len(r.groups)) - hostOwners := make(map[string]map[string]struct{}) - ipOwners := make(map[string]map[string]struct{}) + resources := r.resources() + resources = append(resources, *resource) + return ResolveTopLevelSystems(resources).Count() == r.Count() +} + +func (r TopLevelSystemResolver) resources() []Resource { + resources := make([]Resource, 0, len(r.resourceToGroup)) for _, group := range r.groups { - groupByID[group.id] = group - for host := range group.exactHosts { - addTopLevelSystemOwner(hostOwners, host, group.id) - } - for ip := range group.exactIPs { - addTopLevelSystemOwner(ipOwners, ip, group.id) + for _, resource := range group.resources { + if resource == nil { + continue + } + resources = append(resources, cloneResource(resource)) } } - - targetIDs := candidateExactTargetGroups(candidate, hostOwners, ipOwners, groupByID, monitoredSystemCandidatePriority(candidate)) - return len(targetIDs) == 1 + return resources } func buildTopLevelSystemResolvedGroups( @@ -374,6 +373,11 @@ func topLevelSystemStrongIDs(resource Resource) map[string]struct{} { ids["k8s:"+clusterID] = struct{}{} } } + if resource.VMware != nil { + if hostUUID := strings.TrimSpace(resource.VMware.HostUUID); hostUUID != "" { + ids["vmware-host:"+hostUUID] = struct{}{} + } + } return ids } @@ -386,6 +390,7 @@ func topLevelSystemExactHosts(resource Resource) map[string]struct{} { canonicalPBSHostname(resource), canonicalPMGHostname(resource), canonicalTrueNASHostname(resource), + canonicalVMwareHostname(resource), canonicalProxmoxNodeName(resource), extractHostname(firstTrimmed(topLevelSystemProxmoxHostURL(resource))), extractHostname(firstTrimmed(topLevelSystemPBSHostURL(resource))), @@ -646,16 +651,18 @@ func topLevelSystemStrongIDPriority(prefix string) int { return 2 case "proxmox": return 3 - case "docker": + case "vmware-host": return 4 - case "pbs": + case "docker": return 5 - case "pmg": + case "pbs": return 6 - case "k8s": + case "pmg": return 7 - case "resource": + case "k8s": return 8 + case "resource": + return 9 default: return 99 } @@ -1027,3 +1034,10 @@ func topLevelSystemCanonicalHostname(resource Resource) string { } return strings.TrimSpace(resource.Canonical.Hostname) } + +func canonicalVMwareHostname(resource Resource) string { + if resource.VMware == nil { + return "" + } + return firstTrimmed(firstIdentityHostname(resource.Identity), resource.Name) +} diff --git a/internal/unifiedresources/top_level_systems_test.go b/internal/unifiedresources/top_level_systems_test.go index 39d733f6d1..dd45d532b0 100644 --- a/internal/unifiedresources/top_level_systems_test.go +++ b/internal/unifiedresources/top_level_systems_test.go @@ -37,6 +37,15 @@ func TestResolveTopLevelSystemsTopLevelSourceMatrix(t *testing.T) { wantCount: 1, same: [][2]string{{"agent-host", "truenas-system"}}, }, + { + name: "agent and vmware host share one top-level system", + resources: []Resource{ + topLevelTestAgent("agent-host", "esxi-01.lab.local", "machine-1", "agent-1"), + topLevelTestVMwareHost("vmware-host", "esxi-01.lab.local", "vc-1", "machine-1"), + }, + wantCount: 1, + same: [][2]string{{"agent-host", "vmware-host"}}, + }, { name: "agent and pbs share one top-level system", resources: []Resource{ @@ -304,6 +313,25 @@ func topLevelTestTrueNAS(id, hostname string) Resource { } } +func topLevelTestVMwareHost(id, hostname, connectionID, hostUUID string) Resource { + return Resource{ + ID: id, + Type: ResourceTypeAgent, + Name: hostname, + Status: StatusOnline, + VMware: &VMwareData{ + ConnectionID: connectionID, + ManagedObjectID: id, + EntityType: "host", + HostUUID: hostUUID, + }, + Identity: ResourceIdentity{ + DMIUUID: hostUUID, + Hostnames: []string{hostname}, + }, + } +} + func topLevelTestPBS(id, hostname, instanceID string) Resource { return Resource{ ID: id, diff --git a/pkg/licensing/entitlement_payload.go b/pkg/licensing/entitlement_payload.go index e35de3bab2..6409136c05 100644 --- a/pkg/licensing/entitlement_payload.go +++ b/pkg/licensing/entitlement_payload.go @@ -154,6 +154,10 @@ type LimitStatus struct { // Current is the observed current usage. Current int64 `json:"current"` + // CurrentAvailable reports whether Current reflects a resolved runtime + // usage value rather than an unavailable best-effort fallback. + CurrentAvailable *bool `json:"current_available,omitempty"` + // State describes the over-limit UX state. // Values: "ok", "warning", "enforced" State string `json:"state"` @@ -182,10 +186,11 @@ func (c LegacyConnectionCounts) Total() int64 { } type EntitlementUsageSnapshot struct { - MonitoredSystems int64 - Nodes int64 // legacy compatibility alias for monitored systems - Guests int64 - LegacyConnections LegacyConnectionCounts + MonitoredSystems int64 + MonitoredSystemsAvailable bool + Nodes int64 // legacy compatibility alias for monitored systems + Guests int64 + LegacyConnections LegacyConnectionCounts } func (s EntitlementUsageSnapshot) monitoredSystemCount() int64 { @@ -195,6 +200,13 @@ func (s EntitlementUsageSnapshot) monitoredSystemCount() int64 { return s.Nodes } +func (s EntitlementUsageSnapshot) monitoredSystemCountAvailable() bool { + if s.MonitoredSystemsAvailable { + return true + } + return s.MonitoredSystems > 0 || s.Nodes > 0 +} + // BuildEntitlementPayload constructs the normalized payload from LicenseStatus. func BuildEntitlementPayload(status *LicenseStatus, subscriptionState string) EntitlementPayload { return BuildEntitlementPayloadWithUsage(status, subscriptionState, EntitlementUsageSnapshot{}, nil) @@ -352,10 +364,11 @@ func BuildEntitlementPayloadWithUsage( if status.MaxMonitoredSystems > 0 { currentSystems := usage.monitoredSystemCount() payload.Limits = append(payload.Limits, LimitStatus{ - Key: MaxMonitoredSystemsLicenseGateKey, - Limit: int64(status.MaxMonitoredSystems), - Current: currentSystems, - State: LimitState(currentSystems, int64(status.MaxMonitoredSystems)), + Key: MaxMonitoredSystemsLicenseGateKey, + Limit: int64(status.MaxMonitoredSystems), + Current: currentSystems, + CurrentAvailable: boolPointer(usage.monitoredSystemCountAvailable()), + State: LimitState(currentSystems, int64(status.MaxMonitoredSystems)), }) } if status.MaxGuests > 0 { @@ -413,6 +426,11 @@ func remainingTrialDays(expiresAtUnix, nowUnix int64) int { return daysRemaining } +func boolPointer(value bool) *bool { + v := value + return &v +} + // LimitState returns the over-limit UX state string. func LimitState(current, limit int64) string { if limit <= 0 { diff --git a/pkg/licensing/entitlement_payload_test.go b/pkg/licensing/entitlement_payload_test.go index e405119e91..d2db1c63e2 100644 --- a/pkg/licensing/entitlement_payload_test.go +++ b/pkg/licensing/entitlement_payload_test.go @@ -100,6 +100,9 @@ func TestBuildEntitlementPayloadWithUsage_CurrentValues(t *testing.T) { if agentLimit.Current != 12 { t.Fatalf("expected agent current 12, got %d", agentLimit.Current) } + if agentLimit.CurrentAvailable == nil || !*agentLimit.CurrentAvailable { + t.Fatalf("expected agent current availability to be true, got %+v", agentLimit.CurrentAvailable) + } if guestLimit.Current != 44 { t.Fatalf("expected guest current 44, got %d", guestLimit.Current) } @@ -145,6 +148,26 @@ func TestBuildEntitlementPayload_TrialState(t *testing.T) { } } +func TestBuildEntitlementPayloadWithUsage_MonitoredSystemUsageUnavailable(t *testing.T) { + status := &LicenseStatus{ + Valid: true, + Tier: TierPro, + Features: append([]string(nil), TierFeatures[TierPro]...), + MaxMonitoredSystems: 50, + } + + payload := BuildEntitlementPayloadWithUsage(status, "", EntitlementUsageSnapshot{}, nil) + if len(payload.Limits) != 1 { + t.Fatalf("expected one limit, got %d", len(payload.Limits)) + } + if payload.Limits[0].Current != 0 { + t.Fatalf("expected unresolved current to fall back to 0, got %d", payload.Limits[0].Current) + } + if payload.Limits[0].CurrentAvailable == nil || *payload.Limits[0].CurrentAvailable { + t.Fatalf("expected unresolved current availability to be false, got %+v", payload.Limits[0].CurrentAvailable) + } +} + func TestBuildEntitlementPayload_CopiesStatusDisplayFields(t *testing.T) { tests := []struct { name string @@ -353,6 +376,9 @@ func TestBuildRuntimeCapabilitiesPayloadWithUsage_CurrentValues(t *testing.T) { if payload.Limits[0].Current != 5 { t.Fatalf("expected runtime current 5, got %d", payload.Limits[0].Current) } + if payload.Limits[0].CurrentAvailable == nil || !*payload.Limits[0].CurrentAvailable { + t.Fatalf("expected runtime current availability true, got %+v", payload.Limits[0].CurrentAvailable) + } } func TestBuildRuntimeCapabilitiesPayload_NilStatusDefaultsToFreeTier(t *testing.T) {