[AGNTLOG-174 ] Add telemetry for log compression kind (#36612)

Co-authored-by: DeForest Richards <[email protected]>

Author: DDuongNguyen

comp/core/agenttelemetry/impl/config.go

@@ -213,6 +213,8 @@ var defaultProfiles = `
         - name: logs.decoded
         - name: logs.dropped
         - name: logs.encoded_bytes_sent
+          aggregate_tags:
+            - compression_kind
         - name: logs.sender_latency
         - name: logs.auto_multi_line_aggregator_flush
           aggregate_tags:

pkg/logs/client/http/destination.go

@@ -303,6 +303,11 @@ func (d *Destination) unconditionalSend(payload *message.Payload) (err error) {
 	}
 	metrics.BytesSent.Add(int64(payload.UnencodedSize))
 	var sourceTag string
+	compressionKind := "none"
+
+	if d.endpoint.UseCompression {
+		compressionKind = d.endpoint.CompressionKind
+	}
 
 	if strings.Contains(d.Metadata().TelemetryName(), "logs") {
 		sourceTag = "logs"
@@ -312,7 +317,7 @@ func (d *Destination) unconditionalSend(payload *message.Payload) (err error) {
 
 	metrics.TlmBytesSent.Add(float64(payload.UnencodedSize), sourceTag)
 	metrics.EncodedBytesSent.Add(int64(len(payload.Encoded)))
-	metrics.TlmEncodedBytesSent.Add(float64(len(payload.Encoded)), sourceTag)
+	metrics.TlmEncodedBytesSent.Add(float64(len(payload.Encoded)), sourceTag, compressionKind)
 
 	req, err := http.NewRequest("POST", d.url, bytes.NewReader(payload.Encoded))
 	if err != nil {

pkg/logs/client/http/destination_test.go

@@ -573,7 +573,7 @@ func TestDestinationSourceTagBasedOnTelemetryName(t *testing.T) {
 	// Create telemetry mock
 	telemetryMock := fxutil.Test[telemetry.Component](t, telemetryimpl.MockModule())
 	metrics.TlmBytesSent = telemetryMock.NewCounter("logs", "bytes_sent", []string{"source"}, "")
-	metrics.TlmEncodedBytesSent = telemetryMock.NewCounter("logs", "encoded_bytes_sent", []string{"source"}, "")
+	metrics.TlmEncodedBytesSent = telemetryMock.NewCounter("logs", "encoded_bytes_sent", []string{"source", "compression_kind"}, "")
 
 	// Create a new server
 	server := NewTestServer(200, cfg)
@@ -610,7 +610,7 @@ func TestDestinationSourceTagEPForwarder(t *testing.T) {
 	// Create telemetry mock
 	telemetryMock := fxutil.Test[telemetry.Component](t, telemetryimpl.MockModule())
 	metrics.TlmBytesSent = telemetryMock.NewCounter("logs", "bytes_sent", []string{"source"}, "")
-	metrics.TlmEncodedBytesSent = telemetryMock.NewCounter("logs", "encoded_bytes_sent", []string{"source"}, "")
+	metrics.TlmEncodedBytesSent = telemetryMock.NewCounter("logs", "encoded_bytes_sent", []string{"source", "compression_kind"}, "")
 
 	// Create a new server
 	server := NewTestServer(200, cfg)
@@ -638,3 +638,57 @@ func TestDestinationSourceTagEPForwarder(t *testing.T) {
 	assert.Len(t, metric, 1)
 	assert.Equal(t, "epforwarder", metric[0].Tags()["source"])
 }
+
+// TestDestinationCompression tests what the compression kind is set when compression is used
+func TestDestinationCompression(t *testing.T) {
+	cfg := configmock.New(t)
+
+	// Create telemetry mock
+	telemetryMock := fxutil.Test[telemetry.Component](t, telemetryimpl.MockModule())
+	metrics.TlmEncodedBytesSent = telemetryMock.NewCounter("logs", "encoded_bytes_sent", []string{"source", "compression_kind"}, "")
+
+	// Create a new server with compression enabled
+	server := NewTestServer(200, cfg)
+	defer server.httpServer.Close()
+
+	// Enable compression and set zstdcompression kind
+	server.Destination.endpoint.UseCompression = true
+	server.Destination.endpoint.CompressionKind = "zstd"
+
+	// Test case 1: Telemetry uses zstd compression
+	server.Destination.destMeta = client.NewDestinationMetadata("dbm", "1", "reliable", "0")
+	payload := &message.Payload{
+		Encoded:       []byte("payload"),
+		UnencodedSize: 7, // len("payload")
+	}
+
+	err := server.Destination.unconditionalSend(payload)
+	assert.Nil(t, err)
+	assert.Equal(t, "dbm_1_reliable_0", server.Destination.destMeta.TelemetryName())
+
+	// Verify the compression tag is set correctly
+	metric, err := telemetryMock.(telemetry.Mock).GetCountMetric("logs", "encoded_bytes_sent")
+	assert.NoError(t, err)
+	assert.Len(t, metric, 1)
+	assert.Equal(t, "zstd", metric[0].Tags()["compression_kind"])
+
+	// Test case 2: Telemetry uses gzip compression
+	// Enable compression and set gzip compression kind
+	server.Destination.endpoint.CompressionKind = "gzip"
+
+	server.Destination.destMeta = client.NewDestinationMetadata("dbm", "2", "reliable", "0")
+	payload2 := &message.Payload{
+		Encoded:       []byte("payload"),
+		UnencodedSize: 7,
+	}
+
+	err = server.Destination.unconditionalSend(payload2)
+	assert.Nil(t, err)
+	assert.Equal(t, "dbm_2_reliable_0", server.Destination.destMeta.TelemetryName())
+
+	// Verify the compression tag is set correctly
+	metric, err = telemetryMock.(telemetry.Mock).GetCountMetric("logs", "encoded_bytes_sent")
+	assert.NoError(t, err)
+	assert.Len(t, metric, 2)
+	assert.Equal(t, "gzip", metric[0].Tags()["compression_kind"])
+}

pkg/logs/client/tcp/destination.go

@@ -124,10 +124,12 @@ func (d *Destination) sendAndRetry(payload *message.Payload, output chan *messag
 
 		// TCP is only used for logs data, so we always use "logs" as the source tag
 		sourceTag := "logs"
+		// Default Compression for TCP is none
+		compressionKind := "none"
 
 		metrics.TlmBytesSent.Add(float64(payload.UnencodedSize), sourceTag)
 		metrics.EncodedBytesSent.Add(int64(len(payload.Encoded)))
-		metrics.TlmEncodedBytesSent.Add(float64(len(payload.Encoded)), sourceTag)
+		metrics.TlmEncodedBytesSent.Add(float64(len(payload.Encoded)), sourceTag, compressionKind)
 		output <- payload
 
 		if d.connManager.ShouldReset(d.connCreationTime) {

pkg/logs/metrics/metrics.go

@@ -57,7 +57,7 @@ var (
 	EncodedBytesSent = expvar.Int{}
 	// TlmEncodedBytesSent is the total number of sent bytes after encoding if any
 	TlmEncodedBytesSent = telemetry.NewCounter("logs", "encoded_bytes_sent",
-		[]string{"source"}, "Total number of sent bytes after encoding if any")
+		[]string{"source", "compression_kind"}, "Total number of sent bytes after encoding if any")
 	// BytesMissed is the number of bytes lost before they could be consumed by the agent, such as after a log rotation
 	BytesMissed = expvar.Int{}
 	// TlmBytesMissed is the number of bytes lost before they could be consumed by the agent, such as after log rotation

releasenotes/notes/log-agent-cross-org-telemetry-93a9c76063e51aec.yaml

@@ -0,0 +1,4 @@
+---
+enhancements:
+  - |
+    Adds a compression_kind tag to the ``logs.encoded_bytes_sent`` telemetry metric, enabling aggregation and monitoring of log compression type usage during rollout.