refactor: addressing review comments

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

@@ -433,6 +433,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
@@ -458,6 +459,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
 }