Search backend healthcheck process

changelog.d/search-healthcheck.add

@@ -0,0 +1 @@
+Monitoring of search backend health to control the processing of jobs in the search indexing Oban queue

config/config.exs

@@ -579,7 +579,7 @@
     attachments_cleanup: 1,
     new_users_digest: 1,
     mute_expire: 5,
-    search_indexing: 10,
+    search_indexing: [limit: 10, paused: true],
     rich_media_expiration: 2
   ],
   plugins: [Oban.Plugins.Pruner],

lib/pleroma/application.ex

@@ -109,7 +109,8 @@ def start(_type, _args) do
         streamer_registry() ++
         background_migrators() ++
         shout_child(shout_enabled?()) ++
-        [Pleroma.Gopher.Server]
+        [Pleroma.Gopher.Server] ++
+        [Pleroma.Search.Healthcheck]
 
     # See http://elixir-lang.org/docs/stable/elixir/Supervisor.html
     # for other strategies and supported options

lib/pleroma/search.ex

@@ -14,4 +14,9 @@ def search(query, options) do
 
     search_module.search(options[:for_user], query, options)
   end
+
+  def healthcheck_endpoints do
+    search_module = Pleroma.Config.get([Pleroma.Search, :module], Pleroma.Activity)
+    search_module.healthcheck_endpoints
+  end
 end

lib/pleroma/search/database_search.ex

@@ -48,6 +48,9 @@ def add_to_index(_activity), do: :ok
   @impl true
   def remove_from_index(_object), do: :ok
 
+  @impl true
+  def healthcheck_endpoints, do: nil
+
   def maybe_restrict_author(query, %User{} = author) do
     Activity.Queries.by_author(query, author)
   end

lib/pleroma/search/healthcheck.ex

@@ -0,0 +1,85 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2024 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+defmodule Pleroma.Search.Healthcheck do
+  @doc """
+  Monitors health of search backend to control processing of events based on health and availability.
+  """
+  use GenServer
+  require Logger
+
+  @tick :timer.seconds(60)
+  @queue :search_indexing
+
+  def start_link(_) do
+    GenServer.start_link(__MODULE__, [], name: __MODULE__)
+  end
+
+  @impl true
+  def init(_) do
+    state = %{healthy: false}
+    {:ok, state, {:continue, :start}}
+  end
+
+  @impl true
+  def handle_continue(:start, state) do
+    tick()
+    {:noreply, state}
+  end
+
+  @impl true
+  def handle_info(:check, state) do
+    urls = Pleroma.Search.healthcheck_endpoints()
+
+    new_state =
+      if healthy?(urls) do
+        Oban.resume_queue(queue: @queue)
+        Map.put(state, :healthy, true)
+      else
+        Oban.pause_queue(queue: @queue)
+        Map.put(state, :healthy, false)
+      end
+
+    maybe_log_state_change(state, new_state)
+
+    tick()
+    {:noreply, new_state}
+  end
+
+  @impl true
+  def handle_call(:check, _from, state) do
+    status = Map.get(state, :healthy)
+
+    {:reply, status, state, :hibernate}
+  end
+
+  defp healthy?([]), do: true
+
+  defp healthy?(urls) when is_list(urls) do
+    Enum.all?(
+      urls,
+      fn url ->
+        case Pleroma.HTTP.get(url) do
+          {:ok, %{status: 200}} -> true
+          _ -> false
+        end
+      end
+    )
+  end
+
+  defp healthy?(_), do: true
+
+  defp tick do
+    Process.send_after(self(), :check, @tick)
+  end
+
+  defp maybe_log_state_change(%{healthy: true}, %{healthy: false}) do
+    Logger.error("Pausing Oban queue #{@queue} due to search backend healthcheck failure")
+  end
+
+  defp maybe_log_state_change(%{healthy: false}, %{healthy: true}) do
+    Logger.info("Resuming Oban queue #{@queue} due to search backend healthcheck pass")
+  end
+
+  defp maybe_log_state_change(_, _), do: :ok
+end

lib/pleroma/search/meilisearch.ex

@@ -178,4 +178,15 @@ def add_to_index(activity) do
   def remove_from_index(object) do
     meili_delete("/indexes/objects/documents/#{object.id}")
   end
+
+  @impl true
+  def healthcheck_endpoints do
+    endpoint =
+      Config.get([Pleroma.Search.Meilisearch, :url])
+      |> URI.parse()
+      |> Map.put(:path, "/health")
+      |> URI.to_string()
+
+    [endpoint]
+  end
 end

lib/pleroma/search/search_backend.ex

@@ -21,4 +21,12 @@ defmodule Pleroma.Search.SearchBackend do
   from index.
   """
   @callback remove_from_index(object :: Pleroma.Object.t()) :: :ok | {:error, any()}
+
+  @doc """
+  Healthcheck endpoints of search backend infrastructure to monitor for controlling
+  processing of jobs in the Oban queue.
+
+  It is expected a 200 response is healthy and other responses are unhealthy.
+  """
+  @callback healthcheck_endpoints :: list() | nil
 end