Files
2026-09-12 08:57:29 +08:00

83 lines
4.3 KiB
Go

package observability
import (
"context"
"time"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/prometheus/client_golang/prometheus"
)
// RegisterBusinessMetrics exposes bounded-cardinality workflow state for
// operational alerting. It is registered by the single worker process so
// horizontally scaled API instances do not publish duplicate database gauges.
func RegisterBusinessMetrics(pool *pgxpool.Pool) {
registerStatusCounts(pool, "onboarding_activation_jobs", "Physical device activation jobs by durable status.", "device_activation_job", []string{"NOT_STARTED", "RESERVED", "CONTACT_POSSIBLE", "REGISTERED_OBSERVED", "READY", "RECONCILE_REQUIRED", "BLOCKED"})
registerStatusCounts(pool, "onboarding_claims", "Physical confirmation claims by status.", "device_binding_claim", []string{"PENDING", "CONFIRMED", "CANCELLED", "EXPIRED"})
registerStatusCounts(pool, "conversations", "Current conversations by lifecycle status.", "conversation", []string{"PREPARING", "ROOM_ISSUED", "STARTING_AGENT", "ACTIVE", "STOPPING", "COMPLETED", "FAILED", "EXPIRED"})
registerStatusCounts(pool, "privacy_requests", "Current privacy requests by status.", "privacy_request", []string{"SUBMITTED", "VERIFYING", "EXECUTING", "COMPLETED", "PARTIAL_FAILED"})
registerStatusCounts(pool, "firmware_rollouts", "Current firmware rollouts by status.", "firmware_rollout", []string{"DRAFT", "RUNNING", "PAUSED", "COMPLETED", "ROLLED_BACK"})
registerStatusCounts(pool, "firmware_assignments", "Current firmware assignments by status.", "firmware_assignment", []string{"ASSIGNED", "DOWNLOADING", "INSTALLING", "CONFIRMED", "FAILED", "ROLLED_BACK"})
prometheus.MustRegister(prometheus.NewGaugeFunc(prometheus.GaugeOpts{
Namespace: "voice_platform",
Subsystem: "firmware",
Name: "stalled_rollouts",
Help: "Running firmware rollouts not updated for at least 15 minutes.",
}, func() float64 {
return queryCount(pool, "SELECT count(*)::float8 FROM firmware_rollout WHERE status='RUNNING' AND updated_at < now()-interval '15 minutes'")
}))
prometheus.MustRegister(prometheus.NewGaugeFunc(prometheus.GaugeOpts{
Namespace: "voice_platform", Subsystem: "provider", Name: "notification_backlog",
Help: "Provider callback events pending or failed for more than one minute.",
}, func() float64 {
return queryCount(pool, "SELECT count(*)::float8 FROM provider_event WHERE processing_status IN ('pending','failed') AND received_at<now()-interval '1 minute'")
}))
for _, window := range []struct {
label string
days int
}{{"7d", 7}, {"30d", 30}} {
window := window
prometheus.MustRegister(prometheus.NewGaugeFunc(prometheus.GaugeOpts{
Namespace: "voice_platform", Subsystem: "security", Name: "certificates_expiring",
Help: "Active managed certificate references expiring in the selected window.",
ConstLabels: prometheus.Labels{"window": window.label},
}, func() float64 {
return queryCount(pool, "SELECT count(*)::float8 FROM ops_resource_reference WHERE resource_type='CERTIFICATE' AND status='ACTIVE' AND expires_at<=now()+$1*interval '1 day'", window.days)
}))
}
for _, category := range []string{"BUSINESS", "TECHNICAL", "SECURITY", "COST"} {
category := category
prometheus.MustRegister(prometheus.NewGaugeFunc(prometheus.GaugeOpts{
Namespace: "voice_platform", Name: "alert_incidents",
Help: "Current non-resolved alert incident count by category.",
ConstLabels: prometheus.Labels{"category": category},
}, func() float64 {
return queryCount(pool, "SELECT count(*)::float8 FROM alert_incident WHERE category=$1 AND status<>'RESOLVED'", category)
}))
}
}
func registerStatusCounts(pool *pgxpool.Pool, name, help, table string, statuses []string) {
for _, status := range statuses {
status := status
prometheus.MustRegister(prometheus.NewGaugeFunc(prometheus.GaugeOpts{
Namespace: "voice_platform",
Name: name,
Help: help,
ConstLabels: prometheus.Labels{"status": status},
}, func() float64 {
return queryCount(pool, "SELECT count(*)::float8 FROM "+table+" WHERE status=$1", status)
}))
}
}
func queryCount(pool *pgxpool.Pool, query string, args ...any) float64 {
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
var count float64
if err := pool.QueryRow(ctx, query, args...).Scan(&count); err != nil {
return 0
}
return count
}