Improve data streams performance (#7749)

* update queue used

* remove deprecated features

* only tag spans with schema definition sampled

* fmt

* add fields to reflect config

Author: piochelepiotr

dd-java-agent/agent-bootstrap/src/main/resources/META-INF/native-image/com.datadoghq/dd-java-agent/reflect-config.json

@@ -133,5 +133,23 @@
     "fields": [
       {"name": "consumerIndex", "allowUnsafeAccess": true}
     ]
+  },
+  {
+    "name" : "datadog.jctools.queues.MpscArrayQueueProducerIndexField",
+    "fields": [
+      {"name": "producerIndex", "allowUnsafeAccess": true}
+    ]
+  },
+  {
+    "name" : "datadog.jctools.queues.MpscArrayQueueProducerLimitField",
+    "fields": [
+      {"name": "producerLimit", "allowUnsafeAccess": true}
+    ]
+  },
+  {
+    "name" : "datadog.jctools.queues.MpscArrayQueueConsumerIndexField",
+    "fields": [
+      {"name": "consumerIndex", "allowUnsafeAccess": true}
+    ]
   }
 ]

dd-java-agent/instrumentation/avro/src/main/java/datadog/trace/instrumentation/avro/SchemaExtractor.java

@@ -143,9 +143,6 @@ public static void attachSchemaOnSpan(
       return;
     }
     AgentDataStreamsMonitoring dsm = AgentTracer.get().getDataStreamsMonitoring();
-    span.setTag(DDTags.SCHEMA_TYPE, AVRO);
-    span.setTag(DDTags.SCHEMA_NAME, schema.getFullName());
-    span.setTag(DDTags.SCHEMA_OPERATION, operation);
 
     if (!dsm.canSampleSchema(operation)) {
       return;
@@ -162,6 +159,9 @@ public static void attachSchemaOnSpan(
     }
 
     Schema schemaData = SchemaExtractor.extractSchemas(schema);
+    span.setTag(DDTags.SCHEMA_TYPE, AVRO);
+    span.setTag(DDTags.SCHEMA_NAME, schema.getFullName());
+    span.setTag(DDTags.SCHEMA_OPERATION, operation);
     span.setTag(DDTags.SCHEMA_DEFINITION, schemaData.definition);
     span.setTag(DDTags.SCHEMA_WEIGHT, weight);
     span.setTag(DDTags.SCHEMA_ID, schemaData.id);

dd-java-agent/instrumentation/protobuf/src/main/java/datadog/trace/instrumentation/protobuf_java/SchemaExtractor.java

@@ -183,9 +183,6 @@ public static void attachSchemaOnSpan(
       return;
     }
     AgentDataStreamsMonitoring dsm = AgentTracer.get().getDataStreamsMonitoring();
-    span.setTag(DDTags.SCHEMA_TYPE, PROTOBUF);
-    span.setTag(DDTags.SCHEMA_NAME, descriptor.getFullName());
-    span.setTag(DDTags.SCHEMA_OPERATION, operation);
     // do a check against the schema sampler to avoid forcing the trace sampling decision too often.
     if (!dsm.canSampleSchema(operation)) {
       return;
@@ -200,6 +197,9 @@ public static void attachSchemaOnSpan(
       return;
     }
     Schema schema = SchemaExtractor.extractSchemas(descriptor);
+    span.setTag(DDTags.SCHEMA_TYPE, PROTOBUF);
+    span.setTag(DDTags.SCHEMA_NAME, descriptor.getFullName());
+    span.setTag(DDTags.SCHEMA_OPERATION, operation);
     span.setTag(DDTags.SCHEMA_DEFINITION, schema.definition);
     span.setTag(DDTags.SCHEMA_WEIGHT, weight);
     span.setTag(DDTags.SCHEMA_ID, schema.id);

dd-trace-core/src/main/java/datadog/trace/core/datastreams/DefaultDataStreamsMonitoring.java

@@ -1,7 +1,6 @@
 package datadog.trace.core.datastreams;
 
 import static datadog.communication.ddagent.DDAgentFeaturesDiscovery.V01_DATASTREAMS_ENDPOINT;
-import static datadog.trace.api.DDTags.PATHWAY_HASH;
 import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activeSpan;
 import static datadog.trace.core.datastreams.TagsProcessor.DIRECTION_IN;
 import static datadog.trace.core.datastreams.TagsProcessor.DIRECTION_OUT;
@@ -42,11 +41,10 @@
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
-import org.jctools.queues.MpscBlockingConsumerArrayQueue;
+import org.jctools.queues.MpscArrayQueue;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -61,7 +59,7 @@ public class DefaultDataStreamsMonitoring implements DataStreamsMonitoring, Even
       new StatsPoint(Collections.emptyList(), 0, 0, 0, 0, 0, 0, 0);
 
   private final Map<Long, StatsBucket> timeToBucket = new HashMap<>();
-  private final BlockingQueue<InboxItem> inbox = new MpscBlockingConsumerArrayQueue<>(1024);
+  private final MpscArrayQueue<InboxItem> inbox = new MpscArrayQueue<>(1024);
   private final DatastreamsPayloadWriter payloadWriter;
   private final DDAgentFeaturesDiscovery features;
   private final TimeSource timeSource;
@@ -240,9 +238,6 @@ public void setCheckpoint(
     PathwayContext pathwayContext = span.context().getPathwayContext();
     if (pathwayContext != null) {
       pathwayContext.setCheckpoint(sortedTags, this::add, defaultTimestamp, payloadSizeBytes);
-      if (pathwayContext.getHash() != 0) {
-        span.setTag(PATHWAY_HASH, Long.toUnsignedString(pathwayContext.getHash()));
-      }
     }
   }
 
@@ -318,7 +313,11 @@ public void run() {
       Thread currentThread = Thread.currentThread();
       while (!currentThread.isInterrupted()) {
         try {
-          InboxItem payload = inbox.take();
+          InboxItem payload = inbox.poll();
+          if (payload == null) {
+            Thread.sleep(10);
+            continue;
+          }
 
           if (payload == REPORT) {
             checkDynamicConfig();
@@ -350,8 +349,6 @@ public void run() {
               statsBucket.addBacklog(backlog);
             }
           }
-        } catch (InterruptedException e) {
-          currentThread.interrupt();
         } catch (Exception e) {
           log.debug("Error monitoring data streams", e);
         }

dd-trace-core/src/main/java/datadog/trace/core/datastreams/DefaultPathwayContext.java

@@ -442,15 +442,9 @@ private long generateNodeHash(PathwayHashBuilder pathwayHashBuilder) {
   }
 
   private long generatePathwayHash(long nodeHash, long parentHash) {
-    lock.lock();
-    try {
-      outputBuffer.clear();
-      outputBuffer.writeLongLE(nodeHash);
-      outputBuffer.writeLongLE(parentHash);
-
-      return FNV64Hash.generateHash(outputBuffer.backingArray(), 0, 16, FNV64Hash.Version.v1);
-    } finally {
-      lock.unlock();
-    }
+    outputBuffer.clear();
+    outputBuffer.writeLongLE(nodeHash);
+    outputBuffer.writeLongLE(parentHash);
+    return FNV64Hash.generateHash(outputBuffer.backingArray(), 0, 16, FNV64Hash.Version.v1);
   }
 }