feat(audit): adding kafka audit exporter (#13357)

<!--A few plain bullets saying what changed and why, for a reviewer
skimming it - not a wall of text, not a restatement of the diff, not
generated boilerplate.-->
#### Description
- add `kafka` auditor provider that produces audit logs to a
kafka/redpanda topic as `otlp_proto`
- uses franz-go v1.21.7.
- no delivery timeout or retry cap: on franz-go < v1.22.2 these can ack
a record that was never written (twmb/franz-go#1484).
- added unit tests

<!--Reference issues using `Closes #issue-number` to enable automatic
closure on merge. -->
#### Issues closed by this PR
closes https://github.com/SigNoz/keystone-pod/issues/2

<!--Please delete paragraphs that you did not use before submitting.-->

cmd/enterprise/server.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/SigNoz/signoz/cmd"
 	"github.com/SigNoz/signoz/ee/auditor/fileauditor"
+	"github.com/SigNoz/signoz/ee/auditor/kafkaauditor"
 	"github.com/SigNoz/signoz/ee/auditor/otlphttpauditor"
 	"github.com/SigNoz/signoz/ee/authn/callbackauthn/oidccallbackauthn"
 	"github.com/SigNoz/signoz/ee/authn/callbackauthn/samlcallbackauthn"
@@ -156,6 +157,9 @@ func runServer(ctx context.Context, config signoz.Config, logger *slog.Logger) e
 			if err := factories.Add(fileauditor.NewFactory(licensing, version.Info)); err != nil {
 				panic(err)
 			}
+			if err := factories.Add(kafkaauditor.NewFactory(licensing, version.Info)); err != nil {
+				panic(err)
+			}
 			return factories
 		},
 		func(ctx context.Context, providerSettings factory.ProviderSettings, flagger pkgflagger.Flagger, licensing licensing.Licensing, telemetryStore telemetrystore.TelemetryStore, retentionGetter retention.Getter, orgGetter organization.Getter, zeus zeus.Zeus) (factory.NamedMap[factory.ProviderFactory[meterreporter.Reporter, meterreporter.Config]], string) {

conf/example.yaml

@@ -436,6 +436,7 @@ auditor:
   # Specifies the auditor provider to use.
   # noop: discards all audit events (community default).
   # otlphttp: exports audit events via OTLP HTTP (enterprise).
+  # kafka: exports audit events to a Kafka topic as OTLP protobuf (enterprise).
   provider: noop
   # The async channel capacity for audit events. Events are dropped when full (fail-open).
   buffer_size: 1000
@@ -461,6 +462,18 @@ auditor:
       max_interval: 30s
       # The total maximum time spent retrying.
       max_elapsed_time: 60s
+  kafka:
+    # The seed broker addresses (host:port) used to discover the cluster.
+    brokers:
+      - localhost:9092
+    # The topic audit events are written to. It must already exist.
+    topic: ""
+    # The client id the producer reports to the brokers.
+    client_id: signoz
+    # The compression codec for produced batches: none, gzip, snappy, lz4 or zstd.
+    compression: lz4
+    # The maximum time to drain buffered events on shutdown. Undelivered events are dropped after it.
+    shutdown_timeout: 30s
 
 ##################### Cloud Integration #####################
 cloudintegration:

ee/auditor/kafkaauditor/export.go

@@ -0,0 +1,31 @@
+package kafkaauditor
+
+import (
+	"context"
+	"log/slog"
+
+	"github.com/twmb/franz-go/pkg/kgo"
+
+	"github.com/SigNoz/signoz/pkg/auditor"
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/types/audittypes"
+)
+
+// Produces the whole batch as a single record so the collector's otlp_proto unmarshaler reads it as one ExportLogsServiceRequest.
+func (provider *provider) export(ctx context.Context, events []audittypes.AuditEvent) error {
+	logs := audittypes.NewPLogsFromAuditEvents(events, "signoz", provider.build.Version(), "signoz.audit")
+
+	value, err := provider.marshaler.MarshalLogs(logs)
+	if err != nil {
+		return errors.Wrapf(err, errors.TypeInternal, auditor.ErrCodeAuditExportFailed, "failed to marshal audit logs")
+	}
+
+	record := &kgo.Record{Topic: provider.config.Kafka.Topic, Value: value}
+	if err := provider.client.ProduceSync(ctx, record).FirstErr(); err != nil {
+		provider.settings.Logger().ErrorContext(ctx, "audit export failed", errors.Attr(err), slog.Int("dropped_log_records", len(events)))
+
+		return errors.Wrapf(err, errors.TypeInternal, auditor.ErrCodeAuditExportFailed, "failed to produce audit logs to topic %q", provider.config.Kafka.Topic)
+	}
+
+	return nil
+}

ee/auditor/kafkaauditor/provider.go

@@ -0,0 +1,132 @@
+package kafkaauditor
+
+import (
+	"context"
+	"log/slog"
+
+	"github.com/twmb/franz-go/pkg/kgo"
+	"github.com/twmb/franz-go/plugin/kslog"
+	"go.opentelemetry.io/collector/pdata/plog"
+
+	"github.com/SigNoz/signoz/pkg/auditor"
+	"github.com/SigNoz/signoz/pkg/auditor/auditorserver"
+	"github.com/SigNoz/signoz/pkg/factory"
+	"github.com/SigNoz/signoz/pkg/licensing"
+	"github.com/SigNoz/signoz/pkg/types/audittypes"
+	"github.com/SigNoz/signoz/pkg/version"
+)
+
+var _ auditor.Auditor = (*provider)(nil)
+
+type provider struct {
+	settings  factory.ScopedProviderSettings
+	config    auditor.Config
+	licensing licensing.Licensing
+	build     version.Build
+	server    *auditorserver.Server
+	marshaler plog.ProtoMarshaler
+	client    *kgo.Client
+}
+
+func NewFactory(licensing licensing.Licensing, build version.Build) factory.ProviderFactory[auditor.Auditor, auditor.Config] {
+	return factory.NewProviderFactory(factory.MustNewName("kafka"), func(ctx context.Context, providerSettings factory.ProviderSettings, config auditor.Config) (auditor.Auditor, error) {
+		return newProvider(ctx, providerSettings, config, licensing, build)
+	})
+}
+
+func newProvider(_ context.Context, providerSettings factory.ProviderSettings, config auditor.Config, licensing licensing.Licensing, build version.Build) (auditor.Auditor, error) {
+	settings := factory.NewScopedProviderSettings(providerSettings, "github.com/SigNoz/signoz/ee/auditor/kafkaauditor")
+
+	// Acks from all in-sync replicas, idempotent writes and unbounded record retries are franz-go defaults and are kept as is.
+	// A delivery timeout or retry limit lets franz-go < v1.22.2 ack a record that was never written (twmb/franz-go#1484).
+	client, err := kgo.NewClient(
+		kgo.SeedBrokers(config.Kafka.Brokers...),
+		kgo.ClientID(config.Kafka.ClientID),
+		kgo.ProducerBatchCompression(compressionCodec(config.Kafka.Compression)),
+		kgo.ProducerLinger(0),
+		kgo.WithLogger(kslog.New(settings.Logger())),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	provider := &provider{
+		settings:  settings,
+		config:    config,
+		licensing: licensing,
+		build:     build,
+		marshaler: plog.ProtoMarshaler{},
+		client:    client,
+	}
+
+	server, err := auditorserver.New(settings,
+		auditorserver.Config{
+			BufferSize:    config.BufferSize,
+			BatchSize:     config.BatchSize,
+			FlushInterval: config.FlushInterval,
+		},
+		provider.export,
+	)
+	if err != nil {
+		client.Close()
+		return nil, err
+	}
+
+	provider.server = server
+	return provider, nil
+}
+
+func (provider *provider) Start(ctx context.Context) error {
+	return provider.server.Start(ctx)
+}
+
+func (provider *provider) Audit(ctx context.Context, event audittypes.AuditEvent) {
+	if event.PrincipalAttributes.PrincipalOrgID.IsZero() {
+		provider.settings.Logger().WarnContext(ctx, "audit event dropped as org_id is zero")
+		return
+	}
+
+	if _, err := provider.licensing.GetActive(ctx, event.PrincipalAttributes.PrincipalOrgID); err != nil {
+		return
+	}
+
+	provider.server.Add(ctx, event)
+}
+
+func (provider *provider) Healthy() <-chan struct{} {
+	return provider.server.Healthy()
+}
+
+func (provider *provider) Stop(ctx context.Context) error {
+	ctx, cancel := context.WithTimeout(ctx, provider.config.Kafka.ShutdownTimeout)
+	defer cancel()
+
+	stoppedC := make(chan error, 1)
+	go func() { stoppedC <- provider.server.Stop(ctx) }()
+
+	select {
+	case err := <-stoppedC:
+		provider.client.Close()
+		return err
+	case <-ctx.Done():
+		provider.settings.Logger().WarnContext(ctx, "audit drain timed out, closing kafka client", slog.Duration("shutdown_timeout", provider.config.Kafka.ShutdownTimeout))
+		// Closing fails buffered and in-flight records, so the blocked export and the rest of the drain return immediately.
+		provider.client.Close()
+		return <-stoppedC
+	}
+}
+
+func compressionCodec(name string) kgo.CompressionCodec {
+	switch name {
+	case "gzip":
+		return kgo.GzipCompression()
+	case "snappy":
+		return kgo.SnappyCompression()
+	case "lz4":
+		return kgo.Lz4Compression()
+	case "zstd":
+		return kgo.ZstdCompression()
+	default:
+		return kgo.NoCompression()
+	}
+}

ee/auditor/kafkaauditor/provider_test.go

@@ -0,0 +1,227 @@
+package kafkaauditor
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/twmb/franz-go/pkg/kerr"
+	"github.com/twmb/franz-go/pkg/kfake"
+	"github.com/twmb/franz-go/pkg/kgo"
+	"go.opentelemetry.io/collector/pdata/plog"
+
+	"github.com/SigNoz/signoz/pkg/auditor"
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
+	"github.com/SigNoz/signoz/pkg/types/audittypes"
+	"github.com/SigNoz/signoz/pkg/types/coretypes"
+	"github.com/SigNoz/signoz/pkg/version"
+)
+
+func newTestCluster(t *testing.T, topics ...string) *kfake.Cluster {
+	t.Helper()
+
+	cluster, err := kfake.NewCluster(kfake.NumBrokers(1), kfake.SeedTopics(1, topics...))
+	require.NoError(t, err)
+	t.Cleanup(cluster.Close)
+
+	return cluster
+}
+
+func newTestProvider(t *testing.T, brokers []string, topic string, compression string) *provider {
+	t.Helper()
+
+	config := auditor.Config{
+		Provider:      "kafka",
+		BufferSize:    10,
+		BatchSize:     5,
+		FlushInterval: 50 * time.Millisecond,
+		Kafka: auditor.KafkaConfig{
+			Brokers:         brokers,
+			Topic:           topic,
+			ClientID:        "signoz-test",
+			Compression:     compression,
+			ShutdownTimeout: 30 * time.Second,
+		},
+	}
+
+	auditor, err := newProvider(context.Background(), instrumentationtest.New().ToProviderSettings(), config, nil, version.Info)
+	require.NoError(t, err)
+
+	return auditor.(*provider)
+}
+
+func newTestEvent(resource coretypes.Resource, action coretypes.Verb, body string) audittypes.AuditEvent {
+	return audittypes.AuditEvent{
+		Timestamp: time.Now(),
+		Body:      body,
+		EventName: audittypes.NewEventName(resource.Kind(), action),
+		AuditAttributes: audittypes.AuditAttributes{
+			Action:  action,
+			Outcome: audittypes.OutcomeSuccess,
+		},
+		ResourceAttributes: audittypes.ResourceAttributes{
+			Resource: resource,
+		},
+	}
+}
+
+// Polls the topic until the consumed records hold wantLogRecords audit log records in total.
+func consumeRecords(t *testing.T, brokers []string, topic string, wantLogRecords int) []*kgo.Record {
+	t.Helper()
+
+	client, err := kgo.NewClient(
+		kgo.SeedBrokers(brokers...),
+		kgo.ConsumeTopics(topic),
+		kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()),
+	)
+	require.NoError(t, err)
+	defer client.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	var records []*kgo.Record
+	gotLogRecords := 0
+	for gotLogRecords < wantLogRecords {
+		fetches := client.PollFetches(ctx)
+		require.NoError(t, ctx.Err(), "timed out waiting for %d log records, got %d", wantLogRecords, gotLogRecords)
+		require.Empty(t, fetches.Errors())
+
+		for _, record := range fetches.Records() {
+			logs, err := (&plog.ProtoUnmarshaler{}).UnmarshalLogs(record.Value)
+			require.NoError(t, err)
+			gotLogRecords += logs.LogRecordCount()
+			records = append(records, record)
+		}
+	}
+
+	return records
+}
+
+func TestExport(t *testing.T) {
+	testCases := []struct {
+		name            string
+		compression     string
+		compressionType uint8
+	}{
+		{name: "CompressionNone", compression: "none", compressionType: 0},
+		{name: "CompressionGzip", compression: "gzip", compressionType: 1},
+		{name: "CompressionSnappy", compression: "snappy", compressionType: 2},
+		{name: "CompressionLz4", compression: "lz4", compressionType: 3},
+		{name: "CompressionZstd", compression: "zstd", compressionType: 4},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			cluster := newTestCluster(t, "tenant_audit")
+			provider := newTestProvider(t, cluster.ListenAddrs(), "tenant_audit", testCase.compression)
+			defer provider.client.Close()
+
+			events := []audittypes.AuditEvent{
+				newTestEvent(coretypes.ResourceUser, coretypes.VerbUpdate, "user updated"),
+				newTestEvent(coretypes.ResourceMetaResourceDashboard, coretypes.VerbCreate, "dashboard created"),
+				newTestEvent(coretypes.ResourceMetaResourceDashboard, coretypes.VerbDelete, "dashboard deleted"),
+			}
+			require.NoError(t, provider.export(context.Background(), events))
+
+			records := consumeRecords(t, cluster.ListenAddrs(), "tenant_audit", len(events))
+			require.Len(t, records, 1)
+			assert.Equal(t, "tenant_audit", records[0].Topic)
+			assert.Equal(t, testCase.compressionType, records[0].Attrs.CompressionType())
+
+			logs, err := (&plog.ProtoUnmarshaler{}).UnmarshalLogs(records[0].Value)
+			require.NoError(t, err)
+			assert.Equal(t, len(events), logs.LogRecordCount())
+
+			serviceName, ok := logs.ResourceLogs().At(0).Resource().Attributes().Get("service.name")
+			require.True(t, ok)
+			assert.Equal(t, "signoz", serviceName.Str())
+		})
+	}
+}
+
+func TestExport_RecordTooLarge(t *testing.T) {
+	cluster := newTestCluster(t, "large_audit")
+	provider := newTestProvider(t, cluster.ListenAddrs(), "large_audit", "none")
+	defer provider.client.Close()
+
+	event := newTestEvent(coretypes.ResourceUser, coretypes.VerbCreate, strings.Repeat("a", 2<<20))
+
+	err := provider.export(context.Background(), []audittypes.AuditEvent{event})
+	require.Error(t, err)
+	assert.True(t, errors.Is(err, kerr.MessageTooLarge))
+}
+
+func TestExport_UnknownTopic(t *testing.T) {
+	cluster := newTestCluster(t)
+	provider := newTestProvider(t, cluster.ListenAddrs(), "missing_audit", "lz4")
+	defer provider.client.Close()
+
+	err := provider.export(context.Background(), []audittypes.AuditEvent{newTestEvent(coretypes.ResourceUser, coretypes.VerbCreate, "user created")})
+	require.Error(t, err)
+	assert.True(t, errors.Is(err, kerr.UnknownTopicOrPartition))
+}
+
+func TestExport_BrokerDown_RetriesUntilContextDone(t *testing.T) {
+	cluster := newTestCluster(t, "down_audit")
+	brokers := cluster.ListenAddrs()
+	cluster.Close()
+
+	provider := newTestProvider(t, brokers, "down_audit", "lz4")
+	defer provider.client.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+	defer cancel()
+
+	start := time.Now()
+	err := provider.export(ctx, []audittypes.AuditEvent{newTestEvent(coretypes.ResourceUser, coretypes.VerbCreate, "user created")})
+	require.Error(t, err)
+	assert.True(t, errors.Is(err, context.DeadlineExceeded))
+	assert.GreaterOrEqual(t, time.Since(start), 500*time.Millisecond)
+}
+
+func TestStop_DrainsQueuedEvents(t *testing.T) {
+	cluster := newTestCluster(t, "drain_audit")
+	provider := newTestProvider(t, cluster.ListenAddrs(), "drain_audit", "lz4")
+
+	done := make(chan error, 1)
+	go func() { done <- provider.Start(context.Background()) }()
+	<-provider.Healthy()
+
+	for range 7 {
+		provider.server.Add(context.Background(), newTestEvent(coretypes.ResourceUser, coretypes.VerbUpdate, "user updated"))
+	}
+
+	require.NoError(t, provider.Stop(context.Background()))
+	require.NoError(t, <-done)
+
+	records := consumeRecords(t, cluster.ListenAddrs(), "drain_audit", 7)
+	assert.NotEmpty(t, records)
+}
+
+func TestStop_BrokerDown_ClosesAfterShutdownTimeout(t *testing.T) {
+	cluster := newTestCluster(t, "stuck_audit")
+	brokers := cluster.ListenAddrs()
+	cluster.Close()
+
+	provider := newTestProvider(t, brokers, "stuck_audit", "lz4")
+	provider.config.Kafka.ShutdownTimeout = 500 * time.Millisecond
+
+	done := make(chan error, 1)
+	go func() { done <- provider.Start(context.Background()) }()
+	<-provider.Healthy()
+
+	for range 7 {
+		provider.server.Add(context.Background(), newTestEvent(coretypes.ResourceUser, coretypes.VerbUpdate, "user updated"))
+	}
+
+	start := time.Now()
+	require.NoError(t, provider.Stop(context.Background()))
+	require.NoError(t, <-done)
+	assert.GreaterOrEqual(t, time.Since(start), 500*time.Millisecond)
+	assert.Less(t, time.Since(start), 5*time.Second)
+}

go.mod

@@ -61,6 +61,9 @@ require (
 	github.com/swaggest/jsonschema-go v0.3.78
 	github.com/swaggest/rest v0.2.75
 	github.com/tidwall/gjson v1.18.0
+	github.com/twmb/franz-go v1.21.7
+	github.com/twmb/franz-go/pkg/kfake v0.0.0-20260421215025-4e7a1e1569ac
+	github.com/twmb/franz-go/plugin/kslog v1.0.0
 	github.com/uptrace/bun v1.2.9
 	github.com/uptrace/bun/dialect/pgdialect v1.2.9
 	github.com/uptrace/bun/dialect/sqlitedialect v1.2.9
@@ -144,6 +147,7 @@ require (
 	github.com/swaggest/refl v1.4.0 // indirect
 	github.com/swaggest/usecase v1.3.1 // indirect
 	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
+	github.com/twmb/franz-go/pkg/kmsg v1.13.1 // indirect
 	github.com/ugorji/go/codec v1.3.0 // indirect
 	github.com/uptrace/opentelemetry-go-extra/otelsql v0.3.2 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
@@ -243,7 +247,7 @@ require (
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/jpillora/backoff v1.0.0 // indirect
 	github.com/jtolds/gls v4.20.0+incompatible // indirect
-	github.com/klauspost/compress v1.18.5 // indirect
+	github.com/klauspost/compress v1.19.2 // indirect
 	github.com/klauspost/cpuid/v2 v2.3.0 // indirect
 	github.com/lann/builder v0.0.0-20180802200727-47ae307949d0 // indirect
 	github.com/lann/ps v0.0.0-20150810152359-62de8c46ede0 // indirect
@@ -272,7 +276,7 @@ require (
 	github.com/openfga/openfga v1.14.1
 	github.com/paulmach/orb v0.12.0 // indirect
 	github.com/pelletier/go-toml/v2 v2.2.4 // indirect
-	github.com/pierrec/lz4/v4 v4.1.25 // indirect
+	github.com/pierrec/lz4/v4 v4.1.26 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect
 	github.com/pressly/goose/v3 v3.27.0 // indirect

go.sum

@@ -675,8 +675,8 @@ github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/kisielk/sqlstruct v0.0.0-20201105191214-5f3e10d3ab46/go.mod h1:yyMNCyc/Ib3bDTKd379tNMpB/7/H5TjM2Y9QJ5THLbE=
 github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
-github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE=
-github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
+github.com/klauspost/compress v1.19.2 h1:hMRETovs/pu/dVWN7zIT1PGG8t509MwT6bO7XSi26R8=
+github.com/klauspost/compress v1.19.2/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
 github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y=
 github.com/klauspost/cpuid/v2 v2.3.0/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0=
 github.com/knadh/koanf v1.5.0 h1:q2TSd/3Pyc/5yP9ldIrSdIz26MCcyNQzW0pEAugLPNs=
@@ -853,8 +853,8 @@ github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8
 github.com/perses/spec v0.1.2 h1:yGoygcR3ZusuGDCmRMwsVXCMvMwi1qZndKV6NYNreEw=
 github.com/perses/spec v0.1.2/go.mod h1:NoGI5jmGwRdkdPgyYSZJTBL4/Py+dqIPKS2QV8NOvGE=
 github.com/pierrec/lz4 v2.0.5+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
-github.com/pierrec/lz4/v4 v4.1.25 h1:kocOqRffaIbU5djlIBr7Wh+cx82C0vtFb0fOurZHqD0=
-github.com/pierrec/lz4/v4 v4.1.25/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4=
+github.com/pierrec/lz4/v4 v4.1.26 h1:GrpZw1gZttORinvzBdXPUXATeqlJjqUG/D87TKMnhjY=
+github.com/pierrec/lz4/v4 v4.1.26/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4=
 github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c h1:+mdjkGKdHQG3305AYmdv1U2eRNDiU2ErMBj1gwrq8eQ=
 github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c/go.mod h1:7rwL4CYBLnjLxUqIJNnCWiEdr3bn6IUYi15bNlnbCCU=
 github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=
@@ -1055,6 +1055,16 @@ github.com/tmthrgd/go-hex v0.0.0-20190904060850-447a3041c3bc/go.mod h1:bciPuU6GH
 github.com/tv42/httpunix v0.0.0-20150427012821-b75d8614f926/go.mod h1:9ESjWnEqriFuLhtthL60Sar/7RFoluCcXsuvEwTV5KM=
 github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI=
 github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08=
+github.com/twmb/franz-go v1.21.7 h1:/DkA/o8wQN55gZWtpj2QNb9SIdxwFR7M+NecQWMdmc0=
+github.com/twmb/franz-go v1.21.7/go.mod h1:89kLt1uhE1GkyossLHGdpAMFNK9mV8GYk1lfWu9FiNs=
+github.com/twmb/franz-go/pkg/kadm v1.18.0 h1:WRf/LZmDdcDXwX7WMbtDU++v+b3NzYh2bCGoPMmzirw=
+github.com/twmb/franz-go/pkg/kadm v1.18.0/go.mod h1:XeLhGoLXLFzK8/ryv5FfpxPxGwj4oFEGpPJMB/x6KDE=
+github.com/twmb/franz-go/pkg/kfake v0.0.0-20260421215025-4e7a1e1569ac h1:6Dn8at7J4+5A8H7NgVLcPj4dfVm4pRtUZMHxXmBKxJ4=
+github.com/twmb/franz-go/pkg/kfake v0.0.0-20260421215025-4e7a1e1569ac/go.mod h1:6ofUZL6jAhajJ5WAaJTX9rK3B1ozGDOwKyZczdGK8uE=
+github.com/twmb/franz-go/pkg/kmsg v1.13.1 h1:fG5kItwysTk5UXqVwb64EpQEy3TydF3vYYK21nUQ+bI=
+github.com/twmb/franz-go/pkg/kmsg v1.13.1/go.mod h1:+DPt4NC8RmI6hqb8G09+3giKObE6uD2Eya6CfqBpeJY=
+github.com/twmb/franz-go/plugin/kslog v1.0.0 h1:I64oEmF+0PDvmyLgwrlOtg4mfpSE9GwlcLxM4af2t60=
+github.com/twmb/franz-go/plugin/kslog v1.0.0/go.mod h1:8pMjK3OJJJNNYddBSbnXZkIK5dCKFIk9GcVVCDgvnQc=
 github.com/twmb/murmur3 v1.1.8 h1:8Yt9taO/WN3l08xErzjeschgZU2QSrwm1kclYq+0aRg=
 github.com/twmb/murmur3 v1.1.8/go.mod h1:Qq/R7NUyOfr65zD+6Q5IHKsJLwP7exErjN6lyyq3OSQ=
 github.com/ua-parser/uap-go v0.0.0-20240611065828-3a4781585db6 h1:SIKIoA4e/5Y9ZOl0DCe3eVMLPOQzJxgZpfdHHeauNTM=

pkg/auditor/config.go

@@ -27,6 +27,8 @@ type Config struct {
 	OTLPHTTP OTLPHTTPConfig `mapstructure:"otlphttp"`
 
 	File FileConfig `mapstructure:"file"`
+
+	Kafka KafkaConfig `mapstructure:"kafka"`
 }
 
 // OTLPHTTPConfig holds configuration for the OTLP HTTP exporter provider.
@@ -54,6 +56,23 @@ type FileConfig struct {
 	Path string `mapstructure:"path"`
 }
 
+type KafkaConfig struct {
+	// Brokers are the seed broker addresses (host:port) used to discover the cluster.
+	Brokers []string `mapstructure:"brokers"`
+
+	// Topic must already exist; the producer does not create it.
+	Topic string `mapstructure:"topic"`
+
+	// ClientID identifies this producer in broker logs and metrics.
+	ClientID string `mapstructure:"client_id"`
+
+	// Compression is the codec for produced batches: none, gzip, snappy, lz4 or zstd.
+	Compression string `mapstructure:"compression"`
+
+	// ShutdownTimeout bounds draining buffered events on stop; events still undelivered when it expires are dropped.
+	ShutdownTimeout time.Duration `mapstructure:"shutdown_timeout"`
+}
+
 // RetryConfig configures exponential backoff for the OTLP HTTP exporter.
 type RetryConfig struct {
 	// Enabled controls whether retries are attempted on transient failures.
@@ -89,6 +108,12 @@ func newConfig() factory.Config {
 				MaxElapsedTime:  time.Minute,
 			},
 		},
+		Kafka: KafkaConfig{
+			Brokers:         []string{"localhost:9092"},
+			ClientID:        "signoz",
+			Compression:     "lz4",
+			ShutdownTimeout: 30 * time.Second,
+		},
 	}
 }
 
@@ -125,5 +150,25 @@ func (c Config) Validate() error {
 		}
 	}
 
+	if c.Provider == "kafka" {
+		if len(c.Kafka.Brokers) == 0 {
+			return errors.New(errors.TypeInvalidInput, errors.CodeInvalidInput, "auditor::kafka::brokers must be set when provider is kafka")
+		}
+
+		if c.Kafka.Topic == "" {
+			return errors.New(errors.TypeInvalidInput, errors.CodeInvalidInput, "auditor::kafka::topic must be set when provider is kafka")
+		}
+
+		switch c.Kafka.Compression {
+		case "none", "gzip", "snappy", "lz4", "zstd":
+		default:
+			return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "auditor::kafka::compression must be one of none, gzip, snappy, lz4 or zstd, got %q", c.Kafka.Compression)
+		}
+
+		if c.Kafka.ShutdownTimeout <= 0 {
+			return errors.New(errors.TypeInvalidInput, errors.CodeInvalidInput, "auditor::kafka::shutdown_timeout must be greater than 0")
+		}
+	}
+
 	return nil
 }

tests/integration/tests/logspipelines/01_logs_pipelines.py

@@ -12,6 +12,16 @@ def test_create_logs_pipeline_success(
     create_user_admin: types.Operation,  # pylint: disable=unused-argument
     get_token: Callable[[str, str], str],
 ) -> None:
+    """
+    Setup:
+    Create a new log parsing pipeline with a simple regex parser.
+
+    Tests:
+    1. Create a pipeline with valid configuration
+    2. Verify the pipeline is created successfully
+    3. Verify the response contains version information
+    4. Verify the pipeline is returned in the response
+    """
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
     pipeline_payload = {
@@ -89,6 +99,15 @@ def test_list_logs_pipelines_success(
     create_user_admin: types.Operation,  # pylint: disable=unused-argument
     get_token: Callable[[str, str], str],
 ) -> None:
+    """
+    Setup:
+    Create a pipeline first, then list all pipelines.
+
+    Tests:
+    1. Create a pipeline
+    2. List all pipelines and verify the created pipeline is present
+    3. Verify the response structure
+    """
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
     # Create a pipeline first
@@ -169,6 +188,16 @@ def test_list_logs_pipelines_by_version(
     create_user_admin: types.Operation,  # pylint: disable=unused-argument
     get_token: Callable[[str, str], str],
 ) -> None:
+    """
+    Setup:
+    Create multiple pipeline versions and list by specific version.
+
+    Tests:
+    1. Create initial pipeline (version 1)
+    2. Create updated pipeline (version 2)
+    3. List pipelines by version 1 and verify it returns the original
+    4. List pipelines by version 2 and verify it returns the updated version
+    """
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
     # Create version 1
@@ -321,6 +350,14 @@ def test_preview_logs_pipelines_success(
     create_user_admin: types.Operation,  # pylint: disable=unused-argument
     get_token: Callable[[str, str], str],
 ) -> None:
+    """
+    Setup:
+    Preview a json_parser pipeline with one JSON log and one plain-text log.
+
+    Tests:
+    1. JSON body gets parsed into attributes
+    2. Non-JSON body passes through unchanged instead of being dropped
+    """
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
     empty_log_fields = {
@@ -417,6 +454,15 @@ def test_create_multiple_pipelines_success(
     create_user_admin: types.Operation,  # pylint: disable=unused-argument
     get_token: Callable[[str, str], str],
 ) -> None:
+    """
+    Setup:
+    Create multiple pipelines in a single request.
+
+    Tests:
+    1. Create multiple pipelines with different configurations
+    2. Verify all pipelines are created
+    3. Verify pipelines are ordered correctly
+    """
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
     multi_pipeline_payload = {
@@ -522,6 +568,16 @@ def test_delete_all_pipelines_success(
     create_user_admin: types.Operation,  # pylint: disable=unused-argument
     get_token: Callable[[str, str], str],
 ) -> None:
+    """
+    Setup:
+    Create a pipeline first, then delete all pipelines by sending empty array.
+
+    Tests:
+    1. Create a pipeline
+    2. Delete all pipelines by sending empty pipelines array
+    3. Verify pipelines are deleted
+    4. Verify new version is created
+    """
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
     # Create a pipeline first

tests/integration/tests/querier_json_body/04_unknown_key_collisions.py

@@ -40,8 +40,15 @@ def test_json_body_unknown_key_collision_groupby(
     insert_logs_with_unknown_keys: Callable[[list[Logs], set[str]], None],
 ) -> None:
     """
-    `field` is stripped from the attribute metadata, so every place it lives is
-    reached through the synthesized multiIf, attribute maps then body.
+    Setup:
+    6 logs carrying `field` in different physical places: 2 as a string attribute
+    ("sa"), 1 as a number attribute (10), 1 as a bool attribute (True), 2 in the body
+    JSON ("bs"). `field` is stripped from attribute metadata, so it is unknown
+    everywhere and takes the synthesize path.
+
+    Tests:
+    Grouping by the unknown `field` walks the whole multiIf (attribute maps then body)
+    and buckets every record by its value, no matter which place holds it.
     """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     logs = [
@@ -95,8 +102,14 @@ def test_json_body_unknown_key_collision_filter(
     expected_rids: set[str],
 ) -> None:
     """
-    Each record carries a unique `rid` resource, so the result is checked by record
-    identity, not by count.
+    Setup:
+    The attribute/body collision dataset, each record tagged with a unique `rid`
+    resource so the filter result is checked by exact record identity, not just count.
+
+    Tests:
+    Filtering the unknown `field` resolves attributes OR the body JSON path and selects
+    exactly the expected records: a typed operand hits its attribute map, a body value
+    is matched via the body branch, and `exists` finds all six records.
     """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     logs = [
@@ -131,8 +144,14 @@ def test_json_body_unknown_key_attribute_parity(
     insert_logs_with_unknown_keys: Callable[[list[Logs], set[str]], None],
 ) -> None:
     """
-    With use_json_body on, the synthesized multiIf carries a trailing body branch.
-    It must not change how the attribute maps resolve.
+    Setup:
+    4 logs carrying the same value under known_key (in metadata) and unknown_key
+    (stripped), string / number / bool across records — no body values this time.
+
+    Tests:
+    Even with use_json_body on (the synthesized multiIf carries a trailing body
+    branch), grouping by the unknown attribute key still matches grouping by the known
+    one: the extra body branch does not perturb attribute-map resolution.
     """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     specs = [("log-a", "alpha"), ("log-b", 42), ("log-c", 42), ("log-d", True)]
@@ -178,6 +197,16 @@ def test_json_body_unknown_key_aggregation_over_body(
     get_token: Callable[[str, str], str],
     insert_logs_with_unknown_keys: Callable[[list[Logs], set[str]], None],
 ) -> None:
+    """
+    Setup:
+    The same attribute/body collision dataset — `field` carried as a string / number /
+    bool attribute and in the body JSON, one place per record.
+
+    Tests:
+    Aggregating over the unknown `field` walks the whole multiIf including the body
+    branch: count() sees every record (all 6) and count_distinct() counts the distinct
+    stringified values across attribute maps AND the body ("sa", "10", "true", "bs").
+    """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     logs = [
         Logs(timestamp=now - timedelta(seconds=1), resources={"service.name": SERVICE}, attributes={"field": "sa"}, body="attr-str-0"),
@@ -221,8 +250,17 @@ def test_json_body_unknown_key_body_path_parity(
     prefix: str,
 ) -> None:
     """
-    known_field is registered in the body path metadata and unknown_field is not,
-    so one resolves through metadata and the other through the fallback body path.
+    Setup:
+    Logs whose body carries the same value under two paths — known_field and
+    unknown_field — but only known_field is exported to the body path metadata
+    (distributed_field_keys); unknown_field is left unregistered.
+
+    Tests:
+    Grouping, counting distinct values, and selecting by the registered body path
+    (known_field, metadata-driven) and by the unregistered one (unknown_field,
+    fallback body path) return identical results, for the bare key and the
+    body.-prefixed key — the body counterpart of the attribute parity in
+    13_unknown_key_collisions.py.
     """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     logs = []

tests/integration/tests/queriercommon/07_fields_keys_values.py

@@ -35,8 +35,13 @@ def test_fields_keys_by_context(
     absent: list[str],
 ) -> None:
     """
-    The span context also lists attributes, because `span.<attribute>` resolves
-    attributes in queries.
+    Setup:
+    Insert a log with a code.file attribute and a span with an http.method attribute and a host.name resource.
+
+    Tests:
+    1. Keys for a context list that context's intrinsic columns and the stored keys the context maps to,
+       each with its context; intrinsics of other contexts are not listed. The span context also keeps
+       listing attributes because `span.<attribute>` resolves attributes in queries.
     """
     now = datetime.now(tz=UTC)
     insert_logs([Logs(timestamp=now, attributes={"code.file": "/opt/integration.go"}, body="a log line")])
@@ -78,7 +83,8 @@ def test_fields_keys_by_data_type(
     absent: list[str],
 ) -> None:
     """
-    number, int64 and float64 are one family.
+    Tests:
+    1. A data type filter keeps the intrinsic columns of that type; number, int64 and float64 are one family.
     """
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
     response = requests.get(
@@ -111,6 +117,11 @@ def test_fields_keys_search_matches_intrinsics_case_insensitively(
     search_text: str,
     present: list[str],
 ) -> None:
+    """
+    Tests:
+    1. The search text matches intrinsic columns case-insensitively, as it does for stored keys,
+       with or without a context prefix.
+    """
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
     response = requests.get(
         signoz.self.host_configs["8080"].get("/api/v1/fields/keys"),
@@ -147,8 +158,14 @@ def test_fields_values_bool_fields(
     expected: list[bool],
 ) -> None:
     """
-    The tag rows of a bool attribute carry no value, so true and false come from
-    the type, not from the data.
+    Setup:
+    Insert a log with a bool attribute.
+
+    Tests:
+    1. Values for a bool field are true and false (narrowed by the search text): for the calculated span
+       field, for a stored bool attribute whose tag rows carry no value, and for a key the caller
+       declares bool.
+    2. A span scope selector (isRoot) only takes true.
     """
     insert_logs([Logs(timestamp=datetime.now(tz=UTC), attributes={"retry": True}, body="retrying")])
 
@@ -172,9 +189,16 @@ def test_fields_values_start_excludes_span_values_not_seen_since_the_day(
     insert_traces: Callable[[list[Traces]], None],
 ) -> None:
     """
-    The start is floored to the day, the tag table's deduplication unit.
-    Logs are not covered: the logs collector stamps tag rows with the ingestion
-    hour, and the fixture writes the log's timestamp.
+    Setup:
+    Insert a span three days old and a span now, with different service names.
+
+    Tests:
+    1. Values with startUnixMilli an hour ago contain only the service seen today: the start is
+       floored to the day, the tag table's deduplication unit.
+    2. Values without a start contain both services.
+
+    Logs are not covered: the logs collector stamps tag rows with the ingestion hour, not the
+    log's timestamp, and the fixture writes the log's timestamp.
     """
     signal = "traces"
     now = datetime.now(tz=UTC)

tests/integration/tests/querierlogs/01_list.py

@@ -20,6 +20,19 @@ def test_logs_list(
     get_token: Callable[[str, str], str],
     insert_logs: Callable[[list[Logs]], None],
 ) -> None:
+    """
+    Setup:
+    Insert 2 logs with different attributes
+
+    Tests:
+    1. Query logs for the last 10 seconds and check if the logs are returned in the correct order
+    2. Query values of severity_text attribute from the autocomplete API
+    3. Query values of severity_text attribute from the fields API
+    4. Query values of code.file attribute from the autocomplete API
+    5. Query values of code.file attribute from the fields API
+    6. Query values of code.line attribute from the autocomplete API
+    7. Query values of code.line attribute from the fields API
+    """
     insert_logs(
         [
             Logs(
@@ -397,6 +410,10 @@ def test_logs_list_with_order_by(
     order_by_context: str,
     expected_order: list[str],
 ) -> None:
+    """
+    Setup:
+    Insert 3 logs with service.name in attributes and resources
+    """
 
     attribute_resource_pair = [
         [{"id": "log-001", "service.name": "c"}, {}],

tests/integration/tests/querierlogs/02_aggregation.py

@@ -19,6 +19,17 @@ def test_logs_time_series_count(
     get_token: Callable[[str, str], str],
     insert_logs: Callable[[list[Logs]], None],
 ) -> None:
+    """
+    Setup:
+    Insert 17 logs with service.name attribute set to "java" and severity_text attribute set to "DEBUG", 23 logs with service.name attribute set to "erlang" and severity_text attribute set to "ERROR", 29 logs with service.name attribute set to "go" and severity_text attribute set to "WARNING".
+    All logs have incrementing code.line attribute, modulo 2 for host.name and cloud.account.id.
+
+    Tests:
+    1. count() of all logs for the last 5 minutes
+    2. count() of all logs where code.line = 7 for last 5 minutes
+    3. count() of all logs where service.name = "erlang" OR cloud.account.id = "000" for last 5 minutes
+    4. count() of all logs grouped by host.name for the last 5 minutes
+    """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     logs: list[Logs] = []
 
@@ -430,6 +441,16 @@ def test_datatype_collision(
     get_token: Callable[[str, str], str],
     insert_logs: Callable[[list[Logs]], None],
 ) -> None:
+    """
+    Setup:
+    Insert logs with data type collision scenarios to test DataTypeCollisionHandledFieldName function
+
+    Tests:
+    1. severity_number comparison with string value
+    2. http.status_code with mixed string/number values
+    3. response.time with string values in numeric field
+    4. Edge cases: empty strings
+    """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     logs: list[Logs] = []
 

tests/integration/tests/querierlogs/06_json_body.py

@@ -18,6 +18,17 @@ def test_logs_json_body_simple_searches(
     get_token: Callable[[str, str], str],
     insert_logs: Callable[[list[Logs]], None],
 ) -> None:
+    """
+    Setup:
+    Insert logs with JSON bodies containing simple key-value pairs
+
+    Tests:
+    1. Search by body.message Contains "value"
+    2. Search by body.status = 200 (numeric)
+    3. Search by body.active = true (boolean)
+    4. Search by body.level = "error" with CONTAINS
+    5. Search by body.code > 100 (comparison)
+    """
     now = datetime.now(tz=UTC)
 
     # Log with simple JSON body
@@ -294,6 +305,16 @@ def test_logs_json_body_nested_keys(
     get_token: Callable[[str, str], str],
     insert_logs: Callable[[list[Logs]], None],
 ) -> None:
+    """
+    Setup:
+    Insert logs with JSON bodies containing nested objects
+
+    Tests:
+    1. Search by body.user.name = "value"
+    2. Search by body.request.secure = true (boolean)
+    3. Search by body.response.latency = 123.45 (floating point)
+    4. Search by body.response.status.code = 200
+    """
     now = datetime.now(tz=UTC)
 
     log1_body = json.dumps(
@@ -550,6 +571,17 @@ def test_logs_json_body_listing(
     get_token: Callable[[str, str], str],
     insert_logs: Callable[[list[Logs]], None],
 ) -> None:
+    """
+    Setup:
+    Insert multiple logs with JSON bodies
+
+    Tests:
+    1. List all logs with JSON bodies
+    2. List logs with pagination (limit and offset)
+    3. List logs ordered by timestamp
+    4. List logs with multiple filters combined (AND/OR)
+    5. Count logs matching JSON body filters
+    """
     now = datetime.now(tz=UTC)
 
     logs_data = [
@@ -903,9 +935,7 @@ def test_logs_json_body_in_operator(
     expression: str,
     expected_services: set[str],
 ) -> None:
-    """
-    IN over a body JSON path fans out to one comparison per value.
-    """
+    """IN over a body JSON path fans out to one comparison per value."""
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     specs = [("auth", 200, ["alpha", "beta"]), ("payment", 500, ["gamma"]), ("search", 404, ["beta", "alpha"])]
     insert_logs(

tests/integration/tests/querierlogs/07_list_expectations.py

@@ -427,6 +427,12 @@ def test_logs_list_query_timestamp_expectations(
     query: BuilderQuery,
     result: Callable[[list[Logs]], list[Any]],
 ) -> None:
+    """
+    Setup:
+    Insert logs with corrupt data
+
+    Tests:
+    """
     logs = generate_logs_with_corrupt_metadata()
     insert_logs(logs)
 
@@ -717,8 +723,15 @@ def test_logs_list_query_trace_id_expectations(
     results: Callable[[list[Logs]], list[Any]],
 ) -> None:
     """
-    The four logs differ by where trace_id is set: top level, attribute, resource,
-    or only in the body text. DESC ties break in storage order.
+    Justification for expected rows and ordering:
+    Four logs differ by where trace_id is set: top-level (x[0]), attribute (x[1]),
+    resource (x[2]), or only in body text (x[3]). Each parametrized case documents
+    which column AdjustKeys/field mapping reads and why DESC ties break in storage order.
+
+    Setup:
+    Insert logs with corrupt trace_id
+
+    Tests:
     """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
 

tests/integration/tests/querierlogs/10_filter_operators.py

@@ -66,6 +66,15 @@ def test_logs_filter_operators(
     expression: str,
     expected_bodies: set[str],
 ) -> None:
+    """
+    Setup:
+    Five logs across services / environments / methods, each body naming itself,
+    with numeric `resp_code` / `latency_ms` attributes and an `error.type`
+    attribute present on one.
+
+    Tests:
+    Each production-shaped filter operator selects exactly the expected logs.
+    """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     # (body, service, deployment.environment, http.request.method, resp_code, latency_ms, has_error_type)
     specs = [

tests/integration/tests/querierlogs/11_aggregation_functions.py

@@ -37,6 +37,17 @@ def test_logs_aggregate_functions(
     get_token: Callable[[str, str], str],
     insert_logs: Callable[[list[Logs]], None],
 ) -> None:
+    """
+    Setup:
+    svc-a: 3 logs (latency_ms 10/20/30, endpoints /x /x /y); svc-b: 1 log
+    (latency_ms 40, endpoint /z).
+
+    Tests:
+    A grouped scalar query computes count / sum / avg / min / max / p50 / p90 /
+    p95 / p99 over latency_ms, countIf over a numeric threshold, rate and
+    rate_sum over the query window, and count_distinct over a string attribute —
+    all matching values derived from the inserted logs, ordered by count() desc.
+    """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     # A scalar rate divides by the whole query window, so both sides agree on it.
     lookback_minutes = 5
@@ -123,6 +134,16 @@ def test_logs_aggregate_if_predicates_use_filter_syntax(
     get_token: Callable[[str, str], str],
     insert_logs: Callable[[list[Logs]], None],
 ) -> None:
+    """
+    Setup:
+    svc-a: 3 logs, two with a k8s.pod.name and an x-request-id header, two
+    with "timeout" in the body; svc-b: 1 log without either, with "timeout".
+
+    Tests:
+    countIf predicates accept the filter-only operators EXISTS, NOT EXISTS,
+    CONTAINS and NOT CONTAINS, a combination of them, and a key with hyphens;
+    two predicates in one expression each keep their own values.
+    """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     # (service, k8s.pod.name, x-request-id header, body)
     specs = [
@@ -188,6 +209,16 @@ def test_logs_aggregate_arguments_are_field_names(
     get_token: Callable[[str, str], str],
     insert_logs: Callable[[list[Logs]], None],
 ) -> None:
+    """
+    Setup:
+    svc-a: 3 logs, svc-b: 1 log, each with a body size under a key with a
+    hyphen and an app label under a key with slashes.
+
+    Tests:
+    Aggregation arguments are read as field names, so keys with hyphens,
+    slashes and a type suffix resolve, count(*) counts every log, and DISTINCT
+    keeps its meaning.
+    """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     # (service, http.response.body-size, k8s.pod.labels.app.kubernetes.io/name)
     specs = [
@@ -280,6 +311,15 @@ def test_logs_aggregate_having(
     aggregations: list[str],
     expected_services: set[str],
 ) -> None:
+    """
+    Setup:
+    svc-a: 3 logs (latency_ms 10/20/30, sum 60); svc-b: 1 log (latency_ms 40).
+
+    Tests:
+    A grouped scalar query with `having` keeps only the qualifying groups, for a
+    count() predicate (both directions) and a sum() predicate — extending logs
+    HAVING coverage beyond the single count() case in querierscalar/04_having.py.
+    """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     specs = [("svc-a", 10), ("svc-a", 20), ("svc-a", 30), ("svc-b", 40)]
     logs = [

tests/integration/tests/querierlogs/12_aggregation_timeseries.py

@@ -34,6 +34,14 @@ def test_logs_aggregate_min_max(
     get_token: Callable[[str, str], str],
     insert_logs: Callable[[list[Logs]], None],
 ) -> None:
+    """
+    Setup:
+    svc-a: 3 logs (latency_ms 10/20/30); svc-b: 1 log (latency_ms 40).
+
+    Tests:
+    max(latency_ms) / min(latency_ms) as two aggregations in one time_series query
+    grouped by service.name return each service's latency extremes.
+    """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     specs = [("svc-a", 10), ("svc-a", 20), ("svc-a", 30), ("svc-b", 40)]
     logs = [
@@ -96,6 +104,15 @@ def test_logs_aggregate_time_series_limit(
     limit: int,
     direction: str,
 ) -> None:
+    """
+    Setup:
+    Four services with distinct log counts (a=5, b=3, c=7, d=1).
+
+    Tests:
+    A time_series group-by with a limit returns only the N series that the ordered
+    aggregation keeps — top-N for desc, bottom-N for asc — each series summing to
+    that service's log count.
+    """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     counts = {"svc-a": 5, "svc-b": 3, "svc-c": 7, "svc-d": 1}
     logs = [
@@ -146,6 +163,15 @@ def test_logs_aggregate_multi_group_by(
     get_token: Callable[[str, str], str],
     insert_logs: Callable[[list[Logs]], None],
 ) -> None:
+    """
+    Setup:
+    Logs across (service.name, deployment.environment) pairs:
+    (svc-a, prod)=2, (svc-a, dev)=1, (svc-b, prod)=3.
+
+    Tests:
+    A scalar query grouped by two resource keys returns one row per present pair
+    (the cross product, not the full Cartesian) with the correct per-pair count.
+    """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     pair_counts = {("svc-a", "prod"): 2, ("svc-a", "dev"): 1, ("svc-b", "prod"): 3}
     logs = [

tests/integration/tests/querierlogs/13_unknown_key_collisions.py

@@ -62,8 +62,14 @@ def test_logs_unknown_key_collision_groupby_parity(
     insert_logs_with_unknown_keys: Callable[[list[Logs], set[str]], None],
 ) -> None:
     """
-    The shared value is a string, a number or a bool depending on the record, so
-    it spans all three attribute maps.
+    Setup:
+    8 logs sharing a value under known_key (in metadata) and unknown_key (stripped
+    from metadata); the value is a string / number / bool depending on the record, so
+    the shared value spans attributes_string / attributes_number / attributes_bool.
+
+    Tests:
+    Grouping by unknown_key (synthesized multiIf) yields the same buckets and counts
+    as grouping by known_key (metadata-driven), across all three attribute maps.
     """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     logs = [
@@ -124,6 +130,16 @@ def test_logs_unknown_key_collision_filter_parity(
     predicate: str,
     expected_bodies: set[str],
 ) -> None:
+    """
+    Setup:
+    The same collision dataset as the group-by parity test.
+
+    Tests:
+    Filtering unknown_key (synthesized) selects the same records as filtering known_key
+    (metadata-driven) across operators: typed equality (the operand type picks the
+    matching attribute map), `exists` (fans out across all three maps), and a
+    mixed-type `IN` list (string + number operands union their maps).
+    """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     logs = [
         Logs(
@@ -164,6 +180,15 @@ def test_logs_unknown_key_collision_aggregation_parity(
     get_token: Callable[[str, str], str],
     insert_logs_with_unknown_keys: Callable[[list[Logs], set[str]], None],
 ) -> None:
+    """
+    Setup:
+    The same collision dataset as the group-by parity test.
+
+    Tests:
+    Aggregating over unknown_key (synthesized) matches aggregating over known_key
+    (metadata-driven): count() sees every record (all 8 carry the key across the three
+    attribute maps) and count_distinct() sees the 6 distinct stringified values.
+    """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     logs = [
         Logs(

tests/integration/tests/querierlogs/14_select_fields.py

@@ -38,6 +38,14 @@ def test_logs_list_select_field_types(
     expected: Any,
     expected_type: type,
 ) -> None:
+    """
+    Setup:
+    One log carrying typed attributes (string / number / bool) and severity ERROR.
+
+    Tests:
+    Projecting a string / number / bool attribute or the severity_text intrinsic
+    returns the value with the right Python type under its short-name key.
+    """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     insert_logs(
         [
@@ -85,6 +93,15 @@ def test_logs_list_select_projects_columns(
     get_token: Callable[[str, str], str],
     insert_logs: Callable[[list[Logs]], None],
 ) -> None:
+    """
+    Setup:
+    One log with a string attribute plus body and resources.
+
+    Tests:
+    Projecting a single field returns it as a top-level key and drops the default
+    row payload — the resources_string / attributes_* maps and body are no longer
+    present when a selectFields list is given.
+    """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     insert_logs(
         [
@@ -136,9 +153,17 @@ def test_logs_list_unknown_log_context_synthesizes(
     surface: str,
 ) -> None:
     """
-    An unknown key under the `log.` context synthesizes against the attribute maps
-    and the body path instead of erroring: existence is a property of the data,
-    not of the metadata.
+    Setup:
+    One log.
+
+    Tests:
+    The intrinsic `log.` context is forgiving, like bare / `attribute.` /
+    `resource.` keys (and the traces `span.` context): an unknown key there
+    synthesizes against the log attribute maps (and the body path) rather than
+    erroring, because existence is a property of the data, not of metadata. A
+    filter reference matches nothing and surfaces a not-found warning (under the
+    stripped name); a select projects a null column; an order clause succeeds
+    and returns the log.
     """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     service = "log-forgiving-ctx-svc"

tests/integration/tests/querierscalar/06_metrics_aggregations.py

@@ -40,6 +40,14 @@ def test_metrics_scalar_space_aggregations(
     space_aggregation: str,
     expected: float,
 ) -> None:
+    """
+    Setup:
+    Four gauge series with latest values 50 / 30 / 70 / 10.
+
+    Tests:
+    An ungrouped scalar query combines the four series with the given space
+    aggregation: sum=160, avg=40, min=10, max=70, count=4.
+    """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     insert_metrics([Metrics(metric_name="test.metric", labels={"service.name": service}, timestamp=now - timedelta(seconds=1), temporality="Unspecified", type_="Gauge", is_monotonic=False, value=value) for service, value in METRIC_VALUES.items()])
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
@@ -73,6 +81,14 @@ def test_metrics_scalar_having(
     having_expression: str,
     expected_services: set[str],
 ) -> None:
+    """
+    Setup:
+    Four gauge series (per service.name) with latest values 50 / 30 / 70 / 10.
+
+    Tests:
+    A grouped metrics scalar query with `having` on the space-aggregated value keeps
+    only the qualifying services — HAVING was previously logs-only.
+    """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     insert_metrics([Metrics(metric_name="test.metric", labels={"service.name": service}, timestamp=now - timedelta(seconds=1), temporality="Unspecified", type_="Gauge", is_monotonic=False, value=value) for service, value in METRIC_VALUES.items()])
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)

tests/integration/tests/queriertraces/01_list.py

@@ -49,6 +49,20 @@ def test_traces_list(
     insert_traces: Callable[[list[Traces]], None],
     noise: str,
 ) -> None:
+    """
+    Setup:
+    Insert 4 spans across two services.
+    http-service: POST /integration -> SELECT, HTTP PATCH
+    topic-service: topic publish
+
+    Tests:
+    1. A raw list ordered by timestamp desc returns the spans in order with the
+       projected intrinsic / calculated / resource columns (field context and
+       datatype specified inline in the key name).
+    2. isRoot = 'true' returns only the two root spans.
+    Under the corrupt variant the projected columns still resolve to the real
+    intrinsic/calculated/resource values (never the colliding attributes).
+    """
     extra_attrs, extra_resources = trace_noise(noise)
     http_service_trace_id = TraceIdGenerator.trace_id()
     http_service_span_id = TraceIdGenerator.span_id()
@@ -213,8 +227,17 @@ def test_traces_list_select_fields(
     events_links: str,
 ) -> None:
     """
-    Events arrive parsed, and links carry only traceId and spanId: refType is
-    dropped at the consume layer.
+    Setup:
+    Insert a root span with no events/links and a child span carrying two
+    events and one user-supplied link.
+
+    Tests:
+    1. Empty selectFields returns all fields, and the `events` / `links` columns
+       arrive parsed into structured objects (events carry `attributes`, links
+       carry only `traceId`/`spanId` — refType is dropped at the consume layer).
+    2. A non-empty selectField returns that field along with timestamp, trace_id
+       and span_id.
+    The returned key set is invariant to the corrupt-variant collisions.
     """
     extra_attrs, extra_resources = trace_noise(noise)
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
@@ -335,6 +358,17 @@ def test_traces_list_select_field_types(
     expected: Callable[[Traces], Any],
     expected_type: type,
 ) -> None:
+    """
+    Setup:
+    Insert one span carrying typed attributes and db attributes.
+
+    Tests:
+    Projecting a string / number / bool span attribute, a calculated column, an
+    intrinsic column and its deprecated alias returns the value (derived from
+    the inserted span) with the right Python type; a bare non-existent attribute
+    projects as null. Under the corrupt variant the same-named colliding
+    attributes must not shadow the intrinsic/calculated columns.
+    """
     extra_attrs, extra_resources = trace_noise(noise)
     now = datetime.now(tz=UTC).replace(microsecond=0)
     span = Traces(
@@ -383,6 +417,13 @@ def test_traces_list_select_field_dedup(
     insert_traces: Callable[[list[Traces]], None],
     noise: str,
 ) -> None:
+    """
+    Setup:
+    Insert one span.
+
+    Tests:
+    Selecting the same field twice collapses to a single key in the response.
+    """
     extra_attrs, extra_resources = trace_noise(noise)
     now = datetime.now(tz=UTC).replace(microsecond=0)
     span = Traces(
@@ -437,8 +478,16 @@ def test_traces_list_select_calculated_field_collision(
     expected: Callable[[Traces], Any],
 ) -> None:
     """
-    A bare name or the `span.` context returns the derived column and shadows the
-    same-named attribute; the `attribute.` prefix reaches the attribute.
+    Setup:
+    Insert one span whose derived calculated columns (db_name from db.name,
+    http_method from http.request.method) collide with literal attributes of the
+    same name.
+
+    Tests:
+    Selecting a calculated field by its bare name or `span.` context returns the
+    derived column value (the same-named attribute is shadowed); the explicit
+    `attribute.` prefix reaches the colliding attribute value instead. Every form
+    projects under the calculated field's name.
     """
     now = datetime.now(tz=UTC).replace(microsecond=0)
     span = Traces(
@@ -479,6 +528,19 @@ def test_traces_list_response_shape_values(
     insert_traces: Callable[[list[Traces]], None],
     noise: str,
 ) -> None:
+    """
+    Setup:
+    Insert a curated set of spans exercising every derived/intrinsic response
+    field: an error span, a client span with HTTP url/method, a db span, a
+    host span, and remote/kind variants.
+
+    Tests:
+    With empty selectFields every intrinsic/calculated column (has_error,
+    is_remote, external_http_url, db_name, http_host, status_code_string,
+    kind_string, ...) round-trips with its populated value, and the merged
+    `attributes` map preserves string / number / bool types. Under the corrupt
+    variant these columns keep winning over the same-named colliding attributes.
+    """
     extra_attrs, extra_resources = trace_noise(noise)
     now = datetime.now(tz=UTC).replace(microsecond=0)
     trace_id = TraceIdGenerator.trace_id()
@@ -600,6 +662,17 @@ def test_traces_list_ordering(
     sort_value: Callable[[dict[str, Any]], Any],
     direction: str,
 ) -> None:
+    """
+    Setup:
+    Insert 4 spans whose fields are permuted so that each order key produces a
+    distinct ordering.
+
+    Tests:
+    Ordering by an intrinsic (timestamp, duration_nano, name, kind), a
+    calculated column (response_status_code), a resource attribute and a number
+    span attribute all sort correctly in both directions — including when
+    same-named corrupt attributes collide with those columns.
+    """
     extra_attrs, extra_resources = trace_noise(noise)
     now = datetime.now(tz=UTC).replace(microsecond=0)
     trace_id = TraceIdGenerator.trace_id()
@@ -660,6 +733,14 @@ def test_traces_list_order_multi_key_tiebreak(
     insert_traces: Callable[[list[Traces]], None],
     noise: str,
 ) -> None:
+    """
+    Setup:
+    Insert 3 spans sharing the same timestamp with distinct durations.
+
+    Tests:
+    A secondary order key (duration_nano) breaks the timestamp tie
+    deterministically, and flipping its direction flips the tie-break.
+    """
     extra_attrs, extra_resources = trace_noise(noise)
     now = datetime.now(tz=UTC).replace(microsecond=0)
     trace_id = TraceIdGenerator.trace_id()
@@ -714,7 +795,12 @@ def test_traces_list_no_default_order(
     noise: str,
 ) -> None:
     """
-    The builder adds no default ORDER BY, so only membership is pinned.
+    Setup:
+    Insert 4 spans.
+
+    Tests:
+    A raw list query with no order clause succeeds and returns every matching
+    span (the builder adds no default ORDER BY, so only membership is pinned).
     """
     extra_attrs, extra_resources = trace_noise(noise)
     now = datetime.now(tz=UTC).replace(microsecond=0)
@@ -776,6 +862,14 @@ def test_traces_list_pagination(
     limit: int | None,
     offset: int,
 ) -> None:
+    """
+    Setup:
+    Insert 5 spans with strictly increasing timestamps.
+
+    Tests:
+    Ordered by timestamp desc, limit caps the row count and offset pages
+    through the result; an offset past the end returns no rows.
+    """
     extra_attrs, extra_resources = trace_noise(noise)
     now = datetime.now(tz=UTC).replace(microsecond=0)
     trace_id = TraceIdGenerator.trace_id()
@@ -837,8 +931,13 @@ def test_traces_list_time_boundaries(
     noise: str,
 ) -> None:
     """
-    The spans are second-aligned, so the ms to ns conversion lands exactly on the
-    window edges.
+    Setup:
+    Insert spans exactly at the query start, in the middle, and exactly at the
+    query end (all second-aligned so the ms->ns conversion lands exactly).
+
+    Tests:
+    The query window is [start, end): the span at start is included and the
+    span at end is excluded.
     """
     extra_attrs, extra_resources = trace_noise(noise)
     start_dt = datetime.now(tz=UTC).replace(microsecond=0) - timedelta(seconds=30)
@@ -879,6 +978,14 @@ def test_traces_list_empty_result(
     insert_traces: Callable[[list[Traces]], None],
     noise: str,
 ) -> None:
+    """
+    Setup:
+    Insert one span.
+
+    Tests:
+    A filter matching no span, and a time window that excludes the span, both
+    return zero rows (rows is null, surfaced as an empty list).
+    """
     extra_attrs, extra_resources = trace_noise(noise)
     now = datetime.now(tz=UTC).replace(microsecond=0)
     insert_traces(
@@ -938,7 +1045,14 @@ def test_traces_list_span_scope(
     noise: str,
 ) -> None:
     """
-    isEntryPoint reads top_level_operations, so the entry-point span is seeded there.
+    Setup:
+    Insert a root span, a non-root entry-point span (its (name, service) is
+    seeded into top_level_operations) and a non-root internal span.
+
+    Tests:
+    1. isRoot = 'true' returns only the root span.
+    2. isEntryPoint = 'true' returns only the non-root span whose operation is
+       registered in top_level_operations.
     """
     extra_attrs, extra_resources = trace_noise(noise)
     now = datetime.now(tz=UTC).replace(microsecond=0)
@@ -1148,6 +1262,17 @@ def test_traces_list_with_corrupt_data(
     status_code: HTTPStatus,
     expected: Callable[[list[Traces]], dict[str, Any]],
 ) -> None:
+    """
+    Setup:
+    Insert 4 spans whose metadata carries intrinsic/calculated field names as
+    attributes (e.g. attribute.timestamp, attribute.duration_nano).
+
+    Tests:
+    Ordering / selecting a name that exists both as an intrinsic column and as
+    an attribute resolves to the intrinsic column, while the explicit
+    `attribute.` prefix reaches the attribute value. Each expected row is derived
+    from the inserted span objects.
+    """
     traces = generate_traces_with_corrupt_metadata()
     insert_traces(traces)
     # traces[i] occurred before traces[j] where i < j.
@@ -1221,10 +1346,24 @@ def test_traces_list_with_scope_filter(
     expected_indices: list[int],
 ) -> None:
     """
-    `scope.name` and `scope.version` resolve to the declared JSON sub-column and to
-    a same-named scope attribute; `scope.attribute.` addresses the attribute alone.
-    A bare `name` reaches the span name column and a `name` scope attribute, never
-    the declared scope.name.
+    Setup three spans with different scope key resolution:
+    - x[0]: scope.name/version 'io.signoz.checkout'/'2.3.1'; span attribute
+      env.tier='gold'.
+    - x[1]: scope.name/version 'io.signoz.payment'/'4.5.6'; scope attributes
+      telemetry.sdk.language='python', env.tier='gold', and a `name` scope
+      attribute colliding with x[0]'s scope.name value.
+    - x[2]: span name 'io.signoz.checkout' (colliding with x[0]'s scope.name
+      value) and a span attribute literally named `scope.name`.
+
+    Tests:
+    - Filtering on scope.name / scope.version / a scope attribute.
+    - An unprefixed key is resolved across contexts (scope checked alongside
+      attribute / intrinsic), while a `scope.`-prefixed key is scope-only.
+    - `scope.name`/`scope.version` name every home they resolve to: the declared JSON
+      sub-column and a same-named `name`/`version` scope attribute. The explicit
+      `scope.attribute.` prefix addresses the attribute alone.
+    - a bare `name` reaches the span `name` column and a `name` scope attribute, but
+      never the declared scope.name field.
     """
     now = datetime.now(tz=UTC).replace(microsecond=0)
     trace_id = TraceIdGenerator.trace_id()
@@ -1326,8 +1465,18 @@ def test_traces_list_unknown_span_context_synthesizes(
     surface: str,
 ) -> None:
     """
-    An unknown key under the `span.` context synthesizes against the attribute maps
-    instead of erroring: existence is a property of the data, not of the metadata.
+    Setup:
+    Insert one span.
+
+    Tests:
+    The intrinsic `span.` context is forgiving, like bare / `attribute.` /
+    `resource.` keys (and the logs `log.` context): an unknown key there
+    synthesizes against the span attribute maps rather than erroring, because
+    existence is a property of the data, not of metadata. A filter reference
+    matches nothing and surfaces a not-found warning (under the stripped name);
+    a select projects a null column; an order clause succeeds and returns the
+    span. (See test_traces_list_unknown_other_context_synthesizes for the
+    `attribute.`/`resource.` contexts.)
     """
     now = datetime.now(tz=UTC).replace(microsecond=0)
     span = Traces(
@@ -1386,6 +1535,17 @@ def test_traces_list_unknown_other_context_synthesizes(
     context: str,
     surface: str,
 ) -> None:
+    """
+    Setup:
+    Insert 3 spans.
+
+    Tests:
+    Like the intrinsic `span.` context (see
+    test_traces_list_unknown_span_context_synthesizes), the `attribute.`
+    and `resource.` contexts synthesize an unknown key (existence is a property
+    of the data, not metadata): a filter reference matches nothing, a select
+    projects a null column, and an order clause succeeds and returns every span.
+    """
     now = datetime.now(tz=UTC).replace(microsecond=0)
     trace_id = TraceIdGenerator.trace_id()
     spans = [
@@ -1441,6 +1601,15 @@ def test_traces_list_order_unknown_key_synthesizes(
     get_token: Callable[[str, str], str],
     insert_traces: Callable[[list[Traces]], None],
 ) -> None:
+    """
+    Setup:
+    Insert 3 spans.
+
+    Tests:
+    Ordering by a bare (context-less) unknown key synthesizes a null column and
+    succeeds, returning every matching span — the ordering counterpart of the
+    bare-select-null case and the unknown-filter synthesis in 09_unknown_keys.
+    """
     now = datetime.now(tz=UTC).replace(microsecond=0)
     trace_id = TraceIdGenerator.trace_id()
     spans = [

tests/integration/tests/queriertraces/02_aggregation.py

@@ -77,7 +77,14 @@ def test_traces_aggregate_order_by_count(
     expected_status: HTTPStatus,
 ) -> None:
     """
-    Only the forms that specify the context twice are a bad request.
+    Setup:
+    Insert 4 spans of a single service.
+
+    Tests:
+    An aggregation can be referenced in `order by` by its expression (count()),
+    its `span.`-prefixed expression, or its alias — with or without an explicit
+    fieldContext. Only the combinations that double-specify the context are a bad
+    request; the rest resolve and return the span count.
     """
     use_attribute_backend(attribute_backend)
 
@@ -142,6 +149,20 @@ def test_traces_aggregate_functions(
     attribute_backend: str,
     noise: str,
 ) -> None:
+    """
+    Setup:
+    svc-a: 3 spans (durations 1s/2s/3s; one ERROR; one 4XX via
+    http.response.status_code); svc-b: 1 span (duration 4s). A number attribute
+    `latency_ms` rides along on every span.
+
+    Tests:
+    A grouped scalar query computes count / sum / avg / min / max / p50 / p90 over
+    duration_nano, countIf over the intrinsic status_code and the calculated
+    response_status_code, rate and rate_sum over the query window, and avg over a
+    numeric attribute — all matching values derived from the inserted spans. Under
+    the corrupt variant the same-named colliding attributes must not change any of
+    these.
+    """
     use_attribute_backend(attribute_backend)
 
     extra_attrs, extra_resources = trace_noise(noise)
@@ -235,6 +256,14 @@ def test_traces_aggregate_calculated_range_countif(
     use_attribute_backend: Callable[[str], None],
     attribute_backend: str,
 ) -> None:
+    """
+    Setup:
+    3 spans whose derived response_status_code is 200 / 503 / 404.
+
+    Tests:
+    countIf(response_status_code >= 400 AND response_status_code < 500) counts the
+    single 4XX span (404).
+    """
     use_attribute_backend(attribute_backend)
 
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
@@ -282,6 +311,14 @@ def test_traces_aggregate_having(
     attribute_backend: str,
     noise: str,
 ) -> None:
+    """
+    Setup:
+    svc-a: 3 spans, svc-b: 1 span.
+
+    Tests:
+    A grouped scalar query with `having count() > 2` returns only the groups
+    whose count qualifies (svc-a), derived from the inserted spans.
+    """
     use_attribute_backend(attribute_backend)
 
     extra_attrs, extra_resources = trace_noise(noise)
@@ -348,6 +385,16 @@ def test_traces_aggregate_time_series_limit(
     limit: int,
     direction: str,
 ) -> None:
+    """
+    Setup:
+    Four services with distinct span counts (a=5, b=3, c=7, d=1).
+
+    Tests:
+    A time_series group-by with a limit returns only the N series that the
+    ordered aggregation keeps — top-N for desc, bottom-N for asc — each series
+    summing to that service's span count. Under the corrupt variant the grouping
+    (resource.service.name) and count are unaffected by the colliding attributes.
+    """
     use_attribute_backend(attribute_backend)
 
     extra_attrs, extra_resources = trace_noise(noise)
@@ -421,6 +468,14 @@ def test_traces_aggregate_unknown_key(
     expression: str,
     expected_value: Any,
 ) -> None:
+    """
+    Setup:
+    Insert 3 spans.
+
+    Tests:
+    Aggregating a bare unknown key synthesizes a null column: count() of it is 0
+    (no non-null values) and sum()/avg() are null — the query never errors.
+    """
     use_attribute_backend(attribute_backend)
 
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)

tests/integration/tests/queriertraces/03_fill.py

@@ -39,6 +39,14 @@ def test_traces_fill_plain(
     noise: str,
     fill_mode: str,
 ) -> None:
+    """
+    Setup:
+    One span 3 minutes ago.
+
+    Tests:
+    fillGaps / fillZero over count() with no group by leave the single populated
+    bucket at the span count and every other bucket in the window at 0.
+    """
     extra_attrs, extra_resources = trace_noise(noise)
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     span = Traces(
@@ -85,6 +93,14 @@ def test_traces_fill_group_by(
     noise: str,
     fill_mode: str,
 ) -> None:
+    """
+    Setup:
+    service-a span 3 minutes ago, service-b span 2 minutes ago.
+
+    Tests:
+    fillGaps / fillZero over count() grouped by service.name yield one series per
+    service, each with its single populated bucket and zeros elsewhere.
+    """
     extra_attrs, extra_resources = trace_noise(noise)
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     span_a = Traces(
@@ -151,6 +167,14 @@ def test_traces_fill_formula(
     noise: str,
     fill_mode: str,
 ) -> None:
+    """
+    Setup:
+    'test' service span 3 minutes ago, 'another-test' service span 2 minutes ago.
+
+    Tests:
+    A formula F1 = A + B over two disabled per-service count() queries fills empty
+    buckets to 0, so F1 carries a 1 in each service's populated bucket.
+    """
     extra_attrs, extra_resources = trace_noise(noise)
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     span_test = Traces(
@@ -213,6 +237,15 @@ def test_traces_fill_formula_group_by(
     noise: str,
     fill_mode: str,
 ) -> None:
+    """
+    Setup:
+    group1 span 3 minutes ago, group2 span 2 minutes ago.
+
+    Tests:
+    A formula F1 = A + B over two disabled group-by count() queries yields one
+    series per service, each carrying A+B (=2) in its populated bucket and zeros
+    elsewhere.
+    """
     extra_attrs, extra_resources = trace_noise(noise)
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     span_g1 = Traces(

tests/integration/tests/queriertraces/04_trace_id_filter.py

@@ -34,8 +34,15 @@ def test_traces_list_filter_by_trace_id(
     noise: str,
 ) -> None:
     """
-    A window wider than an hour spans several exponential buckets, so the span must
-    come back exactly once.
+    Setup:
+    A target trace (one root span) and an unrelated trace.
+
+    Tests filtering a raw list by trace_id:
+    1. A narrow window (single bucket) returns just the target span.
+    2. A wide window (>1 h, multiple exponential buckets) returns it exactly once
+       (no duplicate-span regression).
+    3. A window that does not contain the trace returns nothing and warns that it
+       lies outside the selected time range.
     """
     extra_attrs, extra_resources = trace_noise(noise)
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
@@ -123,8 +130,14 @@ def test_traces_aggregation_filter_by_trace_id(
     noise: str,
 ) -> None:
     """
-    trace_summary is authoritative: a trace_id with no row there short-circuits to
-    empty without the outside-range warning.
+    Setup:
+    A target trace with two spans (root + child).
+
+    Tests the trace_id time-range optimization for aggregation (time_series):
+    1. A wide window containing the trace returns the correct count (both spans).
+    2. A window outside the trace short-circuits to empty and warns.
+    3. A trace_id with no row in trace_summary short-circuits to empty with no
+       outside-range warning (trace_summary is authoritative).
     """
     extra_attrs, extra_resources = trace_noise(noise)
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)

tests/integration/tests/queriertraces/06_trace_operator.py

@@ -587,9 +587,10 @@ def test_trace_operator_with_adjusted_keys(
     assert_result: Callable[[requests.Response, list[Traces]], None],
 ) -> None:
     """
-    A trace operator builds a CTE per referenced query and an outer query on top.
-    Both layers need the same key adjustment as a regular trace query, or deprecated
-    keys and context-prefixed aliases do not resolve.
+    Trace operators build a CTE per referenced builder query and an outer
+    query on top. Both layers need the same key adjustment as regular trace
+    queries, otherwise deprecated keys and context-prefixed aliases don't
+    resolve.
     """
     use_attribute_backend(attribute_backend)
 
@@ -647,8 +648,24 @@ def test_trace_operator_select_fields(
     expansion: str,
 ) -> None:
     """
-    An explicit selectFields returns only the core columns plus the named fields,
-    with no expansion to the full set.
+    Setup:
+    Insert a parent (operation.type = 'parent') with one event and one
+    user-supplied link, plus a child span (operation.type = 'child').
+
+    Tests:
+    1. With selectFields=[], the `A => B` trace_operator returns every column
+       in ALL_SELECT_FIELDS, mirroring the builder_query path. Events arrive
+       as {name, timeUnixNano, attributes} and links as {traceId, spanId}
+       with refType dropped at the consume layer.
+    2. With an explicit selectFields=[{"name": "service.name"}], only the 6
+       hardcoded core columns plus service.name come back — no auto-expansion
+       to the full set.
+
+    See:
+    - pkg/telemetrytraces/trace_operator_cte_builder.go::buildFinalQuery for
+      the expansion gate.
+    - pkg/telemetrytraces/trace_operator_cte_builder.go::buildListQuery for
+      the per-row SELECT.
     """
     use_attribute_backend(attribute_backend)
 
@@ -778,6 +795,14 @@ def test_trace_operator_scalar_limit(
     use_attribute_backend: Callable[[str], None],
     attribute_backend: str,
 ) -> None:
+    """
+    Setup:
+    Insert 4, 3, 2 and 1 spans for services limit-a..limit-d.
+
+    Tests:
+    1. A scalar trace operator grouped by service.name with limit=2 returns
+       only the top two groups by count.
+    """
     use_attribute_backend(attribute_backend)
 
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)

tests/integration/tests/queriertraces/08_aggregation_min_max.py

@@ -30,6 +30,14 @@ def test_traces_aggregate_min_max(
     insert_traces: Callable[[list[Traces]], None],
     noise: str,
 ) -> None:
+    """
+    Setup:
+    http-service: 3 spans (3s / 0.5s / 1s); topic-service: 1 span (4s).
+
+    Tests:
+    max(duration_nano) / min(duration_nano) grouped by service.name return each
+    service's duration extremes, derived from the inserted spans.
+    """
     extra_attrs, extra_resources = trace_noise(noise)
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     specs = [

tests/integration/tests/queriertraces/10_filter_operators.py

@@ -75,6 +75,14 @@ def test_traces_filter_operators(
     expression: str,
     expected_names: set[str],
 ) -> None:
+    """
+    Setup:
+    Five spans across services / environments / methods / statuses, with a
+    numeric `latency_ms` attribute and an `error.type` attribute present on one.
+
+    Tests:
+    Each production-shaped filter operator selects exactly the expected spans.
+    """
     use_attribute_backend(attribute_backend)
 
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)

tests/integration/tests/queriertraces/11_aggregation_depth.py

@@ -38,10 +38,16 @@ def test_traces_aggregate_percentiles(
     noise: str,
 ) -> None:
     """
-    The percentiles match numpy's linear interpolation, which ClickHouse quantile()
-    uses for small inputs. Under the collision variant a numeric attribute named
-    duration_nano unions with the intrinsic column into a multiIf; the aggregation
-    must resolve to the column, not fail with NO_COMMON_TYPE.
+    Setup:
+    5 spans with durations 1s..5s.
+
+    Tests:
+    p25 / p50 / p90 / p95 / p99 over duration_nano match numpy's linear-interpolated
+    percentiles (ClickHouse quantile() uses the same interpolation for small inputs).
+    Under the "collision" variant a numeric span attribute named duration_nano unions
+    with the intrinsic column into a multiIf; the aggregation must still resolve to the
+    intrinsic column rather than error with ClickHouse NO_COMMON_TYPE (386) — the
+    regression behind the span_percentile 500.
     """
     use_attribute_backend(attribute_backend)
 
@@ -101,6 +107,15 @@ def test_traces_aggregate_group_by_non_resource(
     group_key: str,
     expected: dict[str, int],
 ) -> None:
+    """
+    Setup:
+    3 spans: (name op-a, endpoint /a) x2 and (name op-b, endpoint /b) x1.
+
+    Tests:
+    Grouping by an intrinsic column (name) or a string span attribute (endpoint) —
+    not just a resource key — buckets by the real value; under the corrupt variant a
+    same-named colliding attribute must not divert the grouping.
+    """
     use_attribute_backend(attribute_backend)
 
     extra_attrs, extra_resources = trace_noise(noise)
@@ -146,6 +161,15 @@ def test_traces_aggregate_multi_group_by(
     attribute_backend: str,
     noise: str,
 ) -> None:
+    """
+    Setup:
+    Spans across (service.name, name) pairs: (svc-1, op-a) x2, (svc-1, op-b) x1,
+    (svc-2, op-a) x1.
+
+    Tests:
+    Grouping by two keys (a resource key and an intrinsic column) returns one row per
+    present pair with the correct per-pair count.
+    """
     use_attribute_backend(attribute_backend)
 
     extra_attrs, extra_resources = trace_noise(noise)
@@ -206,6 +230,15 @@ def test_traces_aggregate_having_breadth(
     having_expression: str,
     expected_services: set[str],
 ) -> None:
+    """
+    Setup:
+    svc-a: 3 spans of 1s (sum 3s, count 3); svc-b: 1 span of 5s (sum 5s, count 1).
+
+    Tests:
+    HAVING on a sum() aggregation, on count(), and a compound count()+sum() predicate
+    each keep only the qualifying groups — extending the count()-only HAVING in
+    02_aggregation.py.
+    """
     use_attribute_backend(attribute_backend)
 
     extra_attrs, extra_resources = trace_noise(noise)
@@ -260,8 +293,15 @@ def test_traces_duration_nano_qol_filter_with_collision(
     duration_filter: str,
 ) -> None:
     """
-    The duration string is parsed against the intrinsic column; the colliding
-    string attribute is cast, so there is no NO_COMMON_TYPE.
+    Setup:
+    5 spans (durations 1s..5s) that also carry a same-named `duration_nano` span
+    attribute (numeric on some, string on others).
+
+    Tests the duration_nano QoL filter — as a duration string ('2s') and as raw
+    nanoseconds — resolves to the intrinsic column despite the collision: it parses the
+    duration, filters on the real durations (3 spans exceed 2s), and doesn't error with
+    ClickHouse NO_COMMON_TYPE (386). The string attribute is cast; the intrinsic column
+    stays a bare comparison.
     """
     use_attribute_backend(attribute_backend)
 

tests/integration/tests/queriertraces/12_timestamp_aggregations.py

@@ -68,8 +68,12 @@ def test_timestamp_order_aggregations_return_the_instant(
     bound: str,
 ) -> None:
     """
-    Coerced to Float64 the value would be about 1.7e9 seconds, which reads as
-    January 1970 when a consumer takes it as milliseconds.
+    Setup:
+    Two spans for one service, 3 and 1 minutes ago.
+
+    Tests:
+    max/min answer with the extreme span's instant. Coerced to Float64 the value would be
+    ~1.7e9 seconds, which reads as January 1970 when a consumer treats it as milliseconds.
     """
     now = datetime.now(tz=UTC).replace(microsecond=0)
     oldest, newest = now - timedelta(minutes=3), now - timedelta(minutes=1)
@@ -94,7 +98,9 @@ def test_timestamp_quantile_returns_an_instant_within_range(
     insert_traces: Callable[[list[Traces]], None],
 ) -> None:
     """
-    ClickHouse computes quantiles over DateTime64 natively.
+    Tests:
+    ClickHouse computes quantiles over DateTime64 natively, so p99 lands inside the
+    seeded range rather than erroring or collapsing to epoch.
     """
     now = datetime.now(tz=UTC).replace(microsecond=0)
     oldest, newest = now - timedelta(minutes=3), now - timedelta(minutes=1)
@@ -119,6 +125,11 @@ def test_timestamp_counting_aggregations_are_numeric(
     expression: str,
     expected: int | None,
 ) -> None:
+    """
+    Tests:
+    count and rate count rows rather than reading the column's value, so they stay
+    numeric for a time column.
+    """
     now = datetime.now(tz=UTC).replace(microsecond=0)
     seed_two_spans(insert_traces, now - timedelta(minutes=3), now - timedelta(minutes=1))
 
@@ -154,6 +165,11 @@ def test_timestamp_arithmetic_aggregations_are_unsupported(
     insert_traces: Callable[[list[Traces]], None],
     expression: str,
 ) -> None:
+    """
+    Tests:
+    Aggregations that cannot produce a time from a time column fail rather than
+    answering with a rescaled number.
+    """
     now = datetime.now(tz=UTC).replace(microsecond=0)
     seed_two_spans(insert_traces, now - timedelta(minutes=3), now - timedelta(minutes=1))
 

tests/integration/tests/rawexportdata/01_logs.py

@@ -18,6 +18,10 @@ def test_export_raw_data_get_not_allowed(
     create_user_admin: None,  # pylint: disable=unused-argument
     get_token: Callable[[str, str], str],
 ) -> None:
+    """
+    Tests:
+    1. GET request to export_raw_data is rejected with 405 Method Not Allowed
+    """
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
     response = requests.get(
@@ -37,6 +41,16 @@ def test_export_logs_csv(
     get_token: Callable[[str, str], str],
     insert_logs: Callable[[list[Logs]], None],
 ) -> None:
+    """
+    Setup:
+    Insert 3 logs with different severity levels and attributes.
+
+    Tests:
+    1. Export logs as CSV format
+    2. Verify CSV structure and content
+    3. Validate headers are present
+    4. Check log data is correctly formatted
+    """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
 
     insert_logs(
@@ -140,6 +154,16 @@ def test_export_logs_jsonl(
     get_token: Callable[[str, str], str],
     insert_logs: Callable[[list[Logs]], None],
 ) -> None:
+    """
+    Setup:
+    Insert 2 logs with different attributes.
+
+    Tests:
+    1. Export logs as JSONL format
+    2. Verify JSONL structure and content
+    3. Check each line is valid JSON
+    4. Validate log data is correctly formatted
+    """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
 
     insert_logs(
@@ -226,6 +250,14 @@ def test_export_logs_with_filter(
     get_token: Callable[[str, str], str],
     insert_logs: Callable[[list[Logs]], None],
 ) -> None:
+    """
+    Setup:
+    Insert logs with different severity levels.
+
+    Tests:
+    1. Export logs with filter applied
+    2. Verify only filtered logs are returned
+    """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
 
     insert_logs(
@@ -303,6 +335,14 @@ def test_export_logs_with_limit(
     get_token: Callable[[str, str], str],
     insert_logs: Callable[[list[Logs]], None],
 ) -> None:
+    """
+    Setup:
+    Insert 5 logs.
+
+    Tests:
+    1. Export logs with limit applied
+    2. Verify only limited number of logs are returned
+    """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
 
     logs = []
@@ -363,6 +403,14 @@ def test_export_logs_with_columns(
     get_token: Callable[[str, str], str],
     insert_logs: Callable[[list[Logs]], None],
 ) -> None:
+    """
+    Setup:
+    Insert logs with various attributes.
+
+    Tests:
+    1. Export logs with specific columns
+    2. Verify only specified columns are returned
+    """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
 
     insert_logs(
@@ -441,6 +489,14 @@ def test_export_logs_with_order_by(
     get_token: Callable[[str, str], str],
     insert_logs: Callable[[list[Logs]], None],
 ) -> None:
+    """
+    Setup:
+    Insert logs at different timestamps.
+
+    Tests:
+    1. Export logs with ascending timestamp order
+    2. Verify logs are returned in correct order
+    """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
 
     insert_logs(
@@ -524,6 +580,14 @@ def test_export_logs_with_complex_filter(
     get_token: Callable[[str, str], str],
     insert_logs: Callable[[list[Logs]], None],
 ) -> None:
+    """
+    Setup:
+    Insert logs with various service names and severity levels.
+
+    Tests:
+    1. Export logs with complex filter (multiple conditions)
+    2. Verify only logs matching all conditions are returned
+    """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
 
     insert_logs(

tests/integration/tests/rawexportdata/02_traces.py

@@ -24,6 +24,10 @@ def test_export_raw_data_get_not_allowed(
     create_user_admin: None,  # pylint: disable=unused-argument
     get_token: Callable[[str, str], str],
 ) -> None:
+    """
+    Tests:
+    1. GET request to export_raw_data is rejected with 405 Method Not Allowed
+    """
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
     response = requests.get(
@@ -43,6 +47,16 @@ def test_export_traces_csv(
     get_token: Callable[[str, str], str],
     insert_traces: Callable[[list[Traces]], None],
 ) -> None:
+    """
+    Setup:
+    Insert 3 traces with different attributes.
+
+    Tests:
+    1. Export traces as CSV format
+    2. Verify CSV structure and content
+    3. Validate headers are present
+    4. Check trace data is correctly formatted
+    """
     http_service_trace_id = TraceIdGenerator.trace_id()
     http_service_span_id = TraceIdGenerator.span_id()
     http_service_db_span_id = TraceIdGenerator.span_id()
@@ -176,6 +190,16 @@ def test_export_traces_jsonl(
     get_token: Callable[[str, str], str],
     insert_traces: Callable[[list[Traces]], None],
 ) -> None:
+    """
+    Setup:
+    Insert 2 traces with different attributes.
+
+    Tests:
+    1. Export traces as JSONL format
+    2. Verify JSONL structure and content
+    3. Check each line is valid JSON
+    4. Validate trace data is correctly formatted
+    """
     http_service_trace_id = TraceIdGenerator.trace_id()
     http_service_span_id = TraceIdGenerator.span_id()
     topic_service_trace_id = TraceIdGenerator.trace_id()
@@ -283,6 +307,14 @@ def test_export_traces_with_filter(
     get_token: Callable[[str, str], str],
     insert_traces: Callable[[list[Traces]], None],
 ) -> None:
+    """
+    Setup:
+    Insert traces with different service names.
+
+    Tests:
+    1. Export traces with filter applied
+    2. Verify only filtered traces are returned
+    """
     service_a_trace_id = TraceIdGenerator.trace_id()
     service_a_span_id = TraceIdGenerator.span_id()
     service_b_trace_id = TraceIdGenerator.trace_id()
@@ -374,6 +406,14 @@ def test_export_traces_with_limit(
     get_token: Callable[[str, str], str],
     insert_traces: Callable[[list[Traces]], None],
 ) -> None:
+    """
+    Setup:
+    Insert 5 traces.
+
+    Tests:
+    1. Export traces with limit applied
+    2. Verify only limited number of traces are returned
+    """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
 
     traces = []
@@ -437,6 +477,11 @@ def test_export_traces_multiple_queries_rejected(
     create_user_admin: None,  # pylint: disable=unused-argument
     get_token: Callable[[str, str], str],
 ) -> None:
+    """
+    Tests:
+    1. POST with multiple builder queries but no trace operator is rejected
+    2. Verify 400 error is returned
+    """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     start_ns = int((now - timedelta(minutes=5)).timestamp() * 1e9)
     end_ns = int(now.timestamp() * 1e9)
@@ -483,6 +528,14 @@ def test_export_traces_with_composite_query_trace_operator(
     get_token: Callable[[str, str], str],
     insert_traces: Callable[[list[Traces]], None],
 ) -> None:
+    """
+    Setup:
+    Insert multiple traces with parent-child relationships.
+
+    Tests:
+    1. Export traces using trace operator in composite query (POST)
+    2. Verify trace operator query works correctly
+    """
     parent_trace_id = TraceIdGenerator.trace_id()
     parent_span_id = TraceIdGenerator.span_id()
     child_span_id_1 = TraceIdGenerator.span_id()
@@ -614,6 +667,14 @@ def test_export_traces_with_select_fields(
     get_token: Callable[[str, str], str],
     insert_traces: Callable[[list[Traces]], None],
 ) -> None:
+    """
+    Setup:
+    Insert traces with various attributes.
+
+    Tests:
+    1. Export traces with specific select fields via POST
+    2. Verify only specified fields are returned in the output
+    """
     trace_id = TraceIdGenerator.trace_id()
     span_id = TraceIdGenerator.span_id()
 

tests/integration/tests/spanmapper/01_span_mapper.py

@@ -14,6 +14,22 @@ def test_create_groups_and_simulate_with_backfill(
     create_user_admin: types.Operation,  # pylint: disable=unused-argument
     get_token: Callable[[str, str], str],
 ) -> None:
+    """
+    Setup:
+    Create a mapping group and a mapper through the CRUD API, then hit the
+    simulate endpoint with two groups: the saved one referenced by name only
+    (server backfills its mappers from the store) and a second, unsaved group
+    sent with inline mappers.
+
+    Tests:
+    1. Create a group (POST /span_mapper_groups) and read back its id
+    2. Create a mapper in that group (POST .../{groupId}/span_mappers)
+    3. List groups and mappers and verify what we created is persisted
+    4. Simulate spans against both groups: a matching span gets the saved
+       group's mappers (backfilled, copy) and the inline group's mappers
+       (move) applied, while a span matching no condition passes through
+    5. Delete the group and verify it (and its mappers) are gone
+    """
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
     create_group = requests.post(

tests/integration/tests/spanmapper/02_default_groups.py

@@ -15,8 +15,19 @@ def test_default_groups_are_seeded_and_shipped_items_are_toggle_only(
     get_token: Callable[[str, str], str],
 ) -> None:
     """
-    The reconciler seeds the shipped groups at startup and on org creation, so
-    nothing is created here.
+    Setup:
+    A fresh org. The reconciler seeds the shipped mapping groups at startup
+    and on org creation, so nothing has to be created here.
+
+    Tests:
+    1. The list contains llm, agent and tool as system groups with shipped
+       substrings
+    2. Shipped mappers and their sources are system-owned and enabled
+    3. A shipped name cannot be taken by a user group, and system groups and
+       mappers cannot be deleted
+    4. A shipped source can be switched off and a user override added; both
+       round-trip through PATCH and the simulator honours them
+    5. A shipped substring can be switched off and a user one added
     """
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
     headers = {"authorization": f"Bearer {token}", "content-type": "application/json"}

tests/integration/tests/thirdpartyapi/01_domain_list.py

@@ -30,8 +30,13 @@ def test_domain_list_last_seen_is_the_span_instant(
     noise: str,
 ) -> None:
     """
-    A numeric coercion of the DateTime64 column would give seconds since epoch,
-    which reads as January 1970 once taken as milliseconds.
+    Setup:
+    Two client spans to api.stripe.com, 3 and 1 minutes ago.
+
+    Tests:
+    Last Seen resolves to the newer span's instant (#5824). A numeric coercion of the
+    DateTime64 column would yield seconds since epoch (~1.7e9), which lands in
+    January 1970 once read as milliseconds.
     """
     extra_attrs = {"timestamp": 1.0} if noise == "timestamp_attr" else {}
     now = datetime.now(tz=UTC).replace(microsecond=0)
@@ -87,7 +92,12 @@ def test_domain_list_groups_by_domain_and_counts_endpoints(
     insert_traces: Callable[[list[Traces]], None],
 ) -> None:
     """
-    The columns that only feed the error-rate formula are stripped from the response.
+    Setup:
+    api.github.com with 2 distinct URLs (one called twice); api.openai.com with 1.
+
+    Tests:
+    One row per domain, endpoint counts are distinct-URL counts, and the columns
+    used only to feed the error-rate formula are stripped from the response.
     """
     now = datetime.now(tz=UTC).replace(microsecond=0)
     specs = [
@@ -152,6 +162,14 @@ def test_domain_list_excludes_non_client_and_urlless_spans(
     get_token: Callable[[str, str], str],
     insert_traces: Callable[[list[Traces]], None],
 ) -> None:
+    """
+    Setup:
+    A client span with an http.url, a server span with one, and a client span without.
+
+    Tests:
+    The base filter (`http_url EXISTS AND kind_string = 'Client'`) keeps only the
+    first domain; the other two never reach the response.
+    """
     now = datetime.now(tz=UTC).replace(microsecond=0)
     specs = [
         ("included.example.com", TracesKind.SPAN_KIND_CLIENT, "https://included.example.com/a"),
@@ -203,7 +221,11 @@ def test_domain_list_show_ip_masks_ip_domains(
     show_ip: bool,
 ) -> None:
     """
-    net.ParseIP accepts both IP forms, so show_ip gates IPv4 and IPv6 alike.
+    Setup:
+    One client span each to a hostname, a bare IPv4 and a bare IPv6 address.
+
+    Tests:
+    show_ip gates both IP forms (net.ParseIP accepts either); the hostname always shows.
     """
     now = datetime.now(tz=UTC).replace(microsecond=0)
     specs = [
@@ -251,7 +273,12 @@ def test_domain_list_latency_and_rate_values(
     insert_traces: Callable[[list[Traces]], None],
 ) -> None:
     """
-    rps is rate() over the scalar window, so the busier domain's rate is 3x the other's.
+    Setup:
+    3 client spans of 250ms on one domain, 1 on another.
+
+    Tests:
+    p99 is duration_nano in nanoseconds, and rps is rate() over the scalar window
+    (count / (end - start) seconds), so the busier domain's rate is 3x the other's.
     """
     now = datetime.now(tz=UTC).replace(microsecond=0)
     busy, quiet = "busy.example.com", "quiet.example.com"
@@ -307,6 +334,14 @@ def test_domain_list_error_rate_is_a_percentage_of_spans(
     total: int,
     expected: float,
 ) -> None:
+    """
+    Setup:
+    `total` client spans on one domain, `errors` of them failed.
+
+    Tests:
+    error_rate is (error/total_span)*100 across the range, including the all-failed and
+    none-failed edges.
+    """
     now = datetime.now(tz=UTC).replace(microsecond=0)
     domain = f"errors-{label}.example.com"
     statuses = [TracesStatusCode.STATUS_CODE_ERROR] * errors + [TracesStatusCode.STATUS_CODE_OK] * (total - errors)
@@ -349,6 +384,14 @@ def test_domain_list_user_filter_narrows_the_base_filter(
     get_token: Callable[[str, str], str],
     insert_traces: Callable[[list[Traces]], None],
 ) -> None:
+    """
+    Setup:
+    Two client spans to the same domain from different services.
+
+    Tests:
+    A caller filter is ANDed onto the base filter, so only the named service's span
+    contributes.
+    """
     now = datetime.now(tz=UTC).replace(microsecond=0)
     domain = "filtered.example.com"
     insert_traces(
@@ -391,6 +434,14 @@ def test_domain_list_caller_group_by_adds_a_dimension(
     get_token: Callable[[str, str], str],
     insert_traces: Callable[[list[Traces]], None],
 ) -> None:
+    """
+    Setup:
+    Two client spans to one domain, from two services.
+
+    Tests:
+    A caller groupBy is merged after the domain key, so the domain splits into one row
+    per group value and the extra key gets its own column.
+    """
     now = datetime.now(tz=UTC).replace(microsecond=0)
     domain = "grouped.example.com"
     insert_traces(
@@ -440,6 +491,10 @@ def test_domain_list_is_empty_outside_the_seeded_window(
     create_user_admin: None,  # pylint: disable=unused-argument
     get_token: Callable[[str, str], str],
 ) -> None:
+    """
+    Tests:
+    A window with no matching spans answers 200 with no rows, not an error.
+    """
     now = datetime.now(tz=UTC).replace(microsecond=0)
     long_ago = now - timedelta(days=400)
     response = make_third_party_apis_request(
@@ -454,6 +509,10 @@ def test_domain_list_is_empty_outside_the_seeded_window(
 
 
 def test_domain_list_requires_authentication(signoz: types.SigNoz) -> None:
+    """
+    Tests:
+    The route sits behind ViewAccess, so an unauthenticated call is rejected.
+    """
     now = datetime.now(tz=UTC).replace(microsecond=0)
     response = make_third_party_apis_request(
         signoz,
@@ -470,6 +529,11 @@ def test_domain_list_rejects_inverted_time_range(
     create_user_admin: None,  # pylint: disable=unused-argument
     get_token: Callable[[str, str], str],
 ) -> None:
+    """
+    Tests:
+    The request is validated before any query is built, so start >= end is a client
+    error rather than an empty result.
+    """
     now = datetime.now(tz=UTC).replace(microsecond=0)
     response = make_third_party_apis_request(
         signoz,