Fix snappy/framed-snappy encoding/decoding

.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 @@
 	"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 @@
 		}
 		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 @@
 
 	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" {