Backpropagate peer tags

communication/src/main/java/datadog/communication/ddagent/DDAgentFeaturesDiscovery.java

@@ -1,7 +1,9 @@
 package datadog.communication.ddagent;
 
 import static datadog.communication.serialization.msgpack.MsgPackWriter.FIXARRAY;
+import static java.util.Collections.emptyList;
 import static java.util.Collections.singletonList;
+import static java.util.Collections.unmodifiableList;
 
 import com.squareup.moshi.JsonAdapter;
 import com.squareup.moshi.Moshi;
@@ -86,6 +88,8 @@ public class DDAgentFeaturesDiscovery implements DroppingPolicy {
   private volatile String evpProxyEndpoint;
   private volatile String version;
   private volatile String telemetryProxyEndpoint;
+  private volatile List<String> peerTags = emptyList();
+  private volatile List<String> spanKindsToComputedStats = emptyList();
 
   private long lastTimeDiscovered;
 
@@ -119,6 +123,8 @@ private void reset() {
     version = null;
     lastTimeDiscovered = 0;
     telemetryProxyEndpoint = null;
+    peerTags = emptyList();
+    spanKindsToComputedStats = emptyList();
   }
 
   /** Run feature discovery, unconditionally. */
@@ -287,6 +293,13 @@ private boolean processInfoResponse(String response) {
             null != canDrop
                 && ("true".equalsIgnoreCase(String.valueOf(canDrop))
                     || Boolean.TRUE.equals(canDrop));
+
+        Object peer_tags = map.get("peer_tags");
+        peerTags = peer_tags == null ? emptyList() : unmodifiableList((List<String>) peer_tags);
+
+        Object span_kinds = map.get("span_kinds_stats_computed");
+        spanKindsToComputedStats =
+            span_kinds == null ? emptyList() : unmodifiableList((List<String>) span_kinds);
       }
       try {
         state = Strings.sha256(response);
@@ -344,6 +357,14 @@ public boolean supportsLongRunning() {
     return supportsLongRunning;
   }
 
+  public List<String> peerTags() {
+    return peerTags;
+  }
+
+  public List<String> spanKindsToComputedStats() {
+    return spanKindsToComputedStats;
+  }
+
   public String getMetricsEndpoint() {
     return metricsEndpoint;
   }

communication/src/test/groovy/datadog/communication/ddagent/DDAgentFeaturesDiscoveryTest.groovy

@@ -30,6 +30,8 @@ class DDAgentFeaturesDiscoveryTest extends DDSpecification {
 
   static final String INFO_RESPONSE = loadJsonFile("agent-info.json")
   static final String INFO_STATE = Strings.sha256(INFO_RESPONSE)
+  static final String INFO_WITH_PEER_TAG_BACK_PROPAGATION_RESPONSE = loadJsonFile("agent-info-with-peer-tag-back-propagation.json")
+  static final String INFO_WITH_PEER_TAG_BACK_PROPAGATION_STATE = Strings.sha256(INFO_WITH_PEER_TAG_BACK_PROPAGATION_RESPONSE)
   static final String INFO_WITH_CLIENT_DROPPING_RESPONSE = loadJsonFile("agent-info-with-client-dropping.json")
   static final String INFO_WITH_CLIENT_DROPPING_STATE = Strings.sha256(INFO_WITH_CLIENT_DROPPING_RESPONSE)
   static final String INFO_WITHOUT_METRICS_RESPONSE = loadJsonFile("agent-info-without-metrics.json")
@@ -424,6 +426,46 @@ class DDAgentFeaturesDiscoveryTest extends DDSpecification {
     !features.supportsContentEncodingHeadersWithEvpProxy()
   }
 
+  def "test parse /info response with peer tag back propagation"() {
+    setup:
+    OkHttpClient client = Mock(OkHttpClient)
+    DDAgentFeaturesDiscovery features = new DDAgentFeaturesDiscovery(client, monitoring, agentUrl, true, true)
+
+    when: "/info available"
+    features.discover()
+
+    then:
+    1 * client.newCall(_) >> { Request request -> infoResponse(request, INFO_RESPONSE) }
+
+    when: "/info available with peer tag back propagation"
+    features.discover()
+
+    then:
+    1 * client.newCall(_) >> { Request request -> infoResponse(request, INFO_WITH_PEER_TAG_BACK_PROPAGATION_RESPONSE) }
+    features.state() == INFO_WITH_PEER_TAG_BACK_PROPAGATION_STATE
+    features.supportsDropping()
+    features.peerTags().containsAll(
+      "_dd.base_service",
+      "active_record.db.vendor",
+      "amqp.destination",
+      "amqp.exchange",
+      "amqp.queue",
+      "grpc.host",
+      "hostname",
+      "http.host",
+      "http.server_name",
+      "streamname",
+      "tablename",
+      "topicname"
+      )
+    features.spanKindsToComputedStats().containsAll(
+      "client",
+      "consumer",
+      "producer",
+      "server"
+      )
+  }
+
   def infoResponse(Request request, String json) {
     return Mock(Call) {
       it.execute() >> new Response.Builder()

communication/src/test/resources/agent-features/agent-info-with-peer-tag-back-propagation.json

@@ -0,0 +1,80 @@
+{
+  "version": "7.67.0",
+  "git_commit": "bdf863ccc9",
+  "endpoints": [
+    "/v0.3/traces",
+    "/v0.3/services",
+    "/v0.4/traces",
+    "/v0.4/services",
+    "/v0.5/traces",
+    "/v0.7/traces",
+    "/profiling/v1/input",
+    "/telemetry/proxy/",
+    "/v0.6/stats",
+    "/v0.1/pipeline_stats",
+    "/openlineage/api/v1/lineage",
+    "/evp_proxy/v1/",
+    "/evp_proxy/v2/",
+    "/evp_proxy/v3/",
+    "/evp_proxy/v4/",
+    "/debugger/v1/input",
+    "/debugger/v1/diagnostics",
+    "/symdb/v1/input",
+    "/dogstatsd/v1/proxy",
+    "/dogstatsd/v2/proxy",
+    "/tracer_flare/v1",
+    "/config/set"
+  ],
+  "client_drop_p0s": true,
+  "span_meta_structs": true,
+  "long_running_spans": true,
+  "span_events": true,
+  "config": {
+    "default_env": "prod",
+    "target_tps": 10,
+    "max_eps": 200,
+    "receiver_port": 8127,
+    "receiver_socket": "/var/run/datadog/apm.socket",
+    "connection_limit": 12,
+    "receiver_timeout": 100,
+    "max_request_bytes": 26214400,
+    "statsd_port": 8125,
+    "max_memory": 0,
+    "max_cpu": 0,
+    "analyzed_spans_by_service": {},
+    "obfuscation": {
+      "elastic_search": true,
+      "mongo": true,
+      "sql_exec_plan": true,
+      "sql_exec_plan_normalize": true,
+      "http": {
+        "remove_query_string": true,
+        "remove_path_digits": true
+      },
+      "remove_stack_traces": false,
+      "redis": true,
+      "memcached": false
+    }
+  },
+  "peer_tags": [
+    "_dd.base_service",
+    "active_record.db.vendor",
+    "amqp.destination",
+    "amqp.exchange",
+    "amqp.queue",
+    "grpc.host",
+    "hostname",
+    "http.host",
+    "http.server_name",
+    "streamname",
+    "tablename",
+    "topicname"
+  ],
+  "span_kinds_stats_computed": [
+    "server",
+    "consumer",
+    "client",
+    "producer"
+  ],
+  "obfuscation_version": 1
+}