vrr/Pulse
2
0
Fork 0
mirror of https://github.com/rcourtman/Pulse.git synced 2026-07-09 16:00:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

Fix host metric percent normalization

Browse source 
Apply token-gated command policy when deriving connection fleet state.
This commit is contained in:
rcourtman 2026-07-06 10:08:36 +01:00
parent 44af57b8fc
commit c9c5f34eb4
14 changed files with 320 additions and 33 deletions
Download patch file Download diff file Expand all files Collapse all files

58
internal/ai/parity_test.go
View file

@ -288,15 +288,15 @@ func TestParityHostFields(t *testing.T) {
KernelVersion: "6.8.0",
Architecture: "amd64",
CPUCount: 16,
CPUUsage: 0.25,
Memory: testMemory(),
CPUUsage: 25,
Memory: testMemoryPercent(),
LoadAverage: []float64{0.21, 0.18, 0.22},
Disks: []models.Disk{
{
Total: 107374182400,
Used: 10737418240,
Free: 96636764160,
Usage: float64(10737418240) / float64(107374182400),
Usage: 10,
Mountpoint: "/",
Type: "ext4",
Device: "/dev/sda1",
@ -375,15 +375,15 @@ func TestParityDockerHostFields(t *testing.T) {
CPUs: 8,
TotalMemoryBytes: 17179869184,
UptimeSeconds: 123456,
CPUUsage: 0.15,
CPUUsage: 15,
LoadAverage: []float64{0.12, 0.10, 0.09},
Memory: testMemory(),
Memory: testMemoryPercent(),
Disks: []models.Disk{
{
Total: 107374182400,
Used: 10737418240,
Free: 96636764160,
Usage: float64(10737418240) / float64(107374182400),
Usage: 10,
Mountpoint: "/",
Type: "ext4",
Device: "/dev/sda1",
@ -440,8 +440,8 @@ func TestParityDockerHostFields(t *testing.T) {
require.InEpsilon(t, dockerTemp, v.Temperature(), 1e-9)
require.Len(t, v.NetworkInterfaces(), len(dh.NetworkInterfaces))
require.Len(t, v.Disks(), len(dh.Disks))
require.InEpsilon(t, percentFromUsage(dh.CPUUsage), v.CPUPercent(), 1e-9)
require.InEpsilon(t, percentFromUsage(dh.Memory.Usage), v.MemoryPercent(), 1e-9)
require.InEpsilon(t, dh.CPUUsage, v.CPUPercent(), 1e-9)
require.InEpsilon(t, dh.Memory.Usage, v.MemoryPercent(), 1e-9)
require.True(t, v.LastSeen().Equal(dh.LastSeen))
}
@ -730,9 +730,9 @@ func TestParityResourceCounts(t *testing.T) {
{Name: "eth0", MAC: "00:11:22:33:44:55", Addresses: []string{"192.168.50.10/24"}},
},
Sensors: models.HostSensorSummary{TemperatureCelsius: map[string]float64{"cpu_package": 50.0}},
Disks: []models.Disk{testDisk()},
Memory: testMemory(),
CPUUsage: 0.05,
Disks: []models.Disk{testDiskPercent()},
Memory: testMemoryPercent(),
CPUUsage: 5,
},
// One standalone host agent that should not merge with anything.
models.Host{
@ -750,9 +750,9 @@ func TestParityResourceCounts(t *testing.T) {
{Name: "eth0", MAC: "00:aa:bb:cc:dd:ee", Addresses: []string{"10.250.0.10/24"}},
},
Sensors: models.HostSensorSummary{TemperatureCelsius: map[string]float64{"cpu_package": 45.0}},
Disks: []models.Disk{testDisk()},
Memory: testMemory(),
CPUUsage: 0.07,
Disks: []models.Disk{testDiskPercent()},
Memory: testMemoryPercent(),
CPUUsage: 7,
},
)
@ -820,9 +820,9 @@ func TestExpectedDeltaHostMerge(t *testing.T) {
{Name: "eth0", MAC: "00:11:22:33:44:55", Addresses: []string{"192.168.50.10/24"}},
},
Sensors: models.HostSensorSummary{TemperatureCelsius: map[string]float64{"cpu_package": 51.0}},
Disks: []models.Disk{testDisk()},
Memory: testMemory(),
CPUUsage: 0.05,
Disks: []models.Disk{testDiskPercent()},
Memory: testMemoryPercent(),
CPUUsage: 5,
}
snapshot := models.StateSnapshot{
@ -876,9 +876,9 @@ func TestExpectedDeltaNodesThroughTwoViews(t *testing.T) {
{Name: "eth0", MAC: "00:11:22:33:44:55", Addresses: []string{"192.168.50.10/24"}},
},
Sensors: models.HostSensorSummary{TemperatureCelsius: map[string]float64{"cpu_package": 51.0}},
Disks: []models.Disk{testDisk()},
Memory: testMemory(),
CPUUsage: 0.05,
Disks: []models.Disk{testDiskPercent()},
Memory: testMemoryPercent(),
CPUUsage: 5,
}
snapshot := models.StateSnapshot{
@ -1103,10 +1103,10 @@ func testStateSnapshot() models.StateSnapshot {
CPUs: 8,
TotalMemoryBytes: 17179869184,
UptimeSeconds: 123456,
CPUUsage: 0.15,
CPUUsage: 15,
LoadAverage: []float64{0.12, 0.10, 0.09},
Memory: testMemory(),
Disks: []models.Disk{testDisk()},
Memory: testMemoryPercent(),
Disks: []models.Disk{testDiskPercent()},
NetworkInterfaces: []models.HostNetworkInterface{{Name: "eth0", MAC: "de:ad:be:ef:00:01", Addresses: []string{"10.10.0.10/24"}}},
Status: "online",
LastSeen: now,
@ -1247,6 +1247,12 @@ func testMemory() models.Memory {
}
}
func testMemoryPercent() models.Memory {
mem := testMemory()
mem.Usage = 25
return mem
}
func testDisk() models.Disk {
used := int64(10737418240) // 10GB
total := int64(107374182400) // 100GB
@ -1258,6 +1264,12 @@ func testDisk() models.Disk {
}
}
func testDiskPercent() models.Disk {
disk := testDisk()
disk.Usage = 10
return disk
}
func percentFromUsage(value float64) float64 {
if value <= 1.0 {
return value * 100

36
internal/api/connections_aggregator.go
View file

@ -97,6 +97,7 @@ type aggregatorInputs struct {
availabilityTargets []config.AvailabilityTarget
availabilityStatuses map[string]monitoring.AvailabilityProbeStatus
hosts []models.Host
apiTokens []config.APITokenRecord
agentDesiredConfigs map[string]connectionAgentDesiredConfig
instanceHealth map[string]monitoring.InstanceHealth
expectedAgentVersion string
@ -975,18 +976,20 @@ func connectionConfigFingerprint(version string, payload any) *ConnectionFleetCo
}
}
func connectionAgentDesiredConfigFingerprints(monitor *monitoring.Monitor, hosts []models.Host) map[string]connectionAgentDesiredConfig {
func connectionAgentDesiredConfigFingerprints(monitor *monitoring.Monitor, hosts []models.Host, tokens []config.APITokenRecord) map[string]connectionAgentDesiredConfig {
if monitor == nil || len(hosts) == 0 {
return nil
}
configs := make(map[string]connectionAgentDesiredConfig, len(hosts))
tokenByID := connectionAgentTokenRecordsByID(tokens)
for _, host := range hosts {
hostID := strings.TrimSpace(host.ID)
if hostID == "" {
continue
}
cfg := monitor.GetHostAgentConfig(hostID)
cfg = effectiveConnectionAgentConfig(cfg, host, tokenByID)
desired := connectionAgentDesiredConfig{
CommandsEnabled: cloneBoolPtr(cfg.CommandsEnabled),
}
@ -1003,6 +1006,37 @@ func connectionAgentDesiredConfigFingerprints(monitor *monitoring.Monitor, hosts
return configs
}
func connectionAgentTokenRecordsByID(tokens []config.APITokenRecord) map[string]*config.APITokenRecord {
if len(tokens) == 0 {
return nil
}
records := make(map[string]*config.APITokenRecord, len(tokens))
for i := range tokens {
id := strings.TrimSpace(tokens[i].ID)
if id == "" {
continue
}
records[id] = &tokens[i]
}
return records
}
func effectiveConnectionAgentConfig(cfg monitoring.HostAgentConfig, host models.Host, tokenByID map[string]*config.APITokenRecord) monitoring.HostAgentConfig {
var record *config.APITokenRecord
if tokenByID != nil {
record = tokenByID[strings.TrimSpace(host.TokenID)]
}
cfg = sanitizeHostAgentConfigForToken(cfg, record, host)
metadata, err := remoteconfig.BuildDesiredConfigMetadata(cfg.CommandsEnabled, cfg.Settings)
if err != nil {
cfg.DesiredConfig = nil
return cfg
}
cfg.DesiredConfig = &metadata
return cfg
}
func connectionAgentDesiredConfigForHost(configs map[string]connectionAgentDesiredConfig, hostID string) *connectionAgentDesiredConfig {
if len(configs) == 0 {
return nil

78
internal/api/connections_aggregator_test.go
View file

@ -653,7 +653,7 @@ func TestConnectionAgentDesiredConfigFingerprintsSkipsEmptyDefaultConfig(t *test
t.Cleanup(func() { monitor.Stop() })
hostID := "default-agent"
got := connectionAgentDesiredConfigFingerprints(monitor, []models.Host{{ID: hostID}})
got := connectionAgentDesiredConfigFingerprints(monitor, []models.Host{{ID: hostID}}, nil)
desired, ok := got[hostID]
if !ok {
t.Fatalf("expected resolved desired config entry for %q, got %+v", hostID, got)
@ -669,7 +669,7 @@ func TestConnectionAgentDesiredConfigFingerprintsSkipsEmptyDefaultConfig(t *test
if err := monitor.UpdateHostAgentConfig(hostID, &commandsEnabled); err != nil {
t.Fatalf("UpdateHostAgentConfig: %v", err)
}
got = connectionAgentDesiredConfigFingerprints(monitor, []models.Host{{ID: hostID}})
got = connectionAgentDesiredConfigFingerprints(monitor, []models.Host{{ID: hostID}}, nil)
desired = got[hostID]
if desired.Fingerprint == nil {
t.Fatalf("managed command override should create desired config fingerprint")
@ -679,6 +679,80 @@ func TestConnectionAgentDesiredConfigFingerprintsSkipsEmptyDefaultConfig(t *test
}
}
func TestConnectionAgentDesiredConfigFingerprintsUsesTokenEffectiveCommandPolicy(t *testing.T) {
monitor, err := monitoring.New(&config.Config{DataPath: t.TempDir()})
if err != nil {
t.Fatalf("monitoring.New: %v", err)
}
t.Cleanup(func() { monitor.Stop() })
hostID := "agent-token-gated"
rawDesiredCommands := true
if err := monitor.UpdateHostAgentConfig(hostID, &rawDesiredCommands); err != nil {
t.Fatalf("UpdateHostAgentConfig: %v", err)
}
now := time.Now()
host := models.Host{
ID: hostID,
Hostname: "agent-token-gated.local",
LastSeen: now,
CommandsEnabled: false,
TokenID: "runtime-without-exec",
}
tokens := []config.APITokenRecord{
{
ID: "runtime-without-exec",
Scopes: []string{config.ScopeAgentReport, config.ScopeAgentConfigRead},
},
}
got := connectionAgentDesiredConfigFingerprints(monitor, []models.Host{host}, tokens)
desired, ok := got[hostID]
if !ok {
t.Fatalf("expected resolved desired config entry for %q, got %+v", hostID, got)
}
if desired.CommandsEnabled == nil || *desired.CommandsEnabled {
t.Fatalf("effective desired CommandsEnabled = %+v, want false", desired.CommandsEnabled)
}
if desired.Fingerprint == nil {
t.Fatal("effective command override should still create a desired config fingerprint")
}
effectiveDisabled := false
effectiveMetadata, err := remoteconfig.BuildDesiredConfigMetadata(&effectiveDisabled, nil)
if err != nil {
t.Fatalf("BuildDesiredConfigMetadata disabled: %v", err)
}
rawMetadata, err := remoteconfig.BuildDesiredConfigMetadata(&rawDesiredCommands, nil)
if err != nil {
t.Fatalf("BuildDesiredConfigMetadata raw: %v", err)
}
if desired.Fingerprint.Hash != effectiveMetadata.Hash {
t.Fatalf("desired fingerprint hash = %q, want effective disabled hash %q", desired.Fingerprint.Hash, effectiveMetadata.Hash)
}
if desired.Fingerprint.Hash == rawMetadata.Hash {
t.Fatalf("desired fingerprint reused raw unsanitized command policy hash %q", desired.Fingerprint.Hash)
}
connections := buildConnections(aggregatorInputs{
hosts: []models.Host{host},
agentDesiredConfigs: got,
now: now,
})
if len(connections) != 1 {
t.Fatalf("expected one connection, got %d", len(connections))
}
policy := connections[0].Fleet.CommandPolicy
if policy == nil ||
policy.Status != fleetStateDisabled ||
policy.Desired != fleetStateDisabled ||
policy.Applied != fleetStateDisabled ||
policy.Enforcement != fleetCommandPolicyInSync {
t.Fatalf("command policy = %+v, want effective disabled in sync", policy)
}
}
func TestConnectionFleetAgentConfigDriftComparesAppliedFingerprintsWhenAvailable(t *testing.T) {
now := time.Now()
conn := Connection{

3
internal/api/connections_alerts.go
View file

@ -30,6 +30,7 @@ func buildAggregatorInputsWithRuntimeSources(
inputs.pveInstances = cfg.PVEInstances
inputs.pbsInstances = cfg.PBSInstances
inputs.pmgInstances = cfg.PMGInstances
inputs.apiTokens = append([]config.APITokenRecord(nil), cfg.APITokens...)
}
if persistence != nil {
@ -57,7 +58,7 @@ func buildAggregatorInputsWithRuntimeSources(
if monitor != nil {
inputs.hosts = monitor.HostsSnapshot()
inputs.agentDesiredConfigs = connectionAgentDesiredConfigFingerprints(monitor, inputs.hosts)
inputs.agentDesiredConfigs = connectionAgentDesiredConfigFingerprints(monitor, inputs.hosts, inputs.apiTokens)
inputs.instanceHealth = instanceHealthByKey(monitor.SchedulerHealth())
inputs.availabilityStatuses = monitor.AvailabilityStatusSnapshot()
} else {

57
internal/api/contract_test.go
View file

@ -14701,6 +14701,63 @@ func TestContract_AgentDefaultDesiredConfigDoesNotCreateRolloutAttention(t *test
assertJSONSnapshot(t, body, want)
}
func TestContract_AgentConnectionPayloadUsesTokenEffectiveCommandPolicy(t *testing.T) {
cfg := &config.Config{
DataPath: t.TempDir(),
APITokens: []config.APITokenRecord{
{
ID: "runtime-without-exec",
Scopes: []string{config.ScopeAgentReport, config.ScopeAgentConfigRead},
},
},
}
monitor, err := monitoring.New(cfg)
if err != nil {
t.Fatalf("monitoring.New: %v", err)
}
t.Cleanup(func() { monitor.Stop() })
hostID := "host-token-gated"
rawDesiredCommands := true
if err := monitor.UpdateHostAgentConfig(hostID, &rawDesiredCommands); err != nil {
t.Fatalf("UpdateHostAgentConfig: %v", err)
}
now := time.Date(2026, 5, 14, 10, 30, 0, 0, time.UTC)
host := models.Host{
ID: hostID,
Hostname: "host-token-gated",
ReportIP: "192.0.2.43",
LastSeen: now.Add(-10 * time.Second),
AgentVersion: "6.0.0",
Platform: "linux",
CommandsEnabled: false,
TokenID: "runtime-without-exec",
}
desiredConfigs := connectionAgentDesiredConfigFingerprints(monitor, []models.Host{host}, cfg.APITokens)
connections := buildConnections(aggregatorInputs{
hosts: []models.Host{host},
apiTokens: cfg.APITokens,
agentDesiredConfigs: desiredConfigs,
expectedAgentVersion: "6.0.0",
now: now,
})
if len(connections) != 1 {
t.Fatalf("expected one agent connection, got %d", len(connections))
}
body, err := json.Marshal(connections[0])
if err != nil {
t.Fatalf("marshal agent Connection: %v", err)
}
want := `{"id":"agent:host-token-gated","type":"agent","name":"host-token-gated","address":"host-token-gated","hostAliases":["host-token-gated","192.0.2.43"],"state":"active","enabled":true,"surfaces":["host"],"scope":{"host":true},"lastSeen":"2026-05-14T10:29:50Z","source":"agent","agentIdentity":{"hostname":"host-token-gated","platform":"linux","reportIp":"192.0.2.43"},"agentVersion":"6.0.0","expectedAgentVersion":"6.0.0","fleet":{"enrollmentState":"enrolled","livenessState":"active","versionDrift":"current","adapterHealth":"healthy","configRollout":"reported","credentialStatus":"verified","updateStatus":"current","remoteControl":"disabled","configDrift":{"status":"pending","desired":{"version":"host-agent-config/v1","hash":"sha256:59378fe4db8132c2ce1d9c16b492c16093156579d09357db44e34a0fab494bea"},"reason":"Pulse has not received a comparable applied agent configuration fingerprint yet"},"rollout":{"status":"pending","stage":"pending","reason":"waiting for the agent to report an applied configuration fingerprint"},"credentialHealth":{"status":"verified","kind":"agent-token","rotation":"healthy","lastVerifiedAt":"2026-05-14T10:29:50Z"},"commandPolicy":{"status":"disabled","desired":"disabled","applied":"disabled","enforcement":"in-sync","reason":"agent command execution matches the desired disabled policy"}},"capabilities":{"supportsPause":false,"supportsScope":false,"supportsTest":false}}`
assertJSONSnapshot(t, body, want)
if strings.Contains(string(body), `"desired":"enabled"`) {
t.Fatalf("connections payload used unsanitized command desire: %s", body)
}
}
func TestContract_ConnectionsListIncludesAgentHostsFromUnifiedReadState(t *testing.T) {
cfg := &config.Config{DataPath: t.TempDir()}
monitor, err := monitoring.New(cfg)

2
internal/unifiedresources/adapters.go
View file

@ -828,7 +828,7 @@ func unraidStorageCapacity(host models.Host) (int64, int64, int64, float64) {
for _, disk := range host.Disks {
mount := strings.TrimSpace(disk.Mountpoint)
if mount == "/mnt/user" || mount == "/mnt/user0" {
return disk.Total, disk.Used, disk.Free, percentFromUsage(disk.Usage)
return disk.Total, disk.Used, disk.Free, percentFromReportedPercent(disk.Usage)
}
}

22
internal/unifiedresources/adapters_test.go
View file

@ -985,6 +985,28 @@ func TestResourceFromHostUnraidStorageIncludesTopologyMetadata(t *testing.T) {
}
}
func TestResourceFromHostUnraidStorageKeepsReportedDiskPercent(t *testing.T) {
host := models.Host{
ID: "tower-host",
Hostname: "tower",
DisplayName: "Tower",
Status: "online",
LastSeen: time.Now().UTC(),
Disks: []models.Disk{
{Mountpoint: "/mnt/user", Total: 1000, Used: 4, Free: 996, Usage: 0.4},
},
Unraid: &models.HostUnraidStorage{
ArrayStarted: true,
ArrayState: "STARTED",
},
}
resource, _ := resourceFromHostUnraidStorage(host)
if resource.Metrics == nil || resource.Metrics.Disk == nil || resource.Metrics.Disk.Percent != 0.4 {
t.Fatalf("expected reported /mnt/user disk percent 0.4, got %+v", resource.Metrics)
}
}
func TestResourceFromHostDerivesStorageTopologyRisk(t *testing.T) {
host := models.Host{
ID: "tower-host",

12
internal/unifiedresources/metrics.go
View file

@ -87,16 +87,16 @@ func buildHostMetricPayload(
source DataSource,
) *ResourceMetrics {
metrics := &ResourceMetrics{}
cpuPercent := percentFromUsage(cpuUsage)
cpuPercent := percentFromReportedPercent(cpuUsage)
metrics.CPU = &MetricValue{Value: cpuPercent, Percent: cpuPercent, Unit: "percent", Source: source}
if memory.Total > 0 {
percent := percentFromUsage(memory.Usage)
percent := percentFromReportedPercent(memory.Usage)
metrics.Memory = &MetricValue{Used: &memory.Used, Total: &memory.Total, Percent: percent, Unit: "bytes", Source: source}
}
if len(disks) > 0 {
disk := disks[0]
if disk.Total > 0 {
percent := percentFromUsage(disk.Usage)
percent := percentFromReportedPercent(disk.Usage)
metrics.Disk = &MetricValue{Used: &disk.Used, Total: &disk.Total, Percent: percent, Unit: "bytes", Source: source}
}
}
@ -337,7 +337,7 @@ func metricsFromKubernetesCluster(cluster models.KubernetesCluster, linkedHosts
continue
}
cpuSum += percentFromUsage(host.CPUUsage)
cpuSum += percentFromReportedPercent(host.CPUUsage)
cpuCount++
hostName := strings.TrimSpace(host.Hostname)
@ -792,6 +792,10 @@ func percentFromUsage(value float64) float64 {
return value
}
func percentFromReportedPercent(value float64) float64 {
return clampMetricValue(value, 0, 100)
}
func setNetworkAndDiskIOMetricsHost(metrics *ResourceMetrics, netIn, netOut, diskRead, diskWrite float64, source DataSource) {
if netIn > 0 {
metrics.NetIn = &MetricValue{Value: netIn, Unit: "bytes/s", Source: source}

57
internal/unifiedresources/metrics_test.go
View file

@ -32,6 +32,63 @@ func TestMetricsFromDockerHostIncludesIORates(t *testing.T) {
}
}
func TestMetricsFromHostKeepsReportedPercentValues(t *testing.T) {
host := models.Host{
CPUUsage: 1,
Memory: models.Memory{Total: 1000, Used: 5, Usage: 0.5},
Disks: []models.Disk{{Total: 1000, Used: 2, Usage: 0.2}},
}
metrics := metricsFromHost(host)
if metrics.CPU == nil || metrics.CPU.Percent != 1 {
t.Fatalf("expected host cpu percent 1, got %+v", metrics.CPU)
}
if metrics.Memory == nil || metrics.Memory.Percent != 0.5 {
t.Fatalf("expected host memory percent 0.5, got %+v", metrics.Memory)
}
if metrics.Disk == nil || metrics.Disk.Percent != 0.2 {
t.Fatalf("expected host disk percent 0.2, got %+v", metrics.Disk)
}
}
func TestMetricsFromDockerHostKeepsReportedPercentValues(t *testing.T) {
host := models.DockerHost{
CPUUsage: 1,
Memory: models.Memory{Total: 1000, Used: 5, Usage: 0.5},
Disks: []models.Disk{{Total: 1000, Used: 2, Usage: 0.2}},
}
metrics := metricsFromDockerHost(host)
if metrics.CPU == nil || metrics.CPU.Percent != 1 {
t.Fatalf("expected docker host cpu percent 1, got %+v", metrics.CPU)
}
if metrics.Memory == nil || metrics.Memory.Percent != 0.5 {
t.Fatalf("expected docker host memory percent 0.5, got %+v", metrics.Memory)
}
if metrics.Disk == nil || metrics.Disk.Percent != 0.2 {
t.Fatalf("expected docker host disk percent 0.2, got %+v", metrics.Disk)
}
}
func TestMetricsFromKubernetesClusterAggregatesLinkedHostReportedPercents(t *testing.T) {
cluster := models.KubernetesCluster{ID: "cluster-1", Name: "cluster-1"}
hosts := []*models.Host{
{
Hostname: "worker-1",
CPUUsage: 1,
},
{
Hostname: "worker-2",
CPUUsage: 2,
},
}
metrics := metricsFromKubernetesCluster(cluster, hosts)
if metrics.CPU == nil || metrics.CPU.Percent != 1.5 {
t.Fatalf("expected linked host cluster cpu percent 1.5, got %+v", metrics.CPU)
}
}
func TestMetricsFromKubernetesPod_MockModeIncludesLiveLikeValues(t *testing.T) {
enableMockMode(t)