Something went wrong. Try again.
An ATProto Powered Reputation Network
Something went wrong. Try again.
4.3 kB · 153 lines
Go
at master
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154package telemetry
import ( "context" "errors" "io" "log/slog" "net/http" "net/http/httptest" "strings" "testing" "time"
"go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" sdktrace "go.opentelemetry.io/otel/sdk/trace" "go.opentelemetry.io/otel/trace" "go.opentelemetry.io/otel/trace/noop")
type stalledExporter struct { started chan struct{} release chan struct{}}
func (e *stalledExporter) ExportSpans(ctx context.Context, spans []sdktrace.ReadOnlySpan) error { select { case e.started <- struct{}{}: default: } select { case <-e.release: return errors.New("collector unavailable") case <-ctx.Done(): return ctx.Err() }}
func (e *stalledExporter) Shutdown(context.Context) error { return nil }
func TestQueueDoesNotBlock(t *testing.T) { exporter := &stalledExporter{started: make(chan struct{}, 1), release: make(chan struct{})} p := newProvider(exporter) defer func() { close(exporter.release) ctx, cancel := context.WithTimeout(context.Background(), flushTimeout) defer cancel() _ = p.Shutdown(ctx) }() tracer := p.Tracer("test") for range exportBatchSize { _, span := tracer.Start(context.Background(), "test") span.End() } select { case <-exporter.started: case <-time.After(flushTimeout): t.Fatal("export did not start") } done := make(chan struct{}) go func() { defer close(done) ctx, cancel := context.WithCancel(context.Background()) cancel() for range queueSize * 2 { _, span := tracer.Start(ctx, "test") span.End() } }() select { case <-done: case <-time.After(flushTimeout): t.Fatal("full export queue blocked the caller") } ctx, cancel := context.WithCancel(context.Background()) cancel() if err := p.Shutdown(ctx); !errors.Is(err, context.Canceled) { t.Fatalf("shutdown cancellation: %v", err) }}
func TestDisabledAndInvalid(t *testing.T) { previous := otel.GetTracerProvider() defer otel.SetTracerProvider(previous) otel.SetTracerProvider(noop.NewTracerProvider()) t.Setenv(endpointEnv, "") stop, err := Setup(context.Background()) if err != nil { t.Fatal(err) } stop() ctx, end := Start(context.Background(), "storage.open", slog.LevelDebug) end(nil) if trace.SpanFromContext(ctx).IsRecording() || len(LogAttrs(ctx)) != 0 { t.Fatal("disabled tracing created a trace") } for _, endpoint := range []string{"invalid", "http://user:secret@localhost/v1/traces", "http://localhost/?secret=x"} { t.Setenv(endpointEnv, endpoint) if _, err := Setup(context.Background()); err == nil || strings.Contains(err.Error(), "secret") { t.Fatalf("unsafe endpoint validation: %v", err) } }}
func TestHTTPExportAndPrivacy(t *testing.T) { previous := otel.GetTracerProvider() previousHandler := otel.GetErrorHandler() defer otel.SetTracerProvider(previous) defer otel.SetErrorHandler(previousHandler) requests := make(chan []byte, 1) server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Path != "/insert/opentelemetry/v1/traces" { t.Errorf("unexpected ingest path: %s", r.URL.Path) } body, err := io.ReadAll(r.Body) if err != nil { t.Error(err) } requests <- body w.Header().Set("Content-Type", "application/x-protobuf") w.WriteHeader(http.StatusOK) })) defer server.Close() t.Setenv(endpointEnv, server.URL+"/insert/opentelemetry/v1/traces") stop, err := Setup(context.Background()) if err != nil { t.Fatal(err) } ctx, end := Start(context.Background(), "replay.worker", slog.LevelDebug) if len(LogAttrs(ctx)) == 0 { t.Fatal("trace log correlation is absent") } batch, endBatch := Start(ctx, "replay.batch", slog.LevelDebug, attribute.Int("batch.size", 256)) if trace.SpanContextFromContext(batch).TraceID() == trace.SpanContextFromContext(ctx).TraceID() { t.Fatal("batch joins an unbounded worker trace") } endBatch(errors.New("credential-secret did:plc:private SELECT sensitive")) end(nil) stop() select { case body := <-requests: for _, forbidden := range []string{"credential-secret", "did:plc:private", "SELECT sensitive"} { if strings.Contains(string(body), forbidden) { t.Fatalf("trace contains %s", forbidden) } } if !strings.Contains(string(body), "replay.batch") { t.Fatal("batch span is absent") } case <-time.After(flushTimeout): t.Fatal("no OTLP request") }}