danvics/Pulse
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

style: fix gofmt formatting inconsistencies

Browse source 
Run gofmt -w to fix tab/space inconsistencies across 33 files.
This commit is contained in:
rcourtman 2025-11-26 23:44:36 +00:00
parent b40a33fd2b
commit 2fe7bb6141
33 changed files with 584 additions and 584 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
cmd/pulse-host-agent/service_stub.go
View file

@ -1,3 +1,4 @@
//go:build !windows
// +build !windows // +build !windows
package main package main

1
cmd/pulse-host-agent/service_windows.go
View file

@ -1,3 +1,4 @@
//go:build windows
// +build windows // +build windows
package main package main

104
cmd/pulse-sensor-proxy/audit_test.go
View file

@ -1,70 +1,70 @@
package main package main
import ( import (
"bufio" "bufio"
"encoding/json" "encoding/json"
"os" "os"
"testing" "testing"
) )
type auditRecord map[string]interface{} type auditRecord map[string]interface{}
func TestAuditLogValidationFailure(t *testing.T) { func TestAuditLogValidationFailure(t *testing.T) {
tmp, err := os.CreateTemp("", "audit-test-*.log") tmp, err := os.CreateTemp("", "audit-test-*.log")
if err != nil { if err != nil {
t.Fatalf("temp file: %v", err) t.Fatalf("temp file: %v", err)
} }
path := tmp.Name() path := tmp.Name()
tmp.Close() tmp.Close()
defer os.Remove(path) defer os.Remove(path)
logger := newAuditLogger(path) logger := newAuditLogger(path)
cred := &peerCredentials{uid: 1000, gid: 1000, pid: 4242} cred := &peerCredentials{uid: 1000, gid: 1000, pid: 4242}
logger.LogValidationFailure("corr-123", cred, "remote", "get_temperature", []string{"node"}, "invalid_node") logger.LogValidationFailure("corr-123", cred, "remote", "get_temperature", []string{"node"}, "invalid_node")
logger.Close() logger.Close()
file, err := os.Open(path) file, err := os.Open(path)
if err != nil { if err != nil {
t.Fatalf("open log: %v", err) t.Fatalf("open log: %v", err)
} }
defer file.Close() defer file.Close()
scanner := bufio.NewScanner(file) scanner := bufio.NewScanner(file)
if !scanner.Scan() { if !scanner.Scan() {
t.Fatalf("expected at least one audit entry (file may be empty)") t.Fatalf("expected at least one audit entry (file may be empty)")
} }
line := scanner.Bytes() line := scanner.Bytes()
if len(line) == 0 { if len(line) == 0 {
t.Fatalf("empty line in audit log") t.Fatalf("empty line in audit log")
} }
t.Logf("Audit log line: %s", string(line)) t.Logf("Audit log line: %s", string(line))
var record auditRecord var record auditRecord
if err := json.Unmarshal(line, &record); err != nil { if err := json.Unmarshal(line, &record); err != nil {
t.Fatalf("unmarshal (line=%s): %v", string(line), err) t.Fatalf("unmarshal (line=%s): %v", string(line), err)
} }
t.Logf("Parsed record: %+v", record) t.Logf("Parsed record: %+v", record)
if record["event_type"] != "command.validation_failed" { if record["event_type"] != "command.validation_failed" {
t.Fatalf("unexpected event_type: %v (full record: %+v)", record["event_type"], record) t.Fatalf("unexpected event_type: %v (full record: %+v)", record["event_type"], record)
} }
if record["correlation_id"] != "corr-123" { if record["correlation_id"] != "corr-123" {
t.Fatalf("unexpected correlation id: %v", record["correlation_id"]) t.Fatalf("unexpected correlation id: %v", record["correlation_id"])
} }
if record["command"] != "get_temperature" { if record["command"] != "get_temperature" {
t.Fatalf("unexpected command: %v", record["command"]) t.Fatalf("unexpected command: %v", record["command"])
} }
if record["reason"] != "invalid_node" { if record["reason"] != "invalid_node" {
t.Fatalf("unexpected reason: %v", record["reason"]) t.Fatalf("unexpected reason: %v", record["reason"])
} }
if record["decision"] != "denied" { if record["decision"] != "denied" {
t.Fatalf("unexpected decision: %v", record["decision"]) t.Fatalf("unexpected decision: %v", record["decision"])
} }
if record["event_hash"] == "" { if record["event_hash"] == "" {
t.Fatalf("expected event_hash to be set") t.Fatalf("expected event_hash to be set")
} }
} }

46
cmd/pulse-sensor-proxy/validation_fuzz_test.go
View file

@ -1,31 +1,31 @@
package main package main
import ( import (
"strings" "strings"
"testing" "testing"
) )
func FuzzValidateCommand(f *testing.F) { func FuzzValidateCommand(f *testing.F) {
seeds := []string{ seeds := []string{
"sensors -j", "sensors -j",
"ipmitool sdr", "ipmitool sdr",
"sensors", "sensors",
"ipmitool lan print", "ipmitool lan print",
} }
for _, seed := range seeds { for _, seed := range seeds {
f.Add(seed) f.Add(seed)
} }
f.Fuzz(func(t *testing.T, input string) { f.Fuzz(func(t *testing.T, input string) {
fields := strings.Fields(input) fields := strings.Fields(input)
if len(fields) == 0 { if len(fields) == 0 {
return return
} }
cmd := fields[0] cmd := fields[0]
args := []string{} args := []string{}
if len(fields) > 1 { if len(fields) > 1 {
args = fields[1:] args = fields[1:]
} }
validateCommand(cmd, args) // ensure no panics validateCommand(cmd, args) // ensure no panics
}) })
} }

57
cmd/pulse/metrics_server.go
View file

@ -1,40 +1,39 @@
package main package main
import ( import (
"context" "context"
"net/http" "net/http"
"time" "time"
"github.com/prometheus/client_golang/prometheus/promhttp" "github.com/prometheus/client_golang/prometheus/promhttp"
"github.com/rs/zerolog/log" "github.com/rs/zerolog/log"
) )
func startMetricsServer(ctx context.Context, addr string) { func startMetricsServer(ctx context.Context, addr string) {
mux := http.NewServeMux() mux := http.NewServeMux()
mux.Handle("/metrics", promhttp.Handler()) mux.Handle("/metrics", promhttp.Handler())
srv := &http.Server{ srv := &http.Server{
Addr: addr, Addr: addr,
Handler: mux, Handler: mux,
ReadTimeout: 5 * time.Second, ReadTimeout: 5 * time.Second,
WriteTimeout: 10 * time.Second, WriteTimeout: 10 * time.Second,
IdleTimeout: 30 * time.Second, IdleTimeout: 30 * time.Second,
} }
go func() { go func() {
<-ctx.Done() <-ctx.Done()
shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel() defer cancel()
if err := srv.Shutdown(shutdownCtx); err != nil && err != http.ErrServerClosed { if err := srv.Shutdown(shutdownCtx); err != nil && err != http.ErrServerClosed {
log.Warn().Err(err).Msg("Failed to shut down metrics server cleanly") log.Warn().Err(err).Msg("Failed to shut down metrics server cleanly")
} }
}() }()
go func() { go func() {
log.Info().Str("addr", addr).Msg("Metrics endpoint listening") log.Info().Str("addr", addr).Msg("Metrics endpoint listening")
if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
log.Warn().Err(err).Msg("Metrics server stopped unexpectedly") log.Warn().Err(err).Msg("Metrics server stopped unexpectedly")
} }
}() }()
} }

14
internal/alerts/per_metric_delay_example_test.go
View file

@ -10,12 +10,12 @@ import (
// This test shows how to configure different alert delays for different metrics // This test shows how to configure different alert delays for different metrics
func TestPerMetricDelayConfiguration(t *testing.T) { func TestPerMetricDelayConfiguration(t *testing.T) {
tests := []struct { tests := []struct {
name string name string
config AlertConfig config AlertConfig
resourceType string resourceType string
metricType string metricType string
expectedDelay int expectedDelay int
description string description string
}{ }{
{ {
name: "CPU alert with longer delay than memory", name: "CPU alert with longer delay than memory",
@ -117,7 +117,7 @@ func TestPerMetricDelayConfiguration(t *testing.T) {
}, },
MetricTimeThresholds: map[string]map[string]int{ MetricTimeThresholds: map[string]map[string]int{
"docker": { "docker": {
"restartcount": 10, // Quick notification for container restarts "restartcount": 10, // Quick notification for container restarts
"cpu": 120, // Longer for CPU "cpu": 120, // Longer for CPU
}, },
}, },

12
internal/alerts/time_threshold_test.go
View file

@ -70,12 +70,12 @@ func TestGetTimeThresholdMetricOverrides(t *testing.T) {
metricType string metricType string
expected int expected int
}{ }{
{"vm-resource", "VM", "cpu", 5}, // guest metric override {"vm-resource", "VM", "cpu", 5}, // guest metric override
{"vm-resource", "VM", "memory", 30}, // falls back to guest type delay {"vm-resource", "VM", "memory", 30}, // falls back to guest type delay
{"node-1", "Node", "temperature", 120}, // node metric override {"node-1", "Node", "temperature", 120}, // node metric override
{"node-1", "Node", "cpu", 60}, // node type delay {"node-1", "Node", "cpu", 60}, // node type delay
{"storage-1", "storage", "usage", 90}, // storage type delay {"storage-1", "storage", "usage", 90}, // storage type delay
{"unknown", "unknown", "cpu", 20}, // global default metric override {"unknown", "unknown", "cpu", 20}, // global default metric override
{"unknown", "unknown", "disk", 20}, {"unknown", "unknown", "disk", 20},
} }

52
internal/api/config_handlers_host_test.go
View file

@ -6,19 +6,19 @@ func TestNormalizeNodeHost(t *testing.T) {
tests := []struct { tests := []struct {
name string name string
rawHost string rawHost string
nodeType string nodeType string
want string want string
}{ }{
{ {
name: "adds default port to explicit https without port", name: "adds default port to explicit https without port",
rawHost: "https://example.com", rawHost: "https://example.com",
nodeType: "pve", nodeType: "pve",
want: "https://example.com:8006", want: "https://example.com:8006",
}, },
{ {
name: "adds default port for bare pve host", name: "adds default port for bare pve host",
rawHost: "pve.lan", rawHost: "pve.lan",
nodeType: "pve", nodeType: "pve",
want: "https://pve.lan:8006", want: "https://pve.lan:8006",
}, },
{ {
@ -39,19 +39,19 @@ func TestNormalizeNodeHost(t *testing.T) {
nodeType: "pmg", nodeType: "pmg",
want: "https://[2001:db8::1]:8006", want: "https://[2001:db8::1]:8006",
}, },
{ {
name: "drops path segments", name: "drops path segments",
rawHost: "https://example.com/api", rawHost: "https://example.com/api",
nodeType: "pve", nodeType: "pve",
want: "https://example.com:8006", want: "https://example.com:8006",
}, },
{ {
name: "adds default port to explicit http scheme", name: "adds default port to explicit http scheme",
rawHost: "http://example.com", rawHost: "http://example.com",
nodeType: "pve", nodeType: "pve",
want: "http://example.com:8006", want: "http://example.com:8006",
}, },
} }
for _, tt := range tests { for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) { t.Run(tt.name, func(t *testing.T) {

4
internal/api/csrf_store.go
View file

@ -25,11 +25,11 @@ type CSRFToken struct {
type CSRFTokenStore struct { type CSRFTokenStore struct {
tokens map[string]*CSRFToken tokens map[string]*CSRFToken
mu sync.RWMutex mu sync.RWMutex
saveMu sync.Mutex // Serializes disk writes to prevent save corruption saveMu sync.Mutex // Serializes disk writes to prevent save corruption
dataPath string dataPath string
saveTicker *time.Ticker saveTicker *time.Ticker
stopChan chan bool stopChan chan bool
stopOnce sync.Once // Ensures Stop() can only close channel once stopOnce sync.Once // Ensures Stop() can only close channel once
} }
func csrfSessionKey(sessionID string) string { func csrfSessionKey(sessionID string) string {

1
internal/api/docker_metadata.go
View file

@ -159,4 +159,5 @@ func (h *DockerMetadataHandler) HandleDeleteMetadata(w http.ResponseWriter, r *h
w.WriteHeader(http.StatusNoContent) w.WriteHeader(http.StatusNoContent)
} }
// test comment // test comment

14
internal/api/host_agents_test.go
View file

@ -22,12 +22,12 @@ func TestHandleLookupByIDSuccess(t *testing.T) {
lastSeen := time.Now().UTC() lastSeen := time.Now().UTC()
handler := newHostAgentHandlerForTests(t, models.Host{ handler := newHostAgentHandlerForTests(t, models.Host{
ID: hostID, ID: hostID,
Hostname: "host.local", Hostname: "host.local",
DisplayName:"Host Local", DisplayName: "Host Local",
Status: "online", Status: "online",
TokenID: tokenID, TokenID: tokenID,
LastSeen: lastSeen, LastSeen: lastSeen,
}) })
req := httptest.NewRequest(http.MethodGet, "/api/agents/host/lookup?id="+hostID, nil) req := httptest.NewRequest(http.MethodGet, "/api/agents/host/lookup?id="+hostID, nil)
@ -42,7 +42,7 @@ func TestHandleLookupByIDSuccess(t *testing.T) {
var resp struct { var resp struct {
Success bool `json:"success"` Success bool `json:"success"`
Host struct { Host struct {
ID string `json:"id"` ID string `json:"id"`
Hostname string `json:"hostname"` Hostname string `json:"hostname"`
Status string `json:"status"` Status string `json:"status"`

24
internal/api/types.go
View file

@ -81,18 +81,18 @@ type ConfigResponse struct {
// NodeConfig represents a node configuration // NodeConfig represents a node configuration
type NodeConfig struct { type NodeConfig struct {
ID string `json:"id"` ID string `json:"id"`
Name string `json:"name"` Name string `json:"name"`
Type string `json:"type"` Type string `json:"type"`
Address string `json:"address"` Address string `json:"address"`
Port int `json:"port,omitempty"` Port int `json:"port,omitempty"`
Username string `json:"username,omitempty"` Username string `json:"username,omitempty"`
HasPassword bool `json:"hasPassword"` HasPassword bool `json:"hasPassword"`
HasToken bool `json:"hasToken"` HasToken bool `json:"hasToken"`
SkipTLS bool `json:"skipTLS,omitempty"` SkipTLS bool `json:"skipTLS,omitempty"`
TemperatureMonitoringEnabled *bool `json:"temperatureMonitoringEnabled,omitempty"` TemperatureMonitoringEnabled *bool `json:"temperatureMonitoringEnabled,omitempty"`
Tags []string `json:"tags,omitempty"` Tags []string `json:"tags,omitempty"`
Metadata map[string]string `json:"metadata,omitempty"` Metadata map[string]string `json:"metadata,omitempty"`
} }
// SettingsConfig represents application settings // SettingsConfig represents application settings

42
internal/logging/logging.go
View file

@ -28,13 +28,13 @@ const (
// Config controls logger initialization. // Config controls logger initialization.
type Config struct { type Config struct {
Format string // "json", "console", or "auto" Format string // "json", "console", or "auto"
Level string // "debug", "info", "warn", "error" Level string // "debug", "info", "warn", "error"
Component string // optional component name Component string // optional component name
FilePath string // optional log file path FilePath string // optional log file path
MaxSizeMB int // rotate after this size (MB) MaxSizeMB int // rotate after this size (MB)
MaxAgeDays int // keep rotated logs for this many days MaxAgeDays int // keep rotated logs for this many days
Compress bool // gzip rotated logs Compress bool // gzip rotated logs
} }
// Option customizes logger construction. // Option customizes logger construction.
@ -271,16 +271,16 @@ func collectOptions(opts ...Option) options {
} }
func parseLevel(level string) zerolog.Level { func parseLevel(level string) zerolog.Level {
switch strings.ToLower(strings.TrimSpace(level)) { switch strings.ToLower(strings.TrimSpace(level)) {
case "debug": case "debug":
return zerolog.DebugLevel return zerolog.DebugLevel
case "warn": case "warn":
return zerolog.WarnLevel return zerolog.WarnLevel
case "error": case "error":
return zerolog.ErrorLevel return zerolog.ErrorLevel
default: default:
return zerolog.InfoLevel return zerolog.InfoLevel
} }
} }
func isValidLevel(level string) bool { func isValidLevel(level string) bool {
@ -310,10 +310,10 @@ func selectWriter(format string) io.Writer {
} }
func newConsoleWriter(out io.Writer) io.Writer { func newConsoleWriter(out io.Writer) io.Writer {
return zerolog.ConsoleWriter{ return zerolog.ConsoleWriter{
Out: out, Out: out,
TimeFormat: defaultTimeFmt, TimeFormat: defaultTimeFmt,
} }
} }
func isTerminal(file *os.File) bool { func isTerminal(file *os.File) bool {

30
internal/logging/logging_test.go
View file

@ -222,14 +222,14 @@ func TestContextHelpersWithRequestID(t *testing.T) {
t.Fatalf("expected stored request id %s, got %s", generated, got) t.Fatalf("expected stored request id %s, got %s", generated, got)
} }
var buf bytes.Buffer var buf bytes.Buffer
logger := New("api", WithWriter(&buf)) logger := New("api", WithWriter(&buf))
ctx = WithLogger(ctx, logger) ctx = WithLogger(ctx, logger)
info := FromContext(ctx) info := FromContext(ctx)
info.Info().Msg("ctx-log") info.Info().Msg("ctx-log")
event := readJSONLine(t, &buf) event := readJSONLine(t, &buf)
if event["request_id"] != generated { if event["request_id"] != generated {
t.Fatalf("expected request_id %s, got %v", generated, event["request_id"]) t.Fatalf("expected request_id %s, got %v", generated, event["request_id"])
} }
@ -299,16 +299,16 @@ func TestFromContextWithoutRequestID(t *testing.T) {
Level: "info", Level: "info",
}) })
var buf bytes.Buffer var buf bytes.Buffer
mu.Lock() mu.Lock()
baseLogger = zerolog.New(&buf).With().Timestamp().Logger() baseLogger = zerolog.New(&buf).With().Timestamp().Logger()
baseWriter = &buf baseWriter = &buf
baseComponent = "" baseComponent = ""
log.Logger = baseLogger log.Logger = baseLogger
mu.Unlock() mu.Unlock()
base := FromContext(context.Background()) base := FromContext(context.Background())
base.Info().Msg("no-request") base.Info().Msg("no-request")
event := readJSONLine(t, &buf) event := readJSONLine(t, &buf)
if _, ok := event["request_id"]; ok { if _, ok := event["request_id"]; ok {

72
internal/models/converters_test.go
View file

@ -181,28 +181,28 @@ func TestVMToFrontend(t *testing.T) {
lastBackup := now.Add(-24 * time.Hour) lastBackup := now.Add(-24 * time.Hour)
vm := VM{ vm := VM{
ID: "vm-100", ID: "vm-100",
VMID: 100, VMID: 100,
Name: "test-vm", Name: "test-vm",
Node: "pve1", Node: "pve1",
Instance: "default", Instance: "default",
Status: "running", Status: "running",
Type: "qemu", Type: "qemu",
CPU: 0.15, CPU: 0.15,
CPUs: 4, CPUs: 4,
Memory: Memory{Total: 8000000000, Used: 4000000000}, Memory: Memory{Total: 8000000000, Used: 4000000000},
Disk: Disk{Total: 100000000000, Used: 50000000000}, Disk: Disk{Total: 100000000000, Used: 50000000000},
NetworkIn: 1000000, NetworkIn: 1000000,
NetworkOut: 500000, NetworkOut: 500000,
DiskRead: 100000, DiskRead: 100000,
DiskWrite: 50000, DiskWrite: 50000,
Uptime: 3600, Uptime: 3600,
Tags: []string{"production", "web"}, Tags: []string{"production", "web"},
LastSeen: now, LastSeen: now,
LastBackup: lastBackup, LastBackup: lastBackup,
IPAddresses: []string{"192.168.1.50", "10.0.0.50"}, IPAddresses: []string{"192.168.1.50", "10.0.0.50"},
OSName: "Ubuntu", OSName: "Ubuntu",
OSVersion: "22.04", OSVersion: "22.04",
} }
frontend := vm.ToFrontend() frontend := vm.ToFrontend()
@ -268,21 +268,21 @@ func TestContainerToFrontend(t *testing.T) {
now := time.Now() now := time.Now()
container := Container{ container := Container{
ID: "ct-101", ID: "ct-101",
VMID: 101, VMID: 101,
Name: "test-ct", Name: "test-ct",
Node: "pve1", Node: "pve1",
Status: "running", Status: "running",
Type: "lxc", Type: "lxc",
CPU: 0.10, CPU: 0.10,
CPUs: 2, CPUs: 2,
Memory: Memory{Total: 4000000000, Used: 2000000000}, Memory: Memory{Total: 4000000000, Used: 2000000000},
Disk: Disk{Total: 50000000000, Used: 25000000000}, Disk: Disk{Total: 50000000000, Used: 25000000000},
NetworkIn: 500000, NetworkIn: 500000,
NetworkOut: 250000, NetworkOut: 250000,
Uptime: 7200, Uptime: 7200,
Tags: []string{"dev", "database"}, Tags: []string{"dev", "database"},
LastSeen: now, LastSeen: now,
IPAddresses: []string{"192.168.1.51"}, IPAddresses: []string{"192.168.1.51"},
} }

96
internal/models/models_frontend.go
View file

@ -5,31 +5,31 @@ package models
// NodeFrontend represents a Node with frontend-friendly field names // NodeFrontend represents a Node with frontend-friendly field names
type NodeFrontend struct { type NodeFrontend struct {
ID string `json:"id"` ID string `json:"id"`
Node string `json:"node"` // Maps to Name Node string `json:"node"` // Maps to Name
Name string `json:"name"` Name string `json:"name"`
DisplayName string `json:"displayName"` DisplayName string `json:"displayName"`
Instance string `json:"instance"` Instance string `json:"instance"`
Host string `json:"host,omitempty"` Host string `json:"host,omitempty"`
Status string `json:"status"` Status string `json:"status"`
Type string `json:"type"` Type string `json:"type"`
CPU float64 `json:"cpu"` CPU float64 `json:"cpu"`
Memory *Memory `json:"memory,omitempty"` // Full memory object with usage percentage Memory *Memory `json:"memory,omitempty"` // Full memory object with usage percentage
Mem int64 `json:"mem"` // Maps to Memory.Used (kept for backward compat) Mem int64 `json:"mem"` // Maps to Memory.Used (kept for backward compat)
MaxMem int64 `json:"maxmem"` // Maps to Memory.Total (kept for backward compat) MaxMem int64 `json:"maxmem"` // Maps to Memory.Total (kept for backward compat)
Disk *Disk `json:"disk,omitempty"` // Full disk object with usage percentage Disk *Disk `json:"disk,omitempty"` // Full disk object with usage percentage
MaxDisk int64 `json:"maxdisk"` // Maps to Disk.Total (kept for backward compat) MaxDisk int64 `json:"maxdisk"` // Maps to Disk.Total (kept for backward compat)
Uptime int64 `json:"uptime"` Uptime int64 `json:"uptime"`
LoadAverage []float64 `json:"loadAverage"` LoadAverage []float64 `json:"loadAverage"`
KernelVersion string `json:"kernelVersion"` KernelVersion string `json:"kernelVersion"`
PVEVersion string `json:"pveVersion"` PVEVersion string `json:"pveVersion"`
CPUInfo CPUInfo `json:"cpuInfo"` CPUInfo CPUInfo `json:"cpuInfo"`
Temperature *Temperature `json:"temperature,omitempty"` // CPU/NVMe temperatures Temperature *Temperature `json:"temperature,omitempty"` // CPU/NVMe temperatures
LastSeen int64 `json:"lastSeen"` // Unix timestamp LastSeen int64 `json:"lastSeen"` // Unix timestamp
ConnectionHealth string `json:"connectionHealth"` ConnectionHealth string `json:"connectionHealth"`
IsClusterMember bool `json:"isClusterMember,omitempty"` IsClusterMember bool `json:"isClusterMember,omitempty"`
ClusterName string `json:"clusterName,omitempty"` ClusterName string `json:"clusterName,omitempty"`
TemperatureMonitoringEnabled *bool `json:"temperatureMonitoringEnabled,omitempty"` // Per-node temperature monitoring override TemperatureMonitoringEnabled *bool `json:"temperatureMonitoringEnabled,omitempty"` // Per-node temperature monitoring override
} }
// VMFrontend represents a VM with frontend-friendly field names // VMFrontend represents a VM with frontend-friendly field names
@ -421,27 +421,27 @@ type ReplicationJobFrontend struct {
// StateFrontend represents the state with frontend-friendly field names // StateFrontend represents the state with frontend-friendly field names
type StateFrontend struct { type StateFrontend struct {
Nodes []NodeFrontend `json:"nodes"` Nodes []NodeFrontend `json:"nodes"`
VMs []VMFrontend `json:"vms"` VMs []VMFrontend `json:"vms"`
Containers []ContainerFrontend `json:"containers"` Containers []ContainerFrontend `json:"containers"`
DockerHosts []DockerHostFrontend `json:"dockerHosts"` DockerHosts []DockerHostFrontend `json:"dockerHosts"`
RemovedDockerHosts []RemovedDockerHostFrontend `json:"removedDockerHosts"` RemovedDockerHosts []RemovedDockerHostFrontend `json:"removedDockerHosts"`
Hosts []HostFrontend `json:"hosts"` Hosts []HostFrontend `json:"hosts"`
Storage []StorageFrontend `json:"storage"` Storage []StorageFrontend `json:"storage"`
CephClusters []CephClusterFrontend `json:"cephClusters"` CephClusters []CephClusterFrontend `json:"cephClusters"`
PhysicalDisks []PhysicalDisk `json:"physicalDisks"` PhysicalDisks []PhysicalDisk `json:"physicalDisks"`
PBS []PBSInstance `json:"pbs"` // Keep as is PBS []PBSInstance `json:"pbs"` // Keep as is
PMG []PMGInstance `json:"pmg"` PMG []PMGInstance `json:"pmg"`
PBSBackups []PBSBackup `json:"pbsBackups"` PBSBackups []PBSBackup `json:"pbsBackups"`
PMGBackups []PMGBackup `json:"pmgBackups"` PMGBackups []PMGBackup `json:"pmgBackups"`
Backups Backups `json:"backups"` Backups Backups `json:"backups"`
ReplicationJobs []ReplicationJobFrontend `json:"replicationJobs"` ReplicationJobs []ReplicationJobFrontend `json:"replicationJobs"`
ActiveAlerts []Alert `json:"activeAlerts"` // Active alerts ActiveAlerts []Alert `json:"activeAlerts"` // Active alerts
Metrics map[string]any `json:"metrics"` // Empty object for now Metrics map[string]any `json:"metrics"` // Empty object for now
PVEBackups PVEBackups `json:"pveBackups"` // Keep as is PVEBackups PVEBackups `json:"pveBackups"` // Keep as is
Performance map[string]any `json:"performance"` // Empty object for now Performance map[string]any `json:"performance"` // Empty object for now
ConnectionHealth map[string]bool `json:"connectionHealth"` // Keep as is ConnectionHealth map[string]bool `json:"connectionHealth"` // Keep as is
Stats map[string]any `json:"stats"` // Empty object for now Stats map[string]any `json:"stats"` // Empty object for now
LastUpdate int64 `json:"lastUpdate"` // Unix timestamp LastUpdate int64 `json:"lastUpdate"` // Unix timestamp
TemperatureMonitoringEnabled bool `json:"temperatureMonitoringEnabled"` // Global temperature monitoring setting TemperatureMonitoringEnabled bool `json:"temperatureMonitoringEnabled"` // Global temperature monitoring setting
} }

86
internal/models/state_snapshot.go
View file

@ -4,25 +4,25 @@ import "time"
// StateSnapshot represents a snapshot of the state without mutex // StateSnapshot represents a snapshot of the state without mutex
type StateSnapshot struct { type StateSnapshot struct {
Nodes []Node `json:"nodes"` Nodes []Node `json:"nodes"`
VMs []VM `json:"vms"` VMs []VM `json:"vms"`
Containers []Container `json:"containers"` Containers []Container `json:"containers"`
DockerHosts []DockerHost `json:"dockerHosts"` DockerHosts []DockerHost `json:"dockerHosts"`
RemovedDockerHosts []RemovedDockerHost `json:"removedDockerHosts"` RemovedDockerHosts []RemovedDockerHost `json:"removedDockerHosts"`
Hosts []Host `json:"hosts"` Hosts []Host `json:"hosts"`
Storage []Storage `json:"storage"` Storage []Storage `json:"storage"`
CephClusters []CephCluster `json:"cephClusters"` CephClusters []CephCluster `json:"cephClusters"`
PhysicalDisks []PhysicalDisk `json:"physicalDisks"` PhysicalDisks []PhysicalDisk `json:"physicalDisks"`
PBSInstances []PBSInstance `json:"pbs"` PBSInstances []PBSInstance `json:"pbs"`
PMGInstances []PMGInstance `json:"pmg"` PMGInstances []PMGInstance `json:"pmg"`
PBSBackups []PBSBackup `json:"pbsBackups"` PBSBackups []PBSBackup `json:"pbsBackups"`
PMGBackups []PMGBackup `json:"pmgBackups"` PMGBackups []PMGBackup `json:"pmgBackups"`
Backups Backups `json:"backups"` Backups Backups `json:"backups"`
ReplicationJobs []ReplicationJob `json:"replicationJobs"` ReplicationJobs []ReplicationJob `json:"replicationJobs"`
Metrics []Metric `json:"metrics"` Metrics []Metric `json:"metrics"`
PVEBackups PVEBackups `json:"pveBackups"` PVEBackups PVEBackups `json:"pveBackups"`
Performance Performance `json:"performance"` Performance Performance `json:"performance"`
ConnectionHealth map[string]bool `json:"connectionHealth"` ConnectionHealth map[string]bool `json:"connectionHealth"`
Stats Stats `json:"stats"` Stats Stats `json:"stats"`
ActiveAlerts []Alert `json:"activeAlerts"` ActiveAlerts []Alert `json:"activeAlerts"`
RecentlyResolved []ResolvedAlert `json:"recentlyResolved"` RecentlyResolved []ResolvedAlert `json:"recentlyResolved"`
@ -63,11 +63,11 @@ func (s *State) GetSnapshot() StateSnapshot {
PBS: pbsBackups, PBS: pbsBackups,
PMG: pmgBackups, PMG: pmgBackups,
}, },
ReplicationJobs: append([]ReplicationJob{}, s.ReplicationJobs...), ReplicationJobs: append([]ReplicationJob{}, s.ReplicationJobs...),
Metrics: append([]Metric{}, s.Metrics...), Metrics: append([]Metric{}, s.Metrics...),
PVEBackups: pveBackups, PVEBackups: pveBackups,
Performance: s.Performance, Performance: s.Performance,
ConnectionHealth: make(map[string]bool), ConnectionHealth: make(map[string]bool),
Stats: s.Stats, Stats: s.Stats,
ActiveAlerts: append([]Alert{}, s.ActiveAlerts...), ActiveAlerts: append([]Alert{}, s.ActiveAlerts...),
RecentlyResolved: append([]ResolvedAlert{}, s.RecentlyResolved...), RecentlyResolved: append([]ResolvedAlert{}, s.RecentlyResolved...),
@ -136,25 +136,25 @@ func (s StateSnapshot) ToFrontend() StateFrontend {
} }
return StateFrontend{ return StateFrontend{
Nodes: nodes, Nodes: nodes,
VMs: vms, VMs: vms,
Containers: containers, Containers: containers,
DockerHosts: dockerHosts, DockerHosts: dockerHosts,
RemovedDockerHosts: removedDockerHosts, RemovedDockerHosts: removedDockerHosts,
Hosts: hosts, Hosts: hosts,
Storage: storage, Storage: storage,
CephClusters: cephClusters, CephClusters: cephClusters,
PhysicalDisks: s.PhysicalDisks, PhysicalDisks: s.PhysicalDisks,
PBS: s.PBSInstances, PBS: s.PBSInstances,
PMG: s.PMGInstances, PMG: s.PMGInstances,
PBSBackups: s.PBSBackups, PBSBackups: s.PBSBackups,
PMGBackups: s.PMGBackups, PMGBackups: s.PMGBackups,
Backups: s.Backups, Backups: s.Backups,
ReplicationJobs: replicationJobs, ReplicationJobs: replicationJobs,
ActiveAlerts: s.ActiveAlerts, ActiveAlerts: s.ActiveAlerts,
Metrics: make(map[string]any), Metrics: make(map[string]any),
PVEBackups: s.PVEBackups, PVEBackups: s.PVEBackups,
Performance: make(map[string]any), Performance: make(map[string]any),
ConnectionHealth: s.ConnectionHealth, ConnectionHealth: s.ConnectionHealth,
Stats: make(map[string]any), Stats: make(map[string]any),
LastUpdate: s.LastUpdate.Unix() * 1000, // JavaScript timestamp LastUpdate: s.LastUpdate.Unix() * 1000, // JavaScript timestamp

2
internal/monitoring/backoff_test.go
View file

@ -77,7 +77,7 @@ func TestBackoffConfig_NextDelay(t *testing.T) {
Max: 5 * time.Minute, Max: 5 * time.Minute,
}, },
attempt: 0, attempt: 0,
rng: 0.5, // neutral jitter rng: 0.5, // neutral jitter
wantMin: 8 * time.Second, // 10s * (1 - 0.2) wantMin: 8 * time.Second, // 10s * (1 - 0.2)
wantMax: 12 * time.Second, // 10s * (1 + 0.2) wantMax: 12 * time.Second, // 10s * (1 + 0.2)
}, },

24
internal/monitoring/circuit_breaker.go
View file

@ -23,8 +23,8 @@ type circuitBreaker struct {
maxDelay time.Duration maxDelay time.Duration
openThreshold int openThreshold int
halfOpenWindow time.Duration halfOpenWindow time.Duration
stateSince time.Time stateSince time.Time
lastTransition time.Time lastTransition time.Time
} }
func newCircuitBreaker(openThreshold int, retryInterval, maxDelay, halfOpenWindow time.Duration) *circuitBreaker { func newCircuitBreaker(openThreshold int, retryInterval, maxDelay, halfOpenWindow time.Duration) *circuitBreaker {
@ -40,16 +40,16 @@ func newCircuitBreaker(openThreshold int, retryInterval, maxDelay, halfOpenWindo
if halfOpenWindow <= 0 { if halfOpenWindow <= 0 {
halfOpenWindow = 30 * time.Second halfOpenWindow = 30 * time.Second
} }
now := time.Now() now := time.Now()
return &circuitBreaker{ return &circuitBreaker{
state: breakerClosed, state: breakerClosed,
retryInterval: retryInterval, retryInterval: retryInterval,
maxDelay: maxDelay, maxDelay: maxDelay,
openThreshold: openThreshold, openThreshold: openThreshold,
halfOpenWindow: halfOpenWindow, halfOpenWindow: halfOpenWindow,
stateSince: now, stateSince: now,
lastTransition: now, lastTransition: now,
} }
} }
func (b *circuitBreaker) allow(now time.Time) bool { func (b *circuitBreaker) allow(now time.Time) bool {

2
internal/monitoring/circuit_breaker_test.go
View file

@ -272,7 +272,7 @@ func TestCircuitBreaker_RetryIntervalBackoff(t *testing.T) {
// All subsequent failures should keep it at maxDelay // All subsequent failures should keep it at maxDelay
for i := 0; i < 5; i++ { for i := 0; i < 5; i++ {
cb.allow(now.Add(time.Duration(42+i*2)*time.Minute)) cb.allow(now.Add(time.Duration(42+i*2) * time.Minute))
cb.recordFailure(now.Add(time.Duration(42+i*2) * time.Minute)) cb.recordFailure(now.Add(time.Duration(42+i*2) * time.Minute))
} }

22
internal/monitoring/integration_integration_test.go
View file

@ -163,11 +163,11 @@ func TestAdaptiveSchedulerIntegration(t *testing.T) {
t.Fatalf("expected transient instance to recover with successes, got 0") t.Fatalf("expected transient instance to recover with successes, got 0")
} }
dlqKeys := map[string]struct{}{} dlqKeys := map[string]struct{}{}
for _, task := range report.Health.DeadLetter.Tasks { for _, task := range report.Health.DeadLetter.Tasks {
dlqKeys[instanceKey(task.Type, task.Instance)] = struct{}{} dlqKeys[instanceKey(task.Type, task.Instance)] = struct{}{}
} }
if len(report.Health.Breakers) > len(dlqKeys) { if len(report.Health.Breakers) > len(dlqKeys) {
t.Fatalf("unexpected number of circuit breaker entries: got %d want <= %d", len(report.Health.Breakers), len(dlqKeys)) t.Fatalf("unexpected number of circuit breaker entries: got %d want <= %d", len(report.Health.Breakers), len(dlqKeys))
} }
for _, breaker := range report.Health.Breakers { for _, breaker := range report.Health.Breakers {
@ -218,12 +218,12 @@ func TestAdaptiveSchedulerSoak(t *testing.T) {
t.Skip("skipping soak test (enable with -soak or HARNESS_SOAK_MINUTES)") t.Skip("skipping soak test (enable with -soak or HARNESS_SOAK_MINUTES)")
} }
minutes := 15 minutes := 15
if minutesEnv != "" { if minutesEnv != "" {
if parsed, err := strconv.Atoi(minutesEnv); err == nil && parsed > 0 { if parsed, err := strconv.Atoi(minutesEnv); err == nil && parsed > 0 {
minutes = parsed minutes = parsed
} }
} }
duration := time.Duration(minutes) * time.Minute duration := time.Duration(minutes) * time.Minute
warmup := 2 * time.Minute warmup := 2 * time.Minute

44
internal/monitoring/scheduler.go
View file

@ -52,14 +52,14 @@ type IntervalRequest struct {
// InstanceDescriptor describes a monitored endpoint for scheduling purposes. // InstanceDescriptor describes a monitored endpoint for scheduling purposes.
type InstanceDescriptor struct { type InstanceDescriptor struct {
Name string Name string
Type InstanceType Type InstanceType
LastSuccess time.Time LastSuccess time.Time
LastFailure time.Time LastFailure time.Time
LastScheduled time.Time LastScheduled time.Time
LastInterval time.Duration LastInterval time.Duration
ErrorCount int ErrorCount int
Metadata map[string]any Metadata map[string]any
} }
// ScheduledTask represents a single polling opportunity planned by the scheduler. // ScheduledTask represents a single polling opportunity planned by the scheduler.
@ -90,13 +90,13 @@ func DefaultSchedulerConfig() SchedulerConfig {
// AdaptiveScheduler orchestrates poll execution plans using pluggable scoring strategies. // AdaptiveScheduler orchestrates poll execution plans using pluggable scoring strategies.
type AdaptiveScheduler struct { type AdaptiveScheduler struct {
cfg SchedulerConfig cfg SchedulerConfig
staleness StalenessSource staleness StalenessSource
interval IntervalSelector interval IntervalSelector
enqueuer TaskEnqueuer enqueuer TaskEnqueuer
mu sync.RWMutex mu sync.RWMutex
lastPlan map[string]ScheduledTask lastPlan map[string]ScheduledTask
} }
// NewAdaptiveScheduler constructs a scheduler with safe defaults. // NewAdaptiveScheduler constructs a scheduler with safe defaults.
@ -110,15 +110,15 @@ func NewAdaptiveScheduler(cfg SchedulerConfig, staleness StalenessSource, interv
if cfg.MaxInterval <= 0 || cfg.MaxInterval < cfg.MinInterval { if cfg.MaxInterval <= 0 || cfg.MaxInterval < cfg.MinInterval {
cfg.MaxInterval = DefaultSchedulerConfig().MaxInterval cfg.MaxInterval = DefaultSchedulerConfig().MaxInterval
} }
if staleness == nil { if staleness == nil {
staleness = noopStalenessSource{} staleness = noopStalenessSource{}
} }
if interval == nil { if interval == nil {
interval = newAdaptiveIntervalSelector(cfg) interval = newAdaptiveIntervalSelector(cfg)
} }
if enqueuer == nil { if enqueuer == nil {
enqueuer = noopTaskEnqueuer{} enqueuer = noopTaskEnqueuer{}
} }
return &AdaptiveScheduler{ return &AdaptiveScheduler{
cfg: cfg, cfg: cfg,

266
internal/monitoring/staleness_tracker.go
View file

@ -1,204 +1,204 @@
package monitoring package monitoring
import ( import (
"crypto/sha1" "crypto/sha1"
"encoding/hex" "encoding/hex"
"sync" "sync"
"time" "time"
) )
// FreshnessSnapshot captures the most recent freshness metadata available for a target instance. // FreshnessSnapshot captures the most recent freshness metadata available for a target instance.
type FreshnessSnapshot struct { type FreshnessSnapshot struct {
InstanceType InstanceType InstanceType InstanceType
Instance string Instance string
LastSuccess time.Time LastSuccess time.Time
LastError time.Time LastError time.Time
LastMutated time.Time LastMutated time.Time
ChangeHash string ChangeHash string
} }
// StalenessTracker maintains freshness metadata and exposes normalized staleness scores. // StalenessTracker maintains freshness metadata and exposes normalized staleness scores.
type StalenessTracker struct { type StalenessTracker struct {
mu sync.RWMutex mu sync.RWMutex
entries map[string]FreshnessSnapshot entries map[string]FreshnessSnapshot
baseTTL time.Duration baseTTL time.Duration
maxStale time.Duration maxStale time.Duration
metrics *PollMetrics metrics *PollMetrics
} }
// NewStalenessTracker builds a tracker wired to poll metrics for last-success signal and using default parameters. // NewStalenessTracker builds a tracker wired to poll metrics for last-success signal and using default parameters.
func NewStalenessTracker(metrics *PollMetrics) *StalenessTracker { func NewStalenessTracker(metrics *PollMetrics) *StalenessTracker {
return &StalenessTracker{ return &StalenessTracker{
entries: make(map[string]FreshnessSnapshot), entries: make(map[string]FreshnessSnapshot),
baseTTL: 10 * time.Second, baseTTL: 10 * time.Second,
maxStale: 5 * time.Minute, maxStale: 5 * time.Minute,
metrics: metrics, metrics: metrics,
} }
} }
// SetBounds allows overriding score decay windows. // SetBounds allows overriding score decay windows.
func (t *StalenessTracker) SetBounds(baseTTL, maxStale time.Duration) { func (t *StalenessTracker) SetBounds(baseTTL, maxStale time.Duration) {
t.mu.Lock() t.mu.Lock()
defer t.mu.Unlock() defer t.mu.Unlock()
if baseTTL > 0 { if baseTTL > 0 {
t.baseTTL = baseTTL t.baseTTL = baseTTL
} }
if maxStale > 0 { if maxStale > 0 {
t.maxStale = maxStale t.maxStale = maxStale
} }
} }
// UpdateSuccess records a successful poll along with a change hash derived from the payload. // UpdateSuccess records a successful poll along with a change hash derived from the payload.
func (t *StalenessTracker) UpdateSuccess(instanceType InstanceType, instance string, payload []byte) { func (t *StalenessTracker) UpdateSuccess(instanceType InstanceType, instance string, payload []byte) {
if t == nil { if t == nil {
return return
} }
now := time.Now() now := time.Now()
snapshot := FreshnessSnapshot{ snapshot := FreshnessSnapshot{
InstanceType: instanceType, InstanceType: instanceType,
Instance: instance, Instance: instance,
LastSuccess: now, LastSuccess: now,
} }
if len(payload) > 0 { if len(payload) > 0 {
sum := sha1.Sum(payload) sum := sha1.Sum(payload)
snapshot.ChangeHash = hex.EncodeToString(sum[:]) snapshot.ChangeHash = hex.EncodeToString(sum[:])
snapshot.LastMutated = now snapshot.LastMutated = now
} }
t.setSnapshot(snapshot) t.setSnapshot(snapshot)
} }
// UpdateError records the most recent error time for a target instance. // UpdateError records the most recent error time for a target instance.
func (t *StalenessTracker) UpdateError(instanceType InstanceType, instance string) { func (t *StalenessTracker) UpdateError(instanceType InstanceType, instance string) {
if t == nil { if t == nil {
return return
} }
snapshot := FreshnessSnapshot{ snapshot := FreshnessSnapshot{
InstanceType: instanceType, InstanceType: instanceType,
Instance: instance, Instance: instance,
LastError: time.Now(), LastError: time.Now(),
} }
t.mergeSnapshot(snapshot) t.mergeSnapshot(snapshot)
} }
// SetChangeHash updates the change fingerprint without affecting success timestamps. // SetChangeHash updates the change fingerprint without affecting success timestamps.
func (t *StalenessTracker) SetChangeHash(instanceType InstanceType, instance string, payload []byte) { func (t *StalenessTracker) SetChangeHash(instanceType InstanceType, instance string, payload []byte) {
if t == nil || len(payload) == 0 { if t == nil || len(payload) == 0 {
return return
} }
now := time.Now() now := time.Now()
sum := sha1.Sum(payload) sum := sha1.Sum(payload)
hash := hex.EncodeToString(sum[:]) hash := hex.EncodeToString(sum[:])
t.mu.Lock() t.mu.Lock()
defer t.mu.Unlock() defer t.mu.Unlock()
key := trackerKey(instanceType, instance) key := trackerKey(instanceType, instance)
snap := t.entries[key] snap := t.entries[key]
snap.InstanceType = instanceType snap.InstanceType = instanceType
snap.Instance = instance snap.Instance = instance
snap.ChangeHash = hash snap.ChangeHash = hash
snap.LastMutated = now snap.LastMutated = now
t.entries[key] = snap t.entries[key] = snap
} }
// StalenessScore implements the StalenessSource interface and returns a normalized value in [0,1]. // StalenessScore implements the StalenessSource interface and returns a normalized value in [0,1].
func (t *StalenessTracker) StalenessScore(instanceType InstanceType, instance string) (float64, bool) { func (t *StalenessTracker) StalenessScore(instanceType InstanceType, instance string) (float64, bool) {
if t == nil { if t == nil {
return 0, false return 0, false
} }
snap, ok := t.snapshot(instanceType, instance) snap, ok := t.snapshot(instanceType, instance)
if !ok { if !ok {
return 0, false return 0, false
} }
if !snap.LastSuccess.IsZero() && t.metrics != nil { if !snap.LastSuccess.IsZero() && t.metrics != nil {
if ts, ok := t.metrics.lastSuccessFor(string(instanceType), instance); ok { if ts, ok := t.metrics.lastSuccessFor(string(instanceType), instance); ok {
snap.LastSuccess = ts snap.LastSuccess = ts
} }
} }
if snap.LastSuccess.IsZero() { if snap.LastSuccess.IsZero() {
return 1, true return 1, true
} }
now := time.Now() now := time.Now()
age := now.Sub(snap.LastSuccess) age := now.Sub(snap.LastSuccess)
if age <= 0 { if age <= 0 {
return 0, true return 0, true
} }
max := t.maxStale max := t.maxStale
if max <= 0 { if max <= 0 {
max = 5 * time.Minute max = 5 * time.Minute
} }
score := age.Seconds() / max.Seconds() score := age.Seconds() / max.Seconds()
if score > 1 { if score > 1 {
score = 1 score = 1
} }
if score < 0 { if score < 0 {
score = 0 score = 0
} }
return score, true return score, true
} }
func (t *StalenessTracker) setSnapshot(snapshot FreshnessSnapshot) { func (t *StalenessTracker) setSnapshot(snapshot FreshnessSnapshot) {
t.mu.Lock() t.mu.Lock()
defer t.mu.Unlock() defer t.mu.Unlock()
key := trackerKey(snapshot.InstanceType, snapshot.Instance) key := trackerKey(snapshot.InstanceType, snapshot.Instance)
t.entries[key] = snapshot t.entries[key] = snapshot
} }
func (t *StalenessTracker) mergeSnapshot(snapshot FreshnessSnapshot) { func (t *StalenessTracker) mergeSnapshot(snapshot FreshnessSnapshot) {
t.mu.Lock() t.mu.Lock()
defer t.mu.Unlock() defer t.mu.Unlock()
key := trackerKey(snapshot.InstanceType, snapshot.Instance) key := trackerKey(snapshot.InstanceType, snapshot.Instance)
existing := t.entries[key] existing := t.entries[key]
if snapshot.LastSuccess.After(existing.LastSuccess) { if snapshot.LastSuccess.After(existing.LastSuccess) {
existing.LastSuccess = snapshot.LastSuccess existing.LastSuccess = snapshot.LastSuccess
} }
if snapshot.LastError.After(existing.LastError) { if snapshot.LastError.After(existing.LastError) {
existing.LastError = snapshot.LastError existing.LastError = snapshot.LastError
} }
if snapshot.LastMutated.After(existing.LastMutated) { if snapshot.LastMutated.After(existing.LastMutated) {
existing.LastMutated = snapshot.LastMutated existing.LastMutated = snapshot.LastMutated
} }
if snapshot.ChangeHash != "" { if snapshot.ChangeHash != "" {
existing.ChangeHash = snapshot.ChangeHash existing.ChangeHash = snapshot.ChangeHash
} }
existing.InstanceType = snapshot.InstanceType existing.InstanceType = snapshot.InstanceType
existing.Instance = snapshot.Instance existing.Instance = snapshot.Instance
t.entries[key] = existing t.entries[key] = existing
} }
func (t *StalenessTracker) snapshot(instanceType InstanceType, instance string) (FreshnessSnapshot, bool) { func (t *StalenessTracker) snapshot(instanceType InstanceType, instance string) (FreshnessSnapshot, bool) {
t.mu.RLock() t.mu.RLock()
defer t.mu.RUnlock() defer t.mu.RUnlock()
snap, ok := t.entries[trackerKey(instanceType, instance)] snap, ok := t.entries[trackerKey(instanceType, instance)]
return snap, ok return snap, ok
} }
func trackerKey(instanceType InstanceType, instance string) string { func trackerKey(instanceType InstanceType, instance string) string {
return string(instanceType) + "::" + instance return string(instanceType) + "::" + instance
} }
// StalenessSnapshot represents staleness data for a single instance. // StalenessSnapshot represents staleness data for a single instance.
type StalenessSnapshot struct { type StalenessSnapshot struct {
Instance string `json:"instance"` Instance string `json:"instance"`
Type string `json:"type"` Type string `json:"type"`
Score float64 `json:"score"` Score float64 `json:"score"`
LastSuccess time.Time `json:"lastSuccess"` LastSuccess time.Time `json:"lastSuccess"`
LastError time.Time `json:"lastError,omitempty"` LastError time.Time `json:"lastError,omitempty"`
} }
// Snapshot returns a copy of all staleness data for API exposure. // Snapshot returns a copy of all staleness data for API exposure.

8
internal/notifications/queue.go
View file

@ -9,9 +9,9 @@ import (
"sync" "sync"
"time" "time"
_ "github.com/mattn/go-sqlite3"
"github.com/rcourtman/pulse-go-rewrite/internal/alerts" "github.com/rcourtman/pulse-go-rewrite/internal/alerts"
"github.com/rcourtman/pulse-go-rewrite/internal/utils" "github.com/rcourtman/pulse-go-rewrite/internal/utils"
_ "github.com/mattn/go-sqlite3"
"github.com/rs/zerolog/log" "github.com/rs/zerolog/log"
) )
@ -54,9 +54,9 @@ type NotificationQueue struct {
wg sync.WaitGroup wg sync.WaitGroup
processorTicker *time.Ticker processorTicker *time.Ticker
cleanupTicker *time.Ticker cleanupTicker *time.Ticker
notifyChan chan struct{} // Signal when new notifications are added notifyChan chan struct{} // Signal when new notifications are added
processor func(*QueuedNotification) error // Notification processor function processor func(*QueuedNotification) error // Notification processor function
workerSem chan struct{} // Semaphore for limiting concurrent workers workerSem chan struct{} // Semaphore for limiting concurrent workers
} }
// NewNotificationQueue creates a new persistent notification queue // NewNotificationQueue creates a new persistent notification queue

18
internal/sensors/parser_test.go
View file

@ -377,19 +377,19 @@ func TestExtractTempInput(t *testing.T) {
expected: 45.0, expected: 45.0,
}, },
{ {
name: "no input field", name: "no input field",
input: map[string]interface{}{"temp1_max": 100.0}, input: map[string]interface{}{"temp1_max": 100.0},
isNaN: true, isNaN: true,
}, },
{ {
name: "empty map", name: "empty map",
input: map[string]interface{}{}, input: map[string]interface{}{},
isNaN: true, isNaN: true,
}, },
{ {
name: "wrong suffix", name: "wrong suffix",
input: map[string]interface{}{"temp1_max": 45.0}, input: map[string]interface{}{"temp1_max": 45.0},
isNaN: true, isNaN: true,
}, },
} }

2
internal/system/container_test.go
View file

@ -17,7 +17,7 @@ func TestIsHexString(t *testing.T) {
{"xyz", false}, {"xyz", false},
{"123g", false}, {"123g", false},
{"hello-world", false}, {"hello-world", false},
{"12 34", false}, // space {"12 34", false}, // space
{"12\n34", false}, // newline {"12\n34", false}, // newline
} }

20
internal/updates/queue.go
View file

@ -12,12 +12,12 @@ import (
type JobState string type JobState string
const ( const (
JobStateIdle JobState = "idle" JobStateIdle JobState = "idle"
JobStateQueued JobState = "queued" JobStateQueued JobState = "queued"
JobStateRunning JobState = "running" JobStateRunning JobState = "running"
JobStateCompleted JobState = "completed" JobStateCompleted JobState = "completed"
JobStateFailed JobState = "failed" JobStateFailed JobState = "failed"
JobStateCancelled JobState = "cancelled" JobStateCancelled JobState = "cancelled"
) )
// UpdateJob represents a single update job // UpdateJob represents a single update job
@ -34,10 +34,10 @@ type UpdateJob struct {
// UpdateQueue manages the update job queue ensuring only one update runs at a time // UpdateQueue manages the update job queue ensuring only one update runs at a time
type UpdateQueue struct { type UpdateQueue struct {
mu sync.RWMutex mu sync.RWMutex
currentJob *UpdateJob currentJob *UpdateJob
jobHistory []*UpdateJob jobHistory []*UpdateJob
maxHistory int maxHistory int
} }
// NewUpdateQueue creates a new update queue // NewUpdateQueue creates a new update queue

18
internal/updates/updater.go
View file

@ -7,10 +7,10 @@ import (
// UpdateRequest represents a request to update // UpdateRequest represents a request to update
type UpdateRequest struct { type UpdateRequest struct {
Version string Version string
Channel string Channel string
Force bool Force bool
DryRun bool DryRun bool
} }
// UpdatePlan contains information about how an update will be performed // UpdatePlan contains information about how an update will be performed
@ -26,11 +26,11 @@ type UpdatePlan struct {
// UpdateProgress represents progress updates during an update // UpdateProgress represents progress updates during an update
type UpdateProgress struct { type UpdateProgress struct {
Stage string `json:"stage"` Stage string `json:"stage"`
Progress int `json:"progress"` // 0-100 Progress int `json:"progress"` // 0-100
Message string `json:"message"` Message string `json:"message"`
IsComplete bool `json:"isComplete"` IsComplete bool `json:"isComplete"`
Error string `json:"error,omitempty"` Error string `json:"error,omitempty"`
} }
// ProgressCallback is called during update execution // ProgressCallback is called during update execution

6
pkg/agents/docker/report_test.go
View file

@ -66,9 +66,9 @@ func TestReport_JSONMarshal(t *testing.T) {
IntervalSeconds: 30, IntervalSeconds: 30,
}, },
Host: HostInfo{ Host: HostInfo{
Hostname: "docker-host", Hostname: "docker-host",
DockerVersion: "24.0.0", DockerVersion: "24.0.0",
TotalCPU: 8, TotalCPU: 8,
TotalMemoryBytes: 16000000000, TotalMemoryBytes: 16000000000,
}, },
Containers: []Container{ Containers: []Container{

30
pkg/agents/host/report.go
View file

@ -89,24 +89,24 @@ type Sensors struct {
// RAIDArray represents an mdadm RAID array. // RAIDArray represents an mdadm RAID array.
type RAIDArray struct { type RAIDArray struct {
Device string `json:"device"` // e.g., /dev/md0 Device string `json:"device"` // e.g., /dev/md0
Name string `json:"name,omitempty"` // Array name if set Name string `json:"name,omitempty"` // Array name if set
Level string `json:"level"` // RAID level: raid0, raid1, raid5, raid6, raid10 Level string `json:"level"` // RAID level: raid0, raid1, raid5, raid6, raid10
State string `json:"state"` // clean, active, degraded, recovering, resyncing, etc. State string `json:"state"` // clean, active, degraded, recovering, resyncing, etc.
TotalDevices int `json:"totalDevices"` // Total number of devices in array TotalDevices int `json:"totalDevices"` // Total number of devices in array
ActiveDevices int `json:"activeDevices"` // Number of active devices ActiveDevices int `json:"activeDevices"` // Number of active devices
WorkingDevices int `json:"workingDevices"` // Number of working devices WorkingDevices int `json:"workingDevices"` // Number of working devices
FailedDevices int `json:"failedDevices"` // Number of failed devices FailedDevices int `json:"failedDevices"` // Number of failed devices
SpareDevices int `json:"spareDevices"` // Number of spare devices SpareDevices int `json:"spareDevices"` // Number of spare devices
UUID string `json:"uuid,omitempty"` // Array UUID UUID string `json:"uuid,omitempty"` // Array UUID
Devices []RAIDDevice `json:"devices"` // Individual devices in array Devices []RAIDDevice `json:"devices"` // Individual devices in array
RebuildPercent float64 `json:"rebuildPercent"` // Rebuild/resync progress (0-100) RebuildPercent float64 `json:"rebuildPercent"` // Rebuild/resync progress (0-100)
RebuildSpeed string `json:"rebuildSpeed,omitempty"` // Rebuild speed (e.g., "50000K/sec") RebuildSpeed string `json:"rebuildSpeed,omitempty"` // Rebuild speed (e.g., "50000K/sec")
} }
// RAIDDevice represents a single device in a RAID array. // RAIDDevice represents a single device in a RAID array.
type RAIDDevice struct { type RAIDDevice struct {
Device string `json:"device"` // e.g., /dev/sda1 Device string `json:"device"` // e.g., /dev/sda1
State string `json:"state"` // active, spare, faulty, removed State string `json:"state"` // active, spare, faulty, removed
Slot int `json:"slot"` // Position in array (-1 if not applicable) Slot int `json:"slot"` // Position in array (-1 if not applicable)
} }

14
pkg/discovery/envdetect/envdetect.go
View file

@ -83,13 +83,13 @@ type SubnetPhase struct {
// EnvironmentProfile captures detection results and scanning plan. // EnvironmentProfile captures detection results and scanning plan.
type EnvironmentProfile struct { type EnvironmentProfile struct {
Type Environment // Detected environment. Type Environment // Detected environment.
Phases []SubnetPhase // Subnet scanning phases. Phases []SubnetPhase // Subnet scanning phases.
ExtraTargets []net.IP // IPs to always probe. ExtraTargets []net.IP // IPs to always probe.
Policy ScanPolicy // Applied scan policy. Policy ScanPolicy // Applied scan policy.
Confidence float64 // Overall confidence (0.0 - 1.0). Confidence float64 // Overall confidence (0.0 - 1.0).
Warnings []string // Non-fatal detection warnings. Warnings []string // Non-fatal detection warnings.
Metadata map[string]string // Misc metadata (container type, gateway, etc.). Metadata map[string]string // Misc metadata (container type, gateway, etc.).
} }
// DetectEnvironment performs environment detection and returns a profile. // DetectEnvironment performs environment detection and returns a profile.

34
pkg/proxmox/client.go
View file

@ -1839,23 +1839,23 @@ func (a *VMAgentField) UnmarshalJSON(data []byte) error {
// VMStatus represents detailed VM status returned by Proxmox. // VMStatus represents detailed VM status returned by Proxmox.
type VMStatus struct { type VMStatus struct {
Status string `json:"status"` Status string `json:"status"`
CPU float64 `json:"cpu"` CPU float64 `json:"cpu"`
CPUs int `json:"cpus"` CPUs int `json:"cpus"`
Mem uint64 `json:"mem"` Mem uint64 `json:"mem"`
MaxMem uint64 `json:"maxmem"` MaxMem uint64 `json:"maxmem"`
Balloon uint64 `json:"balloon"` Balloon uint64 `json:"balloon"`
BalloonMin uint64 `json:"balloon_min"` BalloonMin uint64 `json:"balloon_min"`
FreeMem uint64 `json:"freemem"` FreeMem uint64 `json:"freemem"`
MemInfo *VMMemInfo `json:"meminfo,omitempty"` MemInfo *VMMemInfo `json:"meminfo,omitempty"`
Disk uint64 `json:"disk"` Disk uint64 `json:"disk"`
MaxDisk uint64 `json:"maxdisk"` MaxDisk uint64 `json:"maxdisk"`
DiskRead uint64 `json:"diskread"` DiskRead uint64 `json:"diskread"`
DiskWrite uint64 `json:"diskwrite"` DiskWrite uint64 `json:"diskwrite"`
NetIn uint64 `json:"netin"` NetIn uint64 `json:"netin"`
NetOut uint64 `json:"netout"` NetOut uint64 `json:"netout"`
Uptime uint64 `json:"uptime"` Uptime uint64 `json:"uptime"`
Agent VMAgentField `json:"agent"` Agent VMAgentField `json:"agent"`
} }
// GetZFSPoolStatus gets the status of ZFS pools on a node // GetZFSPoolStatus gets the status of ZFS pools on a node

2
pkg/proxmox/client_test.go
View file

@ -246,8 +246,6 @@ func TestMemoryStatusUnmarshalFlexibleValues(t *testing.T) {
} }
} }
// TestMemoryStatusEffectiveAvailable_RegressionIssue435 tests the specific scenarios // TestMemoryStatusEffectiveAvailable_RegressionIssue435 tests the specific scenarios
// reported in GitHub issue #435 where memory calculations incorrectly included cache/buffers // reported in GitHub issue #435 where memory calculations incorrectly included cache/buffers
func TestMemoryStatusEffectiveAvailable_RegressionIssue435(t *testing.T) { func TestMemoryStatusEffectiveAvailable_RegressionIssue435(t *testing.T) {