add exemplar support to OpenCensus bridge

Author: dashpole

CHANGELOG.md

@@ -11,6 +11,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 ### Added
 
 - Add `go.opentelemetry.io/otel/bridge/opencensus.InstallTraceBridge`, which installs the OpenCensus trace bridge, and replaces `opencensus.NewTracer`. (#4567)
+- `go.opentelemetry.io/otel/bridge/opencensus.NewMetricProducer` now supports exemplars from OpenCensus. (#4585)
 
 ### Deprecated
 

bridge/opencensus/doc.go

@@ -59,5 +59,4 @@
 //   - Summary-typed metrics are dropped
 //   - GaugeDistribution-typed metrics are dropped
 //   - Histogram's SumOfSquaredDeviation field is dropped
-//   - Exemplars on Histograms are dropped
 package opencensus // import "go.opentelemetry.io/otel/bridge/opencensus"

bridge/opencensus/internal/ocmetric/metric.go

@@ -17,35 +17,37 @@ package internal // import "go.opentelemetry.io/otel/bridge/opencensus/internal/
 import (
 	"errors"
 	"fmt"
+	"reflect"
+	"sort"
 
 	ocmetricdata "go.opencensus.io/metric/metricdata"
+	octrace "go.opencensus.io/trace"
 
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 )
 
 var (
-	errConversion                   = errors.New("converting from OpenCensus to OpenTelemetry")
-	errAggregationType              = errors.New("unsupported OpenCensus aggregation type")
-	errMismatchedValueTypes         = errors.New("wrong value type for data point")
-	errNumberDataPoint              = errors.New("converting a number data point")
-	errHistogramDataPoint           = errors.New("converting a histogram data point")
-	errNegativeDistributionCount    = errors.New("distribution count is negative")
-	errNegativeBucketCount          = errors.New("distribution bucket count is negative")
-	errMismatchedAttributeKeyValues = errors.New("mismatched number of attribute keys and values")
+	errAggregationType                = errors.New("unsupported OpenCensus aggregation type")
+	errMismatchedValueTypes           = errors.New("wrong value type for data point")
+	errNegativeDistributionCount      = errors.New("distribution count is negative")
+	errNegativeBucketCount            = errors.New("distribution bucket count is negative")
+	errMismatchedAttributeKeyValues   = errors.New("mismatched number of attribute keys and values")
+	errInvalidExemplarSpanContext     = errors.New("span context exemplar attachment did not contain an OpenCensus SpanContext")
+	errInvalidExemplarAttachmentValue = errors.New("exemplar attachment is not a supported OpenTelemetry attribute type")
 )
 
 // ConvertMetrics converts metric data from OpenCensus to OpenTelemetry.
 func ConvertMetrics(ocmetrics []*ocmetricdata.Metric) ([]metricdata.Metrics, error) {
 	otelMetrics := make([]metricdata.Metrics, 0, len(ocmetrics))
-	var errInfo []string
+	var err error
 	for _, ocm := range ocmetrics {
 		if ocm == nil {
 			continue
 		}
-		agg, err := convertAggregation(ocm)
-		if err != nil {
-			errInfo = append(errInfo, err.Error())
+		agg, aggregationErr := convertAggregation(ocm)
+		if aggregationErr != nil {
+			err = errors.Join(err, fmt.Errorf("error converting metric %v: %w", ocm.Descriptor.Name, aggregationErr))
 			continue
 		}
 		otelMetrics = append(otelMetrics, metricdata.Metrics{
@@ -55,11 +57,10 @@ func ConvertMetrics(ocmetrics []*ocmetricdata.Metric) ([]metricdata.Metrics, err
 			Data:        agg,
 		})
 	}
-	var aggregatedError error
-	if len(errInfo) > 0 {
-		aggregatedError = fmt.Errorf("%w: %q", errConversion, errInfo)
+	if err != nil {
+		return otelMetrics, fmt.Errorf("error converting from OpenCensus to OpenTelemetry: %w", err)
 	}
-	return otelMetrics, aggregatedError
+	return otelMetrics, nil
 }
 
 // convertAggregation produces an aggregation based on the OpenCensus Metric.
@@ -97,17 +98,17 @@ func convertSum[N int64 | float64](labelKeys []ocmetricdata.LabelKey, ts []*ocme
 // convertNumberDataPoints converts OpenCensus TimeSeries to OpenTelemetry DataPoints.
 func convertNumberDataPoints[N int64 | float64](labelKeys []ocmetricdata.LabelKey, ts []*ocmetricdata.TimeSeries) ([]metricdata.DataPoint[N], error) {
 	var points []metricdata.DataPoint[N]
-	var errInfo []string
+	var err error
 	for _, t := range ts {
-		attrs, err := convertAttrs(labelKeys, t.LabelValues)
-		if err != nil {
-			errInfo = append(errInfo, err.Error())
+		attrs, attrsErr := convertAttrs(labelKeys, t.LabelValues)
+		if attrsErr != nil {
+			err = errors.Join(err, attrsErr)
 			continue
 		}
 		for _, p := range t.Points {
 			v, ok := p.Value.(N)
 			if !ok {
-				errInfo = append(errInfo, fmt.Sprintf("%v: %q", errMismatchedValueTypes, p.Value))
+				err = errors.Join(err, fmt.Errorf("%w: %q", errMismatchedValueTypes, p.Value))
 				continue
 			}
 			points = append(points, metricdata.DataPoint[N]{
@@ -118,40 +119,35 @@ func convertNumberDataPoints[N int64 | float64](labelKeys []ocmetricdata.LabelKe
 			})
 		}
 	}
-	var aggregatedError error
-	if len(errInfo) > 0 {
-		aggregatedError = fmt.Errorf("%w: %v", errNumberDataPoint, errInfo)
-	}
-	return points, aggregatedError
+	return points, err
 }
 
 // convertHistogram converts OpenCensus Distribution timeseries to an
 // OpenTelemetry Histogram aggregation.
 func convertHistogram(labelKeys []ocmetricdata.LabelKey, ts []*ocmetricdata.TimeSeries) (metricdata.Histogram[float64], error) {
 	points := make([]metricdata.HistogramDataPoint[float64], 0, len(ts))
-	var errInfo []string
+	var err error
 	for _, t := range ts {
-		attrs, err := convertAttrs(labelKeys, t.LabelValues)
-		if err != nil {
-			errInfo = append(errInfo, err.Error())
+		attrs, attrsErr := convertAttrs(labelKeys, t.LabelValues)
+		if attrsErr != nil {
+			err = errors.Join(err, attrsErr)
 			continue
 		}
 		for _, p := range t.Points {
 			dist, ok := p.Value.(*ocmetricdata.Distribution)
 			if !ok {
-				errInfo = append(errInfo, fmt.Sprintf("%v: %d", errMismatchedValueTypes, p.Value))
+				err = errors.Join(err, fmt.Errorf("%w: %d", errMismatchedValueTypes, p.Value))
 				continue
 			}
-			bucketCounts, err := convertBucketCounts(dist.Buckets)
-			if err != nil {
-				errInfo = append(errInfo, err.Error())
+			bucketCounts, exemplars, bucketErr := convertBuckets(dist.Buckets)
+			if bucketErr != nil {
+				err = errors.Join(err, bucketErr)
 				continue
 			}
 			if dist.Count < 0 {
-				errInfo = append(errInfo, fmt.Sprintf("%v: %d", errNegativeDistributionCount, dist.Count))
+				err = errors.Join(err, fmt.Errorf("%w: %d", errNegativeDistributionCount, dist.Count))
 				continue
 			}
-			// TODO: handle exemplars
 			points = append(points, metricdata.HistogramDataPoint[float64]{
 				Attributes:   attrs,
 				StartTime:    t.StartTime,
@@ -160,26 +156,93 @@ func convertHistogram(labelKeys []ocmetricdata.LabelKey, ts []*ocmetricdata.Time
 				Sum:          dist.Sum,
 				Bounds:       dist.BucketOptions.Bounds,
 				BucketCounts: bucketCounts,
+				Exemplars:    exemplars,
 			})
 		}
 	}
-	var aggregatedError error
-	if len(errInfo) > 0 {
-		aggregatedError = fmt.Errorf("%w: %v", errHistogramDataPoint, errInfo)
-	}
-	return metricdata.Histogram[float64]{DataPoints: points, Temporality: metricdata.CumulativeTemporality}, aggregatedError
+	return metricdata.Histogram[float64]{DataPoints: points, Temporality: metricdata.CumulativeTemporality}, err
 }
 
-// convertBucketCounts converts from OpenCensus bucket counts to slice of uint64.
-func convertBucketCounts(buckets []ocmetricdata.Bucket) ([]uint64, error) {
+// convertBuckets converts from OpenCensus bucket counts to slice of uint64,
+// and converts OpenCensus exemplars to OpenTelemetry exemplars.
+func convertBuckets(buckets []ocmetricdata.Bucket) ([]uint64, []metricdata.Exemplar[float64], error) {
 	bucketCounts := make([]uint64, len(buckets))
+	exemplars := []metricdata.Exemplar[float64]{}
+	var err error
 	for i, bucket := range buckets {
 		if bucket.Count < 0 {
-			return nil, fmt.Errorf("%w: %q", errNegativeBucketCount, bucket.Count)
+			err = errors.Join(err, fmt.Errorf("%w: %q", errNegativeBucketCount, bucket.Count))
+		} else {
+			bucketCounts[i] = uint64(bucket.Count)
+		}
+		if bucket.Exemplar != nil {
+			exemplar, exemplarErr := convertExemplar(bucket.Exemplar)
+			if exemplarErr != nil {
+				err = errors.Join(err, exemplarErr)
+			} else {
+				exemplars = append(exemplars, exemplar)
+			}
+		}
+	}
+	return bucketCounts, exemplars, err
+}
+
+// convertExemplar converts an OpenCensus exemplar to an OpenTelemetry exemplar.
+func convertExemplar(ocExemplar *ocmetricdata.Exemplar) (metricdata.Exemplar[float64], error) {
+	exemplar := metricdata.Exemplar[float64]{
+		Value: ocExemplar.Value,
+		Time:  ocExemplar.Timestamp,
+	}
+	if ocExemplar.Attachments == nil {
+		return exemplar, nil
+	}
+	var err error
+	for k, v := range ocExemplar.Attachments {
+		if k == ocmetricdata.AttachmentKeySpanContext {
+			if sc, ok := v.(octrace.SpanContext); ok {
+				exemplar.SpanID = sc.SpanID[:]
+				exemplar.TraceID = sc.TraceID[:]
+			} else {
+				err = errors.Join(err, fmt.Errorf("%w; type: %v", errInvalidExemplarSpanContext, reflect.TypeOf(v)))
+			}
+		} else if kv := convertKV(k, v); kv.Valid() {
+			exemplar.FilteredAttributes = append(exemplar.FilteredAttributes, kv)
+		} else {
+			err = errors.Join(err, fmt.Errorf("%w; type: %v", errInvalidExemplarAttachmentValue, reflect.TypeOf(v)))
 		}
-		bucketCounts[i] = uint64(bucket.Count)
 	}
-	return bucketCounts, nil
+	sortable := attribute.Sortable(exemplar.FilteredAttributes)
+	sort.Sort(&sortable)
+	return exemplar, err
+}
+
+// convertKV converts an OpenCensus Attachment to an OpenTelemetry KeyValue.
+func convertKV(key string, value any) attribute.KeyValue {
+	switch typedVal := value.(type) {
+	case bool:
+		return attribute.Bool(key, typedVal)
+	case []bool:
+		return attribute.BoolSlice(key, typedVal)
+	case int:
+		return attribute.Int(key, typedVal)
+	case []int:
+		return attribute.IntSlice(key, typedVal)
+	case int64:
+		return attribute.Int64(key, typedVal)
+	case []int64:
+		return attribute.Int64Slice(key, typedVal)
+	case float64:
+		return attribute.Float64(key, typedVal)
+	case []float64:
+		return attribute.Float64Slice(key, typedVal)
+	case string:
+		return attribute.String(key, typedVal)
+	case []string:
+		return attribute.StringSlice(key, typedVal)
+	case fmt.Stringer:
+		return attribute.Stringer(key, typedVal)
+	}
+	return attribute.KeyValue{}
 }
 
 // convertAttrs converts from OpenCensus attribute keys and values to an