Add GetMonitoredResource option (#46)

This allows customizing the MonitoredResource based on tag values.

Author: Ramon Nogueira

monitoredresource/monitored_resources_test.go

@@ -101,11 +101,3 @@ func TestAWSEC2InstanceMonitoredResources(t *testing.T) {
 		t.Errorf("AWSEC2InstanceMonitoredResource Failed: %v", autoDetected)
 	}
 }
-
-func TestNullMonitoredResources(t *testing.T) {
-	os.Setenv("KUBERNETES_SERVICE_HOST", "")
-	mr := Autodetect()
-	if mr != nil {
-		t.Errorf("Expected nil MonitoredResource but found %v", mr)
-	}
-}

stackdriver.go

@@ -57,6 +57,7 @@ import (
 	traceapi "cloud.google.com/go/trace/apiv2"
 	"contrib.go.opencensus.io/exporter/stackdriver/monitoredresource"
 	"go.opencensus.io/stats/view"
+	"go.opencensus.io/tag"
 	"go.opencensus.io/trace"
 	"golang.org/x/oauth2/google"
 	"google.golang.org/api/option"
@@ -183,18 +184,36 @@ type Options struct {
 	// the Resource you set uniquely identifies this Go process.
 	DefaultMonitoringLabels *Labels
 
-	// Context allows users to provide a custom context for API calls.
+	// Context allows you to provide a custom context for API calls.
 	//
 	// This context will be used several times: first, to create Stackdriver
 	// trace and metric clients, and then every time a new batch of traces or
 	// stats needs to be uploaded.
 	//
 	// If unset, context.Background() will be used.
 	Context context.Context
+
+	// GetMonitoredResource may be provided to supply the details of the
+	// monitored resource dynamically based on the tags associated with each
+	// data point. Most users will not need to set this, but should instead
+	// set the MonitoredResource field.
+	//
+	// GetMonitoredResource may add or remove tags by returning a new set of
+	// tags. It is safe for the function to mutate its argument and return it.
+	//
+	// See the documentation on the MonitoredResource field for guidance on the
+	// interaction between monitored resources and labels.
+	//
+	// The MonitoredResource field is ignored if this field is set to a non-nil
+	// value.
+	GetMonitoredResource func(*view.View, []tag.Tag) ([]tag.Tag, monitoredresource.Interface)
 }
 
-// Exporter is a stats.Exporter and trace.Exporter
-// implementation that uploads data to Stackdriver.
+// Exporter is a stats and trace exporter that uploads data to Stackdriver.
+//
+// You can create a single Exporter and register it as both a trace exporter
+// (to export to Stackdriver Trace) and a stats exporter (to integrate with
+// Stackdriver Monitoring).
 type Exporter struct {
 	traceExporter *traceExporter
 	statsExporter *statsExporter

stats.go

@@ -87,6 +87,7 @@ func newStatsExporter(o Options) (*statsExporter, error) {
 		o:            o,
 		createdViews: make(map[string]*metricpb.MetricDescriptor),
 	}
+
 	if o.DefaultMonitoringLabels != nil {
 		e.defaultLabels = o.DefaultMonitoringLabels.m
 	} else {
@@ -107,6 +108,21 @@ func newStatsExporter(o Options) (*statsExporter, error) {
 	return e, nil
 }
 
+func (e *statsExporter) getMonitoredResource(v *view.View, tags []tag.Tag) ([]tag.Tag, *monitoredrespb.MonitoredResource) {
+	if get := e.o.GetMonitoredResource; get != nil {
+		newTags, mr := get(v, tags)
+		return newTags, convertMonitoredResourceToPB(mr)
+	} else {
+		resource := e.o.Resource
+		if resource == nil {
+			resource = &monitoredrespb.MonitoredResource{
+				Type: "global",
+			}
+		}
+		return tags, resource
+	}
+}
+
 // ExportView exports to the Stackdriver Monitoring if view data
 // has one or more rows.
 func (e *statsExporter) ExportView(vd *view.Data) {
@@ -177,20 +193,13 @@ func (e *statsExporter) uploadStats(vds []*view.Data) error {
 func (e *statsExporter) makeReq(vds []*view.Data, limit int) []*monitoringpb.CreateTimeSeriesRequest {
 	var reqs []*monitoringpb.CreateTimeSeriesRequest
 	var timeSeries []*monitoringpb.TimeSeries
-
-	resource := e.o.Resource
-	if resource == nil {
-		resource = &monitoredrespb.MonitoredResource{
-			Type: "global",
-		}
-	}
-
 	for _, vd := range vds {
 		for _, row := range vd.Rows {
+			tags, resource := e.getMonitoredResource(vd.View, append([]tag.Tag(nil), row.Tags...))
 			ts := &monitoringpb.TimeSeries{
 				Metric: &metricpb.Metric{
 					Type:   e.metricType(vd.View),
-					Labels: newLabels(e.defaultLabels, row.Tags),
+					Labels: newLabels(e.defaultLabels, tags),
 				},
 				Resource: resource,
 				Points:   []*monitoringpb.Point{newPoint(vd.View, row, vd.Start, vd.End)},

stats_test.go

@@ -16,8 +16,8 @@ package stackdriver
 
 import (
 	"context"
+	"contrib.go.opencensus.io/exporter/stackdriver/monitoredresource"
 	"fmt"
-	"reflect"
 	"testing"
 	"time"
 
@@ -39,6 +39,8 @@ import (
 
 var authOptions = []option.ClientOption{option.WithGRPCConn(&grpc.ClientConn{})}
 
+var testOptions = Options{ProjectID: "opencensus-test", MonitoringClientOptions: authOptions}
+
 func TestRejectBlankProjectID(t *testing.T) {
 	ids := []string{"", "     ", " "}
 	for _, projectID := range ids {
@@ -492,7 +494,10 @@ func TestExporter_makeReq_batching(t *testing.T) {
 			vds = append(vds, newTestViewData(v, time.Now(), time.Now(), count1, count2))
 		}
 
-		e := &statsExporter{}
+		e, err := newStatsExporter(testOptions)
+		if err != nil {
+			t.Fatal(err)
+		}
 		resps := e.makeReq(vds, tt.limit)
 		if len(resps) != tt.wantReqs {
 			t.Errorf("%v: got %v; want %d requests", tt.name, resps, tt.wantReqs)
@@ -646,7 +651,7 @@ func TestEqualAggWindowTagKeys(t *testing.T) {
 			wantErr: false,
 		},
 	}
-	e, err := newStatsExporter(Options{ProjectID: "opencensus-test", MonitoringClientOptions: authOptions})
+	e, err := newStatsExporter(testOptions)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -860,8 +865,18 @@ func TestExporter_makeReq_withCustomMonitoredResource(t *testing.T) {
 	taskValue := getTaskValue()
 
 	resource := &monitoredrespb.MonitoredResource{
-		Type:   "gce_instance",
-		Labels: map[string]string{"instance_id": "instance", "zone": "us-west-1a"},
+		Type: "gce_instance",
+		Labels: map[string]string{
+			"project_id":  "proj-id",
+			"instance_id": "instance",
+			"zone":        "us-west-1a",
+		},
+	}
+
+	gceInst := &monitoredresource.GCEInstance{
+		ProjectID:  "proj-id",
+		InstanceID: "instance",
+		Zone:       "us-west-1a",
 	}
 
 	tests := []struct {
@@ -934,6 +949,150 @@ func TestExporter_makeReq_withCustomMonitoredResource(t *testing.T) {
 				},
 			}},
 		},
+		{
+			name: "with MonitoredResource and labels",
+			opts: func() Options {
+				var labels Labels
+				labels.Set("pid", "1234", "Process identifier")
+				return Options{
+					MonitoredResource:       gceInst,
+					DefaultMonitoringLabels: &labels,
+				}
+			}(),
+			vd: newTestViewData(v, start, end, count1, count2),
+			want: []*monitoringpb.CreateTimeSeriesRequest{{
+				Name: monitoring.MetricProjectPath("proj-id"),
+				TimeSeries: []*monitoringpb.TimeSeries{
+					{
+						Metric: &metricpb.Metric{
+							Type: "custom.googleapis.com/opencensus/testview",
+							Labels: map[string]string{
+								"test_key": "test-value-1",
+								"pid":      "1234",
+							},
+						},
+						Resource: resource,
+						Points: []*monitoringpb.Point{
+							{
+								Interval: &monitoringpb.TimeInterval{
+									StartTime: &timestamp.Timestamp{
+										Seconds: start.Unix(),
+										Nanos:   int32(start.Nanosecond()),
+									},
+									EndTime: &timestamp.Timestamp{
+										Seconds: end.Unix(),
+										Nanos:   int32(end.Nanosecond()),
+									},
+								},
+								Value: &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_Int64Value{
+									Int64Value: 10,
+								}},
+							},
+						},
+					},
+					{
+						Metric: &metricpb.Metric{
+							Type: "custom.googleapis.com/opencensus/testview",
+							Labels: map[string]string{
+								"test_key": "test-value-2",
+								"pid":      "1234",
+							},
+						},
+						Resource: resource,
+						Points: []*monitoringpb.Point{
+							{
+								Interval: &monitoringpb.TimeInterval{
+									StartTime: &timestamp.Timestamp{
+										Seconds: start.Unix(),
+										Nanos:   int32(start.Nanosecond()),
+									},
+									EndTime: &timestamp.Timestamp{
+										Seconds: end.Unix(),
+										Nanos:   int32(end.Nanosecond()),
+									},
+								},
+								Value: &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_Int64Value{
+									Int64Value: 16,
+								}},
+							},
+						},
+					},
+				},
+			}},
+		},
+		{
+			name: "GetMonitoredResource and labels",
+			opts: func() Options {
+				var labels Labels
+				labels.Set("pid", "1234", "Process identifier")
+				return Options{
+					GetMonitoredResource: func(v *view.View, t []tag.Tag) ([]tag.Tag, monitoredresource.Interface) {
+						return t, gceInst
+					},
+					DefaultMonitoringLabels: &labels,
+				}
+			}(),
+			vd: newTestViewData(v, start, end, count1, count2),
+			want: []*monitoringpb.CreateTimeSeriesRequest{{
+				Name: monitoring.MetricProjectPath("proj-id"),
+				TimeSeries: []*monitoringpb.TimeSeries{
+					{
+						Metric: &metricpb.Metric{
+							Type: "custom.googleapis.com/opencensus/testview",
+							Labels: map[string]string{
+								"test_key": "test-value-1",
+								"pid":      "1234",
+							},
+						},
+						Resource: resource,
+						Points: []*monitoringpb.Point{
+							{
+								Interval: &monitoringpb.TimeInterval{
+									StartTime: &timestamp.Timestamp{
+										Seconds: start.Unix(),
+										Nanos:   int32(start.Nanosecond()),
+									},
+									EndTime: &timestamp.Timestamp{
+										Seconds: end.Unix(),
+										Nanos:   int32(end.Nanosecond()),
+									},
+								},
+								Value: &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_Int64Value{
+									Int64Value: 10,
+								}},
+							},
+						},
+					},
+					{
+						Metric: &metricpb.Metric{
+							Type: "custom.googleapis.com/opencensus/testview",
+							Labels: map[string]string{
+								"test_key": "test-value-2",
+								"pid":      "1234",
+							},
+						},
+						Resource: resource,
+						Points: []*monitoringpb.Point{
+							{
+								Interval: &monitoringpb.TimeInterval{
+									StartTime: &timestamp.Timestamp{
+										Seconds: start.Unix(),
+										Nanos:   int32(start.Nanosecond()),
+									},
+									EndTime: &timestamp.Timestamp{
+										Seconds: end.Unix(),
+										Nanos:   int32(end.Nanosecond()),
+									},
+								},
+								Value: &monitoringpb.TypedValue{Value: &monitoringpb.TypedValue_Int64Value{
+									Int64Value: 16,
+								}},
+							},
+						},
+					},
+				},
+			}},
+		},
 		{
 			name: "custom default monitoring labels",
 			opts: func() Options {
@@ -1075,19 +1234,19 @@ func TestExporter_makeReq_withCustomMonitoredResource(t *testing.T) {
 			opts := tt.opts
 			opts.MonitoringClientOptions = authOptions
 			opts.ProjectID = "proj-id"
-			e, err := newStatsExporter(opts)
+			e, err := NewExporter(opts)
 			if err != nil {
 				t.Fatal(err)
 			}
-			resps := e.makeReq([]*view.Data{tt.vd}, maxTimeSeriesPerUpload)
+			resps := e.statsExporter.makeReq([]*view.Data{tt.vd}, maxTimeSeriesPerUpload)
 			if got, want := len(resps), len(tt.want); got != want {
 				t.Fatalf("%v: Exporter.makeReq() returned %d responses; want %d", tt.name, got, want)
 			}
 			if len(tt.want) == 0 {
 				return
 			}
-			if !reflect.DeepEqual(resps, tt.want) {
-				t.Errorf("%v: Exporter.makeReq() = \n      %v\nwant: %v", tt.name, resps, tt.want)
+			if diff := cmp.Diff(resps, tt.want); diff != "" {
+				t.Errorf("Requests differ, -got +want: %s", diff)
 			}
 		})
 	}
@@ -1105,7 +1264,7 @@ func TestExporter_customContext(t *testing.T) {
 	var timedOut = 0
 	createMetricDescriptor = func(ctx context.Context, c *monitoring.MetricClient, mdr *monitoringpb.CreateMetricDescriptorRequest) (*metric.MetricDescriptor, error) {
 		select {
-		case <-time.After(15 * time.Millisecond):
+		case <-time.After(1 * time.Second):
 			fmt.Println("createMetricDescriptor did not time out")
 		case <-ctx.Done():
 			timedOut++
@@ -1114,7 +1273,7 @@ func TestExporter_customContext(t *testing.T) {
 	}
 	createTimeSeries = func(ctx context.Context, c *monitoring.MetricClient, ts *monitoringpb.CreateTimeSeriesRequest) error {
 		select {
-		case <-time.After(15 * time.Millisecond):
+		case <-time.After(1 * time.Second):
 			fmt.Println("createTimeSeries did not time out")
 		case <-ctx.Done():
 			timedOut++