From cdd2249e001b5068a6d7a34ac98675e398355320 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Sun, 7 Dec 2025 23:02:56 +0000 Subject: [PATCH] phase 1: add unified resources to WebSocket state broadcasts - Extended StateFrontend with Resources field containing unified resource data - Added ResourceFrontend and related types for frontend-compatible resource data - Extended ResourceStoreInterface to include GetAll() method - Monitor now injects resources into WebSocket broadcasts - Added helper method getResourcesForBroadcast() to convert resources to frontend format - All existing tests pass This enables the frontend to access unified resources via WebSocket state. --- internal/models/converters.go | 138 +++++++++++++++++++++++++++++ internal/models/models_frontend.go | 74 ++++++++++++++++ internal/monitoring/monitor.go | 121 ++++++++++++++++++++++++- 3 files changed, 330 insertions(+), 3 deletions(-) diff --git a/internal/models/converters.go b/internal/models/converters.go index 5bf5334..e655c28 100644 --- a/internal/models/converters.go +++ b/internal/models/converters.go @@ -732,3 +732,141 @@ func zeroIfNegative(val int64) int64 { } return val } + +// ResourceToFrontend converts a resources.Resource to ResourceFrontend. +// This function is in models package to avoid circular imports. +// It takes individual fields rather than the whole Resource to avoid importing resources package. +type ResourceConvertInput struct { + ID string + Type string + Name string + DisplayName string + PlatformID string + PlatformType string + SourceType string + ParentID string + ClusterID string + Status string + CPU *ResourceMetricInput + Memory *ResourceMetricInput + Disk *ResourceMetricInput + NetworkRX int64 + NetworkTX int64 + HasNetwork bool + Temperature *float64 + Uptime *int64 + Tags []string + Labels map[string]string + LastSeenUnix int64 + Alerts []ResourceAlertInput + Identity *ResourceIdentityInput + PlatformData map[string]any +} + +// ResourceMetricInput represents a metric value for resource conversion. +type ResourceMetricInput struct { + Current float64 + Total *int64 + Used *int64 + Free *int64 +} + +type ResourceAlertInput struct { + ID string + Type string + Level string + Message string + Value float64 + Threshold float64 + StartTimeUnix int64 +} + +type ResourceIdentityInput struct { + Hostname string + MachineID string + IPs []string +} + +// ConvertResourceToFrontend converts input to ResourceFrontend. +func ConvertResourceToFrontend(input ResourceConvertInput) ResourceFrontend { + rf := ResourceFrontend{ + ID: input.ID, + Type: input.Type, + Name: input.Name, + DisplayName: input.DisplayName, + PlatformID: input.PlatformID, + PlatformType: input.PlatformType, + SourceType: input.SourceType, + ParentID: input.ParentID, + ClusterID: input.ClusterID, + Status: input.Status, + Temperature: input.Temperature, + Uptime: input.Uptime, + Tags: input.Tags, + Labels: input.Labels, + LastSeen: input.LastSeenUnix, + PlatformData: input.PlatformData, + } + + // Convert metrics + if input.CPU != nil { + rf.CPU = &ResourceMetricFrontend{ + Current: input.CPU.Current, + Total: input.CPU.Total, + Used: input.CPU.Used, + Free: input.CPU.Free, + } + } + + if input.Memory != nil { + rf.Memory = &ResourceMetricFrontend{ + Current: input.Memory.Current, + Total: input.Memory.Total, + Used: input.Memory.Used, + Free: input.Memory.Free, + } + } + + if input.Disk != nil { + rf.Disk = &ResourceMetricFrontend{ + Current: input.Disk.Current, + Total: input.Disk.Total, + Used: input.Disk.Used, + Free: input.Disk.Free, + } + } + + if input.HasNetwork { + rf.Network = &ResourceNetworkFrontend{ + RXBytes: input.NetworkRX, + TXBytes: input.NetworkTX, + } + } + + // Convert alerts + if len(input.Alerts) > 0 { + rf.Alerts = make([]ResourceAlertFrontend, len(input.Alerts)) + for i, a := range input.Alerts { + rf.Alerts[i] = ResourceAlertFrontend{ + ID: a.ID, + Type: a.Type, + Level: a.Level, + Message: a.Message, + Value: a.Value, + Threshold: a.Threshold, + StartTime: a.StartTimeUnix, + } + } + } + + // Convert identity + if input.Identity != nil { + rf.Identity = &ResourceIdentityFrontend{ + Hostname: input.Identity.Hostname, + MachineID: input.Identity.MachineID, + IPs: input.Identity.IPs, + } + } + + return rf +} diff --git a/internal/models/models_frontend.go b/internal/models/models_frontend.go index 9c6b5ed..a70282e 100644 --- a/internal/models/models_frontend.go +++ b/internal/models/models_frontend.go @@ -446,4 +446,78 @@ type StateFrontend struct { Stats map[string]any `json:"stats"` // Empty object for now LastUpdate int64 `json:"lastUpdate"` // Unix timestamp TemperatureMonitoringEnabled bool `json:"temperatureMonitoringEnabled"` // Global temperature monitoring setting + // Unified resources - the new way to access all monitored entities + Resources []ResourceFrontend `json:"resources,omitempty"` +} + +// ResourceFrontend is the frontend representation of a unified Resource. +// This mirrors resources.Resource but with time.Time converted to Unix milliseconds. +type ResourceFrontend struct { + // Identity + ID string `json:"id"` + Type string `json:"type"` + Name string `json:"name"` + DisplayName string `json:"displayName"` + + // Platform/Source + PlatformID string `json:"platformId"` + PlatformType string `json:"platformType"` + SourceType string `json:"sourceType"` + + // Hierarchy + ParentID string `json:"parentId,omitempty"` + ClusterID string `json:"clusterId,omitempty"` + + // Universal Metrics + Status string `json:"status"` + CPU *ResourceMetricFrontend `json:"cpu,omitempty"` + Memory *ResourceMetricFrontend `json:"memory,omitempty"` + Disk *ResourceMetricFrontend `json:"disk,omitempty"` + Network *ResourceNetworkFrontend `json:"network,omitempty"` + Temperature *float64 `json:"temperature,omitempty"` + Uptime *int64 `json:"uptime,omitempty"` + + // Metadata + Tags []string `json:"tags,omitempty"` + Labels map[string]string `json:"labels,omitempty"` + LastSeen int64 `json:"lastSeen"` // Unix milliseconds + Alerts []ResourceAlertFrontend `json:"alerts,omitempty"` + + // Identity for deduplication + Identity *ResourceIdentityFrontend `json:"identity,omitempty"` + + // Platform-specific data (JSON blob) + PlatformData map[string]any `json:"platformData,omitempty"` +} + +// ResourceMetricFrontend represents a metric value for the frontend. +type ResourceMetricFrontend struct { + Current float64 `json:"current"` + Total *int64 `json:"total,omitempty"` + Used *int64 `json:"used,omitempty"` + Free *int64 `json:"free,omitempty"` +} + +// ResourceNetworkFrontend represents network metrics for the frontend. +type ResourceNetworkFrontend struct { + RXBytes int64 `json:"rxBytes"` + TXBytes int64 `json:"txBytes"` +} + +// ResourceAlertFrontend represents an alert on a resource. +type ResourceAlertFrontend struct { + ID string `json:"id"` + Type string `json:"type"` + Level string `json:"level"` + Message string `json:"message"` + Value float64 `json:"value"` + Threshold float64 `json:"threshold"` + StartTime int64 `json:"startTime"` // Unix milliseconds +} + +// ResourceIdentityFrontend contains identity info for deduplication. +type ResourceIdentityFrontend struct { + Hostname string `json:"hostname,omitempty"` + MachineID string `json:"machineId,omitempty"` + IPs []string `json:"ips,omitempty"` } diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index fee7de3..673e420 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -4,6 +4,7 @@ import ( "context" "crypto/sha1" "encoding/hex" + "encoding/json" stderrors "errors" "fmt" "math" @@ -28,6 +29,7 @@ import ( "github.com/rcourtman/pulse-go-rewrite/internal/mock" "github.com/rcourtman/pulse-go-rewrite/internal/models" "github.com/rcourtman/pulse-go-rewrite/internal/notifications" + "github.com/rcourtman/pulse-go-rewrite/internal/resources" "github.com/rcourtman/pulse-go-rewrite/internal/system" "github.com/rcourtman/pulse-go-rewrite/internal/tempproxy" "github.com/rcourtman/pulse-go-rewrite/internal/types" @@ -80,7 +82,7 @@ type PVEClientInterface interface { GetCephDF(ctx context.Context) (*proxmox.CephDF, error) } -// ResourceStoreInterface provides methods for polling optimization. +// ResourceStoreInterface provides methods for polling optimization and resource access. // When an agent is monitoring a node, we can reduce API polling for that node. type ResourceStoreInterface interface { // ShouldSkipAPIPolling returns true if API polling should be skipped for the hostname @@ -89,6 +91,8 @@ type ResourceStoreInterface interface { // GetPollingRecommendations returns a map of hostname -> polling multiplier. // 0 = skip entirely, 0.5 = half frequency, 1 = normal GetPollingRecommendations() map[string]float64 + // GetAll returns all resources in the store (for WebSocket broadcasts) + GetAll() []resources.Resource } func getNodeDisplayName(instance *config.PVEInstance, nodeName string) string { @@ -3285,7 +3289,10 @@ func (m *Monitor) Start(ctx context.Context, wsHub *websocket.Hub) { Int("physicalDisks", len(state.PhysicalDisks)). Msg("Broadcasting state update (ticker)") // Convert to frontend format before broadcasting (converts time.Time to int64, etc.) - wsHub.BroadcastState(state.ToFrontend()) + frontendState := state.ToFrontend() + // Inject unified resources if resource store is available + frontendState.Resources = m.getResourcesForBroadcast() + wsHub.BroadcastState(frontendState) case <-ctx.Done(): log.Info().Msg("Monitoring loop stopped") @@ -6894,7 +6901,9 @@ func (m *Monitor) SetMockMode(enable bool) { m.mu.RUnlock() if hub != nil { - hub.BroadcastState(m.GetState().ToFrontend()) + frontendState := m.GetState().ToFrontend() + frontendState.Resources = m.getResourcesForBroadcast() + hub.BroadcastState(frontendState) } if !enable && ctx != nil && hub != nil { @@ -7048,6 +7057,112 @@ func (m *Monitor) shouldSkipNodeMetrics(nodeName string) bool { return should } +// getResourcesForBroadcast retrieves all resources from the store and converts them to frontend format. +// Returns nil if no resource store is configured. +func (m *Monitor) getResourcesForBroadcast() []models.ResourceFrontend { + m.mu.RLock() + store := m.resourceStore + m.mu.RUnlock() + + if store == nil { + return nil + } + + allResources := store.GetAll() + if len(allResources) == 0 { + return nil + } + + result := make([]models.ResourceFrontend, len(allResources)) + for i, r := range allResources { + input := models.ResourceConvertInput{ + ID: r.ID, + Type: string(r.Type), + Name: r.Name, + DisplayName: r.DisplayName, + PlatformID: r.PlatformID, + PlatformType: string(r.PlatformType), + SourceType: string(r.SourceType), + ParentID: r.ParentID, + ClusterID: r.ClusterID, + Status: string(r.Status), + Temperature: r.Temperature, + Uptime: r.Uptime, + Tags: r.Tags, + Labels: r.Labels, + LastSeenUnix: r.LastSeen.UnixMilli(), + } + + // Convert metrics + if r.CPU != nil { + input.CPU = &models.ResourceMetricInput{ + Current: r.CPU.Current, + Total: r.CPU.Total, + Used: r.CPU.Used, + Free: r.CPU.Free, + } + } + if r.Memory != nil { + input.Memory = &models.ResourceMetricInput{ + Current: r.Memory.Current, + Total: r.Memory.Total, + Used: r.Memory.Used, + Free: r.Memory.Free, + } + } + if r.Disk != nil { + input.Disk = &models.ResourceMetricInput{ + Current: r.Disk.Current, + Total: r.Disk.Total, + Used: r.Disk.Used, + Free: r.Disk.Free, + } + } + if r.Network != nil { + input.HasNetwork = true + input.NetworkRX = r.Network.RXBytes + input.NetworkTX = r.Network.TXBytes + } + + // Convert alerts + if len(r.Alerts) > 0 { + input.Alerts = make([]models.ResourceAlertInput, len(r.Alerts)) + for j, a := range r.Alerts { + input.Alerts[j] = models.ResourceAlertInput{ + ID: a.ID, + Type: a.Type, + Level: a.Level, + Message: a.Message, + Value: a.Value, + Threshold: a.Threshold, + StartTimeUnix: a.StartTime.UnixMilli(), + } + } + } + + // Convert identity + if r.Identity != nil { + input.Identity = &models.ResourceIdentityInput{ + Hostname: r.Identity.Hostname, + MachineID: r.Identity.MachineID, + IPs: r.Identity.IPs, + } + } + + // Convert platform data from json.RawMessage to map + if len(r.PlatformData) > 0 { + var platformMap map[string]any + if err := json.Unmarshal(r.PlatformData, &platformMap); err == nil { + input.PlatformData = platformMap + } + } + + result[i] = models.ConvertResourceToFrontend(input) + } + + return result +} + // pollStorageBackupsWithNodes polls backups using a provided nodes list to avoid duplicate GetNodes calls func (m *Monitor) pollStorageBackupsWithNodes(ctx context.Context, instanceName string, client PVEClientInterface, nodes []proxmox.Node, nodeEffectiveStatus map[string]string) {