fix(profiler): don't double compress non-delta profiles (#3918)

Our non-delta heap, block, and mutex profiles are getting
double-compressed. That is, they are gzip-compressed, and if you
decompress them then you get another gzip-compressed blob, which is in
turn a compressed pprof file.

This is due to a bug introduced by #3529. In that PR we reworked our
compression logic in order to support re-compression (into zstd in
particular). We incorrectly decided whether a given profile type is a
delta profile by just checking whether there are delta values for the
profile type, without checking whether delta profiling is actually enabled.
As a result, when delta profiling is disabled we use the delta profling
enabled logic, where we assume the input data is uncomprossed and then
gzip-compress it.

Author: nsrip-dd

profiler/profiler.go

@@ -259,7 +259,7 @@ func newProfiler(opts ...Option) (*profiler, error) {
 		types = append(types, executionTrace)
 	}
 	for _, pt := range types {
-		isDelta := len(profileTypes[pt].DeltaValues) > 0
+		isDelta := p.cfg.deltaProfiles && len(profileTypes[pt].DeltaValues) > 0
 		in, out := compressionStrategy(pt, isDelta, p.cfg.compressionConfig)
 		compressor, err := newCompressionPipeline(in, out)
 		if err != nil {

profiler/profiler_test.go

@@ -7,6 +7,7 @@ package profiler
 
 import (
 	"bytes"
+	"compress/gzip"
 	"context"
 	"encoding/json"
 	"fmt"
@@ -35,6 +36,7 @@ import (
 	"github.com/DataDog/dd-trace-go/v2/internal/traceprof"
 	"github.com/DataDog/dd-trace-go/v2/internal/version"
 
+	pprofile "github.com/google/pprof/profile"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -900,3 +902,42 @@ func TestMetricsProfileStopEarlyNoLog(t *testing.T) {
 		}
 	}
 }
+
+func gzipDecompress(data []byte) ([]byte, error) {
+	r, err := gzip.NewReader(bytes.NewReader(data))
+	if err != nil {
+		return nil, err
+	}
+	return io.ReadAll(r)
+}
+
+func TestHeapProfileCompression(t *testing.T) {
+	t.Run("delta", func(t *testing.T) { testHeapProfileCompression(t, true) })
+	t.Run("non-delta", func(t *testing.T) { testHeapProfileCompression(t, false) })
+}
+
+func testHeapProfileCompression(t *testing.T, delta bool) {
+	profiles := startTestProfiler(t, 1,
+		WithPeriod(10*time.Millisecond), WithProfileTypes(HeapProfile), WithDeltaProfiles(delta),
+	)
+	p := <-profiles
+	attachment := "heap.pprof"
+	if delta {
+		attachment = "delta-heap.pprof"
+	}
+	data, ok := p.attachments[attachment]
+	if !ok {
+		t.Fatalf("no heap profile, got %s", p.event.Attachments)
+	}
+	decompressed, err := gzipDecompress(data)
+	if err != nil {
+		t.Fatalf("decompressing the heap profile failed: %s", err)
+	}
+	t.Logf("%x", decompressed[:16])
+	// We assume the profile is gzip compressed. The pprof pacakge
+	// can parse gzip-compressed profiles (it checks for the magic number).
+	// So we should be able to parse the original data
+	if _, err := pprofile.ParseData(data); err != nil {
+		t.Fatalf("parsing profile data failed: %s", err)
+	}
+}