Merge #150655

150655: pkg/util/log: add http mode to otlp sink r=TheComputerM a=TheComputerM

Added HTTP support for OTLP sink as many platforms support directly ingesting logs through otlp+http but not gRPC, it enables users to perform agentless deployments (no intermediate service like otel-collector or datadog-agent) where the logs are directly ingested by the target app (i.e datadog or grafana)

Had to use proto.Marshal instead of protoutil, because the .proto definitions for OpenTelemetry are not generated by gogoproto and therefore don't have the additional methods that protoutil requires

Also refactored tests to be table driven and make them run on both modes, merged the datadriven tests into the table driven ones during the process.

Epic: none

Release note (general change): Added HTTP mode to OTLP log sink that can export logs to OpenTelemetry-compatible targets over HTTP.

Co-authored-by: Mudit Somani <[email protected]>

Author: craig[bot]

docs/generated/logsinks.md

@@ -289,6 +289,7 @@ Type-specific configuration options:
 |--|--|
 | `channels` | the list of logging channels that use this sink. See the [channel selection configuration](#channel-format) section for details.  |
 | `address` | the network address of the gRPC endpoint for ingestion of logs on your OpenTelemetry Collector/Platform. The host/address and port parts are separated with a colon. |
+| `mode` | decides which protocol to use for exporting logs, can be "grpc" or "http". Set to "grpc" by default. Inherited from `otlp-defaults.mode` if not specified. |
 | `compression` | can be "none" or "gzip" to enable gzip compression. Set to "gzip" by default. Inherited from `otlp-defaults.compression` if not specified. |
 
 

pkg/cli/log_flags_test.go

@@ -58,6 +58,7 @@ func TestSetupLogging(t *testing.T) {
 		`max-buffer-size: 50MiB, ` +
 		`format: newline}}`
 	const defaultOTLPConfig = `otlp-defaults: {` +
+		`mode: grpc, ` +
 		`compression: gzip, ` +
 		`filter: INFO, ` +
 		`format: json, ` +

pkg/testutils/lint/lint_test.go

@@ -1444,6 +1444,7 @@ func TestLint(t *testing.T) {
 			":!sql/types/types_jsonpb.go",
 			":!sql/schemachanger/scplan/scviz/maps.go",
 			":!workload/schemachange/tracing.go",
+			":!util/log/otlp_client.go",
 		)
 		if err != nil {
 			t.Fatal(err)
@@ -1492,6 +1493,7 @@ func TestLint(t *testing.T) {
 			":!storage/mvcc_value.go",
 			":!roachpb/data.go",
 			":!sql/types/types_jsonpb.go",
+			":!util/log/otlp_client_test.go",
 		)
 		if err != nil {
 			t.Fatal(err)

pkg/util/log/BUILD.bazel

@@ -98,6 +98,7 @@ go_library(
         "@org_golang_google_grpc//encoding/gzip",
         "@org_golang_google_grpc//stats",
         "@org_golang_google_grpc//status",
+        "@org_golang_google_protobuf//proto",
     ] + select({
         "@io_bazel_rules_go//go/platform:aix": [
             "@org_golang_x_sys//unix",
@@ -195,6 +196,7 @@ go_test(
         "//pkg/util/caller",
         "//pkg/util/ctxgroup",
         "//pkg/util/encoding",
+        "//pkg/util/httputil",
         "//pkg/util/leaktest",
         "//pkg/util/log/channel",
         "//pkg/util/log/logconfig",
@@ -216,8 +218,8 @@ go_test(
         "@com_github_stretchr_testify//assert",
         "@com_github_stretchr_testify//require",
         "@io_opentelemetry_go_proto_otlp//collector/logs/v1:logs",
-        "@io_opentelemetry_go_proto_otlp//common/v1:common",
         "@org_golang_google_grpc//:grpc",
+        "@org_golang_google_protobuf//proto",
         "@org_golang_x_sys//unix",
     ],
 )

pkg/util/log/flags.go

@@ -123,11 +123,8 @@ func ApplyConfig(
 	// closes the underlying gRPC connection of OTLP sinks.
 	closeOTLPSinks := func() {
 		for _, fc := range sinkInfos {
-			if sink, ok := fc.sink.(*otlpSink); ok && sink.isNotShutdown() {
-				// The reason for nolint:grpcconnclose is that we are not using *rpc.Context
-				// as it is primarily used for communication between crdb nodes, and doesn't
-				// fit this usecase.
-				if err := sink.conn.Close(); err != nil { // nolint:grpcconnclose
+			if sink, ok := fc.sink.(*otlpSink); ok {
+				if err := sink.client.Close(); err != nil {
 					fmt.Fprintf(OrigStderr, "# OTLP Sink Cleanup Warning: %s\n", err.Error())
 				}
 			}

pkg/util/log/logconfig/config.go

@@ -589,7 +589,14 @@ type HTTPSinkConfig struct {
 	sinkName string
 }
 
+var OTLPModeGRPC = "grpc"
+var OTLPModeHTTP = "http"
+
 type OTLPDefaults struct {
+	// Mode decides which protocol to use for exporting logs, can be "grpc" or "http".
+	// Set to "grpc" by default.
+	Mode *string `yaml:",omitempty"`
+
 	// Compression can be "none" or "gzip" to enable gzip compression.
 	// Set to "gzip" by default.
 	Compression *string `yaml:",omitempty"`

pkg/util/log/logconfig/validate.go

@@ -124,6 +124,7 @@ func (c *Config) Validate(defaultLogDir *string) (resErr error) {
 			},
 		},
 		Compression: &GzipCompression,
+		Mode:        &OTLPModeGRPC,
 	}
 
 	propagateCommonDefaults(&baseFileDefaults.CommonSinkConfig, baseCommonSinkConfig)
@@ -528,6 +529,10 @@ func (c *Config) validateOTLPSinkConfig(otsc *OTLPSinkConfig) error {
 		return errors.New("compression must be 'gzip' or 'none'")
 	}
 
+	if *otsc.Mode != OTLPModeGRPC && *otsc.Mode != OTLPModeHTTP {
+		return errors.New("mode must be 'grpc' or 'http'")
+	}
+
 	return c.ValidateCommonSinkConfig(otsc.CommonSinkConfig)
 }
 

pkg/util/log/otlp_client.go

@@ -6,11 +6,16 @@
 package log
 
 import (
+	"bytes"
+	"compress/gzip"
 	"context"
+	"io"
+	"net/http"
 	"strings"
 	"sync"
 
 	"github.com/cockroachdb/cockroach/pkg/cli/exit"
+	"github.com/cockroachdb/cockroach/pkg/util/httputil"
 	"github.com/cockroachdb/cockroach/pkg/util/log/logconfig"
 	collpb "go.opentelemetry.io/proto/otlp/collector/logs/v1"
 	cpb "go.opentelemetry.io/proto/otlp/common/v1"
@@ -20,9 +25,10 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials/insecure"
-	"google.golang.org/grpc/encoding/gzip"
+	grpc_gzip "google.golang.org/grpc/encoding/gzip"
 	"google.golang.org/grpc/stats"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
 )
 
 const (
@@ -44,37 +50,21 @@ var otlpLogRecordPool = sync.Pool{
 	},
 }
 
+type otlpSinkClient interface {
+	Export(ctx context.Context, in *collpb.ExportLogsServiceRequest) (*collpb.ExportLogsServiceResponse, error)
+	Close() error
+}
+
 // OpenTelemetry log sink
 type otlpSink struct {
-	conn *grpc.ClientConn
-	lsc  collpb.LogsServiceClient
-
+	client otlpSinkClient
 	// requestObject should not be modified concurrently as it is reused
 	// between requests
 	requestObject *collpb.ExportLogsServiceRequest
 }
 
-var statsHandlerOption = &otlpStatsHandler{}
-
 func newOTLPSink(config logconfig.OTLPSinkConfig) (*otlpSink, error) {
-	dialOpts := []grpc.DialOption{
-		grpc.WithTransportCredentials(insecure.NewCredentials()),
-		grpc.WithStatsHandler(statsHandlerOption),
-	}
-
-	if *config.Compression == logconfig.GzipCompression {
-		dialOpts = append(dialOpts, grpc.WithDefaultCallOptions(grpc.UseCompressor(gzip.Name)))
-	}
-
-	conn, err := grpc.Dial(config.Address, dialOpts...)
-	if err != nil {
-		return nil, err
-	}
-
-	lsc := collpb.NewLogsServiceClient(conn)
 	sink := &otlpSink{
-		conn: conn,
-		lsc:  lsc,
 		requestObject: &collpb.ExportLogsServiceRequest{
 			ResourceLogs: []*lpb.ResourceLogs{
 				{
@@ -100,11 +90,16 @@ func newOTLPSink(config logconfig.OTLPSinkConfig) (*otlpSink, error) {
 		},
 	}
 
-	return sink, nil
-}
+	setClient := sink.setHTTPClient
+	if *config.Mode == logconfig.OTLPModeGRPC {
+		setClient = sink.setGRPCClient
+	}
+
+	if err := setClient(&config); err != nil {
+		return nil, err
+	}
 
-func (sink *otlpSink) isNotShutdown() bool {
-	return sink.conn.GetState() != connectivity.Shutdown
+	return sink, nil
 }
 
 func (sink *otlpSink) active() bool {
@@ -154,7 +149,7 @@ func (sink *otlpSink) output(b []byte, opts sinkOutputOptions) error {
 	sink.requestObject.ResourceLogs[0].InstrumentationLibraryLogs[0].Logs = records
 
 	// transmit the log over the network
-	_, err := sink.lsc.Export(ctx, sink.requestObject)
+	_, err := sink.client.Export(ctx, sink.requestObject)
 
 	// put the records back into the pool
 	for _, record := range records {
@@ -201,3 +196,125 @@ func (h *otlpStatsHandler) HandleRPC(ctx context.Context, rpcInfo stats.RPCStats
 }
 
 var _ stats.Handler = (*otlpStatsHandler)(nil)
+
+// client used when sink is using gRPC for exporting logs
+type otlpGRPCClient struct {
+	conn *grpc.ClientConn
+	lsc  collpb.LogsServiceClient
+}
+
+func (c *otlpGRPCClient) Close() error {
+	if c.conn.GetState() == connectivity.Shutdown {
+		return nil
+	}
+	// The reason for nolint:grpcconnclose is that we are not using *rpc.Context
+	// as it is primarily used for communication between crdb nodes, and doesn't
+	// fit this usecase.
+	return c.conn.Close() // nolint:grpcconnclose
+}
+
+func (c *otlpGRPCClient) Export(
+	ctx context.Context, in *collpb.ExportLogsServiceRequest,
+) (*collpb.ExportLogsServiceResponse, error) {
+	return c.lsc.Export(ctx, in)
+}
+
+var _ otlpSinkClient = (*otlpGRPCClient)(nil)
+
+var statsHandlerOption = &otlpStatsHandler{}
+
+func (sink *otlpSink) setGRPCClient(config *logconfig.OTLPSinkConfig) error {
+	dialOpts := []grpc.DialOption{
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithStatsHandler(statsHandlerOption),
+	}
+
+	if *config.Compression == logconfig.GzipCompression {
+		dialOpts = append(dialOpts, grpc.WithDefaultCallOptions(grpc.UseCompressor(grpc_gzip.Name)))
+	}
+
+	conn, err := grpc.Dial(config.Address, dialOpts...)
+	if err != nil {
+		return err
+	}
+	lsc := collpb.NewLogsServiceClient(conn)
+
+	sink.client = &otlpGRPCClient{
+		conn: conn,
+		lsc:  lsc,
+	}
+
+	return nil
+}
+
+// client used when sink is using HTTP for exporting logs
+type otlpHTTPClient struct {
+	client      *http.Client
+	request     *http.Request
+	compression string
+	gzipWriter  *gzip.Writer
+}
+
+func (c *otlpHTTPClient) Close() error {
+	return nil
+}
+
+func (c *otlpHTTPClient) Export(
+	ctx context.Context, in *collpb.ExportLogsServiceRequest,
+) (*collpb.ExportLogsServiceResponse, error) {
+	body, err := proto.Marshal(in)
+	if err != nil {
+		return nil, err
+	}
+
+	request := c.request.Clone(context.Background())
+	switch c.compression {
+	case logconfig.NoneCompression:
+		request.Body = io.NopCloser(bytes.NewReader(body))
+	case logconfig.GzipCompression:
+		// Content-Encoding header is set when the sink is initialized
+		// so no need to set it here
+		var buf bytes.Buffer
+		c.gzipWriter.Reset(&buf)
+		if _, err := c.gzipWriter.Write(body); err != nil {
+			return nil, err
+		}
+		if err := c.gzipWriter.Close(); err != nil {
+			return nil, err
+		}
+		request.Body = io.NopCloser(bytes.NewReader(buf.Bytes()))
+	}
+
+	resp, err := c.client.Do(request)
+	resp.Body.Close()
+	return nil, err
+}
+
+var _ otlpSinkClient = (*otlpHTTPClient)(nil)
+
+func (sink *otlpSink) setHTTPClient(config *logconfig.OTLPSinkConfig) error {
+	hc := &http.Client{
+		Transport: &http.Transport{
+			ForceAttemptHTTP2: true,
+		},
+	}
+	request, err := http.NewRequest(http.MethodPost, config.Address, http.NoBody)
+	if err != nil {
+		return err
+	}
+	request.Header.Set(httputil.ContentTypeHeader, httputil.ProtoContentType)
+
+	compression := *config.Compression
+	if compression == logconfig.GzipCompression {
+		request.Header.Set(httputil.ContentEncodingHeader, httputil.GzipEncoding)
+	}
+
+	sink.client = &otlpHTTPClient{
+		client:      hc,
+		request:     request,
+		compression: compression,
+		gzipWriter:  gzip.NewWriter(io.Discard),
+	}
+
+	return nil
+}