Exemplar: Add new record APIs that take exemplar attachments and Span…

…Context 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.

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{}))
+}