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.
This commit is contained in:
rcourtman
parent
ef35120afb
commit
cdd2249e00
3 changed files with 330 additions and 3 deletions
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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"`
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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) {
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue