Add "compression" option to otlphttp exporter

exporter/otlphttpexporter/README.md

@@ -33,6 +33,8 @@ The following settings can be optionally configured:
 - `key_file` path to the TLS key to use for TLS required connections. Should
   only be used if `insecure` is set to false.
 
+- `compression` (default = none): Compression type to use (only gzip is supported today)
+
 - `timeout` (default = 30s): HTTP request time limit. For details see https://golang.org/pkg/net/http/#Client
 - `read_buffer_size` (default = 0): ReadBufferSize for HTTP client.
 - `write_buffer_size` (default = 512 * 1024): WriteBufferSize for HTTP client.

exporter/otlphttpexporter/config.go

@@ -35,4 +35,8 @@ type Config struct {
 
 	// The URL to send logs to. If omitted the Endpoint + "/v1/logs" will be used.
 	LogsEndpoint string `mapstructure:"logs_endpoint"`
+
+	// The compression key for supported compression types within
+	// collector. Currently the only supported mode is `gzip`.
+	Compression string `mapstructure:"compression"`
 }

exporter/otlphttpexporter/config_test.go

@@ -81,5 +81,6 @@ func TestLoadConfig(t *testing.T) {
 				WriteBufferSize: 345,
 				Timeout:         time.Second * 10,
 			},
+			Compression: "gzip",
 		})
 }

exporter/otlphttpexporter/otlp.go

@@ -24,16 +24,19 @@ import (
 	"net/http"
 	"net/url"
 	"strconv"
+	"strings"
 	"time"
 
 	"go.uber.org/zap"
 	"google.golang.org/genproto/googleapis/rpc/status"
 	"google.golang.org/protobuf/proto"
 
+	"go.opentelemetry.io/collector/config/configgrpc"
 	"go.opentelemetry.io/collector/config/configmodels"
 	"go.opentelemetry.io/collector/consumer/consumererror"
 	"go.opentelemetry.io/collector/consumer/pdata"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
+	"go.opentelemetry.io/collector/internal/middleware"
 )
 
 type exporterImp struct {
@@ -67,6 +70,14 @@ func newExporter(cfg configmodels.Exporter, logger *zap.Logger) (*exporterImp, e
 		return nil, err
 	}
 
+	if oCfg.Compression != "" {
+		if strings.ToLower(oCfg.Compression) == configgrpc.CompressionGzip {
+			client.Transport = middleware.NewCompressRoundTripper(client.Transport)
+		} else {
+			return nil, fmt.Errorf("unsupported compression type %q", oCfg.Compression)
+		}
+	}
+
 	return &exporterImp{
 		config: oCfg,
 		client: client,

exporter/otlphttpexporter/otlp_test.go

@@ -132,6 +132,61 @@ func TestTraceRoundTrip(t *testing.T) {
 	}
 }
 
+func TestCompressionOptions(t *testing.T) {
+	addr := testutil.GetAvailableLocalAddress(t)
+
+	tests := []struct {
+		name        string
+		baseURL     string
+		compression string
+		err         bool
+	}{
+		{
+			name:        "no compression",
+			baseURL:     fmt.Sprintf("http://%s", addr),
+			compression: "",
+		},
+		{
+			name:        "gzip",
+			baseURL:     fmt.Sprintf("http://%s", addr),
+			compression: "gzip",
+		},
+		{
+			name:        "incorrect compression",
+			baseURL:     fmt.Sprintf("http://%s", addr),
+			compression: "gzip2",
+			err:         true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			sink := new(consumertest.TracesSink)
+			startTraceReceiver(t, addr, sink)
+
+			factory := NewFactory()
+			cfg := createExporterConfig(test.baseURL, factory.CreateDefaultConfig())
+			cfg.Compression = test.compression
+			exp, err := factory.CreateTracesExporter(context.Background(), component.ExporterCreateParams{Logger: zap.NewNop()}, cfg)
+			if test.err {
+				assert.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+			startAndCleanup(t, exp)
+
+			td := testdata.GenerateTraceDataOneSpan()
+			assert.NoError(t, exp.ConsumeTraces(context.Background(), td))
+			require.Eventually(t, func() bool {
+				return sink.SpansCount() > 0
+			}, 1*time.Second, 10*time.Millisecond)
+			allTraces := sink.AllTraces()
+			require.Len(t, allTraces, 1)
+			assert.EqualValues(t, td, allTraces[0])
+		})
+	}
+}
+
 func TestMetricsError(t *testing.T) {
 	addr := testutil.GetAvailableLocalAddress(t)
 

exporter/otlphttpexporter/testdata/config.yaml

@@ -28,6 +28,7 @@ exporters:
       "can you have a . here?": "F0000000-0000-0000-0000-000000000000"
       header1: 234
       another: "somevalue"
+    compression: gzip
 
 service:
   pipelines:

internal/middleware/compression.go

@@ -15,12 +15,69 @@
 package middleware
 
 import (
+	"bytes"
 	"compress/gzip"
 	"compress/zlib"
 	"io"
 	"net/http"
 )
 
+const (
+	headerContentEncoding = "Content-Encoding"
+	headerValueGZIP       = "gzip"
+)
+
+type CompressRoundTripper struct {
+	http.RoundTripper
+}
+
+func NewCompressRoundTripper(rt http.RoundTripper) *CompressRoundTripper {
+	return &CompressRoundTripper{
+		RoundTripper: rt,
+	}
+}
+
+func (r *CompressRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
+
+	if req.Header.Get(headerContentEncoding) != "" {
+		// If the header already specifies a content encoding then skip compression
+		// since we don't want to compress it again. This is a safeguard that normally
+		// should not happen since CompressRoundTripper is not intended to be used
+		// with http clients which already do their own compression.
+		return r.RoundTripper.RoundTrip(req)
+	}
+
+	// Gzip the body.
+	buf := bytes.NewBuffer([]byte{})
+	gzipWriter := gzip.NewWriter(buf)
+	_, copyErr := io.Copy(gzipWriter, req.Body)
+	closeErr := req.Body.Close()
+
+	if err := gzipWriter.Close(); err != nil {
+		return nil, err
+	}
+
+	if copyErr != nil {
+		return nil, copyErr
+	}
+	if closeErr != nil {
+		return nil, closeErr
+	}
+
+	// Create a new request since the docs say that we cannot modify the "req"
+	// (see https://golang.org/pkg/net/http/#RoundTripper).
+	cReq, err := http.NewRequestWithContext(req.Context(), req.Method, req.URL.String(), buf)
+	if err != nil {
+		return nil, err
+	}
+
+	// Clone the headers and add gzip encoding header.
+	cReq.Header = req.Header.Clone()
+	cReq.Header.Add(headerContentEncoding, headerValueGZIP)
+
+	return r.RoundTripper.RoundTrip(cReq)
+}
+
 type ErrorHandler func(w http.ResponseWriter, r *http.Request, errorMsg string, statusCode int)
 
 type decompressor struct {

internal/middleware/compression_test.go

@@ -31,6 +31,67 @@ import (
 	"go.opentelemetry.io/collector/testutil"
 )
 
+func TestHTTPClientCompression(t *testing.T) {
+	testBody := []byte("uncompressed_text")
+	compressedBody, _ := compressGzip(testBody)
+
+	tests := []struct {
+		name     string
+		encoding string
+		reqBody  []byte
+	}{
+		{
+			name:     "NoCompression",
+			encoding: "",
+			reqBody:  testBody,
+		},
+		{
+			name:     "ValidGzip",
+			encoding: "gzip",
+			reqBody:  compressedBody.Bytes(),
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				body, err := ioutil.ReadAll(r.Body)
+				require.NoError(t, err, "failed to read request body: %v", err)
+				assert.EqualValues(t, tt.reqBody, body)
+				w.WriteHeader(200)
+			})
+
+			addr := testutil.GetAvailableLocalAddress(t)
+			ln, err := net.Listen("tcp", addr)
+			require.NoError(t, err, "failed to create listener: %v", err)
+			srv := &http.Server{
+				Handler: handler,
+			}
+			go func() {
+				_ = srv.Serve(ln)
+			}()
+			// Wait for the servers to start
+			<-time.After(10 * time.Millisecond)
+
+			serverURL := fmt.Sprintf("http://%s", ln.Addr().String())
+			reqBody := bytes.NewBuffer(testBody)
+
+			req, err := http.NewRequest("GET", serverURL, reqBody)
+			require.NoError(t, err, "failed to create request to test handler")
+
+			client := http.Client{}
+			if tt.encoding == "gzip" {
+				client.Transport = NewCompressRoundTripper(http.DefaultTransport)
+			}
+			res, err := client.Do(req)
+			require.NoError(t, err)
+
+			ioutil.ReadAll(res.Body)
+			require.NoError(t, res.Body.Close(), "failed to close request body: %v", err)
+			require.NoError(t, srv.Close())
+		})
+	}
+}
+
 func TestHTTPContentDecompressionHandler(t *testing.T) {
 	testBody := []byte("uncompressed_text")
 	tests := []struct {

testbed/testbed/receivers.go

@@ -186,17 +186,18 @@ func (jr *JaegerDataReceiver) ProtocolName() string {
 	return "jaeger"
 }
 
-// baseOTLPDataReceiver implements the OTLP format receiver.
-type baseOTLPDataReceiver struct {
+// BaseOTLPDataReceiver implements the OTLP format receiver.
+type BaseOTLPDataReceiver struct {
 	DataReceiverBase
 	// One of the "otlp" for OTLP over gRPC or "otlphttp" for OTLP over HTTP.
 	exporterType    string
 	traceReceiver   component.TracesReceiver
 	metricsReceiver component.MetricsReceiver
 	logReceiver     component.LogsReceiver
+	compression     string
 }
 
-func (bor *baseOTLPDataReceiver) Start(tc consumer.TracesConsumer, mc consumer.MetricsConsumer, lc consumer.LogsConsumer) error {
+func (bor *BaseOTLPDataReceiver) Start(tc consumer.TracesConsumer, mc consumer.MetricsConsumer, lc consumer.LogsConsumer) error {
 	factory := otlpreceiver.NewFactory()
 	cfg := factory.CreateDefaultConfig().(*otlpreceiver.Config)
 	cfg.SetName(bor.exporterType)
@@ -228,7 +229,12 @@ func (bor *baseOTLPDataReceiver) Start(tc consumer.TracesConsumer, mc consumer.M
 	return bor.logReceiver.Start(context.Background(), bor)
 }
 
-func (bor *baseOTLPDataReceiver) Stop() error {
+func (bor *BaseOTLPDataReceiver) WithCompression(compression string) *BaseOTLPDataReceiver {
+	bor.compression = compression
+	return bor
+}
+
+func (bor *BaseOTLPDataReceiver) Stop() error {
 	if err := bor.traceReceiver.Shutdown(context.Background()); err != nil {
 		return err
 	}
@@ -238,37 +244,44 @@ func (bor *baseOTLPDataReceiver) Stop() error {
 	return bor.logReceiver.Shutdown(context.Background())
 }
 
-func (bor *baseOTLPDataReceiver) ProtocolName() string {
+func (bor *BaseOTLPDataReceiver) ProtocolName() string {
 	return bor.exporterType
 }
 
-func (bor *baseOTLPDataReceiver) GenConfigYAMLStr() string {
+func (bor *BaseOTLPDataReceiver) GenConfigYAMLStr() string {
 	addr := fmt.Sprintf("localhost:%d", bor.Port)
 	if bor.exporterType == "otlphttp" {
 		addr = "http://" + addr
 	}
 	// Note that this generates an exporter config for agent.
-	return fmt.Sprintf(`
+	str := fmt.Sprintf(`
   %s:
     endpoint: "%s"
     insecure: true`, bor.exporterType, addr)
+
+	if bor.compression != "" {
+		str += fmt.Sprintf(`
+    compression: "%s"`, bor.compression)
+	}
+
+	return str
 }
 
 const DefaultOTLPPort = 55680
 
 // NewOTLPDataReceiver creates a new OTLP DataReceiver that will listen on the specified port after Start
 // is called.
-func NewOTLPDataReceiver(port int) DataReceiver {
-	return &baseOTLPDataReceiver{
+func NewOTLPDataReceiver(port int) *BaseOTLPDataReceiver {
+	return &BaseOTLPDataReceiver{
 		DataReceiverBase: DataReceiverBase{Port: port},
 		exporterType:     "otlp",
 	}
 }
 
 // NewOTLPDataReceiver creates a new OTLP/HTTP DataReceiver that will listen on the specified port after Start
 // is called.
-func NewOTLPHTTPDataReceiver(port int) DataReceiver {
-	return &baseOTLPDataReceiver{
+func NewOTLPHTTPDataReceiver(port int) *BaseOTLPDataReceiver {
+	return &BaseOTLPDataReceiver{
 		DataReceiverBase: DataReceiverBase{Port: port},
 		exporterType:     "otlphttp",
 	}

testbed/tests/trace_test.go

@@ -66,14 +66,23 @@ func TestTrace10kSPS(t *testing.T) {
 			},
 		},
 		{
-			"OTLP",
+			"OTLP-gRPC",
 			testbed.NewOTLPTraceDataSender(testbed.DefaultHost, testbed.GetAvailablePort(t)),
 			testbed.NewOTLPDataReceiver(testbed.GetAvailablePort(t)),
 			testbed.ResourceSpec{
 				ExpectedMaxCPU: 20,
 				ExpectedMaxRAM: 70,
 			},
 		},
+		{
+			"OTLP-gRPC-gzip",
+			testbed.NewOTLPTraceDataSender(testbed.DefaultHost, testbed.GetAvailablePort(t)),
+			testbed.NewOTLPDataReceiver(testbed.GetAvailablePort(t)).WithCompression("gzip"),
+			testbed.ResourceSpec{
+				ExpectedMaxCPU: 30,
+				ExpectedMaxRAM: 100,
+			},
+		},
 		{
 			"OTLP-HTTP",
 			testbed.NewOTLPHTTPTraceDataSender(testbed.DefaultHost, testbed.GetAvailablePort(t)),
@@ -83,6 +92,15 @@ func TestTrace10kSPS(t *testing.T) {
 				ExpectedMaxRAM: 100,
 			},
 		},
+		{
+			"OTLP-HTTP-gzip",
+			testbed.NewOTLPHTTPTraceDataSender(testbed.DefaultHost, testbed.GetAvailablePort(t)),
+			testbed.NewOTLPHTTPDataReceiver(testbed.GetAvailablePort(t)).WithCompression("gzip"),
+			testbed.ResourceSpec{
+				ExpectedMaxCPU: 25,
+				ExpectedMaxRAM: 100,
+			},
+		},
 		{
 			"Zipkin",
 			testbed.NewZipkinDataSender(testbed.DefaultHost, testbed.GetAvailablePort(t)),