mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 09:01:31 +00:00
refactor: consume bluebottle otelsetup, drop the local bootstrap
otelsetup/otelsetup.go and otelsetup/slog.go were a near-copy of the shared bootstrap. Entrypoints now call bluebottle directly, passing WithEnvironmentFallback for SPINIFEX_CI_ENV; the package keeps only the HTTP middleware and request metrics. Init reattaches the OTLP bridge to a default logger installed before it ran, so the local addFanoutHandler is no longer needed. The bridge is now level-gated, so services no longer ship Debug records to the sink while running at Info.
This commit is contained in:
@@ -18,8 +18,8 @@ import (
|
||||
"os/signal"
|
||||
"syscall"
|
||||
|
||||
"github.com/mulgadc/bluebottle/pkg/otelsetup"
|
||||
"github.com/mulgadc/spinifex/spinifex/lbagent"
|
||||
"github.com/mulgadc/spinifex/spinifex/otelsetup"
|
||||
|
||||
_ "github.com/mulgadc/spinifex/internal/fipsboot"
|
||||
)
|
||||
@@ -28,7 +28,7 @@ func main() {
|
||||
// Without this the agent falls back to slog's default text handler, which folds
|
||||
// attributes into the message so the console scraper recovers a level and nothing
|
||||
// else. No OTLP export: the agent's only egress is the gateway it is watching.
|
||||
otelsetup.SetDefaultJSONLogger(slog.LevelInfo)
|
||||
otelsetup.SetDefaultJSONLogger("lb-agent", slog.LevelInfo)
|
||||
|
||||
var (
|
||||
lbID string
|
||||
|
||||
@@ -9,10 +9,10 @@ import (
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/mulgadc/bluebottle/pkg/otelsetup"
|
||||
"github.com/mulgadc/spinifex/spinifex/config"
|
||||
handlers_dns "github.com/mulgadc/spinifex/spinifex/handlers/dns"
|
||||
"github.com/mulgadc/spinifex/spinifex/network/external"
|
||||
"github.com/mulgadc/spinifex/spinifex/otelsetup"
|
||||
"github.com/mulgadc/spinifex/spinifex/service"
|
||||
"github.com/mulgadc/spinifex/spinifex/services/awsgw"
|
||||
"github.com/mulgadc/spinifex/spinifex/services/nats"
|
||||
@@ -39,9 +39,10 @@ func initTelemetry(serviceName string, debug bool) func() {
|
||||
if debug {
|
||||
level = slog.LevelDebug
|
||||
}
|
||||
otelsetup.SetDefaultJSONLogger(level)
|
||||
otelsetup.SetDefaultJSONLogger(serviceName, level)
|
||||
|
||||
shutdown, err := otelsetup.Init(context.Background(), serviceName)
|
||||
shutdown, err := otelsetup.Init(context.Background(), serviceName,
|
||||
otelsetup.WithEnvironmentFallback("SPINIFEX_CI_ENV"))
|
||||
if err != nil {
|
||||
slog.Warn("otel init", "service", serviceName, "error", err)
|
||||
return func() {}
|
||||
|
||||
@@ -18,6 +18,7 @@ import (
|
||||
"github.com/google/uuid"
|
||||
"github.com/mulgadc/bluebottle/pkg/auth"
|
||||
"github.com/mulgadc/bluebottle/pkg/iampolicy"
|
||||
bbotel "github.com/mulgadc/bluebottle/pkg/otelsetup"
|
||||
"github.com/mulgadc/bluebottle/pkg/ratelimit"
|
||||
"github.com/mulgadc/bluebottle/pkg/sigv4"
|
||||
"github.com/mulgadc/spinifex/spinifex/awserrors"
|
||||
@@ -204,7 +205,7 @@ func (gw *GatewayConfig) SetupRoutes() http.Handler {
|
||||
|
||||
// Adjust the level only. Reinstalling the default logger here would drop the OTLP
|
||||
// bridge Init fanned on at startup, blinding the sink to every line after this.
|
||||
otelsetup.SetLevel(logLevel)
|
||||
bbotel.SetLevel(logLevel)
|
||||
|
||||
if gw.RateLimiter == nil {
|
||||
gw.RateLimiter = NewAuthRateLimiter()
|
||||
|
||||
@@ -1,3 +1,7 @@
|
||||
// Package otelsetup carries spinifex's HTTP request instrumentation: a tracing
|
||||
// middleware and the request counter/duration histogram it feeds. The OTel
|
||||
// bootstrap itself lives in bluebottle/pkg/otelsetup, which service entrypoints
|
||||
// call directly.
|
||||
package otelsetup
|
||||
|
||||
import (
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
bbotel "github.com/mulgadc/bluebottle/pkg/otelsetup"
|
||||
"go.opentelemetry.io/otel"
|
||||
"go.opentelemetry.io/otel/codes"
|
||||
sdktrace "go.opentelemetry.io/otel/sdk/trace"
|
||||
@@ -103,10 +104,10 @@ func TestHTTPMiddleware5xxSetsErrorStatus(t *testing.T) {
|
||||
|
||||
func TestHTTPMiddlewareExtractsTraceparent(t *testing.T) {
|
||||
sr := withRecorder(t)
|
||||
// Extraction needs the W3C propagator installed (Init does this even
|
||||
// without an endpoint).
|
||||
// Extraction needs the W3C propagator installed (the shared Init does this
|
||||
// even without an endpoint).
|
||||
t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "")
|
||||
if _, err := Init(t.Context(), "test-svc"); err != nil {
|
||||
if _, err := bbotel.Init(t.Context(), "test-svc"); err != nil {
|
||||
t.Fatalf("Init: %v", err)
|
||||
}
|
||||
|
||||
|
||||
@@ -1,190 +0,0 @@
|
||||
// Package otelsetup configures OpenTelemetry tracing and metrics for Mulga
|
||||
// services. Export is gated on the standard OTLP environment variables; with
|
||||
// no endpoint configured Init is a functional no-op, so instrumented binaries
|
||||
// ship everywhere at zero cost.
|
||||
package otelsetup
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"os"
|
||||
"runtime/debug"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"go.opentelemetry.io/contrib/bridges/otelslog"
|
||||
otelruntime "go.opentelemetry.io/contrib/instrumentation/runtime"
|
||||
"go.opentelemetry.io/otel"
|
||||
"go.opentelemetry.io/otel/attribute"
|
||||
"go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc"
|
||||
"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc"
|
||||
"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
|
||||
loggerglobal "go.opentelemetry.io/otel/log/global"
|
||||
"go.opentelemetry.io/otel/propagation"
|
||||
sdklog "go.opentelemetry.io/otel/sdk/log"
|
||||
sdkmetric "go.opentelemetry.io/otel/sdk/metric"
|
||||
"go.opentelemetry.io/otel/sdk/resource"
|
||||
sdktrace "go.opentelemetry.io/otel/sdk/trace"
|
||||
semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
|
||||
)
|
||||
|
||||
const shutdownTimeout = 10 * time.Second
|
||||
|
||||
// Init installs global tracer and meter providers exporting OTLP over gRPC,
|
||||
// plus the W3C trace-context propagator. The returned shutdown func flushes
|
||||
// and stops both providers; it is always safe to call. When no OTLP endpoint
|
||||
// is configured (or OTEL_SDK_DISABLED=true) only the propagator is installed
|
||||
// and the globals stay no-op.
|
||||
func Init(ctx context.Context, serviceName string) (func(context.Context) error, error) {
|
||||
otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(
|
||||
propagation.TraceContext{}, propagation.Baggage{}))
|
||||
|
||||
if !exportEnabled() {
|
||||
return func(context.Context) error { return nil }, nil
|
||||
}
|
||||
|
||||
res, err := newResource(ctx, serviceName)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("otel resource for %s: %w", serviceName, err)
|
||||
}
|
||||
|
||||
traceExp, err := otlptracegrpc.New(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("otlp trace exporter for %s: %w", serviceName, err)
|
||||
}
|
||||
tp := sdktrace.NewTracerProvider(
|
||||
sdktrace.WithBatcher(traceExp),
|
||||
sdktrace.WithResource(res),
|
||||
sdktrace.WithSampler(rootSampler()),
|
||||
)
|
||||
otel.SetTracerProvider(tp)
|
||||
|
||||
metricExp, err := otlpmetricgrpc.New(ctx)
|
||||
if err != nil {
|
||||
shutdownCtx, cancel := context.WithTimeout(context.Background(), shutdownTimeout)
|
||||
defer cancel()
|
||||
return nil, errors.Join(
|
||||
fmt.Errorf("otlp metric exporter for %s: %w", serviceName, err),
|
||||
tp.Shutdown(shutdownCtx))
|
||||
}
|
||||
mp := sdkmetric.NewMeterProvider(
|
||||
sdkmetric.WithReader(sdkmetric.NewPeriodicReader(metricExp)),
|
||||
sdkmetric.WithResource(res),
|
||||
)
|
||||
otel.SetMeterProvider(mp)
|
||||
|
||||
logExp, err := otlploggrpc.New(ctx)
|
||||
if err != nil {
|
||||
shutdownCtx, cancel := context.WithTimeout(context.Background(), shutdownTimeout)
|
||||
defer cancel()
|
||||
return nil, errors.Join(
|
||||
fmt.Errorf("otlp log exporter for %s: %w", serviceName, err),
|
||||
tp.Shutdown(shutdownCtx), mp.Shutdown(shutdownCtx))
|
||||
}
|
||||
lp := sdklog.NewLoggerProvider(
|
||||
sdklog.WithResource(res),
|
||||
sdklog.WithProcessor(sdklog.NewBatchProcessor(logExp)),
|
||||
)
|
||||
loggerglobal.SetLoggerProvider(lp)
|
||||
addFanoutHandler(otelslog.NewHandler(serviceName, otelslog.WithLoggerProvider(lp)))
|
||||
|
||||
if err := otelruntime.Start(); err != nil {
|
||||
slog.Warn("otel runtime metrics disabled", "err", err)
|
||||
}
|
||||
|
||||
shutdown := func(ctx context.Context) error {
|
||||
ctx, cancel := context.WithTimeout(ctx, shutdownTimeout)
|
||||
defer cancel()
|
||||
return errors.Join(tp.Shutdown(ctx), mp.Shutdown(ctx), lp.Shutdown(ctx))
|
||||
}
|
||||
return shutdown, nil
|
||||
}
|
||||
|
||||
// exportEnabled reports whether any standard OTLP endpoint is configured and
|
||||
// the SDK is not explicitly disabled.
|
||||
// rootSampler samples locally-rooted traces at MULGA_ROOT_TRACE_RATIO
|
||||
// (default 1.0) while always honoring an inbound sampled traceparent. Lets
|
||||
// chatty services shed root noise without losing any request-linked span.
|
||||
func rootSampler() sdktrace.Sampler {
|
||||
ratio := 1.0
|
||||
if v := os.Getenv("MULGA_ROOT_TRACE_RATIO"); v != "" {
|
||||
if f, err := strconv.ParseFloat(v, 64); err == nil && f >= 0 && f <= 1 {
|
||||
ratio = f
|
||||
} else {
|
||||
slog.Warn("invalid MULGA_ROOT_TRACE_RATIO, using 1.0", "value", v)
|
||||
}
|
||||
}
|
||||
return sdktrace.ParentBased(sdktrace.TraceIDRatioBased(ratio))
|
||||
}
|
||||
|
||||
func exportEnabled() bool {
|
||||
if strings.EqualFold(os.Getenv("OTEL_SDK_DISABLED"), "true") {
|
||||
return false
|
||||
}
|
||||
for _, key := range []string{
|
||||
"OTEL_EXPORTER_OTLP_ENDPOINT",
|
||||
"OTEL_EXPORTER_OTLP_TRACES_ENDPOINT",
|
||||
"OTEL_EXPORTER_OTLP_METRICS_ENDPOINT",
|
||||
} {
|
||||
if os.Getenv(key) != "" {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// newResource builds the service resource: identity attrs set here, plus
|
||||
// host detection and anything in OTEL_RESOURCE_ATTRIBUTES (ci.run_id etc.).
|
||||
func newResource(ctx context.Context, serviceName string) (*resource.Resource, error) {
|
||||
attrs := []attribute.KeyValue{semconv.ServiceName(serviceName)}
|
||||
if v := buildVersion(); v != "" {
|
||||
attrs = append(attrs, semconv.ServiceVersion(v))
|
||||
}
|
||||
env := os.Getenv("MULGA_ENV")
|
||||
if env == "" {
|
||||
env = os.Getenv("SPINIFEX_CI_ENV")
|
||||
}
|
||||
if env != "" {
|
||||
// deployment.environment maps to service.environment in Elastic APM,
|
||||
// enabling the native environment selector across the APM UI.
|
||||
attrs = append(attrs,
|
||||
attribute.String("mulga.env", env),
|
||||
semconv.DeploymentEnvironment(env))
|
||||
}
|
||||
if src := os.Getenv("MULGA_SOURCE"); src != "" {
|
||||
attrs = append(attrs, attribute.String("mulga.source", src))
|
||||
}
|
||||
|
||||
res, err := resource.New(ctx,
|
||||
resource.WithFromEnv(),
|
||||
resource.WithHost(),
|
||||
resource.WithTelemetrySDK(),
|
||||
resource.WithAttributes(attrs...),
|
||||
)
|
||||
// Schema-URL conflicts between detectors still yield a usable merged
|
||||
// resource; only a nil resource is fatal.
|
||||
if err != nil && res == nil {
|
||||
return nil, err
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
|
||||
// buildVersion returns the module version or embedded VCS revision, if any.
|
||||
func buildVersion() string {
|
||||
info, ok := debug.ReadBuildInfo()
|
||||
if !ok {
|
||||
return ""
|
||||
}
|
||||
if v := info.Main.Version; v != "" && v != "(devel)" {
|
||||
return v
|
||||
}
|
||||
for _, s := range info.Settings {
|
||||
if s.Key == "vcs.revision" && len(s.Value) >= 12 {
|
||||
return s.Value[:12]
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
@@ -1,403 +0,0 @@
|
||||
package otelsetup
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"go.opentelemetry.io/otel"
|
||||
sdklog "go.opentelemetry.io/otel/sdk/log"
|
||||
"go.opentelemetry.io/otel/trace"
|
||||
"go.opentelemetry.io/otel/trace/noop"
|
||||
|
||||
"go.opentelemetry.io/contrib/bridges/otelslog"
|
||||
loggerglobal "go.opentelemetry.io/otel/log/global"
|
||||
lognoop "go.opentelemetry.io/otel/log/noop"
|
||||
)
|
||||
|
||||
func TestInitWithoutEndpointIsNoop(t *testing.T) {
|
||||
t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "")
|
||||
t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "")
|
||||
t.Setenv("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT", "")
|
||||
// Pin a known no-op provider: the otel global delegator cannot be reset
|
||||
// once another test has installed a real provider.
|
||||
otel.SetTracerProvider(noop.NewTracerProvider())
|
||||
|
||||
shutdown, err := Init(context.Background(), "test-svc")
|
||||
if err != nil {
|
||||
t.Fatalf("Init: %v", err)
|
||||
}
|
||||
_, span := otel.Tracer("test").Start(context.Background(), "op")
|
||||
defer span.End()
|
||||
if span.SpanContext().IsValid() {
|
||||
t.Error("expected no-op tracer without endpoint, got recording span")
|
||||
}
|
||||
if err := shutdown(context.Background()); err != nil {
|
||||
t.Errorf("shutdown: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestInitWithEndpointInstallsProviders(t *testing.T) {
|
||||
// Point at a dead endpoint: exporters dial lazily, so Init must still
|
||||
// succeed and install real (recording) providers.
|
||||
t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://127.0.0.1:1")
|
||||
prevTracer := otel.GetTracerProvider()
|
||||
prevMeter := otel.GetMeterProvider()
|
||||
defer otel.SetTracerProvider(prevTracer)
|
||||
defer otel.SetMeterProvider(prevMeter)
|
||||
|
||||
shutdown, err := Init(context.Background(), "test-svc")
|
||||
if err != nil {
|
||||
t.Fatalf("Init: %v", err)
|
||||
}
|
||||
_, span := otel.Tracer("test").Start(context.Background(), "op")
|
||||
if !span.SpanContext().IsValid() {
|
||||
t.Error("expected recording tracer with endpoint set, got no-op span")
|
||||
}
|
||||
span.End()
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
|
||||
defer cancel()
|
||||
// Flush hits the dead endpoint; only assert it returns rather than hangs.
|
||||
_ = shutdown(ctx)
|
||||
}
|
||||
|
||||
func TestExportEnabled(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
env map[string]string
|
||||
want bool
|
||||
}{
|
||||
{"nothing set", nil, false},
|
||||
{"endpoint set", map[string]string{"OTEL_EXPORTER_OTLP_ENDPOINT": "http://localhost:4317"}, true},
|
||||
{"traces endpoint only", map[string]string{"OTEL_EXPORTER_OTLP_TRACES_ENDPOINT": "http://localhost:4317"}, true},
|
||||
{"disabled overrides endpoint", map[string]string{
|
||||
"OTEL_EXPORTER_OTLP_ENDPOINT": "http://localhost:4317",
|
||||
"OTEL_SDK_DISABLED": "true",
|
||||
}, false},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
for _, key := range []string{
|
||||
"OTEL_EXPORTER_OTLP_ENDPOINT",
|
||||
"OTEL_EXPORTER_OTLP_TRACES_ENDPOINT",
|
||||
"OTEL_EXPORTER_OTLP_METRICS_ENDPOINT",
|
||||
"OTEL_SDK_DISABLED",
|
||||
} {
|
||||
t.Setenv(key, tt.env[key])
|
||||
}
|
||||
if got := exportEnabled(); got != tt.want {
|
||||
t.Errorf("exportEnabled() = %v, want %v", got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewResourceAttributes(t *testing.T) {
|
||||
t.Setenv("MULGA_ENV", "env19")
|
||||
t.Setenv("MULGA_SOURCE", "ci")
|
||||
t.Setenv("OTEL_RESOURCE_ATTRIBUTES", "ci.run_id=12345")
|
||||
|
||||
res, err := newResource(context.Background(), "test-svc")
|
||||
if err != nil {
|
||||
t.Fatalf("newResource: %v", err)
|
||||
}
|
||||
got := map[string]string{}
|
||||
for _, kv := range res.Attributes() {
|
||||
got[string(kv.Key)] = kv.Value.String()
|
||||
}
|
||||
for key, want := range map[string]string{
|
||||
"service.name": "test-svc",
|
||||
"mulga.env": "env19",
|
||||
"mulga.source": "ci",
|
||||
"ci.run_id": "12345",
|
||||
} {
|
||||
if got[key] != want {
|
||||
t.Errorf("resource attr %s = %q, want %q", key, got[key], want)
|
||||
}
|
||||
}
|
||||
if got["host.name"] == "" {
|
||||
t.Error("resource attr host.name missing")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSlogHandlerStampsTraceIDs(t *testing.T) {
|
||||
var buf bytes.Buffer
|
||||
logger := slog.New(NewSlogHandler(slog.NewJSONHandler(&buf, nil)))
|
||||
|
||||
sc := trace.NewSpanContext(trace.SpanContextConfig{
|
||||
TraceID: trace.TraceID{0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f, 0x10},
|
||||
SpanID: trace.SpanID{0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f, 0x01, 0x02},
|
||||
})
|
||||
ctx := trace.ContextWithSpanContext(context.Background(), sc)
|
||||
|
||||
logger.InfoContext(ctx, "with span")
|
||||
|
||||
var line map[string]any
|
||||
if err := json.Unmarshal(buf.Bytes(), &line); err != nil {
|
||||
t.Fatalf("unmarshal log line: %v", err)
|
||||
}
|
||||
if line["trace_id"] != sc.TraceID().String() {
|
||||
t.Errorf("trace_id = %v, want %s", line["trace_id"], sc.TraceID())
|
||||
}
|
||||
if line["span_id"] != sc.SpanID().String() {
|
||||
t.Errorf("span_id = %v, want %s", line["span_id"], sc.SpanID())
|
||||
}
|
||||
}
|
||||
|
||||
func TestSlogHandlerNoSpanNoStamp(t *testing.T) {
|
||||
var buf bytes.Buffer
|
||||
logger := slog.New(NewSlogHandler(slog.NewJSONHandler(&buf, nil)))
|
||||
|
||||
logger.InfoContext(context.Background(), "no span")
|
||||
|
||||
var line map[string]any
|
||||
if err := json.Unmarshal(buf.Bytes(), &line); err != nil {
|
||||
t.Fatalf("unmarshal log line: %v", err)
|
||||
}
|
||||
if _, ok := line["trace_id"]; ok {
|
||||
t.Error("trace_id present on record without span")
|
||||
}
|
||||
if _, ok := line["span_id"]; ok {
|
||||
t.Error("span_id present on record without span")
|
||||
}
|
||||
}
|
||||
|
||||
// TestInitWithoutEndpointInstallsNoLoggerProvider is the prod-safety
|
||||
// guarantee: with no OTLP endpoint configured, Init must not install a
|
||||
// LoggerProvider (the global stays whatever was there before) and must not
|
||||
// touch the process default slog handler.
|
||||
func TestInitWithoutEndpointInstallsNoLoggerProvider(t *testing.T) {
|
||||
t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "")
|
||||
t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "")
|
||||
t.Setenv("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT", "")
|
||||
|
||||
prevLP := loggerglobal.GetLoggerProvider()
|
||||
defer loggerglobal.SetLoggerProvider(prevLP)
|
||||
sentinelLP := lognoop.NewLoggerProvider()
|
||||
loggerglobal.SetLoggerProvider(sentinelLP)
|
||||
|
||||
prevHandler := slog.Default().Handler()
|
||||
defer slog.SetDefault(slog.New(prevHandler))
|
||||
sentinelHandler := slog.NewJSONHandler(&bytes.Buffer{}, nil)
|
||||
slog.SetDefault(slog.New(sentinelHandler))
|
||||
|
||||
shutdown, err := Init(context.Background(), "test-svc")
|
||||
if err != nil {
|
||||
t.Fatalf("Init: %v", err)
|
||||
}
|
||||
defer func() { _ = shutdown(context.Background()) }()
|
||||
|
||||
if _, ok := loggerglobal.GetLoggerProvider().(lognoop.LoggerProvider); !ok {
|
||||
t.Errorf("expected global LoggerProvider to remain the noop sentinel, got %T", loggerglobal.GetLoggerProvider())
|
||||
}
|
||||
if got := slog.Default().Handler(); got != slog.Handler(sentinelHandler) {
|
||||
t.Errorf("expected default slog handler unchanged, got %T", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestInitWithEndpointInstallsLoggerProvider proves the same gate installs a
|
||||
// real LoggerProvider and fans the default logger out once an OTLP endpoint
|
||||
// is configured.
|
||||
func TestInitWithEndpointInstallsLoggerProvider(t *testing.T) {
|
||||
t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://127.0.0.1:1")
|
||||
prevLP := loggerglobal.GetLoggerProvider()
|
||||
defer loggerglobal.SetLoggerProvider(prevLP)
|
||||
prevTracer := otel.GetTracerProvider()
|
||||
prevMeter := otel.GetMeterProvider()
|
||||
defer otel.SetTracerProvider(prevTracer)
|
||||
defer otel.SetMeterProvider(prevMeter)
|
||||
prevHandler := slog.Default().Handler()
|
||||
defer slog.SetDefault(slog.New(prevHandler))
|
||||
|
||||
shutdown, err := Init(context.Background(), "test-svc")
|
||||
if err != nil {
|
||||
t.Fatalf("Init: %v", err)
|
||||
}
|
||||
|
||||
if _, ok := loggerglobal.GetLoggerProvider().(*sdklog.LoggerProvider); !ok {
|
||||
t.Errorf("expected a real sdk/log LoggerProvider, got %T", loggerglobal.GetLoggerProvider())
|
||||
}
|
||||
if _, ok := slog.Default().Handler().(*fanoutHandler); !ok {
|
||||
t.Errorf("expected default slog handler rewrapped as fanoutHandler, got %T", slog.Default().Handler())
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
|
||||
defer cancel()
|
||||
_ = shutdown(ctx)
|
||||
}
|
||||
|
||||
// recordingLogExporter is a minimal sdk/log.Exporter that records every
|
||||
// exported record for assertions, used since v0.20.0 ships no built-in
|
||||
// in-memory test exporter for external consumers.
|
||||
type recordingLogExporter struct {
|
||||
mu sync.Mutex
|
||||
records []sdklog.Record
|
||||
}
|
||||
|
||||
var _ sdklog.Exporter = (*recordingLogExporter)(nil)
|
||||
|
||||
func (e *recordingLogExporter) Export(_ context.Context, records []sdklog.Record) error {
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
for _, r := range records {
|
||||
e.records = append(e.records, r.Clone())
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *recordingLogExporter) Shutdown(context.Context) error { return nil }
|
||||
func (e *recordingLogExporter) ForceFlush(context.Context) error { return nil }
|
||||
|
||||
func (e *recordingLogExporter) snapshot() []sdklog.Record {
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
return append([]sdklog.Record(nil), e.records...)
|
||||
}
|
||||
|
||||
// TestLogRecordCarriesResourceAndTraceID exercises the exact pipeline Init
|
||||
// wires (resource -> LoggerProvider -> otelslog bridge) with a recording
|
||||
// exporter standing in for the OTLP gRPC exporter, proving exported records
|
||||
// carry the full resource (incl. ci.run_id from OTEL_RESOURCE_ATTRIBUTES) and
|
||||
// trace_id/span_id from the logging context.
|
||||
func TestLogRecordCarriesResourceAndTraceID(t *testing.T) {
|
||||
t.Setenv("OTEL_RESOURCE_ATTRIBUTES", "ci.run_id=TESTRUN")
|
||||
|
||||
res, err := newResource(context.Background(), "test-svc")
|
||||
if err != nil {
|
||||
t.Fatalf("newResource: %v", err)
|
||||
}
|
||||
|
||||
exp := &recordingLogExporter{}
|
||||
lp := sdklog.NewLoggerProvider(
|
||||
sdklog.WithResource(res),
|
||||
sdklog.WithProcessor(sdklog.NewSimpleProcessor(exp)),
|
||||
)
|
||||
defer func() { _ = lp.Shutdown(context.Background()) }()
|
||||
|
||||
bridge := otelslog.NewHandler("test-svc", otelslog.WithLoggerProvider(lp))
|
||||
logger := slog.New(bridge)
|
||||
|
||||
sc := trace.NewSpanContext(trace.SpanContextConfig{
|
||||
TraceID: trace.TraceID{0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f, 0x10},
|
||||
SpanID: trace.SpanID{0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f, 0x01, 0x02},
|
||||
TraceFlags: trace.FlagsSampled,
|
||||
})
|
||||
ctx := trace.ContextWithSpanContext(context.Background(), sc)
|
||||
|
||||
logger.InfoContext(ctx, "hello from bridge")
|
||||
|
||||
records := exp.snapshot()
|
||||
if len(records) != 1 {
|
||||
t.Fatalf("got %d exported records, want 1", len(records))
|
||||
}
|
||||
rec := records[0]
|
||||
|
||||
if rec.TraceID() != sc.TraceID() {
|
||||
t.Errorf("record TraceID = %s, want %s", rec.TraceID(), sc.TraceID())
|
||||
}
|
||||
if rec.SpanID() != sc.SpanID() {
|
||||
t.Errorf("record SpanID = %s, want %s", rec.SpanID(), sc.SpanID())
|
||||
}
|
||||
|
||||
attrs := map[string]string{}
|
||||
for _, kv := range rec.Resource().Attributes() {
|
||||
attrs[string(kv.Key)] = kv.Value.String()
|
||||
}
|
||||
if attrs["ci.run_id"] != "TESTRUN" {
|
||||
t.Errorf("resource ci.run_id = %q, want TESTRUN", attrs["ci.run_id"])
|
||||
}
|
||||
if attrs["service.name"] != "test-svc" {
|
||||
t.Errorf("resource service.name = %q, want test-svc", attrs["service.name"])
|
||||
}
|
||||
}
|
||||
|
||||
// recordingSlogHandler is a bare slog.Handler that records whether Handle
|
||||
// was called, standing in for the pre-existing stdout/journald handler in
|
||||
// fan-out tests.
|
||||
type recordingSlogHandler struct {
|
||||
mu sync.Mutex
|
||||
called int
|
||||
}
|
||||
|
||||
var _ slog.Handler = (*recordingSlogHandler)(nil)
|
||||
|
||||
func (h *recordingSlogHandler) Enabled(context.Context, slog.Level) bool { return true }
|
||||
|
||||
func (h *recordingSlogHandler) Handle(context.Context, slog.Record) error {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
h.called++
|
||||
return nil
|
||||
}
|
||||
|
||||
func (h *recordingSlogHandler) WithAttrs([]slog.Attr) slog.Handler { return h }
|
||||
func (h *recordingSlogHandler) WithGroup(string) slog.Handler { return h }
|
||||
|
||||
func (h *recordingSlogHandler) count() int {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
return h.called
|
||||
}
|
||||
|
||||
// TestFanoutHandlerWritesToAllHandlers proves the existing stdout/journald
|
||||
// handler still fires when the OTLP bridge handler is fanned in alongside it.
|
||||
func TestFanoutHandlerWritesToAllHandlers(t *testing.T) {
|
||||
stdout := &recordingSlogHandler{}
|
||||
bridge := &recordingSlogHandler{}
|
||||
logger := slog.New(newFanoutHandler(stdout, bridge))
|
||||
|
||||
logger.Info("dual write")
|
||||
|
||||
if stdout.count() != 1 {
|
||||
t.Errorf("stdout handler called %d times, want 1", stdout.count())
|
||||
}
|
||||
if bridge.count() != 1 {
|
||||
t.Errorf("bridge handler called %d times, want 1", bridge.count())
|
||||
}
|
||||
}
|
||||
|
||||
// TestAddFanoutHandlerPreservesExistingDefault proves addFanoutHandler
|
||||
// rewraps rather than replaces whatever default handler was already
|
||||
// installed (e.g. by SetDefaultJSONLogger before Init runs).
|
||||
func TestAddFanoutHandlerPreservesExistingDefault(t *testing.T) {
|
||||
prevHandler := slog.Default().Handler()
|
||||
defer slog.SetDefault(slog.New(prevHandler))
|
||||
|
||||
existing := &recordingSlogHandler{}
|
||||
slog.SetDefault(slog.New(existing))
|
||||
|
||||
bridge := &recordingSlogHandler{}
|
||||
addFanoutHandler(bridge)
|
||||
|
||||
slog.Info("after rewrap")
|
||||
|
||||
if existing.count() != 1 {
|
||||
t.Errorf("pre-existing handler called %d times, want 1", existing.count())
|
||||
}
|
||||
if bridge.count() != 1 {
|
||||
t.Errorf("bridge handler called %d times, want 1", bridge.count())
|
||||
}
|
||||
}
|
||||
|
||||
func TestSlogHandlerPreservesWrapperThroughWithAttrs(t *testing.T) {
|
||||
var buf bytes.Buffer
|
||||
logger := slog.New(NewSlogHandler(slog.NewJSONHandler(&buf, nil))).
|
||||
With("component", "test").WithGroup("grp")
|
||||
|
||||
sc := trace.NewSpanContext(trace.SpanContextConfig{
|
||||
TraceID: trace.TraceID{0xff, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f, 0x10},
|
||||
SpanID: trace.SpanID{0xfa, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f, 0x01, 0x02},
|
||||
})
|
||||
ctx := trace.ContextWithSpanContext(context.Background(), sc)
|
||||
|
||||
logger.InfoContext(ctx, "wrapped", "k", "v")
|
||||
|
||||
if !bytes.Contains(buf.Bytes(), []byte(sc.TraceID().String())) {
|
||||
t.Errorf("trace_id lost after With/WithGroup: %s", buf.String())
|
||||
}
|
||||
}
|
||||
@@ -1,126 +0,0 @@
|
||||
package otelsetup
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"os"
|
||||
|
||||
"go.opentelemetry.io/otel/trace"
|
||||
)
|
||||
|
||||
// defaultLevel backs the stdout handler's level. Held in a LevelVar so verbosity can
|
||||
// change without reinstalling the handler, which would discard the OTLP bridge Init
|
||||
// fans onto whatever default is installed at the time.
|
||||
var defaultLevel = new(slog.LevelVar)
|
||||
|
||||
// SetDefaultJSONLogger installs the process-wide slog default: JSON to
|
||||
// stdout (journald) at the given level, with trace_id/span_id stamping.
|
||||
// Call once, before Init; use SetLevel to change verbosity afterwards.
|
||||
func SetDefaultJSONLogger(level slog.Level) {
|
||||
defaultLevel.Set(level)
|
||||
slog.SetDefault(slog.New(NewSlogHandler(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
|
||||
Level: defaultLevel,
|
||||
}))))
|
||||
}
|
||||
|
||||
// SetLevel changes the stdout handler's level in place, leaving the handler chain
|
||||
// intact. Callers that only need to adjust verbosity must use this: calling
|
||||
// SetDefaultJSONLogger again replaces the default and silently unbolts OTLP export.
|
||||
func SetLevel(level slog.Level) {
|
||||
defaultLevel.Set(level)
|
||||
}
|
||||
|
||||
var _ slog.Handler = (*traceHandler)(nil)
|
||||
|
||||
// traceHandler stamps trace_id/span_id from the record's context onto every
|
||||
// log line so any log can be pivoted to its trace in the backend.
|
||||
type traceHandler struct {
|
||||
inner slog.Handler
|
||||
}
|
||||
|
||||
// NewSlogHandler wraps inner so records logged with a context carrying an
|
||||
// active span gain trace_id and span_id attributes. Records without a span
|
||||
// pass through unchanged.
|
||||
func NewSlogHandler(inner slog.Handler) slog.Handler {
|
||||
return &traceHandler{inner: inner}
|
||||
}
|
||||
|
||||
func (h *traceHandler) Enabled(ctx context.Context, level slog.Level) bool {
|
||||
return h.inner.Enabled(ctx, level)
|
||||
}
|
||||
|
||||
func (h *traceHandler) Handle(ctx context.Context, r slog.Record) error {
|
||||
if sc := trace.SpanContextFromContext(ctx); sc.IsValid() {
|
||||
r.AddAttrs(
|
||||
slog.String("trace_id", sc.TraceID().String()),
|
||||
slog.String("span_id", sc.SpanID().String()),
|
||||
)
|
||||
}
|
||||
return h.inner.Handle(ctx, r)
|
||||
}
|
||||
|
||||
func (h *traceHandler) WithAttrs(attrs []slog.Attr) slog.Handler {
|
||||
return &traceHandler{inner: h.inner.WithAttrs(attrs)}
|
||||
}
|
||||
|
||||
func (h *traceHandler) WithGroup(name string) slog.Handler {
|
||||
return &traceHandler{inner: h.inner.WithGroup(name)}
|
||||
}
|
||||
|
||||
var _ slog.Handler = (*fanoutHandler)(nil)
|
||||
|
||||
// fanoutHandler writes every record to all inner handlers, e.g. so OTLP
|
||||
// export is additive to the existing stdout/journald handler rather than
|
||||
// replacing it.
|
||||
type fanoutHandler struct {
|
||||
handlers []slog.Handler
|
||||
}
|
||||
|
||||
// newFanoutHandler returns a handler that fans out to all of handlers.
|
||||
func newFanoutHandler(handlers ...slog.Handler) slog.Handler {
|
||||
return &fanoutHandler{handlers: handlers}
|
||||
}
|
||||
|
||||
func (h *fanoutHandler) Enabled(ctx context.Context, level slog.Level) bool {
|
||||
for _, inner := range h.handlers {
|
||||
if inner.Enabled(ctx, level) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func (h *fanoutHandler) Handle(ctx context.Context, r slog.Record) error {
|
||||
var errs error
|
||||
for _, inner := range h.handlers {
|
||||
if inner.Enabled(ctx, r.Level) {
|
||||
errs = errors.Join(errs, inner.Handle(ctx, r.Clone()))
|
||||
}
|
||||
}
|
||||
return errs
|
||||
}
|
||||
|
||||
func (h *fanoutHandler) WithAttrs(attrs []slog.Attr) slog.Handler {
|
||||
next := make([]slog.Handler, len(h.handlers))
|
||||
for i, inner := range h.handlers {
|
||||
next[i] = inner.WithAttrs(attrs)
|
||||
}
|
||||
return &fanoutHandler{handlers: next}
|
||||
}
|
||||
|
||||
func (h *fanoutHandler) WithGroup(name string) slog.Handler {
|
||||
next := make([]slog.Handler, len(h.handlers))
|
||||
for i, inner := range h.handlers {
|
||||
next[i] = inner.WithGroup(name)
|
||||
}
|
||||
return &fanoutHandler{handlers: next}
|
||||
}
|
||||
|
||||
// addFanoutHandler rewraps the process slog default so records also reach
|
||||
// extra, in addition to whatever handler is already installed (e.g. the
|
||||
// stdout/journald handler from SetDefaultJSONLogger). Used by Init to bolt
|
||||
// the OTLP bridge onto an already-configured default logger.
|
||||
func addFanoutHandler(extra slog.Handler) {
|
||||
slog.SetDefault(slog.New(newFanoutHandler(slog.Default().Handler(), extra)))
|
||||
}
|
||||
@@ -1,66 +0,0 @@
|
||||
package otelsetup
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"log/slog"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// recordingHandler stands in for the OTLP bridge: it only needs to prove it still
|
||||
// receives records after a level change.
|
||||
type recordingHandler struct{ msgs *[]string }
|
||||
|
||||
func (recordingHandler) Enabled(context.Context, slog.Level) bool { return true }
|
||||
func (h recordingHandler) Handle(_ context.Context, r slog.Record) error {
|
||||
*h.msgs = append(*h.msgs, r.Message)
|
||||
return nil
|
||||
}
|
||||
func (h recordingHandler) WithAttrs([]slog.Attr) slog.Handler { return h }
|
||||
func (h recordingHandler) WithGroup(string) slog.Handler { return h }
|
||||
|
||||
// A gateway that reinstalled the default logger to change verbosity silently detached
|
||||
// the OTLP bridge, so only startup lines ever reached the sink.
|
||||
func TestSetLevelKeepsFanoutAttached(t *testing.T) {
|
||||
prev := slog.Default()
|
||||
t.Cleanup(func() { slog.SetDefault(prev) })
|
||||
|
||||
var got []string
|
||||
SetDefaultJSONLogger(slog.LevelInfo)
|
||||
addFanoutHandler(recordingHandler{msgs: &got})
|
||||
|
||||
SetLevel(slog.LevelDebug)
|
||||
slog.Info("after level change")
|
||||
|
||||
if len(got) != 1 || got[0] != "after level change" {
|
||||
t.Fatalf("fanout handler lost after SetLevel, got %v", got)
|
||||
}
|
||||
|
||||
// The hazard SetLevel exists to avoid: reinstalling the default drops the fanout,
|
||||
// so nothing logged afterwards reaches the bridge.
|
||||
SetDefaultJSONLogger(slog.LevelInfo)
|
||||
slog.Info("after reinstall")
|
||||
if len(got) != 1 {
|
||||
t.Fatalf("expected reinstall to detach the fanout, got %v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSetLevelChangesVerbosityInPlace(t *testing.T) {
|
||||
prev := slog.Default()
|
||||
t.Cleanup(func() { slog.SetDefault(prev) })
|
||||
|
||||
var buf bytes.Buffer
|
||||
slog.SetDefault(slog.New(slog.NewJSONHandler(&buf, &slog.HandlerOptions{Level: defaultLevel})))
|
||||
|
||||
SetLevel(slog.LevelError)
|
||||
slog.Info("suppressed")
|
||||
if buf.Len() != 0 {
|
||||
t.Fatalf("info logged at error level: %s", buf.String())
|
||||
}
|
||||
|
||||
SetLevel(slog.LevelDebug)
|
||||
slog.Debug("emitted")
|
||||
if !bytes.Contains(buf.Bytes(), []byte("emitted")) {
|
||||
t.Fatalf("debug not logged after level lowered: %s", buf.String())
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user