obs_test.go

package obs

import (
	"context"
	"errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/http/httptest"
	"net/url"
	"os"
	"strings"
	"sync/atomic"
	"syscall"
	"testing"

	"github.com/jackc/pgx/v5/pgxpool"
)

// resetForTest clears the package-level Init guard so independent
// tests can drive a fresh provider. Real services call Init exactly
// once; this only exists to make the tests isolated.
func resetForTest(t *testing.T) {
	t.Helper()
	initOnce.Store(false)
	promRegistry = nil
	meterProvider = nil
	tracerProvider = nil
	loggerProvider = nil
	jobDuration = nil
	jobCount = nil
}

func TestInitRequiresServiceName(t *testing.T) {
	resetForTest(t)
	if _, err := Init(context.Background(), Config{}); err == nil {
		t.Fatal("expected error for missing ServiceName")
	}
}

func TestInitIsIdempotent(t *testing.T) {
	resetForTest(t)
	shutdown, err := Init(context.Background(), Config{ServiceName: "test"})
	if err != nil {
		t.Fatalf("first Init: %v", err)
	}
	t.Cleanup(func() { _ = shutdown(context.Background()) })

	if _, err := Init(context.Background(), Config{ServiceName: "test"}); err == nil {
		t.Fatal("expected error on second Init")
	}
}

func TestJobEmitsOutcomeAndDuration(t *testing.T) {
	resetForTest(t)
	shutdown, err := Init(context.Background(), Config{ServiceName: "test"})
	if err != nil {
		t.Fatalf("Init: %v", err)
	}
	t.Cleanup(func() { _ = shutdown(context.Background()) })

	if err := Job(context.Background(), "ok-job", func(context.Context) error { return nil }); err != nil {
		t.Fatalf("ok-job: %v", err)
	}
	sentinel := errors.New("boom")
	if err := Job(context.Background(), "bad-job", func(context.Context) error { return sentinel }); !errors.Is(err, sentinel) {
		t.Fatalf("bad-job: want %v, got %v", sentinel, err)
	}

	body := scrape(t)
	for _, want := range []string{
		"worker_job_count",
		"worker_job_duration",
		`name="ok-job"`,
		`outcome="success"`,
		`name="bad-job"`,
		`outcome="error"`,
	} {
		if !strings.Contains(body, want) {
			t.Errorf("scrape missing %q", want)
		}
	}
}

func TestMiddlewareEmitsHTTPServerMetrics(t *testing.T) {
	resetForTest(t)
	shutdown, err := Init(context.Background(), Config{ServiceName: "test"})
	if err != nil {
		t.Fatalf("Init: %v", err)
	}
	t.Cleanup(func() { _ = shutdown(context.Background()) })

	var hits int32
	mux := http.NewServeMux()
	mux.HandleFunc("GET /ping", func(w http.ResponseWriter, _ *http.Request) {
		atomic.AddInt32(&hits, 1)
		_, _ = w.Write([]byte("pong"))
	})

	srv := httptest.NewServer(Middleware(mux))
	t.Cleanup(srv.Close)

	resp, err := http.Get(srv.URL + "/ping")
	if err != nil {
		t.Fatalf("GET /ping: %v", err)
	}
	_, _ = io.Copy(io.Discard, resp.Body)
	_ = resp.Body.Close()

	if hits != 1 {
		t.Fatalf("handler hit count: want 1, got %d", hits)
	}
	body := scrape(t)
	if !strings.Contains(body, "http_server_request") {
		t.Errorf("scrape missing http_server_request*; got:\n%s", body)
	}
}

func TestTransportEmitsHTTPClientMetrics(t *testing.T) {
	resetForTest(t)
	shutdown, err := Init(context.Background(), Config{ServiceName: "test"})
	if err != nil {
		t.Fatalf("Init: %v", err)
	}
	t.Cleanup(func() { _ = shutdown(context.Background()) })

	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		_, _ = w.Write([]byte("pong"))
	}))
	t.Cleanup(srv.Close)

	client := HTTPClient(nil)
	resp, err := client.Get(srv.URL)
	if err != nil {
		t.Fatalf("client GET: %v", err)
	}
	_, _ = io.Copy(io.Discard, resp.Body)
	_ = resp.Body.Close()

	body := scrape(t)
	if !strings.Contains(body, "http_client_request_duration") {
		t.Errorf("scrape missing http_client_request_duration*; got:\n%s", body)
	}
}

// TestClassify pins the error.type each transport failure is recorded under,
// and that classifying one changes nothing a caller can observe but its type.
func TestClassify(t *testing.T) {
	opErr := func(op string, err error) *net.OpError {
		return &net.OpError{Op: op, Net: "tcp", Err: err}
	}
	dnsFail := opErr("dial", &net.DNSError{Err: "no such host", Name: "pypi.org", IsNotFound: true})
	refused := opErr("dial", os.NewSyscallError("connect", syscall.ECONNREFUSED))
	reset := opErr("read", os.NewSyscallError("read", syscall.ECONNRESET))
	brokenPipe := opErr("write", os.NewSyscallError("write", syscall.EPIPE))
	unreachable := opErr("dial", os.NewSyscallError("connect", syscall.EHOSTUNREACH))
	netUnreachable := opErr("dial", os.NewSyscallError("connect", syscall.ENETUNREACH))
	timedOut := opErr("dial", os.ErrDeadlineExceeded)
	odd := opErr("dial", errors.New("something else"))

	for _, tc := range []struct {
		name     string
		err      error
		wantType string // as otelhttp records it, minus the package path
		timeout  bool
	}{
		{"dns", dnsFail, "obs.dnsError", false},
		{"refused", refused, "obs.refusedError", false},
		{"reset", reset, "obs.resetError", false},
		{"broken pipe", brokenPipe, "obs.resetError", false},
		{"host unreachable", unreachable, "obs.unreachableError", false},
		{"network unreachable", netUnreachable, "obs.unreachableError", false},
		{"timeout", timedOut, "obs.timeoutError", true},
		{"dial timeout wrapped by a caller", fmt.Errorf("fetch: %w", timedOut), "obs.timeoutError", true},
	} {
		t.Run(tc.name, func(t *testing.T) {
			got := classify(tc.err)
			if typ := fmt.Sprintf("%T", got); typ != tc.wantType {
				t.Fatalf("type = %s, want %s", typ, tc.wantType)
			}
			if got.Error() != tc.err.Error() {
				t.Errorf("Error() = %q, want %q", got.Error(), tc.err.Error())
			}
			if !errors.Is(got, tc.err) {
				t.Error("errors.Is(classified, original) = false")
			}
			var op *net.OpError
			if !errors.As(got, &op) {
				t.Error("errors.As(classified, *net.OpError) = false")
			}
			ne, ok := got.(net.Error)
			if !ok {
				t.Fatal("classified error is not a net.Error")
			}
			if ne.Timeout() != tc.timeout {
				t.Errorf("Timeout() = %v, want %v", ne.Timeout(), tc.timeout)
			}
			// http.Client wraps transport errors in *url.Error, whose Timeout
			// asserts the inner error directly rather than unwrapping.
			if ue := (&url.Error{Op: "Get", URL: "https://pypi.org/", Err: got}); ue.Timeout() != tc.timeout {
				t.Errorf("url.Error.Timeout() = %v, want %v", ue.Timeout(), tc.timeout)
			}
		})
	}

	if !errors.Is(classify(refused), syscall.ECONNREFUSED) {
		t.Error("refused no longer matches syscall.ECONNREFUSED")
	}

	// Errors that already name their cause, and OpErrors with a cause outside
	// the five, come back as the same value.
	for _, err := range []error{
		context.DeadlineExceeded,
		context.Canceled,
		io.ErrUnexpectedEOF,
		odd,
	} {
		if got := classify(err); got != err {
			t.Errorf("classify(%T %v) = %T, want it unchanged", err, err, got)
		}
	}
}

// TestTransportLabelsErrorCause runs a real refused connection through the
// instrumented client and reads the label back off the metrics endpoint.
func TestTransportLabelsErrorCause(t *testing.T) {
	resetForTest(t)
	shutdown, err := Init(context.Background(), Config{ServiceName: "test"})
	if err != nil {
		t.Fatalf("Init: %v", err)
	}
	t.Cleanup(func() { _ = shutdown(context.Background()) })

	// A port that was listening a moment ago and is not now refuses at once.
	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatalf("listen: %v", err)
	}
	addr := ln.Addr().String()
	_ = ln.Close()

	_, err = HTTPClient(nil).Get("http://" + addr + "/")
	if !errors.Is(err, syscall.ECONNREFUSED) {
		t.Fatalf("GET closed port: err = %v, want ECONNREFUSED", err)
	}

	body := scrape(t)
	if want := `error_type="github.com/atomdrift-project/obs.refusedError"`; !strings.Contains(body, want) {
		t.Errorf("scrape missing %s; got:\n%s", want, body)
	}
}

func TestMetricsHandlerBeforeInit(t *testing.T) {
	resetForTest(t)
	rec := httptest.NewRecorder()
	MetricsHandler().ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/_/metrik", nil))
	if rec.Code != http.StatusServiceUnavailable {
		t.Fatalf("pre-Init code: want 503, got %d", rec.Code)
	}
}

func TestParseEndpoint(t *testing.T) {
	cases := []struct {
		raw        string
		wantHost   string
		wantPath   string
		wantInsec  bool
		forceInsec bool
	}{
		{"http://prom:9090/api/v1/otlp", "prom:9090", "/api/v1/otlp", true, false},
		{"https://otlp.example.com", "otlp.example.com", "", false, false},
		{"https://otlp.example.com", "otlp.example.com", "", true, true},
		{"prom:9090", "prom:9090", "", true, false}, // scheme-less defaults to http
	}
	for _, c := range cases {
		host, path, insec, err := parseEndpoint(c.raw, c.forceInsec)
		if err != nil {
			t.Errorf("parseEndpoint(%q): %v", c.raw, err)
			continue
		}
		if host != c.wantHost || path != c.wantPath || insec != c.wantInsec {
			t.Errorf("parseEndpoint(%q, force=%v) = (%q,%q,%v); want (%q,%q,%v)",
				c.raw, c.forceInsec, host, path, insec, c.wantHost, c.wantPath, c.wantInsec)
		}
	}
}

// clearOTLPEnv blanks every OTLP endpoint env var so a case starts from a
// known state regardless of the ambient environment.
func clearOTLPEnv(t *testing.T) {
	t.Helper()
	for _, k := range []string{
		"OTEL_EXPORTER_OTLP_ENDPOINT",
		"OTEL_EXPORTER_OTLP_METRICS_ENDPOINT",
		"OTEL_EXPORTER_OTLP_LOGS_ENDPOINT",
		"OTEL_EXPORTER_OTLP_TRACES_ENDPOINT",
	} {
		t.Setenv(k, "")
	}
}

func TestLogRoutingDefaultsToLoki(t *testing.T) {
	clearOTLPEnv(t)
	t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://otel:9090/api/v1/otlp")

	// Metrics keep the base endpoint (Prometheus), with /v1/<signal> appended.
	if got := signalEndpoint(Config{}, "METRICS"); got.endpoint != "http://otel:9090/api/v1/otlp" || !got.appendSignal {
		t.Errorf("metrics: got %+v, want base with append", got)
	}
	// Logs fall back to Loki, used verbatim (no /v1/logs appended).
	if got := resolveLogs(Config{}); got.endpoint != DefaultLogsEndpoint || got.appendSignal {
		t.Errorf("logs: got %+v, want Loki default verbatim", got)
	}
	// Traces stay disabled: no trace backend behind the base endpoint.
	if got := resolveTraces(Config{}); got.endpoint != "" {
		t.Errorf("traces: got %+v, want disabled", got)
	}
}

func TestLogRoutingHonorsExplicitEndpoint(t *testing.T) {
	clearOTLPEnv(t)
	t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://otel:9090/api/v1/otlp")
	t.Setenv("OTEL_EXPORTER_OTLP_LOGS_ENDPOINT", "http://loki-2:3100/otlp/v1/logs")
	if got := resolveLogs(Config{}); got.endpoint != "http://loki-2:3100/otlp/v1/logs" || got.appendSignal {
		t.Errorf("logs: got %+v, want explicit endpoint verbatim", got)
	}
}

func TestTraceRoutingRequiresExplicitEndpoint(t *testing.T) {
	clearOTLPEnv(t)
	t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://otel:9090/api/v1/otlp")
	if got := resolveTraces(Config{}); got.endpoint != "" {
		t.Fatalf("traces from base: got %+v, want disabled", got)
	}
	t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "http://tempo:4318/v1/traces")
	if got := resolveTraces(Config{}); got.endpoint != "http://tempo:4318/v1/traces" || got.appendSignal {
		t.Errorf("traces: got %+v, want explicit endpoint verbatim", got)
	}
}

// TestDefaultsToSelfHostedFleet verifies that, with no OTLP env vars and no
// Config.Endpoint, metrics push to [DefaultEndpoint], logs divert to
// [DefaultLogsEndpoint], and traces stay off (no Tempo behind the base).
func TestDefaultsToSelfHostedFleet(t *testing.T) {
	clearOTLPEnv(t)
	if got := signalEndpoint(Config{}, "METRICS"); got.endpoint != DefaultEndpoint || !got.appendSignal {
		t.Errorf("metrics: got %+v, want DefaultEndpoint with append", got)
	}
	if got := resolveLogs(Config{}); got.endpoint != DefaultLogsEndpoint || got.appendSignal {
		t.Errorf("logs: got %+v, want Loki default verbatim", got)
	}
	if got := resolveTraces(Config{}); got.endpoint != "" {
		t.Errorf("traces: got %+v, want disabled", got)
	}
}

// TestExplicitEndpointOverridesDefault confirms an env base endpoint still
// wins over [DefaultEndpoint], while keeping the self-hosted log/trace split.
func TestExplicitEndpointOverridesDefault(t *testing.T) {
	clearOTLPEnv(t)
	t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://prom-alt:9090/api/v1/otlp")

	got := signalEndpoint(Config{}, "METRICS")
	if got.endpoint != "http://prom-alt:9090/api/v1/otlp" || !got.appendSignal {
		t.Errorf("metrics: got %+v, want explicit base with append", got)
	}
	if got := resolveLogs(Config{}); got.endpoint != DefaultLogsEndpoint {
		t.Errorf("logs: got %+v, want Loki default", got)
	}
	if got := resolveTraces(Config{}); got.endpoint != "" {
		t.Errorf("traces: got %+v, want disabled", got)
	}
}

// TestPoolStatsExposesContentionMetrics registers a lazily-created pool (no
// live database needed — Stat works on an unconnected pool) and asserts the
// pool gauges, including the saturation and contention series, reach the scrape.
func TestPoolStatsExposesContentionMetrics(t *testing.T) {
	resetForTest(t)
	shutdown, err := Init(context.Background(), Config{ServiceName: "test"})
	if err != nil {
		t.Fatalf("Init: %v", err)
	}
	t.Cleanup(func() { _ = shutdown(context.Background()) })

	cfg, err := pgxpool.ParseConfig("postgres://u:p@127.0.0.1:1/db")
	if err != nil {
		t.Fatalf("ParseConfig: %v", err)
	}
	pool, err := pgxpool.NewWithConfig(context.Background(), cfg)
	if err != nil {
		t.Fatalf("NewWithConfig: %v", err)
	}
	t.Cleanup(pool.Close)

	if err := PoolStats("test", pool); err != nil {
		t.Fatalf("PoolStats: %v", err)
	}

	body := scrape(t)
	for _, w := range []string{
		"db_pool_max", "db_pool_acquired", "db_pool_constructing",
		"db_pool_canceled_acquire_total", "db_pool_new_conns_total",
		`pool="test"`,
	} {
		if !strings.Contains(body, w) {
			t.Errorf("scrape missing %q", w)
		}
	}
}

// TestRuntimeMetricsExposed asserts Init wires the standard Go-runtime and
// process collectors so every service publishes memory, CPU, goroutine, and
// file-descriptor metrics under their canonical names without extra wiring.
func TestRuntimeMetricsExposed(t *testing.T) {
	resetForTest(t)
	shutdown, err := Init(context.Background(), Config{ServiceName: "test"})
	if err != nil {
		t.Fatalf("Init: %v", err)
	}
	t.Cleanup(func() { _ = shutdown(context.Background()) })

	body := scrape(t)
	// process_* metrics depend on /proc and are absent on non-Linux; gate on
	// them only where they exist so the suite stays portable. go_* are always
	// present.
	want := []string{"go_goroutines", "go_memstats_heap_inuse_bytes", "go_gc_duration_seconds"}
	if strings.Contains(body, "process_cpu_seconds_total") {
		want = append(want, "process_resident_memory_bytes", "process_open_fds")
	}
	for _, w := range want {
		if !strings.Contains(body, w) {
			t.Errorf("scrape missing %q", w)
		}
	}
}

func scrape(t *testing.T) string {
	t.Helper()
	rec := httptest.NewRecorder()
	MetricsHandler().ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/_/metrik", nil))
	if rec.Code != http.StatusOK {
		t.Fatalf("scrape code: want 200, got %d", rec.Code)
	}
	return rec.Body.String()
}