Merge branch 'master' into sarahchen6/update-dd-octo-sts

Author: sarahchen6

.github/workflows/update-docker-build-image.yaml

@@ -50,9 +50,10 @@ jobs:
           echo "::notice::Using Docker build image tag: ${TAG}"
       - name: Update the Docker build image in GitLab CI config
         run: |
-          sed -i '' -E 's|(BUILDER_IMAGE_VERSION_PREFIX:)[^#]*([#].*)|\1 "${{ steps.define-tag.outputs.tag }}-" \2|' .gitlab-ci.yml
-      - name: Commit changes
-        id: create-commit
+          sed -i -E 's|(BUILDER_IMAGE_VERSION_PREFIX:)[^#]*([#].*)|\1 "${{ steps.define-tag.outputs.tag }}-" \2|' .gitlab-ci.yml
+      - name: Commit and push changes
+        env:
+          GITHUB_TOKEN: ${{ steps.octo-sts.outputs.token }}
         run: |
           git config user.name "github-actions[bot]"
           git config user.email "41898282+github-actions[bot]@users.noreply.github.com"

dd-java-agent/testing/src/main/groovy/datadog/trace/agent/test/utils/PortUtils.java

@@ -105,7 +105,8 @@ private static boolean isPortOpen(String host, int port) {
 
   public static void waitForPortToOpen(
       final int port, final long timeout, final TimeUnit unit, final Process process) {
-    final long waitUntil = System.currentTimeMillis() + unit.toMillis(timeout);
+    final long startedAt = System.currentTimeMillis();
+    final long waitUntil = startedAt + unit.toMillis(timeout);
 
     while (System.currentTimeMillis() < waitUntil) {
       try {
@@ -133,7 +134,13 @@ public static void waitForPortToOpen(
       }
     }
 
-    throw new RuntimeException("Timed out waiting for port " + port + " to be opened");
+    throw new RuntimeException(
+        "Timed out waiting for port "
+            + port
+            + " to be opened, started to wait at: "
+            + startedAt
+            + ", timed out at: "
+            + System.currentTimeMillis());
   }
 
   public static void waitForPortToOpen(String host, int port, long timeout, TimeUnit unit) {

dd-java-agent/testing/src/test/groovy/utils/PortUtilsTest.groovy

@@ -0,0 +1,70 @@
+package utils
+
+import datadog.trace.agent.test.AgentTestRunner
+import datadog.trace.agent.test.utils.PortUtils
+
+import java.util.concurrent.TimeUnit
+
+class PortUtilsTest extends AgentTestRunner {
+  def "expect waitForPortToOpen succeed"() {
+    given:
+    int port = PortUtils.randomOpenPort()
+    def socket = new ServerSocket(port) // Emulating port opened.
+
+    def process = Mock(Process)
+    process.isAlive() >> true
+
+    when:
+    PortUtils.waitForPortToOpen(port, 1, TimeUnit.SECONDS, process)
+
+    then:
+    noExceptionThrown()
+
+    cleanup:
+    socket.close()
+  }
+
+  def "expect to handle port open timeout"() {
+    given:
+    int port = PortUtils.randomOpenPort()
+    def process = Mock(Process)
+    process.isAlive() >> true
+
+    when:
+    PortUtils.waitForPortToOpen(port, 1, TimeUnit.SECONDS, process)
+
+    then:
+    def ex = thrown(RuntimeException)
+    ex.message.startsWith("Timed out waiting for port $port to be opened, started to wait at:")
+  }
+
+  def "expect to handle process abnormal termination"() {
+    given:
+    int port = PortUtils.randomOpenPort()
+    def process = Mock(Process)
+    process.isAlive() >> false
+    process.exitValue() >> 123
+
+    when:
+    PortUtils.waitForPortToOpen(port, 1, TimeUnit.SECONDS, process)
+
+    then:
+    def ex = thrown(RuntimeException)
+    ex.message == "Process exited abnormally exitCode=123 before port=$port was opened"
+  }
+
+  def "expect to handle process termination before port opened"() {
+    given:
+    int port = PortUtils.randomOpenPort()
+    def process = Mock(Process)
+    process.isAlive() >> false
+    process.exitValue() >> 0
+
+    when:
+    PortUtils.waitForPortToOpen(port, 1, TimeUnit.SECONDS, process)
+
+    then:
+    def ex = thrown(RuntimeException)
+    ex.message == "Process finished before port=$port was opened"
+  }
+}

dd-trace-core/src/main/java/datadog/trace/core/baggage/BaggagePropagator.java

@@ -7,6 +7,7 @@
 import datadog.context.propagation.CarrierVisitor;
 import datadog.context.propagation.Propagator;
 import datadog.trace.api.Config;
+import datadog.trace.api.metrics.BaggageMetrics;
 import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
 import datadog.trace.bootstrap.instrumentation.api.AgentSpanContext;
 import datadog.trace.bootstrap.instrumentation.api.Baggage;
@@ -26,6 +27,7 @@
 public class BaggagePropagator implements Propagator {
   private static final Logger LOG = LoggerFactory.getLogger(BaggagePropagator.class);
   private static final PercentEscaper UTF_ESCAPER = PercentEscaper.create();
+  private static final BaggageMetrics BAGGAGE_METRICS = BaggageMetrics.getInstance();
   static final String BAGGAGE_KEY = "baggage";
   private final boolean injectBaggage;
   private final boolean extractBaggage;
@@ -89,11 +91,13 @@ public <C> void inject(Context context, C carrier, CarrierSetter<C> setter) {
       processedItems++;
       // reached the max number of baggage items allowed
       if (processedItems == this.maxItems) {
+        BAGGAGE_METRICS.onBaggageTruncatedByItemLimit();
         break;
       }
       // Drop newest k/v pair if adding it leads to exceeding the limit
       if (currentBytes + escapedKey.size + escapedVal.size + extraBytes > this.maxBytes) {
         baggageText.setLength(currentBytes);
+        BAGGAGE_METRICS.onBaggageTruncatedByByteLimit();
         break;
       }
       currentBytes += escapedKey.size + escapedVal.size + extraBytes;
@@ -103,6 +107,9 @@ public <C> void inject(Context context, C carrier, CarrierSetter<C> setter) {
     // Save header as cache to re-inject it later if baggage did not change
     baggage.setW3cHeader(headerValue);
     setter.set(carrier, BAGGAGE_KEY, headerValue);
+
+    // Record successful baggage injection for telemetry
+    BAGGAGE_METRICS.onBaggageInjected();
   }
 
   @Override
@@ -117,6 +124,9 @@ public <C> Context extract(Context context, C carrier, CarrierVisitor<C> visitor
       return context;
     }
 
+    // Record successful baggage extraction for telemetry
+    BAGGAGE_METRICS.onBaggageExtracted();
+
     // TODO: consider a better way to link baggage with the extracted (legacy) TagContext
     AgentSpan extractedSpan = AgentSpan.fromContext(context);
     if (extractedSpan != null) {
@@ -158,12 +168,14 @@ private Map<String, String> parseBaggageHeaders(String input) {
         if (kvSeparatorInd > end) {
           LOG.debug(
               "Dropping baggage headers due to key with no value {}", input.substring(start, end));
+          BAGGAGE_METRICS.onBaggageMalformed();
           return emptyMap();
         }
         String key = decode(input.substring(start, kvSeparatorInd).trim());
         String value = decode(input.substring(kvSeparatorInd + 1, end).trim());
         if (key.isEmpty() || value.isEmpty()) {
           LOG.debug("Dropping baggage headers due to empty k/v {}:{}", key, value);
+          BAGGAGE_METRICS.onBaggageMalformed();
           return emptyMap();
         }
         baggage.put(key, value);

dd-trace-core/src/test/groovy/datadog/trace/core/baggage/BaggagePropagatorTelemetryTest.groovy

@@ -0,0 +1,120 @@
+package datadog.trace.core.baggage
+
+import datadog.context.Context
+import datadog.trace.api.Config
+import datadog.trace.api.metrics.BaggageMetrics
+import datadog.trace.api.telemetry.CoreMetricCollector
+import spock.lang.Specification
+
+class BaggagePropagatorTelemetryTest extends Specification {
+
+  def "should directly increment baggage metrics"() {
+    given:
+    def baggageMetrics = BaggageMetrics.getInstance()
+    def collector = CoreMetricCollector.getInstance()
+
+    when:
+    baggageMetrics.onBaggageInjected()
+    collector.prepareMetrics()
+    def metrics = collector.drain()
+
+    then:
+    def baggageMetric = metrics.find { it.metricName == "context_header_style.injected" }
+    baggageMetric != null
+    baggageMetric.value >= 1
+    baggageMetric.tags.contains("header_style:baggage")
+  }
+
+  def "should increment telemetry counter when baggage is successfully extracted"() {
+    given:
+    def config = Mock(Config) {
+      isBaggageExtract() >> true
+      isBaggageInject() >> true
+      getBaggageMaxItems() >> 64
+      getBaggageMaxBytes() >> 8192
+    }
+    def propagator = new BaggagePropagator(config)
+    def context = Context.root()
+    def carrier = ["baggage": "key1=value1,key2=value2"]
+    def visitor = { map, consumer ->
+      map.each { k, v -> consumer.accept(k, v) }
+    }
+    def collector = CoreMetricCollector.getInstance()
+
+    when:
+    propagator.extract(context, carrier, visitor)
+    collector.prepareMetrics()
+    def metrics = collector.drain()
+
+    then:
+    def baggageMetric = metrics.find { it.metricName == "context_header_style.extracted" }
+    baggageMetric != null
+    baggageMetric.value >= 1
+    baggageMetric.tags.contains("header_style:baggage")
+  }
+
+  def "should directly increment all baggage metrics"() {
+    given:
+    def baggageMetrics = BaggageMetrics.getInstance()
+    def collector = CoreMetricCollector.getInstance()
+
+    when:
+    baggageMetrics.onBaggageInjected()
+    baggageMetrics.onBaggageMalformed()
+    baggageMetrics.onBaggageTruncatedByByteLimit()
+    baggageMetrics.onBaggageTruncatedByItemLimit()
+    collector.prepareMetrics()
+    def metrics = collector.drain()
+
+    then:
+    def injectedMetric = metrics.find { it.metricName == "context_header_style.injected" }
+    injectedMetric != null
+    injectedMetric.value == 1
+    injectedMetric.tags.contains("header_style:baggage")
+
+    def malformedMetric = metrics.find { it.metricName == "context_header_style.malformed" }
+    malformedMetric != null
+    malformedMetric.value == 1
+    malformedMetric.tags.contains("header_style:baggage")
+
+    def bytesTruncatedMetric = metrics.find {
+      it.metricName == "context_header_style.truncated" &&
+        it.tags.contains("truncation_reason:baggage_byte_count_exceeded")
+    }
+    bytesTruncatedMetric != null
+    bytesTruncatedMetric.value == 1
+
+    def itemsTruncatedMetric = metrics.find {
+      it.metricName == "context_header_style.truncated" &&
+        it.tags.contains("truncation_reason:baggage_item_count_exceeded")
+    }
+    itemsTruncatedMetric != null
+    itemsTruncatedMetric.value == 1
+  }
+
+  def "should not increment telemetry counter when baggage extraction fails"() {
+    given:
+    def config = Mock(Config) {
+      isBaggageExtract() >> true
+      isBaggageInject() >> true
+      getBaggageMaxItems() >> 64
+      getBaggageMaxBytes() >> 8192
+    }
+    def propagator = new BaggagePropagator(config)
+    def context = Context.root()
+    def carrier = [:] // No baggage header
+    def visitor = { map, consumer ->
+      map.each { k, v -> consumer.accept(k, v) }
+    }
+    def collector = CoreMetricCollector.getInstance()
+
+    when:
+    propagator.extract(context, carrier, visitor)
+    collector.prepareMetrics()
+    def metrics = collector.drain()
+
+    then:
+    def foundMetrics = metrics.findAll { it.metricName.startsWith("context_header_style.") }
+    foundMetrics.isEmpty() // No extraction occurred, so no metrics should be created
+  }
+}