Support exporting Exemplar. (#124)

* Support exporting Exemplar.

* Update a switch statement and add TODOs.

Author: songy23

metrics.go

@@ -23,6 +23,8 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/golang/protobuf/proto"
+	"github.com/golang/protobuf/ptypes/any"
 	"github.com/golang/protobuf/ptypes/timestamp"
 	"go.opencensus.io/trace"
 
@@ -41,6 +43,14 @@ var (
 	errUnspecifiedMetricKind = errors.New("metric kind is unpsecified")
 )
 
+const (
+	exemplarAttachmentTypeString  = "type.googleapis.com/google.protobuf.StringValue"
+	exemplarAttachmentTypeSpanCtx = "type.googleapis.com/google.monitoring.v3.SpanContext"
+
+	// TODO(songy23): add support for this.
+	// exemplarAttachmentTypeDroppedLabels = "type.googleapis.com/google.monitoring.v3.DroppedLabels"
+)
+
 // ExportMetrics exports OpenCensus Metrics to Stackdriver Monitoring.
 func (se *statsExporter) ExportMetrics(ctx context.Context, metrics []*metricdata.Metric) error {
 	if len(metrics) == 0 {
@@ -340,7 +350,7 @@ func (se *statsExporter) metricTsToMpbPoint(ts *metricdata.TimeSeries, metricKin
 			startTime = nil
 		}
 
-		spt, err := metricPointToMpbPoint(startTime, &pt)
+		spt, err := metricPointToMpbPoint(startTime, &pt, se.o.ProjectID)
 		if err != nil {
 			return nil, err
 		}
@@ -349,12 +359,12 @@ func (se *statsExporter) metricTsToMpbPoint(ts *metricdata.TimeSeries, metricKin
 	return sptl, nil
 }
 
-func metricPointToMpbPoint(startTime *timestamp.Timestamp, pt *metricdata.Point) (*monitoringpb.Point, error) {
+func metricPointToMpbPoint(startTime *timestamp.Timestamp, pt *metricdata.Point, projectID string) (*monitoringpb.Point, error) {
 	if pt == nil {
 		return nil, nil
 	}
 
-	mptv, err := metricPointToMpbValue(pt)
+	mptv, err := metricPointToMpbValue(pt, projectID)
 	if err != nil {
 		return nil, err
 	}
@@ -369,7 +379,7 @@ func metricPointToMpbPoint(startTime *timestamp.Timestamp, pt *metricdata.Point)
 	return mpt, nil
 }
 
-func metricPointToMpbValue(pt *metricdata.Point) (*monitoringpb.TypedValue, error) {
+func metricPointToMpbValue(pt *metricdata.Point, projectID string) (*monitoringpb.TypedValue, error) {
 	if pt == nil {
 		return nil, nil
 	}
@@ -423,18 +433,67 @@ func metricPointToMpbValue(pt *metricdata.Point) (*monitoringpb.TypedValue, erro
 				},
 			}
 		}
-		mv.DistributionValue.BucketCounts = addZeroBucketCountOnCondition(insertZeroBound, metricBucketToBucketCounts(dv.Buckets)...)
+		bucketCounts, exemplars := metricBucketToBucketCountsAndExemplars(dv.Buckets, projectID)
+		mv.DistributionValue.BucketCounts = addZeroBucketCountOnCondition(insertZeroBound, bucketCounts...)
+		mv.DistributionValue.Exemplars = exemplars
 
 		tval = &monitoringpb.TypedValue{Value: mv}
 	}
 
 	return tval, err
 }
 
-func metricBucketToBucketCounts(buckets []metricdata.Bucket) []int64 {
+func metricBucketToBucketCountsAndExemplars(buckets []metricdata.Bucket, projectID string) ([]int64, []*distributionpb.Distribution_Exemplar) {
 	bucketCounts := make([]int64, len(buckets))
+	var exemplars []*distributionpb.Distribution_Exemplar
 	for i, bucket := range buckets {
 		bucketCounts[i] = bucket.Count
+		if bucket.Exemplar != nil {
+			exemplars = append(exemplars, metricExemplarToPbExemplar(bucket.Exemplar, projectID))
+		}
+	}
+	return bucketCounts, exemplars
+}
+
+func metricExemplarToPbExemplar(exemplar *metricdata.Exemplar, projectID string) *distributionpb.Distribution_Exemplar {
+	return &distributionpb.Distribution_Exemplar{
+		Value:       exemplar.Value,
+		Timestamp:   timestampProto(exemplar.Timestamp),
+		Attachments: attachmentsToPbAttachments(exemplar.Attachments, projectID),
+	}
+}
+
+func attachmentsToPbAttachments(attachments metricdata.Attachments, projectID string) []*any.Any {
+	var pbAttachments []*any.Any
+	for _, v := range attachments {
+		switch v.(type) {
+		case trace.SpanContext:
+			spanCtx, _ := v.(trace.SpanContext)
+			pbAttachments = append(pbAttachments, toPbSpanCtxAttachment(spanCtx, projectID))
+		default:
+			// Treat everything else as plain string for now.
+			// TODO(songy23): add support for dropped label attachments.
+			pbAttachments = append(pbAttachments, toPbStringAttachment(v))
+		}
+	}
+	return pbAttachments
+}
+
+func toPbStringAttachment(v interface{}) *any.Any {
+	s := fmt.Sprintf("%v", v)
+	return &any.Any{
+		TypeUrl: exemplarAttachmentTypeString,
+		Value:   []byte(s),
+	}
+}
+
+func toPbSpanCtxAttachment(spanCtx trace.SpanContext, projectID string) *any.Any {
+	pbSpanCtx := monitoringpb.SpanContext{
+		SpanName: fmt.Sprintf("projects/%s/traces/%s/spans/%s", projectID, spanCtx.TraceID.String(), spanCtx.SpanID.String()),
+	}
+	bytes, _ := proto.Marshal(&pbSpanCtx)
+	return &any.Any{
+		TypeUrl: exemplarAttachmentTypeSpanCtx,
+		Value:   bytes,
 	}
-	return bucketCounts
 }

metrics_test.go

@@ -16,9 +16,13 @@ package stackdriver
 
 import (
 	"context"
+	"fmt"
 	"strings"
 	"testing"
+	"time"
 
+	"github.com/golang/protobuf/proto"
+	"github.com/golang/protobuf/ptypes/any"
 	"github.com/golang/protobuf/ptypes/timestamp"
 
 	distributionpb "google.golang.org/genproto/googleapis/api/distribution"
@@ -29,6 +33,7 @@ import (
 
 	"go.opencensus.io/metric/metricdata"
 	"go.opencensus.io/resource"
+	"go.opencensus.io/trace"
 )
 
 var se = &statsExporter{
@@ -85,6 +90,7 @@ func TestMetricToCreateTimeSeriesRequest(t *testing.T) {
 		Seconds: 1543160298,
 		Nanos:   100000090,
 	}
+	startTime := time.Unix(1543160298, 100000090)
 	endTimestamp := &timestamp.Timestamp{
 		Seconds: 1543160298,
 		Nanos:   100000997,
@@ -116,7 +122,11 @@ func TestMetricToCreateTimeSeriesRequest(t *testing.T) {
 									Sum:                   11.9,
 									SumOfSquaredDeviation: 0,
 									Buckets: []metricdata.Bucket{
-										{Count: 1}, {}, {}, {},
+										{
+											Count:    1,
+											Exemplar: &metricdata.Exemplar{Value: 11.9, Timestamp: startTime, Attachments: map[string]interface{}{"key": "value"}},
+										},
+										{}, {}, {},
 									},
 									BucketOptions: &metricdata.BucketOptions{
 										Bounds: []float64{10, 20, 30, 40},
@@ -158,6 +168,18 @@ func TestMetricToCreateTimeSeriesRequest(t *testing.T) {
 														},
 													},
 												},
+												Exemplars: []*distributionpb.Distribution_Exemplar{
+													{
+														Value:     11.9,
+														Timestamp: startTimestamp,
+														Attachments: []*any.Any{
+															{
+																TypeUrl: exemplarAttachmentTypeString,
+																Value:   []byte("value"),
+															},
+														},
+													},
+												},
 											},
 										},
 									},
@@ -408,11 +430,21 @@ func TestMetricsToMonitoringMetrics_fromProtoPoint(t *testing.T) {
 		Seconds: 1543160298,
 		Nanos:   100000090,
 	}
+	startTime := time.Unix(1543160298, 100000090)
 	endTimestamp := &timestamp.Timestamp{
 		Seconds: 1543160298,
 		Nanos:   100000997,
 	}
 
+	traceID := trace.TraceID{1, 2, 3, 4, 5, 6, 7, 8, 1, 2, 4, 8, 16, 32, 64, 128}
+	spanID := trace.SpanID{1, 2, 4, 8, 16, 32, 64, 128}
+	spanCtx := trace.SpanContext{
+		TraceID:      traceID,
+		SpanID:       spanID,
+		TraceOptions: 1,
+	}
+	wantSpanCtxBytes, _ := proto.Marshal(&monitoringpb.SpanContext{SpanName: fmt.Sprintf("projects/foo/traces/%s/spans/%s", traceID.String(), spanID.String())})
+
 	tests := []struct {
 		in      *metricdata.Point
 		want    *monitoringpb.Point
@@ -426,7 +458,13 @@ func TestMetricsToMonitoringMetrics_fromProtoPoint(t *testing.T) {
 					Sum:                   11.9,
 					SumOfSquaredDeviation: 0,
 					Buckets: []metricdata.Bucket{
-						{}, {Count: 1}, {}, {}, {},
+						{},
+						{
+							Count:    1,
+							Exemplar: &metricdata.Exemplar{Value: 11.9, Timestamp: startTime, Attachments: map[string]interface{}{"SpanContext": spanCtx}}},
+						{},
+						{},
+						{},
 					},
 					BucketOptions: &metricdata.BucketOptions{
 						Bounds: []float64{0, 10, 20, 30, 40},
@@ -452,6 +490,18 @@ func TestMetricsToMonitoringMetrics_fromProtoPoint(t *testing.T) {
 									},
 								},
 							},
+							Exemplars: []*distributionpb.Distribution_Exemplar{
+								{
+									Value:     11.9,
+									Timestamp: startTimestamp,
+									Attachments: []*any.Any{
+										{
+											TypeUrl: exemplarAttachmentTypeSpanCtx,
+											Value:   wantSpanCtxBytes,
+										},
+									},
+								},
+							},
 						},
 					},
 				},
@@ -490,7 +540,7 @@ func TestMetricsToMonitoringMetrics_fromProtoPoint(t *testing.T) {
 	}
 
 	for i, tt := range tests {
-		mpt, err := metricPointToMpbPoint(startTimestamp, tt.in)
+		mpt, err := metricPointToMpbPoint(startTimestamp, tt.in, "foo")
 		if tt.wantErr != "" {
 			continue
 		}