yovision/Sense/internal/metrics/metrics.go

// Package metrics exposes a deliberately small, low-cardinality Prometheus
// surface without adding a runtime dependency. Tenant, Site, device and Path
// identifiers never become labels.
package metrics

import (
	"crypto/rand"
	"encoding/hex"
	"errors"
	"fmt"
	"net/http"
	"strconv"
	"strings"
	"sync/atomic"
	"time"
)

func GenerateInstanceID() (string, error) {
	value := make([]byte, 12)
	if _, err := rand.Read(value); err != nil {
		return "", errors.New("generate Sense instance identifier")
	}
	return "ins_" + hex.EncodeToString(value), nil
}

type Registry struct {
	instanceID string
	version    string

	reconcileRunOK       atomic.Uint64
	reconcileRunError    atomic.Uint64
	reconcileItemOK      atomic.Uint64
	reconcileItemError   atomic.Uint64
	reconcileItemLost    atomic.Uint64
	reconcileDurationNS  atomic.Uint64
	reconcileTotal       atomic.Int64
	reconcileUnconverged atomic.Int64

	orphanScanOK       atomic.Uint64
	orphanScanError    atomic.Uint64
	orphanObserved     atomic.Int64
	orphanOwnedStale   atomic.Int64
	orphanUnowned      atomic.Int64
	orphanBlockedRatio atomic.Uint64
	orphanBlockedAge   atomic.Uint64
	orphanBlockedScope atomic.Uint64
	orphanDeleted      atomic.Uint64
	orphanDeleteFailed atomic.Uint64
}

func New(instanceID, version string) *Registry {
	return &Registry{instanceID: instanceID, version: version}
}

func (r *Registry) ObserveReconcileRun(err error, duration time.Duration) {
	if err == nil {
		r.reconcileRunOK.Add(1)
	} else {
		r.reconcileRunError.Add(1)
	}
	if duration < 0 {
		duration = 0
	}
	r.reconcileDurationNS.Store(uint64(duration))
}

func (r *Registry) ObserveReconcileItem(result string) {
	switch result {
	case "success":
		r.reconcileItemOK.Add(1)
	case "lease_lost":
		r.reconcileItemLost.Add(1)
	default:
		r.reconcileItemError.Add(1)
	}
}

func (r *Registry) SetConvergence(total, unconverged int) {
	r.reconcileTotal.Store(int64(total))
	r.reconcileUnconverged.Store(int64(unconverged))
}

func (r *Registry) ObserveOrphanScan(observed, ownedStale, unowned int, err error) {
	if err == nil {
		r.orphanScanOK.Add(1)
		r.orphanObserved.Store(int64(observed))
		r.orphanOwnedStale.Store(int64(ownedStale))
		r.orphanUnowned.Store(int64(unowned))
	} else {
		r.orphanScanError.Add(1)
	}
}

func (r *Registry) ObserveOrphanCleanupBlocked(reason string) {
	switch reason {
	case "snapshot_expired":
		r.orphanBlockedAge.Add(1)
	case "ratio_exceeded":
		r.orphanBlockedRatio.Add(1)
	default:
		r.orphanBlockedScope.Add(1)
	}
}

func (r *Registry) ObserveOrphanCleanup(deleted, failed int) {
	if deleted > 0 {
		r.orphanDeleted.Add(uint64(deleted))
	}
	if failed > 0 {
		r.orphanDeleteFailed.Add(uint64(failed))
	}
}

func (r *Registry) Handler() http.Handler {
	return http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
		writer.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
		writer.Header().Set("Cache-Control", "no-store")
		_, _ = writer.Write([]byte(r.render()))
	})
}

func (r *Registry) render() string {
	var output strings.Builder
	writeMetric(&output, "sense_build_info", "Sense process build and instance information.",
		fmt.Sprintf("{instance_id=%s,version=%s}", quoteLabel(r.instanceID), quoteLabel(r.version)), 1)
	writeMetric(&output, "sense_reconcile_runs_total", "Reconciliation runs by result.",
		`{result="success"}`, r.reconcileRunOK.Load())
	writeSample(&output, "sense_reconcile_runs_total", `{result="error"}`, r.reconcileRunError.Load())
	writeMetric(&output, "sense_reconcile_items_total", "Reconciliation items by fixed result.",
		`{result="success"}`, r.reconcileItemOK.Load())
	writeSample(&output, "sense_reconcile_items_total", `{result="error"}`, r.reconcileItemError.Load())
	writeSample(&output, "sense_reconcile_items_total", `{result="lease_lost"}`, r.reconcileItemLost.Load())
	writeMetric(&output, "sense_reconcile_last_duration_seconds", "Duration of the last reconciliation run.",
		"", float64(r.reconcileDurationNS.Load())/float64(time.Second))
	writeMetric(&output, "sense_reconcile_devices", "Enabled video devices by convergence state.",
		`{state="total"}`, r.reconcileTotal.Load())
	writeSample(&output, "sense_reconcile_devices", `{state="unconverged"}`, r.reconcileUnconverged.Load())
	writeMetric(&output, "sense_orphan_scan_runs_total", "MediaMTX orphan scans by result.",
		`{result="success"}`, r.orphanScanOK.Load())
	writeSample(&output, "sense_orphan_scan_runs_total", `{result="error"}`, r.orphanScanError.Load())
	writeMetric(&output, "sense_orphan_paths", "Path counts from the last successful orphan scan.",
		`{classification="observed"}`, r.orphanObserved.Load())
	writeSample(&output, "sense_orphan_paths", `{classification="owned_stale"}`, r.orphanOwnedStale.Load())
	writeSample(&output, "sense_orphan_paths", `{classification="unowned"}`, r.orphanUnowned.Load())
	writeMetric(&output, "sense_orphan_cleanup_blocked_total", "Orphan cleanups blocked by a fixed safety reason.",
		`{reason="ratio_exceeded"}`, r.orphanBlockedRatio.Load())
	writeSample(&output, "sense_orphan_cleanup_blocked_total", `{reason="snapshot_expired"}`, r.orphanBlockedAge.Load())
	writeSample(&output, "sense_orphan_cleanup_blocked_total", `{reason="scope_changed"}`, r.orphanBlockedScope.Load())
	writeMetric(&output, "sense_orphan_cleanup_items_total", "Owned stale path cleanup results.",
		`{result="deleted"}`, r.orphanDeleted.Load())
	writeSample(&output, "sense_orphan_cleanup_items_total", `{result="failed"}`, r.orphanDeleteFailed.Load())
	return output.String()
}

func writeMetric(builder *strings.Builder, name, help, labels string, value any) {
	metricType := "gauge"
	if strings.HasSuffix(name, "_total") {
		metricType = "counter"
	}
	builder.WriteString("# HELP ")
	builder.WriteString(name)
	builder.WriteByte(' ')
	builder.WriteString(help)
	builder.WriteByte('\n')
	builder.WriteString("# TYPE ")
	builder.WriteString(name)
	builder.WriteByte(' ')
	builder.WriteString(metricType)
	builder.WriteByte('\n')
	writeSample(builder, name, labels, value)
}

func writeSample(builder *strings.Builder, name, labels string, value any) {
	builder.WriteString(name)
	builder.WriteString(labels)
	builder.WriteByte(' ')
	switch typed := value.(type) {
	case float64:
		builder.WriteString(strconv.FormatFloat(typed, 'f', 6, 64))
	default:
		builder.WriteString(fmt.Sprint(typed))
	}
	builder.WriteByte('\n')
}

func quoteLabel(value string) string {
	replacer := strings.NewReplacer(`\`, `\\`, "\n", `\n`, `"`, `\"`)
	return `"` + replacer.Replace(value) + `"`
}