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

Canonicalize monitored-system admission enforcement

Browse source
This commit is contained in:
rcourtman 2026-04-07 20:11:57 +01:00
parent e327ff9726
commit 7a47f0923e
30 changed files with 2060 additions and 201 deletions
Download patch file Download diff file Expand all files Collapse all files

7
docs/release-control/v6/internal/subsystems/agent-lifecycle.md
View file

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

7
docs/release-control/v6/internal/subsystems/api-contracts.md
View file

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

11
docs/release-control/v6/internal/subsystems/cloud-paid.md
View file

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

6
docs/release-control/v6/internal/subsystems/performance-and-scalability.md
View file

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

12
docs/release-control/v6/internal/subsystems/storage-recovery.md
View file

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

19
docs/release-control/v6/internal/subsystems/unified-resources.md
View file

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

79
internal/api/config_handlers_add_test.go
View file

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

4
internal/api/config_handlers_auto_register_test.go
View file

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

75
internal/api/config_handlers_update_test.go
View file

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

288
internal/api/config_node_handlers.go
View file

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

6
internal/api/config_setup_handlers.go
View file

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

32
internal/api/contract_test.go
View file

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

18
internal/api/deploy_handlers.go
View file

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

23
internal/api/entitlement_handlers_test.go
View file

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

6
internal/api/license_limits_test_helpers_test.go
View file

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

284
internal/api/monitored_system_limit_enforcement.go
View file

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

1
internal/api/router.go
View file

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

18
internal/api/subscription_entitlements.go
View file

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

84
internal/api/truenas_handlers.go
View file

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

140
internal/api/truenas_handlers_test.go
View file

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

135
internal/api/vmware_handlers.go
View file

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

283
internal/api/vmware_handlers_test.go
View file

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

341
internal/unifiedresources/monitored_system_projection.go Normal file
View file

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

3
internal/unifiedresources/monitored_systems.go
View file

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

173
internal/unifiedresources/monitored_systems_test.go
View file

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

56
internal/unifiedresources/registry_test.go
View file

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

62
internal/unifiedresources/top_level_systems.go
View file

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

28
internal/unifiedresources/top_level_systems_test.go
View file

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

34
pkg/licensing/entitlement_payload.go
View file

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

26
pkg/licensing/entitlement_payload_test.go
View file

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