fix(loki sinks): Fix loki event timestamp out of range panic

src/internal_events/loki.rs

@@ -71,3 +71,18 @@ impl InternalEvent for LokiOutOfOrderEventRewritten {
         counter!("rewritten_timestamp_events_total", self.count as u64);
     }
 }
+
+#[derive(Debug)]
+pub struct LokiEventTimestampOutOfRangeError;
+
+impl InternalEvent for LokiEventTimestampOutOfRangeError {
+    fn emit(self) {
+        error!(
+            message = "Event timestamp out of range.",
+            reason = "out_of_range",
+            internal_log_rate_limit = true,
+        );
+
+        counter!("timestamp_out_of_range_total", 1,);
+    }
+}

src/sinks/loki/sink.rs

@@ -16,8 +16,8 @@ use crate::sinks::loki::event::LokiBatchEncoding;
 use crate::{
     http::{get_http_scheme_from_uri, HttpClient},
     internal_events::{
-        LokiEventUnlabeledError, LokiOutOfOrderEventDroppedError, LokiOutOfOrderEventRewritten,
-        SinkRequestBuildError,
+        LokiEventTimestampOutOfRangeError, LokiEventUnlabeledError,
+        LokiOutOfOrderEventDroppedError, LokiOutOfOrderEventRewritten, SinkRequestBuildError,
     },
     sinks::prelude::*,
 };
@@ -253,7 +253,13 @@ 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 => {
+                    emit!(LokiEventTimestampOutOfRangeError);
+                    return None;
+                }
+            },
             _ => chrono::Utc::now()
                 .timestamp_nanos_opt()
                 .expect("Timestamp out of range"),

src/sinks/loki/tests.rs

@@ -146,3 +146,32 @@ 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");
+    use vector_lib::config::log_schema;
+    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());
+}