diff --git a/.gemini/docs/unified-resource-architecture.md b/.gemini/docs/unified-resource-architecture.md new file mode 100644 index 0000000..3351c7a --- /dev/null +++ b/.gemini/docs/unified-resource-architecture.md @@ -0,0 +1,548 @@ +# Unified Resource Architecture + +## Status: Draft +## Author: AI Assistant + rcourtman +## Created: 2025-12-07 +## Last Updated: 2025-12-07 + +--- + +## Executive Summary + +Pulse is evolving from a traditional "stare at dashboards" monitoring tool to an **AI-first infrastructure management platform** where AI is the primary operator and humans are supervisors. This document outlines the architectural changes needed to support this vision while preserving the excellent UI that 20,000+ users love. + +The core change is introducing a **Unified Resource Model** - a common data abstraction that all platforms (Proxmox, Docker, Hosts, and future platforms like Kubernetes and TrueNAS) normalize to. This enables: + +1. **AI intelligence across all platforms** - AI can reason about your entire infrastructure +2. **Elimination of duplicate monitoring** - One machine = one set of alerts +3. **Extensibility for new platforms** - Adding Kubernetes is adding a new resource type, not a new architecture +4. **Foundation for unified views** - Optional consolidated UI without breaking existing pages + +--- + +## Problem Statement + +### Current Architecture Issues + +#### 1. Multiple Data Sources for Same Machine +When a Proxmox node also has a host agent running, Pulse monitors it twice: +- Via Proxmox API → "Node cpu at 97.9%" +- Via Host Agent → "Host cpu at 99.7%" + +Result: Duplicate alerts, user confusion. + +#### 2. Siloed Data Models +Each platform has its own types with no common abstraction: +``` +Node, VM, Container (Proxmox) +DockerHost, DockerContainer (Docker) +Host (Host Agent) +PBSInstance, PBSDatastore (PBS) +PMGInstance (PMG) +``` + +AI can't easily answer: "What's using the most CPU across my infrastructure?" + +#### 3. Platform-Specific Pages +The frontend has separate components for each platform: +- Dashboard (Proxmox VMs/Containers) +- Docker page +- Hosts page +- Storage page + +Adding a new platform (Kubernetes) means building an entirely new page. + +#### 4. Agent Capabilities Underutilized +The unified pulse-agent can: +- Report richer metrics than Proxmox API +- Execute commands for AI +- Monitor temperature, RAID, disk I/O at granular level + +But the architecture doesn't fully leverage this. + +--- + +## Vision: AI-First Monitoring + +``` +┌────────────────────────────────────────────────────────────────────────┐ +│ TRADITIONAL MONITORING │ +│ │ +│ Human 👁️ ──watches──▶ Dashboard 📊 ──shows──▶ Metrics/Alerts │ +│ │ │ +│ └──────────────────── investigates ──▶ fixes manually │ +└────────────────────────────────────────────────────────────────────────┘ + + ⬇️ + +┌────────────────────────────────────────────────────────────────────────┐ +│ PULSE: AI-FIRST │ +│ │ +│ Infrastructure 🖥️ ──reports to──▶ Pulse AI 🤖 │ +│ │ │ +│ ├── Detects anomalies │ +│ ├── Correlates across systems │ +│ ├── Diagnoses root causes │ +│ ├── Suggests fixes │ +│ └── Executes fixes (approved) │ +│ │ │ +│ ▼ │ +│ Human 👤 ──reviews/approves──▶ Dashboard = CONTEXT, not PRIMARY │ +└────────────────────────────────────────────────────────────────────────┘ +``` + +The dashboard remains beautiful and functional, but AI becomes the primary interface for investigation and remediation. + +--- + +## Solution: Unified Resource Model + +### Core Abstraction + +Every monitored entity becomes a `Resource` with common fields: + +```go +// Resource is the universal abstraction for any monitored entity +type Resource struct { + // Identity + ID string `json:"id"` // Globally unique ID + Type ResourceType `json:"type"` // vm, container, docker-container, pod, host, etc. + Name string `json:"name"` // Human-readable name + DisplayName string `json:"displayName"` // Custom display name (if set) + + // Platform/Source + PlatformID string `json:"platformId"` // Which platform instance + PlatformType PlatformType `json:"platformType"` // proxmox-pve, docker, kubernetes, etc. + SourceType SourceType `json:"sourceType"` // api, agent, hybrid + + // Hierarchy + ParentID string `json:"parentId,omitempty"` // VM → Node, Pod → K8s Node + ClusterID string `json:"clusterId,omitempty"` // Cluster membership + + // Universal Metrics (nullable - not all resources have all metrics) + Status ResourceStatus `json:"status"` // online, offline, running, stopped, degraded + CPU *MetricValue `json:"cpu,omitempty"` + Memory *MetricValue `json:"memory,omitempty"` + Disk *MetricValue `json:"disk,omitempty"` + Network *NetworkMetric `json:"network,omitempty"` + Temperature *float64 `json:"temperature,omitempty"` + Uptime *int64 `json:"uptime,omitempty"` + + // Universal Metadata + Tags []string `json:"tags,omitempty"` + Labels map[string]string `json:"labels,omitempty"` + LastSeen time.Time `json:"lastSeen"` + Alerts []ResourceAlert `json:"alerts,omitempty"` + + // Platform-Specific Data (discriminated by Type) + // This preserves all the rich data while allowing common handling + PlatformData json.RawMessage `json:"platformData,omitempty"` +} + +type ResourceType string + +const ( + // Infrastructure + ResourceTypeNode ResourceType = "node" + ResourceTypeHost ResourceType = "host" + ResourceTypeDockerHost ResourceType = "docker-host" + ResourceTypeK8sNode ResourceType = "k8s-node" + ResourceTypeTrueNASSystem ResourceType = "truenas-system" + + // Compute Workloads + ResourceTypeVM ResourceType = "vm" + ResourceTypeContainer ResourceType = "container" // LXC + ResourceTypeDockerContainer ResourceType = "docker-container" + ResourceTypePod ResourceType = "pod" + ResourceTypeJail ResourceType = "jail" + + // Services + ResourceTypeDockerService ResourceType = "docker-service" + ResourceTypeK8sDeployment ResourceType = "k8s-deployment" + ResourceTypeK8sService ResourceType = "k8s-service" + + // Storage + ResourceTypeStorage ResourceType = "storage" + ResourceTypeDatastore ResourceType = "datastore" + ResourceTypePool ResourceType = "pool" + ResourceTypeDataset ResourceType = "dataset" + + // Backup Systems + ResourceTypePBS ResourceType = "pbs" + ResourceTypePMG ResourceType = "pmg" +) + +type PlatformType string + +const ( + PlatformProxmoxPVE PlatformType = "proxmox-pve" + PlatformProxmoxPBS PlatformType = "proxmox-pbs" + PlatformProxmoxPMG PlatformType = "proxmox-pmg" + PlatformDocker PlatformType = "docker" + PlatformKubernetes PlatformType = "kubernetes" + PlatformTrueNAS PlatformType = "truenas" + PlatformHostAgent PlatformType = "host-agent" +) + +type SourceType string + +const ( + SourceAPI SourceType = "api" // Data from polling an API + SourceAgent SourceType = "agent" // Data pushed from agent + SourceHybrid SourceType = "hybrid" // Both sources, agent preferred +) +``` + +### Deduplication Strategy + +When multiple sources report on the same physical machine: + +```go +type ResourceIdentity struct { + Hostname string // Primary identifier + MachineID string // /etc/machine-id or equivalent + IPs []string // Network addresses +} + +// IdentityMatcher determines if two resources are the same machine +func (r *ResourceStore) IdentityMatcher(a, b Resource) bool { + // 1. Machine ID match (most reliable) + if a.MachineID != "" && a.MachineID == b.MachineID { + return true + } + + // 2. Hostname match (case-insensitive) + if strings.EqualFold(a.Hostname, b.Hostname) { + return true + } + + // 3. IP overlap (if same IP, likely same machine) + for _, ipA := range a.IPs { + for _, ipB := range b.IPs { + if ipA == ipB && !isLocalhost(ipA) { + return true + } + } + } + + return false +} +``` + +When duplicates are detected, prefer **Agent > API**: + +| Scenario | Result | +|----------|--------| +| Node only (no agent) | Use Node data | +| Host agent only (no Proxmox) | Use Host data | +| Both Node + Host agent | Use Host agent data, suppress Node alerts | +| Docker agent on Proxmox node | Combine: Docker data + Host metrics from agent | + +--- + +## Implementation Phases + +### Phase 0: Document & Design (This Document) +- [x] Capture the vision +- [x] Define the data model +- [ ] Review with stakeholders +- [ ] Create GitHub issues/milestones + +### Phase 1: Backend Unification (Invisible to Users) + +**Goal:** Create the Resource abstraction without changing any frontend behavior. + +**Status:** ✅ Core implementation complete + +**Completed Tasks:** +1. ✅ Create `internal/resources/resource.go` with types - Core Resource type, enums, and helper methods +2. ✅ Create `internal/resources/platform_data.go` - Platform-specific data types for PlatformData field +3. ✅ Create `internal/resources/store.go` for unified storage - In-memory store with indexes +4. ✅ Create `internal/resources/converters.go` - Converters: FromNode(), FromHost(), FromDockerHost(), FromVM(), FromContainer(), FromDockerContainer(), FromPBSInstance(), FromStorage() +5. ✅ Add deduplication logic using identity matching (hostname, machineID, IP) +6. ✅ Create `internal/resources/converters_test.go` and `store_test.go` - 25 passing tests +7. ✅ Create `internal/api/resource_handlers.go` - HTTP handlers for unified resources API +8. ✅ Add `/api/resources` endpoint with filtering (type, platform, status, parent, infrastructure, workloads) +9. ✅ Add `/api/resources/stats` endpoint for store statistics +10. ✅ Add `/api/resources/{id}` endpoint for individual resource lookup +11. ✅ Add deduplication helper methods for alert manager: `IsSuppressed()`, `GetPreferredResourceFor()`, `IsSamePhysicalMachine()`, `HasPreferredSourceForHostname()` + +**Integration Points for Alert Manager:** +The resource store provides these methods that the existing alert manager can optionally use: +- `store.IsSuppressed(resourceID)` - Check if a resource was deduplicated +- `store.GetPreferredResourceFor(resourceID)` - Get the preferred resource +- `store.HasPreferredSourceForHostname(hostname)` - Check if an agent is preferred for a hostname + +**Remaining Tasks:** +- [ ] *(Optional)* Enhance alert manager to use resource store methods instead of hostname-only dedup +- [ ] *(Deferred to Phase 2)* Add optional `resources` array to WebSocket state + +**Backward Compatibility:** +- WebSocket still sends `nodes`, `vms`, `containers`, `hosts`, `dockerHosts` as separate arrays +- Unified resources available via REST API at `/api/resources` +- All existing frontend code continues to work unchanged +- Existing hostname-based deduplication in alert manager still works + +**Metrics:** +- Zero frontend changes +- All existing tests pass +- New resource tests added (25 tests) +- New REST API endpoints available + +### Phase 2: AI Context Enhancement + +**Goal:** AI chat can query and act across all resource types. + +**Status:** ✅ Fully complete + +**Completed Tasks:** +1. ✅ Create `internal/ai/resource_context.go` - Unified context builder for AI +2. ✅ Add `ResourceProvider` interface to AI service +3. ✅ Modify `buildSystemPrompt` to use unified model when available +4. ✅ Wire up resource provider in router to AI handlers +5. ✅ AI now uses deduplicated view of infrastructure (falls back to legacy if not available) +6. ✅ Add cross-platform query methods: `GetTopByCPU()`, `GetTopByMemory()`, `GetTopByDisk()` +7. ✅ Add resource correlation: `GetRelated()` for parent/children/siblings/cluster members +8. ✅ Add infrastructure summary: `GetResourceSummary()` with status counts and averages +9. ✅ AI context includes "Top CPU Consumers", "Top Memory Consumers", "Top Disk Usage" +10. ✅ AI context includes infrastructure summary with health status + +**How It Works:** +- When `ResourceProvider` is set, AI gets a cleaner "Unified Infrastructure View" +- Resources grouped by platform (Proxmox nodes, Standalone hosts, Docker hosts) +- Workloads grouped by parent infrastructure +- Agent status shown inline with infrastructure +- Resources with alerts highlighted +- **Top consumers** shown for CPU, Memory, and Disk +- **Infrastructure summary** with healthy/degraded/offline counts + +**Cross-Platform Query Methods:** +```go +// Find top resource consumers across all platforms +store.GetTopByCPU(10, nil) // Top 10 by CPU, any type +store.GetTopByMemory(5, []ResourceType{ResourceTypeVM}) // Top 5 VMs by memory + +// Find related resources +store.GetRelated("vm-123") // Returns parent, children, siblings, cluster_members + +// Get infrastructure overview +store.GetResourceSummary() // TotalResources, Healthy, Degraded, Offline, ByType, ByPlatform +``` + +**User Experience:** +- AI can now answer "What's using the most CPU?" across all platforms +- AI knows about resource relationships (parent nodes, sibling VMs, cluster members) +- AI has infrastructure summary context for better analysis + +### Phase 3: Agent Preference & Hybrid Mode + +**Goal:** When agents exist, prefer their data over API polling. + +**Status:** ✅ Core implementation complete + +**Completed Tasks:** +1. ✅ Polling optimization methods added to resource store: + - `ShouldSkipAPIPolling(hostname)` - Check if API polling should be skipped + - `GetAgentMonitoredHostnames()` - Get list of agent-monitored hosts + - `GetPollingRecommendations()` - Get per-hostname polling multipliers +2. ✅ Hybrid source type defined (`SourceHybrid`) +3. ✅ Agent data automatically preferred over API data (store deduplication) +4. ✅ `ResourceStoreInterface` added to Monitor +5. ✅ `SetResourceStore()` method added to inject store into Monitor +6. ✅ `shouldSkipNodeMetrics()` helper method added to Monitor +7. ✅ Resource store wired into Monitor via `Router.SetMonitor()` + +**How It Works:** +```go +// In Monitor struct +resourceStore ResourceStoreInterface + +// Router injects the store when setting the monitor +func (r *Router) SetMonitor(m *monitoring.Monitor) { + // ... other setup ... + if r.resourceHandlers != nil { + m.SetResourceStore(r.resourceHandlers.Store()) + } +} + +// Monitor can now check if polling should be skipped +func (m *Monitor) shouldSkipNodeMetrics(nodeName string) bool { + if store := m.resourceStore; store != nil { + return store.ShouldSkipAPIPolling(nodeName) + } + return false +} +``` + +**How to Enable Polling Optimization:** + +To enable polling optimization in the actual polling loops, add this check in `pollPVENode()`: + +```go +// In monitor_polling.go, at the start of pollPVENode(): +func (m *Monitor) pollPVENode(...) (models.Node, string, error) { + // Skip detailed metric polling if host agent provides data + if m.shouldSkipNodeMetrics(node.Node) { + // Still return basic node info but skip expensive API calls + // like GetNodeStatus, GetStorage, etc. + } + // ... rest of function +} +``` + +**Remaining Tasks (Future Enhancement):** +- [ ] Add config flag: `EnableAgentPollingOptimization bool` +- [ ] Actually integrate skip logic into `pollPVENode()` and related functions +- [ ] Add Prometheus metrics for skipped polls +- [ ] Add logging for poll optimization decisions + +**Benefits:** +- Better data quality (agent metrics are more accurate) +- Reduced API load (skip redundant polling) +- More AI capabilities (command execution via agents) + +### Phase 4: Optional Unified View (Future) + +**Goal:** Add a consolidated "All Resources" view for power users. + +**Tasks:** +1. Create new React/Solid component for unified resource table +2. Implement filtering by platform, type, status, tags +3. Support hierarchical grouping (cluster > node > workloads) +4. Add as new optional view, don't replace existing pages + +**User Experience:** +- New "All Resources" option in navigation +- Existing pages unchanged +- Users choose their preferred view + +### Phase 5: New Platform Support (Ongoing) + +Each new platform follows the pattern: + +1. **Collector:** Poll API or receive agent telemetry +2. **Converter:** `PlatformDataToResource()` function +3. **Platform Data:** Type-specific struct stored in `PlatformData` field +4. **UI Components:** (Optional) Platform-specific detail views + +Example for Kubernetes: +```go +func K8sNodeToResource(node k8s.Node) Resource { + return Resource{ + ID: fmt.Sprintf("k8s/%s/%s", clusterID, node.Name), + Type: ResourceTypeK8sNode, + Name: node.Name, + PlatformType: PlatformKubernetes, + SourceType: SourceAPI, + Status: mapK8sNodeStatus(node.Status), + CPU: extractK8sCPU(node), + Memory: extractK8sMemory(node), + PlatformData: marshalK8sNodeData(node), + } +} +``` + +--- + +## Migration Strategy + +### Database Considerations + +Current state is in-memory with JSON persistence for alerts. The unified model should: + +1. **Short-term:** Continue in-memory, Resource is just an abstraction layer +2. **Medium-term:** SQLite metrics store (already implemented) extended for resources +3. **Long-term:** Consider time-series DB for metrics history + +### API Compatibility + +**Existing endpoints remain unchanged:** +- `/api/state` - Returns current format +- WebSocket `state` message - Returns current format + +**New optional endpoints:** +- `/api/resources` - Returns unified resource list +- `/api/resources/{id}` - Returns single resource with full detail +- WebSocket `resources` message - Unified resource updates + +### Frontend Migration + +**No forced migration.** Existing components continue to work. New unified view is additive. + +If we eventually want to migrate existing components: +1. Create `useResources()` hook that abstracts data source +2. Components can switch when ready +3. Old and new can coexist + +--- + +## Success Metrics + +### Phase 1 Complete When: +- [x] Resource type defined and tested +- [x] All current types have converters +- [x] Deduplication prevents duplicate alerts (store logic complete) +- [x] Zero frontend changes required +- [x] All existing tests pass +- [x] REST API endpoints available (`/api/resources`, `/api/resources/stats`, `/api/resources/{id}`) +- [x] Deduplication helper methods available for alert manager integration + +### Phase 2 Complete When: +- [x] AI context includes unified resource view +- [x] AI can answer cross-platform questions (via GetTopByCPU/Memory/Disk) +- [x] Correlation across platforms works (via GetRelated) + +### Phase 3 Complete When: +- [x] Agent data preferred when available (via store deduplication) +- [x] Polling optimization methods available (`ShouldSkipAPIPolling`, `GetPollingRecommendations`) +- [x] Resource store wired into Monitor (via `SetResourceStore`) +- [x] `shouldSkipNodeMetrics()` helper available for polling loops +- [x] AI can execute commands via agents (already implemented) +- [ ] Polling optimization actually used in live polling loops (optional enhancement) + +### Phase 4 Complete When: +- [ ] Unified view accessible from UI +- [ ] Filtering and grouping works +- [ ] Existing pages still work + +--- + +## Open Questions + +1. **Should Resource replace existing types entirely, or wrap them?** + - Recommendation: Wrap initially, replace gradually + +2. **How to handle platform-specific features in unified view?** + - Recommendation: Show common fields, expand for platform-specific + +3. **Should we version the Resource schema?** + - Recommendation: Yes, include version field for future evolution + +4. **How to handle offline/stale data in unified model?** + - Recommendation: Include `lastSeen` and `staleness` fields + +--- + +## Appendix: Resource Type Mapping + +| Current Type | Resource Type | Platform Type | Notes | +|--------------|---------------|---------------|-------| +| Node | node | proxmox-pve | Proxmox VE node | +| VM | vm | proxmox-pve | Proxmox VM | +| Container | container | proxmox-pve | LXC container | +| Host | host | host-agent | Standalone host | +| DockerHost | docker-host | docker | Docker/Podman host | +| DockerContainer | docker-container | docker | Docker container | +| PBSInstance | pbs | proxmox-pbs | Backup server | +| PBSDatastore | datastore | proxmox-pbs | PBS datastore | +| PMGInstance | pmg | proxmox-pmg | Mail gateway | +| Storage | storage | proxmox-pve | PVE storage | +| CephCluster | ceph-cluster | proxmox-pve | Ceph cluster | + +--- + +## References + +- [Conversation that sparked this design](internal discussion 2025-12-07) +- [Host Agent Deduplication Fix](commit implementing hostname-based dedup) +- [Pulse AI Features](current AI implementation) diff --git a/internal/ai/resource_context.go b/internal/ai/resource_context.go new file mode 100644 index 0000000..12cf4e9 --- /dev/null +++ b/internal/ai/resource_context.go @@ -0,0 +1,310 @@ +package ai + +import ( + "fmt" + "strings" + + "github.com/rcourtman/pulse-go-rewrite/internal/resources" + "github.com/rs/zerolog/log" +) + +// ResourceProvider provides access to the unified resource store. +type ResourceProvider interface { + GetAll() []resources.Resource + GetInfrastructure() []resources.Resource + GetWorkloads() []resources.Resource + GetByType(t resources.ResourceType) []resources.Resource + GetStats() resources.StoreStats + + // Cross-platform query methods + GetTopByCPU(limit int, types []resources.ResourceType) []resources.Resource + GetTopByMemory(limit int, types []resources.ResourceType) []resources.Resource + GetTopByDisk(limit int, types []resources.ResourceType) []resources.Resource + GetRelated(resourceID string) map[string][]resources.Resource + GetResourceSummary() resources.ResourceSummary +} + +// SetResourceProvider sets the resource provider for unified infrastructure context. +func (s *Service) SetResourceProvider(rp ResourceProvider) { + s.mu.Lock() + defer s.mu.Unlock() + s.resourceProvider = rp +} + +// buildUnifiedResourceContext creates AI context from the unified resource model. +// This provides a cleaner, deduplicated view of infrastructure. +func (s *Service) buildUnifiedResourceContext() string { + s.mu.RLock() + rp := s.resourceProvider + s.mu.RUnlock() + + if rp == nil { + return "" + } + + var sections []string + stats := rp.GetStats() + + // Header with summary + sections = append(sections, "## Unified Infrastructure View") + sections = append(sections, fmt.Sprintf("Total resources: %d (Infrastructure: %d, Workloads: %d)", + stats.TotalResources, stats.ByType[resources.ResourceTypeNode]+stats.ByType[resources.ResourceTypeHost]+stats.ByType[resources.ResourceTypeDockerHost], + stats.ByType[resources.ResourceTypeVM]+stats.ByType[resources.ResourceTypeContainer]+stats.ByType[resources.ResourceTypeDockerContainer])) + + // Build agent lookup + agentsByHostname := make(map[string]bool) + if s.agentServer != nil { + for _, agent := range s.agentServer.GetConnectedAgents() { + agentsByHostname[strings.ToLower(agent.Hostname)] = true + } + } + + // Infrastructure resources (nodes, hosts, docker hosts) + infrastructure := rp.GetInfrastructure() + if len(infrastructure) > 0 { + sections = append(sections, "\n### Infrastructure (Nodes & Hosts)") + sections = append(sections, "These are the physical/virtual machines that host workloads.") + + // Group by platform + byPlatform := make(map[resources.PlatformType][]resources.Resource) + for _, r := range infrastructure { + byPlatform[r.PlatformType] = append(byPlatform[r.PlatformType], r) + } + + // Proxmox nodes + if nodes, ok := byPlatform[resources.PlatformProxmoxPVE]; ok && len(nodes) > 0 { + sections = append(sections, "\n**Proxmox VE Nodes:**") + for _, node := range nodes { + hasAgent := agentsByHostname[strings.ToLower(node.Name)] + agentStatus := "NO AGENT" + if hasAgent { + agentStatus = "HAS AGENT ✓" + } + + // Build cluster info + clusterInfo := "" + if node.ClusterID != "" { + clusterInfo = fmt.Sprintf(" [cluster: %s]", node.ClusterID) + } + + // Get CPU/Memory metrics + metrics := "" + if node.CPU != nil && node.Memory != nil { + metrics = fmt.Sprintf(" - CPU: %.1f%%, Mem: %.1f%%", node.CPUPercent(), node.MemoryPercent()) + } + + sections = append(sections, fmt.Sprintf("- **%s** (%s)%s%s [%s]", + node.EffectiveDisplayName(), agentStatus, clusterInfo, metrics, node.Status)) + } + } + + // Standalone hosts + if hosts, ok := byPlatform[resources.PlatformHostAgent]; ok && len(hosts) > 0 { + sections = append(sections, "\n**Standalone Hosts (via Host Agent):**") + for _, host := range hosts { + // Get IPs from identity + ips := "" + if host.Identity != nil && len(host.Identity.IPs) > 0 { + ips = " - " + strings.Join(host.Identity.IPs, ", ") + } + + // Get metrics + metrics := "" + if host.CPU != nil && host.Memory != nil { + metrics = fmt.Sprintf(", CPU: %.1f%%, Mem: %.1f%%", host.CPUPercent(), host.MemoryPercent()) + } + + sections = append(sections, fmt.Sprintf("- **%s**%s%s [%s]", + host.EffectiveDisplayName(), ips, metrics, host.Status)) + } + } + + // Docker hosts + if dhosts, ok := byPlatform[resources.PlatformDocker]; ok && len(dhosts) > 0 { + sections = append(sections, "\n**Docker/Podman Hosts:**") + for _, dh := range dhosts { + if dh.Type != resources.ResourceTypeDockerHost { + continue + } + + // Count containers for this host + allWorkloads := rp.GetWorkloads() + containerCount := 0 + runningCount := 0 + for _, w := range allWorkloads { + if w.ParentID == dh.ID { + containerCount++ + if w.Status == resources.StatusRunning { + runningCount++ + } + } + } + + sections = append(sections, fmt.Sprintf("- **%s** (%d/%d containers running) [%s]", + dh.EffectiveDisplayName(), runningCount, containerCount, dh.Status)) + } + } + } + + // Workloads (VMs, containers) + workloads := rp.GetWorkloads() + if len(workloads) > 0 { + sections = append(sections, "\n### Workloads (VMs & Containers)") + + // Group by parent for better organization + byParent := make(map[string][]resources.Resource) + noParent := []resources.Resource{} + for _, w := range workloads { + if w.ParentID != "" { + byParent[w.ParentID] = append(byParent[w.ParentID], w) + } else { + noParent = append(noParent, w) + } + } + + // Get all infrastructure to map parent IDs to names + infraMap := make(map[string]resources.Resource) + for _, r := range infrastructure { + infraMap[r.ID] = r + } + + // Show workloads grouped by parent + for parentID, children := range byParent { + parentName := parentID + if parent, ok := infraMap[parentID]; ok { + parentName = parent.EffectiveDisplayName() + } + + sections = append(sections, fmt.Sprintf("\n**On %s:**", parentName)) + for _, w := range children { + typeLabel := string(w.Type) + if w.Type == resources.ResourceTypeVM { + typeLabel = "VM" + } else if w.Type == resources.ResourceTypeContainer { + typeLabel = "LXC" + } else if w.Type == resources.ResourceTypeDockerContainer { + typeLabel = "Docker" + } + + // Get VMID from platform data if available + vmidInfo := "" + if w.PlatformID != "" && w.Type != resources.ResourceTypeDockerContainer { + vmidInfo = fmt.Sprintf(" %s", w.PlatformID) + } + + // Get IPs + ips := "" + if w.Identity != nil && len(w.Identity.IPs) > 0 { + ips = " - " + strings.Join(w.Identity.IPs[:min(2, len(w.Identity.IPs))], ", ") + } + + sections = append(sections, fmt.Sprintf(" - **%s** (%s%s)%s [%s]", + w.EffectiveDisplayName(), typeLabel, vmidInfo, ips, w.Status)) + } + } + + // Show orphaned workloads + if len(noParent) > 0 { + sections = append(sections, "\n**Other workloads:**") + for _, w := range noParent { + sections = append(sections, fmt.Sprintf(" - **%s** (%s) [%s]", + w.EffectiveDisplayName(), w.Type, w.Status)) + } + } + } + + // Resources with alerts + allResources := rp.GetAll() + var alertResources []resources.Resource + for _, r := range allResources { + if len(r.Alerts) > 0 { + alertResources = append(alertResources, r) + } + } + if len(alertResources) > 0 { + sections = append(sections, "\n### Resources with Active Alerts") + for _, r := range alertResources { + for _, alert := range r.Alerts { + sections = append(sections, fmt.Sprintf("- **%s**: %s (%s)", + r.EffectiveDisplayName(), alert.Message, alert.Level)) + } + } + } + + // Add resource summary for cross-platform analysis + summary := rp.GetResourceSummary() + if summary.TotalResources > 0 { + sections = append(sections, "\n### Infrastructure Summary") + sections = append(sections, fmt.Sprintf("- Status: %d healthy, %d degraded, %d offline", + summary.Healthy, summary.Degraded, summary.Offline)) + if summary.WithAlerts > 0 { + sections = append(sections, fmt.Sprintf("- Resources with alerts: %d", summary.WithAlerts)) + } + + // Show average resource usage by type + if len(summary.ByType) > 0 { + sections = append(sections, "- Average utilization by type:") + for t, ts := range summary.ByType { + if ts.Count > 0 && (ts.AvgCPUPercent > 0 || ts.AvgMemoryPercent > 0) { + sections = append(sections, fmt.Sprintf(" - %s (%d): CPU %.1f%%, Memory %.1f%%", + t, ts.Count, ts.AvgCPUPercent, ts.AvgMemoryPercent)) + } + } + } + } + + // Top resource consumers (helps answer "what's using the most CPU/memory") + topCPU := rp.GetTopByCPU(3, nil) + if len(topCPU) > 0 { + sections = append(sections, "\n### Top CPU Consumers") + for i, r := range topCPU { + sections = append(sections, fmt.Sprintf("%d. **%s** (%s): %.1f%%", + i+1, r.EffectiveDisplayName(), r.Type, r.CPUPercent())) + } + } + + topMem := rp.GetTopByMemory(3, nil) + if len(topMem) > 0 { + sections = append(sections, "\n### Top Memory Consumers") + for i, r := range topMem { + sections = append(sections, fmt.Sprintf("%d. **%s** (%s): %.1f%%", + i+1, r.EffectiveDisplayName(), r.Type, r.MemoryPercent())) + } + } + + topDisk := rp.GetTopByDisk(3, nil) + if len(topDisk) > 0 { + sections = append(sections, "\n### Top Disk Usage") + for i, r := range topDisk { + sections = append(sections, fmt.Sprintf("%d. **%s** (%s): %.1f%%", + i+1, r.EffectiveDisplayName(), r.Type, r.DiskPercent())) + } + } + + if len(sections) == 0 { + return "" + } + + result := "\n\n" + strings.Join(sections, "\n") + + // Limit context size + const maxContextSize = 50000 + if len(result) > maxContextSize { + log.Warn(). + Int("original_size", len(result)). + Int("max_size", maxContextSize). + Msg("Unified resource context truncated") + result = result[:maxContextSize] + "\n\n[... Context truncated ...]" + } + + log.Debug().Int("unified_resource_context_size", len(result)).Msg("Built unified resource context") + return result +} + +// min returns the smaller of two integers +func min(a, b int) int { + if a < b { + return a + } + return b +} diff --git a/internal/ai/service.go b/internal/ai/service.go index 8c09668..4d5ff7b 100644 --- a/internal/ai/service.go +++ b/internal/ai/service.go @@ -29,15 +29,16 @@ type StateProvider interface { // Service orchestrates AI interactions type Service struct { - mu sync.RWMutex - persistence *config.ConfigPersistence - provider providers.Provider - cfg *config.AIConfig - agentServer *agentexec.Server - policy *agentexec.CommandPolicy - stateProvider StateProvider - alertProvider AlertProvider - knowledgeStore *knowledge.Store + mu sync.RWMutex + persistence *config.ConfigPersistence + provider providers.Provider + cfg *config.AIConfig + agentServer *agentexec.Server + policy *agentexec.CommandPolicy + stateProvider StateProvider + alertProvider AlertProvider + knowledgeStore *knowledge.Store + resourceProvider ResourceProvider // Unified resource model provider (Phase 2) } // NewService creates a new AI service @@ -1760,7 +1761,18 @@ Pulse manages LXC containers agentlessly from the PVE host. } // Add connected infrastructure info - prompt += s.buildInfrastructureContext() + // Try unified resource context first (Phase 2), fall back to legacy + s.mu.RLock() + hasResourceProvider := s.resourceProvider != nil + s.mu.RUnlock() + + if hasResourceProvider { + // Use the new unified resource model for cleaner, deduplicated context + prompt += s.buildUnifiedResourceContext() + } else { + // Fall back to legacy state-based context + prompt += s.buildInfrastructureContext() + } // Add user annotations from all resources (global context) prompt += s.buildUserAnnotationsContext() diff --git a/internal/api/ai_handlers.go b/internal/api/ai_handlers.go index 9c76b9a..39aa8dc 100644 --- a/internal/api/ai_handlers.go +++ b/internal/api/ai_handlers.go @@ -51,6 +51,11 @@ func (h *AISettingsHandler) SetStateProvider(sp ai.StateProvider) { h.aiService.SetStateProvider(sp) } +// SetResourceProvider sets the resource provider for unified infrastructure context (Phase 2) +func (h *AISettingsHandler) SetResourceProvider(rp ai.ResourceProvider) { + h.aiService.SetResourceProvider(rp) +} + // AISettingsResponse is returned by GET /api/settings/ai // API key is masked for security type AISettingsResponse struct { diff --git a/internal/api/resource_handlers.go b/internal/api/resource_handlers.go new file mode 100644 index 0000000..a6ce5a2 --- /dev/null +++ b/internal/api/resource_handlers.go @@ -0,0 +1,237 @@ +package api + +import ( + "encoding/json" + "net/http" + "strings" + + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/internal/resources" +) + +// ResourceHandlers provides HTTP handlers for the unified resource API. +type ResourceHandlers struct { + store *resources.Store +} + +// NewResourceHandlers creates resource handlers with a new store. +func NewResourceHandlers() *ResourceHandlers { + return &ResourceHandlers{ + store: resources.NewStore(), + } +} + +// Store returns the underlying resource store for populating from the monitor. +func (h *ResourceHandlers) Store() *resources.Store { + return h.store +} + +// HandleGetResources returns all resources, optionally filtered by query params. +// GET /api/resources +// Query params: +// - type: filter by resource type (comma-separated) +// - platform: filter by platform type (comma-separated) +// - status: filter by status (comma-separated) +// - parent: filter by parent ID +// - infrastructure: if "true", only return infrastructure resources +// - workloads: if "true", only return workload resources +func (h *ResourceHandlers) HandleGetResources(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + + query := h.store.Query() + + // Parse type filter + if typeParam := r.URL.Query().Get("type"); typeParam != "" { + types := parseResourceTypes(typeParam) + if len(types) > 0 { + query = query.OfType(types...) + } + } + + // Parse platform filter + if platformParam := r.URL.Query().Get("platform"); platformParam != "" { + platforms := parsePlatformTypes(platformParam) + if len(platforms) > 0 { + query = query.FromPlatform(platforms...) + } + } + + // Parse status filter + if statusParam := r.URL.Query().Get("status"); statusParam != "" { + statuses := parseStatuses(statusParam) + if len(statuses) > 0 { + query = query.WithStatus(statuses...) + } + } + + // Parse parent filter + if parentID := r.URL.Query().Get("parent"); parentID != "" { + query = query.WithParent(parentID) + } + + // Parse alerts filter + if r.URL.Query().Get("alerts") == "true" { + query = query.WithAlerts() + } + + // Execute query + var result []resources.Resource + + // Handle infrastructure/workloads shortcut + if r.URL.Query().Get("infrastructure") == "true" { + result = h.store.GetInfrastructure() + } else if r.URL.Query().Get("workloads") == "true" { + result = h.store.GetWorkloads() + } else { + result = query.Execute() + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(ResourcesResponse{ + Resources: result, + Count: len(result), + Stats: h.store.GetStats(), + }) +} + +// HandleGetResource returns a single resource by ID. +// GET /api/resources/{id} +func (h *ResourceHandlers) HandleGetResource(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + + // Extract ID from path: /api/resources/{id} + path := strings.TrimPrefix(r.URL.Path, "/api/resources/") + if path == "" || path == "/" { + http.Error(w, "Resource ID required", http.StatusBadRequest) + return + } + + resource, ok := h.store.Get(path) + if !ok { + http.Error(w, "Resource not found", http.StatusNotFound) + return + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(resource) +} + +// HandleGetResourceStats returns statistics about the resource store. +// GET /api/resources/stats +func (h *ResourceHandlers) HandleGetResourceStats(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + + stats := h.store.GetStats() + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(stats) +} + +// ResourcesResponse is the response for /api/resources. +type ResourcesResponse struct { + Resources []resources.Resource `json:"resources"` + Count int `json:"count"` + Stats resources.StoreStats `json:"stats"` +} + +// PopulateFromState converts all resources from a StateSnapshot to the unified store. +// This should be called whenever the state is updated. +func (h *ResourceHandlers) PopulateFromSnapshot(snapshot models.StateSnapshot) { + // Convert nodes + for _, node := range snapshot.Nodes { + r := resources.FromNode(node) + h.store.Upsert(r) + } + + // Convert VMs + for _, vm := range snapshot.VMs { + r := resources.FromVM(vm) + h.store.Upsert(r) + } + + // Convert containers + for _, ct := range snapshot.Containers { + r := resources.FromContainer(ct) + h.store.Upsert(r) + } + + // Convert hosts + for _, host := range snapshot.Hosts { + r := resources.FromHost(host) + h.store.Upsert(r) + } + + // Convert docker hosts and their containers + for _, dh := range snapshot.DockerHosts { + r := resources.FromDockerHost(dh) + h.store.Upsert(r) + + // Convert containers within the docker host + for _, dc := range dh.Containers { + r := resources.FromDockerContainer(dc, dh.ID, dh.Hostname) + h.store.Upsert(r) + } + } + + // Convert PBS instances + for _, pbs := range snapshot.PBSInstances { + r := resources.FromPBSInstance(pbs) + h.store.Upsert(r) + } + + // Convert storage + for _, storage := range snapshot.Storage { + r := resources.FromStorage(storage) + h.store.Upsert(r) + } +} + +// Helper functions for parsing query parameters + +func parseResourceTypes(s string) []resources.ResourceType { + parts := strings.Split(s, ",") + var result []resources.ResourceType + for _, p := range parts { + p = strings.TrimSpace(p) + if p == "" { + continue + } + result = append(result, resources.ResourceType(p)) + } + return result +} + +func parsePlatformTypes(s string) []resources.PlatformType { + parts := strings.Split(s, ",") + var result []resources.PlatformType + for _, p := range parts { + p = strings.TrimSpace(p) + if p == "" { + continue + } + result = append(result, resources.PlatformType(p)) + } + return result +} + +func parseStatuses(s string) []resources.ResourceStatus { + parts := strings.Split(s, ",") + var result []resources.ResourceStatus + for _, p := range parts { + p = strings.TrimSpace(p) + if p == "" { + continue + } + result = append(result, resources.ResourceStatus(p)) + } + return result +} diff --git a/internal/api/router.go b/internal/api/router.go index 20028e3..e7f0cc3 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -54,6 +54,7 @@ type Router struct { temperatureProxyHandlers *TemperatureProxyHandlers systemSettingsHandler *SystemSettingsHandler aiSettingsHandler *AISettingsHandler + resourceHandlers *ResourceHandlers agentExecServer *agentexec.Server wsHub *websocket.Hub reloadFunc func() error @@ -182,6 +183,7 @@ func (r *Router) setupRoutes() { r.dockerAgentHandlers = NewDockerAgentHandlers(r.monitor, r.wsHub) r.hostAgentHandlers = NewHostAgentHandlers(r.monitor, r.wsHub) r.temperatureProxyHandlers = NewTemperatureProxyHandlers(r.config, r.persistence, r.reloadFunc) + r.resourceHandlers = NewResourceHandlers() // API routes r.mux.HandleFunc("/api/health", r.handleHealth) @@ -221,6 +223,11 @@ func (r *Router) setupRoutes() { r.mux.HandleFunc("/api/backups/pbs", RequireAuth(r.config, RequireScope(config.ScopeMonitoringRead, r.handleBackupsPBS))) r.mux.HandleFunc("/api/snapshots", RequireAuth(r.config, RequireScope(config.ScopeMonitoringRead, r.handleSnapshots))) + // Unified resources API (Phase 1 of unified resource architecture) + r.mux.HandleFunc("/api/resources", RequireAuth(r.config, RequireScope(config.ScopeMonitoringRead, r.resourceHandlers.HandleGetResources))) + r.mux.HandleFunc("/api/resources/stats", RequireAuth(r.config, RequireScope(config.ScopeMonitoringRead, r.resourceHandlers.HandleGetResourceStats))) + r.mux.HandleFunc("/api/resources/", RequireAuth(r.config, RequireScope(config.ScopeMonitoringRead, r.resourceHandlers.HandleGetResource))) + // Guest metadata routes r.mux.HandleFunc("/api/guests/metadata", RequireAuth(r.config, RequireScope(config.ScopeMonitoringRead, guestMetadataHandler.HandleGetMetadata))) r.mux.HandleFunc("/api/guests/metadata/", RequireAuth(r.config, func(w http.ResponseWriter, req *http.Request) { @@ -1029,6 +1036,10 @@ func (r *Router) setupRoutes() { r.aiSettingsHandler.SetAlertProvider(ai.NewAlertManagerAdapter(alertManager)) } } + // Inject unified resource provider for Phase 2 AI context (cleaner, deduplicated view) + if r.resourceHandlers != nil { + r.aiSettingsHandler.SetResourceProvider(r.resourceHandlers.Store()) + } r.mux.HandleFunc("/api/settings/ai", RequireAdmin(r.config, RequireScope(config.ScopeSettingsRead, r.aiSettingsHandler.HandleGetAISettings))) r.mux.HandleFunc("/api/settings/ai/update", RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, r.aiSettingsHandler.HandleUpdateAISettings))) r.mux.HandleFunc("/api/ai/test", RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, r.aiSettingsHandler.HandleTestAIConnection))) @@ -1265,6 +1276,10 @@ func (r *Router) SetMonitor(m *monitoring.Monitor) { mgr.SetPublicURL(url) } } + // Inject resource store for polling optimization + if r.resourceHandlers != nil { + m.SetResourceStore(r.resourceHandlers.Store()) + } } } @@ -2469,6 +2484,13 @@ func (r *Router) handleState(w http.ResponseWriter, req *http.Request) { } state := r.monitor.GetState() + + // Also populate the unified resource store (Phase 1 of unified architecture) + // This runs on every state request to keep resources up-to-date + if r.resourceHandlers != nil { + r.resourceHandlers.PopulateFromSnapshot(state) + } + frontendState := state.ToFrontend() if err := utils.WriteJSONResponse(w, frontendState); err != nil { diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index dacada5..fee7de3 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -80,6 +80,17 @@ type PVEClientInterface interface { GetCephDF(ctx context.Context) (*proxmox.CephDF, error) } +// ResourceStoreInterface provides methods for polling optimization. +// 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 + // because an agent is providing richer data. + ShouldSkipAPIPolling(hostname string) bool + // GetPollingRecommendations returns a map of hostname -> polling multiplier. + // 0 = skip entirely, 0.5 = half frequency, 1 = normal + GetPollingRecommendations() map[string]float64 +} + func getNodeDisplayName(instance *config.PVEInstance, nodeName string) string { baseName := strings.TrimSpace(nodeName) if baseName == "" { @@ -604,6 +615,7 @@ type Monitor struct { pollStatusMap map[string]*pollStatus dlqInsightMap map[string]*dlqInsight nodeLastOnline map[string]time.Time // Track last time each node was seen online (for grace period) + resourceStore ResourceStoreInterface // Optional unified resource store for polling optimization } type rrdMemCacheEntry struct { @@ -6990,6 +7002,16 @@ func (m *Monitor) GetAlertManager() *alerts.Manager { return m.alertManager } +// SetResourceStore sets the resource store for polling optimization. +// When set, the monitor will check if it should reduce polling frequency +// for nodes that have host agents providing data. +func (m *Monitor) SetResourceStore(store ResourceStoreInterface) { + m.mu.Lock() + defer m.mu.Unlock() + m.resourceStore = store + log.Info().Msg("Resource store set for polling optimization") +} + // GetNotificationManager returns the notification manager func (m *Monitor) GetNotificationManager() *notifications.NotificationManager { return m.notificationMgr @@ -7005,6 +7027,27 @@ func (m *Monitor) GetMetricsStore() *metrics.Store { return m.metricsStore } +// shouldSkipNodeMetrics returns true if we should skip detailed metric polling +// for the given node because a host agent is providing richer data. +// This helps reduce API load when agents are active. +func (m *Monitor) shouldSkipNodeMetrics(nodeName string) bool { + m.mu.RLock() + store := m.resourceStore + m.mu.RUnlock() + + if store == nil { + return false + } + + should := store.ShouldSkipAPIPolling(nodeName) + if should { + log.Debug(). + Str("node", nodeName). + Msg("Skipping detailed node metrics - host agent provides data") + } + return should +} + // 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) { diff --git a/internal/resources/converters.go b/internal/resources/converters.go new file mode 100644 index 0000000..f5218c3 --- /dev/null +++ b/internal/resources/converters.go @@ -0,0 +1,847 @@ +package resources + +import ( + "encoding/json" + "fmt" + "strings" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/models" +) + +// FromNode converts a Proxmox Node to a unified Resource. +func FromNode(n models.Node) Resource { + // Calculate primary temperature if available + var temp *float64 + if n.Temperature != nil && n.Temperature.Available && n.Temperature.HasCPU { + // Use CPU package temperature if available, otherwise average core temps + if n.Temperature.CPUPackage > 0 { + temp = &n.Temperature.CPUPackage + } else if len(n.Temperature.Cores) > 0 { + avg := 0.0 + for _, c := range n.Temperature.Cores { + avg += c.Temp + } + avg /= float64(len(n.Temperature.Cores)) + temp = &avg + } + } + + // Build memory metric + var memory *MetricValue + if n.Memory.Total > 0 { + memory = &MetricValue{ + Current: n.Memory.Usage, + Total: &n.Memory.Total, + Used: &n.Memory.Used, + Free: &n.Memory.Free, + } + } + + // Build disk metric + var disk *MetricValue + if n.Disk.Total > 0 { + disk = &MetricValue{ + Current: n.Disk.Usage, + Total: &n.Disk.Total, + Used: &n.Disk.Used, + Free: &n.Disk.Free, + } + } + + // Build platform data + platformData := NodePlatformData{ + Instance: n.Instance, + Host: n.Host, + GuestURL: n.GuestURL, + PVEVersion: n.PVEVersion, + KernelVersion: n.KernelVersion, + LoadAverage: n.LoadAverage, + IsClusterMember: n.IsClusterMember, + ClusterName: n.ClusterName, + ConnectionHealth: n.ConnectionHealth, + CPUInfo: CPUInfo{ + Model: n.CPUInfo.Model, + Cores: n.CPUInfo.Cores, + Sockets: n.CPUInfo.Sockets, + }, + } + platformDataJSON, _ := json.Marshal(platformData) + + // Determine cluster ID + clusterID := "" + if n.IsClusterMember && n.ClusterName != "" { + clusterID = fmt.Sprintf("pve-cluster/%s", n.ClusterName) + } + + return Resource{ + ID: n.ID, + Type: ResourceTypeNode, + Name: n.Name, + DisplayName: n.DisplayName, + PlatformID: n.Instance, + PlatformType: PlatformProxmoxPVE, + SourceType: SourceAPI, + ClusterID: clusterID, + Status: mapNodeStatus(n.Status), + CPU: &MetricValue{ + Current: n.CPU * 100, // Node CPU is 0-1, convert to percentage + }, + Memory: memory, + Disk: disk, + Temperature: temp, + Uptime: &n.Uptime, + LastSeen: n.LastSeen, + PlatformData: platformDataJSON, + Identity: &ResourceIdentity{ + Hostname: n.Name, + }, + SchemaVersion: CurrentSchemaVersion, + } +} + +// FromVM converts a Proxmox VM to a unified Resource. +func FromVM(vm models.VM) Resource { + // Build memory metric + var memory *MetricValue + if vm.Memory.Total > 0 { + memory = &MetricValue{ + Current: vm.Memory.Usage, + Total: &vm.Memory.Total, + Used: &vm.Memory.Used, + Free: &vm.Memory.Free, + } + } + + // Build disk metric + var disk *MetricValue + if vm.Disk.Total > 0 { + disk = &MetricValue{ + Current: vm.Disk.Usage, + Total: &vm.Disk.Total, + Used: &vm.Disk.Used, + Free: &vm.Disk.Free, + } + } + + // Build network metric + network := &NetworkMetric{ + RXBytes: vm.NetworkIn, + TXBytes: vm.NetworkOut, + } + + // Build platform data + var lastBackup *time.Time + if !vm.LastBackup.IsZero() { + lastBackup = &vm.LastBackup + } + platformData := VMPlatformData{ + VMID: vm.VMID, + Node: vm.Node, + Instance: vm.Instance, + CPUs: vm.CPUs, + Template: vm.Template, + Lock: vm.Lock, + AgentVersion: vm.AgentVersion, + OSName: vm.OSName, + OSVersion: vm.OSVersion, + IPAddresses: vm.IPAddresses, + NetworkIn: vm.NetworkIn, + NetworkOut: vm.NetworkOut, + DiskRead: vm.DiskRead, + DiskWrite: vm.DiskWrite, + LastBackup: lastBackup, + } + platformDataJSON, _ := json.Marshal(platformData) + + // Parent is the node + parentID := fmt.Sprintf("%s/node/%s", vm.Instance, vm.Node) + + return Resource{ + ID: vm.ID, + Type: ResourceTypeVM, + Name: vm.Name, + PlatformID: vm.Instance, + PlatformType: PlatformProxmoxPVE, + SourceType: SourceAPI, + ParentID: parentID, + Status: mapGuestStatus(vm.Status), + CPU: &MetricValue{ + Current: vm.CPU * 100, // VM CPU is 0-1, convert to percentage + }, + Memory: memory, + Disk: disk, + Network: network, + Uptime: &vm.Uptime, + Tags: vm.Tags, + LastSeen: vm.LastSeen, + PlatformData: platformDataJSON, + SchemaVersion: CurrentSchemaVersion, + } +} + +// FromContainer converts a Proxmox LXC Container to a unified Resource. +func FromContainer(ct models.Container) Resource { + // Build memory metric + var memory *MetricValue + if ct.Memory.Total > 0 { + memory = &MetricValue{ + Current: ct.Memory.Usage, + Total: &ct.Memory.Total, + Used: &ct.Memory.Used, + Free: &ct.Memory.Free, + } + } + + // Build disk metric + var disk *MetricValue + if ct.Disk.Total > 0 { + disk = &MetricValue{ + Current: ct.Disk.Usage, + Total: &ct.Disk.Total, + Used: &ct.Disk.Used, + Free: &ct.Disk.Free, + } + } + + // Build network metric + network := &NetworkMetric{ + RXBytes: ct.NetworkIn, + TXBytes: ct.NetworkOut, + } + + // Build platform data + var lastBackup *time.Time + if !ct.LastBackup.IsZero() { + lastBackup = &ct.LastBackup + } + platformData := ContainerPlatformData{ + VMID: ct.VMID, + Node: ct.Node, + Instance: ct.Instance, + CPUs: ct.CPUs, + Template: ct.Template, + Lock: ct.Lock, + OSName: ct.OSName, + IPAddresses: ct.IPAddresses, + NetworkIn: ct.NetworkIn, + NetworkOut: ct.NetworkOut, + DiskRead: ct.DiskRead, + DiskWrite: ct.DiskWrite, + LastBackup: lastBackup, + } + platformDataJSON, _ := json.Marshal(platformData) + + // Parent is the node + parentID := fmt.Sprintf("%s/node/%s", ct.Instance, ct.Node) + + return Resource{ + ID: ct.ID, + Type: ResourceTypeContainer, + Name: ct.Name, + PlatformID: ct.Instance, + PlatformType: PlatformProxmoxPVE, + SourceType: SourceAPI, + ParentID: parentID, + Status: mapGuestStatus(ct.Status), + CPU: &MetricValue{ + Current: ct.CPU * 100, // Container CPU is 0-1, convert to percentage + }, + Memory: memory, + Disk: disk, + Network: network, + Uptime: &ct.Uptime, + Tags: ct.Tags, + LastSeen: ct.LastSeen, + PlatformData: platformDataJSON, + SchemaVersion: CurrentSchemaVersion, + } +} + +// FromHost converts a Host agent to a unified Resource. +func FromHost(h models.Host) Resource { + // Build memory metric + var memory *MetricValue + if h.Memory.Total > 0 { + memory = &MetricValue{ + Current: h.Memory.Usage, + Total: &h.Memory.Total, + Used: &h.Memory.Used, + Free: &h.Memory.Free, + } + } + + // Combine disk metrics from multiple disks + var disk *MetricValue + var totalDisk, usedDisk, freeDisk int64 + for _, d := range h.Disks { + totalDisk += d.Total + usedDisk += d.Used + freeDisk += d.Free + } + if totalDisk > 0 { + usage := float64(usedDisk) / float64(totalDisk) * 100 + disk = &MetricValue{ + Current: usage, + Total: &totalDisk, + Used: &usedDisk, + Free: &freeDisk, + } + } + + // Calculate network totals + var rxTotal, txTotal int64 + for _, iface := range h.NetworkInterfaces { + rxTotal += int64(iface.RXBytes) + txTotal += int64(iface.TXBytes) + } + network := &NetworkMetric{ + RXBytes: rxTotal, + TXBytes: txTotal, + } + + // Get primary temperature + var temp *float64 + if len(h.Sensors.TemperatureCelsius) > 0 { + // Pick the first available temperature + for _, t := range h.Sensors.TemperatureCelsius { + temp = &t + break + } + } + + // Build platform data + disks := make([]DiskInfo, len(h.Disks)) + for i, d := range h.Disks { + disks[i] = DiskInfo{ + Mountpoint: d.Mountpoint, + Device: d.Device, + Type: d.Type, + Total: d.Total, + Used: d.Used, + Free: d.Free, + Usage: d.Usage, + } + } + + interfaces := make([]NetworkInterface, len(h.NetworkInterfaces)) + for i, iface := range h.NetworkInterfaces { + interfaces[i] = NetworkInterface{ + Name: iface.Name, + MAC: iface.MAC, + Addresses: iface.Addresses, + RXBytes: iface.RXBytes, + TXBytes: iface.TXBytes, + SpeedMbps: iface.SpeedMbps, + } + } + + diskIO := make([]DiskIOStats, len(h.DiskIO)) + for i, d := range h.DiskIO { + diskIO[i] = DiskIOStats{ + Device: d.Device, + ReadBytes: d.ReadBytes, + WriteBytes: d.WriteBytes, + ReadOps: d.ReadOps, + WriteOps: d.WriteOps, + ReadTimeMs: d.ReadTime, + WriteTimeMs: d.WriteTime, + IOTimeMs: d.IOTime, + } + } + + raid := make([]HostRAIDArray, len(h.RAID)) + for i, r := range h.RAID { + devices := make([]HostRAIDDevice, len(r.Devices)) + for j, d := range r.Devices { + devices[j] = HostRAIDDevice{ + Device: d.Device, + State: d.State, + Slot: d.Slot, + } + } + raid[i] = HostRAIDArray{ + Device: r.Device, + Name: r.Name, + Level: r.Level, + State: r.State, + TotalDevices: r.TotalDevices, + ActiveDevices: r.ActiveDevices, + WorkingDevices: r.WorkingDevices, + FailedDevices: r.FailedDevices, + SpareDevices: r.SpareDevices, + UUID: r.UUID, + Devices: devices, + RebuildPercent: r.RebuildPercent, + RebuildSpeed: r.RebuildSpeed, + } + } + + platformData := HostPlatformData{ + Platform: h.Platform, + OSName: h.OSName, + OSVersion: h.OSVersion, + KernelVersion: h.KernelVersion, + Architecture: h.Architecture, + CPUCount: h.CPUCount, + LoadAverage: h.LoadAverage, + AgentVersion: h.AgentVersion, + IsLegacy: h.IsLegacy, + Disks: disks, + Interfaces: interfaces, + DiskIO: diskIO, + RAID: raid, + TokenID: h.TokenID, + TokenName: h.TokenName, + TokenHint: h.TokenHint, + TokenLastUsedAt: h.TokenLastUsedAt, + Sensors: HostSensorSummary{ + TemperatureCelsius: h.Sensors.TemperatureCelsius, + FanRPM: h.Sensors.FanRPM, + Additional: h.Sensors.Additional, + }, + } + platformDataJSON, _ := json.Marshal(platformData) + + // Collect IPs for identity + var ips []string + for _, iface := range h.NetworkInterfaces { + ips = append(ips, iface.Addresses...) + } + + return Resource{ + ID: h.ID, + Type: ResourceTypeHost, + Name: h.Hostname, + DisplayName: h.DisplayName, + PlatformID: "host-agent", // No specific platform ID for host agents + PlatformType: PlatformHostAgent, + SourceType: SourceAgent, + Status: mapHostStatus(h.Status), + CPU: &MetricValue{ + Current: h.CPUUsage, + }, + Memory: memory, + Disk: disk, + Network: network, + Temperature: temp, + Uptime: &h.UptimeSeconds, + Tags: h.Tags, + LastSeen: h.LastSeen, + PlatformData: platformDataJSON, + Identity: &ResourceIdentity{ + Hostname: h.Hostname, + IPs: ips, + }, + SchemaVersion: CurrentSchemaVersion, + } +} + +// FromDockerHost converts a DockerHost to a unified Resource. +func FromDockerHost(dh models.DockerHost) Resource { + // Build memory metric + var memory *MetricValue + if dh.Memory.Total > 0 { + memory = &MetricValue{ + Current: dh.Memory.Usage, + Total: &dh.Memory.Total, + Used: &dh.Memory.Used, + Free: &dh.Memory.Free, + } + } + + // Combine disk metrics + var disk *MetricValue + var totalDisk, usedDisk, freeDisk int64 + for _, d := range dh.Disks { + totalDisk += d.Total + usedDisk += d.Used + freeDisk += d.Free + } + if totalDisk > 0 { + usage := float64(usedDisk) / float64(totalDisk) * 100 + disk = &MetricValue{ + Current: usage, + Total: &totalDisk, + Used: &usedDisk, + Free: &freeDisk, + } + } + + // Calculate network totals + var rxTotal, txTotal int64 + for _, iface := range dh.NetworkInterfaces { + rxTotal += int64(iface.RXBytes) + txTotal += int64(iface.TXBytes) + } + network := &NetworkMetric{ + RXBytes: rxTotal, + TXBytes: txTotal, + } + + // Build platform data + disks := make([]DiskInfo, len(dh.Disks)) + for i, d := range dh.Disks { + disks[i] = DiskInfo{ + Mountpoint: d.Mountpoint, + Device: d.Device, + Type: d.Type, + Total: d.Total, + Used: d.Used, + Free: d.Free, + Usage: d.Usage, + } + } + + interfaces := make([]NetworkInterface, len(dh.NetworkInterfaces)) + for i, iface := range dh.NetworkInterfaces { + interfaces[i] = NetworkInterface{ + Name: iface.Name, + MAC: iface.MAC, + Addresses: iface.Addresses, + RXBytes: iface.RXBytes, + TXBytes: iface.TXBytes, + SpeedMbps: iface.SpeedMbps, + } + } + + var swarm *DockerSwarmInfo + if dh.Swarm != nil { + swarm = &DockerSwarmInfo{ + NodeID: dh.Swarm.NodeID, + NodeRole: dh.Swarm.NodeRole, + LocalState: dh.Swarm.LocalState, + ControlAvailable: dh.Swarm.ControlAvailable, + ClusterID: dh.Swarm.ClusterID, + ClusterName: dh.Swarm.ClusterName, + Scope: dh.Swarm.Scope, + Error: dh.Swarm.Error, + } + } + + platformData := DockerHostPlatformData{ + AgentID: dh.AgentID, + MachineID: dh.MachineID, + OS: dh.OS, + KernelVersion: dh.KernelVersion, + Architecture: dh.Architecture, + Runtime: dh.Runtime, + RuntimeVersion: dh.RuntimeVersion, + DockerVersion: dh.DockerVersion, + LoadAverage: dh.LoadAverage, + AgentVersion: dh.AgentVersion, + CPUs: dh.CPUs, + IsLegacy: dh.IsLegacy, + Disks: disks, + Interfaces: interfaces, + CustomDisplayName: dh.CustomDisplayName, + Hidden: dh.Hidden, + PendingUninstall: dh.PendingUninstall, + Swarm: swarm, + TokenID: dh.TokenID, + TokenName: dh.TokenName, + TokenHint: dh.TokenHint, + TokenLastUsedAt: dh.TokenLastUsedAt, + } + platformDataJSON, _ := json.Marshal(platformData) + + // Collect IPs for identity + var ips []string + for _, iface := range dh.NetworkInterfaces { + ips = append(ips, iface.Addresses...) + } + + // Determine display name + displayName := dh.DisplayName + if dh.CustomDisplayName != "" { + displayName = dh.CustomDisplayName + } + + // Determine cluster ID from swarm + clusterID := "" + if dh.Swarm != nil && dh.Swarm.ClusterID != "" { + clusterID = fmt.Sprintf("docker-swarm/%s", dh.Swarm.ClusterID) + } + + return Resource{ + ID: dh.ID, + Type: ResourceTypeDockerHost, + Name: dh.Hostname, + DisplayName: displayName, + PlatformID: dh.AgentID, + PlatformType: PlatformDocker, + SourceType: SourceAgent, + ClusterID: clusterID, + Status: mapDockerHostStatus(dh.Status), + CPU: &MetricValue{ + Current: dh.CPUUsage, + }, + Memory: memory, + Disk: disk, + Network: network, + Uptime: &dh.UptimeSeconds, + LastSeen: dh.LastSeen, + PlatformData: platformDataJSON, + Identity: &ResourceIdentity{ + Hostname: dh.Hostname, + MachineID: dh.MachineID, + IPs: ips, + }, + SchemaVersion: CurrentSchemaVersion, + } +} + +// FromDockerContainer converts a DockerContainer to a unified Resource. +func FromDockerContainer(dc models.DockerContainer, hostID, hostName string) Resource { + // Build memory metric + memTotal := dc.MemoryLimit + memUsed := dc.MemoryUsage + var memory *MetricValue + if memTotal > 0 { + memory = &MetricValue{ + Current: dc.MemoryPercent, + Total: &memTotal, + Used: &memUsed, + } + } + + // Build platform data + ports := make([]ContainerPort, len(dc.Ports)) + for i, p := range dc.Ports { + ports[i] = ContainerPort{ + PrivatePort: p.PrivatePort, + PublicPort: p.PublicPort, + Protocol: p.Protocol, + IP: p.IP, + } + } + + networks := make([]ContainerNetwork, len(dc.Networks)) + for i, n := range dc.Networks { + networks[i] = ContainerNetwork{ + Name: n.Name, + IPv4: n.IPv4, + IPv6: n.IPv6, + } + } + + var podman *PodmanContainerInfo + if dc.Podman != nil { + podman = &PodmanContainerInfo{ + PodName: dc.Podman.PodName, + PodID: dc.Podman.PodID, + Infra: dc.Podman.Infra, + ComposeProject: dc.Podman.ComposeProject, + ComposeService: dc.Podman.ComposeService, + } + } + + platformData := DockerContainerPlatformData{ + HostID: hostID, + HostName: hostName, + Image: dc.Image, + State: dc.State, + Status: dc.Status, + Health: dc.Health, + RestartCount: dc.RestartCount, + ExitCode: dc.ExitCode, + CreatedAt: dc.CreatedAt, + StartedAt: dc.StartedAt, + FinishedAt: dc.FinishedAt, + Labels: dc.Labels, + Ports: ports, + Networks: networks, + Podman: podman, + } + platformDataJSON, _ := json.Marshal(platformData) + + // Create unique ID combining host and container + resourceID := fmt.Sprintf("%s/%s", hostID, dc.ID) + + return Resource{ + ID: resourceID, + Type: ResourceTypeDockerContainer, + Name: dc.Name, + PlatformID: hostID, + PlatformType: PlatformDocker, + SourceType: SourceAgent, + ParentID: hostID, + Status: mapDockerContainerStatus(dc.State), + CPU: &MetricValue{ + Current: dc.CPUPercent, + }, + Memory: memory, + Uptime: &dc.UptimeSeconds, + Labels: dc.Labels, + LastSeen: time.Now(), // Containers don't have their own LastSeen + PlatformData: platformDataJSON, + SchemaVersion: CurrentSchemaVersion, + } +} + +// FromPBSInstance converts a PBS instance to a unified Resource. +func FromPBSInstance(pbs models.PBSInstance) Resource { + // Build memory metric + var memory *MetricValue + if pbs.MemoryTotal > 0 { + memory = &MetricValue{ + Current: pbs.Memory, + Total: &pbs.MemoryTotal, + Used: &pbs.MemoryUsed, + } + } + + platformData := PBSPlatformData{ + Host: pbs.Host, + Version: pbs.Version, + ConnectionHealth: pbs.ConnectionHealth, + MemoryUsed: pbs.MemoryUsed, + MemoryTotal: pbs.MemoryTotal, + NumDatastores: len(pbs.Datastores), + } + platformDataJSON, _ := json.Marshal(platformData) + + return Resource{ + ID: pbs.ID, + Type: ResourceTypePBS, + Name: pbs.Name, + PlatformID: pbs.Host, + PlatformType: PlatformProxmoxPBS, + SourceType: SourceAPI, + Status: mapPBSStatus(pbs.Status, pbs.ConnectionHealth), + CPU: &MetricValue{ + Current: pbs.CPU, + }, + Memory: memory, + Uptime: &pbs.Uptime, + LastSeen: pbs.LastSeen, + PlatformData: platformDataJSON, + SchemaVersion: CurrentSchemaVersion, + } +} + +// FromStorage converts a Proxmox Storage to a unified Resource. +func FromStorage(s models.Storage) Resource { + var disk *MetricValue + if s.Total > 0 { + disk = &MetricValue{ + Current: s.Usage, + Total: &s.Total, + Used: &s.Used, + Free: &s.Free, + } + } + + platformData := StoragePlatformData{ + Instance: s.Instance, + Node: s.Node, + Nodes: s.Nodes, + Type: s.Type, + Content: s.Content, + Shared: s.Shared, + Enabled: s.Enabled, + Active: s.Active, + } + platformDataJSON, _ := json.Marshal(platformData) + + status := StatusOnline + if !s.Active { + status = StatusOffline + } else if !s.Enabled { + status = StatusStopped + } + + return Resource{ + ID: s.ID, + Type: ResourceTypeStorage, + Name: s.Name, + PlatformID: s.Instance, + PlatformType: PlatformProxmoxPVE, + SourceType: SourceAPI, + ParentID: fmt.Sprintf("%s/node/%s", s.Instance, s.Node), + Status: status, + Disk: disk, + LastSeen: time.Now(), // Storage doesn't have LastSeen + PlatformData: platformDataJSON, + SchemaVersion: CurrentSchemaVersion, + } +} + +// Status mapping helpers + +func mapNodeStatus(status string) ResourceStatus { + switch strings.ToLower(status) { + case "online": + return StatusOnline + case "offline": + return StatusOffline + default: + return StatusUnknown + } +} + +func mapGuestStatus(status string) ResourceStatus { + switch strings.ToLower(status) { + case "running": + return StatusRunning + case "stopped": + return StatusStopped + case "paused": + return StatusPaused + default: + return StatusUnknown + } +} + +func mapHostStatus(status string) ResourceStatus { + switch strings.ToLower(status) { + case "online": + return StatusOnline + case "offline": + return StatusOffline + case "degraded": + return StatusDegraded + default: + return StatusUnknown + } +} + +func mapDockerHostStatus(status string) ResourceStatus { + switch strings.ToLower(status) { + case "online": + return StatusOnline + case "offline": + return StatusOffline + default: + return StatusUnknown + } +} + +func mapDockerContainerStatus(state string) ResourceStatus { + switch strings.ToLower(state) { + case "running": + return StatusRunning + case "exited", "dead": + return StatusStopped + case "paused": + return StatusPaused + case "restarting", "created": + return StatusUnknown + default: + return StatusUnknown + } +} + +func mapPBSStatus(status, connectionHealth string) ResourceStatus { + if connectionHealth != "healthy" { + return StatusDegraded + } + switch strings.ToLower(status) { + case "online": + return StatusOnline + case "offline": + return StatusOffline + default: + return StatusUnknown + } +} diff --git a/internal/resources/converters_test.go b/internal/resources/converters_test.go new file mode 100644 index 0000000..e4c4a57 --- /dev/null +++ b/internal/resources/converters_test.go @@ -0,0 +1,546 @@ +package resources + +import ( + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/models" +) + +func TestFromNode(t *testing.T) { + node := models.Node{ + ID: "pve1/node/node1", + Name: "node1", + DisplayName: "Production Node 1", + Instance: "pve1", + Host: "https://192.168.1.100:8006", + Status: "online", + CPU: 0.25, // 25% + Memory: models.Memory{ + Total: 16 * 1024 * 1024 * 1024, // 16GB + Used: 8 * 1024 * 1024 * 1024, // 8GB + Free: 8 * 1024 * 1024 * 1024, + Usage: 50.0, + }, + Disk: models.Disk{ + Total: 500 * 1024 * 1024 * 1024, // 500GB + Used: 200 * 1024 * 1024 * 1024, // 200GB + Free: 300 * 1024 * 1024 * 1024, + Usage: 40.0, + }, + Uptime: 86400, + KernelVersion: "6.8.4-2-pve", + PVEVersion: "8.2.2", + IsClusterMember: true, + ClusterName: "production", + LastSeen: time.Now(), + LoadAverage: []float64{1.5, 2.0, 1.8}, + } + + r := FromNode(node) + + // Check basic fields + if r.ID != node.ID { + t.Errorf("Expected ID %s, got %s", node.ID, r.ID) + } + if r.Type != ResourceTypeNode { + t.Errorf("Expected type %s, got %s", ResourceTypeNode, r.Type) + } + if r.Name != node.Name { + t.Errorf("Expected name %s, got %s", node.Name, r.Name) + } + if r.DisplayName != node.DisplayName { + t.Errorf("Expected displayName %s, got %s", node.DisplayName, r.DisplayName) + } + if r.PlatformType != PlatformProxmoxPVE { + t.Errorf("Expected platform %s, got %s", PlatformProxmoxPVE, r.PlatformType) + } + if r.SourceType != SourceAPI { + t.Errorf("Expected source %s, got %s", SourceAPI, r.SourceType) + } + if r.Status != StatusOnline { + t.Errorf("Expected status %s, got %s", StatusOnline, r.Status) + } + + // Check metrics + if r.CPU == nil { + t.Fatal("CPU should not be nil") + } + if r.CPU.Current != 25.0 { // Converted from 0-1 to percentage + t.Errorf("Expected CPU 25%%, got %f%%", r.CPU.Current) + } + + if r.Memory == nil { + t.Fatal("Memory should not be nil") + } + if r.Memory.Current != 50.0 { + t.Errorf("Expected memory 50%%, got %f%%", r.Memory.Current) + } + + if r.Disk == nil { + t.Fatal("Disk should not be nil") + } + if r.Disk.Current != 40.0 { + t.Errorf("Expected disk 40%%, got %f%%", r.Disk.Current) + } + + if r.Uptime == nil || *r.Uptime != 86400 { + t.Errorf("Expected uptime 86400, got %v", r.Uptime) + } + + // Check cluster info + if r.ClusterID != "pve-cluster/production" { + t.Errorf("Expected clusterID pve-cluster/production, got %s", r.ClusterID) + } + + // Check identity + if r.Identity == nil { + t.Fatal("Identity should not be nil") + } + if r.Identity.Hostname != "node1" { + t.Errorf("Expected hostname node1, got %s", r.Identity.Hostname) + } + + // Check platform data + var pd NodePlatformData + if err := r.GetPlatformData(&pd); err != nil { + t.Fatalf("Failed to get platform data: %v", err) + } + if pd.PVEVersion != "8.2.2" { + t.Errorf("Expected PVE version 8.2.2, got %s", pd.PVEVersion) + } + if pd.KernelVersion != "6.8.4-2-pve" { + t.Errorf("Expected kernel version 6.8.4-2-pve, got %s", pd.KernelVersion) + } + if len(pd.LoadAverage) != 3 { + t.Errorf("Expected 3 load average values, got %d", len(pd.LoadAverage)) + } +} + +func TestFromVM(t *testing.T) { + vm := models.VM{ + ID: "pve1/qemu/100", + VMID: 100, + Name: "webserver", + Node: "node1", + Instance: "pve1", + Status: "running", + Type: "qemu", + CPU: 0.15, + CPUs: 4, + Memory: models.Memory{ + Total: 8 * 1024 * 1024 * 1024, + Used: 4 * 1024 * 1024 * 1024, + Free: 4 * 1024 * 1024 * 1024, + Usage: 50.0, + }, + Disk: models.Disk{ + Total: 100 * 1024 * 1024 * 1024, + Used: 30 * 1024 * 1024 * 1024, + Free: 70 * 1024 * 1024 * 1024, + Usage: 30.0, + }, + NetworkIn: 1000000, + NetworkOut: 500000, + DiskRead: 2000000, + DiskWrite: 1000000, + Uptime: 3600, + Tags: []string{"production", "web"}, + LastSeen: time.Now(), + } + + r := FromVM(vm) + + if r.ID != vm.ID { + t.Errorf("Expected ID %s, got %s", vm.ID, r.ID) + } + if r.Type != ResourceTypeVM { + t.Errorf("Expected type %s, got %s", ResourceTypeVM, r.Type) + } + if r.Status != StatusRunning { + t.Errorf("Expected status %s, got %s", StatusRunning, r.Status) + } + if r.ParentID != "pve1/node/node1" { + t.Errorf("Expected parent pve1/node/node1, got %s", r.ParentID) + } + + // Check CPU is converted to percentage + if r.CPU.Current != 15.0 { + t.Errorf("Expected CPU 15%%, got %f%%", r.CPU.Current) + } + + // Check network + if r.Network == nil { + t.Fatal("Network should not be nil") + } + if r.Network.RXBytes != 1000000 { + t.Errorf("Expected RXBytes 1000000, got %d", r.Network.RXBytes) + } + if r.Network.TXBytes != 500000 { + t.Errorf("Expected TXBytes 500000, got %d", r.Network.TXBytes) + } + + // Check tags + if len(r.Tags) != 2 { + t.Errorf("Expected 2 tags, got %d", len(r.Tags)) + } + + // Check platform data + var pd VMPlatformData + if err := r.GetPlatformData(&pd); err != nil { + t.Fatalf("Failed to get platform data: %v", err) + } + if pd.VMID != 100 { + t.Errorf("Expected VMID 100, got %d", pd.VMID) + } + if pd.CPUs != 4 { + t.Errorf("Expected 4 CPUs, got %d", pd.CPUs) + } +} + +func TestFromContainer(t *testing.T) { + ct := models.Container{ + ID: "pve1/lxc/101", + VMID: 101, + Name: "database", + Node: "node1", + Instance: "pve1", + Status: "stopped", + Type: "lxc", + CPU: 0.0, + CPUs: 2, + Memory: models.Memory{ + Total: 4 * 1024 * 1024 * 1024, + Used: 0, + Free: 4 * 1024 * 1024 * 1024, + Usage: 0.0, + }, + Uptime: 0, + Tags: []string{"database"}, + IPAddresses: []string{"192.168.1.50"}, + LastSeen: time.Now(), + } + + r := FromContainer(ct) + + if r.Type != ResourceTypeContainer { + t.Errorf("Expected type %s, got %s", ResourceTypeContainer, r.Type) + } + if r.Status != StatusStopped { + t.Errorf("Expected status %s, got %s", StatusStopped, r.Status) + } + if r.ParentID != "pve1/node/node1" { + t.Errorf("Expected parent pve1/node/node1, got %s", r.ParentID) + } + + var pd ContainerPlatformData + if err := r.GetPlatformData(&pd); err != nil { + t.Fatalf("Failed to get platform data: %v", err) + } + if len(pd.IPAddresses) != 1 || pd.IPAddresses[0] != "192.168.1.50" { + t.Errorf("Expected IP 192.168.1.50, got %v", pd.IPAddresses) + } +} + +func TestFromHost(t *testing.T) { + host := models.Host{ + ID: "host-abc123", + Hostname: "standalone-server", + DisplayName: "Standalone Server", + Platform: "linux", + OSName: "Ubuntu", + OSVersion: "22.04", + KernelVersion: "5.15.0-generic", + Architecture: "amd64", + CPUCount: 8, + CPUUsage: 45.5, + Memory: models.Memory{ + Total: 32 * 1024 * 1024 * 1024, + Used: 16 * 1024 * 1024 * 1024, + Free: 16 * 1024 * 1024 * 1024, + Usage: 50.0, + }, + LoadAverage: []float64{2.5, 2.0, 1.5}, + Disks: []models.Disk{ + { + Mountpoint: "/", + Total: 500 * 1024 * 1024 * 1024, + Used: 200 * 1024 * 1024 * 1024, + Free: 300 * 1024 * 1024 * 1024, + Usage: 40.0, + }, + }, + NetworkInterfaces: []models.HostNetworkInterface{ + { + Name: "eth0", + MAC: "aa:bb:cc:dd:ee:ff", + Addresses: []string{"192.168.1.100", "fe80::1"}, + RXBytes: 10000000, + TXBytes: 5000000, + }, + }, + Sensors: models.HostSensorSummary{ + TemperatureCelsius: map[string]float64{ + "cpu_temp": 55.0, + }, + }, + Status: "online", + UptimeSeconds: 86400 * 30, // 30 days + AgentVersion: "1.2.3", + Tags: []string{"production"}, + LastSeen: time.Now(), + } + + r := FromHost(host) + + if r.Type != ResourceTypeHost { + t.Errorf("Expected type %s, got %s", ResourceTypeHost, r.Type) + } + if r.PlatformType != PlatformHostAgent { + t.Errorf("Expected platform %s, got %s", PlatformHostAgent, r.PlatformType) + } + if r.SourceType != SourceAgent { + t.Errorf("Expected source %s, got %s", SourceAgent, r.SourceType) + } + + // CPU is already a percentage for hosts + if r.CPU.Current != 45.5 { + t.Errorf("Expected CPU 45.5%%, got %f%%", r.CPU.Current) + } + + // Check temperature + if r.Temperature == nil || *r.Temperature != 55.0 { + t.Errorf("Expected temperature 55.0, got %v", r.Temperature) + } + + // Check identity includes IP + if r.Identity == nil { + t.Fatal("Identity should not be nil") + } + if r.Identity.Hostname != "standalone-server" { + t.Errorf("Expected hostname standalone-server, got %s", r.Identity.Hostname) + } + if len(r.Identity.IPs) < 1 { + t.Error("Expected at least 1 IP in identity") + } + + // Check platform data + var pd HostPlatformData + if err := r.GetPlatformData(&pd); err != nil { + t.Fatalf("Failed to get platform data: %v", err) + } + if pd.CPUCount != 8 { + t.Errorf("Expected 8 CPUs, got %d", pd.CPUCount) + } + if pd.AgentVersion != "1.2.3" { + t.Errorf("Expected agent version 1.2.3, got %s", pd.AgentVersion) + } +} + +func TestFromDockerHost(t *testing.T) { + dh := models.DockerHost{ + ID: "docker-host-1", + AgentID: "agent-xyz", + Hostname: "docker-server", + DisplayName: "Docker Server", + MachineID: "machine-id-123", + OS: "linux", + Architecture: "amd64", + Runtime: "docker", + DockerVersion: "24.0.5", + CPUs: 16, + CPUUsage: 35.0, + Memory: models.Memory{ + Total: 64 * 1024 * 1024 * 1024, + Used: 32 * 1024 * 1024 * 1024, + Free: 32 * 1024 * 1024 * 1024, + Usage: 50.0, + }, + UptimeSeconds: 86400, + Status: "online", + LastSeen: time.Now(), + Swarm: &models.DockerSwarmInfo{ + NodeID: "swarm-node-1", + NodeRole: "manager", + ClusterID: "swarm-cluster-1", + ClusterName: "production-swarm", + }, + NetworkInterfaces: []models.HostNetworkInterface{ + { + Name: "docker0", + Addresses: []string{"172.17.0.1"}, + }, + }, + } + + r := FromDockerHost(dh) + + if r.Type != ResourceTypeDockerHost { + t.Errorf("Expected type %s, got %s", ResourceTypeDockerHost, r.Type) + } + if r.PlatformType != PlatformDocker { + t.Errorf("Expected platform %s, got %s", PlatformDocker, r.PlatformType) + } + + // Check cluster ID from swarm + if r.ClusterID != "docker-swarm/swarm-cluster-1" { + t.Errorf("Expected clusterID docker-swarm/swarm-cluster-1, got %s", r.ClusterID) + } + + // Check identity includes machine ID + if r.Identity == nil { + t.Fatal("Identity should not be nil") + } + if r.Identity.MachineID != "machine-id-123" { + t.Errorf("Expected machineID machine-id-123, got %s", r.Identity.MachineID) + } + + var pd DockerHostPlatformData + if err := r.GetPlatformData(&pd); err != nil { + t.Fatalf("Failed to get platform data: %v", err) + } + if pd.DockerVersion != "24.0.5" { + t.Errorf("Expected Docker version 24.0.5, got %s", pd.DockerVersion) + } + if pd.Swarm == nil || pd.Swarm.NodeRole != "manager" { + t.Error("Expected swarm info with manager role") + } +} + +func TestFromDockerContainer(t *testing.T) { + dc := models.DockerContainer{ + ID: "container-123", + Name: "nginx", + Image: "nginx:latest", + State: "running", + Status: "Up 2 hours", + Health: "healthy", + CPUPercent: 5.5, + MemoryUsage: 256 * 1024 * 1024, // 256MB + MemoryLimit: 512 * 1024 * 1024, // 512MB + MemoryPercent: 50.0, + UptimeSeconds: 7200, + RestartCount: 0, + CreatedAt: time.Now().Add(-24 * time.Hour), + Labels: map[string]string{ + "app": "web", + }, + Ports: []models.DockerContainerPort{ + {PrivatePort: 80, PublicPort: 8080, Protocol: "tcp"}, + }, + } + + r := FromDockerContainer(dc, "docker-host-1", "docker-server") + + if r.Type != ResourceTypeDockerContainer { + t.Errorf("Expected type %s, got %s", ResourceTypeDockerContainer, r.Type) + } + if r.ParentID != "docker-host-1" { + t.Errorf("Expected parent docker-host-1, got %s", r.ParentID) + } + if r.Status != StatusRunning { + t.Errorf("Expected status %s, got %s", StatusRunning, r.Status) + } + + if r.CPU.Current != 5.5 { + t.Errorf("Expected CPU 5.5%%, got %f%%", r.CPU.Current) + } + if r.Memory == nil || r.Memory.Current != 50.0 { + t.Errorf("Expected memory 50%%, got %v", r.Memory) + } + + if r.Labels["app"] != "web" { + t.Error("Expected label app=web") + } + + var pd DockerContainerPlatformData + if err := r.GetPlatformData(&pd); err != nil { + t.Fatalf("Failed to get platform data: %v", err) + } + if pd.Image != "nginx:latest" { + t.Errorf("Expected image nginx:latest, got %s", pd.Image) + } + if pd.Health != "healthy" { + t.Errorf("Expected health healthy, got %s", pd.Health) + } + if len(pd.Ports) != 1 || pd.Ports[0].PublicPort != 8080 { + t.Error("Expected port 8080 mapping") + } +} + +func TestResourceMethods(t *testing.T) { + // Test IsInfrastructure + nodeResource := Resource{Type: ResourceTypeNode} + if !nodeResource.IsInfrastructure() { + t.Error("Node should be infrastructure") + } + if nodeResource.IsWorkload() { + t.Error("Node should not be workload") + } + + vmResource := Resource{Type: ResourceTypeVM} + if vmResource.IsInfrastructure() { + t.Error("VM should not be infrastructure") + } + if !vmResource.IsWorkload() { + t.Error("VM should be workload") + } + + // Test EffectiveDisplayName + r := Resource{Name: "name", DisplayName: ""} + if r.EffectiveDisplayName() != "name" { + t.Error("Should return Name when DisplayName is empty") + } + r.DisplayName = "custom" + if r.EffectiveDisplayName() != "custom" { + t.Error("Should return DisplayName when set") + } + + // Test CPUPercent + r = Resource{} + if r.CPUPercent() != 0 { + t.Error("CPUPercent should be 0 when CPU is nil") + } + r.CPU = &MetricValue{Current: 75.5} + if r.CPUPercent() != 75.5 { + t.Errorf("Expected 75.5, got %f", r.CPUPercent()) + } + + // Test MemoryPercent + r = Resource{} + if r.MemoryPercent() != 0 { + t.Error("MemoryPercent should be 0 when Memory is nil") + } + r.Memory = &MetricValue{Current: 60.0} + if r.MemoryPercent() != 60.0 { + t.Errorf("Expected 60.0, got %f", r.MemoryPercent()) + } +} + +func TestStatusMapping(t *testing.T) { + tests := []struct { + input string + expected ResourceStatus + mapper func(string) ResourceStatus + }{ + {"online", StatusOnline, mapNodeStatus}, + {"offline", StatusOffline, mapNodeStatus}, + {"unknown", StatusUnknown, mapNodeStatus}, + {"running", StatusRunning, mapGuestStatus}, + {"stopped", StatusStopped, mapGuestStatus}, + {"paused", StatusPaused, mapGuestStatus}, + {"online", StatusOnline, mapHostStatus}, + {"degraded", StatusDegraded, mapHostStatus}, + {"running", StatusRunning, mapDockerContainerStatus}, + {"exited", StatusStopped, mapDockerContainerStatus}, + {"dead", StatusStopped, mapDockerContainerStatus}, + {"paused", StatusPaused, mapDockerContainerStatus}, + } + + for _, test := range tests { + result := test.mapper(test.input) + if result != test.expected { + t.Errorf("Mapping %s: expected %s, got %s", test.input, test.expected, result) + } + } +} diff --git a/internal/resources/platform_data.go b/internal/resources/platform_data.go new file mode 100644 index 0000000..27974a8 --- /dev/null +++ b/internal/resources/platform_data.go @@ -0,0 +1,287 @@ +package resources + +import "time" + +// NodePlatformData contains Proxmox VE node-specific fields. +// Stored in Resource.PlatformData when Type is ResourceTypeNode. +type NodePlatformData struct { + Instance string `json:"instance"` // Proxmox instance URL + Host string `json:"host"` // Full host URL from config + GuestURL string `json:"guestURL"` // Optional guest-accessible URL + PVEVersion string `json:"pveVersion"` // Proxmox VE version + KernelVersion string `json:"kernelVersion"` // Linux kernel version + CPUInfo CPUInfo `json:"cpuInfo"` // CPU details + LoadAverage []float64 `json:"loadAverage"` // 1, 5, 15 minute load averages + + // Cluster information + IsClusterMember bool `json:"isClusterMember"` + ClusterName string `json:"clusterName"` + + // Connection status + ConnectionHealth string `json:"connectionHealth"` +} + +// CPUInfo contains CPU hardware details. +type CPUInfo struct { + Model string `json:"model"` + Cores int `json:"cores"` + Sockets int `json:"sockets"` +} + +// VMPlatformData contains Proxmox VM-specific fields. +// Stored in Resource.PlatformData when Type is ResourceTypeVM. +type VMPlatformData struct { + VMID int `json:"vmid"` + Node string `json:"node"` // Proxmox node hosting this VM + Instance string `json:"instance"` // Proxmox instance URL + CPUs int `json:"cpus"` // Number of vCPUs + Template bool `json:"template"` + Lock string `json:"lock,omitempty"` // Lock status (backup, migrate, etc.) + AgentVersion string `json:"agentVersion,omitempty"` + OSName string `json:"osName,omitempty"` + OSVersion string `json:"osVersion,omitempty"` + IPAddresses []string `json:"ipAddresses,omitempty"` + + // I/O stats + NetworkIn int64 `json:"networkIn"` + NetworkOut int64 `json:"networkOut"` + DiskRead int64 `json:"diskRead"` + DiskWrite int64 `json:"diskWrite"` + + // Backup info + LastBackup *time.Time `json:"lastBackup,omitempty"` +} + +// ContainerPlatformData contains Proxmox LXC container-specific fields. +// Stored in Resource.PlatformData when Type is ResourceTypeContainer. +type ContainerPlatformData struct { + VMID int `json:"vmid"` + Node string `json:"node"` // Proxmox node hosting this container + Instance string `json:"instance"` // Proxmox instance URL + CPUs int `json:"cpus"` // Number of vCPUs + Template bool `json:"template"` + Lock string `json:"lock,omitempty"` + OSName string `json:"osName,omitempty"` + IPAddresses []string `json:"ipAddresses,omitempty"` + + // I/O stats + NetworkIn int64 `json:"networkIn"` + NetworkOut int64 `json:"networkOut"` + DiskRead int64 `json:"diskRead"` + DiskWrite int64 `json:"diskWrite"` + + // Backup info + LastBackup *time.Time `json:"lastBackup,omitempty"` +} + +// HostPlatformData contains host-agent specific fields. +// Stored in Resource.PlatformData when Type is ResourceTypeHost. +type HostPlatformData struct { + Platform string `json:"platform,omitempty"` // linux, windows, darwin + OSName string `json:"osName,omitempty"` // e.g., "Ubuntu 22.04" + OSVersion string `json:"osVersion,omitempty"` // OS version string + KernelVersion string `json:"kernelVersion,omitempty"` // Kernel version + Architecture string `json:"architecture,omitempty"` // amd64, arm64, etc. + CPUCount int `json:"cpuCount,omitempty"` // Number of CPUs + LoadAverage []float64 `json:"loadAverage,omitempty"` // 1, 5, 15 minute loads + AgentVersion string `json:"agentVersion,omitempty"` // Pulse agent version + IsLegacy bool `json:"isLegacy,omitempty"` // Legacy agent indicator + Sensors HostSensorSummary `json:"sensors,omitempty"` // Temperature/fan sensors + RAID []HostRAIDArray `json:"raid,omitempty"` // RAID arrays + DiskIO []DiskIOStats `json:"diskIO,omitempty"` // Per-disk I/O stats + Disks []DiskInfo `json:"disks,omitempty"` // Disk usage info + Interfaces []NetworkInterface `json:"interfaces,omitempty"` // Network interfaces + + // Token information + TokenID string `json:"tokenId,omitempty"` + TokenName string `json:"tokenName,omitempty"` + TokenHint string `json:"tokenHint,omitempty"` + TokenLastUsedAt *time.Time `json:"tokenLastUsedAt,omitempty"` +} + +// HostSensorSummary captures sensor readings from a host. +type HostSensorSummary struct { + TemperatureCelsius map[string]float64 `json:"temperatureCelsius,omitempty"` + FanRPM map[string]float64 `json:"fanRpm,omitempty"` + Additional map[string]float64 `json:"additional,omitempty"` +} + +// HostRAIDArray represents an mdadm RAID array. +type HostRAIDArray struct { + Device string `json:"device"` + Name string `json:"name,omitempty"` + Level string `json:"level"` + State string `json:"state"` + TotalDevices int `json:"totalDevices"` + ActiveDevices int `json:"activeDevices"` + WorkingDevices int `json:"workingDevices"` + FailedDevices int `json:"failedDevices"` + SpareDevices int `json:"spareDevices"` + UUID string `json:"uuid,omitempty"` + Devices []HostRAIDDevice `json:"devices"` + RebuildPercent float64 `json:"rebuildPercent"` + RebuildSpeed string `json:"rebuildSpeed,omitempty"` +} + +// HostRAIDDevice represents a device in a RAID array. +type HostRAIDDevice struct { + Device string `json:"device"` + State string `json:"state"` + Slot int `json:"slot"` +} + +// DiskIOStats captures I/O statistics for a disk device. +type DiskIOStats struct { + Device string `json:"device"` + ReadBytes uint64 `json:"readBytes,omitempty"` + WriteBytes uint64 `json:"writeBytes,omitempty"` + ReadOps uint64 `json:"readOps,omitempty"` + WriteOps uint64 `json:"writeOps,omitempty"` + ReadTimeMs uint64 `json:"readTimeMs,omitempty"` + WriteTimeMs uint64 `json:"writeTimeMs,omitempty"` + IOTimeMs uint64 `json:"ioTimeMs,omitempty"` +} + +// DiskInfo represents disk/partition usage. +type DiskInfo struct { + Mountpoint string `json:"mountpoint,omitempty"` + Device string `json:"device,omitempty"` + Type string `json:"type,omitempty"` + Total int64 `json:"total"` + Used int64 `json:"used"` + Free int64 `json:"free"` + Usage float64 `json:"usage"` +} + +// NetworkInterface represents a network interface. +type NetworkInterface struct { + Name string `json:"name"` + MAC string `json:"mac,omitempty"` + Addresses []string `json:"addresses,omitempty"` + RXBytes uint64 `json:"rxBytes,omitempty"` + TXBytes uint64 `json:"txBytes,omitempty"` + SpeedMbps *int64 `json:"speedMbps,omitempty"` +} + +// DockerHostPlatformData contains Docker host-specific fields. +// Stored in Resource.PlatformData when Type is ResourceTypeDockerHost. +type DockerHostPlatformData struct { + AgentID string `json:"agentId"` + MachineID string `json:"machineId,omitempty"` + OS string `json:"os,omitempty"` + KernelVersion string `json:"kernelVersion,omitempty"` + Architecture string `json:"architecture,omitempty"` + Runtime string `json:"runtime,omitempty"` // docker, podman + RuntimeVersion string `json:"runtimeVersion,omitempty"` + DockerVersion string `json:"dockerVersion,omitempty"` + LoadAverage []float64 `json:"loadAverage,omitempty"` + AgentVersion string `json:"agentVersion,omitempty"` + CPUs int `json:"cpus"` + IsLegacy bool `json:"isLegacy,omitempty"` + Disks []DiskInfo `json:"disks,omitempty"` + Interfaces []NetworkInterface `json:"interfaces,omitempty"` + CustomDisplayName string `json:"customDisplayName,omitempty"` + Hidden bool `json:"hidden"` + PendingUninstall bool `json:"pendingUninstall"` + + // Swarm information + Swarm *DockerSwarmInfo `json:"swarm,omitempty"` + + // Token information + TokenID string `json:"tokenId,omitempty"` + TokenName string `json:"tokenName,omitempty"` + TokenHint string `json:"tokenHint,omitempty"` + TokenLastUsedAt *time.Time `json:"tokenLastUsedAt,omitempty"` +} + +// DockerSwarmInfo captures Docker Swarm membership details. +type DockerSwarmInfo struct { + NodeID string `json:"nodeId,omitempty"` + NodeRole string `json:"nodeRole,omitempty"` + LocalState string `json:"localState,omitempty"` + ControlAvailable bool `json:"controlAvailable,omitempty"` + ClusterID string `json:"clusterId,omitempty"` + ClusterName string `json:"clusterName,omitempty"` + Scope string `json:"scope,omitempty"` + Error string `json:"error,omitempty"` +} + +// DockerContainerPlatformData contains Docker container-specific fields. +// Stored in Resource.PlatformData when Type is ResourceTypeDockerContainer. +type DockerContainerPlatformData struct { + HostID string `json:"hostId"` // Parent Docker host ID + HostName string `json:"hostName"` // Parent Docker host name + Image string `json:"image"` // Container image + State string `json:"state"` // created, running, paused, restarting, exited, dead + Status string `json:"status"` // Human-readable status + Health string `json:"health"` // healthy, unhealthy, starting, none + RestartCount int `json:"restartCount"` + ExitCode int `json:"exitCode"` + CreatedAt time.Time `json:"createdAt"` + StartedAt *time.Time `json:"startedAt,omitempty"` + FinishedAt *time.Time `json:"finishedAt,omitempty"` + Labels map[string]string `json:"labels,omitempty"` + Ports []ContainerPort `json:"ports,omitempty"` + Networks []ContainerNetwork `json:"networks,omitempty"` + + // Podman-specific + Podman *PodmanContainerInfo `json:"podman,omitempty"` +} + +// ContainerPort describes a port mapping. +type ContainerPort struct { + PrivatePort int `json:"privatePort"` + PublicPort int `json:"publicPort,omitempty"` + Protocol string `json:"protocol"` + IP string `json:"ip,omitempty"` +} + +// ContainerNetwork describes a container's network attachment. +type ContainerNetwork struct { + Name string `json:"name"` + IPv4 string `json:"ipv4,omitempty"` + IPv6 string `json:"ipv6,omitempty"` +} + +// PodmanContainerInfo captures Podman-specific container info. +type PodmanContainerInfo struct { + PodName string `json:"podName,omitempty"` + PodID string `json:"podId,omitempty"` + Infra bool `json:"infra,omitempty"` + ComposeProject string `json:"composeProject,omitempty"` + ComposeService string `json:"composeService,omitempty"` +} + +// PBSPlatformData contains PBS-specific fields. +// Stored in Resource.PlatformData when Type is ResourceTypePBS. +type PBSPlatformData struct { + Host string `json:"host"` + Version string `json:"version"` + ConnectionHealth string `json:"connectionHealth"` + MemoryUsed int64 `json:"memoryUsed"` + MemoryTotal int64 `json:"memoryTotal"` + NumDatastores int `json:"numDatastores"` +} + +// DatastorePlatformData contains PBS datastore-specific fields. +// Stored in Resource.PlatformData when Type is ResourceTypeDatastore. +type DatastorePlatformData struct { + PBSInstanceID string `json:"pbsInstanceId"` + PBSInstanceName string `json:"pbsInstanceName"` + Content string `json:"content,omitempty"` + Error string `json:"error,omitempty"` + DeduplicationFactor float64 `json:"deduplicationFactor,omitempty"` +} + +// StoragePlatformData contains Proxmox storage-specific fields. +// Stored in Resource.PlatformData when Type is ResourceTypeStorage. +type StoragePlatformData struct { + Instance string `json:"instance"` + Node string `json:"node"` // Primary node + Nodes []string `json:"nodes,omitempty"` // All nodes (for shared storage) + Type string `json:"type"` // zfspool, lvmthin, cephfs, etc. + Content string `json:"content"` + Shared bool `json:"shared"` + Enabled bool `json:"enabled"` + Active bool `json:"active"` +} diff --git a/internal/resources/resource.go b/internal/resources/resource.go new file mode 100644 index 0000000..d51fa5f --- /dev/null +++ b/internal/resources/resource.go @@ -0,0 +1,249 @@ +// Package resources provides a unified abstraction for all monitored entities +// across different platforms (Proxmox, Docker, Kubernetes, TrueNAS, etc.). +// +// The Resource type is the core abstraction that normalizes platform-specific +// types like Node, Host, DockerHost, VM, Container into a common data model. +// This enables: +// - AI intelligence across all platforms +// - Elimination of duplicate monitoring (one machine = one set of alerts) +// - Extensibility for new platforms +// - Foundation for unified views +package resources + +import ( + "encoding/json" + "time" +) + +// Resource is the universal abstraction for any monitored entity. +// All platform-specific types (Node, VM, Container, DockerHost, etc.) can be +// converted to this common type for unified handling. +type Resource struct { + // Identity + ID string `json:"id"` // Globally unique ID + Type ResourceType `json:"type"` // vm, container, docker-container, pod, host, etc. + Name string `json:"name"` // Human-readable name + DisplayName string `json:"displayName"` // Custom display name (if set) + + // Platform/Source + PlatformID string `json:"platformId"` // Which platform instance (e.g., cluster URL) + PlatformType PlatformType `json:"platformType"` // proxmox-pve, docker, kubernetes, etc. + SourceType SourceType `json:"sourceType"` // api, agent, hybrid + + // Hierarchy + ParentID string `json:"parentId,omitempty"` // VM → Node, Pod → K8s Node + ClusterID string `json:"clusterId,omitempty"` // Cluster membership + + // Universal Metrics (nullable - not all resources have all metrics) + Status ResourceStatus `json:"status"` // online, offline, running, stopped, degraded + CPU *MetricValue `json:"cpu,omitempty"` // CPU usage percentage + Memory *MetricValue `json:"memory,omitempty"` // Memory usage + Disk *MetricValue `json:"disk,omitempty"` // Primary disk usage + Network *NetworkMetric `json:"network,omitempty"` // Network I/O + Temperature *float64 `json:"temperature,omitempty"` // Temperature in Celsius + Uptime *int64 `json:"uptime,omitempty"` // Uptime in seconds + + // Universal Metadata + Tags []string `json:"tags,omitempty"` + Labels map[string]string `json:"labels,omitempty"` + LastSeen time.Time `json:"lastSeen"` + Alerts []ResourceAlert `json:"alerts,omitempty"` + + // Platform-Specific Data (discriminated by Type) + // This preserves all the rich data while allowing common handling. + // Use GetPlatformData() to unmarshal into the appropriate type. + PlatformData json.RawMessage `json:"platformData,omitempty"` + + // Identity information for deduplication + Identity *ResourceIdentity `json:"identity,omitempty"` + + // Schema version for future evolution + SchemaVersion int `json:"schemaVersion"` +} + +// ResourceType identifies the kind of monitored entity. +type ResourceType string + +const ( + // Infrastructure - physical or virtual hosts that run workloads + ResourceTypeNode ResourceType = "node" // Proxmox VE node + ResourceTypeHost ResourceType = "host" // Standalone host (via host-agent) + ResourceTypeDockerHost ResourceType = "docker-host" // Docker/Podman host + ResourceTypeK8sNode ResourceType = "k8s-node" // Kubernetes node + ResourceTypeTrueNAS ResourceType = "truenas" // TrueNAS system + + // Compute Workloads - individual running instances + ResourceTypeVM ResourceType = "vm" // Proxmox VM + ResourceTypeContainer ResourceType = "container" // LXC container + ResourceTypeDockerContainer ResourceType = "docker-container" // Docker container + ResourceTypePod ResourceType = "pod" // Kubernetes pod + ResourceTypeJail ResourceType = "jail" // BSD jail / TrueNAS jail + + // Services - logical groupings of workloads + ResourceTypeDockerService ResourceType = "docker-service" // Docker Swarm service + ResourceTypeK8sDeployment ResourceType = "k8s-deployment" // Kubernetes deployment + ResourceTypeK8sService ResourceType = "k8s-service" // Kubernetes service + + // Storage - storage resources + ResourceTypeStorage ResourceType = "storage" // Generic storage + ResourceTypeDatastore ResourceType = "datastore" // PBS datastore + ResourceTypePool ResourceType = "pool" // ZFS/Ceph pool + ResourceTypeDataset ResourceType = "dataset" // ZFS dataset + + // Backup Systems + ResourceTypePBS ResourceType = "pbs" // Proxmox Backup Server + ResourceTypePMG ResourceType = "pmg" // Proxmox Mail Gateway +) + +// PlatformType identifies the source platform/system. +type PlatformType string + +const ( + PlatformProxmoxPVE PlatformType = "proxmox-pve" + PlatformProxmoxPBS PlatformType = "proxmox-pbs" + PlatformProxmoxPMG PlatformType = "proxmox-pmg" + PlatformDocker PlatformType = "docker" + PlatformKubernetes PlatformType = "kubernetes" + PlatformTrueNAS PlatformType = "truenas" + PlatformHostAgent PlatformType = "host-agent" +) + +// SourceType indicates how data is collected for this resource. +type SourceType string + +const ( + SourceAPI SourceType = "api" // Data from polling an API + SourceAgent SourceType = "agent" // Data pushed from agent + SourceHybrid SourceType = "hybrid" // Both sources, agent preferred +) + +// ResourceStatus represents the operational state of a resource. +type ResourceStatus string + +const ( + StatusOnline ResourceStatus = "online" + StatusOffline ResourceStatus = "offline" + StatusRunning ResourceStatus = "running" + StatusStopped ResourceStatus = "stopped" + StatusDegraded ResourceStatus = "degraded" + StatusPaused ResourceStatus = "paused" + StatusUnknown ResourceStatus = "unknown" +) + +// MetricValue represents a metric with current value and optional limits. +type MetricValue struct { + Current float64 `json:"current"` // Current value (percentage for CPU, bytes for memory/disk) + Total *int64 `json:"total,omitempty"` // Total capacity (bytes) - nil for percentages like CPU + Used *int64 `json:"used,omitempty"` // Used amount (bytes) - nil for percentages + Free *int64 `json:"free,omitempty"` // Free amount (bytes) - nil for percentages +} + +// NetworkMetric captures network I/O. +type NetworkMetric struct { + RXBytes int64 `json:"rxBytes"` // Total bytes received + TXBytes int64 `json:"txBytes"` // Total bytes transmitted +} + +// ResourceAlert represents an alert associated with a resource. +type ResourceAlert struct { + ID string `json:"id"` + Type string `json:"type"` // cpu, memory, disk, temperature, etc. + Level string `json:"level"` // warning, critical + Message string `json:"message"` + Value float64 `json:"value"` + Threshold float64 `json:"threshold"` + StartTime time.Time `json:"startTime"` +} + +// ResourceIdentity contains information used for deduplication. +// When multiple sources report on the same physical machine, we use +// these fields to identify and merge them. +type ResourceIdentity struct { + Hostname string `json:"hostname,omitempty"` // Primary identifier + MachineID string `json:"machineId,omitempty"` // /etc/machine-id or equivalent + IPs []string `json:"ips,omitempty"` // Network addresses +} + +// CurrentSchemaVersion is the current version of the Resource schema. +const CurrentSchemaVersion = 1 + +// GetPlatformData unmarshals the PlatformData into the provided type. +// Example: var nodeData NodePlatformData; r.GetPlatformData(&nodeData) +func (r *Resource) GetPlatformData(v interface{}) error { + if r.PlatformData == nil { + return nil + } + return json.Unmarshal(r.PlatformData, v) +} + +// SetPlatformData marshals the provided value into PlatformData. +func (r *Resource) SetPlatformData(v interface{}) error { + data, err := json.Marshal(v) + if err != nil { + return err + } + r.PlatformData = data + return nil +} + +// IsInfrastructure returns true if this resource is an infrastructure host +// (node, host, docker-host) rather than a workload (vm, container). +func (r *Resource) IsInfrastructure() bool { + switch r.Type { + case ResourceTypeNode, ResourceTypeHost, ResourceTypeDockerHost, ResourceTypeK8sNode, ResourceTypeTrueNAS: + return true + default: + return false + } +} + +// IsWorkload returns true if this resource is a workload (vm, container, pod) +// rather than infrastructure. +func (r *Resource) IsWorkload() bool { + switch r.Type { + case ResourceTypeVM, ResourceTypeContainer, ResourceTypeDockerContainer, ResourceTypePod, ResourceTypeJail: + return true + default: + return false + } +} + +// EffectiveDisplayName returns DisplayName if set, otherwise Name. +func (r *Resource) EffectiveDisplayName() string { + if r.DisplayName != "" { + return r.DisplayName + } + return r.Name +} + +// CPUPercent returns the CPU usage as a percentage, or 0 if not available. +func (r *Resource) CPUPercent() float64 { + if r.CPU == nil { + return 0 + } + return r.CPU.Current +} + +// MemoryPercent returns the memory usage as a percentage, or 0 if not available. +func (r *Resource) MemoryPercent() float64 { + if r.Memory == nil { + return 0 + } + // If we have used/total, calculate percentage + if r.Memory.Total != nil && *r.Memory.Total > 0 && r.Memory.Used != nil { + return float64(*r.Memory.Used) / float64(*r.Memory.Total) * 100 + } + // Otherwise, Current is the percentage + return r.Memory.Current +} + +// DiskPercent returns the disk usage as a percentage, or 0 if not available. +func (r *Resource) DiskPercent() float64 { + if r.Disk == nil { + return 0 + } + if r.Disk.Total != nil && *r.Disk.Total > 0 && r.Disk.Used != nil { + return float64(*r.Disk.Used) / float64(*r.Disk.Total) * 100 + } + return r.Disk.Current +} diff --git a/internal/resources/store.go b/internal/resources/store.go new file mode 100644 index 0000000..b24dba0 --- /dev/null +++ b/internal/resources/store.go @@ -0,0 +1,1016 @@ +package resources + +import ( + "strings" + "sync" + "time" +) + +// Store maintains a unified collection of all resources with deduplication. +type Store struct { + mu sync.RWMutex + resources map[string]*Resource // Keyed by Resource.ID + + // Index by identity for deduplication + byHostname map[string][]string // hostname (lower) -> resource IDs + byMachineID map[string]string // machine-id -> resource ID + byIP map[string][]string // IP -> resource IDs + + // Track merged resources (one source is preferred over another) + mergedFrom map[string]string // suppressed ID -> preferred ID +} + +// NewStore creates a new resource store. +func NewStore() *Store { + return &Store{ + resources: make(map[string]*Resource), + byHostname: make(map[string][]string), + byMachineID: make(map[string]string), + byIP: make(map[string][]string), + mergedFrom: make(map[string]string), + } +} + +// Upsert adds or updates a resource in the store, performing deduplication. +// Returns the ID of the resource (may differ if merged with existing). +func (s *Store) Upsert(r Resource) string { + s.mu.Lock() + defer s.mu.Unlock() + + // Check for duplicates + if r.Identity != nil { + if existingID := s.findDuplicate(&r); existingID != "" { + // Found a duplicate - determine which to prefer + existing := s.resources[existingID] + preferred := s.preferredResource(existing, &r) + + if preferred == &r { + // New resource is preferred, replace the old one + s.removeFromIndexes(existing) + delete(s.resources, existingID) + s.mergedFrom[existingID] = r.ID + } else { + // Existing resource is preferred, mark this as merged + s.mergedFrom[r.ID] = existingID + return existingID + } + } + } + + // Add/update the resource + s.resources[r.ID] = &r + s.addToIndexes(&r) + + return r.ID +} + +// Get retrieves a resource by ID. +func (s *Store) Get(id string) (*Resource, bool) { + s.mu.RLock() + defer s.mu.RUnlock() + + // Check if this ID was merged into another + if preferredID, merged := s.mergedFrom[id]; merged { + r, ok := s.resources[preferredID] + return r, ok + } + + r, ok := s.resources[id] + return r, ok +} + +// GetAll returns all resources (excluding suppressed duplicates). +func (s *Store) GetAll() []Resource { + s.mu.RLock() + defer s.mu.RUnlock() + + result := make([]Resource, 0, len(s.resources)) + for _, r := range s.resources { + result = append(result, *r) + } + return result +} + +// GetByType returns all resources of a specific type. +func (s *Store) GetByType(t ResourceType) []Resource { + s.mu.RLock() + defer s.mu.RUnlock() + + var result []Resource + for _, r := range s.resources { + if r.Type == t { + result = append(result, *r) + } + } + return result +} + +// GetByPlatform returns all resources from a specific platform. +func (s *Store) GetByPlatform(p PlatformType) []Resource { + s.mu.RLock() + defer s.mu.RUnlock() + + var result []Resource + for _, r := range s.resources { + if r.PlatformType == p { + result = append(result, *r) + } + } + return result +} + +// GetInfrastructure returns all infrastructure resources (nodes, hosts). +func (s *Store) GetInfrastructure() []Resource { + s.mu.RLock() + defer s.mu.RUnlock() + + var result []Resource + for _, r := range s.resources { + if r.IsInfrastructure() { + result = append(result, *r) + } + } + return result +} + +// GetWorkloads returns all workload resources (VMs, containers). +func (s *Store) GetWorkloads() []Resource { + s.mu.RLock() + defer s.mu.RUnlock() + + var result []Resource + for _, r := range s.resources { + if r.IsWorkload() { + result = append(result, *r) + } + } + return result +} + +// GetChildren returns all resources with the specified parent ID. +func (s *Store) GetChildren(parentID string) []Resource { + s.mu.RLock() + defer s.mu.RUnlock() + + var result []Resource + for _, r := range s.resources { + if r.ParentID == parentID { + result = append(result, *r) + } + } + return result +} + +// Remove removes a resource from the store. +func (s *Store) Remove(id string) { + s.mu.Lock() + defer s.mu.Unlock() + + if r, ok := s.resources[id]; ok { + s.removeFromIndexes(r) + delete(s.resources, id) + } + + // Also clean up any merge references + delete(s.mergedFrom, id) + for k, v := range s.mergedFrom { + if v == id { + delete(s.mergedFrom, k) + } + } +} + +// IsSuppressed returns true if the resource ID has been merged/suppressed by another. +func (s *Store) IsSuppressed(id string) bool { + s.mu.RLock() + defer s.mu.RUnlock() + + _, suppressed := s.mergedFrom[id] + return suppressed +} + +// GetPreferredID returns the preferred resource ID if this one is suppressed. +func (s *Store) GetPreferredID(id string) string { + s.mu.RLock() + defer s.mu.RUnlock() + + if preferredID, ok := s.mergedFrom[id]; ok { + return preferredID + } + return id +} + +// GetStats returns statistics about the store. +func (s *Store) GetStats() StoreStats { + s.mu.RLock() + defer s.mu.RUnlock() + + stats := StoreStats{ + TotalResources: len(s.resources), + SuppressedResources: len(s.mergedFrom), + ByType: make(map[ResourceType]int), + ByPlatform: make(map[PlatformType]int), + } + + for _, r := range s.resources { + stats.ByType[r.Type]++ + stats.ByPlatform[r.PlatformType]++ + } + + return stats +} + +// StoreStats contains statistics about the resource store. +type StoreStats struct { + TotalResources int + SuppressedResources int + ByType map[ResourceType]int + ByPlatform map[PlatformType]int +} + +// GetPreferredResourceFor returns the preferred resource for a given ID. +// If the ID was merged into another resource, returns that preferred resource. +// If not merged, returns the resource itself. Returns nil if not found. +func (s *Store) GetPreferredResourceFor(resourceID string) *Resource { + s.mu.RLock() + defer s.mu.RUnlock() + + // Check if this ID was merged + if preferredID, merged := s.mergedFrom[resourceID]; merged { + if r, ok := s.resources[preferredID]; ok { + return r + } + } + + // Return the resource itself if it exists + if r, ok := s.resources[resourceID]; ok { + return r + } + return nil +} + +// IsSamePhysicalMachine checks if two resource IDs represent the same physical machine. +// This is useful for alert deduplication between different source types (API vs Agent). +func (s *Store) IsSamePhysicalMachine(id1, id2 string) bool { + s.mu.RLock() + defer s.mu.RUnlock() + + // Check if they're literally the same + if id1 == id2 { + return true + } + + // Check if both map to the same preferred resource + preferred1 := id1 + if pid, merged := s.mergedFrom[id1]; merged { + preferred1 = pid + } + + preferred2 := id2 + if pid, merged := s.mergedFrom[id2]; merged { + preferred2 = pid + } + + return preferred1 == preferred2 +} + +// HasPreferredSourceForHostname checks if there's a resource with a preferred source +// (like host-agent) monitoring a machine with the given hostname. +// This helps alert managers determine if they should skip alerts from API sources. +func (s *Store) HasPreferredSourceForHostname(hostname string) bool { + if hostname == "" { + return false + } + + s.mu.RLock() + defer s.mu.RUnlock() + + hostnameLower := strings.ToLower(hostname) + resourceIDs, exists := s.byHostname[hostnameLower] + if !exists { + return false + } + + // Check if any resource with this hostname has a preferred source type + for _, id := range resourceIDs { + if r, ok := s.resources[id]; ok { + // Host agent and Docker agent sources are preferred over API + if r.SourceType == SourceAgent || r.SourceType == SourceHybrid { + return true + } + } + } + + return false +} + +// ============================================================================ +// Polling Optimization Methods (Phase 3) +// These methods help reduce redundant API polling when agents are active +// ============================================================================ + +// ShouldSkipAPIPolling returns true if API polling should be skipped for the +// given hostname because an agent is providing richer, more frequent data. +// This is useful for reducing load when both Proxmox API and host agents monitor +// the same machine. +func (s *Store) ShouldSkipAPIPolling(hostname string) bool { + return s.HasPreferredSourceForHostname(hostname) +} + +// GetAgentMonitoredHostnames returns a list of hostnames that are being monitored +// by agents (host-agent, docker-agent). This can be used by the monitoring loop +// to adjust polling behavior for these hosts. +func (s *Store) GetAgentMonitoredHostnames() []string { + s.mu.RLock() + defer s.mu.RUnlock() + + var hostnames []string + seen := make(map[string]bool) + + for _, r := range s.resources { + if r.SourceType != SourceAgent && r.SourceType != SourceHybrid { + continue + } + if r.Identity == nil || r.Identity.Hostname == "" { + continue + } + hostnameLower := strings.ToLower(r.Identity.Hostname) + if !seen[hostnameLower] { + seen[hostnameLower] = true + hostnames = append(hostnames, r.Identity.Hostname) + } + } + + return hostnames +} + +// GetPollingRecommendations returns recommendations for optimizing polling. +// Returns a map of hostname -> recommended polling interval multiplier. +// - Value of 0 means skip entirely (agent provides all needed data) +// - Value > 1 means reduce frequency (poll less often) +// - Value of 1 means normal polling +func (s *Store) GetPollingRecommendations() map[string]float64 { + s.mu.RLock() + defer s.mu.RUnlock() + + recommendations := make(map[string]float64) + + for _, r := range s.resources { + if r.Identity == nil || r.Identity.Hostname == "" { + continue + } + hostname := strings.ToLower(r.Identity.Hostname) + + switch r.SourceType { + case SourceAgent: + // Agent provides all data - skip API polling for metrics + // (we may still want to poll for cluster-wide info) + recommendations[hostname] = 0 + case SourceHybrid: + // Hybrid mode - reduce frequency but don't skip + recommendations[hostname] = 0.5 // Poll at half frequency + } + } + + return recommendations +} + +// findDuplicate looks for an existing resource that represents the same machine. +// Must be called with the lock held. +func (s *Store) findDuplicate(r *Resource) string { + if r.Identity == nil { + return "" + } + + // 1. Machine ID match (most reliable) + if r.Identity.MachineID != "" { + if existingID, ok := s.byMachineID[r.Identity.MachineID]; ok && existingID != r.ID { + return existingID + } + } + + // 2. Hostname match (case-insensitive) + if r.Identity.Hostname != "" { + hostnameLower := strings.ToLower(r.Identity.Hostname) + if existingIDs, ok := s.byHostname[hostnameLower]; ok { + for _, existingID := range existingIDs { + if existingID != r.ID { + existing := s.resources[existingID] + // Only match infrastructure resources with each other + if existing.IsInfrastructure() && r.IsInfrastructure() { + return existingID + } + } + } + } + } + + // 3. IP overlap (if same non-localhost IP, likely same machine) + for _, ip := range r.Identity.IPs { + if isLocalhost(ip) { + continue + } + if existingIDs, ok := s.byIP[ip]; ok { + for _, existingID := range existingIDs { + if existingID != r.ID { + existing := s.resources[existingID] + // Only match infrastructure resources with each other + if existing.IsInfrastructure() && r.IsInfrastructure() { + return existingID + } + } + } + } + } + + return "" +} + +// preferredResource determines which of two duplicate resources should be kept. +// Agent data is preferred over API data. +func (s *Store) preferredResource(a, b *Resource) *Resource { + // Prefer agent over API + aScore := s.sourceScore(a.SourceType) + bScore := s.sourceScore(b.SourceType) + + if aScore > bScore { + return a + } + if bScore > aScore { + return b + } + + // Same source type - prefer the one with more recent data + if a.LastSeen.After(b.LastSeen) { + return a + } + return b +} + +func (s *Store) sourceScore(st SourceType) int { + switch st { + case SourceAgent: + return 3 // Agent data is most preferred + case SourceHybrid: + return 2 // Hybrid is second + case SourceAPI: + return 1 // API is least preferred + default: + return 0 + } +} + +func (s *Store) addToIndexes(r *Resource) { + if r.Identity == nil { + return + } + + if r.Identity.MachineID != "" { + s.byMachineID[r.Identity.MachineID] = r.ID + } + + if r.Identity.Hostname != "" { + hostnameLower := strings.ToLower(r.Identity.Hostname) + s.byHostname[hostnameLower] = append(s.byHostname[hostnameLower], r.ID) + } + + for _, ip := range r.Identity.IPs { + if !isLocalhost(ip) { + s.byIP[ip] = append(s.byIP[ip], r.ID) + } + } +} + +func (s *Store) removeFromIndexes(r *Resource) { + if r.Identity == nil { + return + } + + if r.Identity.MachineID != "" { + if s.byMachineID[r.Identity.MachineID] == r.ID { + delete(s.byMachineID, r.Identity.MachineID) + } + } + + if r.Identity.Hostname != "" { + hostnameLower := strings.ToLower(r.Identity.Hostname) + s.byHostname[hostnameLower] = removeFromSlice(s.byHostname[hostnameLower], r.ID) + if len(s.byHostname[hostnameLower]) == 0 { + delete(s.byHostname, hostnameLower) + } + } + + for _, ip := range r.Identity.IPs { + s.byIP[ip] = removeFromSlice(s.byIP[ip], r.ID) + if len(s.byIP[ip]) == 0 { + delete(s.byIP, ip) + } + } +} + +func removeFromSlice(slice []string, item string) []string { + result := make([]string, 0, len(slice)) + for _, s := range slice { + if s != item { + result = append(result, s) + } + } + return result +} + +func isLocalhost(ip string) bool { + return ip == "127.0.0.1" || ip == "::1" || strings.HasPrefix(ip, "127.") +} + +// MarkStale marks resources that haven't been updated recently. +func (s *Store) MarkStale(threshold time.Duration) []string { + s.mu.Lock() + defer s.mu.Unlock() + + now := time.Now() + var staleIDs []string + + for id, r := range s.resources { + if now.Sub(r.LastSeen) > threshold { + // Mark as offline/degraded + if r.Status == StatusOnline || r.Status == StatusRunning { + r.Status = StatusDegraded + staleIDs = append(staleIDs, id) + } + } + } + + return staleIDs +} + +// PruneStale removes resources that have been stale for too long. +func (s *Store) PruneStale(staleThreshold, removeThreshold time.Duration) []string { + s.mu.Lock() + defer s.mu.Unlock() + + now := time.Now() + var removedIDs []string + + for id, r := range s.resources { + if now.Sub(r.LastSeen) > removeThreshold { + s.removeFromIndexes(r) + delete(s.resources, id) + removedIDs = append(removedIDs, id) + } + } + + return removedIDs +} + +// Query provides a fluent interface for querying resources. +func (s *Store) Query() *ResourceQuery { + return &ResourceQuery{store: s} +} + +// ResourceQuery provides a fluent query interface. +type ResourceQuery struct { + store *Store + types []ResourceType + platforms []PlatformType + statuses []ResourceStatus + parentID *string + clusterID *string + hasAlerts *bool + sortBy string + sortDesc bool + limit int + offset int +} + +// OfType filters by resource types. +func (q *ResourceQuery) OfType(types ...ResourceType) *ResourceQuery { + q.types = types + return q +} + +// FromPlatform filters by platform types. +func (q *ResourceQuery) FromPlatform(platforms ...PlatformType) *ResourceQuery { + q.platforms = platforms + return q +} + +// WithStatus filters by resource status. +func (q *ResourceQuery) WithStatus(statuses ...ResourceStatus) *ResourceQuery { + q.statuses = statuses + return q +} + +// WithParent filters by parent ID. +func (q *ResourceQuery) WithParent(parentID string) *ResourceQuery { + q.parentID = &parentID + return q +} + +// InCluster filters by cluster ID. +func (q *ResourceQuery) InCluster(clusterID string) *ResourceQuery { + q.clusterID = &clusterID + return q +} + +// WithAlerts filters to resources that have active alerts. +func (q *ResourceQuery) WithAlerts() *ResourceQuery { + hasAlerts := true + q.hasAlerts = &hasAlerts + return q +} + +// SortBy sets the sort field. +func (q *ResourceQuery) SortBy(field string, desc bool) *ResourceQuery { + q.sortBy = field + q.sortDesc = desc + return q +} + +// Limit sets the maximum number of results. +func (q *ResourceQuery) Limit(n int) *ResourceQuery { + q.limit = n + return q +} + +// Offset sets the offset for pagination. +func (q *ResourceQuery) Offset(n int) *ResourceQuery { + q.offset = n + return q +} + +// Execute runs the query and returns matching resources. +func (q *ResourceQuery) Execute() []Resource { + q.store.mu.RLock() + defer q.store.mu.RUnlock() + + var results []Resource + + for _, r := range q.store.resources { + if q.matches(r) { + results = append(results, *r) + } + } + + // TODO: Implement sorting + + // Apply pagination + if q.offset > 0 { + if q.offset >= len(results) { + return []Resource{} + } + results = results[q.offset:] + } + + if q.limit > 0 && q.limit < len(results) { + results = results[:q.limit] + } + + return results +} + +// Count returns the number of matching resources without fetching them. +func (q *ResourceQuery) Count() int { + q.store.mu.RLock() + defer q.store.mu.RUnlock() + + count := 0 + for _, r := range q.store.resources { + if q.matches(r) { + count++ + } + } + return count +} + +func (q *ResourceQuery) matches(r *Resource) bool { + // Type filter + if len(q.types) > 0 { + found := false + for _, t := range q.types { + if r.Type == t { + found = true + break + } + } + if !found { + return false + } + } + + // Platform filter + if len(q.platforms) > 0 { + found := false + for _, p := range q.platforms { + if r.PlatformType == p { + found = true + break + } + } + if !found { + return false + } + } + + // Status filter + if len(q.statuses) > 0 { + found := false + for _, s := range q.statuses { + if r.Status == s { + found = true + break + } + } + if !found { + return false + } + } + + // Parent filter + if q.parentID != nil && r.ParentID != *q.parentID { + return false + } + + // Cluster filter + if q.clusterID != nil && r.ClusterID != *q.clusterID { + return false + } + + // Alerts filter + if q.hasAlerts != nil && *q.hasAlerts && len(r.Alerts) == 0 { + return false + } + + return true +} + +// ============================================================================ +// Cross-Platform Analysis Methods +// These methods support AI queries like "what's using the most CPU" +// ============================================================================ + +// GetTopByCPU returns resources sorted by CPU usage (highest first). +// Optionally filter by resource types. Pass nil to include all types. +func (s *Store) GetTopByCPU(limit int, types []ResourceType) []Resource { + s.mu.RLock() + defer s.mu.RUnlock() + + var resources []Resource + for _, r := range s.resources { + if r.CPU == nil || r.CPU.Current == 0 { + continue + } + if len(types) > 0 { + matched := false + for _, t := range types { + if r.Type == t { + matched = true + break + } + } + if !matched { + continue + } + } + resources = append(resources, *r) + } + + // Sort by CPU usage descending + for i := 0; i < len(resources)-1; i++ { + for j := i + 1; j < len(resources); j++ { + if resources[j].CPUPercent() > resources[i].CPUPercent() { + resources[i], resources[j] = resources[j], resources[i] + } + } + } + + if limit > 0 && limit < len(resources) { + return resources[:limit] + } + return resources +} + +// GetTopByMemory returns resources sorted by memory usage (highest first). +// Optionally filter by resource types. Pass nil to include all types. +func (s *Store) GetTopByMemory(limit int, types []ResourceType) []Resource { + s.mu.RLock() + defer s.mu.RUnlock() + + var resources []Resource + for _, r := range s.resources { + if r.Memory == nil || r.Memory.Current == 0 { + continue + } + if len(types) > 0 { + matched := false + for _, t := range types { + if r.Type == t { + matched = true + break + } + } + if !matched { + continue + } + } + resources = append(resources, *r) + } + + // Sort by memory usage descending + for i := 0; i < len(resources)-1; i++ { + for j := i + 1; j < len(resources); j++ { + if resources[j].MemoryPercent() > resources[i].MemoryPercent() { + resources[i], resources[j] = resources[j], resources[i] + } + } + } + + if limit > 0 && limit < len(resources) { + return resources[:limit] + } + return resources +} + +// GetTopByDisk returns resources sorted by disk usage (highest first). +// Optionally filter by resource types. Pass nil to include all types. +func (s *Store) GetTopByDisk(limit int, types []ResourceType) []Resource { + s.mu.RLock() + defer s.mu.RUnlock() + + var resources []Resource + for _, r := range s.resources { + if r.Disk == nil || r.Disk.Current == 0 { + continue + } + if len(types) > 0 { + matched := false + for _, t := range types { + if r.Type == t { + matched = true + break + } + } + if !matched { + continue + } + } + resources = append(resources, *r) + } + + // Sort by disk usage descending + for i := 0; i < len(resources)-1; i++ { + for j := i + 1; j < len(resources); j++ { + if resources[j].DiskPercent() > resources[i].DiskPercent() { + resources[i], resources[j] = resources[j], resources[i] + } + } + } + + if limit > 0 && limit < len(resources) { + return resources[:limit] + } + return resources +} + +// GetRelated returns resources that are related to the given resource. +// This includes: parent, children, siblings (same parent), and co-located resources (same cluster). +func (s *Store) GetRelated(resourceID string) map[string][]Resource { + s.mu.RLock() + defer s.mu.RUnlock() + + result := make(map[string][]Resource) + + r, ok := s.resources[resourceID] + if !ok { + return result + } + + // Get parent + if r.ParentID != "" { + if parent, ok := s.resources[r.ParentID]; ok { + result["parent"] = []Resource{*parent} + } + } + + // Get children + var children []Resource + for _, other := range s.resources { + if other.ParentID == resourceID { + children = append(children, *other) + } + } + if len(children) > 0 { + result["children"] = children + } + + // Get siblings (same parent) + if r.ParentID != "" { + var siblings []Resource + for _, other := range s.resources { + if other.ID != resourceID && other.ParentID == r.ParentID { + siblings = append(siblings, *other) + } + } + if len(siblings) > 0 { + result["siblings"] = siblings + } + } + + // Get co-located resources (same cluster) + if r.ClusterID != "" { + var colocated []Resource + for _, other := range s.resources { + if other.ID != resourceID && other.ClusterID == r.ClusterID { + colocated = append(colocated, *other) + } + } + if len(colocated) > 0 { + result["cluster_members"] = colocated + } + } + + return result +} + +// GetResourceSummary returns a summary of resource utilization across the infrastructure. +// This is useful for AI to get a quick overview of the infrastructure state. +func (s *Store) GetResourceSummary() ResourceSummary { + s.mu.RLock() + defer s.mu.RUnlock() + + summary := ResourceSummary{ + ByType: make(map[ResourceType]TypeSummary), + ByPlatform: make(map[PlatformType]PlatformSummary), + } + + for _, r := range s.resources { + summary.TotalResources++ + + // Count by status + switch r.Status { + case StatusOnline, StatusRunning: + summary.Healthy++ + case StatusDegraded: + summary.Degraded++ + case StatusOffline, StatusStopped, StatusUnknown: + summary.Offline++ + } + + // Track alerts + if len(r.Alerts) > 0 { + summary.WithAlerts++ + } + + // Aggregate by type + ts := summary.ByType[r.Type] + ts.Count++ + if r.CPU != nil { + ts.TotalCPUPercent += r.CPUPercent() + } + if r.Memory != nil { + ts.TotalMemoryPercent += r.MemoryPercent() + } + summary.ByType[r.Type] = ts + + // Aggregate by platform + ps := summary.ByPlatform[r.PlatformType] + ps.Count++ + summary.ByPlatform[r.PlatformType] = ps + } + + // Calculate averages + for t, ts := range summary.ByType { + if ts.Count > 0 { + ts.AvgCPUPercent = ts.TotalCPUPercent / float64(ts.Count) + ts.AvgMemoryPercent = ts.TotalMemoryPercent / float64(ts.Count) + summary.ByType[t] = ts + } + } + + return summary +} + +// ResourceSummary provides an overview of resource utilization. +type ResourceSummary struct { + TotalResources int + Healthy int + Degraded int + Offline int + WithAlerts int + ByType map[ResourceType]TypeSummary + ByPlatform map[PlatformType]PlatformSummary +} + +// TypeSummary aggregates metrics for a resource type. +type TypeSummary struct { + Count int + TotalCPUPercent float64 + TotalMemoryPercent float64 + AvgCPUPercent float64 + AvgMemoryPercent float64 +} + +// PlatformSummary aggregates counts for a platform. +type PlatformSummary struct { + Count int +} + diff --git a/internal/resources/store_test.go b/internal/resources/store_test.go new file mode 100644 index 0000000..6b3be13 --- /dev/null +++ b/internal/resources/store_test.go @@ -0,0 +1,740 @@ +package resources + +import ( + "testing" + "time" +) + +func TestStoreUpsertAndGet(t *testing.T) { + store := NewStore() + + r := Resource{ + ID: "test-1", + Type: ResourceTypeNode, + Name: "node1", + PlatformType: PlatformProxmoxPVE, + SourceType: SourceAPI, + Status: StatusOnline, + LastSeen: time.Now(), + } + + id := store.Upsert(r) + if id != "test-1" { + t.Errorf("Expected ID test-1, got %s", id) + } + + retrieved, ok := store.Get("test-1") + if !ok { + t.Fatal("Failed to retrieve resource") + } + if retrieved.Name != "node1" { + t.Errorf("Expected name node1, got %s", retrieved.Name) + } +} + +func TestStoreGetAll(t *testing.T) { + store := NewStore() + + store.Upsert(Resource{ID: "1", Type: ResourceTypeNode, LastSeen: time.Now()}) + store.Upsert(Resource{ID: "2", Type: ResourceTypeVM, LastSeen: time.Now()}) + store.Upsert(Resource{ID: "3", Type: ResourceTypeContainer, LastSeen: time.Now()}) + + all := store.GetAll() + if len(all) != 3 { + t.Errorf("Expected 3 resources, got %d", len(all)) + } +} + +func TestStoreGetByType(t *testing.T) { + store := NewStore() + + store.Upsert(Resource{ID: "node1", Type: ResourceTypeNode, LastSeen: time.Now()}) + store.Upsert(Resource{ID: "node2", Type: ResourceTypeNode, LastSeen: time.Now()}) + store.Upsert(Resource{ID: "vm1", Type: ResourceTypeVM, LastSeen: time.Now()}) + + nodes := store.GetByType(ResourceTypeNode) + if len(nodes) != 2 { + t.Errorf("Expected 2 nodes, got %d", len(nodes)) + } + + vms := store.GetByType(ResourceTypeVM) + if len(vms) != 1 { + t.Errorf("Expected 1 VM, got %d", len(vms)) + } +} + +func TestStoreGetByPlatform(t *testing.T) { + store := NewStore() + + store.Upsert(Resource{ID: "1", PlatformType: PlatformProxmoxPVE, LastSeen: time.Now()}) + store.Upsert(Resource{ID: "2", PlatformType: PlatformProxmoxPVE, LastSeen: time.Now()}) + store.Upsert(Resource{ID: "3", PlatformType: PlatformDocker, LastSeen: time.Now()}) + + pve := store.GetByPlatform(PlatformProxmoxPVE) + if len(pve) != 2 { + t.Errorf("Expected 2 PVE resources, got %d", len(pve)) + } + + docker := store.GetByPlatform(PlatformDocker) + if len(docker) != 1 { + t.Errorf("Expected 1 Docker resource, got %d", len(docker)) + } +} + +func TestStoreGetInfrastructureAndWorkloads(t *testing.T) { + store := NewStore() + + store.Upsert(Resource{ID: "node1", Type: ResourceTypeNode, LastSeen: time.Now()}) + store.Upsert(Resource{ID: "host1", Type: ResourceTypeHost, LastSeen: time.Now()}) + store.Upsert(Resource{ID: "vm1", Type: ResourceTypeVM, LastSeen: time.Now()}) + store.Upsert(Resource{ID: "ct1", Type: ResourceTypeContainer, LastSeen: time.Now()}) + store.Upsert(Resource{ID: "dc1", Type: ResourceTypeDockerContainer, LastSeen: time.Now()}) + + infra := store.GetInfrastructure() + if len(infra) != 2 { + t.Errorf("Expected 2 infrastructure resources, got %d", len(infra)) + } + + workloads := store.GetWorkloads() + if len(workloads) != 3 { + t.Errorf("Expected 3 workload resources, got %d", len(workloads)) + } +} + +func TestStoreGetChildren(t *testing.T) { + store := NewStore() + + store.Upsert(Resource{ID: "node1", Type: ResourceTypeNode, LastSeen: time.Now()}) + store.Upsert(Resource{ID: "vm1", Type: ResourceTypeVM, ParentID: "node1", LastSeen: time.Now()}) + store.Upsert(Resource{ID: "vm2", Type: ResourceTypeVM, ParentID: "node1", LastSeen: time.Now()}) + store.Upsert(Resource{ID: "vm3", Type: ResourceTypeVM, ParentID: "node2", LastSeen: time.Now()}) + + children := store.GetChildren("node1") + if len(children) != 2 { + t.Errorf("Expected 2 children of node1, got %d", len(children)) + } +} + +func TestStoreRemove(t *testing.T) { + store := NewStore() + + store.Upsert(Resource{ID: "test-1", LastSeen: time.Now()}) + store.Upsert(Resource{ID: "test-2", LastSeen: time.Now()}) + + if len(store.GetAll()) != 2 { + t.Fatal("Expected 2 resources before remove") + } + + store.Remove("test-1") + + if len(store.GetAll()) != 1 { + t.Error("Expected 1 resource after remove") + } + + _, ok := store.Get("test-1") + if ok { + t.Error("Removed resource should not be retrievable") + } +} + +func TestDeduplicationByHostname(t *testing.T) { + store := NewStore() + + now := time.Now() + + // Add a Proxmox node (API source) + nodeResource := Resource{ + ID: "pve1/node/server1", + Type: ResourceTypeNode, + Name: "server1", + PlatformType: PlatformProxmoxPVE, + SourceType: SourceAPI, + Status: StatusOnline, + CPU: &MetricValue{Current: 50.0}, + LastSeen: now, + Identity: &ResourceIdentity{ + Hostname: "server1", + }, + } + store.Upsert(nodeResource) + + // Add a host agent for the same server (agent source - should be preferred) + hostResource := Resource{ + ID: "host-agent/server1", + Type: ResourceTypeHost, + Name: "server1", + PlatformType: PlatformHostAgent, + SourceType: SourceAgent, + Status: StatusOnline, + CPU: &MetricValue{Current: 55.0}, // Slightly different value + LastSeen: now, + Identity: &ResourceIdentity{ + Hostname: "server1", + }, + } + store.Upsert(hostResource) + + // We should only have 1 resource (the agent one, as it's preferred) + all := store.GetAll() + if len(all) != 1 { + t.Errorf("Expected 1 resource after dedup, got %d", len(all)) + } + + // The agent resource should be the one stored + r, ok := store.Get("host-agent/server1") + if !ok { + t.Fatal("Failed to get agent resource") + } + if r.SourceType != SourceAgent { + t.Errorf("Expected agent source, got %s", r.SourceType) + } + if r.CPU.Current != 55.0 { + t.Errorf("Expected CPU 55.0 from agent, got %f", r.CPU.Current) + } + + // The node resource should redirect to the agent resource + if !store.IsSuppressed("pve1/node/server1") { + t.Error("Node resource should be suppressed") + } + preferred := store.GetPreferredID("pve1/node/server1") + if preferred != "host-agent/server1" { + t.Errorf("Expected preferred ID to be host-agent/server1, got %s", preferred) + } + + // Accessing by suppressed ID should return the preferred resource + r, ok = store.Get("pve1/node/server1") + if !ok { + t.Fatal("Should be able to get by suppressed ID") + } + if r.ID != "host-agent/server1" { + t.Errorf("Expected to get agent resource when accessing by node ID, got %s", r.ID) + } +} + +func TestDeduplicationByMachineID(t *testing.T) { + store := NewStore() + + now := time.Now() + machineID := "abc-123-def-456" + + // Add a Docker host + dockerHost := Resource{ + ID: "docker-host-1", + Type: ResourceTypeDockerHost, + Name: "server-different-name", + PlatformType: PlatformDocker, + SourceType: SourceAgent, + Status: StatusOnline, + LastSeen: now, + Identity: &ResourceIdentity{ + Hostname: "server-different-name", + MachineID: machineID, + }, + } + store.Upsert(dockerHost) + + // Add a host agent with the same machine ID but different hostname + hostAgent := Resource{ + ID: "host-agent-1", + Type: ResourceTypeHost, + Name: "server-production", + PlatformType: PlatformHostAgent, + SourceType: SourceAgent, + Status: StatusOnline, + LastSeen: now.Add(time.Second), // Slightly newer + Identity: &ResourceIdentity{ + Hostname: "server-production", + MachineID: machineID, + }, + } + store.Upsert(hostAgent) + + // Should only have 1 resource (the newer one, since both are agent sources) + all := store.GetAll() + if len(all) != 1 { + t.Errorf("Expected 1 resource after dedup by machineID, got %d", len(all)) + } +} + +func TestDeduplicationByIP(t *testing.T) { + store := NewStore() + + now := time.Now() + sharedIP := "192.168.1.100" + + // Add a Proxmox node + node := Resource{ + ID: "node-1", + Type: ResourceTypeNode, + Name: "pve-node", + PlatformType: PlatformProxmoxPVE, + SourceType: SourceAPI, + Status: StatusOnline, + LastSeen: now, + Identity: &ResourceIdentity{ + Hostname: "pve-node", + IPs: []string{sharedIP}, + }, + } + store.Upsert(node) + + // Add a host agent with the same IP + host := Resource{ + ID: "host-1", + Type: ResourceTypeHost, + Name: "different-hostname", + PlatformType: PlatformHostAgent, + SourceType: SourceAgent, // Agent preferred + Status: StatusOnline, + LastSeen: now, + Identity: &ResourceIdentity{ + Hostname: "different-hostname", + IPs: []string{sharedIP}, + }, + } + store.Upsert(host) + + // Should only have 1 resource + all := store.GetAll() + if len(all) != 1 { + t.Errorf("Expected 1 resource after dedup by IP, got %d", len(all)) + } + + // Agent should be preferred + r, _ := store.Get("host-1") + if r == nil || r.SourceType != SourceAgent { + t.Error("Agent resource should be preferred") + } +} + +func TestNoDeduplicationForWorkloads(t *testing.T) { + store := NewStore() + + now := time.Now() + + // VMs with the same hostname should NOT be deduplicated + // (they're workloads, not infrastructure) + vm1 := Resource{ + ID: "pve1/vm/100", + Type: ResourceTypeVM, + Name: "webserver", + PlatformType: PlatformProxmoxPVE, + SourceType: SourceAPI, + LastSeen: now, + Identity: &ResourceIdentity{ + Hostname: "webserver", // Same hostname + }, + } + store.Upsert(vm1) + + vm2 := Resource{ + ID: "pve2/vm/100", + Type: ResourceTypeVM, + Name: "webserver", + PlatformType: PlatformProxmoxPVE, + SourceType: SourceAPI, + LastSeen: now, + Identity: &ResourceIdentity{ + Hostname: "webserver", // Same hostname + }, + } + store.Upsert(vm2) + + // Both VMs should exist (workloads are not deduplicated) + all := store.GetAll() + if len(all) != 2 { + t.Errorf("Expected 2 VMs (no dedup for workloads), got %d", len(all)) + } +} + +func TestNoDeduplicationForLocalhost(t *testing.T) { + store := NewStore() + + now := time.Now() + + host1 := Resource{ + ID: "host-1", + Type: ResourceTypeHost, + Name: "server1", + PlatformType: PlatformHostAgent, + SourceType: SourceAgent, + LastSeen: now, + Identity: &ResourceIdentity{ + Hostname: "server1", + IPs: []string{"127.0.0.1", "192.168.1.1"}, + }, + } + store.Upsert(host1) + + host2 := Resource{ + ID: "host-2", + Type: ResourceTypeHost, + Name: "server2", + PlatformType: PlatformHostAgent, + SourceType: SourceAgent, + LastSeen: now, + Identity: &ResourceIdentity{ + Hostname: "server2", + IPs: []string{"127.0.0.1", "192.168.1.2"}, // Both have localhost + }, + } + store.Upsert(host2) + + // Both should exist (127.0.0.1 shouldn't trigger dedup) + all := store.GetAll() + if len(all) != 2 { + t.Errorf("Expected 2 hosts (localhost shouldn't dedup), got %d", len(all)) + } +} + +func TestStoreStats(t *testing.T) { + store := NewStore() + + store.Upsert(Resource{ID: "1", Type: ResourceTypeNode, PlatformType: PlatformProxmoxPVE, LastSeen: time.Now()}) + store.Upsert(Resource{ID: "2", Type: ResourceTypeNode, PlatformType: PlatformProxmoxPVE, LastSeen: time.Now()}) + store.Upsert(Resource{ID: "3", Type: ResourceTypeVM, PlatformType: PlatformProxmoxPVE, LastSeen: time.Now()}) + store.Upsert(Resource{ID: "4", Type: ResourceTypeDockerHost, PlatformType: PlatformDocker, LastSeen: time.Now()}) + + stats := store.GetStats() + + if stats.TotalResources != 4 { + t.Errorf("Expected 4 total resources, got %d", stats.TotalResources) + } + if stats.ByType[ResourceTypeNode] != 2 { + t.Errorf("Expected 2 nodes, got %d", stats.ByType[ResourceTypeNode]) + } + if stats.ByPlatform[PlatformProxmoxPVE] != 3 { + t.Errorf("Expected 3 PVE resources, got %d", stats.ByPlatform[PlatformProxmoxPVE]) + } +} + +func TestStoreQuery(t *testing.T) { + store := NewStore() + + store.Upsert(Resource{ + ID: "1", + Type: ResourceTypeNode, + PlatformType: PlatformProxmoxPVE, + Status: StatusOnline, + LastSeen: time.Now(), + }) + store.Upsert(Resource{ + ID: "2", + Type: ResourceTypeVM, + PlatformType: PlatformProxmoxPVE, + Status: StatusRunning, + ParentID: "1", + LastSeen: time.Now(), + }) + store.Upsert(Resource{ + ID: "3", + Type: ResourceTypeVM, + PlatformType: PlatformProxmoxPVE, + Status: StatusStopped, + ParentID: "1", + LastSeen: time.Now(), + }) + store.Upsert(Resource{ + ID: "4", + Type: ResourceTypeDockerContainer, + PlatformType: PlatformDocker, + Status: StatusRunning, + LastSeen: time.Now(), + }) + + // Query by type + vms := store.Query().OfType(ResourceTypeVM).Execute() + if len(vms) != 2 { + t.Errorf("Expected 2 VMs, got %d", len(vms)) + } + + // Query by status + running := store.Query().WithStatus(StatusRunning).Execute() + if len(running) != 2 { + t.Errorf("Expected 2 running resources, got %d", len(running)) + } + + // Query by platform + pve := store.Query().FromPlatform(PlatformProxmoxPVE).Execute() + if len(pve) != 3 { + t.Errorf("Expected 3 PVE resources, got %d", len(pve)) + } + + // Query by parent + node1Children := store.Query().WithParent("1").Execute() + if len(node1Children) != 2 { + t.Errorf("Expected 2 children of node 1, got %d", len(node1Children)) + } + + // Combined query + runningVMs := store.Query(). + OfType(ResourceTypeVM). + WithStatus(StatusRunning). + Execute() + if len(runningVMs) != 1 { + t.Errorf("Expected 1 running VM, got %d", len(runningVMs)) + } + + // Count + count := store.Query().OfType(ResourceTypeVM).Count() + if count != 2 { + t.Errorf("Expected count 2, got %d", count) + } + + // Limit + limited := store.Query().Limit(2).Execute() + if len(limited) > 2 { + t.Errorf("Expected at most 2 results, got %d", len(limited)) + } +} + +func TestMarkStale(t *testing.T) { + store := NewStore() + + old := time.Now().Add(-2 * time.Hour) + recent := time.Now() + + store.Upsert(Resource{ + ID: "old-1", + Status: StatusOnline, + LastSeen: old, + }) + store.Upsert(Resource{ + ID: "recent-1", + Status: StatusOnline, + LastSeen: recent, + }) + + stale := store.MarkStale(time.Hour) + + if len(stale) != 1 { + t.Errorf("Expected 1 stale resource, got %d", len(stale)) + } + + r, _ := store.Get("old-1") + if r.Status != StatusDegraded { + t.Errorf("Expected stale resource to be degraded, got %s", r.Status) + } + + r, _ = store.Get("recent-1") + if r.Status != StatusOnline { + t.Errorf("Recent resource should still be online, got %s", r.Status) + } +} + +func TestPruneStale(t *testing.T) { + store := NewStore() + + veryOld := time.Now().Add(-48 * time.Hour) + old := time.Now().Add(-2 * time.Hour) + recent := time.Now() + + store.Upsert(Resource{ID: "very-old", LastSeen: veryOld}) + store.Upsert(Resource{ID: "old", LastSeen: old}) + store.Upsert(Resource{ID: "recent", LastSeen: recent}) + + removed := store.PruneStale(time.Hour, 24*time.Hour) + + if len(removed) != 1 { + t.Errorf("Expected 1 removed resource, got %d", len(removed)) + } + + if len(store.GetAll()) != 2 { + t.Errorf("Expected 2 remaining resources, got %d", len(store.GetAll())) + } +} + +func TestAPIToAgentPreference(t *testing.T) { + store := NewStore() + + now := time.Now() + + // First, add an API resource + apiResource := Resource{ + ID: "api-node", + Type: ResourceTypeNode, + Name: "server", + PlatformType: PlatformProxmoxPVE, + SourceType: SourceAPI, + CPU: &MetricValue{Current: 50.0}, + LastSeen: now, + Identity: &ResourceIdentity{ + Hostname: "server", + }, + } + store.Upsert(apiResource) + + // Then, add an agent resource for the same machine + agentResource := Resource{ + ID: "agent-host", + Type: ResourceTypeHost, + Name: "server", + PlatformType: PlatformHostAgent, + SourceType: SourceAgent, + CPU: &MetricValue{Current: 55.0}, + LastSeen: now, + Identity: &ResourceIdentity{ + Hostname: "server", + }, + } + store.Upsert(agentResource) + + // Only agent resource should exist + all := store.GetAll() + if len(all) != 1 { + t.Fatalf("Expected 1 resource, got %d", len(all)) + } + + if all[0].SourceType != SourceAgent { + t.Errorf("Expected agent source type, got %s", all[0].SourceType) + } +} + +func TestGetTopByCPU(t *testing.T) { + store := NewStore() + now := time.Now() + + store.Upsert(Resource{ + ID: "vm1", + Type: ResourceTypeVM, + Name: "low-cpu-vm", + CPU: &MetricValue{Current: 20.0}, + LastSeen: now, + }) + store.Upsert(Resource{ + ID: "vm2", + Type: ResourceTypeVM, + Name: "high-cpu-vm", + CPU: &MetricValue{Current: 85.0}, + LastSeen: now, + }) + store.Upsert(Resource{ + ID: "node1", + Type: ResourceTypeNode, + Name: "busy-node", + CPU: &MetricValue{Current: 75.0}, + LastSeen: now, + }) + + // Get top 2 by CPU + top := store.GetTopByCPU(2, nil) + if len(top) != 2 { + t.Fatalf("Expected 2 resources, got %d", len(top)) + } + if top[0].Name != "high-cpu-vm" { + t.Errorf("Expected high-cpu-vm first, got %s", top[0].Name) + } + if top[1].Name != "busy-node" { + t.Errorf("Expected busy-node second, got %s", top[1].Name) + } + + // Filter by type + topVMs := store.GetTopByCPU(10, []ResourceType{ResourceTypeVM}) + if len(topVMs) != 2 { + t.Errorf("Expected 2 VMs, got %d", len(topVMs)) + } +} + +func TestGetRelated(t *testing.T) { + store := NewStore() + now := time.Now() + + store.Upsert(Resource{ + ID: "node1", + Type: ResourceTypeNode, + Name: "parent-node", + ClusterID: "cluster1", + LastSeen: now, + }) + store.Upsert(Resource{ + ID: "vm1", + Type: ResourceTypeVM, + Name: "child-vm-1", + ParentID: "node1", + ClusterID: "cluster1", + LastSeen: now, + }) + store.Upsert(Resource{ + ID: "vm2", + Type: ResourceTypeVM, + Name: "child-vm-2", + ParentID: "node1", + ClusterID: "cluster1", + LastSeen: now, + }) + store.Upsert(Resource{ + ID: "node2", + Type: ResourceTypeNode, + Name: "cluster-peer", + ClusterID: "cluster1", + LastSeen: now, + }) + + // Get related resources for vm1 + related := store.GetRelated("vm1") + + // Should have parent + if parent, ok := related["parent"]; !ok || len(parent) != 1 { + t.Error("Expected 1 parent") + } + + // Should have sibling (vm2) + if siblings, ok := related["siblings"]; !ok || len(siblings) != 1 { + t.Errorf("Expected 1 sibling, got %d", len(related["siblings"])) + } + + // Should have cluster members + if cluster, ok := related["cluster_members"]; !ok || len(cluster) != 3 { + t.Errorf("Expected 3 cluster members, got %d", len(related["cluster_members"])) + } +} + +func TestGetResourceSummary(t *testing.T) { + store := NewStore() + now := time.Now() + + store.Upsert(Resource{ + ID: "node1", + Type: ResourceTypeNode, + PlatformType: PlatformProxmoxPVE, + Status: StatusOnline, + CPU: &MetricValue{Current: 50}, + Memory: &MetricValue{Current: 50}, // 50% usage + LastSeen: now, + }) + store.Upsert(Resource{ + ID: "vm1", + Type: ResourceTypeVM, + PlatformType: PlatformProxmoxPVE, + Status: StatusRunning, + CPU: &MetricValue{Current: 70}, + Memory: &MetricValue{Current: 50}, // 50% usage + LastSeen: now, + }) + store.Upsert(Resource{ + ID: "vm2", + Type: ResourceTypeVM, + PlatformType: PlatformProxmoxPVE, + Status: StatusStopped, + LastSeen: now, + }) + + summary := store.GetResourceSummary() + + if summary.TotalResources != 3 { + t.Errorf("Expected 3 total resources, got %d", summary.TotalResources) + } + if summary.Healthy != 2 { + t.Errorf("Expected 2 healthy, got %d", summary.Healthy) + } + if summary.Offline != 1 { + t.Errorf("Expected 1 offline, got %d", summary.Offline) + } + + // Check per-type stats + vmStats := summary.ByType[ResourceTypeVM] + if vmStats.Count != 2 { + t.Errorf("Expected 2 VMs, got %d", vmStats.Count) + } +} +