Add exporterhelper to wrap all exporters with observability. (#484)

* Add exporterhelper to wrap all exporters with observability.

* Remove the nop view exporter, no needed.

* Move constants in constants.go, add TODO to handle gRPC errors.

* Don't export the errors exporterhelper returns.

cmd/occollector/app/collector/processors.go

@@ -248,7 +248,8 @@ func startProcessor(v *viper.Viper, logger *zap.Logger) (processor.SpanProcessor
 	_ = metricsExporters
 
 	if builder.LoggingExporterEnabled(v) {
-		dbgProc := processor.NewTraceExporterProcessor(loggingexporter.NewTraceExporter(logger))
+		tle, _ := loggingexporter.NewTraceExporter(logger)
+		dbgProc := processor.NewTraceExporterProcessor(tle)
 		// TODO: Add this to the exporters list and avoid treating it specially. Don't know all the implications.
 		nameToSpanProcessor["debug"] = dbgProc
 		spanProcessors = append(spanProcessors, dbgProc)

exporter/exporterhelper/common.go

@@ -0,0 +1,66 @@
+// 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 exporterhelper
+
+import (
+	"go.opencensus.io/trace"
+)
+
+var (
+	okStatus = trace.Status{Code: trace.StatusCodeOK}
+)
+
+// ExporterOptions contains options concerning how an Exporter is configured.
+type ExporterOptions struct {
+	// TOOD: Retry logic must be in the same place as metrics recording because
+	// if a request is retried we should not record metrics otherwise number of
+	// spans received + dropped will be different than the number of received spans
+	// in the receiver.
+	recordMetrics bool
+	spanName      string
+}
+
+// ExporterOption apply changes to ExporterOptions.
+type ExporterOption func(*ExporterOptions)
+
+// WithRecordMetrics makes new Exporter to record metrics for every request.
+func WithRecordMetrics(recordMetrics bool) ExporterOption {
+	return func(o *ExporterOptions) {
+		o.recordMetrics = recordMetrics
+	}
+}
+
+// WithSpanName makes new Exporter to wrap every request with a Span.
+func WithSpanName(spanName string) ExporterOption {
+	return func(o *ExporterOptions) {
+		o.spanName = spanName
+	}
+}
+
+// Construct the ExporterOptions from multiple ExporterOption.
+func newExporterOptions(options ...ExporterOption) ExporterOptions {
+	var opts ExporterOptions
+	for _, op := range options {
+		op(&opts)
+	}
+	return opts
+}
+
+func errToStatus(err error) trace.Status {
+	if err != nil {
+		return trace.Status{Code: trace.StatusCodeUnknown, Message: err.Error()}
+	}
+	return okStatus
+}

exporter/exporterhelper/common_test.go

@@ -0,0 +1,61 @@
+// 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 exporterhelper
+
+import (
+	"errors"
+	"testing"
+
+	"go.opencensus.io/trace"
+)
+
+func TestDefaultOptions(t *testing.T) {
+	checkRecordMetrics(t, newExporterOptions(), false)
+	checkSpanName(t, newExporterOptions(), "")
+}
+
+func TestWithRecordMetrics(t *testing.T) {
+	checkRecordMetrics(t, newExporterOptions(WithRecordMetrics(true)), true)
+	checkRecordMetrics(t, newExporterOptions(WithRecordMetrics(false)), false)
+}
+
+func TestWithSpanName(t *testing.T) {
+	checkSpanName(t, newExporterOptions(WithSpanName("my_span")), "my_span")
+	checkSpanName(t, newExporterOptions(WithSpanName("")), "")
+}
+
+func TestErrorToStatus(t *testing.T) {
+	if g, w := errToStatus(nil), okStatus; g != w {
+		t.Fatalf("Status: Want %v Got %v", w, g)
+	}
+
+	errorStatus := trace.Status{Code: trace.StatusCodeUnknown, Message: "my_error"}
+	if g, w := errToStatus(errors.New("my_error")), errorStatus; g != w {
+		t.Fatalf("Status: Want %v Got %v", w, g)
+	}
+}
+
+func checkRecordMetrics(t *testing.T, opts ExporterOptions, recordMetrics bool) {
+	if opts.recordMetrics != recordMetrics {
+		t.Errorf("Wrong recordMetrics Want: %t Got: %t", opts.recordMetrics, recordMetrics)
+		return
+	}
+}
+
+func checkSpanName(t *testing.T, opts ExporterOptions, spanName string) {
+	if opts.spanName != spanName {
+		t.Errorf("Wrong spanName Want: %s Got: %s", opts.spanName, spanName)
+		return
+	}
+}

exporter/exporterhelper/constants.go

@@ -0,0 +1,35 @@
+// 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 exporterhelper
+
+import (
+	"errors"
+)
+
+var (
+	// errEmptyExporterFormat is returned when an empty name is given.
+	errEmptyExporterFormat = errors.New("empty exporter format")
+	// errNilPushTraceData is returned when a nil pushTraceData is given.
+	errNilPushTraceData = errors.New("nil pushTraceData")
+	// errNilPushMetricsData is returned when a nil pushMetricsData is given.
+	errNilPushMetricsData = errors.New("nil pushMetricsData")
+)
+
+const (
+	numDroppedMetricsAttribute  = "num_dropped_metrics"
+	numReceivedMetricsAttribute = "num_received_metrics"
+	numDroppedSpansAttribute    = "num_dropped_spans"
+	numReceivedSpansAttribute   = "num_received_spans"
+)

exporter/exporterhelper/metricshelper.go

@@ -0,0 +1,90 @@
+// 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 exporterhelper
+
+import (
+	"context"
+
+	"github.com/census-instrumentation/opencensus-service/observability"
+	"go.opencensus.io/trace"
+
+	"github.com/census-instrumentation/opencensus-service/data"
+	"github.com/census-instrumentation/opencensus-service/exporter"
+)
+
+// PushMetricsData is a helper function that is similar to ConsumeMetricsData but also returns
+// the number of dropped metrics.
+type PushMetricsData func(ctx context.Context, td data.MetricsData) (droppedMetrics int, err error)
+
+type metricsExporter struct {
+	exporterFormat  string
+	pushMetricsData PushMetricsData
+}
+
+var _ (exporter.MetricsExporter) = (*metricsExporter)(nil)
+
+func (me *metricsExporter) MetricsExportFormat() string {
+	return me.exporterFormat
+}
+
+func (me *metricsExporter) ConsumeMetricsData(ctx context.Context, md data.MetricsData) error {
+	exporterCtx := observability.ContextWithExporterName(ctx, me.exporterFormat)
+	_, err := me.pushMetricsData(exporterCtx, md)
+	return err
+}
+
+// NewMetricsExporter creates an MetricsExporter that can record metrics and can wrap every request with a Span.
+// If no options are passed it just adds the exporter format as a tag in the Context.
+// TODO: Add support for recordMetrics.
+// TODO: Add support for retries.
+func NewMetricsExporter(exporterFormat string, pushMetricsData PushMetricsData, options ...ExporterOption) (exporter.MetricsExporter, error) {
+	if exporterFormat == "" {
+		return nil, errEmptyExporterFormat
+	}
+
+	if pushMetricsData == nil {
+		return nil, errNilPushMetricsData
+	}
+
+	opts := newExporterOptions(options...)
+
+	if opts.spanName != "" {
+		pushMetricsData = pushMetricsDataWithSpan(pushMetricsData, opts.spanName)
+	}
+
+	return &metricsExporter{
+		exporterFormat:  exporterFormat,
+		pushMetricsData: pushMetricsData,
+	}, nil
+}
+
+func pushMetricsDataWithSpan(next PushMetricsData, spanName string) PushMetricsData {
+	return func(ctx context.Context, md data.MetricsData) (int, error) {
+		ctx, span := trace.StartSpan(ctx, spanName)
+		defer span.End()
+		// Call next stage.
+		droppedMetrics, err := next(ctx, md)
+		if span.IsRecordingEvents() {
+			span.AddAttributes(
+				trace.Int64Attribute(numReceivedMetricsAttribute, int64(len(md.Metrics))),
+				trace.Int64Attribute(numDroppedMetricsAttribute, int64(droppedMetrics)),
+			)
+			if err != nil {
+				span.SetStatus(errToStatus(err))
+			}
+		}
+		return droppedMetrics, err
+	}
+}