perf: add benchmarks covering http ingestion and context cache (#2273)

* perf: add benchmarks covering http ingestion and context cache

* chore: formatting

* perf: bandit (#2275)

* Revert "Revert "perf: bandit adaptor (#2204)""

This reverts commit 3a2fcba.

* perf: add bandit benchmark

* chore: verison bump to v1.10.6

VERSION

@@ -1 +1 @@
-1.10.5
+1.10.6

config/config.exs

@@ -28,6 +28,23 @@ config :logflare, Logflare.Source.BigQuery.Schema, updates_per_minute: 6
 
 # Configures the endpoint
 config :logflare, LogflareWeb.Endpoint,
+  adapter: Bandit.PhoenixAdapter,
+  http: [
+    thousand_island_options: [
+      # https://cloud.google.com/load-balancing/docs/https/#timeouts_and_retries
+      # preserves idle keepalive connections up to load balancer max of 600s
+      read_timeout: 600_000,
+      # transport options are passed wholly to :gen_tcp
+      # https://github.com/mtrudel/thousand_island/blob/ae733332892b1bb2482a9cf4e97de03411fba2ad/lib/thousand_island/transports/tcp.ex#L61
+      transport_options: [
+        # https://www.erlang.org/doc/man/inet
+        # both reuseport and reuseport_lb should be provided for linux
+        reuseport: true,
+        reuseport_lb: true
+        #
+      ]
+    ]
+  ],
   url: [host: "localhost", scheme: "http", port: 4000],
   secret_key_base: "DSzZYeAgGaXlfRXPQqMOPiA8hJOYSImhnR2lO8lREOE2vWDmkGn1XWHxoCZoASlP",
   render_errors: [view: LogflareWeb.ErrorView, accepts: ~w(html json)],

config/dev.exs

@@ -7,14 +7,7 @@ config :logflare,
 config :logflare, LogflareWeb.Endpoint,
   server: true,
   http: [
-    port: System.get_env("PORT") || 4000,
-    transport_options: [
-      max_connections: 16_384,
-      num_acceptors: 100,
-      socket_opts: [{:raw, 0xFFFF, 0x0200, <<1::32-native>>}]
-    ],
-    protocol_options: [max_keepalive: 1_000],
-    compress: true
+    port: System.get_env("PORT") || 4000
   ],
   live_view: [
     signing_salt: "eVpFFmpN+OHPrilThexLilWnF+a8zBLbCtdH/OzAayShcm1B3OHOyGiadM6qOezp"

config/prod.exs

@@ -5,24 +5,7 @@ config :logflare, env: :prod
 
 config :logflare, LogflareWeb.Endpoint,
   http: [
-    port: 4000,
-    transport_options: [
-      max_connections: 64_000,
-      num_acceptors: 1_000,
-      # for so_reuseport
-      socket_opts: [{:raw, 1, 15, <<1::32-native>>}]
-    ],
-    # https://blog.percy.io/tuning-nginx-behind-google-cloud-platform-http-s-load-balancer-305982ddb340
-    # https://github.com/ninenines/cowboy/issues/1286#issuecomment-699643478
-    protocol_options: [
-      # https://ninenines.eu/docs/en/cowboy/2.8/manual/cowboy_http/
-      request_timeout: 30_000,
-      # https://cloud.google.com/load-balancing/docs/https/#timeouts_and_retries
-      # must be greater than 600s
-      idle_timeout: 650_000,
-      max_keepalive: :infinity
-    ],
-    compress: true
+    port: 4000
   ],
   cache_static_manifest: "priv/static/cache_manifest.json",
   server: true,

mix.exs

@@ -72,13 +72,13 @@ defmodule Logflare.Mixfile do
   defp deps do
     [
       # Phoenix stuff
-      {:phoenix, "~> 1.7"},
+      {:phoenix, "~> 1.7.14"},
       {:phoenix_live_view, "~> 0.18"},
       {:phoenix_view, "~> 2.0"},
       {:phoenix_pubsub, "~> 2.1"},
       {:phoenix_ecto, "~> 4.4"},
       {:phoenix_live_reload, "~> 1.4", only: :dev},
-      {:bandit, ">= 0.7.7"},
+      {:bandit, "~> 1.5.7"},
       {:plug_crypto, "~> 1.2.2"},
       {:cors_plug, "~> 2.0"},
       {:plug_caisson, "~> 0.2.1"},

test/profiling/context_cache.exs

@@ -0,0 +1,46 @@
+alias Logflare.Sources
+alias Logflare.Users
+require Phoenix.ConnTest
+Mimic.copy(Broadway)
+Mimic.copy(Logflare.Backends)
+Mimic.copy(Logflare.Logs)
+Mimic.copy(Logflare.Partners)
+
+Mimic.stub(Logflare.Backends, :ingest_logs, fn _, _ -> :ok end)
+Mimic.stub(Logflare.Logs, :ingest_logs, fn _, _ -> :ok end)
+# Mimic.stub(Broadway, :push_messages, fn _, _ -> :ok end)
+ver = System.argv() |> Enum.at(0)
+
+v1_source = Sources.get(:"9f37d86e-e4fa-4ef2-a47e-e8d4ac1fceba")
+
+v2_source = Sources.get(:"94d07aab-30f5-460e-8871-eb85f4674e35")
+
+user = Users.get(v1_source.user_id)
+str_key = inspect({:get, [user.id]})
+
+Benchee.run(
+  %{
+    "Cachex.get/3" => fn _ ->
+      Cachex.get(Logflare.Users.Cache, {:get, [user.id]})
+    end,
+    "Logflare.Users.get/1" => fn _ ->
+      Logflare.Users.Cache.get(user.id)
+    end
+  },
+  before_scenario: fn input ->
+    Cachex.put(Logflare.Users.Cache, {:get, [user.id]}, {:cached, user})
+    Cachex.put(Logflare.Users.Cache, str_key, {:cached, user})
+    {input, nil}
+  end,
+  time: 4,
+  memory_time: 0
+)
+
+# Before 090f8d93:
+# Name                               ips        average  deviation         median         99th %
+# Cachex.get/3                    1.40 M      714.86 ns  ±2833.04%         625 ns         833 ns
+# ContextCache.apply_fun/3        1.03 M      968.97 ns  ±1686.53%         792 ns        1167 ns
+
+# Comparison:
+# Cachex.get/3                    1.40 M
+# ContextCache.apply_fun/3        1.03 M - 1.36x slower +254.10 ns

test/profiling/context_cache_cold_start.exs

@@ -0,0 +1,54 @@
+alias Logflare.Sources
+alias Logflare.Users
+require Phoenix.ConnTest
+Mimic.copy(Broadway)
+Mimic.copy(Logflare.Backends)
+Mimic.copy(Logflare.Logs)
+Mimic.copy(Logflare.Partners)
+
+Mimic.stub(Logflare.Backends, :ingest_logs, fn _, _ -> :ok end)
+Mimic.stub(Logflare.Logs, :ingest_logs, fn _, _ -> :ok end)
+# Mimic.stub(Broadway, :push_messages, fn _, _ -> :ok end)
+ver = System.argv() |> Enum.at(0)
+
+v1_source = Sources.get(:"9f37d86e-e4fa-4ef2-a47e-e8d4ac1fceba")
+
+v2_source = Sources.get(:"94d07aab-30f5-460e-8871-eb85f4674e35")
+
+user = Users.get(v1_source.user_id)
+
+apply_args = {:get, [user.id]}
+
+Benchee.run(
+  %{
+    "Cachex.fetch/3" => fn ->
+      Cachex.fetch(Logflare.Users.Cache, apply_args, fn {fun, args} ->
+        value = apply(Logflare.Users, :get, [user.id])
+
+        # keys_key = {{Logflare.Users, Logflare.ContextCache.select_key(value)}, :erlang.phash2(apply_args)}
+        # Cachex.put(Logflare.ContextCache, keys_key, apply_args)
+
+        {:commit, {:cached, value}}
+      end)
+    end,
+    "Users.Cache.get/1" => fn ->
+      Logflare.Users.Cache.get(user.id)
+    end
+  },
+  before_each: fn input ->
+    Cachex.clear(Logflare.Users.Cache)
+    Cachex.clear(Logflare.ContextCache)
+    input
+  end,
+  time: 4,
+  memory_time: 0
+)
+
+# Before 090f8d93:
+# Name                        ips        average  deviation         median         99th %
+# Users.Cache.get/1        1.67 K      598.09 μs    ±69.70%      546.42 μs     1175.40 μs
+# Cachex.fetch/3           1.41 K      707.75 μs    ±38.63%      664.25 μs     1510.41 μs
+
+# Comparison:
+# Users.Cache.get/1        1.67 K
+# Cachex.fetch/3           1.41 K - 1.18x slower +109.67 μs

test/profiling/ingest_http.exs

@@ -0,0 +1,60 @@
+alias Logflare.Sources
+alias Logflare.Users
+require Phoenix.ConnTest
+Mimic.copy(Broadway)
+Mimic.copy(Logflare.Backends)
+Mimic.copy(Logflare.Logs)
+Mimic.copy(Logflare.Partners)
+
+Mimic.stub(Logflare.Backends, :ingest_logs, fn _, _ -> :ok end)
+Mimic.stub(Logflare.Logs, :ingest_logs, fn _, _ -> :ok end)
+# Mimic.stub(Broadway, :push_messages, fn _, _ -> :ok end)
+ver = System.argv() |> Enum.at(0)
+
+v1_source = Sources.get(:"9f37d86e-e4fa-4ef2-a47e-e8d4ac1fceba")
+
+v2_source = Sources.get(:"94d07aab-30f5-460e-8871-eb85f4674e35")
+
+user = Users.get(v1_source.user_id)
+
+Benchee.run(
+  %{
+    "pipeline" => fn input ->
+      Phoenix.ConnTest.build_conn()
+      |> Phoenix.ConnTest.dispatch(
+        LogflareWeb.Endpoint,
+        :post,
+        "/api/logs?source=#{input.token}&api_key=#{user.api_key}",
+        %{
+          message: "some msg",
+          field: "1234",
+          testing: 123
+        }
+      )
+    end
+  },
+  inputs: %{
+    "v1" => v1_source,
+    "v2" => v2_source
+  },
+  time: 10,
+  memory_time: 0
+)
+
+# Before 090f8d93:
+# ##### With input v1 #####
+# Name               ips        average  deviation         median         99th %
+# pipeline         36.05       27.74 ms    ±20.18%       26.72 ms       47.35 ms
+
+# ##### With input v2 #####
+# Name               ips        average  deviation         median         99th %
+# pipeline         37.07       26.98 ms    ±21.35%       25.97 ms       55.37 ms
+
+# With bandit (2024-12-10):
+# ##### With input v1 #####
+# Name               ips        average  deviation         median         99th %
+# pipeline         44.25       22.60 ms    ±14.53%       21.89 ms       34.32 ms
+
+# ##### With input v2 #####
+# Name               ips        average  deviation         median         99th %
+# pipeline         45.08       22.18 ms    ±11.26%       22.16 ms       29.60 ms

test/profiling/ingest_http_plugs.exs

@@ -0,0 +1,135 @@
+alias Logflare.Sources
+alias Logflare.Users
+alias Logflare.Partners
+require Phoenix.ConnTest
+Mimic.copy(Broadway)
+Mimic.copy(Logflare.Backends)
+Mimic.copy(Logflare.Logs)
+Mimic.copy(Logflare.Partners)
+
+Mimic.stub(Logflare.Backends, :ingest_logs, fn _, _ -> :ok end)
+Mimic.stub(Logflare.Logs, :ingest_logs, fn _, _ -> :ok end)
+# Mimic.stub(Broadway, :push_messages, fn _, _ -> :ok end)
+ver = System.argv() |> Enum.at(0)
+
+v1_source = Sources.get(:"9f37d86e-e4fa-4ef2-a47e-e8d4ac1fceba")
+
+v2_source = Sources.get(:"94d07aab-30f5-460e-8871-eb85f4674e35")
+
+user = Users.get(v1_source.user_id)
+
+Benchee.run(
+  %{
+    "VerifyApiAccess" =>
+      {fn {source, conn} ->
+         LogflareWeb.Plugs.VerifyApiAccess.call(conn, scopes: ~w(public))
+       end,
+       before_scenario: fn {source, conn} ->
+         {source, conn}
+       end},
+    "FetchResource" =>
+      {fn {source, conn} ->
+         LogflareWeb.Plugs.FetchResource.call(conn, [])
+       end,
+       before_scenario: fn {source, conn} ->
+         conn =
+           conn
+           |> LogflareWeb.Plugs.VerifyApiAccess.call(scopes: ~w(public))
+
+         {source, conn}
+       end},
+    "VerifyResourceOwnership" =>
+      {fn {source, conn} ->
+         LogflareWeb.Plugs.VerifyResourceOwnership.call(conn, [])
+       end,
+       before_scenario: fn {source, conn} ->
+         conn =
+           conn
+           |> LogflareWeb.Plugs.VerifyApiAccess.call(scopes: ~w(public))
+           |> LogflareWeb.Plugs.FetchResource.call([])
+
+         {source, conn}
+       end},
+    "SetPlanFromCache" =>
+      {fn {source, conn} ->
+         LogflareWeb.Plugs.SetPlanFromCache.call(conn, [])
+       end,
+       before_scenario: fn {source, conn} ->
+         conn =
+           conn
+           |> LogflareWeb.Plugs.VerifyApiAccess.call(scopes: ~w(public))
+           |> LogflareWeb.Plugs.FetchResource.call([])
+           |> LogflareWeb.Plugs.VerifyResourceOwnership.call([])
+
+         {source, conn}
+       end},
+    "RateLimiter" =>
+      {fn {source, conn} ->
+         LogflareWeb.Plugs.RateLimiter.call(conn, [])
+       end,
+       before_scenario: fn {source, conn} ->
+         conn =
+           conn
+           |> LogflareWeb.Plugs.VerifyApiAccess.call(scopes: ~w(public))
+           |> LogflareWeb.Plugs.FetchResource.call([])
+           |> LogflareWeb.Plugs.VerifyResourceOwnership.call([])
+           |> LogflareWeb.Plugs.SetPlanFromCache.call([])
+
+         {source, conn}
+       end},
+    "BufferLimiter" =>
+      {fn {source, conn} ->
+         LogflareWeb.Plugs.BufferLimiter.call(conn, [])
+       end,
+       before_scenario: fn {source, conn} ->
+         conn =
+           conn
+           |> LogflareWeb.Plugs.VerifyApiAccess.call(scopes: ~w(public))
+           |> LogflareWeb.Plugs.FetchResource.call([])
+           |> LogflareWeb.Plugs.VerifyResourceOwnership.call([])
+           |> LogflareWeb.Plugs.SetPlanFromCache.call([])
+           |> LogflareWeb.Plugs.RateLimiter.call([])
+
+         {source, conn}
+       end}
+  },
+  inputs: %{
+    "v1" => v1_source
+    # "v2" => v2_source
+  },
+  before_scenario: fn input ->
+    prepared_conn =
+      Phoenix.ConnTest.build_conn(
+        :post,
+        "/api/logs?source=#{input.token}&api_key=#{user.api_key}",
+        %{
+          message: "some msg",
+          field: "1234",
+          testing: 123
+        }
+      )
+      |> Plug.Conn.assign(:resource_type, :source)
+
+    {input, prepared_conn}
+  end,
+  time: 4,
+  memory_time: 0
+)
+
+# Before 090f8d93:
+# ##### With input v1 #####
+# Name                              ips        average  deviation         median         99th %
+# VerifyResourceOwnership    14120.42 K      0.0708 μs ±25384.39%      0.0830 μs      0.0840 μs
+# FetchResource                668.87 K        1.50 μs  ±1270.86%        1.33 μs        2.58 μs
+# BufferLimiter                332.39 K        3.01 μs   ±428.79%        2.79 μs        4.67 μs
+# VerifyApiAccess              111.55 K        8.96 μs   ±168.32%        7.46 μs       27.17 μs
+# RateLimiter                   50.82 K       19.68 μs    ±23.81%       19.17 μs       26.04 μs
+# SetPlanFromCache              36.09 K       27.71 μs    ±22.85%       25.29 μs       40.29 μs
+
+# Comparison:
+# VerifyResourceOwnership    14120.42 K
+# FetchResource                668.87 K - 21.11x slower +1.42 μs
+# BufferLimiter                332.39 K - 42.48x slower +2.94 μs
+# VerifyApiAccess              111.55 K - 126.58x slower +8.89 μs
+# RateLimiter                   50.82 K - 277.83x slower +19.61 μs
+# SetPlanFromCache              36.09 K - 391.23x slower +27.64 μs