fix(profiler): support gzip compression mode [#4696 backport]

felixge · nsrip-dd · commit ce45805ec3c6 · 2026-04-30T14:15:25.000-04:00
Backport #4696 to the v2.7.x release branch Adds a functional gzip compression mode for the profiler via `DD_PROFILING_DEBUG_COMPRESSION_SETTINGS=gzip`. This was already kinda working, but not really, hence the "fix" title. - Defaults `gzip` to gzip-6, matching the gzip level legacy mode applies when it compresses profiler-produced profiles. - Adds gzip-to-gzip recompression so gzip mode works for runtime profiles that are already gzip-compressed at gzip-1. - Adds tests for gzip mode and gzip recompression. This is done for incident-53386, where zstd caused a memory usage increase. The gzip mode provides an alternative to the default zstd compression while keeping profiler uploads compressed.
diff --git a/profiler/compression.go b/profiler/compression.go
@@ -48,11 +48,20 @@ func compressionStrategy(pt ProfileType, isDelta bool, config string) (compressi
 	if config == "" || config == "legacy" {
 		return legacyCompressionStrategy(pt, isDelta)
 	}
-	algorithm, levelStr, _ := strings.Cut(config, "-")
+	algorithmStr, levelStr, _ := strings.Cut(config, "-")
+	algorithm := compressionAlgorithm(algorithmStr)
 	// Don't bother checking the error. We'll get zero which represents the
 	// default, and we we assume this is only going to get used internally
 	level, _ := strconv.Atoi(levelStr)
-	return inputCompression(pt, isDelta), compression{algorithm: compressionAlgorithm(algorithm), level: level}
+	if levelStr == "" {
+		switch algorithm {
+		case compressionAlgorithmGzip:
+			level = gzip6Compression.level
+		case compressionAlgorithmZstd:
+			level = zstdCompression.level
+		}
+	}
+	return inputCompression(pt, isDelta), compression{algorithm: algorithm, level: level}
 }
 
 // inputCompression maps the given profile type and isDelta flavor to the
@@ -172,6 +181,14 @@ func (b *compressionPipelineBuilder) Build(in compression, out compression) (com
 		return b.getZstdEncoder(getZstdLevelOrDefault(out.level))
 	}
 
+	if in.algorithm == compressionAlgorithmGzip && out.algorithm == compressionAlgorithmGzip {
+		gzipOut, err := kgzip.NewWriterLevel(nil, out.level)
+		if err != nil {
+			return nil, err
+		}
+		return newGzipRecompressor(gzipOut), nil
+	}
+
 	if in.algorithm == compressionAlgorithmGzip && out.algorithm == compressionAlgorithmZstd {
 		encoder, err := b.getZstdEncoder(getZstdLevelOrDefault(out.level))
 		if err != nil {
@@ -213,6 +230,43 @@ func (r *passthroughCompressor) Close() error {
 	return nil
 }
 
+func newGzipRecompressor(gzipOut *kgzip.Writer) *gzipRecompressor {
+	return &gzipRecompressor{gzipOut: gzipOut, err: make(chan error)}
+}
+
+type gzipRecompressor struct {
+	// err synchronizes finishing writes after closing pw and reports any
+	// error during recompression
+	err     chan error
+	pw      io.WriteCloser
+	gzipOut *kgzip.Writer
+}
+
+func (r *gzipRecompressor) Reset(w io.Writer) {
+	r.gzipOut.Reset(w)
+	pr, pw := io.Pipe()
+	go func() {
+		gzr, err := kgzip.NewReader(pr)
+		if err != nil {
+			r.err <- err
+			return
+		}
+		_, err = io.Copy(r.gzipOut, gzr)
+		r.err <- err
+	}()
+	r.pw = pw
+}
+
+func (r *gzipRecompressor) Write(p []byte) (int, error) {
+	return r.pw.Write(p)
+}
+
+func (r *gzipRecompressor) Close() error {
+	r.pw.Close()
+	err := <-r.err
+	return cmp.Or(err, r.gzipOut.Close())
+}
+
 func newZstdRecompressor(encoder *sharedZstdEncoder) *zstdRecompressor {
 	return &zstdRecompressor{zstdOut: encoder, err: make(chan error)}
 }
diff --git a/profiler/compression_test.go b/profiler/compression_test.go
@@ -37,6 +37,8 @@ func TestNewCompressionPipeline(t *testing.T) {
 		{noCompression, noCompression, plainData, plainData},
 		{gzip1Compression, gzip1Compression, gzip1Data, gzip1Data},
 		{gzip6Compression, gzip6Compression, gzip6Data, gzip6Data},
+		{gzip1Compression, gzip6Compression, gzip1Data, gzip6Data},
+		{gzip6Compression, gzip1Compression, gzip6Data, gzip1Data},
 		{gzip1Compression, zstdCompression, gzip1Data, zstdData},
 		{gzip6Compression, zstdCompression, gzip6Data, zstdData},
 	}
@@ -99,6 +101,15 @@ func TestDebugCompressionEnv(t *testing.T) {
 		require.NoError(t, err)
 	})
 
+	t.Run("explicit-gzip-already-gzipped-input", func(t *testing.T) {
+		t.Setenv("DD_PROFILING_DEBUG_COMPRESSION_SETTINGS", "gzip")
+		p := startTestProfiler(t, 1, WithProfileTypes(CPUProfile), WithPeriod(time.Millisecond)).ReceiveProfile(t)
+		r, err := gzip.NewReader(bytes.NewReader(p.attachments["cpu.pprof"]))
+		require.NoError(t, err)
+		_, err = io.Copy(io.Discard, r)
+		require.NoError(t, err)
+	})
+
 	t.Run("zstd-delta", func(t *testing.T) {
 		t.Setenv("DD_PROFILING_DEBUG_COMPRESSION_SETTINGS", "zstd-3")
 		p := startTestProfiler(t, 1, WithProfileTypes(CPUProfile, HeapProfile), WithPeriod(time.Millisecond)).ReceiveProfile(t)
