Add option to provide resource based on metric descriptor.

metrics.go

@@ -126,8 +126,18 @@ func (se *statsExporter) metricToMpbTs(ctx context.Context, metric *metricdata.M
 	if metric == nil {
 		return nil, errNilMetricOrMetricDescriptor
 	}
-
-	resource := se.metricRscToMpbRsc(metric.Resource)
+	var resource *monitoredrespb.MonitoredResource
+	if get := se.o.ResourceByDescriptor; get != nil {
+		mr := get(&metric.Descriptor)
+		// TODO(rghetia): optimize this. It is inefficient to convert this for all metrics.
+		resource = convertMonitoredResourceToPB(mr)
+		if resource.Type == "" {
+			resource.Type = "global"
+			resource.Labels = nil
+		}
+	} else {
+		resource = se.metricRscToMpbRsc(metric.Resource)
+	}
 
 	metricName := metric.Descriptor.Name
 	metricType := se.metricTypeFromProto(metricName)

metrics_test.go

@@ -31,6 +31,7 @@ import (
 	monitoredrespb "google.golang.org/genproto/googleapis/api/monitoredres"
 	monitoringpb "google.golang.org/genproto/googleapis/monitoring/v3"
 
+	"contrib.go.opencensus.io/exporter/stackdriver/monitoredresource"
 	"go.opencensus.io/metric/metricdata"
 	"go.opencensus.io/resource"
 	"go.opencensus.io/trace"
@@ -559,3 +560,238 @@ func TestMetricsToMonitoringMetrics_fromProtoPoint(t *testing.T) {
 		}
 	}
 }
+
+func TestGetMonitorResource(t *testing.T) {
+	startTimestamp := &timestamp.Timestamp{
+		Seconds: 1543160298,
+		Nanos:   100000090,
+	}
+	startTime, _ := ptypes.Timestamp(startTimestamp)
+	endTimestamp := &timestamp.Timestamp{
+		Seconds: 1543160298,
+		Nanos:   100000997,
+	}
+	endTime, _ := ptypes.Timestamp(endTimestamp)
+
+	tests := []struct {
+		in      *metricdata.Metric
+		want    []*monitoringpb.CreateTimeSeriesRequest
+		wantErr string
+	}{
+		{
+			in: &metricdata.Metric{
+				Descriptor: metricdata.Descriptor{
+					Name:        "custom_resource_one",
+					Description: "This is a test",
+					Unit:        metricdata.UnitBytes,
+					Type:        metricdata.TypeCumulativeInt64,
+				},
+				Resource: nil,
+				TimeSeries: []*metricdata.TimeSeries{
+					{
+						StartTime: startTime,
+						Points: []metricdata.Point{
+							{
+								Time:  endTime,
+								Value: int64(5),
+							},
+						},
+					},
+				},
+			},
+			want: []*monitoringpb.CreateTimeSeriesRequest{
+				{
+					Name: "projects/foo",
+					TimeSeries: []*monitoringpb.TimeSeries{
+						{
+							Metric: &googlemetricpb.Metric{
+								Type:   "custom.googleapis.com/opencensus/custom_resource_one",
+								Labels: nil,
+							},
+							Resource: &monitoredrespb.MonitoredResource{
+								Type: "one",
+								Labels: map[string]string{
+									"k11": "v11",
+									"k12": "v12",
+								},
+							},
+							Points: []*monitoringpb.Point{
+								{
+									Interval: &monitoringpb.TimeInterval{
+										StartTime: startTimestamp,
+										EndTime:   endTimestamp,
+									},
+									Value: &monitoringpb.TypedValue{
+										Value: &monitoringpb.TypedValue_Int64Value{
+											Int64Value: 5,
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			in: &metricdata.Metric{
+				Descriptor: metricdata.Descriptor{
+					Name:        "custom_resource_two",
+					Description: "This is a test",
+					Unit:        metricdata.UnitBytes,
+					Type:        metricdata.TypeCumulativeInt64,
+				},
+				Resource: nil,
+				TimeSeries: []*metricdata.TimeSeries{
+					{
+						StartTime: startTime,
+						Points: []metricdata.Point{
+							{
+								Time:  endTime,
+								Value: int64(5),
+							},
+						},
+					},
+				},
+			},
+			want: []*monitoringpb.CreateTimeSeriesRequest{
+				{
+					Name: "projects/foo",
+					TimeSeries: []*monitoringpb.TimeSeries{
+						{
+							Metric: &googlemetricpb.Metric{
+								Type:   "custom.googleapis.com/opencensus/custom_resource_two",
+								Labels: nil,
+							},
+							Resource: &monitoredrespb.MonitoredResource{
+								Type: "two",
+								Labels: map[string]string{
+									"k21": "v21",
+									"k22": "v22",
+								},
+							},
+							Points: []*monitoringpb.Point{
+								{
+									Interval: &monitoringpb.TimeInterval{
+										StartTime: startTimestamp,
+										EndTime:   endTimestamp,
+									},
+									Value: &monitoringpb.TypedValue{
+										Value: &monitoringpb.TypedValue_Int64Value{
+											Int64Value: 5,
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			in: &metricdata.Metric{
+				Descriptor: metricdata.Descriptor{
+					Name:        "custom_resource_other",
+					Description: "This is a test",
+					Unit:        metricdata.UnitBytes,
+					Type:        metricdata.TypeCumulativeInt64,
+				},
+				Resource: nil,
+				TimeSeries: []*metricdata.TimeSeries{
+					{
+						StartTime: startTime,
+						Points: []metricdata.Point{
+							{
+								Time:  endTime,
+								Value: int64(5),
+							},
+						},
+					},
+				},
+			},
+			want: []*monitoringpb.CreateTimeSeriesRequest{
+				{
+					Name: "projects/foo",
+					TimeSeries: []*monitoringpb.TimeSeries{
+						{
+							Metric: &googlemetricpb.Metric{
+								Type:   "custom.googleapis.com/opencensus/custom_resource_other",
+								Labels: nil,
+							},
+							Resource: &monitoredrespb.MonitoredResource{
+								Type: "global",
+							},
+							Points: []*monitoringpb.Point{
+								{
+									Interval: &monitoringpb.TimeInterval{
+										StartTime: startTimestamp,
+										EndTime:   endTimestamp,
+									},
+									Value: &monitoringpb.TypedValue{
+										Value: &monitoringpb.TypedValue_Int64Value{
+											Int64Value: 5,
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	var se = &statsExporter{
+		o: Options{
+			ProjectID:            "foo",
+			ResourceByDescriptor: testGetMonitoringResource,
+		},
+	}
+
+	for i, tt := range tests {
+		tsl, err := se.metricToMpbTs(context.Background(), tt.in)
+		if tt.wantErr != "" {
+			if err == nil || !strings.Contains(err.Error(), tt.wantErr) {
+				t.Errorf("#%d: unmatched error. Got\n\t%v\nWant\n\t%v", i, err, tt.wantErr)
+			}
+			continue
+		}
+		if err != nil {
+			t.Errorf("#%d: unexpected error: %v", i, err)
+			continue
+		}
+
+		got := se.combineTimeSeriesToCreateTimeSeriesRequest(tsl)
+		// Our saving grace is serialization equality since some
+		// unexported fields could be present in the various values.
+		if diff := cmpTSReqs(got, tt.want); diff != "" {
+			t.Fatalf("Test %d failed. Unexpected CreateTimeSeriesRequests -got +want: %s", i, diff)
+		}
+	}
+}
+
+type customResource struct {
+	rt string
+	rm map[string]string
+}
+
+var _ monitoredresource.Interface = customResource{}
+
+func (cr customResource) MonitoredResource() (resType string, labels map[string]string) {
+	return cr.rt, cr.rm
+}
+
+var crOne = customResource{rt: "one", rm: map[string]string{"k11": "v11", "k12": "v12"}}
+var crTwo = customResource{rt: "two", rm: map[string]string{"k21": "v21", "k22": "v22"}}
+var crEmpty = customResource{rt: ""}
+
+func testGetMonitoringResource(md *metricdata.Descriptor) monitoredresource.Interface {
+	switch md.Name {
+	case "custom_resource_one":
+		return crOne
+	case "custom_resource_two":
+		return crTwo
+	default:
+		return crEmpty
+	}
+}

stackdriver.go

@@ -255,6 +255,17 @@ type Options struct {
 	// to Stackdriver Monitoring. This is only used for Proto metrics export
 	// for now. The minimum number of workers is 1.
 	NumberOfWorkers int
+
+	// ResourceByDescriptor may be provided to supply monitored resource dynamically
+	// based on the metric Descriptor. Most users will not need to set this,
+	// but should instead set ResourceDetector.
+	//
+	// The MonitoredResource and ResourceDetector fields are ignored if this
+	// field is set to a non-nil value.
+	//
+	// If the func set to this field does not return valid resource even for one
+	// metric then it will result into an error for the entire CreateTimeSeries request.
+	ResourceByDescriptor func(*metricdata.Descriptor) monitoredresource.Interface
 }
 
 const defaultTimeout = 5 * time.Second