pkg/util/log: reducing allocs in otlp sink

this aims to increase the performance and decrease the number of
allocations made in the otlp sink.

Benchmark results:
```
name         old time/op    new time/op    delta
OTLPSink-12     536µs ± 1%     535µs ± 0%     ~     (p=0.156 n=10+9)

name         old alloc/op   new alloc/op   delta
OTLPSink-12    18.5kB ±25%    15.7kB ±11%  -14.97%  (p=0.004 n=9+9)

name         old allocs/op  new allocs/op  delta
OTLPSink-12       132 ± 0%       123 ± 0%   -6.82%  (p=0.000 n=10+10)
```

benchmark was done in a e2e environment (crdb->otel-collector->datadog)
and with gzip compression turned on.

part of: CRDB-51667

Release note: none

Author: TheComputerM

pkg/util/log/otlp_client.go

@@ -6,8 +6,9 @@
 package log
 
 import (
-	"bytes"
 	"context"
+	"strings"
+	"sync"
 
 	"github.com/cockroachdb/cockroach/pkg/cli/exit"
 	"github.com/cockroachdb/cockroach/pkg/util/log/logconfig"
@@ -31,17 +32,29 @@ const (
 	logAttributeSinkKey      = "sink.name"
 )
 
+// pool for OTEL spec log record objects that we can reuse between requests
+var otlpLogRecordPool = sync.Pool{
+	New: func() any {
+		return &lpb.LogRecord{
+			Body: &cpb.AnyValue{
+				Value: &cpb.AnyValue_StringValue{StringValue: ""},
+			},
+		}
+	},
+}
+
 // OpenTelemetry log sink
 type otlpSink struct {
-	conn     *grpc.ClientConn
-	lsc      collpb.LogsServiceClient
-	name     string
-	resource *rpb.Resource
+	conn *grpc.ClientConn
+	lsc  collpb.LogsServiceClient
+
+	// requestObject should not be modified concurrently as it is reused
+	// between requests
+	requestObject *collpb.ExportLogsServiceRequest
 }
 
 func newOTLPSink(config logconfig.OTLPSinkConfig) (*otlpSink, error) {
 	dialOpts := []grpc.DialOption{
-		grpc.WithUserAgent("CRDB OTLP over gRPC logs exporter"),
 		grpc.WithTransportCredentials(insecure.NewCredentials()),
 	}
 
@@ -55,24 +68,35 @@ func newOTLPSink(config logconfig.OTLPSinkConfig) (*otlpSink, error) {
 	}
 
 	lsc := collpb.NewLogsServiceClient(conn)
-	name := config.SinkName
-	return &otlpSink{
+	sink := &otlpSink{
 		conn: conn,
 		lsc:  lsc,
-		name: name,
-		resource: &rpb.Resource{
-			Attributes: []*cpb.KeyValue{
+		requestObject: &collpb.ExportLogsServiceRequest{
+			ResourceLogs: []*lpb.ResourceLogs{
 				{
-					Key:   logAttributeServiceKey,
-					Value: &cpb.AnyValue{Value: &cpb.AnyValue_StringValue{StringValue: logAttributeServiceValue}},
-				},
-				{
-					Key:   logAttributeSinkKey,
-					Value: &cpb.AnyValue{Value: &cpb.AnyValue_StringValue{StringValue: name}},
+					Resource: &rpb.Resource{
+						Attributes: []*cpb.KeyValue{
+							{
+								Key:   logAttributeServiceKey,
+								Value: &cpb.AnyValue{Value: &cpb.AnyValue_StringValue{StringValue: logAttributeServiceValue}},
+							},
+							{
+								Key:   logAttributeSinkKey,
+								Value: &cpb.AnyValue{Value: &cpb.AnyValue_StringValue{StringValue: config.SinkName}},
+							},
+						},
+					},
+					InstrumentationLibraryLogs: []*lpb.InstrumentationLibraryLogs{
+						{
+							Logs: nil,
+						},
+					},
 				},
 			},
 		},
-	}, nil
+	}
+
+	return sink, nil
 }
 
 func (sink *otlpSink) isNotShutdown() bool {
@@ -91,43 +115,47 @@ func (sink *otlpSink) exitCode() exit.Code {
 	return exit.LoggingNetCollectorUnavailable()
 }
 
-// converts the raw bytes into OTEL log records
-func extractRecordsToOTLP(b []byte) []*lpb.LogRecord {
-	bodies := bytes.Split(b, []byte("\n"))
-	records := make([]*lpb.LogRecord, 0, len(bodies))
-
-	for _, body := range bodies {
-		body = bytes.TrimSpace(body)
-		if len(body) > 0 {
-			records = append(records, &lpb.LogRecord{
-				Body: &cpb.AnyValue{
-					Value: &cpb.AnyValue_StringValue{
-						StringValue: string(body),
-					},
-				},
-			})
+// converts the raw bytes into OTEL log records using the otlpLogRecordPool
+func otlpExtractRecords(b []byte) []*lpb.LogRecord {
+	body := string(b)
+	records := make([]*lpb.LogRecord, 0, strings.Count(body, "\n")+1)
+
+	start := 0
+	for i, ch := range body {
+		if ch == '\n' {
+			if i > start {
+				record := otlpLogRecordPool.Get().(*lpb.LogRecord)
+				record.Body.Value.(*cpb.AnyValue_StringValue).StringValue = body[start:i]
+				records = append(records, record)
+			}
+			start = i + 1
 		}
 	}
 
+	// check at the very end to ensure entire buffer is processed
+	if start < len(body) {
+		record := otlpLogRecordPool.Get().(*lpb.LogRecord)
+		record.Body.Value.(*cpb.AnyValue_StringValue).StringValue = body[start:]
+		records = append(records, record)
+	}
+
 	return records
 }
 
 func (sink *otlpSink) output(b []byte, opts sinkOutputOptions) error {
-	records := extractRecordsToOTLP(b)
 	ctx := context.Background()
 
-	_, err := sink.lsc.Export(ctx, &collpb.ExportLogsServiceRequest{
-		ResourceLogs: []*lpb.ResourceLogs{
-			{
-				Resource: sink.resource,
-				InstrumentationLibraryLogs: []*lpb.InstrumentationLibraryLogs{
-					{
-						Logs: records,
-					},
-				},
-			},
-		},
-	})
+	records := otlpExtractRecords(b)
+	sink.requestObject.ResourceLogs[0].InstrumentationLibraryLogs[0].Logs = records
+
+	// transmit the log over the network
+	_, err := sink.lsc.Export(ctx, sink.requestObject)
+
+	// put the records back into the pool
+	for _, record := range records {
+		record.Body.Value.(*cpb.AnyValue_StringValue).StringValue = ""
+		otlpLogRecordPool.Put(record)
+	}
 
 	if status.Code(err) == codes.OK {
 		return nil

pkg/util/log/otlp_client_test.go

@@ -219,3 +219,50 @@ func TestOTLPClientSeverity(t *testing.T) {
 		return strings.Join(output, "\n")
 	})
 }
+
+func TestOTLPExtractRecords(t *testing.T) {
+	tests := map[string]struct {
+		input  string
+		result []string
+	}{
+		"single_line": {
+			input:  "Hello World",
+			result: []string{"Hello World"},
+		},
+		"multiple_lines": {
+			input:  "Message 1\nMessage 2\n\nMessage 3",
+			result: []string{"Message 1", "Message 2", "Message 3"},
+		},
+		"trailing_newline": {
+			input:  "Message 1\nMessage 2\n",
+			result: []string{"Message 1", "Message 2"},
+		},
+		"leading_newline": {
+			input:  "\nMessage 1\nMessage 2",
+			result: []string{"Message 1", "Message 2"},
+		},
+		"redaction_markers": {
+			input:  "Message 1\n‹Message 2›\nMessage 3",
+			result: []string{"Message 1", "‹Message 2›", "Message 3"},
+		},
+		"empty_string": {
+			input:  "",
+			result: []string{},
+		},
+		"newline_only": {
+			input:  "\n\n",
+			result: []string{},
+		},
+	}
+
+	for name, test := range tests {
+		t.Run(name, func(t *testing.T) {
+			body := []byte(test.input)
+			records := otlpExtractRecords(body)
+			require.Len(t, records, len(test.result))
+			for i, record := range records {
+				require.Equal(t, test.result[i], record.Body.GetStringValue())
+			}
+		})
+	}
+}