chore(datadog_agent source, datadog_metrics sink): Add end to end tests with the Datadog Agent for metrics.

.github/workflows/changes.yml

@@ -119,6 +119,8 @@ on:
         value: ${{ jobs.e2e_tests.outputs.all-e2e }}
       e2e-datadog-logs:
         value: ${{ jobs.e2e_tests.outputs.datadog-logs }}
+      e2e-datadog-metrics:
+        value: ${{ jobs.e2e_tests.outputs.datadog-metrics }}
 
 jobs:
   # Detects changes that are not specific to integration tests
@@ -250,7 +252,10 @@ jobs:
     if: ${{ inputs.e2e_tests }}
     outputs:
       all-e2e: ${{ steps.filter.outputs.all-int}}
+      # TODO: when https://github.com/vectordotdev/vector/issues/18829 is undertaken,
+      # part of that will involve updating this job to also run `cargo vdev e2e ci-paths`
       datadog-logs: ${{ steps.filter.outputs.e2e-datadog-logs }}
+      datadog-metrics: ${{ steps.filter.outputs.e2e-datadog-metrics }}
     steps:
       - uses: actions/checkout@v3
 

.github/workflows/e2e.yml

@@ -83,6 +83,16 @@ jobs:
           max_attempts: 3
           command: bash scripts/ci-integration-test.sh e2e-datadog-logs
 
+      - if: (github.event_name == 'schedule' || needs.changes.outputs.all-e2e == 'true' || needs.changes.outputs.e2e-datadog-metrics == 'true') &&
+          (github.event_name != 'pull_request' || env.PR_HAS_ACCESS_TO_SECRETS == 'true')
+        name: e2e-datadog-metrics
+        uses: nick-fields/retry@v2
+        with:
+          timeout_minutes: 35
+          max_attempts: 3
+          command: bash scripts/ci-integration-test.sh e2e-datadog-metrics
+
+
   e2e-test-suite:
     name: E2E Test Suite
     runs-on: ubuntu-latest

.github/workflows/integration-comment.yml

@@ -467,6 +467,16 @@ jobs:
           max_attempts: 3
           command: bash scripts/ci-integration-test.sh e2e-datadog-logs
 
+      - name: datadog-e2e-metrics
+        if: ${{ contains(github.event.comment.body, '/ci-run-integration-datadog-e2e-metrics')
+          || contains(github.event.comment.body, '/ci-run-integration-all')
+          || contains(github.event.comment.body, '/ci-run-all') }}
+        uses: nick-fields/retry@v2
+        with:
+          timeout_minutes: 35
+          max_attempts: 3
+          command: bash scripts/ci-integration-test.sh datadog-e2e-metrics
+
   update-pr-status:
     name: Signal result to PR
     runs-on: ubuntu-latest

Cargo.toml

@@ -859,7 +859,8 @@ all-e2e-tests = [
 
 e2e-tests-datadog = [
   "sources-datadog_agent",
-  "sinks-datadog_logs"
+  "sinks-datadog_logs",
+  "sinks-datadog_metrics"
 ]
 
 vector-api-tests = [

scripts/e2e/e2e-datadog-logs/compose.yaml

@@ -4,6 +4,9 @@ services:
   # Generates random log data for consumption by the custom Agent check
   log_generator:
     image: docker.io/mingrammer/flog
+    depends_on:
+      - datadog-agent-vector
+      - datadog-agent
     command:
       - "-f"
       - "json"
@@ -16,35 +19,11 @@ services:
     volumes:
       - log_path:/var/log/
 
-  # Receives log data from the `datadog-agent-vector` service and sends
-  # to the `fakeintake-vector` service.
-  vector:
-    depends_on:
-      - log_generator
-      - fakeintake-vector
-    build:
-      context: ${PWD}
-    # re-using the integration test runner image since it already has
-    # compiled vector on it.
-    image: ${CONFIG_VECTOR_IMAGE}
-    environment:
-      - FEATURES=e2e-tests-datadog
-    working_dir: /home/vector
-    network_mode: host
-    command:
-      - "/usr/bin/vector"
-      - "-vvv"
-      - "-c"
-      - "/home/vector/tests/data/e2e/datadog/logs/vector.toml"
-    volumes:
-      - ${PWD}:/home/vector
-
   # Tails a custom log created by `log_generator` and sends log data to
   # the `fakeintake-agent` service
   datadog-agent:
     image: docker.io/datadog/agent:${CONFIG_AGENT_VERSION}
     depends_on:
-      - log_generator
       - fakeintake-agent
     environment:
     - DD_API_KEY=${TEST_DATADOG_API_KEY:?TEST_DATADOG_API_KEY required}
@@ -65,7 +44,6 @@ services:
   datadog-agent-vector:
     image: docker.io/datadog/agent:${CONFIG_AGENT_VERSION}
     depends_on:
-      - log_generator
       - vector
     environment:
     - DD_API_KEY=${TEST_DATADOG_API_KEY:?TEST_DATADOG_API_KEY required}
@@ -81,6 +59,28 @@ services:
     # The custom log to tail, created by the `log_generator` service
     - log_path:/var/log/
 
+  # Receives log data from the `datadog-agent-vector` service and sends
+  # to the `fakeintake-vector` service.
+  vector:
+    depends_on:
+      - fakeintake-vector
+    build:
+      context: ${PWD}
+    # re-using the integration test runner image since it already has
+    # compiled vector on it.
+    image: ${CONFIG_VECTOR_IMAGE}
+    environment:
+      - FEATURES=e2e-tests-datadog
+    working_dir: /home/vector
+    network_mode: host
+    command:
+      - "/usr/bin/vector"
+      - "-vvv"
+      - "-c"
+      - "/home/vector/tests/data/e2e/datadog/logs/vector.toml"
+    volumes:
+      - ${PWD}:/home/vector
+
   # Receives log data from the `datadog-agent` service. Is queried by the test runner
   # which does the validation of consistency with the other fakeintake service.
   fakeintake-agent:

scripts/e2e/e2e-datadog-metrics/compose.yaml

@@ -0,0 +1,82 @@
+version: '3'
+
+services:
+
+  # Emits metrics to the Agent only path
+  dogstatsd-client-agent:
+    build: ./dogstatsd_client
+    environment:
+    - STATSD_HOST=datadog-agent
+    depends_on:
+      - datadog-agent
+
+  # Emits metrics to the Agent-Vector path
+  dogstatsd-client-vector:
+    build: ./dogstatsd_client
+    environment:
+    - STATSD_HOST=datadog-agent-vector
+    depends_on:
+      - datadog-agent-vector
+
+  # Sends metric data received from the Emitter to the `fakeintake-agent` service
+  datadog-agent:
+    image: docker.io/datadog/agent:${CONFIG_AGENT_VERSION}
+    depends_on:
+      - fakeintake-agent
+    environment:
+    - DD_API_KEY=${TEST_DATADOG_API_KEY:?TEST_DATADOG_API_KEY required}
+    - DD_HOSTNAME=datadog-agent
+    volumes:
+    # The Agent config file
+    - ${PWD}/tests/data/e2e/datadog/metrics/agent_only.yaml:/etc/datadog-agent/datadog.yaml
+
+  # Sends metric data received from the Emitter to the `vector` service
+  datadog-agent-vector:
+    image: docker.io/datadog/agent:${CONFIG_AGENT_VERSION}
+    depends_on:
+      - vector
+    environment:
+    - DD_API_KEY=${TEST_DATADOG_API_KEY:?TEST_DATADOG_API_KEY required}
+    - DD_HOSTNAME=datadog-agent-vector
+    volumes:
+    # The Agent config file
+    - ${PWD}/tests/data/e2e/datadog/metrics/agent_vector.yaml:/etc/datadog-agent/datadog.yaml
+
+  # Receives metric data from the `datadog-agent-vector` service and sends
+  # to the `fakeintake-vector` service.
+  vector:
+    depends_on:
+      - fakeintake-vector
+    build:
+      context: ${PWD}
+    # re-using the integration test runner image since it already has
+    # compiled vector on it.
+    image: ${CONFIG_VECTOR_IMAGE}
+    environment:
+      - FEATURES=e2e-tests-datadog
+    working_dir: /home/vector
+    network_mode: host
+    command:
+      - "/usr/bin/vector"
+      - "-vvv"
+      - "-c"
+      - "/home/vector/tests/data/e2e/datadog/metrics/vector.toml"
+    volumes:
+      - ${PWD}:/home/vector
+
+  # Receives metric data from the `datadog-agent` service. Is queried by the test runner
+  # which does the validation of consistency with the other fakeintake service.
+  fakeintake-agent:
+    image: docker.io/datadog/fakeintake:latest
+
+  # Receives metric data from the `datadog-agent-vector` service. Is queried by the test runner
+  # which does the validation of consistency with the other fakeintake service.
+  fakeintake-vector:
+    image: docker.io/datadog/fakeintake:latest
+
+networks:
+  default:
+    name: ${VECTOR_NETWORK}
+
+volumes:
+  target: {}

scripts/e2e/e2e-datadog-metrics/dogstatsd_client/Dockerfile

@@ -0,0 +1,8 @@
+FROM python:3.7-alpine
+
+COPY . /app
+WORKDIR /app
+
+RUN pip install -r requirements.txt
+
+CMD [ "python3", "./client.py"]

scripts/e2e/e2e-datadog-metrics/dogstatsd_client/client.py

@@ -0,0 +1,56 @@
+from datadog import initialize, statsd
+import time
+import os
+import random
+
+STATSD_HOST = os.getenv('STATSD_HOST')
+
+print(f"initializing for {STATSD_HOST}")
+
+options = {
+    'statsd_host':STATSD_HOST,
+    'statsd_port':8125
+}
+
+initialize(**options)
+
+# Give the Agent time to actually spin up.
+# The container may return "ready" but the
+# Agent process is still booting.
+time.sleep(10)
+
+hist_data = [
+    9, 5, 0, 2, 16, 17, 8, 16, 10, 13,
+    15, 3, 9, 13, 11, 17, 5, 18, 14, 9,
+    4, 16, 9, 17, 4, 11, 7, 14, 8, 12,
+    10, 9, 11, 3, 18, 12, 17, 12, 3, 19,
+    9, 11, 19, 9, 15, 2, 7, 10, 4, 14
+]
+
+dist_data = [
+    18, 5, 19, 0, 13, 12, 5, 12, 10, 4,
+    1, 5, 7, 1, 14, 16, 20, 0, 8, 2, 4,
+    20, 8, 4, 20, 6, 20, 3, 10, 11, 12,
+    15, 2, 12, 5, 19, 19, 5, 9, 6, 18,
+    19, 11, 6, 17, 5, 0, 1, 17, 17
+]
+
+for i in range(50):
+    print("rate")
+    statsd.increment('foo_metric.rate', tags=['a_tag:1'])
+
+    print("gauge")
+    statsd.gauge('foo_metric.gauge', i, tags=["a_tag:2"])
+
+    print("set")
+    statsd.set('foo_metric.set', i, tags=["a_tag:3"])
+
+    print("histogram")
+    statsd.histogram('foo_metric.histogram', hist_data[i], tags=["a_tag:4"])
+
+    print("distribution")
+    statsd.distribution('foo_metric.distribution', dist_data[i], tags=["a_tag:5"])
+
+    statsd.flush()
+    time.sleep(0.01)
+

scripts/e2e/e2e-datadog-metrics/dogstatsd_client/requirements.txt

@@ -0,0 +1 @@
+datadog

scripts/e2e/e2e-datadog-metrics/test.yaml

@@ -0,0 +1,27 @@
+features:
+- e2e-tests-datadog
+
+test: "e2e"
+
+test_filter: 'datadog::metrics::'
+
+runner:
+  env:
+    VECTOR_RECEIVE_PORT: '8081'
+    FAKE_INTAKE_AGENT_ENDPOINT: 'http://fakeintake-agent:80'
+    FAKE_INTAKE_VECTOR_ENDPOINT: 'http://fakeintake-vector:80'
+
+matrix:
+  # validate against the Agent latest nightly and also stable v6 and v7
+  agent_version: ['latest', '6', '7']
+
+# changes to these files/paths will invoke the integration test in CI
+# expressions are evaluated using https://github.com/micromatch/picomatch
+paths:
+- "src/common/datadog.rs"
+- "src/sources/datadog_agent/**"
+- "src/internal_events/datadog_*"
+- "src/sinks/datadog/metrics/**"
+- "src/sinks/util/**"
+- "scripts/integration/datadog-e2e/metrics/**"
+- "tests/data/e2e/datadog/metrics/**"

src/common/datadog.rs

@@ -9,39 +9,55 @@ use vector_lib::{event::DatadogMetricOriginMetadata, sensitive_string::Sensitive
 pub(crate) const DD_US_SITE: &str = "datadoghq.com";
 pub(crate) const DD_EU_SITE: &str = "datadoghq.eu";
 
+/// DatadogSeriesMetric
 #[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
-pub(crate) struct DatadogSeriesMetric {
-    pub(crate) metric: String,
-    pub(crate) r#type: DatadogMetricType,
-    pub(crate) interval: Option<u32>,
-    pub(crate) points: Vec<DatadogPoint<f64>>,
-    pub(crate) tags: Option<Vec<String>>,
+pub struct DatadogSeriesMetric {
+    /// metric
+    pub metric: String,
+    /// metric type
+    pub r#type: DatadogMetricType,
+    /// interval
+    pub interval: Option<u32>,
+    /// points
+    pub points: Vec<DatadogPoint<f64>>,
+    /// tags
+    pub tags: Option<Vec<String>>,
+    /// host
     #[serde(skip_serializing_if = "Option::is_none")]
-    pub(crate) host: Option<String>,
+    pub host: Option<String>,
+    /// source_type_name
     #[serde(skip_serializing_if = "Option::is_none")]
-    pub(crate) source_type_name: Option<String>,
+    pub source_type_name: Option<String>,
+    /// device
     #[serde(skip_serializing_if = "Option::is_none")]
-    pub(crate) device: Option<String>,
+    pub device: Option<String>,
+    /// metadata
     #[serde(skip_serializing_if = "Option::is_none")]
-    pub(crate) metadata: Option<DatadogSeriesMetricMetadata>,
+    pub metadata: Option<DatadogSeriesMetricMetadata>,
 }
 
+/// Datadog series metric metadata
 #[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
-pub(crate) struct DatadogSeriesMetricMetadata {
+pub struct DatadogSeriesMetricMetadata {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub(crate) origin: Option<DatadogMetricOriginMetadata>,
 }
 
+/// Datadog Metric Type
 #[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
 #[serde(rename_all = "snake_case")]
-pub(crate) enum DatadogMetricType {
+pub enum DatadogMetricType {
+    /// Gauge
     Gauge,
+    /// Count
     Count,
+    /// Rate
     Rate,
 }
 
+/// Datadog Point
 #[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
-pub(crate) struct DatadogPoint<T>(pub(crate) i64, pub(crate) T);
+pub struct DatadogPoint<T>(pub i64, pub T);
 
 /// Gets the base API endpoint to use for any calls to Datadog.
 ///