test

Author: piochelepiotr

dd-java-agent/instrumentation-testing/src/main/groovy/datadog/trace/agent/test/InstrumentationSpecification.groovy

@@ -361,6 +361,7 @@ abstract class InstrumentationSpecification extends DDSpecification implements A
     long bucketDuration = dataStreamsBucketDuration()
     TEST_DATA_STREAMS_MONITORING = new DefaultDataStreamsMonitoring(sink, features, SystemTimeSource.INSTANCE, { MOCK_DSM_TRACE_CONFIG }, TEST_DATA_STREAMS_WRITER, bucketDuration)
 
+    System.out.println("new test writer")
     TEST_WRITER = new ListWriter()
 
     if (isTestAgentEnabled()) {

dd-java-agent/instrumentation/aws-java/aws-java-sqs-2.0/src/main/java/datadog/trace/instrumentation/aws/v2/sqs/TracingIterator.java

@@ -113,13 +113,6 @@ protected void startNewMessageSpan(Message message) {
           State state = State.FACTORY.create();
           state.captureAndSetContinuation(span);
           messageStateStore.put(message, state);
-          System.out.println(
-              "[TracingIterator] Captured state for SQS message: "
-                  + message.messageId()
-                  + " with span: "
-                  + span.getSpanId()
-                  + " on thread: "
-                  + Thread.currentThread().getId());
         } catch (Exception stateException) {
           log.debug("Problem capturing state for SQS message", stateException);
         }

dd-java-agent/instrumentation/spring/spring-messaging-4.0/src/main/java/datadog/trace/instrumentation/springmessaging/SpringMessageHandlerInstrumentation.java

@@ -70,46 +70,24 @@ public static AgentScope onEnter(
       // First try to get context from continuation (preferred method)
       State state = InstrumentationContext.get(Message.class, State.class).get(message);
       if (null != state) {
-        System.out.println(
-            "[Spring] Found state in Spring message, attempting to activate continuation on thread: "
-                + Thread.currentThread().getId());
         AgentScope.Continuation continuation = state.getAndResetContinuation();
         if (null != continuation) {
           try (AgentScope scope = continuation.activate()) {
             AgentSpan span = startSpan(SPRING_INBOUND);
             DECORATE.afterStart(span);
             span.setResourceName(DECORATE.spanNameForMethod(thiz.getMethod()));
-            System.out.println(
-                "[Spring] Successfully activated continuation from Spring Message with span: "
-                    + span.getSpanId()
-                    + " on thread: "
-                    + Thread.currentThread().getId());
             return activateSpan(span);
           }
-        } else {
-          System.out.println(
-              "[Spring] No continuation found in state on thread: "
-                  + Thread.currentThread().getId());
         }
-      } else {
-        System.out.println(
-            "[Spring] No state found in Spring message 2, falling back to header extraction on thread: "
-                + Thread.currentThread().getId());
       }
 
       // Fallback to existing context or header extraction
       if (null != parent) {
         // prefer existing context, assume it was already extracted from this message
         parentContext = parent.context();
-        System.out.println(
-            "[Spring] Using existing active span context on thread: "
-                + Thread.currentThread().getId());
       } else {
         // otherwise try to re-extract the message context to avoid disconnected trace
         parentContext = extractContextAndGetSpanContext(message, GETTER);
-        System.out.println(
-            "[Spring] Extracted context from message headers on thread: "
-                + Thread.currentThread().getId());
       }
 
       AgentSpan span = startSpan(SPRING_INBOUND, parentContext);

dd-java-agent/instrumentation/spring/spring-sqs/build.gradle

@@ -7,19 +7,32 @@ muzzle {
   }
 }
 
-ext {
-  minJavaVersionForTests = JavaVersion.VERSION_17
-}
-
 apply from: "$rootDir/gradle/java.gradle"
 
 addTestSuiteForDir('latestDepTest', 'test')
 
-[compileTestGroovy, compileLatestDepTestGroovy].each {
-  it.javaLauncher = getJavaLauncherFor(17)
-}
-
 dependencies {
   compileOnly group: 'software.amazon.awssdk', name: 'sqs', version: '2.20.162'
   compileOnly group: 'org.springframework', name: 'spring-messaging', version: '5.3.23'
+
+  testImplementation project(':dd-java-agent:instrumentation:trace-annotation')
+  testImplementation project(':dd-java-agent:instrumentation:aws-java:aws-java-sqs-2.0')
+  testImplementation project(':dd-java-agent:instrumentation:spring:spring-messaging-4.0')
+
+  testImplementation group: 'org.testcontainers', name: 'localstack', version: libs.versions.testcontainers.get()
+  testImplementation group: 'org.springframework', name: 'spring-context', version: '5.3.23'
+  testImplementation group: 'org.springframework', name: 'spring-test', version: '5.3.23'
+  testImplementation group: 'org.springframework', name: 'spring-core', version: '5.3.23'
+  testImplementation group: 'software.amazon.awssdk', name: 'sqs', version: '2.20.162'
+  testImplementation group: 'software.amazon.awssdk', name: 'aws-core', version: '2.20.162'
+
+  latestDepTestImplementation group: 'org.springframework', name: 'spring-context', version: '5.+'
+  latestDepTestImplementation group: 'org.springframework', name: 'spring-test', version: '5.+'
+  latestDepTestImplementation group: 'org.springframework', name: 'spring-core', version: '5.+'
+  latestDepTestImplementation group: 'software.amazon.awssdk', name: 'sqs', version: '2.+'
+  latestDepTestImplementation group: 'software.amazon.awssdk', name: 'aws-core', version: '2.+'
+}
+
+tasks.withType(Test).configureEach {
+  usesService(testcontainersLimit)
 }

dd-java-agent/instrumentation/spring/spring-sqs/src/main/java/datadog/trace/instrumentation/springsqs/AbstractMessagingMessageConverterToMessagingInstrumentation.java

@@ -71,18 +71,7 @@ public static void onExit(
           // Transfer state from SQS Message to Spring Message
           ContextStore<Message, State> to = InstrumentationContext.get(Message.class, State.class);
           to.put(springMessage, state);
-          System.out.println(
-              "[ToMessaging] Transferred state from SQS message to Spring message on thread: "
-                  + Thread.currentThread().getId());
-        } else {
-          System.out.println(
-              "[ToMessaging] No state found in SQS message during conversion on thread: "
-                  + Thread.currentThread().getId());
         }
-      } else {
-        System.out.println(
-            "[ToMessaging] Skipping transfer - not an SQS message or null message on thread: "
-                + Thread.currentThread().getId());
       }
     }
   }

dd-java-agent/instrumentation/spring/spring-sqs/src/test/groovy/SpringSqsTest.groovy

@@ -0,0 +1,135 @@
+import datadog.trace.agent.test.InstrumentationSpecification
+import datadog.trace.agent.test.utils.PortUtils
+import org.testcontainers.containers.localstack.LocalStackContainer
+import org.testcontainers.utility.DockerImageName
+import sqs.MessagingSqsApplication
+import sqs.Receiver
+import sqs.Sender
+import sqs.SqsMessageListener
+import spock.lang.Shared
+import software.amazon.awssdk.services.sqs.SqsAsyncClient
+
+import java.util.concurrent.TimeUnit
+
+import static datadog.trace.agent.test.utils.TraceUtils.runUnderTrace
+import static org.testcontainers.containers.localstack.LocalStackContainer.Service.SQS
+
+class SpringSqsTest extends InstrumentationSpecification {
+
+  @Shared
+  LocalStackContainer localstack
+
+  @Override
+  def setupSpec() {
+    localstack = new LocalStackContainer(DockerImageName.parse("localstack/localstack:latest"))
+        .withServices(SQS)
+    localstack.start()
+    
+    def endpoint = localstack.getEndpointOverride(SQS).toString()
+    MessagingSqsApplication.endpoint = endpoint
+    
+    // Wait for LocalStack to be ready
+    PortUtils.waitForPortToOpen(
+        localstack.getHost(), 
+        localstack.getMappedPort(4566), 
+        10, 
+        TimeUnit.SECONDS)
+    
+    // Create the test queue with retry logic
+    def sqsClient = SqsAsyncClient.builder()
+        .endpointOverride(new URI(endpoint))
+        .region(software.amazon.awssdk.regions.Region.US_EAST_1)
+        .credentialsProvider(software.amazon.awssdk.auth.credentials.StaticCredentialsProvider.create(
+            software.amazon.awssdk.auth.credentials.AwsBasicCredentials.create("test", "test")))
+        .build()
+    
+    // Retry queue creation
+    def maxRetries = 5
+    def retryCount = 0
+    while (retryCount < maxRetries) {
+      try {
+        MessagingSqsApplication.createQueue(sqsClient, MessagingSqsApplication.queueName)
+        break
+      } catch (Exception e) {
+        retryCount++
+        if (retryCount >= maxRetries) {
+          throw e
+        }
+        Thread.sleep(2000) // Wait 2 seconds before retry
+      }
+    }
+  }
+
+  @Override
+  def cleanupSpec() {
+    if (null != localstack) {
+      localstack.close()
+    }
+  }
+
+  def "test basic SQS message send and receive"() {
+    setup:
+    println "=== TEST STARTING ==="
+    def application = MessagingSqsApplication.run()
+    def sender = application.getBean(Sender)
+    def receiver = application.getBean(Receiver)
+    def messageListener = application.getBean(SqsMessageListener)
+    println "=== APPLICATION STARTED ==="
+
+    when:
+    runUnderTrace("parent") {
+      println "Sending message..."
+      sender.send("hello")
+      println "Message sent"
+      
+      // Wait a bit for the message to be available
+      Thread.sleep(1000)
+      
+      // Poll for messages
+      println "Polling for messages..."
+      messageListener.receiveMessage()
+      println "Message polling completed"
+    }
+    
+    then: "message was received and processed"
+    assert receiver.latch.await(5, TimeUnit.SECONDS)
+    
+    // Debug: Check writer state before waiting
+    println "=== WRITER DEBUG: Before waitForTraces ==="
+    println "TEST_WRITER type: ${TEST_WRITER.getClass()}"
+    println "TEST_WRITER traces size: ${TEST_WRITER.traces.size()}"
+    println "TEST_WRITER traces: ${TEST_WRITER.traces}"
+    
+    // Wait for traces to be captured (with shorter timeout for debugging)
+    try {
+      TEST_WRITER.waitForTraces(1, 5, java.util.concurrent.TimeUnit.SECONDS)
+    } catch (Exception e) {
+      println "Timeout waiting for traces: ${e.message}"
+      println "Current traces: ${TEST_WRITER.traces}"
+    }
+
+    // Print all generated spans for debugging
+    println "=== Generated Spans ==="
+    if (TEST_WRITER.traces == null) {
+      println "TEST_WRITER.traces is null"
+    } else if (TEST_WRITER.traces.isEmpty()) {
+      println "TEST_WRITER.traces is empty"
+    } else {
+      TEST_WRITER.traces.eachWithIndex { trace, traceIndex ->
+        println "Trace $traceIndex:"
+        if (trace == null) {
+          println "  Trace is null"
+        } else {
+          trace.eachWithIndex { span, spanIndex ->
+            println "  Span $spanIndex: ${span.operationName} (${span.resourceName}) - Type: ${span.spanType} - Parent: ${span.parentId}"
+          }
+        }
+      }
+      println "Total traces: ${TEST_WRITER.traces.size()}"
+    }
+    println "======================="
+
+    cleanup:
+    application.close()
+  }
+}

dd-java-agent/instrumentation/spring/spring-sqs/src/test/java/sqs/MessagingSqsApplication.java

@@ -0,0 +1,54 @@
+package sqs;
+
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import software.amazon.awssdk.services.sqs.SqsAsyncClient;
+import software.amazon.awssdk.services.sqs.model.CreateQueueRequest;
+import software.amazon.awssdk.services.sqs.model.GetQueueUrlRequest;
+
+@Configuration
+public class MessagingSqsApplication {
+
+  public static String endpoint = "http://localhost:4566";
+  public static String region = "us-east-1";
+  public static String queueName = "test-queue";
+
+  @Bean
+  SqsAsyncClient sqsAsyncClient() {
+    return SqsAsyncClient.builder()
+        .endpointOverride(java.net.URI.create(endpoint))
+        .region(software.amazon.awssdk.regions.Region.of(region))
+        .credentialsProvider(software.amazon.awssdk.auth.credentials.StaticCredentialsProvider.create(
+            software.amazon.awssdk.auth.credentials.AwsBasicCredentials.create("test", "test")))
+        .build();
+  }
+
+  public static ConfigurableApplicationContext run() {
+    AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
+    context.register(MessagingSqsApplication.class);
+    context.register(Receiver.class);
+    context.register(Sender.class);
+    context.register(SqsMessageListener.class);
+    context.refresh();
+    return context;
+  }
+
+  public static String getQueueUrl(SqsAsyncClient sqsClient, String queueName) {
+    try {
+      return sqsClient.getQueueUrl(GetQueueUrlRequest.builder().queueName(queueName).build()).get()
+          .queueUrl();
+    } catch (Exception e) {
+      throw new RuntimeException("Failed to get queue URL", e);
+    }
+  }
+
+  public static void createQueue(SqsAsyncClient sqsClient, String queueName) {
+    try {
+      sqsClient.createQueue(CreateQueueRequest.builder().queueName(queueName).build()).get();
+    } catch (Exception e) {
+      throw new RuntimeException("Failed to create queue", e);
+    }
+  }
+}

dd-java-agent/instrumentation/spring/spring-sqs/src/test/java/sqs/Receiver.java

@@ -0,0 +1,18 @@
+package sqs;
+
+import datadog.trace.api.Trace;
+import datadog.trace.bootstrap.instrumentation.api.AgentTracer;
+import java.util.concurrent.CountDownLatch;
+import org.springframework.stereotype.Component;
+
+@Component
+public class Receiver {
+
+  public final CountDownLatch latch = new CountDownLatch(1);
+
+  public void receiveMessage(String message) {
+    assert null != AgentTracer.activeSpan() : "no active span during message receipt";
+    System.out.println("Processing message: " + message);
+    latch.countDown();
+  }
+}

dd-java-agent/instrumentation/spring/spring-sqs/src/test/java/sqs/Sender.java

@@ -0,0 +1,31 @@
+package sqs;
+
+import org.springframework.stereotype.Component;
+import software.amazon.awssdk.services.sqs.SqsAsyncClient;
+import software.amazon.awssdk.services.sqs.model.SendMessageRequest;
+
+@Component
+public class Sender {
+
+  private final SqsAsyncClient sqsClient;
+  private final String queueUrl;
+
+  public Sender(SqsAsyncClient sqsClient) {
+    this.sqsClient = sqsClient;
+    this.queueUrl = MessagingSqsApplication.getQueueUrl(sqsClient, MessagingSqsApplication.queueName);
+  }
+
+  public void send(String message) {
+    try {
+      sqsClient
+          .sendMessage(
+              SendMessageRequest.builder()
+                  .queueUrl(queueUrl)
+                  .messageBody(message)
+                  .build())
+          .get();
+    } catch (Exception e) {
+      throw new RuntimeException("Failed to send message", e);
+    }
+  }
+}