From bcae416e1438eb5f884c86e909db5e0a0a5d5b98 Mon Sep 17 00:00:00 2001 From: Liam Date: Sat, 11 Jan 2025 16:58:45 -0500 Subject: [PATCH 1/2] Add maintenance stream wrapper --- lib/philomena/maintenance.ex | 108 +++++++++++++++++++++++++++++++++++ 1 file changed, 108 insertions(+) create mode 100644 lib/philomena/maintenance.ex diff --git a/lib/philomena/maintenance.ex b/lib/philomena/maintenance.ex new file mode 100644 index 00000000..22aa5b41 --- /dev/null +++ b/lib/philomena/maintenance.ex @@ -0,0 +1,108 @@ +defmodule Philomena.Maintenance do + @moduledoc """ + Functions related to online and offline maintenance tasks. + """ + + @typedoc "Progress from a stream job." + @type progress_report :: %{ + curr: integer(), + rate: number(), + remaining_time: number() + } + + @doc """ + Periodically stream progress reports for a stream task that produces a range + of integers between `min` and `max`, estimating the rate of progress and time + remaining. + """ + @spec stream_progress( + id_stream :: Enumerable.t({:ok, integer()}), + min :: integer(), + max :: integer(), + report_period :: number() + ) :: Enumerable.t(progress_report()) + def stream_progress(id_stream, min, max, report_period \\ 1.0) do + # Reference point for comparison during the stream. + begin = now() + + # Estimate progress counters based on how many objects have been + # processed since the initial reference point. + create_report = fn state, curr_id -> + curr_rate = (curr_id - min) / max(now() - begin, 1) + remaining_time = (max - curr_id) / max(curr_rate, 1) + + %{ + state: state, + curr: curr_id, + rate: round(curr_rate), + remaining_time: remaining_time + } + end + + # Convert input items received after every period elapses into + # a report, then concatenate an additional report after all items + # are processed. + id_stream + |> Stream.transform(begin, fn {:ok, curr_id}, prev_time -> + curr_time = now() + + if curr_time - prev_time > report_period do + {[create_report.(:in_progress, curr_id)], curr_time} + else + {[], prev_time} + end + end) + |> Stream.concat(Stream.map([[]], fn _ -> create_report.(:done, max) end)) + end + + @doc """ + Write progress reports to the console for a stream task that produces a range + of integers between `min` and `max`, estimating the rate of progress and time + remaining. + """ + @spec log_progress( + id_stream :: Enumerable.t({:ok, integer()}), + label :: String.t(), + min :: integer(), + max :: integer(), + report_period :: number() + ) :: :ok + def log_progress(id_stream, label, min, max, report_period \\ 1.0) do + id_stream + |> stream_progress(min, max, report_period) + |> Enum.each(fn p -> + # Clear line + IO.write("\e[2K\r") + + # Newline on report depends on whether stream is finished + case p.state do + :in_progress -> + eta = format_eta(p.remaining_time) + + IO.write("#{label}: #{p.curr}/#{max} [#{p.rate}/sec], ETA: #{eta}") + + :done -> + IO.puts("#{label}: #{p.curr}/#{max} [#{p.rate}/sec], done.") + end + end) + end + + @spec format_eta(number()) :: String.t() + defp format_eta(remaining_time) do + seconds = round(remaining_time) + minutes = div(seconds, 60) + hours = div(minutes, 60) + + cond do + seconds < 45 -> "about #{seconds} second(s)" + seconds < 90 -> "about a minute" + minutes < 45 -> "about #{minutes} minute(s)" + true -> "about #{hours} hour(s)" + end + end + + @spec now() :: float() + defp now do + :erlang.system_time(:microsecond) / 1_000_000 + end +end From 07ee436d6d5c387a7271ed130fbdc1f52d8dd9c3 Mon Sep 17 00:00:00 2001 From: Liam Date: Sat, 11 Jan 2025 17:31:05 -0500 Subject: [PATCH 2/2] Add reindexing command line with rate and time estimate --- lib/mix/tasks/reindex_all.ex | 2 +- lib/philomena/search_indexer.ex | 61 ++++++++++++++++++++++----------- lib/philomena_query/search.ex | 61 +++++++++++++++++++++++++++++---- 3 files changed, 97 insertions(+), 27 deletions(-) diff --git a/lib/mix/tasks/reindex_all.ex b/lib/mix/tasks/reindex_all.ex index b1af4887..41b6f85d 100644 --- a/lib/mix/tasks/reindex_all.ex +++ b/lib/mix/tasks/reindex_all.ex @@ -11,6 +11,6 @@ defmodule Mix.Tasks.ReindexAll do raise "do not run this task unless you know what you're doing" end - SearchIndexer.recreate_reindex_all_destructive!() + SearchIndexer.recreate_reindex_all_destructive!(maintenance: false) end end diff --git a/lib/philomena/search_indexer.ex b/lib/philomena/search_indexer.ex index cf564a5c..9dc8cf0d 100644 --- a/lib/philomena/search_indexer.ex +++ b/lib/philomena/search_indexer.ex @@ -17,7 +17,9 @@ defmodule Philomena.SearchIndexer do alias Philomena.Tags alias Philomena.Tags.Tag + alias Philomena.Maintenance alias Philomena.Polymorphic + alias Philomena.Repo import Ecto.Query @schemas [ @@ -60,10 +62,10 @@ defmodule Philomena.SearchIndexer do :ok """ - @spec recreate_reindex_all_destructive! :: :ok - def recreate_reindex_all_destructive! do + @spec recreate_reindex_all_destructive!(opts :: Keyword.t()) :: :ok + def recreate_reindex_all_destructive!(opts \\ []) do @schemas - |> Stream.map(&recreate_reindex_schema_destructive!/1) + |> Stream.map(&recreate_reindex_schema_destructive!(&1, opts)) |> Stream.run() end @@ -77,12 +79,12 @@ defmodule Philomena.SearchIndexer do :ok """ - @spec recreate_reindex_schema_destructive!(schema :: module()) :: :ok - def recreate_reindex_schema_destructive!(schema) when schema in @schemas do + @spec recreate_reindex_schema_destructive!(schema :: module(), opts :: Keyword.t()) :: :ok + def recreate_reindex_schema_destructive!(schema, opts \\ []) when schema in @schemas do Search.delete_index!(schema) Search.create_index!(schema) - reindex_schema(schema) + reindex_schema(schema, opts) end @doc """ @@ -94,10 +96,10 @@ defmodule Philomena.SearchIndexer do :ok """ - @spec reindex_all :: :ok - def reindex_all do + @spec reindex_all(opts :: Keyword.t()) :: :ok + def reindex_all(opts \\ []) do @schemas - |> Stream.map(&reindex_schema/1) + |> Stream.map(&reindex_schema(&1, opts)) |> Stream.run() end @@ -110,10 +112,30 @@ defmodule Philomena.SearchIndexer do :ok """ - @spec reindex_schema(schema :: module()) :: :ok - def reindex_schema(schema) + @spec reindex_schema(schema :: module(), opts :: Keyword.t()) :: :ok + def reindex_schema(schema, opts \\ []) do + maintenance = Keyword.get(opts, :maintenance, true) - def reindex_schema(Report) do + if maintenance do + query = limit(schema, 1) + min = Repo.one(order_by(query, asc: :id)).id + max = Repo.one(order_by(query, desc: :id)).id + + schema + |> reindex_schema_impl(opts) + |> Maintenance.log_progress(inspect(schema), min, max) + else + schema + |> reindex_schema_impl(opts) + |> Stream.run() + end + end + + @spec reindex_schema_impl(schema :: module(), opts :: Keyword.t()) :: + Enumerable.t({:ok, integer()}) + defp reindex_schema_impl(schema, opts) + + defp reindex_schema_impl(Report, opts) do # Reports currently require handling for their polymorphic nature Report |> preload([:user, :admin]) @@ -125,25 +147,24 @@ defmodule Philomena.SearchIndexer do |> Enum.map(&Search.index_document(&1, Report)) end, timeout: :infinity, - max_concurrency: max_concurrency() + max_concurrency: max_concurrency(opts) ) - |> Stream.run() end - def reindex_schema(schema) when schema in @schemas do + defp reindex_schema_impl(schema, opts) when schema in @schemas do # Normal schemas can simply be reindexed with indexing_preloads context = Map.fetch!(@contexts, schema) schema |> preload(^context.indexing_preloads()) - |> Search.reindex(schema, + |> Search.reindex_stream(schema, batch_size: @batch_sizes[schema], - max_concurrency: max_concurrency() + max_concurrency: max_concurrency(opts) ) end - @spec max_concurrency() :: pos_integer() - defp max_concurrency do - System.schedulers_online() + @spec max_concurrency(opts :: Keyword.t()) :: pos_integer() + defp max_concurrency(opts) do + Keyword.get(opts, :max_concurrency, System.schedulers_online()) end end diff --git a/lib/philomena_query/search.ex b/lib/philomena_query/search.ex index ff6a29b1..cfc3a93e 100644 --- a/lib/philomena_query/search.ex +++ b/lib/philomena_query/search.ex @@ -189,6 +189,10 @@ defmodule PhilomenaQuery.Search do Note that indexing is near real-time and requires an index refresh before documents will become visible. Unless changed in the mapping, this happens after 5 seconds have elapsed. + > #### Warning {: .warning} + > The returned stream must be enumerated for the reindex to process. If you do not care + > about the progress IDs yielded, use `reindex/3` instead. + ## Example query = @@ -196,11 +200,14 @@ defmodule PhilomenaQuery.Search do where: i.id < 100_000, preload: ^Images.indexing_preloads() - Search.reindex(query, Image, batch_size: 5000) + query + |> Search.reindex_stream(Image, batch_size: 1024) + |> Enum.each(&IO.inspect/1) """ - @spec reindex(queryable(), schema_module(), Batch.batch_options()) :: :ok - def reindex(queryable, module, opts \\ []) do + @spec reindex_stream(queryable(), schema_module(), Batch.batch_options()) :: + Enumerable.t({:ok, integer()}) + def reindex_stream(queryable, module, opts \\ []) do max_concurrency = Keyword.get(opts, :max_concurrency, 1) index = @policy.index_for(module) @@ -208,10 +215,10 @@ defmodule PhilomenaQuery.Search do |> Batch.query_batches(opts) |> Task.async_stream( fn query -> + records = Repo.all(query) + lines = - query - |> Repo.all() - |> Enum.flat_map(fn record -> + Enum.flat_map(records, fn record -> doc = index.as_json(record) [ @@ -221,10 +228,52 @@ defmodule PhilomenaQuery.Search do end) Api.bulk(@policy.opensearch_url(), lines) + + last_id(records) end, timeout: :infinity, max_concurrency: max_concurrency ) + |> flatten_stream() + end + + defp last_id([]), do: [] + defp last_id(records), do: [Enum.max_by(records, & &1.id).id] + + @spec flatten_stream(Enumerable.t({:ok, [integer()]})) :: Enumerable.t({:ok, integer()}) + defp flatten_stream(stream) do + # Converts [{:ok, [1, 2]}] into [{:ok, 1}, {:ok, 2}] + Stream.transform(stream, [], fn {:ok, last_id}, _ -> + {Enum.map(last_id, &{:ok, &1}), []} + end) + end + + @doc """ + Efficiently index a batch of documents in the index named by the module. + + This function is substantially more efficient than running `index_document/2` for + each instance of a schema struct and can index with hundreds of times the throughput. + + The queryable should be a schema type with its indexing preloads included in + the query. The options are forwarded to `PhilomenaQuery.Batch.record_batches/3`. + + Note that indexing is near real-time and requires an index refresh before documents will + become visible. Unless changed in the mapping, this happens after 5 seconds have elapsed. + + ## Example + + query = + from i in Image, + where: i.id < 100_000, + preload: ^Images.indexing_preloads() + + Search.reindex(query, Image, batch_size: 1024) + + """ + @spec reindex(queryable(), schema_module(), Batch.batch_options()) :: :ok + def reindex(queryable, module, opts \\ []) do + queryable + |> reindex_stream(module, opts) |> Stream.run() end