Add opencensus_task labels to proto metrics if no default labels are provided. (#148)

* Add opencensus_task labels to proto metrics if no default labels are provided.

* Make default labels stateless.

Author: songy23

equivalence_test.go

@@ -97,7 +97,7 @@ func TestStatsAndMetricsEquivalence(t *testing.T) {
 		if err != nil {
 			t.Errorf("#%d: Stats.viewToMetricDescriptor: %v", i, err)
 		}
-		pMD, err := se.protoMetricDescriptorToCreateMetricDescriptorRequest(ctx, last.Metrics[0])
+		pMD, err := se.protoMetricDescriptorToCreateMetricDescriptorRequest(ctx, last.Metrics[0], nil)
 		if err != nil {
 			t.Errorf("#%d: Stats.protoMetricDescriptorToMetricDescriptor: %v", i, err)
 		}
@@ -107,7 +107,7 @@ func TestStatsAndMetricsEquivalence(t *testing.T) {
 
 		vdl := []*view.Data{vd}
 		sctreql := se.makeReq(vdl, maxTimeSeriesPerUpload)
-		tsl, _ := se.protoMetricToTimeSeries(ctx, last.Node, last.Resource, last.Metrics[0])
+		tsl, _ := se.protoMetricToTimeSeries(ctx, last.Node, last.Resource, last.Metrics[0], nil)
 		pctreql := se.combineTimeSeriesToCreateTimeSeriesRequest(tsl)
 		if diff := cmpTSReqs(pctreql, sctreql); diff != "" {
 			t.Fatalf("TimeSeries Mismatch -FromMetrics +FromStats: %s", diff)

metrics_proto.go

@@ -47,9 +47,10 @@ var errNilMetric = errors.New("expecting a non-nil metric")
 var errNilMetricDescriptor = errors.New("expecting a non-nil metric descriptor")
 
 type metricProtoPayload struct {
-	node     *commonpb.Node
-	resource *resourcepb.Resource
-	metric   *metricspb.Metric
+	node             *commonpb.Node
+	resource         *resourcepb.Resource
+	metric           *metricspb.Metric
+	additionalLabels map[string]labelValue
 }
 
 // ExportMetricsProto exports OpenCensus Metrics Proto to Stackdriver Monitoring.
@@ -58,11 +59,18 @@ func (se *statsExporter) ExportMetricsProto(ctx context.Context, node *commonpb.
 		return errNilMetric
 	}
 
+	additionalLabels := se.defaultLabels
+	if additionalLabels == nil {
+		// additionalLabels must be stateless because each node is different
+		additionalLabels = getDefaultLabelsFromNode(node)
+	}
+
 	for _, metric := range metrics {
 		payload := &metricProtoPayload{
-			metric:   metric,
-			resource: rsc,
-			node:     node,
+			metric:           metric,
+			resource:         rsc,
+			node:             node,
+			additionalLabels: additionalLabels,
 		}
 		se.protoMetricsBundler.Add(payload, 1)
 	}
@@ -83,15 +91,15 @@ func (se *statsExporter) handleMetricsProtoUpload(payloads []*metricProtoPayload
 
 	for _, payload := range payloads {
 		// Now create the metric descriptor remotely.
-		if err := se.createMetricDescriptor(ctx, payload.metric); err != nil {
+		if err := se.createMetricDescriptor(ctx, payload.metric, payload.additionalLabels); err != nil {
 			span.SetStatus(trace.Status{Code: 2, Message: err.Error()})
 			return err
 		}
 	}
 
 	var allTimeSeries []*monitoringpb.TimeSeries
 	for _, payload := range payloads {
-		tsl, err := se.protoMetricToTimeSeries(ctx, payload.node, payload.resource, payload.metric)
+		tsl, err := se.protoMetricToTimeSeries(ctx, payload.node, payload.resource, payload.metric, payload.additionalLabels)
 		if err != nil {
 			span.SetStatus(trace.Status{Code: 2, Message: err.Error()})
 			return err
@@ -196,7 +204,7 @@ func (se *statsExporter) combineTimeSeriesToCreateTimeSeriesRequest(ts []*monito
 
 // protoMetricToTimeSeries converts a metric into a Stackdriver Monitoring v3 API CreateTimeSeriesRequest
 // but it doesn't invoke any remote API.
-func (se *statsExporter) protoMetricToTimeSeries(ctx context.Context, node *commonpb.Node, rsc *resourcepb.Resource, metric *metricspb.Metric) ([]*monitoringpb.TimeSeries, error) {
+func (se *statsExporter) protoMetricToTimeSeries(ctx context.Context, node *commonpb.Node, rsc *resourcepb.Resource, metric *metricspb.Metric, additionalLabels map[string]labelValue) ([]*monitoringpb.TimeSeries, error) {
 	if metric == nil {
 		return nil, errNilMetric
 	}
@@ -223,7 +231,7 @@ func (se *statsExporter) protoMetricToTimeSeries(ctx context.Context, node *comm
 
 		// Each TimeSeries has labelValues which MUST be correlated
 		// with that from the MetricDescriptor
-		labels, err := labelsPerTimeSeries(se.defaultLabels, metricLabelKeys, protoTimeSeries.GetLabelValues())
+		labels, err := labelsPerTimeSeries(additionalLabels, metricLabelKeys, protoTimeSeries.GetLabelValues())
 		if err != nil {
 			// TODO: (@odeke-em) perhaps log this error from labels extraction, if non-nil.
 			continue
@@ -261,10 +269,10 @@ func labelsPerTimeSeries(defaults map[string]labelValue, labelKeys []*metricspb.
 	return labels, nil
 }
 
-func (se *statsExporter) protoMetricDescriptorToCreateMetricDescriptorRequest(ctx context.Context, metric *metricspb.Metric) (*monitoringpb.CreateMetricDescriptorRequest, error) {
+func (se *statsExporter) protoMetricDescriptorToCreateMetricDescriptorRequest(ctx context.Context, metric *metricspb.Metric, additionalLabels map[string]labelValue) (*monitoringpb.CreateMetricDescriptorRequest, error) {
 	// Otherwise, we encountered a cache-miss and
 	// should create the metric descriptor remotely.
-	inMD, err := se.protoToMonitoringMetricDescriptor(metric)
+	inMD, err := se.protoToMonitoringMetricDescriptor(metric, additionalLabels)
 	if err != nil {
 		return nil, err
 	}
@@ -279,7 +287,7 @@ func (se *statsExporter) protoMetricDescriptorToCreateMetricDescriptorRequest(ct
 
 // createMetricDescriptor creates a metric descriptor from the OpenCensus proto metric
 // and then creates it remotely using Stackdriver's API.
-func (se *statsExporter) createMetricDescriptor(ctx context.Context, metric *metricspb.Metric) error {
+func (se *statsExporter) createMetricDescriptor(ctx context.Context, metric *metricspb.Metric, additionalLabels map[string]labelValue) error {
 	se.protoMu.Lock()
 	defer se.protoMu.Unlock()
 
@@ -290,7 +298,7 @@ func (se *statsExporter) createMetricDescriptor(ctx context.Context, metric *met
 
 	// Otherwise, we encountered a cache-miss and
 	// should create the metric descriptor remotely.
-	inMD, err := se.protoToMonitoringMetricDescriptor(metric)
+	inMD, err := se.protoToMonitoringMetricDescriptor(metric, additionalLabels)
 	if err != nil {
 		return err
 	}
@@ -337,7 +345,7 @@ func (se *statsExporter) protoTimeSeriesToMonitoringPoints(ts *metricspb.TimeSer
 	return sptl, nil
 }
 
-func (se *statsExporter) protoToMonitoringMetricDescriptor(metric *metricspb.Metric) (*googlemetricpb.MetricDescriptor, error) {
+func (se *statsExporter) protoToMonitoringMetricDescriptor(metric *metricspb.Metric, additionalLabels map[string]labelValue) (*googlemetricpb.MetricDescriptor, error) {
 	if metric == nil {
 		return nil, errNilMetric
 	}
@@ -358,7 +366,7 @@ func (se *statsExporter) protoToMonitoringMetricDescriptor(metric *metricspb.Met
 		Type:        metricType,
 		MetricKind:  metricKind,
 		ValueType:   valueType,
-		Labels:      labelDescriptorsFromProto(se.defaultLabels, metric.GetMetricDescriptor().GetLabelKeys()),
+		Labels:      labelDescriptorsFromProto(additionalLabels, metric.GetMetricDescriptor().GetLabelKeys()),
 	}
 
 	return sdm, nil
@@ -573,3 +581,13 @@ func protoResourceToMonitoredResource(rsp *resourcepb.Resource) *monitoredrespb.
 	}
 	return mrsp
 }
+
+func getDefaultLabelsFromNode(node *commonpb.Node) map[string]labelValue {
+	taskValue := fmt.Sprintf("%s-%d@%s", strings.ToLower(node.LibraryInfo.GetLanguage().String()), node.Identifier.Pid, node.Identifier.HostName)
+	return map[string]labelValue{
+		opencensusTaskKey: {
+			val:  taskValue,
+			desc: opencensusTaskDescription,
+		},
+	}
+}

metrics_proto_test.go

@@ -16,6 +16,7 @@ package stackdriver
 
 import (
 	"context"
+	"reflect"
 	"strings"
 	"testing"
 
@@ -27,6 +28,7 @@ import (
 	monitoredrespb "google.golang.org/genproto/googleapis/api/monitoredres"
 	monitoringpb "google.golang.org/genproto/googleapis/monitoring/v3"
 
+	commonpb "github.com/census-instrumentation/opencensus-proto/gen-go/agent/common/v1"
 	metricspb "github.com/census-instrumentation/opencensus-proto/gen-go/metrics/v1"
 	resourcepb "github.com/census-instrumentation/opencensus-proto/gen-go/resource/v1"
 )
@@ -180,7 +182,7 @@ func TestProtoMetricToCreateTimeSeriesRequest(t *testing.T) {
 		if se == nil {
 			se = new(statsExporter)
 		}
-		tsl, err := se.protoMetricToTimeSeries(context.Background(), nil, nil, tt.in)
+		tsl, err := se.protoMetricToTimeSeries(context.Background(), nil, nil, tt.in, nil)
 		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)
@@ -241,7 +243,7 @@ func TestProtoToMonitoringMetricDescriptor(t *testing.T) {
 		if se == nil {
 			se = new(statsExporter)
 		}
-		got, err := se.protoToMonitoringMetricDescriptor(tt.in)
+		got, err := se.protoToMonitoringMetricDescriptor(tt.in, nil)
 		if tt.wantErr != "" {
 			if err == nil || !strings.Contains(err.Error(), tt.wantErr) {
 				t.Errorf("#%d: \nGot %v\nWanted error substring %q", i, err, tt.wantErr)
@@ -474,3 +476,42 @@ func TestCombineTimeSeriesAndDeduplication(t *testing.T) {
 		}
 	}
 }
+
+func TestNodeToDefaultLabels(t *testing.T) {
+	tests := []struct {
+		in   *commonpb.Node
+		want map[string]labelValue
+	}{
+		{
+			in: &commonpb.Node{
+				Identifier:  &commonpb.ProcessIdentifier{HostName: "host1", Pid: 8081},
+				LibraryInfo: &commonpb.LibraryInfo{Language: commonpb.LibraryInfo_JAVA},
+			},
+			want: map[string]labelValue{
+				"opencensus_task": {
+					val:  "java-8081@host1",
+					desc: "Opencensus task identifier",
+				},
+			},
+		},
+		{
+			in: &commonpb.Node{
+				Identifier:  &commonpb.ProcessIdentifier{HostName: "host2", Pid: 9090},
+				LibraryInfo: &commonpb.LibraryInfo{Language: commonpb.LibraryInfo_PYTHON},
+			},
+			want: map[string]labelValue{
+				"opencensus_task": {
+					val:  "python-9090@host2",
+					desc: "Opencensus task identifier",
+				},
+			},
+		},
+	}
+
+	for i, tt := range tests {
+		got := getDefaultLabelsFromNode(tt.in)
+		if !reflect.DeepEqual(got, tt.want) {
+			t.Fatalf("Test %d failed. Default labels mismatch. Want %v\nGot %v\n", i, tt.want, got)
+		}
+	}
+}