Add support for span id remapper in logs pipelines processors

.apigentools-info

@@ -4,13 +4,13 @@
     "spec_versions": {
         "v1": {
             "apigentools_version": "1.6.6",
-            "regenerated": "2025-02-21 08:00:35.115496",
-            "spec_repo_commit": "8f2d39c3"
+            "regenerated": "2025-02-21 18:16:27.438843",
+            "spec_repo_commit": "5de91bd6"
         },
         "v2": {
             "apigentools_version": "1.6.6",
-            "regenerated": "2025-02-21 08:00:35.131024",
-            "spec_repo_commit": "8f2d39c3"
+            "regenerated": "2025-02-21 18:16:27.454329",
+            "spec_repo_commit": "5de91bd6"
         }
     }
 }

.generator/schemas/v1/openapi.yaml

@@ -6049,6 +6049,7 @@ components:
       - $ref: '#/components/schemas/LogsLookupProcessor'
       - $ref: '#/components/schemas/ReferenceTableLogsLookupProcessor'
       - $ref: '#/components/schemas/LogsTraceRemapper'
+      - $ref: '#/components/schemas/LogsSpanRemapper'
     LogsQueryCompute:
       description: Define computation for a log query.
       properties:
@@ -6160,6 +6161,43 @@ components:
       x-enum-varnames:
       - TIME_ASCENDING
       - TIME_DESCENDING
+    LogsSpanRemapper:
+      description: "There are two ways to define correlation between application spans
+        and logs:\n\n  1. Follow the documentation on [how to inject a span ID in
+        the application logs](https://docs.datadoghq.com/tracing/connect_logs_and_traces).\n
+        \ Log integrations automatically handle all remaining setup steps by default.\n\n
+        \ 2. Use the span remapper processor to define a log attribute as its associated
+        span ID."
+      properties:
+        is_enabled:
+          default: false
+          description: Whether or not the processor is enabled.
+          type: boolean
+        name:
+          description: Name of the processor.
+          type: string
+        sources:
+          default:
+          - dd.span_id
+          description: Array of source attributes.
+          items:
+            description: Attribute to extract the span ID from.
+            type: string
+          type: array
+        type:
+          $ref: '#/components/schemas/LogsSpanRemapperType'
+      required:
+      - type
+      type: object
+    LogsSpanRemapperType:
+      default: span-id-remapper
+      description: Type of logs span remapper.
+      enum:
+      - span-id-remapper
+      example: span-id-remapper
+      type: string
+      x-enum-varnames:
+      - SPAN_ID_REMAPPER
     LogsStatusRemapper:
       description: "Use this Processor if you want to assign some attributes as the
         official status.\n\nEach incoming status value is mapped as follows.\n\n  -
@@ -6275,7 +6313,7 @@ components:
           - dd.trace_id
           description: Array of source attributes.
           items:
-            description: Attribute to extract the Trace ID from.
+            description: Attribute to extract the trace ID from.
             type: string
           type: array
         type:

examples/v1_logs-pipelines_CreateLogsPipeline_2707101123.rs

@@ -0,0 +1,29 @@
+// Create a pipeline with Span Id Remapper returns "OK" response
+use datadog_api_client::datadog;
+use datadog_api_client::datadogV1::api_logs_pipelines::LogsPipelinesAPI;
+use datadog_api_client::datadogV1::model::LogsFilter;
+use datadog_api_client::datadogV1::model::LogsPipeline;
+use datadog_api_client::datadogV1::model::LogsProcessor;
+use datadog_api_client::datadogV1::model::LogsSpanRemapper;
+use datadog_api_client::datadogV1::model::LogsSpanRemapperType;
+
+#[tokio::main]
+async fn main() {
+    let body = LogsPipeline::new("testPipeline".to_string())
+        .filter(LogsFilter::new().query("source:python".to_string()))
+        .processors(vec![LogsProcessor::LogsSpanRemapper(Box::new(
+            LogsSpanRemapper::new(LogsSpanRemapperType::SPAN_ID_REMAPPER)
+                .is_enabled(true)
+                .name("test_filter".to_string())
+                .sources(vec!["dd.span_id".to_string()]),
+        ))])
+        .tags(vec![]);
+    let configuration = datadog::Configuration::new();
+    let api = LogsPipelinesAPI::with_config(configuration);
+    let resp = api.create_logs_pipeline(body).await;
+    if let Ok(value) = resp {
+        println!("{:#?}", value);
+    } else {
+        println!("{:#?}", resp.unwrap_err());
+    }
+}

src/datadogV1/model/mod.rs

@@ -900,6 +900,10 @@ pub mod model_logs_trace_remapper;
 pub use self::model_logs_trace_remapper::LogsTraceRemapper;
 pub mod model_logs_trace_remapper_type;
 pub use self::model_logs_trace_remapper_type::LogsTraceRemapperType;
+pub mod model_logs_span_remapper;
+pub use self::model_logs_span_remapper::LogsSpanRemapper;
+pub mod model_logs_span_remapper_type;
+pub use self::model_logs_span_remapper_type::LogsSpanRemapperType;
 pub mod model_logs_processor;
 pub use self::model_logs_processor::LogsProcessor;
 pub mod model_logs_pipeline_processor_type;

src/datadogV1/model/model_logs_processor.rs

@@ -26,6 +26,7 @@ pub enum LogsProcessor {
         Box<crate::datadogV1::model::ReferenceTableLogsLookupProcessor>,
     ),
     LogsTraceRemapper(Box<crate::datadogV1::model::LogsTraceRemapper>),
+    LogsSpanRemapper(Box<crate::datadogV1::model::LogsSpanRemapper>),
     UnparsedObject(crate::datadog::UnparsedObject),
 }
 
@@ -150,6 +151,13 @@ impl<'de> Deserialize<'de> for LogsProcessor {
                 return Ok(LogsProcessor::LogsTraceRemapper(_v));
             }
         }
+        if let Ok(_v) =
+            serde_json::from_value::<Box<crate::datadogV1::model::LogsSpanRemapper>>(value.clone())
+        {
+            if !_v._unparsed {
+                return Ok(LogsProcessor::LogsSpanRemapper(_v));
+            }
+        }
 
         return Ok(LogsProcessor::UnparsedObject(
             crate::datadog::UnparsedObject { value },

src/datadogV1/model/model_logs_span_remapper.rs

@@ -0,0 +1,156 @@
+// Unless explicitly stated otherwise all files in this repository are licensed under the Apache-2.0 License.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2019-Present Datadog, Inc.
+use serde::de::{Error, MapAccess, Visitor};
+use serde::{Deserialize, Deserializer, Serialize};
+use serde_with::skip_serializing_none;
+use std::fmt::{self, Formatter};
+
+/// There are two ways to define correlation between application spans and logs:
+///
+///   1. Follow the documentation on [how to inject a span ID in the application logs](<https://docs.datadoghq.com/tracing/connect_logs_and_traces>).
+///   Log integrations automatically handle all remaining setup steps by default.
+///
+///   2. Use the span remapper processor to define a log attribute as its associated span ID.
+#[non_exhaustive]
+#[skip_serializing_none]
+#[derive(Clone, Debug, PartialEq, Serialize)]
+pub struct LogsSpanRemapper {
+    /// Whether or not the processor is enabled.
+    #[serde(rename = "is_enabled")]
+    pub is_enabled: Option<bool>,
+    /// Name of the processor.
+    #[serde(rename = "name")]
+    pub name: Option<String>,
+    /// Array of source attributes.
+    #[serde(rename = "sources")]
+    pub sources: Option<Vec<String>>,
+    /// Type of logs span remapper.
+    #[serde(rename = "type")]
+    pub type_: crate::datadogV1::model::LogsSpanRemapperType,
+    #[serde(flatten)]
+    pub additional_properties: std::collections::BTreeMap<String, serde_json::Value>,
+    #[serde(skip)]
+    #[serde(default)]
+    pub(crate) _unparsed: bool,
+}
+
+impl LogsSpanRemapper {
+    pub fn new(type_: crate::datadogV1::model::LogsSpanRemapperType) -> LogsSpanRemapper {
+        LogsSpanRemapper {
+            is_enabled: None,
+            name: None,
+            sources: None,
+            type_,
+            additional_properties: std::collections::BTreeMap::new(),
+            _unparsed: false,
+        }
+    }
+
+    pub fn is_enabled(mut self, value: bool) -> Self {
+        self.is_enabled = Some(value);
+        self
+    }
+
+    pub fn name(mut self, value: String) -> Self {
+        self.name = Some(value);
+        self
+    }
+
+    pub fn sources(mut self, value: Vec<String>) -> Self {
+        self.sources = Some(value);
+        self
+    }
+
+    pub fn additional_properties(
+        mut self,
+        value: std::collections::BTreeMap<String, serde_json::Value>,
+    ) -> Self {
+        self.additional_properties = value;
+        self
+    }
+}
+
+impl<'de> Deserialize<'de> for LogsSpanRemapper {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        struct LogsSpanRemapperVisitor;
+        impl<'a> Visitor<'a> for LogsSpanRemapperVisitor {
+            type Value = LogsSpanRemapper;
+
+            fn expecting(&self, f: &mut Formatter<'_>) -> fmt::Result {
+                f.write_str("a mapping")
+            }
+
+            fn visit_map<M>(self, mut map: M) -> Result<Self::Value, M::Error>
+            where
+                M: MapAccess<'a>,
+            {
+                let mut is_enabled: Option<bool> = None;
+                let mut name: Option<String> = None;
+                let mut sources: Option<Vec<String>> = None;
+                let mut type_: Option<crate::datadogV1::model::LogsSpanRemapperType> = None;
+                let mut additional_properties: std::collections::BTreeMap<
+                    String,
+                    serde_json::Value,
+                > = std::collections::BTreeMap::new();
+                let mut _unparsed = false;
+
+                while let Some((k, v)) = map.next_entry::<String, serde_json::Value>()? {
+                    match k.as_str() {
+                        "is_enabled" => {
+                            if v.is_null() {
+                                continue;
+                            }
+                            is_enabled = Some(serde_json::from_value(v).map_err(M::Error::custom)?);
+                        }
+                        "name" => {
+                            if v.is_null() {
+                                continue;
+                            }
+                            name = Some(serde_json::from_value(v).map_err(M::Error::custom)?);
+                        }
+                        "sources" => {
+                            if v.is_null() {
+                                continue;
+                            }
+                            sources = Some(serde_json::from_value(v).map_err(M::Error::custom)?);
+                        }
+                        "type" => {
+                            type_ = Some(serde_json::from_value(v).map_err(M::Error::custom)?);
+                            if let Some(ref _type_) = type_ {
+                                match _type_ {
+                                    crate::datadogV1::model::LogsSpanRemapperType::UnparsedObject(_type_) => {
+                                        _unparsed = true;
+                                    },
+                                    _ => {}
+                                }
+                            }
+                        }
+                        &_ => {
+                            if let Ok(value) = serde_json::from_value(v.clone()) {
+                                additional_properties.insert(k, value);
+                            }
+                        }
+                    }
+                }
+                let type_ = type_.ok_or_else(|| M::Error::missing_field("type_"))?;
+
+                let content = LogsSpanRemapper {
+                    is_enabled,
+                    name,
+                    sources,
+                    type_,
+                    additional_properties,
+                    _unparsed,
+                };
+
+                Ok(content)
+            }
+        }
+
+        deserializer.deserialize_any(LogsSpanRemapperVisitor)
+    }
+}

src/datadogV1/model/model_logs_span_remapper_type.rs

@@ -0,0 +1,48 @@
+// Unless explicitly stated otherwise all files in this repository are licensed under the Apache-2.0 License.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2019-Present Datadog, Inc.
+
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
+
+#[non_exhaustive]
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum LogsSpanRemapperType {
+    SPAN_ID_REMAPPER,
+    UnparsedObject(crate::datadog::UnparsedObject),
+}
+
+impl ToString for LogsSpanRemapperType {
+    fn to_string(&self) -> String {
+        match self {
+            Self::SPAN_ID_REMAPPER => String::from("span-id-remapper"),
+            Self::UnparsedObject(v) => v.value.to_string(),
+        }
+    }
+}
+
+impl Serialize for LogsSpanRemapperType {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        match self {
+            Self::UnparsedObject(v) => v.serialize(serializer),
+            _ => serializer.serialize_str(self.to_string().as_str()),
+        }
+    }
+}
+
+impl<'de> Deserialize<'de> for LogsSpanRemapperType {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let s: String = String::deserialize(deserializer)?;
+        Ok(match s.as_str() {
+            "span-id-remapper" => Self::SPAN_ID_REMAPPER,
+            _ => Self::UnparsedObject(crate::datadog::UnparsedObject {
+                value: serde_json::Value::String(s.into()),
+            }),
+        })
+    }
+}

tests/scenarios/cassettes/v1/logs_pipelines/Create-a-pipeline-with-Span-Id-Remapper-returns-OK-response.frozen

@@ -0,0 +1 @@
+2025-02-20T15:44:02.905Z