Fix snappy/framed-snappy encoding/decoding (#12911)

#### Description
This is an alternative PR to #12825.

I'm taking all the commits from that PR and adding the feature gate on
the client side, as requested by reviews. The server behavior of peeking
into the initial bytes to identify the encoding is kept :)

If you've reviewed the previous PR, you can just review the latest
commit!

<!-- Issue number if applicable -->
#### Link to tracking issue
Fixes #10584

---------

Signed-off-by: Arthur Silva Sens <[email protected]>
Co-authored-by: Michael Graff <[email protected]>
Co-authored-by: Alex Boten <[email protected]>
Co-authored-by: Pablo Baeyens <[email protected]>
Co-authored-by: Jonathan <[email protected]>

Author: 5 people

.chloggen/snappy-done-right.yaml

@@ -0,0 +1,36 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: confighttp and configcompression
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Fix handling of `snappy` content-encoding in a backwards-compatible way"
+
+# One or more tracking issues or pull requests related to the change
+issues: [10584, 12825]
+
+# (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: |
+  The collector used the Snappy compression type of "framed" to handle the HTTP
+  content-encoding "snappy".  However, this encoding is typically used to indicate
+  the "block" compression variant of "snappy".  This change allows the collector to:
+  - When receiving a request with encoding 'snappy', the server endpoints will peek
+    at the first bytes of the payload to determine if it is "framed" or "block" snappy,
+    and will decompress accordingly.  This is a backwards-compatible change.
+
+  If the feature-gate "confighttp.framedSnappy" is enabled, you'll see new behavior for both client and server:
+  - Client compression type "snappy" will now compress to the "block" variant of snappy
+    instead of "framed". Client compression type "x-snappy-framed" will now compress to the "framed" variant of snappy.
+  - Servers will accept both "snappy" and "x-snappy-framed" as valid content-encodings.
+
+# 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]

config/configcompression/compressiontype.go

@@ -24,6 +24,7 @@ const (
 	TypeZlib                Type = "zlib"
 	TypeDeflate             Type = "deflate"
 	TypeSnappy              Type = "snappy"
+	TypeSnappyFramed        Type = "x-snappy-framed"
 	TypeZstd                Type = "zstd"
 	TypeLz4                 Type = "lz4"
 	typeNone                Type = "none"
@@ -43,6 +44,7 @@ func (ct *Type) UnmarshalText(in []byte) error {
 		typ == TypeZlib ||
 		typ == TypeDeflate ||
 		typ == TypeSnappy ||
+		typ == TypeSnappyFramed ||
 		typ == TypeZstd ||
 		typ == TypeLz4 ||
 		typ == typeNone ||

config/configcompression/compressiontype_test.go

@@ -42,6 +42,12 @@ func TestUnmarshalText(t *testing.T) {
 			shouldError:     false,
 			isCompressed:    true,
 		},
+		{
+			name:            "ValidSnappyFramed",
+			compressionName: []byte("x-snappy-framed"),
+			shouldError:     false,
+			isCompressed:    true,
+		},
 		{
 			name:            "ValidZstd",
 			compressionName: []byte("zstd"),
@@ -128,6 +134,17 @@ func TestValidateParams(t *testing.T) {
 			compressionLevel: 1,
 			shouldError:      true,
 		},
+		{
+			name:            "ValidSnappyFramed",
+			compressionName: []byte("x-snappy-framed"),
+			shouldError:     false,
+		},
+		{
+			name:             "InvalidSnappyFramed",
+			compressionName:  []byte("x-snappy-framed"),
+			compressionLevel: 1,
+			shouldError:      true,
+		},
 		{
 			name:             "ValidZstd",
 			compressionName:  []byte("zstd"),

config/confighttp/README.md

@@ -48,6 +48,8 @@ README](../configtls/README.md).
       - SpeedBestCompression: `11`
     - `snappy`
       No compression levels supported yet
+    - `x-snappy-framed` (When feature gate `confighttp.framedSnappy` is enabled)
+      No compression levels supported yet
 - [`max_idle_conns`](https://golang.org/pkg/net/http/#Transport)
 - [`max_idle_conns_per_host`](https://golang.org/pkg/net/http/#Transport)
 - [`max_conns_per_host`](https://golang.org/pkg/net/http/#Transport)
@@ -105,6 +107,7 @@ will not be enabled.
 - `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", "lz4"]
+  - `x-snappy-framed` can be used if feature gate `confighttp.snappyFramed` is enabled.
 - [`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

@@ -6,9 +6,11 @@
 package confighttp // import "go.opentelemetry.io/collector/config/confighttp"
 
 import (
+	"bufio"
 	"bytes"
 	"compress/gzip"
 	"compress/zlib"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -18,6 +20,15 @@ import (
 	"github.com/pierrec/lz4/v4"
 
 	"go.opentelemetry.io/collector/config/configcompression"
+	"go.opentelemetry.io/collector/featuregate"
+)
+
+var enableFramedSnappy = featuregate.GlobalRegistry().MustRegister(
+	"confighttp.framedSnappy",
+	featuregate.StageAlpha,
+	featuregate.WithRegisterFromVersion("v0.125.0"),
+	featuregate.WithRegisterDescription("Content encoding 'snappy' will compress/decompress block snappy format while 'x-snappy-framed' will compress/decompress framed snappy format."),
+	featuregate.WithRegisterReferenceURL("https://github.com/open-telemetry/opentelemetry-collector/issues/10584"),
 )
 
 type compressRoundTripper struct {
@@ -60,23 +71,65 @@ var availableDecoders = map[string]func(body io.ReadCloser) (io.ReadCloser, erro
 		}
 		return zr, nil
 	},
+	"snappy": snappyHandler,
 	//nolint:unparam // Ignoring the linter request to remove error return since it needs to match the method signature
-	"snappy": func(body io.ReadCloser) (io.ReadCloser, error) {
-		// Lazy Reading content to improve memory efficiency
+	"lz4": func(body io.ReadCloser) (io.ReadCloser, error) {
 		return &compressReadCloser{
-			Reader: snappy.NewReader(body),
+			Reader: lz4.NewReader(body),
 			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) {
+	"x-snappy-framed": func(body io.ReadCloser) (io.ReadCloser, error) {
 		return &compressReadCloser{
-			Reader: lz4.NewReader(body),
+			Reader: snappy.NewReader(body),
 			orig:   body,
 		}, nil
 	},
 }
 
+// snappyFramingHeader is always the first 10 bytes of a snappy framed stream.
+var snappyFramingHeader = []byte{
+	0xff, 0x06, 0x00, 0x00,
+	0x73, 0x4e, 0x61, 0x50, 0x70, 0x59, // "sNaPpY"
+}
+
+// snappyHandler returns an io.ReadCloser that auto-detects the snappy format.
+// This is necessary because the collector previously used "content-encoding: snappy"
+// but decompressed and compressed the payloads using the snappy framing format.
+// However, "content-encoding: snappy" is uses the block format, and "x-snappy-framed"
+// is the framing format.  This handler is a (hopefully temporary) hack to
+// make this work in a backwards-compatible way.
+//
+// See https://github.com/google/snappy/blob/6af9287fbdb913f0794d0148c6aa43b58e63c8e3/framing_format.txt#L27-L36
+// for more details on the framing format.
+func snappyHandler(body io.ReadCloser) (io.ReadCloser, error) {
+	br := bufio.NewReader(body)
+
+	peekBytes, err := br.Peek(len(snappyFramingHeader))
+	if err != nil && !errors.Is(err, io.EOF) {
+		return nil, err
+	}
+
+	isFramed := len(peekBytes) >= len(snappyFramingHeader) && bytes.Equal(peekBytes[:len(snappyFramingHeader)], snappyFramingHeader)
+
+	if isFramed {
+		return &compressReadCloser{
+			Reader: snappy.NewReader(br),
+			orig:   body,
+		}, nil
+	}
+	compressed, err := io.ReadAll(br)
+	if err != nil {
+		return nil, err
+	}
+	decoded, err := snappy.Decode(nil, compressed)
+	if err != nil {
+		return nil, err
+	}
+	return io.NopCloser(bytes.NewReader(decoded)), nil
+}
+
 func newCompressionParams(level configcompression.Level) configcompression.CompressionParams {
 	return configcompression.CompressionParams{
 		Level: level,
@@ -143,6 +196,9 @@ func httpContentDecompressor(h http.Handler, maxRequestBodySize int64, eh func(w
 
 	enabled := map[string]func(body io.ReadCloser) (io.ReadCloser, error){}
 	for _, dec := range enableDecoders {
+		if dec == "x-frame-snappy" && !enableFramedSnappy.IsEnabled() {
+			continue
+		}
 		enabled[dec] = availableDecoders[dec]
 
 		if dec == "deflate" {