Merge #148525

148525: pkg/util/log: add otlp log sink r=TheComputerM a=TheComputerM

OpenTelemetry is now an industry standard for o11y and CRDB
should support its logging protocol so that it works
seamlessly with OTEL compatible tooling.

This commit implements the log sink functionality, it uses
gRPC to transmit logs in the OTEL-standard protobuf format.
This sink can be used to ship logs on OTEL compatible targets
like otel-collector, datadog-agent or even directly to
platforms like datadog and Loki.

Parsing the config for the otlp log sink was done in
a previous commit.

Resolves: #143049

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

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

Author: craig[bot]

docs/generated/logsinks.md

@@ -289,7 +289,6 @@ 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. |
-| `insecure` | Disables transport security for the underlying gRPC connection.  Inherited from `otlp-defaults.insecure` 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

@@ -57,8 +57,7 @@ func TestSetupLogging(t *testing.T) {
 		`flush-trigger-size: 1.0MiB, ` +
 		`max-buffer-size: 50MiB, ` +
 		`format: newline}}`
-	const defaultOtlpConfig = `otlp-defaults: {` +
-		`insecure: false, ` +
+	const defaultOTLPConfig = `otlp-defaults: {` +
 		`compression: gzip, ` +
 		`filter: INFO, ` +
 		`format: json, ` +
@@ -200,7 +199,7 @@ func TestSetupLogging(t *testing.T) {
 		// Shorten the configuration for legibility during reviews of test changes.
 		actual = strings.ReplaceAll(actual, defaultFluentConfig, "<fluentDefaults>")
 		actual = strings.ReplaceAll(actual, defaultHTTPConfig, "<httpDefaults>")
-		actual = strings.ReplaceAll(actual, defaultOtlpConfig, "<otlpDefaults>")
+		actual = strings.ReplaceAll(actual, defaultOTLPConfig, "<otlpDefaults>")
 		actual = stdFileDefaultsRe.ReplaceAllString(actual, "<stdFileDefaults($path)>")
 		actual = fileDefaultsNoMaxSizeRe.ReplaceAllString(actual, "<fileDefaultsNoMaxSize($path)>")
 		actual = strings.ReplaceAll(actual, fileDefaultsNoDir, "<fileDefaultsNoDir>")

pkg/testutils/lint/lint_test.go

@@ -1242,6 +1242,7 @@ func TestLint(t *testing.T) {
 			":!rpc/context.go",
 			":!rpc/nodedialer/nodedialer_test.go",
 			":!util/grpcutil/grpc_util_test.go",
+			":!util/log/otlp_client_test.go",
 			":!server/server_obs_service.go",
 			":!server/testserver.go",
 			":!util/tracing/*_test.go",

pkg/util/log/BUILD.bazel

@@ -34,6 +34,7 @@ go_library(
         "log_entry.go",
         "log_flush.go",
         "metric.go",
+        "otlp_client.go",
         "redact.go",
         "registry.go",
         "report.go",
@@ -86,6 +87,16 @@ go_library(
         "@com_github_cockroachdb_redact//interfaces",
         "@com_github_cockroachdb_ttycolor//:ttycolor",
         "@com_github_petermattis_goid//:goid",
+        "@io_opentelemetry_go_proto_otlp//collector/logs/v1:logs",
+        "@io_opentelemetry_go_proto_otlp//common/v1:common",
+        "@io_opentelemetry_go_proto_otlp//logs/v1:logs",
+        "@io_opentelemetry_go_proto_otlp//resource/v1:resource",
+        "@org_golang_google_grpc//:grpc",
+        "@org_golang_google_grpc//codes",
+        "@org_golang_google_grpc//connectivity",
+        "@org_golang_google_grpc//credentials/insecure",
+        "@org_golang_google_grpc//encoding/gzip",
+        "@org_golang_google_grpc//status",
     ] + select({
         "@io_bazel_rules_go//go/platform:aix": [
             "@org_golang_x_sys//unix",
@@ -164,6 +175,7 @@ go_test(
         "intercept_test.go",
         "log_decoder_test.go",
         "main_test.go",
+        "otlp_client_test.go",
         "redact_test.go",
         "registry_test.go",
         "secondary_log_test.go",
@@ -202,6 +214,9 @@ go_test(
         "@com_github_pmezard_go_difflib//difflib",
         "@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_x_sys//unix",
     ],
 )

pkg/util/log/flags.go

@@ -119,6 +119,21 @@ func ApplyConfig(
 	fd2CaptureCleanupFn := func() {}
 
 	closer := newBufferedSinkCloser()
+
+	// 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
+					fmt.Fprintf(OrigStderr, "# OTLP Sink Cleanup Warning: %s\n", err.Error())
+				}
+			}
+		}
+	}
+
 	// logShutdownFn is the returned cleanup function, whose purpose
 	// is to tear down the work we are doing here.
 	logShutdownFn = func() {
@@ -127,6 +142,7 @@ func ApplyConfig(
 		logging.setChannelLoggers(make(map[Channel]*loggerT), &si)
 		fd2CaptureCleanupFn()
 		secLoggersCancel()
+		closeOTLPSinks()
 		if err := closer.Close(defaultCloserTimeout); err != nil {
 			fmt.Printf("# WARNING: %s\n", err.Error())
 		}
@@ -367,16 +383,16 @@ func ApplyConfig(
 	}
 
 	// Create the OpenTelemetry sinks.
-	for _, fc := range config.Sinks.OtlpServers {
+	for _, fc := range config.Sinks.OTLPServers {
 		if fc.Filter == severity.NONE {
 			continue
 		}
-		optlSinkInfo, err := newOtlpSinkInfo(*fc)
+		otplSinkInfo, err := newOTLPSinkInfo(*fc)
 		if err != nil {
 			return nil, err
 		}
-		attachBufferWrapper(optlSinkInfo, fc.CommonSinkConfig.Buffering, closer)
-		attachSinkInfo(optlSinkInfo, &fc.Channels)
+		attachBufferWrapper(otplSinkInfo, fc.CommonSinkConfig.Buffering, closer)
+		attachSinkInfo(otplSinkInfo, &fc.Channels)
 	}
 
 	// Prepend the interceptor sink to all channels.
@@ -446,9 +462,20 @@ func newHTTPSinkInfo(c logconfig.HTTPSinkConfig) (*sinkInfo, error) {
 	return info, nil
 }
 
-func newOtlpSinkInfo(_ logconfig.OtlpSinkConfig) (*sinkInfo, error) {
-	// TODO(mudit): Implement newOtlpSink
-	return nil, nil
+func newOTLPSinkInfo(c logconfig.OTLPSinkConfig) (*sinkInfo, error) {
+	info := &sinkInfo{}
+
+	if err := info.applyConfig(c.CommonSinkConfig); err != nil {
+		return nil, err
+	}
+	info.applyFilters(c.Channels)
+
+	otlpSink, err := newOTLPSink(c)
+	if err != nil {
+		return nil, err
+	}
+	info.sink = otlpSink
+	return info, nil
 }
 
 // applyFilters applies the channel filters to a sinkInfo.

pkg/util/log/logconfig/config.go

@@ -35,9 +35,9 @@ const DefaultFluentFormat = `json-fluent-compact`
 // when not specified in a configuration.
 const DefaultHTTPFormat = `json-compact`
 
-// DefaultOtlpFormat is the entry format for OpenTelemetry sinks
+// DefaultOTLPFormat is the entry format for OpenTelemetry sinks
 // when not specified in a configuration.
-const DefaultOtlpFormat = `json`
+const DefaultOTLPFormat = `json`
 
 // DefaultFilePerms is the default permissions used in file-defaults. It
 // is applied literally via os.Chmod, without considering the umask.
@@ -78,7 +78,7 @@ http-defaults:
       max-buffer-size: 50mib
 otlp-defaults:
     filter: INFO
-    format: ` + DefaultOtlpFormat + `
+    format: ` + DefaultOTLPFormat + `
     redactable: true
     exit-on-error: false
     buffering:
@@ -138,10 +138,10 @@ type Config struct {
 	// configuration value.
 	HTTPDefaults HTTPDefaults `yaml:"http-defaults,omitempty"`
 
-	// OtlpDefaults represents the default configuration for OpenTelemetry sinks,
+	// OTLPDefaults represents the default configuration for OpenTelemetry sinks,
 	// inherited when a specific OpenTelemetry sink config does not provide a
 	// configuration value.
-	OtlpDefaults OtlpDefaults `yaml:"otlp-defaults,omitempty"`
+	OTLPDefaults OTLPDefaults `yaml:"otlp-defaults,omitempty"`
 
 	// Sinks represents the sink configurations.
 	Sinks SinkConfig `yaml:",omitempty"`
@@ -267,8 +267,8 @@ type SinkConfig struct {
 	FluentServers map[string]*FluentSinkConfig `yaml:"fluent-servers,omitempty"`
 	// HTTPServers represents the list of configured http sinks.
 	HTTPServers map[string]*HTTPSinkConfig `yaml:"http-servers,omitempty"`
-	// OtlpServers represents the list of configured opentelemetry sinks.
-	OtlpServers map[string]*OtlpSinkConfig `yaml:"otlp-servers,omitempty"`
+	// OTLPServers represents the list of configured opentelemetry sinks.
+	OTLPServers map[string]*OTLPSinkConfig `yaml:"otlp-servers,omitempty"`
 	// Stderr represents the configuration for the stderr sink.
 	Stderr StderrSinkConfig `yaml:",omitempty"`
 }
@@ -589,18 +589,15 @@ type HTTPSinkConfig struct {
 	sinkName string
 }
 
-type OtlpDefaults struct {
-	// Disables transport security for the underlying gRPC connection.
-	Insecure *bool `yaml:",omitempty"`
-
+type OTLPDefaults struct {
 	// Compression can be "none" or "gzip" to enable gzip compression.
 	// Set to "gzip" by default.
 	Compression *string `yaml:",omitempty"`
 
 	CommonSinkConfig `yaml:",inline"`
 }
 
-// OtlpSinkConfig represents the configuration for one OpenTelemetry sink.
+// OTLPSinkConfig represents the configuration for one OpenTelemetry sink.
 //
 // User-facing documentation follows.
 // TITLE: Output to OpenTelemetry compatible collectors.
@@ -637,7 +634,7 @@ type OtlpDefaults struct {
 // {{site.data.alerts.callout_info}}
 // Run `cockroach debug check-log-config` to verify the effect of defaults inheritance.
 // {{site.data.alerts.end}}
-type OtlpSinkConfig struct {
+type OTLPSinkConfig struct {
 	// Channels is the list of logging channels that use this sink.
 	Channels ChannelFilters `yaml:",omitempty,flow"`
 
@@ -646,7 +643,7 @@ type OtlpSinkConfig struct {
 	// The host/address and port parts are separated with a colon.
 	Address string `yaml:""`
 
-	OtlpDefaults `yaml:",inline"`
+	OTLPDefaults `yaml:",inline"`
 
 	// SinkName is populated during validation.
 	SinkName string

pkg/util/log/logconfig/validate.go

@@ -111,9 +111,9 @@ func (c *Config) Validate(defaultLogDir *string) (resErr error) {
 		}(),
 		Compression: &GzipCompression,
 	}
-	baseOtlpDefaults := OtlpDefaults{
+	baseOTLPDefaults := OTLPDefaults{
 		CommonSinkConfig: CommonSinkConfig{
-			Format: func() *string { s := DefaultOtlpFormat; return &s }(),
+			Format: func() *string { s := DefaultOTLPFormat; return &s }(),
 			Buffering: CommonBufferSinkConfigWrapper{
 				CommonBufferSinkConfig: CommonBufferSinkConfig{
 					MaxStaleness:     &defaultBufferedStaleness,
@@ -123,19 +123,18 @@ func (c *Config) Validate(defaultLogDir *string) (resErr error) {
 				},
 			},
 		},
-		Insecure:    &bf,
 		Compression: &GzipCompression,
 	}
 
 	propagateCommonDefaults(&baseFileDefaults.CommonSinkConfig, baseCommonSinkConfig)
 	propagateCommonDefaults(&baseFluentDefaults.CommonSinkConfig, baseCommonSinkConfig)
 	propagateCommonDefaults(&baseHTTPDefaults.CommonSinkConfig, baseCommonSinkConfig)
-	propagateCommonDefaults(&baseOtlpDefaults.CommonSinkConfig, baseCommonSinkConfig)
+	propagateCommonDefaults(&baseOTLPDefaults.CommonSinkConfig, baseCommonSinkConfig)
 
 	propagateFileDefaults(&c.FileDefaults, baseFileDefaults)
 	propagateFluentDefaults(&c.FluentDefaults, baseFluentDefaults)
 	propagateHTTPDefaults(&c.HTTPDefaults, baseHTTPDefaults)
-	propagateOtlpDefaults(&c.OtlpDefaults, baseOtlpDefaults)
+	propagateOTLPDefaults(&c.OTLPDefaults, baseOTLPDefaults)
 
 	// Normalize the directory.
 	if err := normalizeDir(&c.FileDefaults.Dir); err != nil {
@@ -177,13 +176,13 @@ func (c *Config) Validate(defaultLogDir *string) (resErr error) {
 		}
 	}
 
-	for sinkName, fc := range c.Sinks.OtlpServers {
+	for sinkName, fc := range c.Sinks.OTLPServers {
 		if fc == nil {
-			fc = &OtlpSinkConfig{Channels: SelectChannels()}
-			c.Sinks.OtlpServers[sinkName] = fc
+			fc = &OTLPSinkConfig{Channels: SelectChannels()}
+			c.Sinks.OTLPServers[sinkName] = fc
 		}
 		fc.SinkName = sinkName
-		if err := c.validateOtlpSinkConfig(fc); err != nil {
+		if err := c.validateOTLPSinkConfig(fc); err != nil {
 			fmt.Fprintf(&errBuf, "otlp server %q: %v\n", sinkName, err)
 		}
 	}
@@ -273,7 +272,7 @@ func (c *Config) Validate(defaultLogDir *string) (resErr error) {
 		}
 	}
 
-	for serverName, fc := range c.Sinks.OtlpServers {
+	for serverName, fc := range c.Sinks.OTLPServers {
 		if len(fc.Channels.Filters) == 0 {
 			fmt.Fprintf(&errBuf, "otlp server %q: no channel selected\n", serverName)
 		}
@@ -518,8 +517,8 @@ func (c *Config) validateHTTPSinkConfig(hsc *HTTPSinkConfig) error {
 	return c.ValidateCommonSinkConfig(hsc.CommonSinkConfig)
 }
 
-func (c *Config) validateOtlpSinkConfig(otsc *OtlpSinkConfig) error {
-	propagateOtlpDefaults(&otsc.OtlpDefaults, c.OtlpDefaults)
+func (c *Config) validateOTLPSinkConfig(otsc *OTLPSinkConfig) error {
+	propagateOTLPDefaults(&otsc.OTLPDefaults, c.OTLPDefaults)
 	otsc.Address = strings.TrimSpace(otsc.Address)
 	if len(otsc.Address) == 0 {
 		return errors.New("address cannot be empty")
@@ -566,7 +565,7 @@ func propagateHTTPDefaults(target *HTTPDefaults, source HTTPDefaults) {
 	propagateDefaults(target, source)
 }
 
-func propagateOtlpDefaults(target *OtlpDefaults, source OtlpDefaults) {
+func propagateOTLPDefaults(target *OTLPDefaults, source OTLPDefaults) {
 	propagateDefaults(target, source)
 }
 

pkg/util/log/logconfig/validate_test.go

@@ -58,7 +58,7 @@ func clearExpectedValues(c *Config) {
 	c.FileDefaults = FileDefaults{}
 	c.FluentDefaults = FluentDefaults{}
 	c.HTTPDefaults = HTTPDefaults{}
-	c.OtlpDefaults = OtlpDefaults{}
+	c.OTLPDefaults = OTLPDefaults{}
 
 	for _, f := range c.Sinks.FileGroups {
 		if *f.Dir == "/default-dir" {