feat(ourlogs): Always set observed time for logs (#4559)

### Summary
Observed time nanos should be set by the collector in OTel. In the case
of Sentry the collector will always be Relay. We may want to allow
passing observed time nanos in the future, but we can address that after
we move to using EAP items consumers since they rely on 'received'
instead of observed time.

Author: k-fish

CHANGELOG.md

@@ -27,6 +27,7 @@
 - Update mapping of OTLP spans to Sentry spans in the experimental OTL traces endpoint. ([#4505](https://github.com/getsentry/relay/pull/4505))
 - Expose metrics for the `AsyncPool`. ([#4538](https://github.com/getsentry/relay/pull/4538))
 - Expose service utilization metrics through the internal relay metric endpoint. ([#4543](https://github.com/getsentry/relay/pull/4543))
+- Always set observed time for OTel logs in Relay. ([#4559](https://github.com/getsentry/relay/pull/4559))
 
 ## 25.2.0
 

Cargo.lock

@@ -109,6 +109,11 @@ impl UnixTimestamp {
         self.0
     }
 
+    /// Returns the number of nanoseconds since the UNIX epoch start. Precision limited to seconds.
+    pub fn as_nanos(self) -> u64 {
+        self.as_secs() * 1_000_000_000
+    }
+
     /// Returns the timestamp as chrono datetime.
     pub fn as_datetime(self) -> Option<DateTime<Utc>> {
         DateTime::from_timestamp(self.0 as i64, 0)

relay-common/src/time.rs

@@ -23,6 +23,7 @@ opentelemetry-proto = { workspace = true, features = [
 ] }
 relay-event-schema = { workspace = true }
 relay-protocol = { workspace = true }
+relay-common = { workspace = true }
 serde_json = { workspace = true }
 
 [dev-dependencies]

relay-ourlogs/Cargo.toml

@@ -1,6 +1,7 @@
 use opentelemetry_proto::tonic::common::v1::any_value::Value as OtelValue;
 
 use crate::OtelLog;
+use relay_common::time::UnixTimestamp;
 use relay_event_schema::protocol::{AttributeValue, OurLog, SpanId, TraceId};
 use relay_protocol::{Annotated, Object};
 
@@ -29,6 +30,10 @@ pub fn otel_to_sentry_log(otel_log: OtelLog) -> OurLog {
 
     let mut attribute_data = Object::new();
 
+    // We ignore the passed observed time since Relay always acts as the collector in Sentry.
+    // We may change this in the future with forwarding Relays.
+    let observed_time_unix_nano = UnixTimestamp::now().as_nanos();
+
     for attribute in attributes.into_iter() {
         if let Some(value) = attribute.value.and_then(|v| v.value) {
             let key = attribute.key;
@@ -58,7 +63,7 @@ pub fn otel_to_sentry_log(otel_log: OtelLog) -> OurLog {
 
     OurLog {
         timestamp_nanos: Annotated::new(otel_log.time_unix_nano),
-        observed_timestamp_nanos: Annotated::new(otel_log.observed_time_unix_nano),
+        observed_timestamp_nanos: Annotated::new(observed_time_unix_nano),
         trace_id: TraceId(trace_id).into(),
         span_id: Annotated::new(SpanId(span_id)),
         trace_flags: Annotated::new(0),
@@ -201,4 +206,61 @@ mod tests {
             Some(&"SELECT \"table\".\"col\" FROM \"table\" WHERE \"table\".\"col\" = %s".into())
         );
     }
+
+    #[test]
+    fn parse_log_without_observed_time() {
+        let json_without_observed_time = r#"{
+            "timeUnixNano": "1544712660300000000",
+            "observedTimeUnixNano": "0",
+            "severityNumber": 10,
+            "severityText": "Information",
+            "traceId": "5B8EFFF798038103D269B633813FC60C",
+            "spanId": "EEE19B7EC3C1B174",
+            "body": {
+                "stringValue": "Example log record"
+            },
+            "attributes": []
+        }"#;
+
+        let before_test = UnixTimestamp::now().as_nanos();
+        let otel_log: OtelLog = serde_json::from_str(json_without_observed_time).unwrap();
+        let our_log: OurLog = otel_to_sentry_log(otel_log);
+        let after_test = UnixTimestamp::now().as_nanos();
+
+        let observed_time = our_log.observed_timestamp_nanos.value().unwrap();
+        assert!(*observed_time > 0);
+        assert!(*observed_time >= before_test);
+        assert!(*observed_time <= after_test);
+    }
+
+    #[test]
+    fn parse_log_ignores_observed_time() {
+        let json_with_observed_time = r#"{
+            "timeUnixNano": "1544712660300000000",
+            "observedTimeUnixNano": "1544712660300000000",
+            "severityNumber": 10,
+            "severityText": "Information",
+            "traceId": "5B8EFFF798038103D269B633813FC60C",
+            "spanId": "EEE19B7EC3C1B174",
+            "body": {
+                "stringValue": "Example log record"
+            },
+            "attributes": []
+        }"#;
+
+        let before_test = UnixTimestamp::now().as_nanos();
+        let otel_log: OtelLog = serde_json::from_str(json_with_observed_time).unwrap();
+        let our_log: OurLog = otel_to_sentry_log(otel_log);
+        let after_test = UnixTimestamp::now().as_nanos();
+
+        let observed_time = our_log.observed_timestamp_nanos.value().unwrap();
+        assert!(*observed_time > 0);
+        assert!(*observed_time >= before_test);
+        assert!(*observed_time <= after_test);
+
+        assert_ne!(
+            our_log.observed_timestamp_nanos,
+            Annotated::new(1544712660300000000)
+        );
+    }
 }