Utilizing otlp logger so error/info/debug logs are directed to otel logs sink. Also includes other logging related cleanup, as per code review comments.

Author: kamalchaturvedi

internal/collector/logsgzipprocessor/processor.go

@@ -10,7 +10,6 @@ import (
 	"context"
 	"fmt"
 	"io"
-	"log/slog"
 	"sync"
 
 	"go.opentelemetry.io/collector/component"
@@ -19,6 +18,7 @@ import (
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/processor"
 	"go.uber.org/multierr"
+	"go.uber.org/zap"
 )
 
 // nolint: ireturn
@@ -34,11 +34,14 @@ func NewFactory() processor.Factory {
 
 // nolint: ireturn
 func createLogsGzipProcessor(_ context.Context,
-	_ processor.Settings,
+	settings processor.Settings,
 	cfg component.Config,
 	logs consumer.Logs,
 ) (processor.Logs, error) {
-	return newLogsGzipProcessor(logs), nil
+	logger := settings.Logger
+	logger.Info("Creating logs gzip processor")
+
+	return newLogsGzipProcessor(logs, settings), nil
 }
 
 // logsGzipProcessor is a custom-processor implementation for compressing individual log records into
@@ -51,7 +54,8 @@ type logsGzipProcessor struct {
 	// Otherwise, creating a new compressor for every log record would result in frequent memory allocations
 	// and increased garbage collection overhead, especially under high-throughput workload like this one.
 	// By pooling these objects, we minimize allocation churn, reduce GC pressure, and improve overall performance.
-	pool *sync.Pool
+	pool     *sync.Pool
+	settings processor.Settings
 }
 
 type GzipWriter interface {
@@ -60,14 +64,15 @@ type GzipWriter interface {
 	Reset(w io.Writer)
 }
 
-func newLogsGzipProcessor(logs consumer.Logs) *logsGzipProcessor {
+func newLogsGzipProcessor(logs consumer.Logs, settings processor.Settings) *logsGzipProcessor {
 	return &logsGzipProcessor{
 		nextConsumer: logs,
 		pool: &sync.Pool{
 			New: func() any {
 				return gzip.NewWriter(nil)
 			},
 		},
+		settings: settings,
 	}
 }
 
@@ -77,7 +82,7 @@ func (p *logsGzipProcessor) ConsumeLogs(ctx context.Context, ld plog.Logs) error
 	for i := range resourceLogs.Len() {
 		scopeLogs := resourceLogs.At(i).ScopeLogs()
 		for j := range scopeLogs.Len() {
-			err := p.processLogRecords(ctx, scopeLogs.At(j).LogRecords())
+			err := p.processLogRecords(scopeLogs.At(j).LogRecords())
 			if err != nil {
 				errs = multierr.Append(errs, err)
 			}
@@ -90,15 +95,19 @@ func (p *logsGzipProcessor) ConsumeLogs(ctx context.Context, ld plog.Logs) error
 	return p.nextConsumer.ConsumeLogs(ctx, ld)
 }
 
-func (p *logsGzipProcessor) processLogRecords(ctx context.Context, logRecords plog.LogRecordSlice) error {
+func (p *logsGzipProcessor) processLogRecords(logRecords plog.LogRecordSlice) error {
 	var errs error
 	// Filter out unsupported data types in the log before processing
 	logRecords.RemoveIf(func(lr plog.LogRecord) bool {
 		body := lr.Body()
 		// Keep only STRING or BYTES types
+		if body.Type() != pcommon.ValueTypeStr &&
+			body.Type() != pcommon.ValueTypeBytes {
+			p.settings.Logger.Debug("Skipping log record with unsupported body type", zap.Any("type", body.Type()))
+			return true
+		}
 
-		return body.Type() != pcommon.ValueTypeStr &&
-			body.Type() != pcommon.ValueTypeBytes
+		return false
 	})
 	// Process remaining valid records
 	for k := range logRecords.Len() {
@@ -114,15 +123,13 @@ func (p *logsGzipProcessor) processLogRecords(ctx context.Context, logRecords pl
 		}
 		gzipped, err := p.gzipCompress(data)
 		if err != nil {
-			slog.ErrorContext(ctx, "failed to compress log record", slog.Any("error", err))
-			errs = multierr.Append(errs, err)
+			errs = multierr.Append(errs, fmt.Errorf("failed to compress log record: %w", err))
 
 			continue
 		}
 		err = record.Body().FromRaw(gzipped)
 		if err != nil {
-			slog.ErrorContext(ctx, "failed to set gzipped data to log record body", slog.Any("error", err))
-			errs = multierr.Append(errs, err)
+			errs = multierr.Append(errs, fmt.Errorf("failed to set gzipped data to log record body: %w", err))
 
 			continue
 		}
@@ -133,18 +140,21 @@ func (p *logsGzipProcessor) processLogRecords(ctx context.Context, logRecords pl
 
 func (p *logsGzipProcessor) gzipCompress(data []byte) ([]byte, error) {
 	var buf bytes.Buffer
+	var err error
 	wIface := p.pool.Get()
 	w, ok := wIface.(GzipWriter)
 	if !ok {
 		return nil, fmt.Errorf("writer of type %T not supported", wIface)
 	}
 	w.Reset(&buf)
 	defer func() {
-		w.Close()
+		if err = w.Close(); err != nil {
+			p.settings.Logger.Error("Failed to close gzip writer", zap.Error(err))
+		}
 		p.pool.Put(w)
 	}()
 
-	_, err := w.Write(data)
+	_, err = w.Write(data)
 	if err != nil {
 		return nil, err
 	}
@@ -162,11 +172,11 @@ func (p *logsGzipProcessor) Capabilities() consumer.Capabilities {
 }
 
 func (p *logsGzipProcessor) Start(ctx context.Context, _ component.Host) error {
-	slog.DebugContext(ctx, "starting logs gzip processor")
+	p.settings.Logger.Info("Starting logs gzip processor")
 	return nil
 }
 
 func (p *logsGzipProcessor) Shutdown(ctx context.Context) error {
-	slog.DebugContext(ctx, "shutting down logs gzip processor")
+	p.settings.Logger.Info("Shutting down logs gzip processor")
 	return nil
 }

internal/collector/logsgzipprocessor/processor_benchmark_test.go

@@ -12,6 +12,7 @@ import (
 
 	"go.opentelemetry.io/collector/consumer/consumertest"
 	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/processor"
 )
 
 // Helper to generate logs with variable size and content
@@ -59,12 +60,12 @@ func BenchmarkGzipProcessor(b *testing.B) {
 		b.Run(bm.name, func(b *testing.B) {
 			b.ReportAllocs()
 			consumer := &consumertest.LogsSink{}
-			processor := newLogsGzipProcessor(consumer)
+			p := newLogsGzipProcessor(consumer, processor.Settings{})
 			logs := generateLogs(bm.numRecords, bm.recordSize)
 
 			b.ResetTimer()
 			for i := 0; i < b.N; i++ {
-				_ = processor.ConsumeLogs(context.Background(), logs)
+				_ = p.ConsumeLogs(context.Background(), logs)
 			}
 		})
 	}
@@ -76,13 +77,13 @@ func BenchmarkGzipProcessor_Concurrent(b *testing.B) {
 	const workers = 8
 	logs := generateLogs(1000, 1000)
 	consumer := &consumertest.LogsSink{}
-	processor := newLogsGzipProcessor(consumer)
+	p := newLogsGzipProcessor(consumer, processor.Settings{})
 
 	b.ReportAllocs()
 	b.ResetTimer()
 	b.RunParallel(func(pb *testing.PB) {
 		for pb.Next() {
-			_ = processor.ConsumeLogs(context.Background(), logs)
+			_ = p.ConsumeLogs(context.Background(), logs)
 		}
 	})
 }

internal/collector/logsgzipprocessor/processor_test.go

@@ -17,6 +17,8 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/consumer/consumertest"
 	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/processor/processortest"
+	"go.uber.org/zap"
 )
 
 var dummyInputStr = "hello world"
@@ -27,22 +29,24 @@ func TestGzipProcessor(t *testing.T) {
 		name  string
 	}{
 		{
-			name:  "string content",
+			name:  "Test 1: string content",
 			input: dummyInputStr,
 		},
 		{
-			name:  "byte content",
+			name:  "Test 2: byte content",
 			input: []byte("binary data"),
 		},
 		{
-			name:  "integer content",
+			name:  "Test 3: integer content",
 			input: 12345,
 		},
 	}
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			ctx := context.Background()
+			settings := processortest.NewNopSettings(processortest.NopType)
+			settings.Logger = zap.NewNop()
 			// Setup: create a log record with the test case content
 			logs := plog.NewLogs()
 			logRecord := logs.ResourceLogs().AppendEmpty().
@@ -60,7 +64,7 @@ func TestGzipProcessor(t *testing.T) {
 			}
 
 			next := &consumertest.LogsSink{}
-			processor := newLogsGzipProcessor(next)
+			processor := newLogsGzipProcessor(next, settings)
 			require.NoError(t, processor.Start(ctx, nil))
 
 			capability := processor.Capabilities()
@@ -121,18 +125,20 @@ func TestGzipProcessorFailure(t *testing.T) {
 		isGzipCloseError bool
 	}{
 		{
-			name:             "gzip write failure",
+			name:             "Test 1: gzip write failure",
 			isGzipWriteError: true,
 		},
 		{
-			name:             "gzip writer close failure",
+			name:             "Test 2: gzip writer close failure",
 			isGzipCloseError: true,
 		},
 	}
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			ctx := context.Background()
+			settings := processortest.NewNopSettings(processortest.NopType)
+			settings.Logger = zap.NewNop()
 			// Setup: create a log record with the test case content
 			logs := plog.NewLogs()
 			logRecord := logs.ResourceLogs().AppendEmpty().
@@ -151,6 +157,7 @@ func TestGzipProcessorFailure(t *testing.T) {
 						return mockWriter
 					},
 				},
+				settings: settings,
 			}
 			require.NoError(t, processor.Start(ctx, nil))