fix(ourlogs): Add observed nanos for parity with OTel logs

CHANGELOG.md

@@ -13,6 +13,7 @@
 
 - Use sentry prefix for browser name/version in logs. ([#4783](https://github.com/getsentry/relay/pull/4783))
 - Do not overcount the number of bytes in logs. ([#4786](https://github.com/getsentry/relay/pull/4786))
+- Record observed time for logs. ([#4795](https://github.com/getsentry/relay/pull/4795))
 
 **Internal**:
 

relay-ourlogs/src/ourlog.rs

@@ -163,6 +163,10 @@ pub fn ourlog_merge_otel(ourlog: &mut Annotated<OurLog>) {
         })
         .unwrap_or_default();
 
+    // This is separate from the sdk provided 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();
+
     attributes.insert(
         "sentry.severity_text".to_owned(),
         Annotated::new(Attribute::new(
@@ -203,7 +207,7 @@ pub fn ourlog_merge_otel(ourlog: &mut Annotated<OurLog>) {
         "sentry.observed_timestamp_nanos".to_owned(),
         Annotated::new(Attribute::new(
             AttributeType::String,
-            Value::String(timestamp_nanos.to_string()),
+            Value::String(observed_time_unix_nano.to_string()),
         )),
     );
     attributes.insert(
@@ -476,8 +480,34 @@ mod tests {
             }
         }"#;
 
+        let before_test = UnixTimestamp::now().as_nanos();
         let mut merged_log = Annotated::<OurLog>::from_json(json).unwrap();
         ourlog_merge_otel(&mut merged_log);
+        let after_test = UnixTimestamp::now().as_nanos();
+
+        // Test the observed timestamp separately
+        let observed_timestamp = merged_log
+            .value()
+            .and_then(|log| log.attribute("sentry.observed_timestamp_nanos"))
+            .and_then(|attr| attr.as_str().and_then(|s| s.parse::<u64>().ok()))
+            .unwrap_or(0);
+
+        assert!(observed_timestamp > 0);
+        assert!(observed_timestamp >= before_test);
+        assert!(observed_timestamp <= after_test);
+
+        // Set observed timestamp to a fixed value for snapshot testing
+        if let Some(log) = merged_log.value_mut() {
+            if let Some(attributes) = log.attributes.value_mut() {
+                attributes.insert(
+                    "sentry.observed_timestamp_nanos".to_owned(),
+                    Annotated::new(Attribute::new(
+                        AttributeType::String,
+                        Value::String("946684800000000000".to_string()),
+                    )),
+                );
+            }
+        }
 
         insta::assert_debug_snapshot!(merged_log, @r###"
         OurLog {
@@ -608,7 +638,34 @@ mod tests {
             body: Annotated::new("somebody".into()),
             ..Default::default()
         });
+
+        let before_test = UnixTimestamp::now().as_nanos();
         ourlog_merge_otel(&mut ourlog);
+        let after_test = UnixTimestamp::now().as_nanos();
+
+        // Test the observed timestamp separately
+        let observed_timestamp = ourlog
+            .value()
+            .and_then(|log| log.attribute("sentry.observed_timestamp_nanos"))
+            .and_then(|attr| attr.as_str().and_then(|s| s.parse::<u64>().ok()))
+            .unwrap_or(0);
+
+        assert!(observed_timestamp > 0);
+        assert!(observed_timestamp >= before_test);
+        assert!(observed_timestamp <= after_test);
+
+        // Set observed timestamp to a fixed value for snapshot testing
+        if let Some(log) = ourlog.value_mut() {
+            if let Some(attributes) = log.attributes.value_mut() {
+                attributes.insert(
+                    "sentry.observed_timestamp_nanos".to_owned(),
+                    Annotated::new(Attribute::new(
+                        AttributeType::String,
+                        Value::String("1638144000000000000".to_string()),
+                    )),
+                );
+            }
+        }
 
         insta::assert_debug_snapshot!(ourlog, @r#"
         OurLog {

tests/integration/test_ourlogs.py

@@ -96,6 +96,10 @@ def test_ourlog_extraction_with_otel_logs(
 
     timestamp_proto.FromSeconds(int(timestamp))
 
+    start_timestamp_nanos = int((start.timestamp() - 1) * 1e9)
+    end = datetime.now(timezone.utc)
+    end_timestamp_nanos = int((end.timestamp() + 1) * 1e9)
+
     expected_logs = [
         MessageToDict(
             TraceItem(
@@ -141,7 +145,13 @@ def test_ourlog_extraction_with_otel_logs(
         expected_log["itemId"] = log["itemId"]
         expected_log["received"] = time_within_delta()
 
-        # This field is set by Relay so we need to remove it
+        observed_timestamp_str = log["attributes"]["sentry.observed_timestamp_nanos"][
+            "stringValue"
+        ]
+        observed_timestamp = int(observed_timestamp_str)
+        assert observed_timestamp > 0
+        assert start_timestamp_nanos < observed_timestamp < end_timestamp_nanos
+
         del log["attributes"]["sentry.observed_timestamp_nanos"]
 
     assert logs == expected_logs
@@ -266,6 +276,10 @@ def test_ourlog_extraction_with_sentry_logs(
 
     relay.send_envelope(project_id, envelope)
 
+    start_timestamp_nanos = int((start.timestamp() - 1) * 1e9)
+    end = datetime.now(timezone.utc)
+    end_timestamp_nanos = int((end.timestamp() + 1) * 1e9)
+
     timestamp_nanos = int(timestamp * 1e6) * 1000
     timestamp_proto = Timestamp()
 
@@ -293,9 +307,6 @@ def test_ourlog_extraction_with_sentry_logs(
                     "sentry.severity_text": AnyValue(string_value="error"),
                     "sentry.span_id": AnyValue(string_value="eee19b7ec3c1b175"),
                     "sentry.trace_flags": AnyValue(int_value=0),
-                    "sentry.observed_timestamp_nanos": AnyValue(
-                        string_value=str(timestamp_nanos)
-                    ),
                     "sentry.timestamp_precise": AnyValue(int_value=timestamp_nanos),
                     "sentry.timestamp_nanos": AnyValue(
                         string_value=str(timestamp_nanos)
@@ -328,9 +339,6 @@ def test_ourlog_extraction_with_sentry_logs(
                     "sentry.severity_text": AnyValue(string_value="info"),
                     "sentry.trace_flags": AnyValue(int_value=0),
                     "sentry.span_id": AnyValue(string_value="eee19b7ec3c1b174"),
-                    "sentry.observed_timestamp_nanos": AnyValue(
-                        string_value=str(timestamp_nanos)
-                    ),
                     "string.attribute": AnyValue(string_value="some string"),
                     "valid_string_with_other": AnyValue(string_value="test"),
                     "sentry.timestamp_precise": AnyValue(int_value=timestamp_nanos),
@@ -348,11 +356,18 @@ def test_ourlog_extraction_with_sentry_logs(
     logs = [MessageToDict(log) for log in ourlogs_consumer.get_ourlogs()]
 
     for log, expected_log in zip(logs, expected_logs):
-        # we can't generate uuid7 with a specific timestamp
-        # in Python just yet so we're overriding it
         expected_log["itemId"] = log["itemId"]
         expected_log["received"] = time_within_delta()
 
+        observed_timestamp_str = log["attributes"]["sentry.observed_timestamp_nanos"][
+            "stringValue"
+        ]
+        observed_timestamp = int(observed_timestamp_str)
+        assert observed_timestamp > 0
+        assert start_timestamp_nanos < observed_timestamp < end_timestamp_nanos
+
+        del log["attributes"]["sentry.observed_timestamp_nanos"]
+
     assert logs == expected_logs
 
     ourlogs_consumer.assert_empty()
@@ -383,6 +398,10 @@ def test_ourlog_extraction_with_sentry_logs_with_missing_fields(
 
     relay.send_envelope(project_id, envelope)
 
+    start_timestamp_nanos = int((start.timestamp() - 1) * 1e9)
+    end = datetime.now(timezone.utc)
+    end_timestamp_nanos = int((end.timestamp() + 1) * 1e9)
+
     timestamp_nanos = int(timestamp * 1e6) * 1000
     timestamp_proto = Timestamp()
 
@@ -405,9 +424,6 @@ def test_ourlog_extraction_with_sentry_logs_with_missing_fields(
                     "sentry.body": AnyValue(string_value="Example log record 2"),
                     "sentry.browser.name": AnyValue(string_value="Python Requests"),
                     "sentry.browser.version": AnyValue(string_value="2.32"),
-                    "sentry.observed_timestamp_nanos": AnyValue(
-                        string_value=str(timestamp_nanos)
-                    ),
                     "sentry.severity_number": AnyValue(int_value=13),
                     "sentry.severity_text": AnyValue(string_value="warn"),
                     "sentry.timestamp_nanos": AnyValue(
@@ -426,11 +442,18 @@ def test_ourlog_extraction_with_sentry_logs_with_missing_fields(
     logs = [MessageToDict(log) for log in ourlogs_consumer.get_ourlogs()]
 
     for log, expected_log in zip(logs, expected_logs):
-        # we can't generate uuid7 with a specific timestamp
-        # in Python just yet so we're overriding it
         expected_log["itemId"] = log["itemId"]
         expected_log["received"] = time_within_delta()
 
+        observed_timestamp_str = log["attributes"]["sentry.observed_timestamp_nanos"][
+            "stringValue"
+        ]
+        observed_timestamp = int(observed_timestamp_str)
+        assert observed_timestamp > 0
+        assert start_timestamp_nanos < observed_timestamp < end_timestamp_nanos
+
+        del log["attributes"]["sentry.observed_timestamp_nanos"]
+
     assert logs == expected_logs
 
     ourlogs_consumer.assert_empty()
@@ -544,6 +567,10 @@ def test_browser_name_version_extraction(
 
     relay.send_envelope(project_id, envelope, headers={"User-Agent": user_agent})
 
+    start_timestamp_nanos = int((start.timestamp() - 1) * 1e9)
+    end = datetime.now(timezone.utc)
+    end_timestamp_nanos = int((end.timestamp() + 1) * 1e9)
+
     timestamp_nanos = int(timestamp * 1e6) * 1000
     timestamp_proto = Timestamp()
 
@@ -571,9 +598,6 @@ def test_browser_name_version_extraction(
                 "sentry.severity_text": AnyValue(string_value="error"),
                 "sentry.span_id": AnyValue(string_value="eee19b7ec3c1b175"),
                 "sentry.trace_flags": AnyValue(int_value=0),
-                "sentry.observed_timestamp_nanos": AnyValue(
-                    string_value=str(timestamp_nanos)
-                ),
                 "sentry.timestamp_precise": AnyValue(int_value=timestamp_nanos),
                 "sentry.timestamp_nanos": AnyValue(string_value=str(timestamp_nanos)),
             },
@@ -593,6 +617,15 @@ def test_browser_name_version_extraction(
     expected_log["itemId"] = log["itemId"]
     expected_log["received"] = time_within_delta()
 
+    observed_timestamp_str = log["attributes"]["sentry.observed_timestamp_nanos"][
+        "stringValue"
+    ]
+    observed_timestamp = int(observed_timestamp_str)
+    assert observed_timestamp > 0
+    assert start_timestamp_nanos < observed_timestamp < end_timestamp_nanos
+
+    del log["attributes"]["sentry.observed_timestamp_nanos"]
+
     assert log == expected_log
 
     ourlogs_consumer.assert_empty()