Move debug processor to exporters, because no next. (#443)

* Move debug processor to exporters, because no next.

* Remove another hardcoded debug exporter.

* Add tests for ExportFormat.

* Renmae debug to logging.

* Renmae New[Trace|Metrics]DataExporter to New[Trace|Metrics]Exporter.

Author: Bogdan Drutu

cmd/occollector/app/builder/builder.go

@@ -36,7 +36,7 @@ const (
 	ocReceiverFlg               = "receive-oc-trace"
 	zipkinReceiverFlg           = "receive-zipkin"
 	zipkinScribeReceiverFlg     = "receive-zipkin-scribe"
-	debugProcessorFlg           = "debug-processor"
+	loggingExporterFlg          = "logging-exporter"
 	useTailSamplingAlwaysSample = "tail-sampling-always-sample"
 )
 
@@ -51,7 +51,7 @@ func Flags(flags *flag.FlagSet) {
 		fmt.Sprintf("Flag to run the Zipkin receiver, default settings: %+v", *NewDefaultZipkinReceiverCfg()))
 	flags.Bool(zipkinScribeReceiverFlg, false,
 		fmt.Sprintf("Flag to run the Zipkin Scribe receiver, default settings: %+v", *NewDefaultZipkinScribeReceiverCfg()))
-	flags.Bool(debugProcessorFlg, false, "Flag to add a debug processor (combine with log level DEBUG to log incoming spans)")
+	flags.Bool(loggingExporterFlg, false, "Flag to add a logging exporter (combine with log level DEBUG to log incoming spans)")
 	flags.Bool(useTailSamplingAlwaysSample, false, "Flag to use a tail-based sampling processor with an always sample policy, "+
 		"unless tail sampling setting is present on configuration file.")
 }
@@ -61,9 +61,9 @@ func GetConfigFile(v *viper.Viper) string {
 	return v.GetString(configCfg)
 }
 
-// DebugProcessorEnabled returns true if the debug processor is enabled, and false otherwise
-func DebugProcessorEnabled(v *viper.Viper) bool {
-	return v.GetBool(debugProcessorFlg)
+// LoggingExporterEnabled returns true if the debug processor is enabled, and false otherwise
+func LoggingExporterEnabled(v *viper.Viper) bool {
+	return v.GetBool(loggingExporterFlg)
 }
 
 // DebugTailSamplingEnabled returns true if the debug processor is enabled, and false otherwise

cmd/occollector/app/collector/processors.go

@@ -26,6 +26,7 @@ import (
 
 	"github.com/census-instrumentation/opencensus-service/cmd/occollector/app/builder"
 	"github.com/census-instrumentation/opencensus-service/cmd/occollector/app/sender"
+	"github.com/census-instrumentation/opencensus-service/exporter/loggingexporter"
 	"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"
@@ -246,8 +247,9 @@ func startProcessor(v *viper.Viper, logger *zap.Logger) (processor.SpanProcessor
 	// TODO: (@pjanotti) make use of metrics exporters
 	_ = metricsExporters
 
-	if builder.DebugProcessorEnabled(v) {
-		dbgProc := processor.NewNoopSpanProcessor(logger)
+	if builder.LoggingExporterEnabled(v) {
+		dbgProc := processor.NewTraceExporterProcessor(loggingexporter.NewTraceExporter(logger))
+		// 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/loggingexporter/logging_exporter.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 loggingexporter
+
+import (
+	"context"
+
+	"github.com/census-instrumentation/opencensus-service/data"
+	"github.com/census-instrumentation/opencensus-service/exporter"
+	"go.uber.org/zap"
+)
+
+const exportFormat = "LoggingExporter"
+
+// A logging exporter that does not sends the data to any destination but logs debugging messages.
+type loggingExporter struct{ logger *zap.Logger }
+
+var _ exporter.TraceDataExporter = (*loggingExporter)(nil)
+var _ exporter.MetricsDataExporter = (*loggingExporter)(nil)
+
+func (sp *loggingExporter) ProcessTraceData(ctx context.Context, td data.TraceData) error {
+	// TODO: Record metrics
+	// TODO: Add ability to record the received data
+	sp.logger.Debug("loggingTraceDataExporter", zap.Int("#spans", len(td.Spans)))
+	return nil
+}
+
+func (sp *loggingExporter) ProcessMetricsData(ctx context.Context, md data.MetricsData) error {
+	sp.logger.Debug("loggingMetricsDataExporter", zap.Int("#metrics", len(md.Metrics)))
+	// TODO: Record metrics
+	// TODO: Add ability to record the received data
+	return nil
+}
+
+func (sp *loggingExporter) ExportFormat() string {
+	return exportFormat
+}
+
+// NewTraceExporter creates an exporter.TraceDataExporter that just drops the
+// received data and logs debugging messages.
+func NewTraceExporter(logger *zap.Logger) exporter.TraceDataExporter {
+	return &loggingExporter{logger: logger}
+}
+
+// NewMetricsExporter creates an exporter.MetricsDataExporter that just drops the
+// received data and logs debugging messages.
+func NewMetricsExporter(logger *zap.Logger) exporter.MetricsDataExporter {
+	return &loggingExporter{logger: logger}
+}

processor/debug_processor_test.go …loggingexporter/logging_exporter_test.goprocessor/debug_processor_test.go renamed to exporter/loggingexporter/logging_exporter_test.go processor/debug_processor_test.go renamed to exporter/loggingexporter/logging_exporter_test.go

@@ -11,7 +11,7 @@
 // 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 processor
+package loggingexporter
 
 import (
 	"context"
@@ -20,27 +20,33 @@ 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"
-	"go.uber.org/zap/zaptest"
+	"go.uber.org/zap"
 )
 
-func TestDebugTraceDataProcessorNoErrors(t *testing.T) {
-	dtdp := NewDebugTraceDataProcessor(zaptest.NewLogger(t))
+func TestLoggingTraceExporterNoErrors(t *testing.T) {
+	dtdp := NewTraceExporter(zap.NewNop())
 	td := data.TraceData{
 		Spans: make([]*tracepb.Span, 7),
 	}
 	if err := dtdp.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())
+	}
 }
 
-func TestDebugMetricsDataProcessorNoErrors(t *testing.T) {
-	dmdp := NewDebugMetricsDataProcessor(zaptest.NewLogger(t))
+func TestLoggingMetricsExporterNoErrors(t *testing.T) {
+	dmdp := NewMetricsExporter(zap.NewNop())
 	md := data.MetricsData{
 		Metrics: make([]*metricspb.Metric, 7),
 	}
 	if err := dmdp.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())
+	}
 }

internal/collector/processor/processor.go

@@ -15,13 +15,10 @@
 package processor
 
 import (
-	"context"
-
 	commonpb "github.com/census-instrumentation/opencensus-proto/gen-go/agent/common/v1"
 	"go.opencensus.io/stats"
 	"go.opencensus.io/stats/view"
 	"go.opencensus.io/tag"
-	"go.uber.org/zap"
 
 	"github.com/census-instrumentation/opencensus-service/data"
 	"github.com/census-instrumentation/opencensus-service/internal/collector/telemetry"
@@ -34,29 +31,6 @@ type SpanProcessor interface {
 	// TODO: (@pjanotti) For shutdown improvement, the interface needs a method to attempt that.
 }
 
-// An initial processor that does not sends the data to any destination but helps debugging.
-type debugSpanProcessor struct{ logger *zap.Logger }
-
-var _ SpanProcessor = (*debugSpanProcessor)(nil)
-
-func (sp *debugSpanProcessor) ProcessSpans(td data.TraceData, spanFormat string) error {
-	if td.Node == nil {
-		sp.logger.Warn("Received batch with nil Node", zap.String("format", spanFormat))
-	}
-
-	statsTags := StatsTagsForBatch("debug", ServiceNameForNode(td.Node), spanFormat)
-	numSpans := len(td.Spans)
-	stats.RecordWithTags(context.Background(), statsTags, StatReceivedSpanCount.M(int64(numSpans)))
-
-	sp.logger.Debug("debugSpanProcessor", zap.String("originalFormat", spanFormat), zap.Int("#spans", numSpans))
-	return nil
-}
-
-// NewNoopSpanProcessor creates an OC SpanProcessor that just drops the received data.
-func NewNoopSpanProcessor(logger *zap.Logger) SpanProcessor {
-	return &debugSpanProcessor{logger: logger}
-}
-
 // Keys and stats for telemetry.
 var (
 	TagSourceFormatKey, _ = tag.NewKey("format")

processor/debug_processor.go

@@ -16,17 +16,17 @@ package receiver_test
 
 import (
 	"context"
-	"encoding/json"
 	"log"
 	"time"
 
+	"github.com/census-instrumentation/opencensus-service/exporter/loggingexporter"
+
 	"contrib.go.opencensus.io/exporter/ocagent"
 	"go.opencensus.io/trace"
 
-	"github.com/census-instrumentation/opencensus-service/data"
-	"github.com/census-instrumentation/opencensus-service/processor"
 	"github.com/census-instrumentation/opencensus-service/receiver"
 	"github.com/census-instrumentation/opencensus-service/receiver/opencensusreceiver"
+	"go.uber.org/zap"
 )
 
 func Example_endToEnd() {
@@ -43,7 +43,7 @@ func Example_endToEnd() {
 
 	// Once we have the span receiver which will connect to the
 	// various exporter pipeline i.e. *tracepb.Span->OpenCensus.SpanData
-	lsr := new(logSpanSink)
+	lsr := loggingexporter.NewTraceExporter(zap.NewNop())
 	for _, tr := range trl {
 		if err := tr.StartTraceReception(context.Background(), lsr); err != nil {
 			log.Fatalf("Failed to start trace receiver: %v", err)
@@ -90,14 +90,3 @@ func Example_endToEnd() {
 	oce.Flush()
 	<-time.After(5 * time.Second)
 }
-
-type logSpanSink int
-
-var _ processor.TraceDataProcessor = (*logSpanSink)(nil)
-
-func (lsr *logSpanSink) ProcessTraceData(ctx context.Context, td data.TraceData) error {
-	spansBlob, _ := json.MarshalIndent(td.Spans, " ", "  ")
-	log.Printf("\n****\nNode: %#v\nSpans: %s\n****\n", td.Node, spansBlob)
-
-	return nil
-}