perf: reduce broadway memory footprint

cloudbuild/staging/deploy.yaml

@@ -43,7 +43,7 @@ steps:
       - --type=proactive
       - --max-surge=1
       - --max-unavailable=0
-      - --min-ready=60
+      - --min-ready=180
       - --minimal-action=replace
       - --most-disruptive-allowed-action=replace
       - --replacement-method=substitute

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

@@ -15,7 +15,9 @@ defmodule Logflare.Backends.Adaptor.PostgresAdaptor.Pipeline do
     Broadway.start_link(__MODULE__,
       name: adaptor_state.pipeline_name,
       producer: [
-        module: {BufferProducer, [source: adaptor_state.source, backend: adaptor_state.backend]},
+        module:
+          {BufferProducer,
+           [source_id: adaptor_state.source.id, backend_id: adaptor_state.backend.id]},
         transformer: {__MODULE__, :transform, []},
         concurrency: 1
       ],

lib/logflare/backends/adaptor/webhook_adaptor.ex

@@ -105,8 +105,8 @@ defmodule Logflare.Backends.Adaptor.WebhookAdaptor do
           module:
             {BufferProducer,
              [
-               backend: args.backend,
-               source: args.source
+               backend_id: Map.get(args.backend || %{}, :id),
+               source_id: args.source.id
              ]},
           transformer: {__MODULE__, :transform, []},
           concurrency: 1

lib/logflare/backends/buffer_producer.ex

@@ -15,12 +15,14 @@ defmodule Logflare.Backends.BufferProducer do
 
   @impl GenStage
   def init(opts) do
+    source = Sources.Cache.get_by_id(opts[:source_id])
+
     state = %{
       demand: 0,
       # TODO: broadcast by id instead.
-      source_id: opts[:source].id,
-      source_token: opts[:source].token,
-      backend_id: Map.get(opts[:backend] || %{}, :id),
+      source_id: opts[:source_id],
+      source_token: source.token,
+      backend_id: opts[:backend_id],
       # discard logging backoff
       last_discard_log_dt: nil,
       interval: Keyword.get(opts, :interval, @default_interval)

lib/logflare/source/bigquery/pipeline.ex

@@ -45,8 +45,8 @@ defmodule Logflare.Source.BigQuery.Pipeline do
             module:
               {BufferProducer,
                [
-                 source: source,
-                 backend: backend
+                 source_id: source.id,
+                 backend_id: backend.id
                ]},
             transformer: {__MODULE__, :transform, []}
           ],

test/logflare/backends/buffer_producer_test.exs

@@ -16,7 +16,10 @@ defmodule Logflare.Backends.BufferProducerTest do
     source = insert(:source, user: user)
 
     le = build(:log_event, source: source)
-    buffer_producer_pid = start_supervised!({BufferProducer, backend: nil, source: source})
+
+    buffer_producer_pid =
+      start_supervised!({BufferProducer, backend_id: nil, source_id: source.id})
+
     sid_bid_pid = {source.id, nil, buffer_producer_pid}
     :timer.sleep(100)
     :ok = IngestEventQueue.add_to_table(sid_bid_pid, [le])
@@ -37,7 +40,10 @@ defmodule Logflare.Backends.BufferProducerTest do
     startup_key = {source.id, nil, nil}
     IngestEventQueue.upsert_tid(startup_key)
     :ok = IngestEventQueue.add_to_table(startup_key, [le])
-    buffer_producer_pid = start_supervised!({BufferProducer, backend: nil, source: source})
+
+    buffer_producer_pid =
+      start_supervised!({BufferProducer, backend_id: nil, source_id: source.id})
+
     sid_bid_pid = {source.id, nil, buffer_producer_pid}
 
     GenStage.stream([{buffer_producer_pid, max_demand: 1}])
@@ -53,7 +59,7 @@ defmodule Logflare.Backends.BufferProducerTest do
     source = insert(:source, user: user)
 
     pid =
-      start_supervised!({BufferProducer, backend: nil, source: source, buffer_size: 10})
+      start_supervised!({BufferProducer, backend_id: nil, source_id: source.id, buffer_size: 10})
 
     le = build(:log_event)
     items = for _ <- 1..100, do: le

test/logflare/backends/loki_adaptor_test.exs

@@ -52,6 +52,7 @@ defmodule Logflare.Backends.Adaptor.LokiAdaptorTest do
 
   describe "logs ingestion" do
     setup do
+      insert(:plan)
       user = insert(:user)
       source = insert(:source, user: user)