WIP: Hotfix/fix exemplars for counters

client/amqp/v2/amqp.go

@@ -12,7 +12,7 @@ import (
 	patronerrors "github.com/beatlabs/patron/errors"
 	"github.com/beatlabs/patron/log"
 	"github.com/beatlabs/patron/trace"
-	opentracing "github.com/opentracing/opentracing-go"
+	"github.com/opentracing/opentracing-go"
 	"github.com/opentracing/opentracing-go/ext"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/streadway/amqp"
@@ -100,7 +100,7 @@ func (tc *Publisher) Publish(ctx context.Context, exchange, key string, mandator
 	start := time.Now()
 	err := tc.channel.Publish(exchange, key, mandatory, immediate, msg)
 
-	observePublish(sp, start, exchange, err)
+	observePublish(ctx, sp, start, exchange, err)
 	if err != nil {
 		return fmt.Errorf("failed to publish message: %w", err)
 	}
@@ -120,7 +120,11 @@ func (c amqpHeadersCarrier) Set(key, val string) {
 	c[key] = val
 }
 
-func observePublish(span opentracing.Span, start time.Time, exchange string, err error) {
+func observePublish(ctx context.Context, span opentracing.Span, start time.Time, exchange string, err error) {
 	trace.SpanComplete(span, err)
-	publishDurationMetrics.WithLabelValues(exchange, strconv.FormatBool(err != nil)).Observe(time.Since(start).Seconds())
+
+	durationHistogram := trace.Histogram{
+		Observer: publishDurationMetrics.WithLabelValues(exchange, strconv.FormatBool(err == nil)),
+	}
+	durationHistogram.Observe(ctx, time.Since(start).Seconds())
 }

client/es/elasticsearch.go

@@ -17,7 +17,7 @@ import (
 	elasticsearch "github.com/elastic/go-elasticsearch/v8"
 	"github.com/elastic/go-elasticsearch/v8/esapi"
 	"github.com/elastic/go-elasticsearch/v8/estransport"
-	opentracing "github.com/opentracing/opentracing-go"
+	"github.com/opentracing/opentracing-go"
 	"github.com/opentracing/opentracing-go/ext"
 	"github.com/prometheus/client_golang/prometheus"
 )
@@ -120,8 +120,10 @@ func (c *transportClient) Perform(req *http.Request) (*http.Response, error) {
 
 func observeResponse(req *http.Request, sp opentracing.Span, rsp *http.Response, start time.Time) {
 	endSpan(sp, rsp)
-	reqDurationMetrics.WithLabelValues(req.Method, req.URL.Host, strconv.Itoa(rsp.StatusCode)).
-		Observe(time.Since(start).Seconds())
+	durationHistogram := trace.Histogram{
+		Observer: reqDurationMetrics.WithLabelValues(req.Method, req.URL.Host, strconv.Itoa(rsp.StatusCode)),
+	}
+	durationHistogram.Observe(req.Context(), time.Since(start).Seconds())
 }
 
 // Config is a wrapper for elasticsearch.Config

client/grpc/grpc.go

@@ -86,9 +86,11 @@ func unaryInterceptor(target string) grpc.UnaryClientInterceptor {
 		invokeDuration := time.Since(invokeTime)
 
 		rpcStatus, _ := status.FromError(err) // codes.OK if err == nil, codes.Unknown if !ok
-		rpcDurationMetrics.
-			WithLabelValues(unary, target, method, rpcStatus.Code().String()).
-			Observe(invokeDuration.Seconds())
+
+		durationHistogram := trace.Histogram{
+			Observer: rpcDurationMetrics.WithLabelValues(unary, target, method, rpcStatus.Code().String()),
+		}
+		durationHistogram.Observe(ctx, invokeDuration.Seconds())
 
 		if err != nil {
 			trace.SpanError(span)

client/http/http.go

@@ -93,9 +93,10 @@ func (tc *TracedClient) Do(req *http.Request) (*http.Response, error) {
 	}
 
 	ext.HTTPStatusCode.Set(ht.Span(), uint16(rsp.StatusCode))
-	reqDurationMetrics.
-		WithLabelValues(req.Method, req.URL.Host, strconv.Itoa(rsp.StatusCode)).
-		Observe(time.Since(start).Seconds())
+	durationHistogram := trace.Histogram{
+		Observer: reqDurationMetrics.WithLabelValues(req.Method, req.URL.Host, strconv.Itoa(rsp.StatusCode)),
+	}
+	durationHistogram.Observe(req.Context(), time.Since(start).Seconds())
 
 	if hdr := req.Header.Get(encoding.AcceptEncodingHeader); hdr != "" {
 		rsp.Body = decompress(hdr, rsp)

client/redis/redis.go

@@ -9,7 +9,7 @@ import (
 	"github.com/beatlabs/patron/trace"
 	"github.com/go-redis/redis/extra/rediscmd"
 	"github.com/go-redis/redis/v8"
-	opentracing "github.com/opentracing/opentracing-go"
+	"github.com/opentracing/opentracing-go"
 	"github.com/opentracing/opentracing-go/ext"
 	"github.com/prometheus/client_golang/prometheus"
 )
@@ -89,7 +89,10 @@ func (th tracingHook) AfterProcessPipeline(ctx context.Context, cmds []redis.Cmd
 
 func observeDuration(ctx context.Context, cmd string, err error) {
 	dur := time.Since(ctx.Value(duration{}).(time.Time))
-	cmdDurationMetrics.WithLabelValues(cmd, strconv.FormatBool(err != nil)).Observe(dur.Seconds())
+	durationHistogram := trace.Histogram{
+		Observer: cmdDurationMetrics.WithLabelValues(cmd, strconv.FormatBool(err == nil)),
+	}
+	durationHistogram.Observe(ctx, dur.Seconds())
 }
 
 func startSpan(ctx context.Context, address, opName string) (opentracing.Span, context.Context) {

client/sns/v2/publisher.go

@@ -69,10 +69,10 @@ func (p Publisher) Publish(ctx context.Context, input *sns.PublishInput) (messag
 	start := time.Now()
 	out, err := p.api.PublishWithContext(ctx, input)
 	if input.TopicArn != nil {
-		observePublish(span, start, *input.TopicArn, err)
+		observePublish(ctx, span, start, *input.TopicArn, err)
 	}
 	if input.TargetArn != nil {
-		observePublish(span, start, *input.TargetArn, err)
+		observePublish(ctx, span, start, *input.TargetArn, err)
 	}
 	if err != nil {
 		return "", fmt.Errorf("failed to publish message: %w", err)
@@ -124,7 +124,11 @@ func injectHeaders(span opentracing.Span, input *sns.PublishInput) error {
 	return nil
 }
 
-func observePublish(span opentracing.Span, start time.Time, topic string, err error) {
+func observePublish(ctx context.Context, span opentracing.Span, start time.Time, topic string, err error) {
 	trace.SpanComplete(span, err)
-	publishDurationMetrics.WithLabelValues(topic, strconv.FormatBool(err != nil)).Observe(time.Since(start).Seconds())
+
+	durationHistogram := trace.Histogram{
+		Observer: publishDurationMetrics.WithLabelValues(topic, strconv.FormatBool(err == nil)),
+	}
+	durationHistogram.Observe(ctx, time.Since(start).Seconds())
 }