chore(dev): add multiservice opentelemetry trace e2e test

.github/actions/setup/action.yml

@@ -95,7 +95,7 @@ inputs:
   vdev:
     required: false
     default: true
-    description: "Install vdev CLI tool via cargo binstall (version from vdev/Cargo.toml)."
+    description: "Install vdev CLI, building from checkout for PRs that change vdev/."
 
   # prepare.sh - npm
   actionlint:
@@ -216,6 +216,27 @@ runs:
         path: scripts/environment/npm-tools/node_modules
         key: ${{ runner.os }}-npm-tools-${{ hashFiles('scripts/environment/npm-tools/*') }}
 
+    - name: Detect vdev source changes
+      if: ${{ inputs.vdev == 'true' && (github.event.pull_request || github.event.merge_group || github.event_name == 'workflow_dispatch') }}
+      shell: bash
+      env:
+        VDEV_BASE_SHA: ${{ github.event.pull_request.base.sha || github.event.merge_group.base_sha }}
+      run: |
+        # Manual runs have no comparison base, so exercise the checked-out harness.
+        if [[ -z "$VDEV_BASE_SHA" ]]
+        then
+          echo "VDEV_BUILD_FROM_SOURCE=true" >> "$GITHUB_ENV"
+          exit 0
+        fi
+        if ! git cat-file -e "${VDEV_BASE_SHA}^{commit}" 2>/dev/null
+        then
+          git fetch --no-tags --depth=1 origin "$VDEV_BASE_SHA"
+        fi
+        if ! git diff --quiet "$VDEV_BASE_SHA" HEAD -- vdev/
+        then
+          echo "VDEV_BUILD_FROM_SOURCE=true" >> "$GITHUB_ENV"
+        fi
+
     - name: Run prepare.sh
       shell: bash
       env:

.github/workflows/ci-integration-review.yml

@@ -187,10 +187,11 @@ jobs:
           command: vdev integration run ${{ matrix.service }} ${{ runner.debug == '1' && '--always-show-logs' || '' }}
 
   e2e-tests:
+    # The dedicated command also starts with the ordinary opentelemetry-traces command.
+    if: github.repository == 'vectordotdev/vector' && !startsWith(github.event.review.body, '/ci-run-e2e-opentelemetry-traces-multiservice')
     needs:
       - prep-pr
       - build-test-runner
-    if: github.repository == 'vectordotdev/vector'
     runs-on: ubuntu-24.04-8core
     timeout-minutes: 30
     permissions:
@@ -252,13 +253,81 @@ jobs:
           max_attempts: 3
           command: vdev e2e run ${{ matrix.service }} ${{ runner.debug == '1' && '--always-show-logs' || '' }}
 
+  otel-traces-multiservice:
+    name: Multiservice OpenTelemetry trace validation
+    if: github.repository == 'vectordotdev/vector' && startsWith(github.event.review.body, '/ci-run-e2e-opentelemetry-traces-multiservice')
+    needs:
+      - prep-pr
+      - build-test-runner
+    runs-on: ubuntu-24.04-8core
+    timeout-minutes: 60
+    permissions:
+      contents: read
+      packages: read # Required to pull test runner image from GHCR
+      id-token: write
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          submodules: "recursive"
+          ref: ${{ github.event.review.commit_id }}
+
+      - uses: ./.github/actions/dd-token
+
+      - uses: ./.github/actions/setup
+        env:
+          # Include harness changes inherited from the PR's base branch.
+          VDEV_BUILD_FROM_SOURCE: true
+        with:
+          vdev: true
+          datadog-ci: true
+
+      - name: Pull test runner image
+        uses: ./.github/actions/pull-test-runner
+        with:
+          github_token: ${{ secrets.GITHUB_TOKEN }}
+          commit_sha: ${{ github.event.review.commit_id }}
+
+      - name: Multiservice OpenTelemetry trace validation
+        run: |
+          mkdir -p target/otel-traces-multiservice-evidence
+          set -o pipefail
+          vdev e2e run opentelemetry-traces-multiservice --always-show-logs 2>&1 | tee target/otel-traces-multiservice-evidence/runner.log
+
+      - name: Collect multiservice trace evidence
+        if: always()
+        run: |
+          mkdir -p target/otel-traces-multiservice-evidence
+          git rev-parse HEAD > target/otel-traces-multiservice-evidence/revision.txt
+          cp -R tests/e2e/opentelemetry-traces-multiservice/config tests/e2e/opentelemetry-traces-multiservice/data target/otel-traces-multiservice-evidence/
+          if docker volume inspect opentelemetry_traces_multiservice_evidence >/dev/null 2>&1; then
+            docker run --rm \
+              -v opentelemetry_traces_multiservice_evidence:/output/opentelemetry-traces-multiservice:ro \
+              -v "$PWD/target/otel-traces-multiservice-evidence:/evidence" \
+              alpine:3.20 sh -c 'if [ -d /output/opentelemetry-traces-multiservice ]; then cp -R /output/opentelemetry-traces-multiservice/. /evidence/; fi'
+          fi
+          summaries=(target/otel-traces-multiservice-evidence/*/summary.md)
+          if [ -f "${summaries[0]}" ]; then
+            cat "${summaries[@]}" >> "$GITHUB_STEP_SUMMARY"
+          else
+            echo 'Multiservice trace test did not produce a validation summary. See runner.log and the failed job steps.' >> "$GITHUB_STEP_SUMMARY"
+          fi
+
+      - name: Upload multiservice trace evidence
+        if: always()
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: otel-traces-multiservice-${{ github.event.review.commit_id }}
+          path: target/otel-traces-multiservice-evidence/
+          retention-days: 14
+
   update-pr-status:
     name: Signal result to PR
     runs-on: ubuntu-24.04
     timeout-minutes: 5
     needs:
       - integration-tests
       - e2e-tests
+      - otel-traces-multiservice
     if: always() && (startsWith(github.event.review.body, '/ci-run-integration') || contains(github.event.review.body, '/ci-run-all') || contains(github.event.review.body, '/ci-run-e2e'))
     permissions:
       statuses: write # Required to set final commit status

Cargo.toml

@@ -1159,7 +1159,8 @@ e2e-tests-opentelemetry = [
   "sinks-console",
   "sinks-file",
   "codecs-opentelemetry",
-  "dep:prost-reflect"
+  "dep:prost-reflect",
+  "dep:bollard"
 ]
 
 vector-api-tests = [

scripts/environment/prepare.sh

@@ -185,6 +185,12 @@ maybe_install_cargo_tool() {
   # code of an unmerged checkout. If no prebuilt binary exists, build the
   # current checkout instead.
   if [[ "$tool" == "vdev" ]]; then
+    if [[ "${VDEV_BUILD_FROM_SOURCE:-false}" == "true" ]]
+    then
+      echo "Building vdev from the working tree to test harness changes..."
+      cargo install -f --path vdev --locked
+      return 0
+    fi
     local vdev_installer=("${cargo_tool_installer[@]}")
     if [[ "${vdev_installer[0]}" == "binstall" ]]; then
       vdev_installer+=(--force --disable-strategies compile)

tests/e2e/datadog/metrics/mod.rs

@@ -1,8 +1,13 @@
+use std::time::Duration;
+
 use async_compression::tokio::bufread::{ZstdDecoder, ZstdEncoder};
 use base64::{Engine, prelude::BASE64_STANDARD};
 use bytes::Bytes;
 use serde_json::Value;
-use tokio::io::{AsyncReadExt, BufReader};
+use tokio::{
+    io::{AsyncReadExt, BufReader},
+    time::{Instant, sleep},
+};
 use tracing::{debug, warn};
 use vector::test_util::{compression::is_zstd, trace_init};
 use vector_common::Result;
@@ -110,13 +115,12 @@ where
 async fn validate() {
     trace_init();
 
-    // Even with configuring docker service dependencies, we need a small buffer of time
-    // to ensure events flow through to fakeintake before asking for them
-    std::thread::sleep(std::time::Duration::from_secs(2));
-
-    series::validate().await;
+    // Wait for workload data, not compilation time, to make both pipelines ready.
+    // Leave time to report failures before nextest's two-minute test timeout.
+    let deadline = Instant::now() + Duration::from_secs(90);
+    series::validate(deadline).await;
 
-    sketches::validate().await;
+    sketches::validate(deadline).await;
 }
 
 async fn compress_with_zstd(data: &[u8]) -> Vec<u8> {

tests/e2e/datadog/metrics/series.rs

@@ -233,36 +233,50 @@ fn assert_v2_resource(series: &[MetricPayload]) {
     );
 }
 
-async fn get_v2_series_from_pipeline(address: String) -> SeriesIntake {
-    info!("getting v2 series payloads");
-    let payloads =
-        get_fakeintake_payloads::<FakeIntakeResponseRaw>(&address, SERIES_ENDPOINT_V2).await;
-
-    info!("unpacking payloads");
-    let payloads = unpack_proto_payloads::<MetricPayload>(&payloads)
-        .await
-        .expect("Failed to unpack v2 series payloads");
-    assert_v2_resource(&payloads);
-
-    info!("generating series intake");
-    let intake = generate_series_intake(&payloads);
-
-    common_series_assertions(&intake);
-
-    info!("{intake:?}");
-
-    intake
+fn has_expected_series(payloads: &[MetricPayload]) -> bool {
+    ["rate", "gauge", "set", "histogram", "resource"]
+        .iter()
+        .all(|suffix| {
+            let prefix = format!("foo_metric.{suffix}");
+            payloads
+                .iter()
+                .flat_map(|payload| &payload.series)
+                .any(|series| series.metric.starts_with(&prefix))
+        })
 }
 
-async fn get_v3_series_from_pipeline(address: String) -> SeriesIntake {
-    info!("getting v3 series payloads");
-    let payloads =
-        get_fakeintake_payloads::<FakeIntakeResponseJson>(&address, SERIES_ENDPOINT_V3).await;
+async fn get_series_from_pipeline(
+    address: String,
+    api_version: &str,
+    deadline: Instant,
+) -> SeriesIntake {
+    let payloads = loop {
+        info!("getting {api_version} series payloads from {address}");
+        let payloads = match api_version {
+            "v2" => {
+                let payloads =
+                    get_fakeintake_payloads::<FakeIntakeResponseRaw>(&address, SERIES_ENDPOINT_V2)
+                        .await;
+                unpack_proto_payloads::<MetricPayload>(&payloads)
+                    .await
+                    .expect("Failed to unpack v2 series payloads")
+            }
+            "v3" => {
+                let payloads =
+                    get_fakeintake_payloads::<FakeIntakeResponseJson>(&address, SERIES_ENDPOINT_V3)
+                        .await;
+                unpack_v3_series(&payloads.payloads)
+            }
+            v => panic!("Unknown CONFIG_SERIES_API_VERSION: {v}"),
+        };
+
+        if has_expected_series(&payloads) || Instant::now() >= deadline {
+            break payloads;
+        }
+        sleep(Duration::from_secs(1)).await;
+    };
 
-    info!("unpacking payloads");
-    let payloads = unpack_v3_series(&payloads.payloads);
-    // V3 carries resources in a dedicated column just like V2 does, so the V2 expectations
-    // (structured resource present, not duplicated back into the tags) hold unchanged.
+    // Both V2 and V3 carry resources separately from tags.
     assert_v2_resource(&payloads);
 
     info!("generating series intake");
@@ -358,19 +372,37 @@ fn compare_intakes(agent_intake: &SeriesIntake, vector_intake: &SeriesIntake) {
         });
 }
 
-pub(super) async fn validate() {
+pub(super) async fn validate(deadline: Instant) {
     let api_version = std::env::var("CONFIG_SERIES_API_VERSION")
         .expect("CONFIG_SERIES_API_VERSION must be set ('v2' or 'v3')");
 
     info!("==== getting series data from agent-only pipeline ==== ");
-    let agent_intake = get_v2_series_from_pipeline(fake_intake_agent_address()).await;
+    let agent_intake = get_series_from_pipeline(fake_intake_agent_address(), "v2", deadline).await;
 
     info!("==== getting series data from agent-vector pipeline ({api_version}) ====");
-    let vector_intake = match api_version.as_str() {
-        "v2" => get_v2_series_from_pipeline(fake_intake_vector_address()).await,
-        "v3" => get_v3_series_from_pipeline(fake_intake_vector_address()).await,
-        v => panic!("Unknown CONFIG_SERIES_API_VERSION: {v}"),
-    };
+    let vector_intake =
+        get_series_from_pipeline(fake_intake_vector_address(), &api_version, deadline).await;
 
     compare_intakes(&agent_intake, &vector_intake);
 }
+
+#[test]
+fn series_readiness_requires_all_workload_metrics() {
+    let names = [
+        "foo_metric.rate",
+        "foo_metric.gauge",
+        "foo_metric.set",
+        "foo_metric.histogram.avg",
+        RESOURCE_METRIC_NAME,
+    ];
+    let mut payload = MetricPayload::default();
+    for name in names {
+        assert!(!has_expected_series(std::slice::from_ref(&payload)));
+        payload.series.push(MetricSeries {
+            metric: name.into(),
+            ..Default::default()
+        });
+    }
+    assert!(has_expected_series(std::slice::from_ref(&payload)));
+    assert!(!has_expected_series(&[]));
+}

tests/e2e/datadog/metrics/sketches.rs

@@ -87,18 +87,26 @@ fn common_sketch_assertions(sketches: &SketchIntake) {
     assert!(found, "Didn't receive metric type distribution");
 }
 
-async fn get_sketches_from_pipeline(address: String) -> SketchIntake {
-    info!("getting sketch payloads");
-    let payloads =
-        get_fakeintake_payloads::<FakeIntakeResponseRaw>(&address, SKETCHES_ENDPOINT).await;
-
-    info!("unpacking payloads");
-    let payloads = unpack_proto_payloads(&payloads)
-        .await
-        .expect("Failed to unpack sketch payloads");
-
-    info!("generating sketch intake");
-    let sketches = generate_sketch_intake(payloads);
+async fn get_sketches_from_pipeline(address: String, deadline: Instant) -> SketchIntake {
+    let sketches = loop {
+        info!("getting sketch payloads from {address}");
+        let payloads =
+            get_fakeintake_payloads::<FakeIntakeResponseRaw>(&address, SKETCHES_ENDPOINT).await;
+
+        let payloads = unpack_proto_payloads(&payloads)
+            .await
+            .expect("Failed to unpack sketch payloads");
+        let sketches = generate_sketch_intake(payloads);
+
+        if sketches
+            .keys()
+            .any(|context| context.metric_name.starts_with("foo_metric.distribution"))
+            || Instant::now() >= deadline
+        {
+            break sketches;
+        }
+        sleep(Duration::from_secs(1)).await;
+    };
 
     common_sketch_assertions(&sketches);
 
@@ -107,12 +115,12 @@ async fn get_sketches_from_pipeline(address: String) -> SketchIntake {
     sketches
 }
 
-pub(super) async fn validate() {
+pub(super) async fn validate(deadline: Instant) {
     info!("==== getting sketch data from agent-only pipeline ==== ");
-    let agent_sketches = get_sketches_from_pipeline(fake_intake_agent_address()).await;
+    let agent_sketches = get_sketches_from_pipeline(fake_intake_agent_address(), deadline).await;
 
     info!("==== getting sketch data from agent-vector pipeline ====");
-    let vector_sketches = get_sketches_from_pipeline(fake_intake_vector_address()).await;
+    let vector_sketches = get_sketches_from_pipeline(fake_intake_vector_address(), deadline).await;
 
     agent_sketches
         .iter()