Source: metrics_proto.go in package contrib.go.opencensus.io/exporter/stackdriver

Copyright 2018, 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 stackdriver

The code in this file is responsible for converting OpenCensus Proto metricsdirectly to Stackdriver Metrics.


import (
	"context"
	"errors"
	"fmt"
	"path"
	"strings"

	"go.opencensus.io/resource"

	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"
	timestamppb "github.com/golang/protobuf/ptypes/timestamp"
	distributionpb "google.golang.org/genproto/googleapis/api/distribution"
	labelpb "google.golang.org/genproto/googleapis/api/label"
	googlemetricpb "google.golang.org/genproto/googleapis/api/metric"
	monitoredrespb "google.golang.org/genproto/googleapis/api/monitoredres"
	monitoringpb "google.golang.org/genproto/googleapis/monitoring/v3"
)

var errNilMetricOrMetricDescriptor = errors.New("non-nil metric or metric descriptor")
var percentileLabelKey = &metricspb.LabelKey{
	Key:         "percentile",
	Description: "the value at a given percentile of a distribution",
}
var globalResource = &resource.Resource{Type: "global"}
var domains = []string{"googleapis.com", "kubernetes.io", "istio.io", "knative.dev"}

PushMetricsProto exports OpenCensus Metrics Proto to Stackdriver Monitoring synchronously, without de-duping or adding proto metrics to the bundler.

func (se *statsExporter) PushMetricsProto(ctx context.Context, node *commonpb.Node, rsc *resourcepb.Resource, metrics []*metricspb.Metric) (int, error) {
	if len(metrics) == 0 {
		return 0, errNilMetricOrMetricDescriptor
	}

Caches the resources seen so far

	seenResources := make(map[*resourcepb.Resource]*monitoredrespb.MonitoredResource)

	mb := newMetricsBatcher(ctx, se.o.ProjectID, se.o.NumberOfWorkers, se.c, se.o.Timeout)
	for _, metric := range metrics {

No TimeSeries to export, skip this metric.

			continue
		}
		mappedRsc := se.getResource(rsc, metric, seenResources)
		if metric.GetMetricDescriptor().GetType() == metricspb.MetricDescriptor_SUMMARY {
			summaryMtcs := se.convertSummaryMetrics(metric)
			for _, summaryMtc := range summaryMtcs {
				if err := se.createMetricDescriptorFromMetricProto(ctx, summaryMtc); err != nil {
					mb.recordDroppedTimeseries(len(summaryMtc.GetTimeseries()), err)
					continue
				}
				se.protoMetricToTimeSeries(ctx, mappedRsc, summaryMtc, mb)
			}
		} else {
			if err := se.createMetricDescriptorFromMetricProto(ctx, metric); err != nil {
				mb.recordDroppedTimeseries(len(metric.GetTimeseries()), err)
				continue
			}
			se.protoMetricToTimeSeries(ctx, mappedRsc, metric, mb)
		}
	}

	return mb.droppedTimeSeries, mb.close(ctx)
}

func (se *statsExporter) convertSummaryMetrics(summary *metricspb.Metric) []*metricspb.Metric {
	var metrics []*metricspb.Metric

	for _, ts := range summary.Timeseries {
		var percentileTss []*metricspb.TimeSeries
		var countTss []*metricspb.TimeSeries
		var sumTss []*metricspb.TimeSeries
		lvs := ts.GetLabelValues()

		startTime := ts.StartTimestamp
		for _, pt := range ts.GetPoints() {
			ptTimestamp := pt.GetTimestamp()
			summaryValue := pt.GetSummaryValue()
			if summaryValue.Sum != nil {
				sumTs := &metricspb.TimeSeries{
					LabelValues:    lvs,
					StartTimestamp: startTime,
					Points: []*metricspb.Point{
						{
							Value: &metricspb.Point_DoubleValue{
								DoubleValue: summaryValue.Sum.Value,
							},
							Timestamp: ptTimestamp,
						},
					},
				}
				sumTss = append(sumTss, sumTs)
			}

			if summaryValue.Count != nil {
				countTs := &metricspb.TimeSeries{
					LabelValues:    lvs,
					StartTimestamp: startTime,
					Points: []*metricspb.Point{
						{
							Value: &metricspb.Point_Int64Value{
								Int64Value: summaryValue.Count.Value,
							},
							Timestamp: ptTimestamp,
						},
					},
				}
				countTss = append(countTss, countTs)
			}

			snapshot := summaryValue.GetSnapshot()
			for _, percentileValue := range snapshot.GetPercentileValues() {
				lvsWithPercentile := lvs[0:]
				lvsWithPercentile = append(lvsWithPercentile, &metricspb.LabelValue{
					HasValue: true,
					Value:    fmt.Sprintf("%f", percentileValue.Percentile),
				})
				percentileTs := &metricspb.TimeSeries{
					LabelValues:    lvsWithPercentile,
					StartTimestamp: nil,
					Points: []*metricspb.Point{
						{
							Value: &metricspb.Point_DoubleValue{
								DoubleValue: percentileValue.Value,
							},
							Timestamp: ptTimestamp,
						},
					},
				}
				percentileTss = append(percentileTss, percentileTs)
			}
		}

		if len(sumTss) > 0 {
			metric := &metricspb.Metric{
				MetricDescriptor: &metricspb.MetricDescriptor{
					Name:        fmt.Sprintf("%s_summary_sum", summary.GetMetricDescriptor().GetName()),
					Description: summary.GetMetricDescriptor().GetDescription(),
					Type:        metricspb.MetricDescriptor_CUMULATIVE_DOUBLE,
					Unit:        summary.GetMetricDescriptor().GetUnit(),
					LabelKeys:   summary.GetMetricDescriptor().GetLabelKeys(),
				},
				Timeseries: sumTss,
				Resource:   summary.Resource,
			}
			metrics = append(metrics, metric)
		}
		if len(countTss) > 0 {
			metric := &metricspb.Metric{
				MetricDescriptor: &metricspb.MetricDescriptor{
					Name:        fmt.Sprintf("%s_summary_count", summary.GetMetricDescriptor().GetName()),
					Description: summary.GetMetricDescriptor().GetDescription(),
					Type:        metricspb.MetricDescriptor_CUMULATIVE_INT64,
					Unit:        "1",
					LabelKeys:   summary.GetMetricDescriptor().GetLabelKeys(),
				},
				Timeseries: countTss,
				Resource:   summary.Resource,
			}
			metrics = append(metrics, metric)
		}
		if len(percentileTss) > 0 {
			lks := summary.GetMetricDescriptor().GetLabelKeys()[0:]
			lks = append(lks, percentileLabelKey)
			metric := &metricspb.Metric{
				MetricDescriptor: &metricspb.MetricDescriptor{
					Name:        fmt.Sprintf("%s_summary_percentile", summary.GetMetricDescriptor().GetName()),
					Description: summary.GetMetricDescriptor().GetDescription(),
					Type:        metricspb.MetricDescriptor_GAUGE_DOUBLE,
					Unit:        summary.GetMetricDescriptor().GetUnit(),
					LabelKeys:   lks,
				},
				Timeseries: percentileTss,
				Resource:   summary.Resource,
			}
			metrics = append(metrics, metric)
		}
	}
	return metrics
}

func (se *statsExporter) getResource(rsc *resourcepb.Resource, metric *metricspb.Metric, seenRscs map[*resourcepb.Resource]*monitoredrespb.MonitoredResource) *monitoredrespb.MonitoredResource {
	var resource = rsc
	if metric.Resource != nil {
		resource = metric.Resource
	}
	mappedRsc, ok := seenRscs[resource]
	if !ok {
		mappedRsc = se.o.MapResource(resourcepbToResource(resource))
		seenRscs[resource] = mappedRsc
	}
	return mappedRsc
}

func resourcepbToResource(rsc *resourcepb.Resource) *resource.Resource {
	if rsc == nil {
		return globalResource
	}
	res := &resource.Resource{
		Type:   rsc.Type,
		Labels: make(map[string]string, len(rsc.Labels)),
	}

	for k, v := range rsc.Labels {
		res.Labels[k] = v
	}
	return res
}

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, mappedRsc *monitoredrespb.MonitoredResource, metric *metricspb.Metric, mb *metricsBatcher) {
	if metric == nil || metric.MetricDescriptor == nil {
		mb.recordDroppedTimeseries(len(metric.GetTimeseries()), errNilMetricOrMetricDescriptor)
	}

	metricType := se.metricTypeFromProto(metric.GetMetricDescriptor().GetName())
	metricLabelKeys := metric.GetMetricDescriptor().GetLabelKeys()
	metricKind, valueType := protoMetricDescriptorTypeToMetricKind(metric)
	labelKeys := make([]string, 0, len(metricLabelKeys))
	for _, key := range metricLabelKeys {
		labelKeys = append(labelKeys, sanitize(key.GetKey()))
	}

	for _, protoTimeSeries := range metric.Timeseries {

No points to send just move forward.

			continue
		}

		sdPoints, err := se.protoTimeSeriesToMonitoringPoints(protoTimeSeries, metricKind)
		if err != nil {
			mb.recordDroppedTimeseries(1, err)
			continue
		}

Each TimeSeries has labelValues which MUST be correlated with that from the MetricDescriptor

		labels, err := labelsPerTimeSeries(se.defaultLabels, labelKeys, protoTimeSeries.GetLabelValues())
		if err != nil {
			mb.recordDroppedTimeseries(1, err)
			continue
		}
		mb.addTimeSeries(&monitoringpb.TimeSeries{
			Metric: &googlemetricpb.Metric{
				Type:   metricType,
				Labels: labels,
			},
			MetricKind: metricKind,
			ValueType:  valueType,
			Resource:   mappedRsc,
			Points:     sdPoints,
		})
	}
}

func labelsPerTimeSeries(defaults map[string]labelValue, labelKeys []string, labelValues []*metricspb.LabelValue) (map[string]string, error) {
	if len(labelKeys) != len(labelValues) {
		return nil, fmt.Errorf("length mismatch: len(labelKeys)=%d len(labelValues)=%d", len(labelKeys), len(labelValues))
	}

No labels for this metric

		return nil, nil
	}

Fill in the defaults firstly, irrespective of if the labelKeys and labelValues are mismatched.

	for key, label := range defaults {
		labels[key] = label.val
	}

	for i, labelKey := range labelKeys {
		labelValue := labelValues[i]
		if !labelValue.GetHasValue() {
			continue
		}
		labels[labelKey] = labelValue.GetValue()
	}

	return labels, nil
}

Skip create metric descriptor if configured

	if se.o.SkipCMD {
		return nil
	}

	ctx, cancel := newContextWithTimeout(ctx, se.o.Timeout)
	defer cancel()

	se.protoMu.Lock()
	defer se.protoMu.Unlock()

	name := metric.GetMetricDescriptor().GetName()
	if _, created := se.protoMetricDescriptors[name]; created {
		return nil
	}

	if builtinMetric(se.metricTypeFromProto(name)) {
		se.protoMetricDescriptors[name] = true
		return nil
	}

Otherwise, we encountered a cache-miss and should create the metric descriptor remotely.

	inMD, err := se.protoToMonitoringMetricDescriptor(metric, se.defaultLabels)
	if err != nil {
		return err
	}

	if err = se.createMetricDescriptor(ctx, inMD); err != nil {
		return err
	}

	se.protoMetricDescriptors[name] = true
	return nil
}

func (se *statsExporter) protoTimeSeriesToMonitoringPoints(ts *metricspb.TimeSeries, metricKind googlemetricpb.MetricDescriptor_MetricKind) ([]*monitoringpb.Point, error) {
	sptl := make([]*monitoringpb.Point, 0, len(ts.Points))

If we have a last value aggregation point i.e. MetricDescriptor_GAUGE StartTime should be nil.

		startTime := ts.StartTimestamp
		if metricKind == googlemetricpb.MetricDescriptor_GAUGE {
			startTime = nil
		}
		spt, err := fromProtoPoint(startTime, pt)
		if err != nil {
			return nil, err
		}
		sptl = append(sptl, spt)
	}
	return sptl, nil
}

func (se *statsExporter) protoToMonitoringMetricDescriptor(metric *metricspb.Metric, additionalLabels map[string]labelValue) (*googlemetricpb.MetricDescriptor, error) {
	if metric == nil || metric.MetricDescriptor == nil {
		return nil, errNilMetricOrMetricDescriptor
	}

	md := metric.GetMetricDescriptor()
	metricName := md.GetName()
	unit := md.GetUnit()
	description := md.GetDescription()
	metricType := se.metricTypeFromProto(metricName)
	displayName := se.displayName(metricName)
	metricKind, valueType := protoMetricDescriptorTypeToMetricKind(metric)

	sdm := &googlemetricpb.MetricDescriptor{
		Name:        fmt.Sprintf("projects/%s/metricDescriptors/%s", se.o.ProjectID, metricType),
		DisplayName: displayName,
		Description: description,
		Unit:        unit,
		Type:        metricType,
		MetricKind:  metricKind,
		ValueType:   valueType,
		Labels:      labelDescriptorsFromProto(additionalLabels, metric.GetMetricDescriptor().GetLabelKeys()),
	}

	return sdm, nil
}

func labelDescriptorsFromProto(defaults map[string]labelValue, protoLabelKeys []*metricspb.LabelKey) []*labelpb.LabelDescriptor {
	labelDescriptors := make([]*labelpb.LabelDescriptor, 0, len(defaults)+len(protoLabelKeys))

Fill in the defaults first.

	for key, lbl := range defaults {
		labelDescriptors = append(labelDescriptors, &labelpb.LabelDescriptor{
			Key:         sanitize(key),
			Description: lbl.desc,
			ValueType:   labelpb.LabelDescriptor_STRING,
		})
	}

Now fill in those from the metric.

	for _, protoKey := range protoLabelKeys {
		labelDescriptors = append(labelDescriptors, &labelpb.LabelDescriptor{
			Key:         sanitize(protoKey.GetKey()),
			Description: protoKey.GetDescription(),
			ValueType:   labelpb.LabelDescriptor_STRING, // We only use string tags
		})
	}
	return labelDescriptors
}

func (se *statsExporter) metricTypeFromProto(name string) string {
	prefix := se.o.MetricPrefix
	if se.o.GetMetricPrefix != nil {
		prefix = se.o.GetMetricPrefix(name)
	}
	if prefix != "" {
		name = path.Join(prefix, name)
	}

Still needed because the name may or may not have a "/" at the beginning.

		name = path.Join(defaultDomain, name)
	}
	return name
}

hasDomain checks if the metric name already has a domain in it.

func hasDomain(name string) bool {
	for _, domain := range domains {
		if strings.Contains(name, domain) {
			return true
		}
	}
	return false
}

func fromProtoPoint(startTime *timestamppb.Timestamp, pt *metricspb.Point) (*monitoringpb.Point, error) {
	if pt == nil {
		return nil, nil
	}

	mptv, err := protoToMetricPoint(pt.Value)
	if err != nil {
		return nil, err
	}

	return &monitoringpb.Point{
		Value: mptv,
		Interval: &monitoringpb.TimeInterval{
			StartTime: startTime,
			EndTime:   pt.Timestamp,
		},
	}, nil
}

func protoToMetricPoint(value interface{}) (*monitoringpb.TypedValue, error) {
	if value == nil {
		return nil, nil
	}

	switch v := value.(type) {

All the other types are not yet handled. TODO: (@odeke-em, @songy23) talk to the Stackdriver team to determine the use cases for: *TypedValue_BoolValue *TypedValue_StringValue and then file feature requests on OpenCensus-Specs and then OpenCensus-Proto, lest we shall error here. TODO: Add conversion from SummaryValue when https://github.com/census-ecosystem/opencensus-go-exporter-stackdriver/issues/66 has been figured out.

		return nil, fmt.Errorf("protoToMetricPoint: unknown Data type: %T", value)

	case *metricspb.Point_Int64Value:
		return &monitoringpb.TypedValue{
			Value: &monitoringpb.TypedValue_Int64Value{
				Int64Value: v.Int64Value,
			},
		}, nil

	case *metricspb.Point_DoubleValue:
		return &monitoringpb.TypedValue{
			Value: &monitoringpb.TypedValue_DoubleValue{
				DoubleValue: v.DoubleValue,
			},
		}, nil

	case *metricspb.Point_DistributionValue:
		dv := v.DistributionValue
		var mv *monitoringpb.TypedValue_DistributionValue
		if dv != nil {
			var mean float64
			if dv.Count > 0 {
				mean = float64(dv.Sum) / float64(dv.Count)
			}
			mv = &monitoringpb.TypedValue_DistributionValue{
				DistributionValue: &distributionpb.Distribution{
					Count:                 dv.Count,
					Mean:                  mean,
					SumOfSquaredDeviation: dv.SumOfSquaredDeviation,
				},
			}

			insertZeroBound := false
			if bopts := dv.BucketOptions; bopts != nil && bopts.Type != nil {
				bexp, ok := bopts.Type.(*metricspb.DistributionValue_BucketOptions_Explicit_)
				if ok && bexp != nil && bexp.Explicit != nil {
					insertZeroBound = shouldInsertZeroBound(bexp.Explicit.Bounds...)
					mv.DistributionValue.BucketOptions = &distributionpb.Distribution_BucketOptions{
						Options: &distributionpb.Distribution_BucketOptions_ExplicitBuckets{

The first bucket bound should be 0.0 because the Metrics first bucket is [0, first_bound) but Stackdriver monitoring bucket bounds begin with -infinity (first bucket is (-infinity, 0))

								Bounds: addZeroBoundOnCondition(insertZeroBound, bexp.Explicit.Bounds...),
							},
						},
					}
				}
			}
			mv.DistributionValue.BucketCounts = addZeroBucketCountOnCondition(insertZeroBound, bucketCounts(dv.Buckets)...)

		}
		return &monitoringpb.TypedValue{Value: mv}, nil
	}
}

func bucketCounts(buckets []*metricspb.DistributionValue_Bucket) []int64 {
	bucketCounts := make([]int64, len(buckets))
	for i, bucket := range buckets {
		if bucket != nil {
			bucketCounts[i] = bucket.Count
		}
	}
	return bucketCounts
}

func protoMetricDescriptorTypeToMetricKind(m *metricspb.Metric) (googlemetricpb.MetricDescriptor_MetricKind, googlemetricpb.MetricDescriptor_ValueType) {
	dt := m.GetMetricDescriptor()
	if dt == nil {
		return googlemetricpb.MetricDescriptor_METRIC_KIND_UNSPECIFIED, googlemetricpb.MetricDescriptor_VALUE_TYPE_UNSPECIFIED
	}

	switch dt.Type {
	case metricspb.MetricDescriptor_CUMULATIVE_INT64:
		return googlemetricpb.MetricDescriptor_CUMULATIVE, googlemetricpb.MetricDescriptor_INT64

	case metricspb.MetricDescriptor_CUMULATIVE_DOUBLE:
		return googlemetricpb.MetricDescriptor_CUMULATIVE, googlemetricpb.MetricDescriptor_DOUBLE

	case metricspb.MetricDescriptor_CUMULATIVE_DISTRIBUTION:
		return googlemetricpb.MetricDescriptor_CUMULATIVE, googlemetricpb.MetricDescriptor_DISTRIBUTION

	case metricspb.MetricDescriptor_GAUGE_DOUBLE:
		return googlemetricpb.MetricDescriptor_GAUGE, googlemetricpb.MetricDescriptor_DOUBLE

	case metricspb.MetricDescriptor_GAUGE_INT64:
		return googlemetricpb.MetricDescriptor_GAUGE, googlemetricpb.MetricDescriptor_INT64

	case metricspb.MetricDescriptor_GAUGE_DISTRIBUTION:
		return googlemetricpb.MetricDescriptor_GAUGE, googlemetricpb.MetricDescriptor_DISTRIBUTION

	default:
		return googlemetricpb.MetricDescriptor_METRIC_KIND_UNSPECIFIED, googlemetricpb.MetricDescriptor_VALUE_TYPE_UNSPECIFIED
	}