Move observability, add test helper for observability and record metrics in loggingexporter. (#459)

Author: Bogdan Drutu

cmd/ocagent/main.go

@@ -34,11 +34,11 @@ import (
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 
-	"github.com/census-instrumentation/opencensus-service/internal"
 	"github.com/census-instrumentation/opencensus-service/internal/config"
 	"github.com/census-instrumentation/opencensus-service/internal/config/viperutils"
 	"github.com/census-instrumentation/opencensus-service/internal/pprofserver"
 	"github.com/census-instrumentation/opencensus-service/internal/version"
+	"github.com/census-instrumentation/opencensus-service/observability"
 	"github.com/census-instrumentation/opencensus-service/processor"
 	"github.com/census-instrumentation/opencensus-service/receiver/jaegerreceiver"
 	"github.com/census-instrumentation/opencensus-service/receiver/opencensusreceiver"
@@ -230,7 +230,7 @@ func runOCReceiver(logger *zap.Logger, acfg *config.Config, tdp processor.TraceD
 	if err != nil {
 		return nil, fmt.Errorf("Failed to create the OpenCensus receiver on address %q: error %v", addr, err)
 	}
-	if err := view.Register(internal.AllViews...); err != nil {
+	if err := view.Register(observability.AllViews...); err != nil {
 		return nil, fmt.Errorf("Failed to register internal.AllViews: %v", err)
 	}
 

cmd/occollector/app/collector/telemetry.go

@@ -25,12 +25,12 @@ import (
 	"go.opencensus.io/stats/view"
 	"go.uber.org/zap"
 
-	"github.com/census-instrumentation/opencensus-service/internal"
 	"github.com/census-instrumentation/opencensus-service/internal/collector/processor"
 	"github.com/census-instrumentation/opencensus-service/internal/collector/processor/nodebatcher"
 	"github.com/census-instrumentation/opencensus-service/internal/collector/processor/queued"
 	"github.com/census-instrumentation/opencensus-service/internal/collector/processor/tailsampling"
 	"github.com/census-instrumentation/opencensus-service/internal/collector/telemetry"
+	"github.com/census-instrumentation/opencensus-service/observability"
 )
 
 const (
@@ -59,7 +59,7 @@ func initTelemetry(asyncErrorChannel chan<- error, v *viper.Viper, logger *zap.L
 	views := processor.MetricViews(level)
 	views = append(views, queued.MetricViews(level)...)
 	views = append(views, nodebatcher.MetricViews(level)...)
-	views = append(views, internal.AllViews...)
+	views = append(views, observability.AllViews...)
 	views = append(views, tailsampling.SamplingProcessorMetricViews(level)...)
 	processMetricsViews := telemetry.NewProcessMetricsViews()
 	views = append(views, processMetricsViews.Views()...)

exporter/exportertest/exportertest.go

@@ -0,0 +1,25 @@
+// 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 exportertest
+
+import (
+	"encoding/json"
+)
+
+// ToJSON marshals a generic interface to JSON to enable easy comparisons.
+func ToJSON(v interface{}) []byte {
+	b, _ := json.MarshalIndent(v, "", "  ")
+	return b
+}

exporter/exportertest/sink_exporter.go

@@ -16,7 +16,6 @@ package exportertest
 
 import (
 	"context"
-	"encoding/json"
 	"sync"
 
 	"github.com/census-instrumentation/opencensus-service/data"
@@ -86,9 +85,3 @@ func (sme *SinkMetricsExporter) AllMetrics() []data.MetricsData {
 
 	return sme.metrics[:]
 }
-
-// ToJSON marshals a generic interface to JSON to enable easy comparisons.
-func ToJSON(v interface{}) []byte {
-	b, _ := json.MarshalIndent(v, "", "  ")
-	return b
-}

exporter/loggingexporter/logging_exporter.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/census-instrumentation/opencensus-service/data"
 	"github.com/census-instrumentation/opencensus-service/exporter"
+	"github.com/census-instrumentation/opencensus-service/observability"
 	"go.uber.org/zap"
 )
 
@@ -30,21 +31,23 @@ type loggingExporter struct{ logger *zap.Logger }
 var _ exporter.TraceExporter = (*loggingExporter)(nil)
 var _ exporter.MetricsExporter = (*loggingExporter)(nil)
 
-func (sp *loggingExporter) ProcessTraceData(ctx context.Context, td data.TraceData) error {
-	// TODO: Record metrics
+func (le *loggingExporter) ProcessTraceData(ctx context.Context, td data.TraceData) error {
+	le.logger.Debug("loggingTraceExporter", zap.Int("#spans", len(td.Spans)))
 	// TODO: Add ability to record the received data
-	sp.logger.Debug("loggingTraceExporter", zap.Int("#spans", len(td.Spans)))
+
+	// Even though we just log all the spans, we record 0 dropped spans.
+	observability.RecordTraceExporterMetrics(observability.ContextWithExporterName(ctx, "logging_trace"), len(td.Spans), 0)
 	return nil
 }
 
-func (sp *loggingExporter) ProcessMetricsData(ctx context.Context, md data.MetricsData) error {
-	sp.logger.Debug("loggingMetricsExporter", zap.Int("#metrics", len(md.Metrics)))
-	// TODO: Record metrics
+func (le *loggingExporter) ProcessMetricsData(ctx context.Context, md data.MetricsData) error {
+	le.logger.Debug("loggingMetricsExporter", zap.Int("#metrics", len(md.Metrics)))
 	// TODO: Add ability to record the received data
+	// TODO: Record metrics
 	return nil
 }
 
-func (sp *loggingExporter) ExportFormat() string {
+func (le *loggingExporter) ExportFormat() string {
 	return exportFormat
 }
 

exporter/loggingexporter/logging_exporter_test.go

@@ -20,33 +20,39 @@ import (
 	metricspb "github.com/census-instrumentation/opencensus-proto/gen-go/metrics/v1"
 	tracepb "github.com/census-instrumentation/opencensus-proto/gen-go/trace/v1"
 	"github.com/census-instrumentation/opencensus-service/data"
+	"github.com/census-instrumentation/opencensus-service/observability/observabilitytest"
 	"go.uber.org/zap"
 )
 
 func TestLoggingTraceExporterNoErrors(t *testing.T) {
-	dtdp := NewTraceExporter(zap.NewNop())
+	lte := NewTraceExporter(zap.NewNop())
 	td := data.TraceData{
 		Spans: make([]*tracepb.Span, 7),
 	}
-	if err := dtdp.ProcessTraceData(context.Background(), td); err != nil {
+	if err := lte.ProcessTraceData(context.Background(), td); err != nil {
 		t.Errorf("Wanted nil got error")
 		return
 	}
-	if "LoggingExporter" != dtdp.ExportFormat() {
-		t.Errorf("Wanted LoggingExporter got %v", dtdp.ExportFormat())
+	if "LoggingExporter" != lte.ExportFormat() {
+		t.Errorf("Wanted LoggingExporter got %v", lte.ExportFormat())
 	}
 }
 
+func TestLoggingTraceExporterRecordMetrics(t *testing.T) {
+	lte := NewTraceExporter(zap.NewNop())
+	observabilitytest.CheckRecordedMetricsForTraceExporter(t, lte, "logging_trace")
+}
+
 func TestLoggingMetricsExporterNoErrors(t *testing.T) {
-	dmdp := NewMetricsExporter(zap.NewNop())
+	lme := NewMetricsExporter(zap.NewNop())
 	md := data.MetricsData{
 		Metrics: make([]*metricspb.Metric, 7),
 	}
-	if err := dmdp.ProcessMetricsData(context.Background(), md); err != nil {
+	if err := lme.ProcessMetricsData(context.Background(), md); err != nil {
 		t.Errorf("Wanted nil got error")
 		return
 	}
-	if "LoggingExporter" != dmdp.ExportFormat() {
-		t.Errorf("Wanted LoggingExporter got %v", dmdp.ExportFormat())
+	if "LoggingExporter" != lme.ExportFormat() {
+		t.Errorf("Wanted LoggingExporter got %v", lme.ExportFormat())
 	}
 }

exporter/opencensusexporter/opencensus.go

@@ -24,9 +24,9 @@ import (
 
 	agenttracepb "github.com/census-instrumentation/opencensus-proto/gen-go/agent/trace/v1"
 	"github.com/census-instrumentation/opencensus-service/data"
-	"github.com/census-instrumentation/opencensus-service/internal"
 	"github.com/census-instrumentation/opencensus-service/internal/compression"
 	"github.com/census-instrumentation/opencensus-service/internal/compression/grpc"
+	"github.com/census-instrumentation/opencensus-service/observability"
 	"github.com/census-instrumentation/opencensus-service/processor"
 )
 
@@ -117,14 +117,14 @@ func (oce *ocagentExporter) ProcessTraceData(ctx context.Context, td data.TraceD
 			Node:     td.Node,
 		},
 	)
-	ctxWithExporterName := internal.ContextWithExporterName(ctx, exporterTagValue)
+	ctxWithExporterName := observability.ContextWithExporterName(ctx, exporterTagValue)
 	if err != nil {
 		// TODO: If failed to send all maybe record a different metric. Failed to "Sent", but
 		// this may not be accurate if we have retry outside of this exporter. Maybe the retry
 		// processor should record these metrics. For the moment we assume no retry.
-		internal.RecordTraceExporterMetrics(ctxWithExporterName, len(td.Spans), len(td.Spans))
+		observability.RecordTraceExporterMetrics(ctxWithExporterName, len(td.Spans), len(td.Spans))
 		return err
 	}
-	internal.RecordTraceExporterMetrics(ctxWithExporterName, len(td.Spans), 0)
+	observability.RecordTraceExporterMetrics(ctxWithExporterName, len(td.Spans), 0)
 	return nil
 }

exporter/zipkinexporter/zipkin.go

@@ -31,7 +31,7 @@ import (
 
 	commonpb "github.com/census-instrumentation/opencensus-proto/gen-go/agent/common/v1"
 	"github.com/census-instrumentation/opencensus-service/data"
-	"github.com/census-instrumentation/opencensus-service/internal"
+	"github.com/census-instrumentation/opencensus-service/observability"
 	"github.com/census-instrumentation/opencensus-service/processor"
 	spandatatranslator "github.com/census-instrumentation/opencensus-service/translator/trace/spandata"
 )
@@ -221,7 +221,7 @@ func (ze *zipkinExporter) ProcessTraceData(ctx context.Context, td data.TraceDat
 	}
 
 	// And finally record metrics on the number of exported spans.
-	internal.RecordTraceExporterMetrics(internal.ContextWithExporterName(ctx, "zipkin"), len(td.Spans), len(td.Spans)-goodSpans)
+	observability.RecordTraceExporterMetrics(observability.ContextWithExporterName(ctx, "zipkin"), len(td.Spans), len(td.Spans)-goodSpans)
 
 	return nil
 }

internal/observability.go observability/observability.gointernal/observability.go renamed to observability/observability.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package internal
+package observability
 
 // This file contains helpers that are useful to add observability
 // with metrics and tracing using OpenCensus to the various pieces
@@ -31,22 +31,26 @@ import (
 )
 
 var (
-	tagKeyReceiver, _      = tag.NewKey("oc_receiver")
 	mReceiverReceivedSpans = stats.Int64("oc.io/receiver/received_spans", "Counts the number of spans received by the receiver", "1")
 	mReceiverDroppedSpans  = stats.Int64("oc.io/receiver/dropped_spans", "Counts the number of spans dropped by the receiver", "1")
 
-	tagKeyExporter, _      = tag.NewKey("oc_exporter")
 	mExporterReceivedSpans = stats.Int64("oc.io/exporter/received_spans", "Counts the number of spans received by the exporter", "1")
 	mExporterDroppedSpans  = stats.Int64("oc.io/exporter/dropped_spans", "Counts the number of spans received by the exporter", "1")
 )
 
+// TagKeyReceiver defines tag key for Receiver.
+var TagKeyReceiver, _ = tag.NewKey("oc_receiver")
+
+// TagKeyExporter defines tag key for Exporter.
+var TagKeyExporter, _ = tag.NewKey("oc_exporter")
+
 // ViewReceiverReceivedSpans defines the view for the receiver received spans metric.
 var ViewReceiverReceivedSpans = &view.View{
 	Name:        mReceiverReceivedSpans.Name(),
 	Description: mReceiverReceivedSpans.Description(),
 	Measure:     mReceiverReceivedSpans,
 	Aggregation: view.Sum(),
-	TagKeys:     []tag.Key{tagKeyReceiver},
+	TagKeys:     []tag.Key{TagKeyReceiver},
 }
 
 // ViewReceiverDroppedSpans defines the view for the receiver dropped spans metric.
@@ -55,7 +59,7 @@ var ViewReceiverDroppedSpans = &view.View{
 	Description: mReceiverDroppedSpans.Description(),
 	Measure:     mReceiverDroppedSpans,
 	Aggregation: view.Sum(),
-	TagKeys:     []tag.Key{tagKeyReceiver},
+	TagKeys:     []tag.Key{TagKeyReceiver},
 }
 
 // ViewExporterReceivedSpans defines the view for the exporter received spans metric.
@@ -64,7 +68,7 @@ var ViewExporterReceivedSpans = &view.View{
 	Description: mExporterReceivedSpans.Description(),
 	Measure:     mExporterReceivedSpans,
 	Aggregation: view.Sum(),
-	TagKeys:     []tag.Key{tagKeyReceiver, tagKeyExporter},
+	TagKeys:     []tag.Key{TagKeyReceiver, TagKeyExporter},
 }
 
 // ViewExporterDroppedSpans defines the view for the exporter dropped spans metric.
@@ -73,7 +77,7 @@ var ViewExporterDroppedSpans = &view.View{
 	Description: mExporterDroppedSpans.Description(),
 	Measure:     mExporterDroppedSpans,
 	Aggregation: view.Sum(),
-	TagKeys:     []tag.Key{tagKeyReceiver, tagKeyExporter},
+	TagKeys:     []tag.Key{TagKeyReceiver, TagKeyExporter},
 }
 
 // AllViews has the views for the metrics provided by the agent.
@@ -88,7 +92,7 @@ var AllViews = []*view.View{
 // and returns the newly created context. For receivers that can receive multiple signals it is
 // recommended to encode the signal as suffix (e.g. "oc_trace" and "oc_metrics").
 func ContextWithReceiverName(ctx context.Context, receiverName string) context.Context {
-	ctx, _ = tag.New(ctx, tag.Upsert(tagKeyReceiver, receiverName))
+	ctx, _ = tag.New(ctx, tag.Upsert(TagKeyReceiver, receiverName))
 	return ctx
 }
 
@@ -102,7 +106,7 @@ func RecordTraceReceiverMetrics(ctxWithTraceReceiverName context.Context, receiv
 // and returns the newly created context. For exporters that can export multiple signals it is
 // recommended to encode the signal as suffix (e.g. "oc_trace" and "oc_metrics").
 func ContextWithExporterName(ctx context.Context, exporterName string) context.Context {
-	ctx, _ = tag.New(ctx, tag.Upsert(tagKeyExporter, exporterName))
+	ctx, _ = tag.New(ctx, tag.Upsert(TagKeyExporter, exporterName))
 	return ctx
 }