feat: emit metrics for command duration and errors (#826)

Author: steved

rueidisotel/README.md

@@ -11,6 +11,10 @@ Client side caching metrics:
 - `rueidis_do_cache_miss`: number of cache miss on client side
 - `rueidis_do_cache_hits`: number of cache hits on client side
 
+Client side commmand metrics:
+- `rueidis_command_duration_seconds`: histogram of command duration
+- `rueidis_command_errors`: number of command errors
+
 ```golang
 package main
 

rueidisotel/metrics.go

@@ -37,6 +37,14 @@ func WithMeterProvider(provider metric.MeterProvider) Option {
 	}
 }
 
+// WithOperationMetricAttr sets the operation name as an attribute for duration and error metrics.
+// This may cause memory usage to increase with the number of commands used.
+func WithOperationMetricAttr() Option {
+	return func(cli *otelclient) {
+		cli.commandMetrics.opAttr = true
+	}
+}
+
 type HistogramOption struct {
 	Buckets []float64
 }
@@ -149,6 +157,20 @@ func newClient(opts ...Option) (*otelclient, error) {
 	if err != nil {
 		return nil, err
 	}
+	cli.commandMetrics.addOpts = cli.addOpts
+	cli.commandMetrics.recordOpts = cli.recordOpts
+	cli.commandMetrics.duration, err = cli.meter.Float64Histogram(
+		"rueidis_command_duration_seconds",
+		metric.WithUnit("s"),
+		metric.WithExplicitBucketBoundaries(defaultHistogramBuckets...),
+	)
+	if err != nil {
+		return nil, err
+	}
+	cli.commandMetrics.errors, err = cli.meter.Int64Counter("rueidis_command_errors")
+	if err != nil {
+		return nil, err
+	}
 	return cli, nil
 }
 

rueidisotel/metrics_test.go

@@ -106,8 +106,14 @@ func TestNewClientMeterError(t *testing.T) {
 	tests := []struct {
 		name string
 	}{
-		{"rueidis_dial_attempt"}, {"rueidis_dial_success"}, {"rueidis_do_cache_miss"},
-		{"rueidis_do_cache_hits"}, {"rueidis_dial_conns"}, {"rueidis_dial_latency"},
+		{"rueidis_dial_attempt"},
+		{"rueidis_dial_success"},
+		{"rueidis_do_cache_miss"},
+		{"rueidis_do_cache_hits"},
+		{"rueidis_dial_conns"},
+		{"rueidis_dial_latency"},
+		{"rueidis_command_duration_seconds"},
+		{"rueidis_command_errors"},
 	}
 
 	for _, tt := range tests {
@@ -244,29 +250,29 @@ func TestTrackDialing(t *testing.T) {
 	})
 }
 
-func int64CountMetric(metrics metricdata.ResourceMetrics, name string) int64 {
+func findMetric(metrics metricdata.ResourceMetrics, name string) metricdata.Aggregation {
 	for _, sm := range metrics.ScopeMetrics {
 		for _, m := range sm.Metrics {
 			if m.Name == name {
-				data, ok := m.Data.(metricdata.Sum[int64])
-				if !ok {
-					return 0
-				}
-				return data.DataPoints[0].Value
+				return m.Data
 			}
 		}
 	}
+	return nil
+}
+
+func int64CountMetric(metrics metricdata.ResourceMetrics, name string) int64 {
+	m := findMetric(metrics, name)
+	if data, ok := m.(metricdata.Sum[int64]); ok {
+		return data.DataPoints[0].Value
+	}
 	return 0
 }
 
 func float64HistogramMetric(metrics metricdata.ResourceMetrics, name string) float64 {
-	for _, sm := range metrics.ScopeMetrics {
-		for _, m := range sm.Metrics {
-			if m.Name == name {
-				data := m.Data.(metricdata.Histogram[float64])
-				return data.DataPoints[0].Sum
-			}
-		}
+	m := findMetric(metrics, name)
+	if data, ok := m.(metricdata.Histogram[float64]); ok {
+		return data.DataPoints[0].Sum
 	}
 	return 0
 }

rueidisotel/trace.go

@@ -61,6 +61,32 @@ func WithDBStatement(f StatementFunc) Option {
 // StatementFunc is a the function that maps a command's tokens to a string to put in the db.statement attribute
 type StatementFunc func(cmdTokens []string) string
 
+type commandMetrics struct {
+	duration   metric.Float64Histogram
+	errors     metric.Int64Counter
+	addOpts    []metric.AddOption
+	recordOpts []metric.RecordOption
+	opAttr     bool
+}
+
+func (c *commandMetrics) recordDuration(ctx context.Context, op string, now time.Time) {
+	opts := c.recordOpts
+	if c.opAttr {
+		opts = append(c.recordOpts, metric.WithAttributeSet(attribute.NewSet(attribute.String("operation", op))))
+	}
+	c.duration.Record(ctx, time.Since(now).Seconds(), opts...)
+}
+
+func (c *commandMetrics) recordError(ctx context.Context, op string, err error) {
+	if err != nil && !rueidis.IsRedisNil(err) {
+		opts := c.addOpts
+		if c.opAttr {
+			opts = append(c.addOpts, metric.WithAttributeSet(attribute.NewSet(attribute.String("operation", op))))
+		}
+		c.errors.Add(ctx, 1, opts...)
+	}
+}
+
 type otelclient struct {
 	client          rueidis.Client
 	meterProvider   metric.MeterProvider
@@ -74,50 +100,66 @@ type otelclient struct {
 	addOpts         []metric.AddOption
 	recordOpts      []metric.RecordOption
 	histogramOption HistogramOption
+	commandMetrics
 }
 
 func (o *otelclient) B() rueidis.Builder {
 	return o.client.B()
 }
 
 func (o *otelclient) Do(ctx context.Context, cmd rueidis.Completed) (resp rueidis.RedisResult) {
-	ctx, span := o.start(ctx, first(cmd.Commands()), sum(cmd.Commands()))
+	op := first(cmd.Commands())
+	defer o.recordDuration(ctx, op, time.Now())
+	ctx, span := o.start(ctx, op, sum(cmd.Commands()))
 	if o.dbStmtFunc != nil {
 		span.SetAttributes(dbstmt.String(o.dbStmtFunc(cmd.Commands())))
 	}
 
 	resp = o.client.Do(ctx, cmd)
 	o.end(span, resp.Error())
+	o.recordError(ctx, op, resp.Error())
 	return
 }
 
 func (o *otelclient) DoMulti(ctx context.Context, multi ...rueidis.Completed) (resp []rueidis.RedisResult) {
-	ctx, span := o.start(ctx, multiFirst(multi), multiSum(multi))
+	op := multiFirst(multi)
+	defer o.recordDuration(ctx, op, time.Now())
+	ctx, span := o.start(ctx, op, multiSum(multi))
 	resp = o.client.DoMulti(ctx, multi...)
-	o.end(span, firstError(resp))
+	err := firstError(resp)
+	o.end(span, err)
+	o.recordError(ctx, op, err)
 	return
 }
 
 func (o *otelclient) DoStream(ctx context.Context, cmd rueidis.Completed) (resp rueidis.RedisResultStream) {
-	ctx, span := o.start(ctx, first(cmd.Commands()), sum(cmd.Commands()))
+	op := first(cmd.Commands())
+	defer o.recordDuration(ctx, op, time.Now())
+	ctx, span := o.start(ctx, op, sum(cmd.Commands()))
 	if o.dbStmtFunc != nil {
 		span.SetAttributes(dbstmt.String(o.dbStmtFunc(cmd.Commands())))
 	}
 
 	resp = o.client.DoStream(ctx, cmd)
 	o.end(span, resp.Error())
+	o.recordError(ctx, op, resp.Error())
 	return
 }
 
 func (o *otelclient) DoMultiStream(ctx context.Context, multi ...rueidis.Completed) (resp rueidis.MultiRedisResultStream) {
-	ctx, span := o.start(ctx, multiFirst(multi), multiSum(multi))
+	op := multiFirst(multi)
+	defer o.recordDuration(ctx, op, time.Now())
+	ctx, span := o.start(ctx, op, multiSum(multi))
 	resp = o.client.DoMultiStream(ctx, multi...)
 	o.end(span, resp.Error())
+	o.recordError(ctx, op, resp.Error())
 	return
 }
 
 func (o *otelclient) DoCache(ctx context.Context, cmd rueidis.Cacheable, ttl time.Duration) (resp rueidis.RedisResult) {
-	ctx, span := o.start(ctx, first(cmd.Commands()), sum(cmd.Commands()))
+	op := first(cmd.Commands())
+	defer o.recordDuration(ctx, op, time.Now())
+	ctx, span := o.start(ctx, op, sum(cmd.Commands()))
 	if o.dbStmtFunc != nil {
 		span.SetAttributes(dbstmt.String(o.dbStmtFunc(cmd.Commands())))
 	}
@@ -131,11 +173,14 @@ func (o *otelclient) DoCache(ctx context.Context, cmd rueidis.Cacheable, ttl tim
 		}
 	}
 	o.end(span, resp.Error())
+	o.recordError(ctx, op, resp.Error())
 	return
 }
 
 func (o *otelclient) DoMultiCache(ctx context.Context, multi ...rueidis.CacheableTTL) (resps []rueidis.RedisResult) {
-	ctx, span := o.start(ctx, multiCacheableFirst(multi), multiCacheableSum(multi))
+	op := multiCacheableFirst(multi)
+	defer o.recordDuration(ctx, op, time.Now())
+	ctx, span := o.start(ctx, op, multiCacheableSum(multi))
 	resps = o.client.DoMultiCache(ctx, multi...)
 	for _, resp := range resps {
 		if resp.NonRedisError() == nil {
@@ -146,28 +191,32 @@ func (o *otelclient) DoMultiCache(ctx context.Context, multi ...rueidis.Cacheabl
 			}
 		}
 	}
-	o.end(span, firstError(resps))
+	err := firstError(resps)
+	o.end(span, err)
+	o.recordError(ctx, op, err)
 	return
 }
 
 func (o *otelclient) Dedicated(fn func(rueidis.DedicatedClient) error) (err error) {
 	return o.client.Dedicated(func(client rueidis.DedicatedClient) error {
 		return fn(&dedicated{
-			client:     client,
-			tAttrs:     o.tAttrs,
-			tracer:     o.tracer,
-			dbStmtFunc: o.dbStmtFunc,
+			client:         client,
+			tAttrs:         o.tAttrs,
+			tracer:         o.tracer,
+			dbStmtFunc:     o.dbStmtFunc,
+			commandMetrics: o.commandMetrics,
 		})
 	})
 }
 
 func (o *otelclient) Dedicate() (rueidis.DedicatedClient, func()) {
 	client, cancel := o.client.Dedicate()
 	return &dedicated{
-		client:     client,
-		tAttrs:     o.tAttrs,
-		tracer:     o.tracer,
-		dbStmtFunc: o.dbStmtFunc,
+		client:         client,
+		tAttrs:         o.tAttrs,
+		tracer:         o.tracer,
+		dbStmtFunc:     o.dbStmtFunc,
+		commandMetrics: o.commandMetrics,
 	}, cancel
 }
 
@@ -198,6 +247,7 @@ func (o *otelclient) Nodes() map[string]rueidis.Client {
 			tAttrs:          o.tAttrs,
 			histogramOption: o.histogramOption,
 			dbStmtFunc:      o.dbStmtFunc,
+			commandMetrics:  o.commandMetrics,
 		}
 	}
 	return nodes
@@ -218,27 +268,35 @@ type dedicated struct {
 	tracer     trace.Tracer
 	tAttrs     trace.SpanStartEventOption
 	dbStmtFunc StatementFunc
+	commandMetrics
 }
 
 func (d *dedicated) B() rueidis.Builder {
 	return d.client.B()
 }
 
 func (d *dedicated) Do(ctx context.Context, cmd rueidis.Completed) (resp rueidis.RedisResult) {
-	ctx, span := d.start(ctx, first(cmd.Commands()), sum(cmd.Commands()))
+	op := first(cmd.Commands())
+	defer d.recordDuration(ctx, op, time.Now())
+	ctx, span := d.start(ctx, op, sum(cmd.Commands()))
 	if d.dbStmtFunc != nil {
 		span.SetAttributes(dbstmt.String(d.dbStmtFunc(cmd.Commands())))
 	}
 
 	resp = d.client.Do(ctx, cmd)
 	d.end(span, resp.Error())
+	d.recordError(ctx, op, resp.Error())
 	return
 }
 
 func (d *dedicated) DoMulti(ctx context.Context, multi ...rueidis.Completed) (resp []rueidis.RedisResult) {
-	ctx, span := d.start(ctx, multiFirst(multi), multiSum(multi))
+	op := multiFirst(multi)
+	defer d.recordDuration(ctx, op, time.Now())
+	ctx, span := d.start(ctx, op, multiSum(multi))
 	resp = d.client.DoMulti(ctx, multi...)
-	d.end(span, firstError(resp))
+	err := firstError(resp)
+	d.end(span, err)
+	d.recordError(ctx, op, err)
 	return
 }