Merge #150622

150622: pkg/util/log: add metrics to otlp sink r=TheComputerM a=TheComputerM

adds metrics to monitor internal statistics of otlp sinks, similar to
the metrics exposed for fluent sinks.

`log.otlp.sink.write.attempts`: Number of write attempts experienced by
otlp-server logging sinks

`log.otlp.sink.write.errors`: Number of write errors experienced by
otlp-server logging sinks

`log.otlp.sink.grpc.transparent_retries`: Number of transparent retries
done by otlp-server logging sinks when using GRPC

Epic: none
Release note: none

---

**Happy case**

<img width="1133" height="238" alt="image" src="https://github.com/user-attachments/assets/1c03c159-e88e-42e1-9e09-e4a66a5f2fe1" />

**When OTLP receiver is down**

<img width="1133" height="238" alt="image" src="https://github.com/user-attachments/assets/3cc59391-f6be-4452-b4df-aff2606b34c8" />

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

Author: craig[bot]

docs/generated/metrics/metrics.yaml

@@ -10127,6 +10127,30 @@ layers:
       unit: COUNT
       aggregation: AVG
       derivative: NON_NEGATIVE_DERIVATIVE
+    - name: log.otlp.sink.grpc.transparent_retries
+      exported_name: log_otlp_sink_grpc_transparent_retries
+      description: Number of transparent retries done by otlp-server logging sinks when using GRPC
+      y_axis_label: Retries
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
+    - name: log.otlp.sink.write.attempts
+      exported_name: log_otlp_sink_write_attempts
+      description: Number of write attempts experienced by otlp-server logging sinks
+      y_axis_label: Attempts
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
+    - name: log.otlp.sink.write.errors
+      exported_name: log_otlp_sink_write_errors
+      description: Number of write errors experienced by otlp-server logging sinks
+      y_axis_label: Errors
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
     - name: sys.cgo.allocbytes
       exported_name: sys_cgo_allocbytes
       description: Current bytes of memory allocated by cgo

pkg/util/log/BUILD.bazel

@@ -96,6 +96,7 @@ go_library(
         "@org_golang_google_grpc//connectivity",
         "@org_golang_google_grpc//credentials/insecure",
         "@org_golang_google_grpc//encoding/gzip",
+        "@org_golang_google_grpc//stats",
         "@org_golang_google_grpc//status",
     ] + select({
         "@io_bazel_rules_go//go/platform:aix": [

pkg/util/log/logmetrics/metrics.go

@@ -43,6 +43,27 @@ var (
 		Unit:        metric.Unit_COUNT,
 		MetricType:  io_prometheus_client.MetricType_COUNTER,
 	}
+	otlpSinkWriteAttempts = metric.Metadata{
+		Name:        "log.otlp.sink.write.attempts",
+		Help:        "Number of write attempts experienced by otlp-server logging sinks",
+		Measurement: "Attempts",
+		Unit:        metric.Unit_COUNT,
+		MetricType:  io_prometheus_client.MetricType_COUNTER,
+	}
+	otlpSinkWriteErrors = metric.Metadata{
+		Name:        "log.otlp.sink.write.errors",
+		Help:        "Number of write errors experienced by otlp-server logging sinks",
+		Measurement: "Errors",
+		Unit:        metric.Unit_COUNT,
+		MetricType:  io_prometheus_client.MetricType_COUNTER,
+	}
+	otlpSinkGRPCTransparentRetries = metric.Metadata{
+		Name:        "log.otlp.sink.grpc.transparent_retries",
+		Help:        "Number of transparent retries done by otlp-server logging sinks when using GRPC",
+		Measurement: "Retries",
+		Unit:        metric.Unit_COUNT,
+		MetricType:  io_prometheus_client.MetricType_COUNTER,
+	}
 	bufferedSinkMessagesDropped = metric.Metadata{
 		Name:        "log.buffered.messages.dropped",
 		Help:        "Count of log messages that are dropped by buffered log sinks. When CRDB attempts to buffer a log message in a buffered log sink whose buffer is already full, it drops the oldest buffered messages to make space for the new message",
@@ -87,12 +108,17 @@ var _ log.LogMetrics = (*logMetricsRegistry)(nil)
 func newLogMetricsRegistry() *logMetricsRegistry {
 	return &logMetricsRegistry{
 		counters: []*metric.Counter{
+			log.BufferedSinkMessagesDropped: metric.NewCounter(bufferedSinkMessagesDropped),
+			log.LogMessageCount:             metric.NewCounter(logMessageCount),
+			// fluent sink metrics
 			log.FluentSinkConnectionAttempt: metric.NewCounter(fluentSinkConnAttempts),
 			log.FluentSinkConnectionError:   metric.NewCounter(fluentSinkConnErrors),
 			log.FluentSinkWriteAttempt:      metric.NewCounter(fluentSinkWriteAttempts),
 			log.FluentSinkWriteError:        metric.NewCounter(fluentSinkWriteErrors),
-			log.BufferedSinkMessagesDropped: metric.NewCounter(bufferedSinkMessagesDropped),
-			log.LogMessageCount:             metric.NewCounter(logMessageCount),
+			// otlp sink metrics
+			log.OTLPSinkWriteAttempt:           metric.NewCounter(otlpSinkWriteAttempts),
+			log.OTLPSinkWriteError:             metric.NewCounter(otlpSinkWriteErrors),
+			log.OTLPSinkGRPCTransparentRetries: metric.NewCounter(otlpSinkGRPCTransparentRetries),
 		},
 	}
 }

pkg/util/log/metric.go

@@ -33,10 +33,15 @@ type LogMetrics interface {
 type Metric int
 
 const (
-	FluentSinkConnectionAttempt Metric = iota
+	BufferedSinkMessagesDropped Metric = iota
+	LogMessageCount
+	// fluent sink metrics
+	FluentSinkConnectionAttempt
 	FluentSinkConnectionError
 	FluentSinkWriteAttempt
 	FluentSinkWriteError
-	BufferedSinkMessagesDropped
-	LogMessageCount
+	// otlp sink metrics
+	OTLPSinkWriteAttempt
+	OTLPSinkWriteError
+	OTLPSinkGRPCTransparentRetries
 )

pkg/util/log/otlp_client.go

@@ -21,6 +21,7 @@ import (
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/encoding/gzip"
+	"google.golang.org/grpc/stats"
 	"google.golang.org/grpc/status"
 )
 
@@ -53,9 +54,12 @@ type otlpSink struct {
 	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 {
@@ -143,6 +147,7 @@ func otlpExtractRecords(b []byte) []*lpb.LogRecord {
 }
 
 func (sink *otlpSink) output(b []byte, opts sinkOutputOptions) error {
+	logging.metrics.IncrementCounter(OTLPSinkWriteAttempt, 1)
 	ctx := context.Background()
 
 	records := otlpExtractRecords(b)
@@ -161,5 +166,38 @@ func (sink *otlpSink) output(b []byte, opts sinkOutputOptions) error {
 		return nil
 	}
 
-	return err
+	if err != nil {
+		logging.metrics.IncrementCounter(OTLPSinkWriteError, 1)
+		return err
+	}
+
+	return nil
 }
+
+// otlpStatsHandler implements the stats.Handler interface to and is passed as
+// a dial option to the grpc client in the otlp log sink to get grpc metrics.
+type otlpStatsHandler struct{}
+
+// TagRPC exists to satisfy the stats.Handler interface.
+func (h *otlpStatsHandler) TagRPC(ctx context.Context, info *stats.RPCTagInfo) context.Context {
+	return ctx
+}
+
+// TagConn exists to satisfy the stats.Handler interface.
+func (h *otlpStatsHandler) TagConn(ctx context.Context, info *stats.ConnTagInfo) context.Context {
+	return ctx
+}
+
+// HandleConn exists to satisfy the stats.Handler interface.
+func (h *otlpStatsHandler) HandleConn(ctx context.Context, connInfo stats.ConnStats) {}
+
+func (h *otlpStatsHandler) HandleRPC(ctx context.Context, rpcInfo stats.RPCStats) {
+	switch st := rpcInfo.(type) {
+	case *stats.Begin:
+		if st.IsTransparentRetryAttempt {
+			logging.metrics.IncrementCounter(OTLPSinkGRPCTransparentRetries, 1)
+		}
+	}
+}
+
+var _ stats.Handler = (*otlpStatsHandler)(nil)