From e99e585b66acdb8dcd30cf8cd9074d3fc24bb374 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Mon, 6 Apr 2026 11:42:20 -0500 Subject: [PATCH] Add type filter to backfill enqueue Backfill dashboard now has checkboxes to select which enrichment types to process (HRRR, Weather, Terrain, IEMRE). Prevents weather jobs from flooding the queue when only HRRR needs work. --- config/runtime.exs | 28 +++++++------- .../workers/backfill_enqueue_worker.ex | 38 ++++++++++++------- .../workers/contact_weather_enqueue_worker.ex | 19 +++++----- .../workers/weather_fetch_worker.ex | 4 +- lib/microwaveprop_web/live/backfill_live.ex | 25 ++++++++++-- lib/microwaveprop_web/router.ex | 4 +- tail_logs.sh | 3 ++ .../workers/solar_index_worker_test.exs | 4 +- .../plugs/remote_ip_test.exs | 21 ++++++---- 9 files changed, 93 insertions(+), 53 deletions(-) create mode 100755 tail_logs.sh diff --git a/config/runtime.exs b/config/runtime.exs index bbf8b4d7..ebf04c15 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -46,6 +46,20 @@ if config_env() == :prod do host = System.get_env("PHX_HOST") || "example.com" + config :libcluster, + topologies: [ + k8s: [ + strategy: Cluster.Strategy.Kubernetes, + config: [ + mode: :ip, + kubernetes_ip_lookup_mode: :pods, + kubernetes_node_basename: "microwaveprop", + kubernetes_selector: "app=prop", + kubernetes_namespace: "prop" + ] + ] + ] + # ## SSL Support # # To get SSL working, you will need to add the `https` key @@ -130,20 +144,6 @@ if config_env() == :prod do ]} ] - config :libcluster, - topologies: [ - k8s: [ - strategy: Cluster.Strategy.Kubernetes, - config: [ - mode: :ip, - kubernetes_ip_lookup_mode: :pods, - kubernetes_node_basename: "microwaveprop", - kubernetes_selector: "app=prop", - kubernetes_namespace: "prop" - ] - ] - ] - config :microwaveprop, :email_from, {"NTMS Propagation", "graham@w5isp.com"} config :microwaveprop, hrrr_base_url: System.get_env("HRRR_BASE_URL", "http://skippy.w5isp.com:8080") config :microwaveprop, srtm_tiles_dir: "/srtm" diff --git a/lib/microwaveprop/workers/backfill_enqueue_worker.ex b/lib/microwaveprop/workers/backfill_enqueue_worker.ex index c0ac4086..c88266de 100644 --- a/lib/microwaveprop/workers/backfill_enqueue_worker.ex +++ b/lib/microwaveprop/workers/backfill_enqueue_worker.ex @@ -14,25 +14,24 @@ defmodule Microwaveprop.Workers.BackfillEnqueueWorker do require Logger @enrichable [:pending, :queued, :failed] + @valid_types ~w(hrrr weather terrain iemre) @impl Oban.Worker - def perform(%Oban.Job{args: %{"limit" => limit}}) do + def perform(%Oban.Job{args: %{"limit" => limit} = args}) do + types = parse_types(args) + contacts = - Repo.all( - from(c in Contact, - where: - (not is_nil(c.pos1) or not is_nil(c.grid1)) and - (c.hrrr_status in ^@enrichable or c.weather_status in ^@enrichable or - c.terrain_status in ^@enrichable or c.iemre_status in ^@enrichable), - order_by: [desc: c.qso_timestamp], - limit: ^limit - ) - ) + Contact + |> where([c], not is_nil(c.pos1) or not is_nil(c.grid1)) + |> where(^type_filter(types)) + |> order_by([c], desc: c.qso_timestamp) + |> limit(^limit) + |> Repo.all() count = length(contacts) - Logger.info("BackfillEnqueue: processing #{count} contacts") + Logger.info("BackfillEnqueue: processing #{count} contacts (types: #{inspect(types)})") - Enum.each(contacts, &ContactWeatherEnqueueWorker.enqueue_for_contact/1) + Enum.each(contacts, &ContactWeatherEnqueueWorker.enqueue_for_contact(&1, types)) Phoenix.PubSub.broadcast( Microwaveprop.PubSub, @@ -42,4 +41,17 @@ defmodule Microwaveprop.Workers.BackfillEnqueueWorker do :ok end + + defp parse_types(%{"types" => types}) when is_list(types) do + types |> Enum.filter(&(&1 in @valid_types)) |> Enum.map(&String.to_existing_atom/1) + end + + defp parse_types(_), do: [:hrrr, :weather, :terrain, :iemre] + + defp type_filter(types) do + Enum.reduce(types, dynamic(false), fn type, acc -> + field = :"#{type}_status" + dynamic([c], ^acc or field(c, ^field) in ^@enrichable) + end) + end end diff --git a/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex b/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex index c1989347..727e40f8 100644 --- a/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex +++ b/lib/microwaveprop/workers/contact_weather_enqueue_worker.ex @@ -18,13 +18,14 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do @doc """ Enqueue all enrichment jobs (weather, HRRR, terrain, IEMRE) for a single contact. Called directly from submission flow — no Oban indirection. + Optionally pass a list of types to limit which enrichments run. """ - def enqueue_for_contact(contact) do + def enqueue_for_contact(contact, types \\ [:hrrr, :weather, :terrain, :iemre]) do contact = Radio.ensure_positions!(contact) - weather_jobs = build_weather_jobs([contact]) - hrrr_jobs = build_hrrr_jobs([contact]) - terrain_jobs = build_terrain_jobs([contact]) - iemre_jobs = build_iemre_jobs([contact]) + weather_jobs = if :weather in types, do: build_weather_jobs([contact]), else: [] + hrrr_jobs = if :hrrr in types, do: build_hrrr_jobs([contact]), else: [] + terrain_jobs = if :terrain in types, do: build_terrain_jobs([contact]), else: [] + iemre_jobs = if :iemre in types, do: build_iemre_jobs([contact]), else: [] all_jobs = weather_jobs ++ hrrr_jobs ++ terrain_jobs ++ iemre_jobs @@ -35,13 +36,13 @@ defmodule Microwaveprop.Workers.ContactWeatherEnqueueWorker do ids = [contact.id] if contact.pos1 do - mark_status!(ids, :weather_status, weather_jobs) - mark_status!(ids, :hrrr_status, hrrr_jobs) - mark_status!(ids, :iemre_status, iemre_jobs) + if :weather in types, do: mark_status!(ids, :weather_status, weather_jobs) + if :hrrr in types, do: mark_status!(ids, :hrrr_status, hrrr_jobs) + if :iemre in types, do: mark_status!(ids, :iemre_status, iemre_jobs) end if contact.pos1 && contact.pos2 do - mark_status!(ids, :terrain_status, terrain_jobs) + if :terrain in types, do: mark_status!(ids, :terrain_status, terrain_jobs) end :ok diff --git a/lib/microwaveprop/workers/weather_fetch_worker.ex b/lib/microwaveprop/workers/weather_fetch_worker.ex index 65e3c94b..433ffba7 100644 --- a/lib/microwaveprop/workers/weather_fetch_worker.ex +++ b/lib/microwaveprop/workers/weather_fetch_worker.ex @@ -5,14 +5,14 @@ defmodule Microwaveprop.Workers.WeatherFetchWorker do max_attempts: 20, unique: [period: 300, states: [:available, :scheduled, :executing, :retryable]] - require Logger - alias Microwaveprop.Repo alias Microwaveprop.Weather alias Microwaveprop.Weather.IemClient alias Microwaveprop.Weather.SoundingParams alias Microwaveprop.Weather.Station + require Logger + @impl Oban.Worker def backoff(%Oban.Job{attempt: attempt}) do min(120 * Integer.pow(2, attempt - 1), _six_hours = 21_600) diff --git a/lib/microwaveprop_web/live/backfill_live.ex b/lib/microwaveprop_web/live/backfill_live.ex index 5b9ec8da..4905e952 100644 --- a/lib/microwaveprop_web/live/backfill_live.ex +++ b/lib/microwaveprop_web/live/backfill_live.ex @@ -41,10 +41,18 @@ defmodule MicrowavepropWeb.BackfillLive do end @impl true - def handle_event("enqueue", %{"limit" => limit_str}, socket) do + def handle_event("enqueue", %{"limit" => limit_str} = params, socket) do limit = String.to_integer(limit_str) - %{"limit" => limit} + types = + ~w(hrrr weather terrain iemre) + |> Enum.filter(&(params[&1] == "true")) + |> then(fn + [] -> ~w(hrrr weather terrain iemre) + types -> types + end) + + %{"limit" => limit, "types" => types} |> BackfillEnqueueWorker.new() |> Oban.insert() @@ -295,7 +303,8 @@ defmodule MicrowavepropWeb.BackfillLive do
<%= for {label, complete} <- [{"HRRR", @unprocessed.total - @unprocessed.hrrr}, {"Weather", @unprocessed.total - @unprocessed.weather}, {"Terrain", @unprocessed.total - @unprocessed.terrain}, {"IEMRE", @unprocessed.total - @unprocessed.iemre}] do %> - <% pct = if @unprocessed.total > 0, do: round(complete * 100 / @unprocessed.total), else: 0 %> + <% pct = + if @unprocessed.total > 0, do: round(complete * 100 / @unprocessed.total), else: 0 %>
{label}
-
+
+
+ <%= for type <- ~w(hrrr weather terrain iemre) do %> + + <% end %> +