ITWorx-Pulse-Public/internal/observability/metrics.go

package observability

import (
	"fmt"
	"net/http"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"
)

const maxRequestSeries = 128

type requestKey struct {
	method string
	route  string
	status string
}

type requestValue struct {
	count uint64
	sum   time.Duration
}

// Registry contains only bounded process metrics. Route and status labels are
// normalized to fixed vocabularies before they reach the registry.
type Registry struct {
	mu        sync.Mutex
	startedAt time.Time
	requests  map[requestKey]requestValue
	gauges    map[string]float64
	counters  map[string]uint64
}

func NewRegistry(now time.Time) *Registry {
	if now.IsZero() {
		now = time.Now().UTC()
	}
	return &Registry{startedAt: now.UTC(), requests: make(map[requestKey]requestValue), gauges: make(map[string]float64), counters: make(map[string]uint64)}
}

func (r *Registry) ObserveRequest(method, path string, status int, duration time.Duration) {
	if r == nil {
		return
	}
	key := requestKey{method: normalizeMethod(method), route: RouteCode(path), status: statusClass(status)}
	if duration < 0 {
		duration = 0
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	if _, exists := r.requests[key]; !exists && len(r.requests) >= maxRequestSeries-1 {
		key = requestKey{method: "OTHER", route: "other", status: "other"}
	}
	value := r.requests[key]
	value.count++
	value.sum += duration
	r.requests[key] = value
}

func (r *Registry) SetGauge(name string, value float64) {
	if r == nil || !validMetricName(name) {
		return
	}
	r.mu.Lock()
	r.gauges[name] = value
	r.mu.Unlock()
}

func (r *Registry) IncCounter(name string) {
	if r == nil || !validMetricName(name) {
		return
	}
	r.mu.Lock()
	r.counters[name]++
	r.mu.Unlock()
}

func (r *Registry) Handler() http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) {
		if request.Method != http.MethodGet || request.URL.Path != "/api/v1/system/metrics" {
			http.NotFound(w, request)
			return
		}
		if r != nil {
			r.IncCounter("pulse_metrics_scrapes_total")
		}
		w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
		w.Header().Set("Cache-Control", "no-store")
		_, _ = w.Write([]byte(r.Exposition(time.Now().UTC())))
	})
}

func (r *Registry) Exposition(now time.Time) string {
	if r == nil {
		return ""
	}
	if now.IsZero() {
		now = time.Now().UTC()
	}
	r.mu.Lock()
	startedAt := r.startedAt
	requests := make(map[requestKey]requestValue, len(r.requests))
	for key, value := range r.requests {
		requests[key] = value
	}
	gauges := make(map[string]float64, len(r.gauges))
	for name, value := range r.gauges {
		gauges[name] = value
	}
	counters := make(map[string]uint64, len(r.counters))
	for name, value := range r.counters {
		counters[name] = value
	}
	r.mu.Unlock()

	if startedAt.IsZero() {
		startedAt = now
	}
	lines := []string{
		"# HELP pulse_up Whether the Pulse API process is running.",
		"# TYPE pulse_up gauge",
		"pulse_up 1",
		"# HELP pulse_process_uptime_seconds Process uptime in seconds.",
		"# TYPE pulse_process_uptime_seconds gauge",
		fmt.Sprintf("pulse_process_uptime_seconds %s", formatFloat(maxDuration(now.Sub(startedAt), 0).Seconds())),
		"# HELP pulse_http_requests_total HTTP requests by bounded route and status class.",
		"# TYPE pulse_http_requests_total counter",
		"# HELP pulse_http_request_duration_seconds HTTP request duration by bounded route and status class.",
		"# TYPE pulse_http_request_duration_seconds histogram",
	}
	keys := make([]requestKey, 0, len(requests))
	for key := range requests {
		keys = append(keys, key)
	}
	sort.Slice(keys, func(i, j int) bool {
		if keys[i].route != keys[j].route {
			return keys[i].route < keys[j].route
		}
		if keys[i].method != keys[j].method {
			return keys[i].method < keys[j].method
		}
		return keys[i].status < keys[j].status
	})
	for _, key := range keys {
		value := requests[key]
		labels := fmt.Sprintf(`method="%s",route="%s",status_class="%s"`, escapeLabel(key.method), escapeLabel(key.route), escapeLabel(key.status))
		lines = append(lines,
			fmt.Sprintf("pulse_http_requests_total{%s} %d", labels, value.count),
			fmt.Sprintf("pulse_http_request_duration_seconds_count{%s} %d", labels, value.count),
			fmt.Sprintf("pulse_http_request_duration_seconds_sum{%s} %s", labels, formatFloat(value.sum.Seconds())),
		)
	}
	for _, name := range sortedFloatNames(gauges) {
		lines = append(lines, fmt.Sprintf("%s %s", name, formatFloat(gauges[name])))
	}
	for _, name := range sortedUintNames(counters) {
		if name == "pulse_metrics_scrapes_total" {
			lines = append(lines, "# HELP pulse_metrics_scrapes_total Internal metrics scrapes.", "# TYPE pulse_metrics_scrapes_total counter")
		}
		lines = append(lines, fmt.Sprintf("%s %d", name, counters[name]))
	}
	return strings.Join(lines, "\n") + "\n"
}

func Middleware(registry *Registry, next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) {
		started := time.Now()
		writer := &statusWriter{ResponseWriter: w}
		next.ServeHTTP(writer, request)
		if registry != nil {
			registry.ObserveRequest(request.Method, request.URL.Path, writer.status, time.Since(started))
		}
	})
}

type statusWriter struct {
	http.ResponseWriter
	status      int
	wroteHeader bool
}

func (w *statusWriter) WriteHeader(status int) {
	if w.wroteHeader {
		return
	}
	w.status = status
	w.wroteHeader = true
	w.ResponseWriter.WriteHeader(status)
}

func (w *statusWriter) Write(body []byte) (int, error) {
	if !w.wroteHeader {
		w.WriteHeader(http.StatusOK)
	}
	return w.ResponseWriter.Write(body)
}

func (w *statusWriter) Unwrap() http.ResponseWriter { return w.ResponseWriter }

func (w *statusWriter) Flush() {
	if !w.wroteHeader {
		w.WriteHeader(http.StatusOK)
	}
	if flusher, ok := w.ResponseWriter.(http.Flusher); ok {
		flusher.Flush()
	}
}

func RouteCode(path string) string {
	switch {
	case path == "/healthz":
		return "healthz"
	case path == "/readyz":
		return "readyz"
	case path == "/auth/test-login":
		return "auth_test_login"
	case path == "/session/logout":
		return "session_logout"
	case path == "/api/v1/system/status":
		return "system_status"
	case path == "/api/v1/system/metrics":
		return "system_metrics"
	case path == "/api/v1/system/diagnostics":
		return "system_diagnostics"
	case strings.HasPrefix(path, "/api/v1/metrics/"):
		return "metrics"
	case strings.HasPrefix(path, "/api/v1/live"):
		return "live"
	case strings.HasPrefix(path, "/api/v1/"):
		return "api_other"
	default:
		return "other"
	}
}

func normalizeMethod(method string) string {
	switch method {
	case http.MethodGet, http.MethodHead, http.MethodPost, http.MethodPut, http.MethodPatch, http.MethodDelete, http.MethodOptions:
		return method
	default:
		return "OTHER"
	}
}

func statusClass(status int) string {
	switch {
	case status >= 200 && status < 300:
		return "2xx"
	case status >= 300 && status < 400:
		return "3xx"
	case status >= 400 && status < 500:
		return "4xx"
	case status >= 500 && status < 600:
		return "5xx"
	default:
		return "other"
	}
}

func validMetricName(name string) bool {
	if !strings.HasPrefix(name, "pulse_") || len(name) > 100 {
		return false
	}
	for index, char := range name {
		if (char < 'a' || char > 'z') && (char < '0' || char > '9') && char != '_' {
			return false
		}
		if index == 0 && char >= '0' && char <= '9' {
			return false
		}
	}
	return true
}

func sortedFloatNames(values map[string]float64) []string {
	names := make([]string, 0, len(values))
	for name := range values {
		names = append(names, name)
	}
	sort.Strings(names)
	return names
}

func sortedUintNames(values map[string]uint64) []string {
	names := make([]string, 0, len(values))
	for name := range values {
		names = append(names, name)
	}
	sort.Strings(names)
	return names
}

func escapeLabel(value string) string {
	return strings.NewReplacer(`\`, `\\`, `"`, `\"`, "\n", "\\n").Replace(value)
}
func formatFloat(value float64) string { return strconv.FormatFloat(value, 'f', -1, 64) }
func maxDuration(value, minimum time.Duration) time.Duration {
	if value < minimum {
		return minimum
	}
	return value
}