confighttp: Including lz4 as supported compression encoding (#11108)

#### Description

Adding in support for lz4 compression into the project, I have seen it
perform amazing well in other projects that send a lot of traffic and
were originally using `gzip` or `flate`.

It would be really great to have for environments that are cpu, memory
sensitive.

#### Link to tracking issue
Fixes 9128

#### Testing

I've updated the existing test frameworks to include lz4 in their suite

#### Documentation

I've updated the readmes to include lz4 but I suspect there might be
more places to update.

.chloggen/msg_feature-include-lz4-compression.yaml

@@ -0,0 +1,25 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: "enhancement"
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: "confighttp"
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Adding support for lz4 compression into the project
+
+# One or more tracking issues or pull requests related to the change
+issues: [9128]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

cmd/otelcorecol/go.mod

@@ -64,6 +64,7 @@ require (
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/mostynb/go-grpc-compression v1.2.3 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
+	github.com/pierrec/lz4/v4 v4.1.21 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
 	github.com/prometheus/client_golang v1.20.4 // indirect

config/configcompression/compressiontype.go

@@ -14,6 +14,7 @@ const (
 	TypeDeflate Type = "deflate"
 	TypeSnappy  Type = "snappy"
 	TypeZstd    Type = "zstd"
+	TypeLz4     Type = "lz4"
 	typeNone    Type = "none"
 	typeEmpty   Type = ""
 )
@@ -31,6 +32,7 @@ func (ct *Type) UnmarshalText(in []byte) error {
 		typ == TypeDeflate ||
 		typ == TypeSnappy ||
 		typ == TypeZstd ||
+		typ == TypeLz4 ||
 		typ == typeNone ||
 		typ == typeEmpty {
 		*ct = typ

config/configcompression/compressiontype_test.go

@@ -57,6 +57,12 @@ func TestUnmarshalText(t *testing.T) {
 			compressionName: []byte("none"),
 			shouldError:     false,
 		},
+		{
+			name:            "ValidLz4",
+			compressionName: []byte("lz4"),
+			isCompressed:    true,
+			shouldError:     false,
+		},
 		{
 			name:            "Invalid",
 			compressionName: []byte("ggip"),

config/confighttp/README.md

@@ -23,7 +23,7 @@ README](../configtls/README.md).
 - [`read_buffer_size`](https://golang.org/pkg/net/http/#Transport)
 - [`timeout`](https://golang.org/pkg/net/http/#Client)
 - [`write_buffer_size`](https://golang.org/pkg/net/http/#Transport)
-- `compression`: Compression type to use among `gzip`, `zstd`, `snappy`, `zlib`, and `deflate`.
+- `compression`: Compression type to use among `gzip`, `zstd`, `snappy`, `zlib`, `deflate`, and `lz4`.
   - look at the documentation for the server-side of the communication.
   - `none` will be treated as uncompressed, and any other inputs will cause an error.
 - [`max_idle_conns`](https://golang.org/pkg/net/http/#Transport)
@@ -79,7 +79,7 @@ will not be enabled.
   not set, browsers use a default of 5 seconds.
 - `endpoint`: Valid value syntax available [here](https://github.com/grpc/grpc/blob/master/doc/naming.md)
 - `max_request_body_size`: configures the maximum allowed body size in bytes for a single request. Default: `20971520` (20MiB)
-- `compression_algorithms`: configures the list of compression algorithms the server can accept. Default: ["", "gzip", "zstd", "zlib", "snappy", "deflate"]
+- `compression_algorithms`: configures the list of compression algorithms the server can accept. Default: ["", "gzip", "zstd", "zlib", "snappy", "deflate", "lz4"]
 - [`tls`](../configtls/README.md)
 - [`auth`](../configauth/README.md)
   - `request_params`: a list of query parameter names to add to the auth context, along with the HTTP headers

config/confighttp/compression.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/golang/snappy"
 	"github.com/klauspost/compress/zstd"
+	"github.com/pierrec/lz4/v4"
 
 	"go.opentelemetry.io/collector/config/configcompression"
 )
@@ -66,6 +67,13 @@ var availableDecoders = map[string]func(body io.ReadCloser) (io.ReadCloser, erro
 			orig:   body,
 		}, nil
 	},
+	//nolint:unparam // Ignoring the linter request to remove error return since it needs to match the method signature
+	"lz4": func(body io.ReadCloser) (io.ReadCloser, error) {
+		return &compressReadCloser{
+			Reader: lz4.NewReader(body),
+			orig:   body,
+		}, nil
+	},
 }
 
 func newCompressRoundTripper(rt http.RoundTripper, compressionType configcompression.Type) (*compressRoundTripper, error) {
@@ -119,7 +127,6 @@ type decompressor struct {
 // httpContentDecompressor offloads the task of handling compressed HTTP requests
 // by identifying the compression format in the "Content-Encoding" header and re-writing
 // request body so that the handlers further in the chain can work on decompressed data.
-// It supports gzip and deflate/zlib compression.
 func httpContentDecompressor(h http.Handler, maxRequestBodySize int64, eh func(w http.ResponseWriter, r *http.Request, errorMsg string, statusCode int), enableDecoders []string, decoders map[string]func(body io.ReadCloser) (io.ReadCloser, error)) http.Handler {
 	errHandler := defaultErrorHandler
 	if eh != nil {

config/confighttp/compression_test.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/golang/snappy"
 	"github.com/klauspost/compress/zstd"
+	"github.com/pierrec/lz4/v4"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -33,6 +34,7 @@ func TestHTTPClientCompression(t *testing.T) {
 	compressedDeflateBody := compressZlib(t, testBody)
 	compressedSnappyBody := compressSnappy(t, testBody)
 	compressedZstdBody := compressZstd(t, testBody)
+	compressedLz4Body := compressLz4(t, testBody)
 
 	tests := []struct {
 		name        string
@@ -82,6 +84,12 @@ func TestHTTPClientCompression(t *testing.T) {
 			reqBody:     compressedZstdBody.Bytes(),
 			shouldError: false,
 		},
+		{
+			name:        "ValidLz4",
+			encoding:    configcompression.TypeLz4,
+			reqBody:     compressedLz4Body.Bytes(),
+			shouldError: false,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -199,6 +207,12 @@ func TestHTTPContentDecompressionHandler(t *testing.T) {
 			reqBody:  compressSnappy(t, testBody),
 			respCode: http.StatusOK,
 		},
+		{
+			name:     "ValidLz4",
+			encoding: "lz4",
+			reqBody:  compressLz4(t, testBody),
+			respCode: http.StatusOK,
+		},
 		{
 			name:     "InvalidDeflate",
 			encoding: "deflate",
@@ -365,6 +379,75 @@ func TestOverrideCompressionList(t *testing.T) {
 	require.NoError(t, res.Body.Close(), "failed to close request body: %v", err)
 }
 
+func TestDecompressorAvoidDecompressionBomb(t *testing.T) {
+	t.Parallel()
+
+	for _, tc := range []struct {
+		name     string
+		encoding string
+		compress func(tb testing.TB, payload []byte) *bytes.Buffer
+	}{
+		// None encoding is ignored since it does not
+		// enforce the max body size if content encoding header is not set
+		{
+			name:     "gzip",
+			encoding: "gzip",
+			compress: compressGzip,
+		},
+		{
+			name:     "zstd",
+			encoding: "zstd",
+			compress: compressZstd,
+		},
+		{
+			name:     "zlib",
+			encoding: "zlib",
+			compress: compressZlib,
+		},
+		{
+			name:     "snappy",
+			encoding: "snappy",
+			compress: compressSnappy,
+		},
+		{
+			name:     "lz4",
+			encoding: "lz4",
+			compress: compressLz4,
+		},
+	} {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			h := httpContentDecompressor(
+				http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+					n, err := io.Copy(io.Discard, r.Body)
+					assert.Equal(t, int64(1024), n, "Must have only read the limited value of bytes")
+					assert.EqualError(t, err, "http: request body too large")
+					w.WriteHeader(http.StatusBadRequest)
+				}),
+				1024,
+				defaultErrorHandler,
+				defaultCompressionAlgorithms,
+				availableDecoders,
+			)
+
+			payload := tc.compress(t, make([]byte, 2*1024)) // 2KB uncompressed payload
+			assert.NotEmpty(t, payload.Bytes(), "Must have data available")
+
+			req := httptest.NewRequest(http.MethodPost, "/", payload)
+			req.Header.Set("Content-Encoding", tc.encoding)
+
+			resp := httptest.NewRecorder()
+
+			h.ServeHTTP(resp, req)
+
+			assert.Equal(t, http.StatusBadRequest, resp.Code, "Must match the expected code")
+			assert.Empty(t, resp.Body.String(), "Must match the returned string")
+		})
+	}
+}
+
 func compressGzip(t testing.TB, body []byte) *bytes.Buffer {
 	var buf bytes.Buffer
 	gw := gzip.NewWriter(&buf)
@@ -400,3 +483,12 @@ func compressZstd(t testing.TB, body []byte) *bytes.Buffer {
 	require.NoError(t, zw.Close())
 	return &buf
 }
+
+func compressLz4(tb testing.TB, body []byte) *bytes.Buffer {
+	var buf bytes.Buffer
+	lz := lz4.NewWriter(&buf)
+	_, err := lz.Write(body)
+	require.NoError(tb, err)
+	require.NoError(tb, lz.Close())
+	return &buf
+}

config/confighttp/compressor.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/golang/snappy"
 	"github.com/klauspost/compress/zstd"
+	"github.com/pierrec/lz4/v4"
 
 	"go.opentelemetry.io/collector/config/configcompression"
 )
@@ -32,6 +33,13 @@ var (
 	zStdPool                  = &compressor{pool: sync.Pool{New: func() any { zw, _ := zstd.NewWriter(nil, zstd.WithEncoderConcurrency(1)); return zw }}}
 	_        writeCloserReset = (*zlib.Writer)(nil)
 	zLibPool                  = &compressor{pool: sync.Pool{New: func() any { return zlib.NewWriter(nil) }}}
+	_        writeCloserReset = (*lz4.Writer)(nil)
+	lz4Pool                   = &compressor{pool: sync.Pool{New: func() any {
+		lz := lz4.NewWriter(nil)
+		// Setting concurrency to 1 to disable async decoding by goroutines. This will reduce the overall memory footprint and pool
+		_ = lz.Apply(lz4.ConcurrencyOption(1))
+		return lz
+	}}}
 )
 
 type compressor struct {
@@ -50,6 +58,8 @@ func newCompressor(compressionType configcompression.Type) (*compressor, error)
 		return zStdPool, nil
 	case configcompression.TypeZlib, configcompression.TypeDeflate:
 		return zLibPool, nil
+	case configcompression.TypeLz4:
+		return lz4Pool, nil
 	}
 	return nil, errors.New("unsupported compression type, ")
 }

config/confighttp/confighttp.go

@@ -33,7 +33,7 @@ import (
 
 const headerContentEncoding = "Content-Encoding"
 const defaultMaxRequestBodySize = 20 * 1024 * 1024 // 20MiB
-var defaultCompressionAlgorithms = []string{"", "gzip", "zstd", "zlib", "snappy", "deflate"}
+var defaultCompressionAlgorithms = []string{"", "gzip", "zstd", "zlib", "snappy", "deflate", "lz4"}
 
 // ClientConfig defines settings for creating an HTTP client.
 type ClientConfig struct {

config/confighttp/go.mod

@@ -5,6 +5,7 @@ go 1.22.0
 require (
 	github.com/golang/snappy v0.0.4
 	github.com/klauspost/compress v1.17.10
+	github.com/pierrec/lz4/v4 v4.1.21
 	github.com/rs/cors v1.11.1
 	github.com/stretchr/testify v1.9.0
 	go.opentelemetry.io/collector/client v1.17.0

exporter/otlphttpexporter/go.mod

@@ -42,6 +42,7 @@ require (
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
+	github.com/pierrec/lz4/v4 v4.1.21 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/rs/cors v1.11.1 // indirect
 	go.opentelemetry.io/collector/client v1.17.0 // indirect

extension/zpagesextension/go.mod

@@ -34,6 +34,7 @@ require (
 	github.com/knadh/koanf/v2 v2.1.1 // indirect
 	github.com/mitchellh/copystructure v1.2.0 // indirect
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
+	github.com/pierrec/lz4/v4 v4.1.21 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/rs/cors v1.11.1 // indirect
 	go.opentelemetry.io/collector/client v1.17.0 // indirect

internal/e2e/go.mod

@@ -60,6 +60,7 @@ require (
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/mostynb/go-grpc-compression v1.2.3 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
+	github.com/pierrec/lz4/v4 v4.1.21 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
 	github.com/prometheus/client_golang v1.20.4 // indirect

receiver/otlpreceiver/go.mod

@@ -48,6 +48,7 @@ require (
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/mostynb/go-grpc-compression v1.2.3 // indirect
+	github.com/pierrec/lz4/v4 v4.1.21 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/rs/cors v1.11.1 // indirect
 	go.opentelemetry.io/collector/client v1.17.0 // indirect