Restrict ClickHouse pipeline to log events that are within a 72 window (#2986)

* restrict clickhouse pipeline to log events that are within a 72 window

* add logging to indicate number of events dropped due to old timestamps

Author: amokan

lib/logflare/backends/adaptor/clickhouse_adaptor/pipeline.ex

@@ -24,6 +24,9 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.Pipeline do
   @batch_size 1_500
   @max_retries 1
 
+  # 72 hour max event age, based on timestamp
+  @max_event_age_us 72 * 3_600 * 1_000_000
+
   @doc false
   def max_retries, do: @max_retries
 
@@ -91,6 +94,8 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.Pipeline do
       }
     )
 
+    message_count = length(messages)
+
     result =
       OpenTelemetry.Tracer.with_span :clickhouse_pipeline, %{
         attributes: %{
@@ -103,7 +108,22 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.Pipeline do
       } do
         source = Sources.Cache.get_by_id(source_id)
         backend = Backends.Cache.get_backend(backend_id)
-        events = for %{data: le} <- messages, do: le
+        cutoff_us = System.system_time(:microsecond) - @max_event_age_us
+
+        events =
+          for %{data: le} <- messages,
+              le.body["timestamp"] >= cutoff_us,
+              do: le
+
+        event_count = length(events)
+
+        if event_count < message_count do
+          Logger.warning(
+            "Dropping #{message_count - event_count} of #{message_count} ClickHouse event(s) older than 72 hours",
+            source_token: source_token,
+            backend_id: backend_id
+          )
+        end
 
         ClickHouseAdaptor.insert_log_events({source, backend}, events)
       end

test/logflare/backends/adaptor/clickhouse_adaptor/pipeline_test.exs

@@ -115,6 +115,45 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.PipelineTest do
       assert result == []
     end
 
+    test "filters out events older than 72 hours", %{
+      context: context,
+      source: source,
+      backend: backend
+    } do
+      old_timestamp = System.system_time(:microsecond) - 73 * 3_600 * 1_000_000
+      recent_timestamp = System.system_time(:microsecond) - 1 * 3_600 * 1_000_000
+
+      old_event =
+        build(:log_event, source: source, message: "Old event", timestamp: old_timestamp)
+
+      recent_event =
+        build(:log_event, source: source, message: "Recent event", timestamp: recent_timestamp)
+
+      messages = [
+        %Message{data: old_event, acknowledger: {Pipeline, :ack_id, :ack_data}},
+        %Message{data: recent_event, acknowledger: {Pipeline, :ack_id, :ack_data}}
+      ]
+
+      result = Pipeline.handle_batch(:ch, messages, %{size: 2, trigger: :flush}, context)
+
+      assert result == messages
+
+      Process.sleep(200)
+
+      table_name = ClickHouseAdaptor.clickhouse_ingest_table_name(source)
+
+      {:ok, query_result} =
+        ClickHouseAdaptor.execute_ch_query(
+          backend,
+          "SELECT body FROM #{table_name}"
+        )
+
+      assert length(query_result) == 1
+      [row] = query_result
+      body = Jason.decode!(row["body"])
+      assert body["event_message"] == "Recent event"
+    end
+
     test "handles log events with different field types", %{
       context: context,
       source: source,