Add compression to opencensus receiver/exporter (#347)

Add compression to opencensus receiver/exporter and enable the exporter in the collector

Author: Steven Karis

cmd/ocagent/main.go

@@ -28,6 +28,8 @@ import (
 
 	"go.opencensus.io/stats/view"
 	"go.opencensus.io/zpages"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
 
 	"github.com/census-instrumentation/opencensus-service/data"
 	"github.com/census-instrumentation/opencensus-service/exporter"
@@ -75,7 +77,15 @@ func runOCAgent() {
 		log.Fatalf("Configuration logical error: %v", err)
 	}
 
-	traceExporters, metricsExporters, closeFns, err := config.ExportersFromYAMLConfig(yamlBlob)
+	// TODO: don't hardcode info level logging
+	conf := zap.NewProductionConfig()
+	conf.Level.SetLevel(zapcore.InfoLevel)
+	logger, err := conf.Build()
+	if err != nil {
+		log.Fatalf("Could not instantiate logger: %v", err)
+	}
+
+	traceExporters, metricsExporters, closeFns, err := config.ExportersFromYAMLConfig(logger, yamlBlob)
 	if err != nil {
 		log.Fatalf("Config: failed to create exporters from YAML: %v", err)
 	}

cmd/occollector/app/collector/processors.go

@@ -26,75 +26,41 @@ 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"
-	"github.com/census-instrumentation/opencensus-service/exporter/exporterparser"
 	"github.com/census-instrumentation/opencensus-service/internal/collector/processor"
+	"github.com/census-instrumentation/opencensus-service/internal/config"
 )
 
-func createExporters(v *viper.Viper, logger *zap.Logger) (doneFns []func(), traceExporters []exporter.TraceExporter, metricsExporters []exporter.MetricsExporter) {
+func createExporters(v *viper.Viper, logger *zap.Logger) ([]func(), []exporter.TraceExporter, []exporter.MetricsExporter) {
 	// TODO: (@pjanotti) this is slightly modified from agent but in the end duplication, need to consolidate style and visibility.
-	parseFns := []struct {
-		name string
-		fn   func([]byte) ([]exporter.TraceExporter, []exporter.MetricsExporter, []func() error, error)
-	}{
-		{name: "datadog", fn: exporterparser.DatadogTraceExportersFromYAML},
-		{name: "stackdriver", fn: exporterparser.StackdriverTraceExportersFromYAML},
-		{name: "zipkin", fn: exporterparser.ZipkinExportersFromYAML},
-		{name: "jaeger", fn: exporterparser.JaegerExportersFromYAML},
-		{name: "kafka", fn: exporterparser.KafkaExportersFromYAML},
-	}
 
-	config := builder.GetConfigFile(v)
-	if config == "" {
+	cfg := builder.GetConfigFile(v)
+	if cfg == "" {
 		logger.Info("No config file, exporters can be only configured via the file.")
-		return
+		return nil, nil, nil
 	}
 
-	cfgBlob, err := ioutil.ReadFile(config)
+	cfgBlob, err := ioutil.ReadFile(cfg)
 	if err != nil {
 		logger.Fatal("Cannot read config file for exporters", zap.Error(err))
 	}
 
-	for _, cfg := range parseFns {
-		tes, mes, tesDoneFns, err := cfg.fn(cfgBlob)
-		if err != nil {
-			logger.Fatal("Failed to create config for exporter", zap.String("exporter", cfg.name), zap.Error(err))
-		}
-
-		var anyTraceExporterEnabled, anyMetricsExporterEnabled bool
+	traceExporters, metricsExporters, doneFns, err := config.ExportersFromYAMLConfig(logger, cfgBlob)
+	if err != nil {
+		logger.Fatal("Failed to create config for exporters", zap.Error(err))
+	}
 
-		for _, te := range tes {
-			if te != nil {
-				traceExporters = append(traceExporters, te)
-				anyTraceExporterEnabled = true
+	wrappedDoneFns := make([]func(), 0, len(doneFns))
+	for _, doneFn := range doneFns {
+		wrapperFn := func() {
+			if err := doneFn(); err != nil {
+				logger.Warn("Error when closing exporters", zap.Error(err))
 			}
 		}
-		for _, me := range mes {
-			if me != nil {
-				metricsExporters = append(metricsExporters, me)
-				anyMetricsExporterEnabled = true
-			}
-		}
-		for _, tesDoneFn := range tesDoneFns {
-			if tesDoneFn != nil {
-				wrapperFn := func() {
-					if err := tesDoneFn(); err != nil {
-						logger.Warn("Error when closing exporter", zap.String("exporter", cfg.name), zap.Error(err))
-					}
-				}
-				doneFns = append(doneFns, wrapperFn)
-			}
-		}
-
-		if anyTraceExporterEnabled {
-			logger.Info("Trace Exporter enabled", zap.String("exporter", cfg.name))
-		}
-		if anyMetricsExporterEnabled {
-			logger.Info("Metrices Exporter enabled", zap.String("exporter", cfg.name))
-		}
 
+		wrappedDoneFns = append(wrappedDoneFns, wrapperFn)
 	}
 
-	return doneFns, traceExporters, metricsExporters
+	return wrappedDoneFns, traceExporters, metricsExporters
 }
 
 func buildQueuedSpanProcessor(logger *zap.Logger, opts *builder.QueuedSpanProcessorCfg) (processor.SpanProcessor, error) {

exporter/exporterparser/opencensus.go

@@ -24,10 +24,13 @@ import (
 	"github.com/census-instrumentation/opencensus-service/data"
 	"github.com/census-instrumentation/opencensus-service/exporter"
 	"github.com/census-instrumentation/opencensus-service/internal"
+	"github.com/census-instrumentation/opencensus-service/internal/compression"
+	"github.com/census-instrumentation/opencensus-service/internal/compression/grpc"
 )
 
 type opencensusConfig struct {
-	Endpoint string `yaml:"endpoint,omitempty"`
+	Endpoint    string `yaml:"endpoint,omitempty"`
+	Compression string `yaml:"compression,omitempty"`
 	// TODO: add insecure, service name options.
 }
 
@@ -60,7 +63,16 @@ func OpenCensusTraceExportersFromYAML(config []byte) (tes []exporter.TraceExport
 		return nil, nil, nil, fmt.Errorf("OpenCensus config requires an Endpoint")
 	}
 
-	sde, serr := ocagent.NewExporter(ocagent.WithAddress(ocac.Endpoint), ocagent.WithInsecure())
+	opts := []ocagent.ExporterOption{ocagent.WithAddress(ocac.Endpoint), ocagent.WithInsecure()}
+	if ocac.Compression != "" {
+		if compressionKey := grpc.GetGRPCCompressionKey(ocac.Compression); compressionKey != compression.Unsupported {
+			opts = append(opts, ocagent.UseCompressor(compressionKey))
+		} else {
+			return nil, nil, nil, fmt.Errorf("Unsupported compression type: %s", ocac.Compression)
+		}
+	}
+
+	sde, serr := ocagent.NewExporter(opts...)
 	if serr != nil {
 		return nil, nil, nil, fmt.Errorf("Cannot configure OpenCensus Trace exporter: %v", serr)
 	}

go.mod

@@ -2,7 +2,7 @@ module github.com/census-instrumentation/opencensus-service
 
 require (
 	cloud.google.com/go v0.32.0 // indirect
-	contrib.go.opencensus.io/exporter/ocagent v0.4.2
+	contrib.go.opencensus.io/exporter/ocagent v0.4.3
 	contrib.go.opencensus.io/exporter/stackdriver v0.9.1
 	git.apache.org/thrift.git v0.0.0-20181101003639-92be4f312b88 // indirect
 	github.com/BurntSushi/toml v0.3.1 // indirect

internal/compression/grpc/grpc.go

@@ -0,0 +1,40 @@
+// 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 grpc
+
+import (
+	"strings"
+
+	"google.golang.org/grpc/encoding/gzip"
+
+	"github.com/census-instrumentation/opencensus-service/internal/compression"
+)
+
+var (
+	// Map of opencensus compression types to grpc registered compression types
+	grpcCompressionKeyMap = map[string]string{
+		compression.Gzip: gzip.Name,
+	}
+)
+
+// GetGRPCCompressionKey returns the grpc registered compression key if the
+// passed in compression key is supported, and Unsupported otherwise
+func GetGRPCCompressionKey(compressionType string) string {
+	compressionKey := strings.ToLower(compressionType)
+	if encodingKey, ok := grpcCompressionKeyMap[compressionKey]; ok {
+		return encodingKey
+	}
+	return compression.Unsupported
+}

internal/compression/grpc/grpc_test.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 grpc
+
+import (
+	"testing"
+
+	"github.com/census-instrumentation/opencensus-service/internal/compression"
+)
+
+func TestGetGRPCCompressionKey(t *testing.T) {
+	if GetGRPCCompressionKey("gzip") != compression.Gzip {
+		t.Error("gzip is marked as supported but returned unsupported")
+	}
+
+	if GetGRPCCompressionKey("Gzip") != compression.Gzip {
+		t.Error("Capitalization of Gzip should not matter")
+	}
+
+	if GetGRPCCompressionKey("badType") != compression.Unsupported {
+		t.Error("badType is not supported but was returned as supported")
+	}
+}

internal/compression/grpc/gzip.go

@@ -0,0 +1,17 @@
+// 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 grpc
+
+import _ "google.golang.org/grpc/encoding/gzip" // import the gzip package with auto-registers the gzip grpc compressor

internal/compression/keys.go

@@ -0,0 +1,21 @@
+// 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 compression
+
+// Compression keys for supported compression types within opencensus collector
+const (
+	Unsupported = ""
+	Gzip        = "gzip"
+)

internal/config/config.go

@@ -21,6 +21,7 @@ import (
 	"reflect"
 	"strings"
 
+	"go.uber.org/zap"
 	yaml "gopkg.in/yaml.v2"
 
 	"github.com/census-instrumentation/opencensus-service/exporter"
@@ -299,7 +300,7 @@ func eqLocalHost(host string) bool {
 //  + kafka
 //  + opencensus
 //  + prometheus
-func ExportersFromYAMLConfig(config []byte) (traceExporters []exporter.TraceExporter, metricsExporters []exporter.MetricsExporter, doneFns []func() error, err error) {
+func ExportersFromYAMLConfig(logger *zap.Logger, config []byte) ([]exporter.TraceExporter, []exporter.MetricsExporter, []func() error, error) {
 	parseFns := []struct {
 		name string
 		fn   func([]byte) ([]exporter.TraceExporter, []exporter.MetricsExporter, []func() error, error)
@@ -313,28 +314,35 @@ func ExportersFromYAMLConfig(config []byte) (traceExporters []exporter.TraceExpo
 		{name: "prometheus", fn: exporterparser.PrometheusExportersFromYAML},
 	}
 
+	var traceExporters []exporter.TraceExporter
+	var metricsExporters []exporter.MetricsExporter
+	var doneFns []func() error
 	for _, cfg := range parseFns {
-		tes, mes, tesDoneFns, terr := cfg.fn(config)
+		tes, mes, tesDoneFns, err := cfg.fn(config)
 		if err != nil {
-			err = fmt.Errorf("Failed to create config for %q: %v", cfg.name, terr)
-			return
+			err = fmt.Errorf("Failed to create config for %q: %v", cfg.name, err)
+			return nil, nil, nil, err
 		}
 
 		for _, te := range tes {
 			if te != nil {
 				traceExporters = append(traceExporters, te)
+				logger.Info("Trace Exporter enabled", zap.String("exporter", cfg.name))
 			}
 		}
 
 		for _, me := range mes {
 			if me != nil {
 				metricsExporters = append(metricsExporters, me)
+				logger.Info("Metrices Exporter enabled", zap.String("exporter", cfg.name))
 			}
 		}
 
 		for _, doneFn := range tesDoneFns {
-			doneFns = append(doneFns, doneFn)
+			if doneFn != nil {
+				doneFns = append(doneFns, doneFn)
+			}
 		}
 	}
-	return
+	return traceExporters, metricsExporters, doneFns, nil
 }

receiver/receiver.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/census-instrumentation/opencensus-service/data"
 	"github.com/census-instrumentation/opencensus-service/internal"
+	_ "github.com/census-instrumentation/opencensus-service/internal/compression/grpc" // load in supported grpc compression encodings
 )
 
 // A TraceReceiver is an "arbitrary data"-to-"trace proto span" converter.