orbit/internal/observability/metrics_test.go

package observability

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"math"
	"net/http"
	"net/http/httptest"
	"os"
	"runtime"
	"strconv"
	"strings"
	"sync"
	"testing"
	"time"
)

func TestConcurrentScrapesAndUpdates(t *testing.T) {
	r := NewRegistry()
	labels := Labels{"method": "GET", "route": "tasks", "status_class": "2xx"}
	var wg sync.WaitGroup
	for i := 0; i < 16; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for n := 0; n < 100; n++ {
				r.AddCounter("orbit_http_requests_total", 1, labels)
				r.ObserveHistogram("orbit_http_request_duration_seconds", .01, labels)
				if n%10 == 0 {
					r.Render()
				}
			}
		}()
	}
	wg.Wait()
	out := r.Render()
	for _, want := range []string{
		`orbit_http_requests_total{method="GET",route="tasks",status_class="2xx"} 1600`,
		`orbit_http_request_duration_seconds_count{method="GET",route="tasks",status_class="2xx"} 1600`,
		`orbit_http_request_duration_seconds_bucket{method="GET",route="tasks",status_class="2xx",le="0.005"} 0`,
		`orbit_http_request_duration_seconds_bucket{method="GET",route="tasks",status_class="2xx",le="+Inf"} 1600`,
	} {
		if !strings.Contains(out, want) {
			t.Fatal("lost samples or inconsistent histogram", want)
		}
	}
}
func TestCatalogRejectsNamesTypesAndAllDynamicLabels(t *testing.T) {
	r := NewRegistry()
	families := len(r.families)
	for i := 0; i < 3000; i++ {
		r.AddCounter("untrusted_"+strconv.Itoa(i), 1, nil)
		r.AddCounter("orbit_http_requests_total", 1, Labels{"method": "GET", "route": "private-" + strconv.Itoa(i), "status_class": "2xx"})
	}
	r.ObserveHistogram("orbit_http_requests_in_flight", 1, nil) // Must not panic on a type mismatch.
	r.SetGauge("orbit_http_requests_total", 1, nil)
	r.SetGauge("orbit_go_goroutines", 1, Labels{"tenant": "private"})
	if len(r.families) != families || r.rejected.Load() != 6003 {
		t.Fatal("unbounded family creation or invalid samples accepted")
	}
	if out := r.Render(); strings.Contains(out, "private") || strings.Contains(out, "untrusted") {
		t.Fatal("dynamic data exposed")
	}
	// Exercise every possible HTTP combination, then verify known series continue updating.
	for _, method := range methods {
		for _, route := range routes {
			for _, status := range statusClasses {
				labels := Labels{"method": method, "route": route, "status_class": status}
				r.AddCounter("orbit_http_requests_total", 1, labels)
				r.ObserveHistogram("orbit_http_request_duration_seconds", 1, labels)
			}
		}
	}
	labels := Labels{"method": "GET", "route": "tasks", "status_class": "2xx"}
	r.AddCounter("orbit_http_requests_total", 1, labels)
	if n := len(r.families["orbit_http_requests_total"].series); n != len(methods)*len(routes)*len(statusClasses) {
		t.Fatal("wrong series bound", n)
	}
	if !strings.Contains(r.Render(), `orbit_http_requests_total{method="GET",route="tasks",status_class="2xx"} 2`) {
		t.Fatal("existing series stopped updating")
	}
}
func TestInvalidNumbersAndMonotoneCounters(t *testing.T) {
	r := NewRegistry()
	r.SetCounter("orbit_db_wait_count_total", 10, nil)
	for _, value := range []float64{-1, math.NaN(), math.Inf(1)} {
		r.AddCounter("orbit_db_wait_count_total", value, nil)
		r.ObserveHistogram("orbit_http_request_duration_seconds", value, nil)
	}
	r.SetCounter("orbit_db_wait_count_total", 9, nil)
	if !strings.Contains(r.Render(), "orbit_db_wait_count_total 10\n") || r.rejected.Load() != 7 {
		t.Fatal("invalid counter update")
	}
}
func TestRuntimeObservationDoesNotExposeErrors(t *testing.T) {
	r := NewRegistry()
	r.RecordRuntimeConfig(RuntimeConfig{WorkflowTickInterval: time.Second, ScheduleTickInterval: 1500 * time.Millisecond, RecoveryTickInterval: 2 * time.Second, BatchLimit: 37, ShutdownTimeout: 45 * time.Second, WorkerConnectionLimit: 512, EventConnectionLimit: 256, DispatchWorkers: 16, DispatchCapacity: 2048})
	r.RecordDraining(true)
	r.RecordBatch("workflow", time.Now().Add(-time.Millisecond), nil)
	r.RecordBatch("workflow", time.Now(), errors.New("private-dsn"))
	r.RecordBatch("schedule", time.Now(), context.Canceled)
	r.RecordDatabase(sql.DBStats{MaxOpenConnections: 20, OpenConnections: 4, InUse: 3, Idle: 1, WaitCount: 9, WaitDuration: 2 * time.Second})
	out := r.Render()
	for _, want := range []string{`orbit_runtime_batch_total{component="workflow",outcome="success"} 1`, `orbit_runtime_batch_total{component="workflow",outcome="error"} 1`, `orbit_runtime_batch_total{component="schedule",outcome="cancelled"} 1`, "orbit_db_connections_in_use 3\n", "orbit_db_wait_duration_seconds_total 2\n"} {
		if !strings.Contains(out, want) {
			t.Fatal("missing runtime evidence", want)
		}
	}
	for _, want := range []string{`orbit_runtime_tick_interval_seconds{component="workflow"} 1`, `orbit_runtime_tick_interval_seconds{component="schedule"} 1.5`, `orbit_runtime_tick_interval_seconds{component="execution_recovery"} 2`, "orbit_runtime_batch_limit 37\n", "orbit_shutdown_timeout_seconds 45\n", "orbit_worker_connection_limit 512\n", "orbit_event_connection_limit 256\n", "orbit_dispatch_workers 16\n", "orbit_dispatch_capacity 2048\n", "orbit_process_draining 1\n"} {
		if !strings.Contains(out, want) {
			t.Fatal("missing runtime configuration", want)
		}
	}
	if !strings.Contains(out, `orbit_build_info{go_version="`+runtime.Version()+`",revision="unknown",version="dev"} 1`) {
		t.Fatal("missing build identity")
	}
	if strings.Contains(out, "private-dsn") {
		t.Fatal("error data leaked")
	}
	var disabled *Registry
	disabled.RecordBatch("workflow", time.Now(), nil)
	disabled.RecordDatabase(sql.DBStats{})
}

func TestBusinessSnapshotKeepsLastGoodStateOnFailure(t *testing.T) {
	r := NewRegistry()
	started := time.Now().Add(-25 * time.Millisecond)
	r.RecordBusinessSnapshot(BusinessSnapshot{
		TaskStates:                 map[string]uint64{"WAITING": 3},
		TaskOldestAgeSeconds:       12.5,
		DeliveryStates:             map[string]uint64{"PENDING": 2},
		DeliveryOldestAgeSeconds:   8.25,
		WorkflowStates:             map[string]uint64{"RUNNING": 1},
		WorkflowOldestAgeSeconds:   4.75,
		InboxStates:                map[string]uint64{"PENDING": 6},
		InboxOldestAgeSeconds:      3.25,
		RecentEvents:               map[string]uint64{"task.accepted": 5},
		TaskTerminalLatencySeconds: 2.5, TaskTerminalLatencyMaxSeconds: 7.5,
		WorkflowTerminalLatencySeconds: 3.5, WorkflowTerminalLatencyMaxSeconds: 9.5,
		TaskAcceptToStartLatencySeconds: 1.5, TaskAcceptToStartLatencyMaxSeconds: 4.5,
		TaskAcceptToTerminalLatencySeconds: 5.5, TaskAcceptToTerminalLatencyMaxSeconds: 8.5,
		WorkflowStartToTerminalLatencySeconds: 6.5, WorkflowStartToTerminalLatencyMaxSeconds: 10.5,
		Truncated: true,
	}, started, nil)
	first := r.Render()
	for _, want := range []string{
		`orbit_pulse_tasks_state{state="WAITING"} 3`,
		`orbit_result_deliveries_state{state="PENDING"} 2`,
		`orbit_weave_workflows_state{state="RUNNING"} 1`,
		`orbit_weave_inbox_state{state="PENDING"} 6`,
		`orbit_business_events_recent{event="task.accepted"} 5`,
		"orbit_pulse_task_terminal_latency_seconds 2.5",
		"orbit_weave_workflow_terminal_latency_max_seconds 9.5",
		"orbit_pulse_task_accept_to_start_latency_seconds 1.5",
		"orbit_pulse_task_accept_to_terminal_latency_max_seconds 8.5",
		"orbit_weave_workflow_start_to_terminal_latency_seconds 6.5",
		"orbit_pulse_tasks_oldest_age_seconds 12.5",
		"orbit_weave_inbox_oldest_age_seconds 3.25",
		"orbit_business_snapshot_truncated 1",
		`orbit_business_snapshot_total{outcome="success"} 1`,
	} {
		if !strings.Contains(first, want) {
			t.Fatalf("missing business sample %q", want)
		}
	}
	lastSuccess := r.families["orbit_business_snapshot_last_success_timestamp_seconds"].series[""].value.Load()
	r.RecordBusinessSnapshot(BusinessSnapshot{TaskStates: map[string]uint64{"WAITING": 99}}, time.Now(), context.DeadlineExceeded)
	second := r.Render()
	if !strings.Contains(second, `orbit_business_snapshot_total{outcome="timeout"} 1`) || !strings.Contains(second, `orbit_pulse_tasks_state{state="WAITING"} 3`) {
		t.Fatal("failed sample overwrote good state")
	}
	if got := r.families["orbit_business_snapshot_last_success_timestamp_seconds"].series[""].value.Load(); got != lastSuccess {
		t.Fatal("failed sample changed last success timestamp")
	}
}
func TestHTTPBoundedLabelsAndPanicPreserved(t *testing.T) {
	r := NewRegistry()
	h := r.Middleware(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { w.WriteHeader(202) }))
	for i := 0; i < 100; i++ {
		h.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest("GET", fmt.Sprintf("/v1/tasks/private-%d?token=private-key", i), nil))
	}
	out := r.Render()
	if strings.Contains(out, "private-") || strings.Count(out, "orbit_http_requests_total{") != 1 {
		t.Fatal("request label cardinality or privacy")
	}
	panics := r.Middleware(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { panic("original panic") }))
	func() {
		defer func() {
			if recover() != "original panic" {
				t.Fatal("panic changed")
			}
		}()
		panics.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest("GET", "/v1/queues", nil))
	}()
	out = r.Render()
	if !strings.Contains(out, "orbit_http_requests_in_flight 0\n") || !strings.Contains(out, "orbit_http_panics_total 1\n") || !strings.Contains(out, `route="queues",status_class="5xx"} 1`) {
		t.Fatal("panic leaked in-flight state")
	}
}
func TestScrapeConcurrencyRejectsRatherThanBlockingBusiness(t *testing.T) {
	r := NewRegistry()
	r.scrape <- struct{}{}
	r.scrape <- struct{}{}
	rr := httptest.NewRecorder()
	r.ServeHTTP(rr, httptest.NewRequest("GET", "/metrics", nil))
	if rr.Code != 503 || rr.Header().Get("Retry-After") != "1" {
		t.Fatal("unbounded scrape")
	}
	// Registry recording is independent of the scrape admission slots.
	r.RecordBatch("workflow", time.Now(), nil)
	<-r.scrape
	<-r.scrape
	rr = httptest.NewRecorder()
	r.ServeHTTP(rr, httptest.NewRequest("GET", "/metrics", nil))
	if rr.Code != 200 || !strings.HasSuffix(rr.Body.String(), "\n") {
		t.Fatal("scrape did not recover")
	}
}

// TestExportExpositionFixture optionally exports the full finite HTTP catalog
// for an independent Prometheus promtool check; ordinary tests need no tool.
func TestExportExpositionFixture(t *testing.T) {
	r := NewRegistry()
	for _, method := range methods {
		for _, route := range routes {
			for _, status := range statusClasses {
				labels := Labels{"method": method, "route": route, "status_class": status}
				r.AddCounter("orbit_http_requests_total", 1, labels)
				r.ObserveHistogram("orbit_http_request_duration_seconds", .02, labels)
			}
		}
	}
	r.RecordDatabase(sql.DBStats{MaxOpenConnections: 20})
	r.RecordBatch("workflow", time.Now(), nil)
	body := r.Render()
	if len(body) > 2<<20 {
		t.Fatal("finite catalog exceeded bounded output size")
	}
	if path := os.Getenv("ORBIT_TEST_METRICS_OUTPUT"); path != "" {
		if err := os.WriteFile(path, []byte(body), 0600); err != nil {
			t.Fatal(err)
		}
	}
}

type discardWriter struct{ header http.Header }

func (w *discardWriter) Header() http.Header         { return w.header }
func (w *discardWriter) Write(p []byte) (int, error) { return len(p), nil }
func (w *discardWriter) WriteHeader(int)             {}
func BenchmarkHTTPMiddleware(b *testing.B) {
	for _, enabled := range []bool{false, true} {
		b.Run(strconv.FormatBool(enabled), func(b *testing.B) {
			var h http.Handler = http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(204) })
			if enabled {
				h = NewRegistry().Middleware(h)
			}
			b.ReportAllocs()
			b.ResetTimer()
			b.RunParallel(func(pb *testing.PB) {
				req := httptest.NewRequest("GET", "/v1/tasks/id", nil)
				w := &discardWriter{header: make(http.Header)}
				for pb.Next() {
					h.ServeHTTP(w, req)
				}
			})
		})
	}
}