Merge branch 'main' into rayz/otlp-use-stringbuilder

rayz · rayz · commit 25e29d5c5081 · 2026-02-05T15:14:13.000-05:00
diff --git a/.github/chainguard/self.bump-adp-version.create-pr.sts.yaml b/.github/chainguard/self.bump-adp-version.create-pr.sts.yaml
@@ -0,0 +1,14 @@
+# Docs: https://datadoghq.atlassian.net/wiki/spaces/SECENG/pages/5138645099/User+guide+dd-octo-sts
+issuer: https://token.actions.githubusercontent.com
+
+subject: repo:DataDog/saluki:ref:refs/tags/*
+
+claim_pattern:
+  event_name: (push|workflow_dispatch)
+  ref: refs/tags/[0-9]+\.[0-9]+\.[0-9]+
+  ref_protected: "false"
+  job_workflow_ref: DataDog/saluki/\.github/workflows/bump-adp-version\.yml@refs/tags/[0-9]+\.[0-9]+\.[0-9]+
+
+permissions:
+  contents: write
+  pull_requests: write
diff --git a/.github/workflows/bump-adp-version.yml b/.github/workflows/bump-adp-version.yml
@@ -0,0 +1,167 @@
+name: Bump ADP Version
+
+on:
+  push:
+    tags:
+      - "[0-9]+.[0-9]+.[0-9]+"
+  workflow_dispatch:
+
+jobs:
+  bump-version:
+    runs-on: ubuntu-latest
+    permissions:
+      id-token: write
+    steps:
+      - name: Get access token from dd-octo-sts
+        uses: DataDog/dd-octo-sts-action@acaa02eee7e3bb0839e4272dacb37b8f3b58ba80 # v1.0.3
+        id: octo-sts
+        with:
+          scope: DataDog/saluki
+          policy: self.bump-adp-version.create-pr
+
+      - name: Checkout repository
+        uses: actions/checkout@1af3b93b6815bc44a9784bd300feb67ff0d1eeb3 # v6.0.0
+        with:
+          token: "${{ steps.octo-sts.outputs.token }}"
+          ref: main
+
+      - name: Configure Git
+        run: |
+          git config user.name "github-actions[bot]"
+          git config user.email "github-actions[bot]@users.noreply.github.com"
+
+      - name: Calculate new version
+        id: version
+        run: |
+          CURRENT_VERSION=$(grep -E '^version = "' bin/agent-data-plane/Cargo.toml | head -n 1 | cut -d '"' -f 2)
+          echo "Current version: $CURRENT_VERSION"
+
+          MAJOR=$(echo $CURRENT_VERSION | cut -d '.' -f 1)
+          MINOR=$(echo $CURRENT_VERSION | cut -d '.' -f 2)
+          PATCH=$(echo $CURRENT_VERSION | cut -d '.' -f 3)
+          NEW_PATCH=$((PATCH + 1))
+          NEW_VERSION="${MAJOR}.${MINOR}.${NEW_PATCH}"
+
+          echo "New version: $NEW_VERSION"
+          echo "current_version=$CURRENT_VERSION" >> $GITHUB_OUTPUT
+          echo "new_version=$NEW_VERSION" >> $GITHUB_OUTPUT
+
+      - name: Update version in Cargo.toml
+        run: |
+          CURRENT_VERSION="${{ steps.version.outputs.current_version }}"
+          NEW_VERSION="${{ steps.version.outputs.new_version }}"
+
+          sed -i "s/^version = \"$CURRENT_VERSION\"/version = \"$NEW_VERSION\"/" bin/agent-data-plane/Cargo.toml
+
+          echo "Updated bin/agent-data-plane/Cargo.toml"
+
+      - name: Update Cargo.lock
+        run: cargo update -p agent-data-plane --quiet
+
+      - name: Create remote branch
+        run: |
+          BRANCH_NAME=bump-adp-version-${{ steps.version.outputs.new_version }}
+          git push origin HEAD:refs/heads/${BRANCH_NAME}
+
+      - name: Commit Changes
+        id: commit
+        run: |
+          git add bin/agent-data-plane/Cargo.toml Cargo.lock
+          git commit -m "chore(agent-data-plane): bump version to ${{ steps.version.outputs.new_version }}"
+          echo "commit_sha=$(git rev-parse HEAD)" >> $GITHUB_OUTPUT
+
+      - name: Sign Commit
+        uses: DataDog/commit-headless@5a0f3876e0fbdd3a86b3e008acf4ec562db59eee # v2.0.1
+        with:
+          token: "${{ steps.octo-sts.outputs.token }}"
+          branch: bump-adp-version-${{ steps.version.outputs.new_version }}
+          head-sha: ${{ github.sha }}
+          command: push
+          commits: "${{ steps.commit.outputs.commit_sha }}"
+
+      - name: Create Pull Request
+        uses: actions/github-script@ed597411d8f924073f98dfc5c65a23a2325f34cd # v8.0.0
+        with:
+          github-token: "${{ steps.octo-sts.outputs.token }}"
+          script: |
+            const branchName = `bump-adp-version-${{ steps.version.outputs.new_version }}`;
+            const prTitle = `chore(agent-data-plane): bump version to ${{ steps.version.outputs.new_version }}`;
+            const prBody = `## Summary
+
+            This PR bumps the Agent Data Plane version to ${{ steps.version.outputs.new_version }} for the next development cycle.
+
+            ### Details of change
+
+            - **Previous version:** ${{ steps.version.outputs.current_version }}
+            - **New version:** ${{ steps.version.outputs.new_version }}
+
+            _This PR was automatically generated by the [Bump ADP Version](.github/workflows/bump-adp-version.yml) workflow._
+
+            ## Change Type
+
+            - [ ] Bug fix
+            - [ ] New feature
+            - [x] Non-functional (chore, refactoring, docs)
+            - [ ] Performance`;
+            const prLabel = 'ci/bump-adp-version';
+
+            // Make sure we don't have an open PR for the exact same change.
+            const { data: exactPRData } = await github.rest.search.issuesAndPullRequests({
+              q: `repo:${context.repo.owner}/${context.repo.repo} is:pr is:open head:${branchName}`
+            });
+
+            const exactPRs = exactPRData.items;
+
+            if (exactPRs.length > 0) {
+              console.log(`Pull request for this version already exists: ${exactPRs[0].html_url}`);
+              return;
+            }
+
+            // Create the new pull request.
+            const updatePR = await github.rest.pulls.create({
+              owner: context.repo.owner,
+              repo: context.repo.repo,
+              title: prTitle,
+              head: branchName,
+              base: 'main',
+              body: prBody,
+            });
+            console.log(`Pull request created: ${updatePR.data.html_url}`);
+
+            // Set the right labels on the PR.
+            await github.rest.issues.addLabels({
+              owner: context.repo.owner,
+              repo: context.repo.repo,
+              issue_number: updatePR.data.number,
+              labels: [prLabel, 'automated']
+            });
+
+            // Look for other open PRs for bumping the ADP version, and close them out so this one takes precedence.
+            const { data: labeledPRData } = await github.rest.search.issuesAndPullRequests({
+              q: `repo:${context.repo.owner}/${context.repo.repo} is:pr is:open label:"${prLabel}"`
+            });
+
+            const obsoleteUpdatePRs = labeledPRData.items;
+
+            // Close any existing PRs created from this workflow so that we only have the latest one.
+            for (const obsoleteUpdatePR of obsoleteUpdatePRs) {
+              // Don't close the PR we just opened.
+              if (obsoleteUpdatePR.number == updatePR.data.number) {
+                continue;
+              }
+
+              console.log(`Closing existing PR: ${obsoleteUpdatePR.html_url}`);
+              await github.rest.pulls.update({
+                owner: context.repo.owner,
+                repo: context.repo.repo,
+                pull_number: obsoleteUpdatePR.number,
+                state: 'closed'
+              });
+
+              await github.rest.issues.createComment({
+                owner: context.repo.owner,
+                repo: context.repo.repo,
+                issue_number: obsoleteUpdatePR.number,
+                body: `Closing this pull request as a newer version bump is available: [update PR](https://github.com/${context.repo.owner}/${context.repo.repo}/pull/${updatePR.data.number})`
+              });
+            }
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/bin/agent-data-plane/Cargo.toml b/bin/agent-data-plane/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "agent-data-plane"
-version = "0.1.31"
+version = "0.1.32"
 edition = { workspace = true }
 license = { workspace = true }
 repository = { workspace = true }
diff --git a/lib/saluki-components/src/common/otlp/traces/translator.rs b/lib/saluki-components/src/common/otlp/traces/translator.rs
@@ -1,3 +1,4 @@
+use std::collections::hash_map::IntoIter;
 use std::num::NonZeroUsize;
 use std::sync::Arc;
 
@@ -6,7 +7,7 @@ use otlp_protos::opentelemetry::proto::resource::v1::Resource as OtlpResource;
 use otlp_protos::opentelemetry::proto::trace::v1::ResourceSpans;
 use saluki_common::collections::FastHashMap;
 use saluki_common::strings::StringBuilder;
-use saluki_context::tags::TagSet;
+use saluki_context::tags::{SharedTagSet, TagSet};
 use saluki_core::data_model::event::trace::{Span as DdSpan, Trace, TraceSampling};
 use saluki_core::data_model::event::Event;
 use stringtheory::interning::GenericMapInterner;
@@ -74,7 +75,7 @@ impl OtlpTracesTranslator {
         }
     }
 
-    pub fn translate_resource_spans(&mut self, resource_spans: ResourceSpans, metrics: &Metrics) -> Vec<Event> {
+    pub fn translate_spans(&mut self, resource_spans: ResourceSpans, metrics: &Metrics) -> impl Iterator<Item = Event> {
         let resource: OtlpResource = resource_spans.resource.unwrap_or_default();
         let ignore_missing_fields = self.config.ignore_missing_datadog_fields;
         let compute_top_level = self.config.enable_otlp_compute_top_level_by_span_kind;
@@ -120,23 +121,38 @@ impl OtlpTracesTranslator {
             }
         }
 
-        traces_by_id
-            .into_iter()
-            .filter_map(|(_, entry)| {
-                if entry.spans.is_empty() {
-                    None
-                } else {
-                    let mut trace = Trace::new(entry.spans, resource_tags.clone());
+        OtlpTraceEventsIter {
+            resource_tags,
+            entries: traces_by_id.into_iter(),
+        }
+    }
+}
 
-                    // Set the trace-level sampling priority if one was found
-                    if let Some(priority) = entry.priority {
-                        trace.set_sampling(Some(TraceSampling::new(false, Some(priority), None, None)));
-                    }
+struct OtlpTraceEventsIter {
+    resource_tags: SharedTagSet,
+    entries: IntoIter<u64, TraceEntry>,
+}
 
-                    Some(Event::Trace(trace))
-                }
-            })
-            .collect()
+impl Iterator for OtlpTraceEventsIter {
+    type Item = Event;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        for (_, entry) in self.entries.by_ref() {
+            if entry.spans.is_empty() {
+                continue;
+            }
+
+            let mut trace = Trace::new(entry.spans, self.resource_tags.clone());
+
+            // Set the trace-level sampling priority if one was found
+            if let Some(priority) = entry.priority {
+                trace.set_sampling(Some(TraceSampling::new(false, Some(priority), None, None)));
+            }
+
+            return Some(Event::Trace(trace));
+        }
+
+        None
     }
 }
 
diff --git a/lib/saluki-components/src/decoders/otlp/mod.rs b/lib/saluki-components/src/decoders/otlp/mod.rs
@@ -135,8 +135,7 @@ impl Decoder for OtlpDecoder {
                             };
 
                             for resource_spans in request.resource_spans {
-                                let trace_events = traces_translator.translate_resource_spans(resource_spans, &metrics);
-                                for trace_event in trace_events {
+                                for trace_event in traces_translator.translate_spans(resource_spans, &metrics) {
                                     if let Some(event_buffer) = event_buffer_manager.try_push(trace_event) {
                                         if let Err(e) = context.dispatcher().dispatch(event_buffer).await {
                                             error!(error = %e, "Failed to dispatch trace events.");
diff --git a/lib/saluki-components/src/sources/otlp/metrics/translator.rs b/lib/saluki-components/src/sources/otlp/metrics/translator.rs
@@ -3,6 +3,7 @@
 use std::collections::HashSet;
 use std::sync::LazyLock;
 use std::time::{SystemTime, UNIX_EPOCH};
+use std::vec::IntoIter;
 
 use otlp_protos::opentelemetry::proto::common::v1::KeyValue as OtlpKeyValue;
 use otlp_protos::opentelemetry::proto::metrics::v1::{
@@ -96,11 +97,11 @@ impl OtlpMetricsTranslator {
         }
     }
 
-    /// Translates a batch of OTLP `ResourceMetrics` into a collection of Saluki `Event`s.
+    /// Translates a batch of OTLP `ResourceMetrics` into Saluki `Event`s.
     /// This is the Rust equivalent of the Go `MapMetrics` function.
-    pub fn map_metrics(
+    pub fn translate_metrics(
         &mut self, resource_metrics: OtlpResourceMetrics, metrics: &Metrics,
-    ) -> Result<Vec<Event>, GenericError> {
+    ) -> Result<IntoIter<Event>, GenericError> {
         let mut events = Vec::new();
         let resource = resource_metrics.resource.unwrap_or_default();
         let source = self.attribute_translator.resource_to_source(&resource);
@@ -204,7 +205,7 @@ impl OtlpMetricsTranslator {
 
         metrics.metrics_received().increment(events.len() as u64);
 
-        Ok(events)
+        Ok(events.into_iter())
     }
 
     /// Creates a new `OtlpMetricsTranslator` for tests.
diff --git a/lib/saluki-components/src/sources/otlp/mod.rs b/lib/saluki-components/src/sources/otlp/mod.rs
