core/server/metrics/collector.go

// 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 {
	NodeID     string            `json:"node_id"`
	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 {
	nodeID           string
	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(nodeID string, db *sql.DB, hub ConnCounter, bus BusCounter, stores store.Stores, sandboxFn SandboxStatsFunc, triggerFn TriggerFireCountFunc, startTime time.Time) *Collector {
	return &Collector{
		nodeID:           nodeID,
		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{NodeID: c.nodeID}
	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)
}