Add gzip compression and "disable_compression" to HTTP client exporters

All exporters that use HTTPClientSettings will now have a config option
"compression". The default value is empty and results in no compression,
which is the same as the old behavior. You can also specify "compression: gzip"
which will result in gzip encoding of outgoing http requests.

config/configgrpc/README.md

@@ -16,7 +16,7 @@ configuration. For more information, see [configtls
 README](../configtls/README.md).
 
 - [`balancer_name`](https://github.com/grpc/grpc-go/blob/master/examples/features/load_balancing/README.md)
-- `compression` (default = gzip): Compression type to use (only gzip is supported today)
+- `compression` (default = none): Compression type to use (only gzip is supported today)
 - `endpoint`: Valid value syntax available [here](https://github.com/grpc/grpc/blob/master/doc/naming.md)
 - `headers`: name/value pairs added to the request
 - [`keepalive`](https://godoc.org/google.golang.org/grpc/keepalive#ClientParameters)

config/confighttp/README.md

@@ -15,6 +15,7 @@ configuration. For more information, see [configtls
 README](../configtls/README.md).
 
 - `endpoint`: address:port
+- `compression` (default = none): Compression type to use (only gzip is supported today)
 - `headers`: name/value pairs added to the HTTP request headers
 - [`read_buffer_size`](https://golang.org/pkg/net/http/#Transport)
 - [`timeout`](https://golang.org/pkg/net/http/#Client)

config/confighttp/confighttp.go

@@ -16,12 +16,15 @@ package confighttp
 
 import (
  "crypto/tls"
+ "fmt"
  "net"
  "net/http"
+ "strings"
  "time"
 
  "github.com/rs/cors"
 
+ "go.opentelemetry.io/collector/config/configgrpc"
  "go.opentelemetry.io/collector/config/configtls"
  "go.opentelemetry.io/collector/internal/middleware"
 )
@@ -46,6 +49,10 @@ type HTTPClientSettings struct {
  // Existing header values are overwritten if collision happens.
  Headers map[string]string `mapstructure:"headers,omitempty"`
 
+ // The compression key for supported compression types within
+ // collector. Currently the only supported mode is `gzip`.
+ Compression string `mapstructure:"compression"`
+
  // Custom Round Tripper to allow for individual components to intercept HTTP requests
  CustomRoundTripper func(next http.RoundTripper) (http.RoundTripper, error)
 }
@@ -74,6 +81,14 @@ func (hcs *HTTPClientSettings) ToClient() (*http.Client, error) {
  }
  }
 
+ if hcs.Compression != "" {
+ if strings.ToLower(hcs.Compression) == configgrpc.CompressionGzip {
+ clientTransport = middleware.NewCompressRoundTripper(clientTransport)
+ } else {
+ return nil, fmt.Errorf("unsupported compression type %q", hcs.Compression)
+ }
+ }
+
  if hcs.CustomRoundTripper != nil {
  clientTransport, err = hcs.CustomRoundTripper(clientTransport)
  if err != nil {

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/zipkinexporter/README.md

@@ -26,6 +26,7 @@ As a result, the following parameters are also required:
 
 The following settings are optional:
 
+- `compression` (default = none): Compression type to use (only gzip is supported today)
 - `defaultservicename` (default = `<missing service name>`): What to name
  services missing this information.
 

internal/middleware/compression.go

@@ -15,12 +15,59 @@
 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) {
+
+ gzipWriter := gzip.NewWriter(nil)
+
+ // Gzip the body.
+ buf := bytes.NewBuffer([]byte{})
+ gzipWriter.Reset(buf)
+ _, copyErr := io.Copy(gzipWriter, req.Body)
+ req.Body.Close()
+
+ if err := gzipWriter.Close(); err != nil {
+ return nil, err
+ }
+
+ if copyErr != nil {
+ return nil, copyErr
+ }
+
+ // Create a new request since the docs say that we cannot modify the "req".
+ 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,68 @@ 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")
+ req.Header.Set("Content-Encoding", tt.encoding)
+
+ 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 {