wip: Periodic refresh of cache warmers (follow-up)

.dialyzer_ignore.exs

@@ -43,7 +43,6 @@
   {"lib/logflare_web/controllers/api/endpoint_controller.ex", :pattern_match},
   {"lib/logflare_web/controllers/billing_controller.ex", :unused_fun},
   {"lib/logflare_web/controllers/billing_controller.ex", :pattern_match_cov},
-  {"lib/logflare_web/controllers/health_check_controller.ex", :pattern_match},
   {"lib/logflare_web/controllers/source_controller.ex", :pattern_match},
   {"lib/logflare_web/live/monaco_editor_component.ex", :pattern_match},
   {"lib/logflare_web/live/query_live.ex", :no_return},

AGENTS.md

@@ -84,8 +84,9 @@ gate to go green - fix the code.
   that have pre-existing findings. They are off so the gate is green on existing
   code. Fixing a backlog check's findings and removing it from that list is a
   welcome change on its own. Never add a check to the list to go green.
-- `mix test.slop` runs `ex_dna --max-clones 29`, a duplication ratchet. A new
-  clone fails CI. When you remove clones, lower the number. Never raise it.
+- `mix test.slop` runs `ex_dna --max-clones <Current clone number>`,
+  a duplication ratchet. A new clone fails CI. When you remove clones,
+  lower the number. Never raise it.
 - `mix test.structure` runs `reach.check --smells` against
   `.reach.baseline.json` (194 accepted findings), so only **new** structural
   smells fail. To accept a

config/runtime.exs

@@ -624,6 +624,14 @@ config :logflare, :context_cache_gossip, %{
   max_nodes: cache_gossip_max_nodes
 }
 
+# LOGFLARE_CACHE_WARMER_REFRESH_ENABLED: Re-run cache warmers every third of the cache TTL
+default_cache_warmer_refresh_enabled = if config_env() == :test, do: "false", else: "true"
+
+config :logflare, Logflare.ContextCache.Warmer,
+  refresh_enabled:
+    System.get_env("LOGFLARE_CACHE_WARMER_REFRESH_ENABLED", default_cache_warmer_refresh_enabled) ==
+      "true"
+
 # LOGFLARE_READ_REPLICAS: PostgreSQL read replicas for selected cache queries.
 # An empty list uses the primary database. Entries are bare host names, IP literals, or URIs
 # whose omitted options inherit the primary.

docs/docs.logflare.com/docs/self-hosting/index.md

@@ -50,6 +50,7 @@ All browser authentication will be disabled when in single-tenant mode.
 | `LOGFLARE_PUBSUB_POOL_SIZE`                    | Integer, defaults to `10`                                               | Sets the number of `Phoenix.PubSub.PG2` partitions to be created. Should be configured to the number of cores of your server for optimal multi-node performance.                                                                                                     |
 | `LOGFLARE_ALERTS_ENABLED`                      | Boolean, defaults to `true`                                             | Flag for enabling and disabling query alerts.                                                                                                                                                                                                                        |
 | `LOGFLARE_MIN_CLUSTER_SIZE`                    | Integer, defaults to `1`                                                | Sets the target cluster size, and emits a warning log periodically if the cluster is below the set number of nodes..                                                                                                                                                 |
+| `LOGFLARE_CACHE_WARMER_REFRESH_ENABLED`        | Boolean, defaults to `true`                                             | Reloads cached sources, rules, users, backends and source schemas of recently active sources every third of their cache lifetime, so they are refreshed before they expire. When `false`, they are only loaded on startup.                                           |
 | `LOGFLARE_OTEL_ENDPOINT`                       | String, defaults to `nil`                                               | Sets the OpenTelemetry Endpoint to send traces to via gRPC. Port number can be included, such as `https://logflare.app:443`                                                                                                                                          |
 | `LOGFLARE_OTEL_SOURCE_UUID`                    | String, defaults to `nil`, optionally required for OpenTelemetry.       | Sets the appropriate header for ingesting OpenTelemetry events into a Logflare source.                                                                                                                                                                               |
 | `LOGFLARE_OTEL_ACCESS_TOKEN`                   | String, defaults to `nil`, optionally required for OpenTelemetry.       | Sets the appropriate authentication header for ingesting OpenTelemetry events into a Logflare source.                                                                                                                                                                |

lib/logflare/auth/cache.ex

@@ -4,31 +4,19 @@ defmodule Logflare.Auth.Cache do
   Cachex `expiration`.
   """
 
+  use Logflare.ContextCache
+
   alias Logflare.Auth
+  alias Logflare.Cache.CachexOps
   alias Logflare.OauthAccessTokens.OauthAccessToken
   alias Logflare.User
-  alias Logflare.Utils
 
   def child_spec(_) do
-    stats = Application.get_env(:logflare, :cache_stats, false)
-
-    %{
-      id: __MODULE__,
-      start:
-        {Cachex, :start_link,
-         [
-           __MODULE__,
-           [
-             hooks:
-               [
-                 if(stats, do: Utils.cache_stats()),
-                 Utils.cache_limit(100_000)
-               ]
-               |> Enum.filter(& &1),
-             expiration: Utils.cache_expiration_min(5, 2)
-           ]
-         ]}
-    }
+    CachexOps.child_spec(__MODULE__,
+      limit: 100_000,
+      ttl: to_timeout(minute: 5),
+      purge_interval: to_timeout(minute: 2)
+    )
   end
 
   @spec verify_access_token(OauthAccessToken.t() | String.t()) ::

lib/logflare/backends/cache.ex

@@ -1,33 +1,20 @@
 defmodule Logflare.Backends.Cache do
   @moduledoc false
 
+  use Logflare.ContextCache
+
   alias Logflare.Backends
-  alias Logflare.Utils
-  import Cachex.Spec
+  alias Logflare.Cache.CachexOps
+  alias Logflare.ContextCache.Warmer
 
   def child_spec(_) do
-    stats = Application.get_env(:logflare, :cache_stats, false)
+    ttl = to_timeout(minute: 20)
 
-    %{
-      id: __MODULE__,
-      start:
-        {Cachex, :start_link,
-         [
-           __MODULE__,
-           [
-             warmers: [
-               warmer(required: false, module: Backends.CacheWarmer, name: Backends.CacheWarmer)
-             ],
-             hooks:
-               [
-                 if(stats, do: Utils.cache_stats()),
-                 Utils.cache_limit(100_000)
-               ]
-               |> Enum.filter(& &1),
-             expiration: Utils.cache_expiration_min()
-           ]
-         ]}
-    }
+    CachexOps.child_spec(__MODULE__,
+      limit: 100_000,
+      ttl: ttl,
+      warmer: {Backends.CacheWarmer, interval: Warmer.interval(ttl)}
+    )
   end
 
   def list_backends(arg), do: apply_repo_fun(__ENV__.function, [arg])

lib/logflare/backends/cache_warmer.ex

@@ -1,20 +1,19 @@
 defmodule Logflare.Backends.CacheWarmer do
   alias Logflare.Backends
-  alias Logflare.Repo
+  alias Logflare.Backends.Cache
+  alias Logflare.ContextCache.Warmer
 
   use Cachex.Warmer
 
   @impl true
-  def execute(_state), do: Repo.with_replica(&warm/0)
+  def execute(_state), do: Warmer.warm(Cache, &warm/0)
 
+  @spec warm() :: Warmer.pairs()
   defp warm do
     backends = Backends.list_backends(ingesting: true, limit: 1_000)
 
-    get_kv =
-      for b <- backends do
-        {{:get_backend, [b.id]}, {:cached, b}}
-      end
-
-    {:ok, get_kv}
+    for b <- backends do
+      {{:get_backend, [b.id]}, {:cached, b}}
+    end
   end
 end

lib/logflare/billing/cache.ex

@@ -1,29 +1,17 @@
 defmodule Logflare.Billing.Cache do
   @moduledoc false
 
+  use Logflare.ContextCache
+
   alias Logflare.Billing
-  alias Logflare.Utils
+  alias Logflare.Cache.CachexOps
 
   def child_spec(_) do
-    stats = Application.get_env(:logflare, :cache_stats, false)
-
-    %{
-      id: __MODULE__,
-      start:
-        {Cachex, :start_link,
-         [
-           __MODULE__,
-           [
-             hooks:
-               [
-                 if(stats, do: Utils.cache_stats()),
-                 Utils.cache_limit(100_000)
-               ]
-               |> Enum.filter(& &1),
-             expiration: Utils.cache_expiration_min(180, 10)
-           ]
-         ]}
-    }
+    CachexOps.child_spec(__MODULE__,
+      limit: 100_000,
+      ttl: to_timeout(hour: 3),
+      purge_interval: to_timeout(minute: 10)
+    )
   end
 
   def get_billing_account_by(keyword) do

lib/logflare/cache.ex

@@ -0,0 +1,52 @@
+defmodule Logflare.Cache do
+  @moduledoc """
+  Operational contract of an application cache, independent of its storage backend.
+
+  `use Logflare.Cache` injects default implementations of every callback, delegating to
+  `Logflare.Cache.CachexOps`. A cache on another backend overrides them, or passes
+  `impl: module` naming a `Logflare.Cache.Ops` implementation.
+  """
+
+  alias Logflare.Cache.CachexOps
+
+  @typedoc """
+  Counters since the last `c:reset/0`. Rates are percentages (0-100); `total_heap_size` is in bytes.
+  """
+  @type stats() :: %{
+          evictions: non_neg_integer(),
+          expirations: non_neg_integer(),
+          operations: non_neg_integer(),
+          hits: non_neg_integer(),
+          misses: non_neg_integer(),
+          hit_rate: number(),
+          miss_rate: number(),
+          total_heap_size: non_neg_integer()
+        }
+
+  @doc "Whether the cache on this node can serve requests."
+  @callback healthy?() :: boolean()
+
+  @callback stats() :: stats()
+
+  @doc "Clears all entries and statistics."
+  @callback reset() :: :ok
+
+  defmacro __using__(opts) do
+    impl = Keyword.get(opts, :impl, CachexOps)
+
+    quote do
+      @behaviour Logflare.Cache
+
+      @impl Logflare.Cache
+      def healthy?, do: unquote(impl).healthy?(__MODULE__)
+
+      @impl Logflare.Cache
+      def stats, do: unquote(impl).stats(__MODULE__)
+
+      @impl Logflare.Cache
+      def reset, do: unquote(impl).reset(__MODULE__)
+
+      defoverridable healthy?: 0, stats: 0, reset: 0
+    end
+  end
+end