LOG-7417: Fix loki event timestamp out of range panic

changelog.d/20763_loki_sink_drop_invalid_timestamp_event.fix.md

@@ -0,0 +1,3 @@
+Loki sink now drops events with non-parsable timestamps.
+
+authors: suikammd

src/internal_events/loki.rs

@@ -71,3 +71,29 @@ impl InternalEvent for LokiOutOfOrderEventRewritten {
         counter!("rewritten_timestamp_events_total", self.count as u64);
     }
 }
+
+#[derive(Debug)]
+pub struct LokiTimestampNonParsableEventsDropped;
+
+impl InternalEvent for LokiTimestampNonParsableEventsDropped {
+    fn emit(self) {
+        let reason = "Dropping timestamp non-parsable event(s).";
+
+        error!(
+            message = "Event timestamp non-parsable.",
+            error_code = "non-parsable_timestamp",
+            error_type = error_type::CONDITION_FAILED,
+            stage = error_stage::PROCESSING,
+            internal_log_rate_limit = true,
+        );
+
+        emit!(ComponentEventsDropped::<INTENTIONAL> { count: 1, reason });
+
+        counter!(
+            "component_errors_total", 1,
+            "error_code" => "non-parsable_timestamp",
+            "error_type" => error_type::CONDITION_FAILED,
+            "stage" => error_stage::PROCESSING,
+        );
+    }
+}

src/sinks/loki/sink.rs

@@ -17,7 +17,7 @@ use crate::{
     http::{get_http_scheme_from_uri, HttpClient},
     internal_events::{
         LokiEventUnlabeledError, LokiOutOfOrderEventDroppedError, LokiOutOfOrderEventRewritten,
-        SinkRequestBuildError,
+        LokiTimestampNonParsableEventsDropped, SinkRequestBuildError,
     },
     sinks::prelude::*,
 };
@@ -253,7 +253,14 @@ impl EventEncoder {
         self.remove_label_fields(&mut event);
 
         let timestamp = match event.as_log().get_timestamp() {
-            Some(Value::Timestamp(ts)) => ts.timestamp_nanos_opt().expect("Timestamp out of range"),
+            Some(Value::Timestamp(ts)) => match ts.timestamp_nanos_opt() {
+                Some(timestamp) => timestamp,
+                None => {
+                    finalizers.update_status(EventStatus::Errored);
+                    emit!(LokiTimestampNonParsableEventsDropped);
+                    return None;
+                }
+            },
             _ => chrono::Utc::now()
                 .timestamp_nanos_opt()
                 .expect("Timestamp out of range"),

src/sinks/loki/tests.rs

@@ -7,6 +7,7 @@ use crate::{
     sinks::util::test::{build_test_server, load_sink},
     test_util,
 };
+use vector_lib::config::log_schema;
 
 #[test]
 fn generate_config() {
@@ -146,3 +147,31 @@ async fn healthcheck_grafana_cloud() {
         .await
         .expect("healthcheck failed");
 }
+
+#[tokio::test]
+async fn timestamp_out_of_range() {
+    let (config, cx) = load_sink::<LokiConfig>(
+        r#"
+        endpoint = "http://localhost:3100"
+        labels = {label1 = "{{ foo }}", label2 = "some-static-label", label3 = "{{ foo }}", "{{ foo }}" = "{{ foo }}"}
+        encoding.codec = "json"
+    "#,
+    )
+    .unwrap();
+    let client = config.build_client(cx).unwrap();
+    let mut sink = LokiSink::new(config, client).unwrap();
+
+    let mut e1 = LogEvent::from("hello world");
+    if let Some(timestamp_key) = log_schema().timestamp_key_target_path() {
+        let date = chrono::NaiveDate::from_ymd_opt(1677, 9, 21)
+            .unwrap()
+            .and_hms_nano_opt(0, 12, 43, 145_224_191)
+            .unwrap()
+            .and_local_timezone(chrono::Utc)
+            .unwrap();
+        e1.insert(timestamp_key, date);
+    }
+    let e1 = Event::Log(e1);
+
+    assert!(sink.encoder.encode_event(e1).is_none());
+}