Skip to content

Commit 76964f2

Browse files
authored
[otel-agent] Use zstd compression for all signals (#52758)
### What does this PR do? Makes the DDOT collector (`otel-agent`) compress **every** signal with `zstd`, replacing the previous per-signal mix (zlib for metrics, gzip for traces, zstd for logs): - **Metrics**: use the config-driven metrics compressor (`metricscompression/fx`) instead of the hardcoded zlib module; defaults to **zstd level 3**, overridable via `serializer_zstd_compressor_level`. - **Logs**: pin zstd and default `logs_config.zstd_compression_level` to **3** (overridable). - **Traces**: use the existing `fx-zstd` module (zstd at `BestSpeed`); the trace level is intentionally fixed. - **Host metadata**: rides the same serializer compressor as metrics (no dedicated knob), so it becomes zstd automatically. Also adds tests that guard against future per-signal compression divergence. ### Motivation The DDOT exporter used a different compression algorithm per signal, which is inconsistent and harder to reason about. Standardizing on `zstd` gives a consistent algorithm across metrics/traces/logs and a better compression ratio, with the level configurable per signal. ### Describe how you validated your changes - `dda inv otel-agent.build` passes; `dda inv linter.go` → **0 issues** on changed packages; `gofmt` clean. - **Unit** (`cmd/otel-agent/config`): asserts metrics and logs resolve to the **same** algorithm (zstd) at level 3, that the level stays env-overridable, and that **host metadata** uses that same serializer compressor (so it is zstd too). - **Integration** (`comp/otelcol/otlp/integrationtest`): now decodes trace payloads as zstd — verifies traces ship valid zstd end-to-end. Passes locally (`dda inv otel-agent.integration-test`). - **E2E** (`TestOTelAgentComplete/TestOTLPCompression`): asserts metrics, traces, and logs reach fakeintake with `Content-Encoding: zstd` (requires the e2e infra to run). ### Additional Notes - **Scope is the standalone `otel-agent`.** The core Agent, trace-agent, host-profiler, and OSS exporter are unaffected — they use their own compression modules (`metricscompression/fx`, `fx-zstd`, `fx-gzip`, `fx-otel` respectively). - **The v2 metrics intake is preserved** (`use_v3_api.series.enabled` stays `false`); moving series to v3 is a separate effort. ⚠️ **Reviewers:** please confirm the v2 series intake (`/api/v2/series`) accepts `Content-Encoding: zstd` — the core Agent uses zstd→v3 for series by default, so this exact combination isn't exercised today. The new e2e test validates it against fakeintake. - **Exhaustive compression coverage** (all the DD exporter's submission clients): after this change, every active intake submission in the otel-agent is zstd — **metrics (series/sketches), host metadata, logs, traces** — *except* **APM stats**, which stays **gzip** (hardcoded in `pkg/trace/writer/stats.go`, shared with the trace-agent; out of scope here, separate conversation). The orchestrator/K8s-objects sub-exporter already uses zstd and is disabled by default. Internal exporter telemetry (`metricsclient`, COAT/gateway gauges) is scraped Prometheus/OTel-meter telemetry, not an intake submission. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: jaime.fullaondo <jaime.fullaondo@datadoghq.com>
1 parent 5ba66e3 commit 76964f2

13 files changed

Lines changed: 387 additions & 19 deletions

File tree

cmd/otel-agent/config/BUILD.bazel

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ go_library(
1919
"//comp/otelcol/otlp/components/exporter/datadogexporter",
2020
"//pkg/config/model",
2121
"//pkg/config/setup",
22+
"//pkg/config/setup/constants",
2223
"@com_github_open_telemetry_opentelemetry_collector_contrib_pkg_datadog//featuregates",
2324
"@io_opentelemetry_go_collector_confmap//:confmap",
2425
"@io_opentelemetry_go_collector_confmap_provider_envprovider//:envprovider",
@@ -32,11 +33,15 @@ go_library(
3233

3334
dd_agent_go_test(
3435
name = "config_test",
35-
srcs = ["agent_config_test.go"],
36+
srcs = [
37+
"agent_config_test.go",
38+
"compression_consistency_test.go",
39+
],
3640
data = glob(["testdata/**"]),
3741
embed = [":config"],
3842
deps = [
3943
"//pkg/config/mock",
44+
"//pkg/util/compression",
4045
"//pkg/util/defaultpaths",
4146
"@com_github_stretchr_testify//assert",
4247
"@com_github_stretchr_testify//require",

cmd/otel-agent/config/agent_config.go

Lines changed: 28 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@ import (
3333
"github.com/DataDog/datadog-agent/comp/otelcol/otlp/components/exporter/datadogexporter"
3434
pkgconfigmodel "github.com/DataDog/datadog-agent/pkg/config/model"
3535
pkgconfigsetup "github.com/DataDog/datadog-agent/pkg/config/setup"
36+
"github.com/DataDog/datadog-agent/pkg/config/setup/constants"
3637
)
3738

3839
type logLevel int
@@ -47,6 +48,11 @@ const (
4748
off
4849
)
4950

51+
// ddotZstdCompressionLevel is the default zstd compression level DDOT applies to
52+
// every signal that exposes a configurable level (metrics and logs). It stays
53+
// overridable via the DD_*_ZSTD_*_LEVEL env vars; this is only the default.
54+
const ddotZstdCompressionLevel = 3
55+
5056
// datadog agent log levels: trace, debug, info, warn, error, critical, and off
5157
// otel log levels: disabled, debug, info, warn, error
5258
var logLevelMap = map[string]logLevel{
@@ -222,8 +228,14 @@ func NewConfigComponent(ctx context.Context, ddCfg string, uris []string) (confi
222228
pkgconfig.Set("skip_ssl_validation", ddc.ClientConfig.TLS.InsecureSkipVerify, pkgconfigmodel.SourceFile)
223229
}
224230

225-
// The otel-agent forces zlib compression, which is incompatible with the v3
226-
// metrics intake.
231+
// Compression: the otel-agent (DDOT) uses zstd for every signal (metrics, traces,
232+
// logs) so the compression algorithm stays consistent across signals. The level
233+
// defaults to 3 but stays overridable via DD_SERIALIZER_ZSTD_COMPRESSOR_LEVEL
234+
// (SourceDefault < SourceEnvVar). DDOT deliberately stays on the v2 metrics intake:
235+
// zstd is v3-compatible, but moving to v3 is a separate effort, so v3 is disabled
236+
// here regardless of the compressor.
237+
pkgconfig.Set("serializer_compressor_kind", constants.DefaultCompressorKind, pkgconfigmodel.SourceDefault)
238+
pkgconfig.Set("serializer_zstd_compressor_level", ddotZstdCompressionLevel, pkgconfigmodel.SourceDefault)
227239
pkgconfig.Set("use_v3_api.series.enabled", "false", pkgconfigmodel.SourceAgentRuntime)
228240

229241
// Log configs
@@ -232,7 +244,21 @@ func NewConfigComponent(ctx context.Context, ddCfg string, uris []string) (confi
232244
pkgconfig.Set("logs_config.logs_dd_url", ddc.Logs.Endpoint, pkgconfigmodel.SourceFile)
233245
pkgconfig.Set("logs_config.batch_wait", ddc.Logs.BatchWait, pkgconfigmodel.SourceFile)
234246
pkgconfig.Set("logs_config.use_compression", ddc.Logs.UseCompression, pkgconfigmodel.SourceFile)
247+
// logs_config.compression_level carries the exporter's logs::compression_level
248+
// (a gzip level, 0-9); it only applies when the active log compressor is gzip.
235249
pkgconfig.Set("logs_config.compression_level", ddc.Logs.CompressionLevel, pkgconfigmodel.SourceFile)
250+
// DDOT logs use zstd to match metrics/traces. compression_kind is set at SourceFile
251+
// (not SourceDefault) so config.IsConfigured() is true, which bypasses the logs
252+
// pipeline's fallback to gzip when logs_config.additional_endpoints is set. That
253+
// fallback is a conservative default for non-Datadog intakes (PR #35625), but
254+
// additional_endpoints here are other Datadog endpoints (multi-region / dual-ship /
255+
// MRF) that accept zstd, and the metrics forwarder already sends zstd to all of
256+
// them. The logs pipeline shares one compressor across destinations, so this makes
257+
// every log endpoint use zstd. Override with DD_LOGS_CONFIG_COMPRESSION_KIND=gzip
258+
// if a non-Datadog log endpoint is ever added. The zstd level defaults to 3,
259+
// overridable via DD_LOGS_CONFIG_ZSTD_COMPRESSION_LEVEL.
260+
pkgconfig.Set("logs_config.compression_kind", constants.DefaultLogCompressionKind, pkgconfigmodel.SourceFile)
261+
pkgconfig.Set("logs_config.zstd_compression_level", ddotZstdCompressionLevel, pkgconfigmodel.SourceDefault)
236262

237263
// APM & OTel trace configs
238264
pkgconfig.Set("apm_config.enabled", true, pkgconfigmodel.SourceDefault)
Lines changed: 162 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,162 @@
1+
// Unless explicitly stated otherwise all files in this repository are licensed
2+
// under the Apache License Version 2.0.
3+
// This product includes software developed at Datadog (https://www.datadoghq.com/).
4+
// Copyright 2016-present Datadog, Inc.
5+
6+
package config
7+
8+
import (
9+
"context"
10+
"testing"
11+
12+
"github.com/stretchr/testify/assert"
13+
"github.com/stretchr/testify/require"
14+
15+
configmock "github.com/DataDog/datadog-agent/pkg/config/mock"
16+
"github.com/DataDog/datadog-agent/pkg/util/compression"
17+
)
18+
19+
// TestDDOTCompressionConsistency guards the invariant that the DDOT (otel-agent)
20+
// exporter compresses every signal with a single algorithm — zstd — so the
21+
// per-signal compressors cannot silently diverge.
22+
//
23+
// Metrics and logs are config-driven: their algorithm and level come from the
24+
// agent-config keys set in NewConfigComponent, asserted here directly.
25+
//
26+
// Traces are NOT config-driven: cmd/otel-agent/subcommands/run/command.go wires
27+
// comp/trace/compression/fx-zstd (zstd at BestSpeed; the level is intentionally
28+
// fixed for traces). "zstd" below is therefore the shared invariant all three
29+
// signals must satisfy. The on-the-wire guard that traces actually ship zstd
30+
// lives in the OTLP integration test (Content-Encoding assertion).
31+
func TestDDOTCompressionConsistency(t *testing.T) {
32+
configmock.New(t)
33+
c, err := NewConfigComponent(context.Background(), "", []string{"testdata/config_default.yaml"})
34+
require.NoError(t, err)
35+
36+
const wantAlgo = "zstd"
37+
const wantLevel = ddotZstdCompressionLevel
38+
39+
metricsKind := c.GetString("serializer_compressor_kind")
40+
logsKind := c.GetString("logs_config.compression_kind")
41+
42+
// All signals share one algorithm by default.
43+
assert.Equal(t, wantAlgo, metricsKind, "metrics must default to zstd")
44+
assert.Equal(t, wantAlgo, logsKind, "logs must default to zstd")
45+
assert.Equal(t, metricsKind, logsKind, "metrics and logs compression algorithms must not diverge")
46+
47+
// logs_config.compression_kind must be IsConfigured() (set at a non-default source)
48+
// so the logs pipeline does NOT fall back to gzip when logs_config.additional_endpoints
49+
// is set. additional_endpoints are other Datadog endpoints (multi-region / dual-ship /
50+
// MRF) that accept zstd, so every log endpoint should ship zstd — matching the metrics
51+
// forwarder, which already fans one zstd payload to all endpoints.
52+
assert.True(t, c.IsConfigured("logs_config.compression_kind"),
53+
"logs_config.compression_kind must be configured so the additional_endpoints gzip fallback is bypassed")
54+
55+
// Levels default to 3 for the signals that support a configurable level.
56+
assert.Equal(t, wantLevel, c.GetInt("serializer_zstd_compressor_level"), "metrics zstd level should default to 3")
57+
assert.Equal(t, wantLevel, c.GetInt("logs_config.zstd_compression_level"), "logs zstd level should default to 3")
58+
59+
// DDOT deliberately stays on the v2 metrics intake (v3 is a separate effort).
60+
assert.Equal(t, "false", c.GetString("use_v3_api.series.enabled"), "DDOT must stay on the v2 metrics intake")
61+
}
62+
63+
// TestDDOTCompressionLevelOverridable verifies the per-signal compression level
64+
// stays overridable (e.g. via DD_* env vars) rather than being forced, so the
65+
// SourceDefault precedence chosen in NewConfigComponent does not lock operators out.
66+
func TestDDOTCompressionLevelOverridable(t *testing.T) {
67+
configmock.New(t)
68+
t.Setenv("DD_SERIALIZER_ZSTD_COMPRESSOR_LEVEL", "6")
69+
t.Setenv("DD_LOGS_CONFIG_ZSTD_COMPRESSION_LEVEL", "9")
70+
71+
c, err := NewConfigComponent(context.Background(), "", []string{"testdata/config_default.yaml"})
72+
require.NoError(t, err)
73+
74+
assert.Equal(t, 6, c.GetInt("serializer_zstd_compressor_level"), "metrics zstd level should be overridable via env")
75+
assert.Equal(t, 9, c.GetInt("logs_config.zstd_compression_level"), "logs zstd level should be overridable via env")
76+
77+
// Overriding only the level must not change the algorithm.
78+
assert.Equal(t, "zstd", c.GetString("serializer_compressor_kind"))
79+
assert.Equal(t, "zstd", c.GetString("logs_config.compression_kind"))
80+
}
81+
82+
// TestDDOTHostMetadataCompression explicitly guards that host metadata — a
83+
// separate submission path from metrics — is compressed with zstd.
84+
//
85+
// The DD exporter pushes host metadata through the agent serializer:
86+
// serializer.SendHostMetadata -> sendMetadata -> split.CheckSizeAndSerialize(m,
87+
// true /*compress*/, s.Strategy). s.Strategy is the serializer's compressor, built
88+
// from the SAME serializer_compressor_kind / serializer_zstd_compressor_level keys
89+
// as the metrics series/sketches path — there is no dedicated host-metadata
90+
// compression knob. So host metadata is zstd (level 3) whenever metrics are; this
91+
// test pins that so the two cannot silently diverge.
92+
func TestDDOTHostMetadataCompression(t *testing.T) {
93+
configmock.New(t)
94+
c, err := NewConfigComponent(context.Background(), "", []string{"testdata/config_default.yaml"})
95+
require.NoError(t, err)
96+
97+
assert.Equal(t, "zstd", c.GetString("serializer_compressor_kind"),
98+
"host metadata shares the metrics compressor (serializer_compressor_kind); both must be zstd")
99+
assert.Equal(t, ddotZstdCompressionLevel, c.GetInt("serializer_zstd_compressor_level"),
100+
"host metadata uses the metrics zstd level")
101+
}
102+
103+
// TestDDOTSupportedCompressors covers every compression algorithm DDOT supports
104+
// per signal — not just the zstd default — on two axes:
105+
//
106+
// 1. Selectability: each supported algorithm can be chosen via the DDOT config
107+
// surface (DD_* env override beats the SourceDefault/SourceFile the
108+
// otel-agent sets), so operators are not locked into zstd.
109+
// 2. Wire encoding: the Content-Encoding each algorithm emits — the value
110+
// fakeintake/the intake observe and the e2e utils.TestCompression asserts —
111+
// is pinned here. Note zlib ships as "deflate" (not "zlib") and none ships
112+
// as "identity"; those non-obvious mappings are the main regression risk.
113+
//
114+
// Metrics (serializer_compressor_kind, shared by series/sketches/host-metadata)
115+
// support zstd/zlib/gzip/none. Logs (logs_config.compression_kind) support
116+
// zstd/gzip, used verbatim as the Content-Encoding. Traces are compile-time
117+
// (the fx-zstd/fx-gzip module in command.go), not a runtime config key, so they
118+
// are out of scope for this config-level test; the OTLP integration test proves
119+
// traces actually ship valid zstd on the wire.
120+
func TestDDOTSupportedCompressors(t *testing.T) {
121+
// Pin the on-the-wire Content-Encoding for each algorithm. These are the
122+
// exact strings e2e/intake see; changing one would silently break the e2e
123+
// compression assertions, so guard them here.
124+
assert.Equal(t, "zstd", compression.ZstdEncoding, "zstd must ship Content-Encoding zstd")
125+
assert.Equal(t, "deflate", compression.ZlibEncoding, "zlib must ship Content-Encoding deflate (not \"zlib\")")
126+
assert.Equal(t, "gzip", compression.GzipEncoding, "gzip must ship Content-Encoding gzip")
127+
128+
// metrics: every supported serializer_compressor_kind must be selectable,
129+
// and each maps to the wire encoding recorded above (none -> "identity").
130+
metrics := []struct{ kind, wantEncoding string }{
131+
{compression.ZstdKind, compression.ZstdEncoding},
132+
{compression.ZlibKind, compression.ZlibEncoding},
133+
{compression.GzipKind, compression.GzipEncoding},
134+
{compression.NoneKind, "identity"},
135+
}
136+
for _, tc := range metrics {
137+
t.Run("metrics/"+tc.kind, func(t *testing.T) {
138+
configmock.New(t)
139+
t.Setenv("DD_SERIALIZER_COMPRESSOR_KIND", tc.kind)
140+
c, err := NewConfigComponent(context.Background(), "", []string{"testdata/config_default.yaml"})
141+
require.NoError(t, err)
142+
assert.Equalf(t, tc.kind, c.GetString("serializer_compressor_kind"),
143+
"operators must be able to select %q for metrics via DD_SERIALIZER_COMPRESSOR_KIND (wire encoding %q)", tc.kind, tc.wantEncoding)
144+
})
145+
}
146+
147+
// logs: only zstd and gzip are supported; the pipeline uses the kind
148+
// verbatim as the Content-Encoding, so the kind is the wire value.
149+
logs := []string{compression.ZstdKind, compression.GzipKind}
150+
for _, kind := range logs {
151+
t.Run("logs/"+kind, func(t *testing.T) {
152+
configmock.New(t)
153+
t.Setenv("DD_LOGS_CONFIG_COMPRESSION_KIND", kind)
154+
c, err := NewConfigComponent(context.Background(), "", []string{"testdata/config_default.yaml"})
155+
require.NoError(t, err)
156+
assert.Equalf(t, kind, c.GetString("logs_config.compression_kind"),
157+
"operators must be able to select %q for logs via DD_LOGS_CONFIG_COMPRESSION_KIND", kind)
158+
assert.Truef(t, c.IsConfigured("logs_config.compression_kind"),
159+
"overriding logs_config.compression_kind to %q must keep it IsConfigured() so the additional_endpoints gzip fallback stays bypassed", kind)
160+
})
161+
}
162+
}

cmd/otel-agent/subcommands/run/BUILD.bazel

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -67,10 +67,10 @@ go_library(
6767
"//comp/otelcol/otlp/components/metricsclient",
6868
"//comp/serializer/logscompression/fx",
6969
"//comp/serializer/metricscompression/def",
70-
"//comp/serializer/metricscompression/fx-otel",
70+
"//comp/serializer/metricscompression/fx",
7171
"//comp/trace/agent/fx",
7272
"//comp/trace/agent/impl",
73-
"//comp/trace/compression/fx-gzip",
73+
"//comp/trace/compression/fx-zstd",
7474
"//comp/trace/config/def",
7575
"//comp/trace/config/impl",
7676
"//comp/trace/payload-modifier/fx",

cmd/otel-agent/subcommands/run/command.go

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -72,10 +72,10 @@ import (
7272
"github.com/DataDog/datadog-agent/comp/otelcol/otlp/components/metricsclient"
7373
logscompressionfx "github.com/DataDog/datadog-agent/comp/serializer/logscompression/fx"
7474
metricscompression "github.com/DataDog/datadog-agent/comp/serializer/metricscompression/def"
75-
metricscompressionfx "github.com/DataDog/datadog-agent/comp/serializer/metricscompression/fx-otel"
75+
metricscompressionfx "github.com/DataDog/datadog-agent/comp/serializer/metricscompression/fx"
7676
traceagentfx "github.com/DataDog/datadog-agent/comp/trace/agent/fx"
7777
traceagentcomp "github.com/DataDog/datadog-agent/comp/trace/agent/impl"
78-
gzipfx "github.com/DataDog/datadog-agent/comp/trace/compression/fx-gzip"
78+
zstdfx "github.com/DataDog/datadog-agent/comp/trace/compression/fx-zstd"
7979
traceconfigdef "github.com/DataDog/datadog-agent/comp/trace/config/def"
8080
traceconfigimpl "github.com/DataDog/datadog-agent/comp/trace/config/impl"
8181
payloadmodifierfx "github.com/DataDog/datadog-agent/comp/trace/payload-modifier/fx"
@@ -268,7 +268,7 @@ func commonAgentFxOptions(ctx context.Context, params *cliParams, acfg coreconfi
268268
fx.Provide(func(cfg traceconfigdef.Component) telemetry.TelemetryCollector {
269269
return telemetry.NewCollector(cfg.Object())
270270
}),
271-
gzipfx.Module(),
271+
zstdfx.Module(),
272272
// ctx is required to be supplied from here, as Windows needs to inject its own context
273273
// to allow the agent to work as a service.
274274
fx.Provide(func() context.Context { return ctx }), // fx.Supply(ctx) fails with a missing type error.

comp/otelcol/otlp/components/exporter/serializerexporter/serializer.go

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -90,8 +90,17 @@ func setupSerializer(config pkgconfigmodel.Config, cfg *ExporterConfig) {
9090

9191
config.Set("use_v2_api.series", true, pkgconfigmodel.SourceDefault)
9292

93-
// The serializer exporter forces zlib compression (metricscompressionfx
94-
// fx-otel), which is incompatible with the v3 metrics intake.
93+
// Scope: this function only runs for the standalone OSS Datadog exporter shipped in
94+
// opentelemetry-collector-contrib. That is the single path today where
95+
// serializerexporter builds its own serializer (the `f.s == nil` branch in
96+
// factory.go); DDOT (cmd/otel-agent) and the core Agent's OTLP ingestion both inject
97+
// an already-built serializer and never reach this code.
98+
//
99+
// On this OSS path the compressor is wired by the caller (initSerializerInternal) via
100+
// metricscompression/fx-otel → NewCompressorReqOtel → a hardcoded zlib.New(). The
101+
// serializer uses that compressor object directly, so the serializer_compressor_kind
102+
// = zstd set above has no effect here: series are sent zlib-compressed
103+
// (Content-Encoding: deflate). Only the v2 intake accepts zlib, so v3 is disabled below.
95104
config.Set("use_v3_api.series.enabled", "false", pkgconfigmodel.SourceAgentRuntime)
96105

97106
// Serializer: allow user to blacklist any kind of payload to be sent

comp/otelcol/otlp/integrationtest/BUILD.bazel

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -55,10 +55,10 @@ dd_agent_go_test(
5555
"//comp/otelcol/otlp/testutil",
5656
"//comp/serializer/logscompression/fx",
5757
"//comp/serializer/metricscompression/def",
58-
"//comp/serializer/metricscompression/fx-otel",
58+
"//comp/serializer/metricscompression/fx",
5959
"//comp/trace",
6060
"//comp/trace/agent/impl",
61-
"//comp/trace/compression/fx-gzip",
61+
"//comp/trace/compression/fx-zstd",
6262
"//comp/trace/config/def",
6363
"//comp/trace/payload-modifier/fx",
6464
"//pkg/config/env",
@@ -71,6 +71,7 @@ dd_agent_go_test(
7171
"//pkg/util/fxutil",
7272
"//pkg/util/option",
7373
"@com_github_datadog_datadog_go_v5//statsd",
74+
"@com_github_datadog_zstd//:zstd",
7475
"@com_github_stretchr_testify//assert",
7576
"@com_github_stretchr_testify//require",
7677
"@com_github_tinylib_msgp//msgp",

0 commit comments

Comments
 (0)