Exemplar: Add new record APIs that take exemplar attachments and SpanContext key. (#1123)

* Exemplar: Add SpanContext Attachment key.

* Exemplar: Add new record APIs that take exemplar attachments.

* Use RetrieveData instead of fake exporter to fix race.

* Change map[string]interface to metricdata.Attachments

* Use nil instead of empty map.

* Update to use options for recording attachments.

Author: songy23

metric/metricdata/exemplar.go

@@ -18,6 +18,11 @@ import (
 	"time"
 )
 
+// Exemplars keys.
+const (
+	AttachmentKeySpanContext = "SpanContext"
+)
+
 // Exemplar is an example data point associated with each bucket of a
 // distribution type aggregation.
 //

stats/record.go

@@ -18,6 +18,7 @@ package stats
 import (
 	"context"
 
+	"go.opencensus.io/metric/metricdata"
 	"go.opencensus.io/stats/internal"
 	"go.opencensus.io/tag"
 )
@@ -30,40 +31,87 @@ func init() {
 	}
 }
 
+type recordOptions struct {
+	attachments  metricdata.Attachments
+	mutators     []tag.Mutator
+	measurements []Measurement
+}
+
+// WithAttachments applies provided exemplar attachments.
+func WithAttachments(attachments metricdata.Attachments) Options {
+	return func(ro *recordOptions) {
+		ro.attachments = attachments
+	}
+}
+
+// WithTags applies provided tag mutators.
+func WithTags(mutators ...tag.Mutator) Options {
+	return func(ro *recordOptions) {
+		ro.mutators = mutators
+	}
+}
+
+// WithMeasurements applies provided measurements.
+func WithMeasurements(measurements ...Measurement) Options {
+	return func(ro *recordOptions) {
+		ro.measurements = measurements
+	}
+}
+
+// Options apply changes to recordOptions.
+type Options func(*recordOptions)
+
+func createRecordOption(ros ...Options) *recordOptions {
+	o := &recordOptions{}
+	for _, ro := range ros {
+		ro(o)
+	}
+	return o
+}
+
 // Record records one or multiple measurements with the same context at once.
 // If there are any tags in the context, measurements will be tagged with them.
 func Record(ctx context.Context, ms ...Measurement) {
+	RecordWithOptions(ctx, WithMeasurements(ms...))
+}
+
+// RecordWithTags records one or multiple measurements at once.
+//
+// Measurements will be tagged with the tags in the context mutated by the mutators.
+// RecordWithTags is useful if you want to record with tag mutations but don't want
+// to propagate the mutations in the context.
+func RecordWithTags(ctx context.Context, mutators []tag.Mutator, ms ...Measurement) error {
+	return RecordWithOptions(ctx, WithTags(mutators...), WithMeasurements(ms...))
+}
+
+// RecordWithOptions records measurements from the given options (if any) against context
+// and tags and attachments in the options (if any).
+// If there are any tags in the context, measurements will be tagged with them.
+func RecordWithOptions(ctx context.Context, ros ...Options) error {
+	o := createRecordOption(ros...)
+	if len(o.measurements) == 0 {
+		return nil
+	}
 	recorder := internal.DefaultRecorder
 	if recorder == nil {
-		return
-	}
-	if len(ms) == 0 {
-		return
+		return nil
 	}
 	record := false
-	for _, m := range ms {
+	for _, m := range o.measurements {
 		if m.desc.subscribed() {
 			record = true
 			break
 		}
 	}
 	if !record {
-		return
+		return nil
 	}
-	// TODO(songy23): fix attachments.
-	recorder(tag.FromContext(ctx), ms, map[string]interface{}{})
-}
-
-// RecordWithTags records one or multiple measurements at once.
-//
-// Measurements will be tagged with the tags in the context mutated by the mutators.
-// RecordWithTags is useful if you want to record with tag mutations but don't want
-// to propagate the mutations in the context.
-func RecordWithTags(ctx context.Context, mutators []tag.Mutator, ms ...Measurement) error {
-	ctx, err := tag.New(ctx, mutators...)
-	if err != nil {
-		return err
+	if len(o.mutators) > 0 {
+		var err error
+		if ctx, err = tag.New(ctx, o.mutators...); err != nil {
+			return err
+		}
 	}
-	Record(ctx, ms...)
+	recorder(tag.FromContext(ctx), o.measurements, o.attachments)
 	return nil
 }

stats/record_test.go

@@ -0,0 +1,95 @@
+// Copyright 2019, OpenCensus Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package stats_test
+
+import (
+	"context"
+	"log"
+	"reflect"
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+
+	"go.opencensus.io/metric/metricdata"
+	"go.opencensus.io/stats"
+	"go.opencensus.io/stats/view"
+	"go.opencensus.io/tag"
+	"go.opencensus.io/trace"
+)
+
+var (
+	tid     = trace.TraceID{1, 2, 3, 4, 5, 6, 7, 8, 1, 2, 4, 8, 16, 32, 64, 128}
+	sid     = trace.SpanID{1, 2, 4, 8, 16, 32, 64, 128}
+	spanCtx = trace.SpanContext{
+		TraceID:      tid,
+		SpanID:       sid,
+		TraceOptions: 1,
+	}
+)
+
+func TestRecordWithAttachments(t *testing.T) {
+	k1, _ := tag.NewKey("k1")
+	k2, _ := tag.NewKey("k2")
+	distribution := view.Distribution(5, 10)
+	m := stats.Int64("TestRecordWithAttachments/m1", "", stats.UnitDimensionless)
+	v := &view.View{
+		Name:        "test_view",
+		TagKeys:     []tag.Key{k1, k2},
+		Measure:     m,
+		Aggregation: distribution,
+	}
+	view.SetReportingPeriod(100 * time.Millisecond)
+	if err := view.Register(v); err != nil {
+		log.Fatalf("Failed to register views: %v", err)
+	}
+
+	attachments := map[string]interface{}{metricdata.AttachmentKeySpanContext: spanCtx}
+	stats.RecordWithOptions(context.Background(), stats.WithAttachments(attachments), stats.WithMeasurements(m.M(12)))
+	rows, err := view.RetrieveData("test_view")
+	if err != nil {
+		t.Errorf("Failed to retrieve data %v", err)
+	}
+	if len(rows) == 0 {
+		t.Errorf("No data was recorded.")
+	}
+	data := rows[0].Data
+	dis, ok := data.(*view.DistributionData)
+	if !ok {
+		t.Errorf("want DistributionData, got %+v", data)
+	}
+	wantBuckets := []int64{0, 0, 1}
+	if !reflect.DeepEqual(dis.CountPerBucket, wantBuckets) {
+		t.Errorf("want buckets %v, got %v", wantBuckets, dis.CountPerBucket)
+	}
+	for i, e := range dis.ExemplarsPerBucket {
+		// Exemplar slice should be [nil, nil, exemplar]
+		if i != 2 && e != nil {
+			t.Errorf("want nil exemplar, got %v", e)
+		}
+		if i == 2 {
+			wantExemplar := &metricdata.Exemplar{Value: 12, Attachments: attachments}
+			if diff := cmpExemplar(e, wantExemplar); diff != "" {
+				t.Fatalf("Unexpected Exemplar -got +want: %s", diff)
+			}
+		}
+	}
+}
+
+// Compare exemplars while ignoring exemplar timestamp, since timestamp is non-deterministic.
+func cmpExemplar(got, want *metricdata.Exemplar) string {
+	return cmp.Diff(got, want, cmpopts.IgnoreFields(metricdata.Exemplar{}, "Timestamp"), cmpopts.IgnoreUnexported(metricdata.Exemplar{}))
+}