fix(otlp): accept flexible timestamp formats in JSON payloads (#1108)

## Summary

- Adds normalization layer to convert OTLP JSON timestamps to string
format before deserialization
- Supports string timestamps (proto3 JSON spec), integer timestamps
(common in some SDKs), and object timestamps `{"low": n, "high": m}`
(from buggy older JS SDKs)
- Fixes errors like `"invalid type: integer, expected a string"` seen
with serverless-self-monitoring

## Root Cause

The `opentelemetry-proto` crate's serde deserializer only accepts
string-encoded 64-bit timestamps (per proto3 JSON spec), but some
OpenTelemetry SDKs send timestamps as integers or objects.

**serverless-self-monitoring uses outdated SDK versions:**

| Handler | Package | Version |
|---------|---------|---------|
| node18/20/22 | `@opentelemetry/exporter-trace-otlp-http` | **0.36.1**
(~Feb 2023) |
| python310 | `opentelemetry-exporter-otlp-proto-http` | **1.15.0**
(~Feb 2023) |

The extension's integration tests use **0.54.2** (current), which
properly serializes timestamps as strings, so this issue wasn't caught.

**Known upstream issue:** [opentelemetry-rust
#1662](open-telemetry/opentelemetry-rust#1662)
- The Rust `opentelemetry-proto` crate's serde implementation doesn't
accept both string and integer formats for 64-bit integers, as the
proto3 JSON spec recommends.

**Related JS SDK issue:** [opentelemetry-js
#4216](open-telemetry/opentelemetry-js#4216) -
Older JS SDK versions sent timestamps as `{"low": n, "high": m}` objects
instead of strings.

## Solution

Instead of waiting for upstream fixes or requiring all senders to update
their SDKs, we normalize JSON timestamps before deserialization:

1. Parse JSON into `serde_json::Value`
2. Recursively find timestamp fields (`startTimeUnixNano`,
`endTimeUnixNano`, `timeUnixNano`, `observedTimeUnixNano`)
3. Convert integers → strings, objects `{"low": n, "high": m}` →
reconstructed string
4. Deserialize normalized JSON with `opentelemetry-proto`

## Test plan

- [x] Unit tests for timestamp normalization (string, integer, object
formats)
- [x] Unit tests for nested structure normalization
- [x] Full test suite passes (506 tests)
- [x] Clippy passes with `-D warnings`
- [ ] Integration tests with serverless-self-monitoring

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>

Author: jchrostek-dd

bottlecap/src/otlp/processor.rs

@@ -1,10 +1,75 @@
 use libdd_trace_protobuf::pb::Span as DatadogSpan;
 use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest;
 use prost::Message;
+use serde_json::Value;
 use std::{error::Error, sync::Arc};
 
 use crate::{config::Config, otlp::transform::otel_resource_spans_to_dd_spans};
 
+/// Fields that contain 64-bit nanosecond timestamps and need flexible deserialization.
+/// Per proto3 JSON spec, these should be string-encoded, but some SDKs send integers
+/// or even objects like {"low": ..., "high": ...}.
+const TIMESTAMP_FIELDS: &[&str] = &[
+    "startTimeUnixNano",
+    "endTimeUnixNano",
+    "timeUnixNano",
+    "observedTimeUnixNano",
+];
+
+/// Recursively normalizes timestamp fields in a JSON value.
+/// Converts integer timestamps to strings and handles the {"low": ..., "high": ...}
+/// object format from older/buggy OpenTelemetry JS SDKs.
+fn normalize_timestamps(value: &mut Value) {
+    match value {
+        Value::Object(map) => {
+            for (key, val) in map.iter_mut() {
+                if TIMESTAMP_FIELDS.contains(&key.as_str()) {
+                    normalize_timestamp_value(val);
+                } else {
+                    normalize_timestamps(val);
+                }
+            }
+        }
+        Value::Array(arr) => {
+            for item in arr.iter_mut() {
+                normalize_timestamps(item);
+            }
+        }
+        _ => {}
+    }
+}
+
+/// Normalizes a single timestamp value to a string.
+/// Handles:
+/// - String: already correct, leave as-is
+/// - Integer: convert to string
+/// - Object {"low": n, "high": m}: reconstruct 64-bit value and convert to string
+fn normalize_timestamp_value(value: &mut Value) {
+    match value {
+        Value::Number(n) => {
+            // Integer timestamp - convert to string
+            if let Some(i) = n.as_u64() {
+                *value = Value::String(i.to_string());
+            } else if let Some(i) = n.as_i64() {
+                *value = Value::String(i.to_string());
+            }
+        }
+        Value::Object(map) => {
+            // Handle {"low": n, "high": m} format from buggy JS SDKs
+            // This represents a 64-bit integer split into two 32-bit parts
+            let low_val = map.get("low").and_then(Value::as_u64);
+            let high_val = map.get("high").and_then(Value::as_u64);
+            if let (Some(low), Some(high)) = (low_val, high_val) {
+                // Reconstruct the 64-bit value: high << 32 | low
+                let timestamp = (high << 32) | (low & 0xFFFF_FFFF);
+                *value = Value::String(timestamp.to_string());
+            }
+        }
+        // String or other types: nothing to do
+        _ => {}
+    }
+}
+
 #[derive(Debug, Clone, Copy, PartialEq)]
 pub enum OtlpEncoding {
     Protobuf,
@@ -46,7 +111,11 @@ impl Processor {
         encoding: OtlpEncoding,
     ) -> Result<Vec<Vec<DatadogSpan>>, Box<dyn Error>> {
         let request = match encoding {
-            OtlpEncoding::Json => serde_json::from_slice::<ExportTraceServiceRequest>(body)?,
+            OtlpEncoding::Json => {
+                let mut json_value: Value = serde_json::from_slice(body)?;
+                normalize_timestamps(&mut json_value);
+                serde_json::from_value::<ExportTraceServiceRequest>(json_value)?
+            }
             OtlpEncoding::Protobuf => ExportTraceServiceRequest::decode(body)?,
         };
 
@@ -61,3 +130,66 @@ impl Processor {
         Ok(spans)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    #[test]
+    fn test_integer_timestamp_converted_to_string() {
+        let mut value = json!({"startTimeUnixNano": 1_581_452_772_000_000_321_u64});
+        normalize_timestamps(&mut value);
+        assert_eq!(value["startTimeUnixNano"], json!("1581452772000000321"));
+    }
+
+    #[test]
+    fn test_split_object_timestamp_reconstructed() {
+        // Some old JS SDKs send 64-bit ints as {"low": u32, "high": u32}
+        let mut value =
+            json!({"startTimeUnixNano": {"low": 1_029_784_000_u64, "high": 395_146_000_u64}});
+        normalize_timestamps(&mut value);
+        let expected = (395_146_000_u64 << 32) | 1_029_784_000_u64;
+        assert_eq!(value["startTimeUnixNano"], json!(expected.to_string()));
+    }
+
+    #[test]
+    fn test_non_timestamp_integers_unchanged() {
+        // Verify we only convert timestamp fields, not all integers
+        let mut value = json!({
+            "kind": 1,
+            "droppedAttributesCount": 5,
+            "attributes": [{"value": {"intValue": 42}}],
+            "startTimeUnixNano": 12345_u64
+        });
+        normalize_timestamps(&mut value);
+
+        // These should remain as integers
+        assert_eq!(value["kind"], json!(1));
+        assert_eq!(value["droppedAttributesCount"], json!(5));
+        assert_eq!(value["attributes"][0]["value"]["intValue"], json!(42));
+        // Only this should be converted
+        assert_eq!(value["startTimeUnixNano"], json!("12345"));
+    }
+
+    #[test]
+    fn test_nested_event_timestamps_normalized() {
+        let mut value = json!({
+            "resourceSpans": [{
+                "scopeSpans": [{
+                    "spans": [{
+                        "startTimeUnixNano": 100_u64,
+                        "endTimeUnixNano": "200",
+                        "events": [{"timeUnixNano": 300_u64}]
+                    }]
+                }]
+            }]
+        });
+        normalize_timestamps(&mut value);
+
+        let span = &value["resourceSpans"][0]["scopeSpans"][0]["spans"][0];
+        assert_eq!(span["startTimeUnixNano"], json!("100"));
+        assert_eq!(span["endTimeUnixNano"], json!("200")); // Already string
+        assert_eq!(span["events"][0]["timeUnixNano"], json!("300"));
+    }
+}