Graham McIntire
485676887a
The slow CDS submit/poll/download cycle of the month-batch worker was sharing the era5 queue with the cheap Era5FetchWorker router, and its 2 concurrent slots were permanently pinned by long-running batch jobs while hundreds of router jobs starved. Give the batch worker its own era5_batch queue (also 2 concurrent per pod) so the router has dedicated capacity.
68 lines
2.2 KiB
Elixir
68 lines
2.2 KiB
Elixir
defmodule Microwaveprop.Workers.Era5MonthBatchWorker do
|
||
@moduledoc """
|
||
Fetches one month of ERA5 data for a 2° × 2° tile in a single CDS request
|
||
and bulk-inserts all resulting hourly profiles into `era5_profiles`.
|
||
|
||
This exists because fetching ERA5 per point-hour is tragically slow: each
|
||
CDS request goes through a full submit → poll → assemble → download cycle
|
||
whose overhead dwarfs the data transfer. Grouping by month+tile turns
|
||
thousands of CDS jobs into a handful, and once a tile-month is cached in
|
||
the DB every downstream query for a point inside that window is a local
|
||
lookup.
|
||
|
||
Uniqueness on the full arg set means Oban collapses duplicate enqueues
|
||
for the same tile-month — multiple `Era5FetchWorker` requests for the
|
||
same region collapse into a single batch.
|
||
|
||
Runs on its own `:era5_batch` queue so its slow CDS poll cycles don't
|
||
starve the fast `Era5FetchWorker` router that also lives in the ERA5
|
||
namespace.
|
||
"""
|
||
use Oban.Worker,
|
||
queue: :era5_batch,
|
||
max_attempts: 3,
|
||
unique: [
|
||
period: :infinity,
|
||
states: [:available, :scheduled, :executing, :retryable],
|
||
keys: [:year, :month, :tile_lat, :tile_lon]
|
||
]
|
||
|
||
alias Microwaveprop.Weather.Era5BatchClient
|
||
|
||
require Logger
|
||
|
||
@impl Oban.Worker
|
||
def backoff(%Oban.Job{attempt: attempt}) do
|
||
# A full month fetch can sit in the CDS queue for ages; back off generously.
|
||
min(600 * Integer.pow(2, attempt - 1), _one_day = 86_400)
|
||
end
|
||
|
||
@impl Oban.Worker
|
||
def perform(%Oban.Job{args: args}) do
|
||
params = %{
|
||
year: fetch_int!(args, "year"),
|
||
month: fetch_int!(args, "month"),
|
||
tile_lat: fetch_int!(args, "tile_lat"),
|
||
tile_lon: fetch_int!(args, "tile_lon")
|
||
}
|
||
|
||
case Era5BatchClient.fetch_month_into_db(params) do
|
||
{:ok, _count} ->
|
||
:ok
|
||
|
||
{:error, reason} ->
|
||
Logger.warning(
|
||
"ERA5Batch: #{params.year}-#{params.month} tile #{params.tile_lat},#{params.tile_lon} failed: #{inspect(reason)}"
|
||
)
|
||
|
||
{:error, reason}
|
||
end
|
||
end
|
||
|
||
defp fetch_int!(args, key) do
|
||
case Map.fetch!(args, key) do
|
||
n when is_integer(n) -> n
|
||
n when is_binary(n) -> String.to_integer(n)
|
||
end
|
||
end
|
||
end
|