// Package metrics — collector.go // // On-demand metrics collector for the admin /api/v1/admin/metrics endpoint. // Gathers runtime, database, cluster, and extension stats into a single JSON snapshot. package metrics import ( "context" "database/sql" "encoding/json" "os" "runtime" "time" "switchboard-core/database" "switchboard-core/store" ) // Snapshot is the top-level JSON response from GET /api/v1/admin/metrics. type Snapshot struct { Runtime RuntimeMetrics `json:"runtime"` DB DBMetrics `json:"db"` Cluster *ClusterMetrics `json:"cluster,omitempty"` Extensions ExtensionMetrics `json:"extensions"` } type RuntimeMetrics struct { Goroutines int `json:"goroutines"` HeapAlloc uint64 `json:"heap_alloc"` HeapSys uint64 `json:"heap_sys"` StackInUse uint64 `json:"stack_in_use"` GCCycles uint32 `json:"gc_cycles"` GCPauseNs uint64 `json:"gc_pause_ns"` GCCPUPercent float64 `json:"gc_cpu_pct"` UptimeSec float64 `json:"uptime_sec"` WSClients int `json:"ws_clients"` ExtensionsLoaded int `json:"extensions_loaded"` OpenFDs int `json:"open_fds"` } type DBMetrics struct { LatencyMs float64 `json:"latency_ms"` PoolActive int `json:"pool_active"` PoolIdle int `json:"pool_idle"` PoolMax int `json:"pool_max"` WaitCount int64 `json:"wait_count"` WaitDuration float64 `json:"wait_duration_ms"` // PG-only fields (zero/omitted on SQLite) DeadTuples *int64 `json:"dead_tuples,omitempty"` ActiveBackends *int `json:"active_backends,omitempty"` } type ClusterMetrics struct { Size int `json:"size"` Nodes []ClusterNode `json:"nodes"` } type ClusterNode struct { NodeID string `json:"node_id"` Endpoint string `json:"endpoint"` UptimeSec float64 `json:"uptime_sec"` HeartbeatAge int64 `json:"heartbeat_age_ms"` Stats json.RawMessage `json:"stats"` } type ExtensionMetrics struct { StarlarkExecTotal uint64 `json:"starlark_exec_total"` StarlarkErrorsTotal uint64 `json:"starlark_errors_total"` StarlarkAvgDuration float64 `json:"starlark_avg_duration_ms"` TriggerFiresTotal int64 `json:"trigger_fires_total"` EventBusPublished int64 `json:"event_bus_published"` EventBusDelivered int64 `json:"event_bus_delivered"` } // ConnCounter provides WebSocket connection count (satisfied by events.Hub). type ConnCounter interface { ConnCount() int } // BusCounter provides publish/deliver counts (satisfied by events.Bus). type BusCounter interface { PublishCount() int64 DeliverCount() int64 } // SandboxStatsFunc returns cumulative sandbox execution counters. type SandboxStatsFunc func() (execCount, errorCount uint64, avgDurationMs float64) // TriggerFireCountFunc returns cumulative trigger fire count. type TriggerFireCountFunc func() int64 // Collector gathers metrics on demand for the admin endpoint. type Collector struct { db *sql.DB hub ConnCounter bus BusCounter stores store.Stores sandboxStats SandboxStatsFunc triggerFireCount TriggerFireCountFunc startTime time.Time } // NewCollector creates a metrics collector with all required dependencies. func NewCollector(db *sql.DB, hub ConnCounter, bus BusCounter, stores store.Stores, sandboxFn SandboxStatsFunc, triggerFn TriggerFireCountFunc, startTime time.Time) *Collector { return &Collector{ db: db, hub: hub, bus: bus, stores: stores, sandboxStats: sandboxFn, triggerFireCount: triggerFn, startTime: startTime, } } // Collect gathers all metrics synchronously and returns a snapshot. func (c *Collector) Collect(ctx context.Context) *Snapshot { snap := &Snapshot{} snap.Runtime = c.collectRuntime() snap.DB = c.collectDB(ctx) snap.Cluster = c.collectCluster(ctx) snap.Extensions = c.collectExtensions() return snap } func (c *Collector) collectRuntime() RuntimeMetrics { var m runtime.MemStats runtime.ReadMemStats(&m) extCount := 0 if c.stores.Packages != nil { if pkgs, err := c.stores.Packages.List(context.Background()); err == nil { for _, p := range pkgs { if p.Status == "active" { extCount++ } } } } return RuntimeMetrics{ Goroutines: runtime.NumGoroutine(), HeapAlloc: m.HeapAlloc, HeapSys: m.HeapSys, StackInUse: m.StackInuse, GCCycles: m.NumGC, GCPauseNs: m.PauseNs[(m.NumGC+255)%256], GCCPUPercent: m.GCCPUFraction * 100, UptimeSec: time.Since(c.startTime).Seconds(), WSClients: c.hub.ConnCount(), ExtensionsLoaded: extCount, OpenFDs: countOpenFDs(), } } func (c *Collector) collectDB(ctx context.Context) DBMetrics { dm := DBMetrics{} if c.db == nil { return dm } // Pool stats stats := c.db.Stats() dm.PoolActive = stats.InUse dm.PoolIdle = stats.Idle dm.PoolMax = stats.MaxOpenConnections dm.WaitCount = stats.WaitCount dm.WaitDuration = float64(stats.WaitDuration.Milliseconds()) // Latency probe probeCtx, cancel := context.WithTimeout(ctx, 2*time.Second) defer cancel() start := time.Now() if err := c.db.PingContext(probeCtx); err == nil { dm.LatencyMs = float64(time.Since(start).Microseconds()) / 1000.0 } // PG-only stats if database.IsPostgres() { var deadTuples int64 if err := c.db.QueryRowContext(ctx, `SELECT COALESCE(SUM(n_dead_tup), 0) FROM pg_stat_user_tables`).Scan(&deadTuples); err == nil { dm.DeadTuples = &deadTuples } var activeBackends int if err := c.db.QueryRowContext(ctx, `SELECT count(*) FROM pg_stat_activity WHERE datname = current_database()`).Scan(&activeBackends); err == nil { dm.ActiveBackends = &activeBackends } } return dm } func (c *Collector) collectCluster(ctx context.Context) *ClusterMetrics { if c.stores.Cluster == nil { return nil } nodes, err := c.stores.Cluster.ListNodes(ctx) if err != nil { return nil } if len(nodes) == 0 { return nil } cm := &ClusterMetrics{ Size: len(nodes), Nodes: make([]ClusterNode, len(nodes)), } now := time.Now() for i, n := range nodes { // Extract uptime from stats JSONB var uptimeSec float64 var statsMap map[string]any if json.Unmarshal(n.Stats, &statsMap) == nil { if u, ok := statsMap["uptime_sec"].(float64); ok { uptimeSec = u } } cm.Nodes[i] = ClusterNode{ NodeID: n.NodeID, Endpoint: n.Endpoint, UptimeSec: uptimeSec, HeartbeatAge: now.Sub(n.Heartbeat).Milliseconds(), Stats: n.Stats, } } return cm } func (c *Collector) collectExtensions() ExtensionMetrics { em := ExtensionMetrics{} if c.sandboxStats != nil { em.StarlarkExecTotal, em.StarlarkErrorsTotal, em.StarlarkAvgDuration = c.sandboxStats() } if c.triggerFireCount != nil { em.TriggerFiresTotal = c.triggerFireCount() } if c.bus != nil { em.EventBusPublished = c.bus.PublishCount() em.EventBusDelivered = c.bus.DeliverCount() } return em } // countOpenFDs counts open file descriptors via /proc/self/fd (Linux only). func countOpenFDs() int { entries, err := os.ReadDir("/proc/self/fd") if err != nil { return -1 } return len(entries) }