Revert "What if we used zstd in our trace writers (#23806)" (#25278)

This reverts commit e496f86.

Author: ajgajg1134

pkg/trace/go.mod

@@ -20,7 +20,6 @@ require (
 	github.com/DataDog/datadog-go/v5 v5.5.0
 	github.com/DataDog/opentelemetry-mapping-go/pkg/otlp/attributes v0.14.0
 	github.com/DataDog/sketches-go v1.4.2
-	github.com/DataDog/zstd v1.5.5
 	github.com/Microsoft/go-winio v0.6.1
 	github.com/cihub/seelog v0.0.0-20170130134532-f561c5e57575
 	github.com/davecgh/go-spew v1.1.1

pkg/trace/go.sum

@@ -8,7 +8,6 @@ package writer
 import (
 	"compress/gzip"
 	"errors"
-	"io"
 	"runtime"
 	"strings"
 	"sync"
@@ -22,7 +21,6 @@ import (
 	"github.com/DataDog/datadog-agent/pkg/trace/timing"
 
 	"github.com/DataDog/datadog-go/v5/statsd"
-	"github.com/DataDog/zstd"
 )
 
 // pathTraces is the target host API path for delivering traces.
@@ -71,7 +69,7 @@ type TraceWriter struct {
 	senders      []*sender
 	stop         chan struct{}
 	stats        *info.TraceWriterInfo
-	wg           sync.WaitGroup // waits for compressors
+	wg           sync.WaitGroup // waits for gzippers
 	tick         time.Duration  // flush frequency
 	agentVersion string
 
@@ -87,7 +85,6 @@ type TraceWriter struct {
 	easylog *log.ThrottledLogger
 	statsd  statsd.ClientInterface
 	timing  timing.Reporter
-	useZstd bool
 }
 
 // NewTraceWriter returns a new TraceWriter. It is created for the given agent configuration and
@@ -118,7 +115,6 @@ func NewTraceWriter(
 		telemetryCollector: telemetryCollector,
 		statsd:             statsd,
 		timing:             timing,
-		useZstd:            cfg.HasFeature("zstd-encoding"),
 	}
 	climit := cfg.TraceWriter.ConnectionLimit
 	if climit == 0 {
@@ -286,43 +282,25 @@ func (w *TraceWriter) serializer() {
 			}
 
 			w.stats.BytesUncompressed.Add(int64(len(b)))
-			var p *payload
-			var writer io.WriteCloser
-
-			if w.useZstd {
-				p = newPayload(map[string]string{
-					"Content-Type":     "application/x-protobuf",
-					"Content-Encoding": "zstd",
-					headerLanguages:    strings.Join(info.Languages(), "|"),
-				})
-
-				p.body.Grow(len(b) / 2)
-				writer = zstd.NewWriterLevel(p.body, zstd.BestSpeed)
-
-			} else {
-				p = newPayload(map[string]string{
-					"Content-Type":     "application/x-protobuf",
-					"Content-Encoding": "gzip",
-					headerLanguages:    strings.Join(info.Languages(), "|"),
-				})
-				p.body.Grow(len(b) / 2)
-
-				writer, err = gzip.NewWriterLevel(p.body, gzip.BestSpeed)
-				if err != nil {
-					// it will never happen, unless an invalid compression is chosen;
-					// we know gzip.BestSpeed is valid.
-					log.Errorf("Failed to compress trace paylod: %s", err)
-					return
-				}
+			p := newPayload(map[string]string{
+				"Content-Type":     "application/x-protobuf",
+				"Content-Encoding": "gzip",
+				headerLanguages:    strings.Join(info.Languages(), "|"),
+			})
+			p.body.Grow(len(b) / 2)
+			gzipw, err := gzip.NewWriterLevel(p.body, gzip.BestSpeed)
+			if err != nil {
+				// it will never happen, unless an invalid compression is chosen;
+				// we know gzip.BestSpeed is valid.
+				log.Errorf("gzip.NewWriterLevel: %d", err)
+				return
 			}
-
-			if _, err := writer.Write(b); err != nil {
-				log.Errorf("Error compressing trace payload: %v", err)
+			if _, err := gzipw.Write(b); err != nil {
+				log.Errorf("Error gzipping trace payload: %v", err)
 			}
-			if err := writer.Close(); err != nil {
-				log.Errorf("Error closing compressed stream when writing trace payload: %v", err)
+			if err := gzipw.Close(); err != nil {
+				log.Errorf("Error closing gzip stream when writing trace payload: %v", err)
 			}
-
 			sendPayloads(w.senders, p, w.syncMode)
 		}()
 	}

pkg/trace/writer/trace.go

@@ -7,13 +7,11 @@ package writer
 
 import (
 	"compress/gzip"
-	"fmt"
 	"io"
 	"runtime"
 	"sync"
 	"testing"
 
-	"github.com/DataDog/zstd"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/proto"
@@ -44,45 +42,38 @@ func (s MockSampler) GetTargetTPS() float64 {
 var mockSampler = MockSampler{TargetTPS: 5, Enabled: true}
 
 func TestTraceWriter(t *testing.T) {
-	testCases := []bool{false, true}
+	srv := newTestServer()
+	cfg := &config.AgentConfig{
+		Hostname:   testHostname,
+		DefaultEnv: testEnv,
+		Endpoints: []*config.Endpoint{{
+			APIKey: "123",
+			Host:   srv.URL,
+		}},
+		TraceWriter: &config.WriterConfig{ConnectionLimit: 200, QueueSize: 40},
+	}
 
-	for _, tc := range testCases {
-		srv := newTestServer()
-		cfg := &config.AgentConfig{
-			Hostname:   testHostname,
-			DefaultEnv: testEnv,
-			Endpoints: []*config.Endpoint{{
-				APIKey: "123",
-				Host:   srv.URL,
-			}},
-			TraceWriter: &config.WriterConfig{ConnectionLimit: 200, QueueSize: 40},
+	t.Run("ok", func(t *testing.T) {
+		testSpans := []*SampledChunks{
+			randomSampledSpans(20, 8),
+			randomSampledSpans(10, 0),
+			randomSampledSpans(40, 5),
 		}
-		if tc {
-			cfg.Features = map[string]struct{}{"zstd-encoding": {}}
+		// Use a flush threshold that allows the first two entries to not overflow,
+		// but overflow on the third.
+		defer useFlushThreshold(testSpans[0].Size + testSpans[1].Size + 10)()
+		tw := NewTraceWriter(cfg, mockSampler, mockSampler, mockSampler, telemetry.NewNoopCollector(), &statsd.NoOpClient{}, &timing.NoopReporter{})
+		tw.In = make(chan *SampledChunks)
+		go tw.Run()
+		for _, ss := range testSpans {
+			tw.In <- ss
 		}
-
-		t.Run(fmt.Sprintf("zstd_%t", tc), func(t *testing.T) {
-			testSpans := []*SampledChunks{
-				randomSampledSpans(20, 8),
-				randomSampledSpans(10, 0),
-				randomSampledSpans(40, 5),
-			}
-			// Use a flush threshold that allows the first two entries to not overflow,
-			// but overflow on the third.
-			defer useFlushThreshold(testSpans[0].Size + testSpans[1].Size + 10)()
-			tw := NewTraceWriter(cfg, mockSampler, mockSampler, mockSampler, telemetry.NewNoopCollector(), &statsd.NoOpClient{}, &timing.NoopReporter{})
-			tw.In = make(chan *SampledChunks)
-			go tw.Run()
-			for _, ss := range testSpans {
-				tw.In <- ss
-			}
-			tw.Stop()
-			// One payload flushes due to overflowing the threshold, and the second one
-			// because of stop.
-			assert.Equal(t, 2, srv.Accepted())
-			payloadsContain(t, srv.Payloads(), testSpans, tc)
-		})
-	}
+		tw.Stop()
+		// One payload flushes due to overflowing the threshold, and the second one
+		// because of stop.
+		assert.Equal(t, 2, srv.Accepted())
+		payloadsContain(t, srv.Payloads(), testSpans)
+	})
 }
 
 func TestTraceWriterMultipleEndpointsConcurrent(t *testing.T) {
@@ -131,7 +122,7 @@ func TestTraceWriterMultipleEndpointsConcurrent(t *testing.T) {
 
 	wg.Wait()
 	tw.Stop()
-	payloadsContain(t, srv.Payloads(), testSpans, false)
+	payloadsContain(t, srv.Payloads(), testSpans)
 }
 
 // useFlushThreshold sets n as the number of bytes to be used as the flush threshold
@@ -154,25 +145,14 @@ func randomSampledSpans(spans, events int) *SampledChunks {
 }
 
 // payloadsContain checks that the given payloads contain the given set of sampled spans.
-func payloadsContain(t *testing.T, payloads []*payload, sampledSpans []*SampledChunks, shouldUseZstd bool) {
+func payloadsContain(t *testing.T, payloads []*payload, sampledSpans []*SampledChunks) {
 	t.Helper()
 	var all pb.AgentPayload
 	for _, p := range payloads {
 		assert := assert.New(t)
-		var slurp []byte
-		var err error
-		var reader io.ReadCloser
-
-		if shouldUseZstd {
-			reader = zstd.NewReader(p.body)
-			assert.NotNil(reader)
-		} else {
-			reader, err = gzip.NewReader(p.body)
-			assert.NoError(err)
-		}
-
-		slurp, err = io.ReadAll(reader)
-
+		gzipr, err := gzip.NewReader(p.body)
+		assert.NoError(err)
+		slurp, err := io.ReadAll(gzipr)
 		assert.NoError(err)
 		var payload pb.AgentPayload
 		err = proto.Unmarshal(slurp, &payload)
@@ -227,7 +207,7 @@ func TestTraceWriterFlushSync(t *testing.T) {
 		tw.FlushSync()
 		// Now all trace payloads should be sent
 		assert.Equal(t, 1, srv.Accepted())
-		payloadsContain(t, srv.Payloads(), testSpans, false)
+		payloadsContain(t, srv.Payloads(), testSpans)
 	})
 }
 
@@ -296,7 +276,7 @@ func TestTraceWriterSyncStop(t *testing.T) {
 		tw.Stop()
 		// Now all trace payloads should be sent
 		assert.Equal(t, 1, srv.Accepted())
-		payloadsContain(t, srv.Payloads(), testSpans, false)
+		payloadsContain(t, srv.Payloads(), testSpans)
 	})
 }