From 2da74c5cd87fa94eda772dc54e34cd85d3aad014 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Sat, 18 Apr 2026 16:33:28 -0500 Subject: [PATCH] feat(telemetry): wide instrumentation + bump hrrr to 2 per pod MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Config: - runtime.exs hrrr queue 1 → 2 (6 concurrent HRRR jobs across 3 pods) New helper Microwaveprop.Instrument.span/3 wraps :telemetry.span with a [:microwaveprop | event_suffix] prefix so metrics can key off it. Spans added (each emits duration + result tag): - HrrrClient.fetch_grid / fetch_profile / fetch_idx - NexradClient.fetch_frame + decode_png - IemClient.fetch_asos / fetch_raob - GefsClient.fetch_grid_profiles - NarrClient.fetch_profile_at - UwyoSoundingClient.fetch_sounding - ElevationClient.fetch_elevation_profile - Weather.upsert_hrrr_profiles_batch / upsert_gefs_profiles_batch - Propagation.replace_scores - PropagationGridWorker.compute_scores_algorithm - TerrainAnalysis.analyse - CommonVolumeRadarWorker.aggregate_stats Telemetry catalog (MicrowavepropWeb.Telemetry): - Oban job duration / queue_time / count / exception by (worker, queue, state) - Per-span summary metrics for every instrumented phase above - Periodic (10s) poller emits oban queue depth by (queue, state) — drops into the /admin/dashboard Metrics tab immediately Also drops the now-redundant "fetching n0q frame" and "fetching " info lines from CommonVolumeRadarWorker / NexradClient; the span events cover that and the worker's "ingested" line stays for per-job signal. --- config/runtime.exs | 2 +- lib/microwaveprop/instrument.ex | 26 ++ lib/microwaveprop/propagation.ex | 8 + lib/microwaveprop/terrain/elevation_client.ex | 22 +- lib/microwaveprop/terrain/terrain_analysis.ex | 8 + lib/microwaveprop/weather.ex | 16 ++ lib/microwaveprop/weather/gefs_client.ex | 8 + lib/microwaveprop/weather/hrrr_client.ex | 34 ++- lib/microwaveprop/weather/iem_client.ex | 36 +-- lib/microwaveprop/weather/narr_client.ex | 12 +- lib/microwaveprop/weather/nexrad_client.ex | 57 +++-- .../weather/uwyo_sounding_client.ex | 14 +- .../workers/common_volume_radar_worker.ex | 8 +- .../workers/propagation_grid_worker.ex | 25 +- lib/microwaveprop_web/telemetry.ex | 230 ++++++++++++++++-- 15 files changed, 406 insertions(+), 100 deletions(-) create mode 100644 lib/microwaveprop/instrument.ex diff --git a/config/runtime.exs b/config/runtime.exs index 8353ead8..5bb10894 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -166,7 +166,7 @@ if config_env() == :prod do solar: 1, weather: 3, gefs: 1, - hrrr: 1, + hrrr: 2, terrain: 3, iemre: 3, # Historical backfill for pre-2014 contacts (pre-HRRR archive). diff --git a/lib/microwaveprop/instrument.ex b/lib/microwaveprop/instrument.ex new file mode 100644 index 00000000..dd19e064 --- /dev/null +++ b/lib/microwaveprop/instrument.ex @@ -0,0 +1,26 @@ +defmodule Microwaveprop.Instrument do + @moduledoc """ + Thin wrapper around `:telemetry.span/3` so the rest of the codebase can + instrument a block of work with a single call: + + Instrument.span([:hrrr, :fetch_grid], %{points: length(points)}, fn -> + HrrrClient.fetch_grid(points, valid_time) + end) + + Emits `[:microwaveprop | event_suffix] ++ [:start|:stop|:exception]` + with the automatic `duration` measurement, so metrics registered in + `MicrowavepropWeb.Telemetry` can key off the same prefix. + """ + + @type event_suffix :: [atom(), ...] + + @spec span(event_suffix(), map(), (-> term())) :: term() + def span(event_suffix, metadata \\ %{}, fun) when is_list(event_suffix) and is_function(fun, 0) do + event = [:microwaveprop | event_suffix] + + :telemetry.span(event, metadata, fn -> + result = fun.() + {result, Map.put(metadata, :result, :ok)} + end) + end +end diff --git a/lib/microwaveprop/propagation.ex b/lib/microwaveprop/propagation.ex index 67f5cc77..3f33c74f 100644 --- a/lib/microwaveprop/propagation.ex +++ b/lib/microwaveprop/propagation.ex @@ -166,6 +166,14 @@ defmodule Microwaveprop.Propagation do """ @spec replace_scores(Enumerable.t(), DateTime.t()) :: {:ok, non_neg_integer()} | {:error, term()} def replace_scores(scores, %DateTime{} = valid_time) do + Microwaveprop.Instrument.span( + [:db, :replace_scores], + %{valid_time: valid_time}, + fn -> do_replace_scores(scores, valid_time) end + ) + end + + defp do_replace_scores(scores, valid_time) do {per_band, total} = Enum.reduce(scores, {%{}, 0}, fn score, {acc, count} -> {Map.update(acc, score.band_mhz, [score], &[score | &1]), count + 1} diff --git a/lib/microwaveprop/terrain/elevation_client.ex b/lib/microwaveprop/terrain/elevation_client.ex index 46f99dc1..018cd58a 100644 --- a/lib/microwaveprop/terrain/elevation_client.ex +++ b/lib/microwaveprop/terrain/elevation_client.ex @@ -10,15 +10,21 @@ defmodule Microwaveprop.Terrain.ElevationClient do @spec fetch_elevation_profile(float(), float(), float(), float(), pos_integer(), keyword()) :: {:ok, list(map())} | {:error, String.t()} def fetch_elevation_profile(lat1, lon1, lat2, lon2, n \\ 64, opts \\ []) do - case srtm_tiles_dir() do - nil -> - Logger.debug("SRTM tiles_dir not configured, using API") - fetch_elevation_profile_api(lat1, lon1, lat2, lon2, n) + Microwaveprop.Instrument.span( + [:elevation, :fetch_profile], + %{sample_count: n, source: if(srtm_tiles_dir(), do: :srtm, else: :api)}, + fn -> + case srtm_tiles_dir() do + nil -> + Logger.debug("SRTM tiles_dir not configured, using API") + fetch_elevation_profile_api(lat1, lon1, lat2, lon2, n) - tiles_dir -> - srtm_opts = Keyword.merge([download: true], opts) - Srtm.fetch_elevation_profile(lat1, lon1, lat2, lon2, tiles_dir, n, srtm_opts) - end + tiles_dir -> + srtm_opts = Keyword.merge([download: true], opts) + Srtm.fetch_elevation_profile(lat1, lon1, lat2, lon2, tiles_dir, n, srtm_opts) + end + end + ) end defp fetch_elevation_profile_api(lat1, lon1, lat2, lon2, n) do diff --git a/lib/microwaveprop/terrain/terrain_analysis.ex b/lib/microwaveprop/terrain/terrain_analysis.ex index 37ad9a0b..1b14f138 100644 --- a/lib/microwaveprop/terrain/terrain_analysis.ex +++ b/lib/microwaveprop/terrain/terrain_analysis.ex @@ -54,6 +54,14 @@ defmodule Microwaveprop.Terrain.TerrainAnalysis do """ @spec analyse([elevation_point()], float(), float(), keyword()) :: analysis_result() def analyse(profile, dist_km, freq_ghz, opts \\ []) do + Microwaveprop.Instrument.span( + [:terrain, :analyse], + %{point_count: length(profile), dist_km: dist_km, freq_ghz: freq_ghz}, + fn -> do_analyse(profile, dist_km, freq_ghz, opts) end + ) + end + + defp do_analyse(profile, dist_km, freq_ghz, opts) do ant_ht_a = Keyword.get(opts, :ant_ht_a, 0.0) ant_ht_b = Keyword.get(opts, :ant_ht_b, 0.0) k = Keyword.get(opts, :k_factor, 4 / 3) diff --git a/lib/microwaveprop/weather.ex b/lib/microwaveprop/weather.ex index 4be15ba5..c0b9f19f 100644 --- a/lib/microwaveprop/weather.ex +++ b/lib/microwaveprop/weather.ex @@ -769,6 +769,14 @@ defmodule Microwaveprop.Weather do @spec upsert_gefs_profiles_batch([map()]) :: {non_neg_integer(), nil} def upsert_gefs_profiles_batch(profiles) do + Microwaveprop.Instrument.span( + [:db, :upsert_gefs_profiles], + %{count: length(profiles)}, + fn -> do_upsert_gefs_profiles_batch(profiles) end + ) + end + + defp do_upsert_gefs_profiles_batch(profiles) do now = DateTime.truncate(DateTime.utc_now(), :second) profiles @@ -805,6 +813,14 @@ defmodule Microwaveprop.Weather do @spec upsert_hrrr_profiles_batch([map()], keyword()) :: {non_neg_integer(), nil} def upsert_hrrr_profiles_batch(profiles, _opts \\ []) do + Microwaveprop.Instrument.span( + [:db, :upsert_hrrr_profiles], + %{count: length(profiles)}, + fn -> do_upsert_hrrr_profiles_batch(profiles) end + ) + end + + defp do_upsert_hrrr_profiles_batch(profiles) do now = DateTime.truncate(DateTime.utc_now(), :second) profiles diff --git a/lib/microwaveprop/weather/gefs_client.ex b/lib/microwaveprop/weather/gefs_client.ex index b3dab9a6..e2479520 100644 --- a/lib/microwaveprop/weather/gefs_client.ex +++ b/lib/microwaveprop/weather/gefs_client.ex @@ -199,6 +199,14 @@ defmodule Microwaveprop.Weather.GefsClient do @spec fetch_grid_profiles(Date.t(), non_neg_integer(), non_neg_integer()) :: {:ok, [map()]} | {:error, term()} def fetch_grid_profiles(%Date{} = run_date, run_hour, forecast_hour) do + Microwaveprop.Instrument.span( + [:gefs, :fetch_grid_profiles], + %{forecast_hour: forecast_hour}, + fn -> do_fetch_grid_profiles(run_date, run_hour, forecast_hour) end + ) + end + + defp do_fetch_grid_profiles(run_date, run_hour, forecast_hour) do alias Microwaveprop.Propagation.Grid alias Microwaveprop.Weather.Grib2.Wgrib2 alias Microwaveprop.Weather.HrrrClient diff --git a/lib/microwaveprop/weather/hrrr_client.ex b/lib/microwaveprop/weather/hrrr_client.ex index 953feec6..9e8ff3c7 100644 --- a/lib/microwaveprop/weather/hrrr_client.ex +++ b/lib/microwaveprop/weather/hrrr_client.ex @@ -86,6 +86,14 @@ defmodule Microwaveprop.Weather.HrrrClient do @spec fetch_grid([{float(), float()}], DateTime.t(), keyword()) :: {:ok, %{{float(), float()} => map()}} | {:error, term()} def fetch_grid(points, run_time, opts \\ []) do + Microwaveprop.Instrument.span( + [:hrrr, :fetch_grid], + %{point_count: length(points), forecast_hour: Keyword.get(opts, :forecast_hour, 0)}, + fn -> do_fetch_grid(points, run_time, opts) end + ) + end + + defp do_fetch_grid(points, run_time, opts) do hour_dt = nearest_hrrr_hour(run_time) date = DateTime.to_date(hour_dt) hour = hour_dt.hour @@ -119,6 +127,14 @@ defmodule Microwaveprop.Weather.HrrrClient do @spec fetch_profile(float(), float(), DateTime.t()) :: {:ok, map()} | {:error, term()} def fetch_profile(lat, lon, valid_time) do + Microwaveprop.Instrument.span( + [:hrrr, :fetch_profile], + %{lat: lat, lon: lon}, + fn -> do_fetch_profile(lat, lon, valid_time) end + ) + end + + defp do_fetch_profile(lat, lon, valid_time) do hour_dt = nearest_hrrr_hour(valid_time) date = DateTime.to_date(hour_dt) hour = hour_dt.hour @@ -373,16 +389,18 @@ defmodule Microwaveprop.Weather.HrrrClient do end defp fetch_idx(url) do - case Req.get(url, req_options()) do - {:ok, %{status: 200, body: body}} -> - {:ok, body} + Microwaveprop.Instrument.span([:hrrr, :fetch_idx], %{url: url}, fn -> + case Req.get(url, req_options()) do + {:ok, %{status: 200, body: body}} -> + {:ok, body} - {:ok, %{status: status}} -> - {:error, "HRRR idx HTTP #{status}"} + {:ok, %{status: status}} -> + {:error, "HRRR idx HTTP #{status}"} - {:error, reason} -> - {:error, reason} - end + {:error, reason} -> + {:error, reason} + end + end) end @doc """ diff --git a/lib/microwaveprop/weather/iem_client.ex b/lib/microwaveprop/weather/iem_client.ex index 120d2791..9c7cac22 100644 --- a/lib/microwaveprop/weather/iem_client.ex +++ b/lib/microwaveprop/weather/iem_client.ex @@ -56,32 +56,36 @@ defmodule Microwaveprop.Weather.IemClient do def fetch_asos(station_id, start_dt, end_dt) do url = asos_url(station_id, start_dt, end_dt) - case Req.get(url, req_options()) do - {:ok, %{status: 200, body: body}} -> - {:ok, parse_asos_csv(body)} + Microwaveprop.Instrument.span([:iem, :fetch_asos], %{station: station_id}, fn -> + case Req.get(url, req_options()) do + {:ok, %{status: 200, body: body}} -> + {:ok, parse_asos_csv(body)} - {:ok, %{status: status}} -> - {:error, "IEM ASOS HTTP #{status}"} + {:ok, %{status: status}} -> + {:error, "IEM ASOS HTTP #{status}"} - {:error, reason} -> - {:error, reason} - end + {:error, reason} -> + {:error, reason} + end + end) end @spec fetch_raob(String.t(), DateTime.t()) :: {:ok, [map()]} | {:error, term()} def fetch_raob(station_id, dt) do url = raob_url(station_id, dt) - case Req.get(url, req_options()) do - {:ok, %{status: 200, body: body}} -> - {:ok, parse_raob_json(body)} + Microwaveprop.Instrument.span([:iem, :fetch_raob], %{station: station_id}, fn -> + case Req.get(url, req_options()) do + {:ok, %{status: 200, body: body}} -> + {:ok, parse_raob_json(body)} - {:ok, %{status: status}} -> - {:error, "IEM RAOB HTTP #{status}"} + {:ok, %{status: status}} -> + {:error, "IEM RAOB HTTP #{status}"} - {:error, reason} -> - {:error, reason} - end + {:error, reason} -> + {:error, reason} + end + end) end @spec fetch_iemre(float(), float(), Date.t()) :: {:ok, [map()]} | {:error, term()} diff --git a/lib/microwaveprop/weather/narr_client.ex b/lib/microwaveprop/weather/narr_client.ex index 38e3db36..a29e2462 100644 --- a/lib/microwaveprop/weather/narr_client.ex +++ b/lib/microwaveprop/weather/narr_client.ex @@ -212,12 +212,14 @@ defmodule Microwaveprop.Weather.NarrClient do @spec fetch_profile_at(DateTime.t(), {float(), float()}) :: {:ok, map()} | {:error, term()} def fetch_profile_at(%DateTime{} = valid_time, {lat, lon}) when is_number(lat) and is_number(lon) do - grb_url = url_for(valid_time) + Microwaveprop.Instrument.span([:narr, :fetch_profile], %{lat: lat, lon: lon}, fn -> + grb_url = url_for(valid_time) - with {:ok, index} <- fetch_inventory(valid_time), - {:ok, record_keys} <- pick_records(index) do - do_fetch_profile(grb_url, index, record_keys, lat, lon) - end + with {:ok, index} <- fetch_inventory(valid_time), + {:ok, record_keys} <- pick_records(index) do + do_fetch_profile(grb_url, index, record_keys, lat, lon) + end + end) end defp do_fetch_profile(grb_url, index, record_keys, lat, lon) do diff --git a/lib/microwaveprop/weather/nexrad_client.ex b/lib/microwaveprop/weather/nexrad_client.ex index 403b4f0e..14646241 100644 --- a/lib/microwaveprop/weather/nexrad_client.ex +++ b/lib/microwaveprop/weather/nexrad_client.ex @@ -33,20 +33,20 @@ defmodule Microwaveprop.Weather.NexradClient do req_opts = Application.get_env(:microwaveprop, :nexrad_req_options, []) - Logger.info("NexradClient: fetching #{url}") + Microwaveprop.Instrument.span([:nexrad, :fetch_frame], %{}, fn -> + case Req.get(url, [receive_timeout: 60_000, retry: false] ++ req_opts) do + {:ok, %{status: 200, body: body}} when is_binary(body) -> + process_frame(body, rounded, points_of_interest) - case Req.get(url, [receive_timeout: 60_000, retry: false] ++ req_opts) do - {:ok, %{status: 200, body: body}} when is_binary(body) -> - process_frame(body, rounded, points_of_interest) + {:ok, %{status: status}} -> + Logger.warning("NexradClient: HTTP #{status} for #{url}") + {:error, "NEXRAD n0q HTTP #{status}"} - {:ok, %{status: status}} -> - Logger.warning("NexradClient: HTTP #{status} for #{url}") - {:error, "NEXRAD n0q HTTP #{status}"} - - {:error, reason} -> - Logger.warning("NexradClient: transport error for #{url}: #{inspect(reason)}") - {:error, reason} - end + {:error, reason} -> + Logger.warning("NexradClient: transport error for #{url}: #{inspect(reason)}") + {:error, reason} + end + end) end @doc "Round a DateTime down to the nearest 5-minute boundary." @@ -111,22 +111,31 @@ defmodule Microwaveprop.Weather.NexradClient do defp fetch_and_decode_frame(rounded) do url = frame_url(rounded) + + Microwaveprop.Instrument.span([:nexrad, :fetch_frame], %{}, fn -> + handle_decode_frame_response(url, do_fetch_png(url)) + end) + end + + defp do_fetch_png(url) do req_opts = Application.get_env(:microwaveprop, :nexrad_req_options, []) + Req.get(url, [receive_timeout: 60_000, retry: false] ++ req_opts) + end - Logger.info("NexradClient: fetching #{url}") + defp handle_decode_frame_response(_url, {:ok, %{status: 200, body: body}}) when is_binary(body) do + Microwaveprop.Instrument.span([:nexrad, :decode_png], %{bytes: byte_size(body)}, fn -> + decode_png_to_pixels(body) + end) + end - case Req.get(url, [receive_timeout: 60_000, retry: false] ++ req_opts) do - {:ok, %{status: 200, body: body}} when is_binary(body) -> - decode_png_to_pixels(body) + defp handle_decode_frame_response(url, {:ok, %{status: status}}) do + Logger.warning("NexradClient: HTTP #{status} for #{url}") + {:error, "NEXRAD n0q HTTP #{status}"} + end - {:ok, %{status: status}} -> - Logger.warning("NexradClient: HTTP #{status} for #{url}") - {:error, "NEXRAD n0q HTTP #{status}"} - - {:error, reason} -> - Logger.warning("NexradClient: transport error for #{url}: #{inspect(reason)}") - {:error, reason} - end + defp handle_decode_frame_response(url, {:error, reason}) do + Logger.warning("NexradClient: transport error for #{url}: #{inspect(reason)}") + {:error, reason} end defp extract_rain_cells(pixels, width, center_lat, center_lon, radius_km, min_dbz) do diff --git a/lib/microwaveprop/weather/uwyo_sounding_client.ex b/lib/microwaveprop/weather/uwyo_sounding_client.ex index 0072d3f6..cdd9ef6a 100644 --- a/lib/microwaveprop/weather/uwyo_sounding_client.ex +++ b/lib/microwaveprop/weather/uwyo_sounding_client.ex @@ -33,13 +33,15 @@ defmodule Microwaveprop.Weather.UwyoSoundingClient do @spec fetch_sounding(String.t(), DateTime.t()) :: {:ok, [sounding()]} | {:error, term()} def fetch_sounding(station_code, %DateTime{} = dt) do - url = sounding_url(station_code, dt) + Microwaveprop.Instrument.span([:uwyo, :fetch_sounding], %{station: station_code}, fn -> + url = sounding_url(station_code, dt) - case Req.get(url, req_options()) do - {:ok, %{status: 200, body: body}} -> {:ok, parse_sounding_html(body)} - {:ok, %{status: status}} -> {:error, "UWYO sounding HTTP #{status}"} - {:error, reason} -> {:error, reason} - end + case Req.get(url, req_options()) do + {:ok, %{status: 200, body: body}} -> {:ok, parse_sounding_html(body)} + {:ok, %{status: status}} -> {:error, "UWYO sounding HTTP #{status}"} + {:error, reason} -> {:error, reason} + end + end) end @spec parse_sounding_html(String.t()) :: [sounding()] diff --git a/lib/microwaveprop/workers/common_volume_radar_worker.ex b/lib/microwaveprop/workers/common_volume_radar_worker.ex index da8a8373..1fbfcf6c 100644 --- a/lib/microwaveprop/workers/common_volume_radar_worker.ex +++ b/lib/microwaveprop/workers/common_volume_radar_worker.ex @@ -77,8 +77,6 @@ defmodule Microwaveprop.Workers.CommonVolumeRadarWorker do qso_at = contact.qso_timestamp rounded = NexradClient.round_to_5min(DateTime.from_naive!(qso_at, "Etc/UTC")) - Logger.info("CommonVolumeRadarWorker: #{contact.id} fetching n0q frame for #{DateTime.to_iso8601(rounded)}") - case NexradClient.fetch_decoded_frame(rounded) do {:ok, pixels, width} -> stats = aggregate_stats(pixels, width, pos1, pos2) @@ -121,6 +119,12 @@ defmodule Microwaveprop.Workers.CommonVolumeRadarWorker do coverage_pct: float() | nil } def aggregate_stats(pixels, width, pos1, pos2, opts \\ []) do + Microwaveprop.Instrument.span([:radar, :aggregate_stats], %{}, fn -> + do_aggregate_stats(pixels, width, pos1, pos2, opts) + end) + end + + defp do_aggregate_stats(pixels, width, pos1, pos2, opts) do step = Keyword.get(opts, :step, @pixel_step) case CommonVolume.bounding_box(pos1, pos2, @radius_km) do diff --git a/lib/microwaveprop/workers/propagation_grid_worker.ex b/lib/microwaveprop/workers/propagation_grid_worker.ex index f1b9ff6d..bb4f3c0a 100644 --- a/lib/microwaveprop/workers/propagation_grid_worker.ex +++ b/lib/microwaveprop/workers/propagation_grid_worker.ex @@ -413,16 +413,23 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do end defp compute_scores_algorithm(grid_data, valid_time, include_factors?) do - grid_data - |> Task.async_stream( - &score_one_point(&1, valid_time, include_factors?), - max_concurrency: System.schedulers_online() * 2, - timeout: 30_000 + Microwaveprop.Instrument.span( + [:propagation_grid, :score_band], + %{point_count: length(grid_data)}, + fn -> + grid_data + |> Task.async_stream( + &score_one_point(&1, valid_time, include_factors?), + max_concurrency: System.schedulers_online() * 2, + timeout: 30_000 + ) + |> Stream.flat_map(fn + {:ok, results} -> results + {:exit, _reason} -> [] + end) + |> Enum.to_list() + end ) - |> Stream.flat_map(fn - {:ok, results} -> results - {:exit, _reason} -> [] - end) end defp score_one_point({{lat, lon}, profile}, valid_time, include_factors?) do diff --git a/lib/microwaveprop_web/telemetry.ex b/lib/microwaveprop_web/telemetry.ex index 3a76d1f6..902ca194 100644 --- a/lib/microwaveprop_web/telemetry.ex +++ b/lib/microwaveprop_web/telemetry.ex @@ -22,14 +22,19 @@ defmodule MicrowavepropWeb.Telemetry do end def metrics do + phoenix_metrics() ++ + ecto_metrics() ++ + vm_metrics() ++ + oban_metrics() ++ + http_client_metrics() ++ + db_batch_metrics() ++ + worker_phase_metrics() + end + + defp phoenix_metrics do [ - # Phoenix Metrics - summary("phoenix.endpoint.start.system_time", - unit: {:native, :millisecond} - ), - summary("phoenix.endpoint.stop.duration", - unit: {:native, :millisecond} - ), + summary("phoenix.endpoint.start.system_time", unit: {:native, :millisecond}), + summary("phoenix.endpoint.stop.duration", unit: {:native, :millisecond}), summary("phoenix.router_dispatch.start.system_time", tags: [:route], unit: {:native, :millisecond} @@ -42,19 +47,18 @@ defmodule MicrowavepropWeb.Telemetry do tags: [:route], unit: {:native, :millisecond} ), - summary("phoenix.socket_connected.duration", - unit: {:native, :millisecond} - ), + summary("phoenix.socket_connected.duration", unit: {:native, :millisecond}), sum("phoenix.socket_drain.count"), - summary("phoenix.channel_joined.duration", - unit: {:native, :millisecond} - ), + summary("phoenix.channel_joined.duration", unit: {:native, :millisecond}), summary("phoenix.channel_handled_in.duration", tags: [:event], unit: {:native, :millisecond} - ), + ) + ] + end - # Database Metrics + defp ecto_metrics do + [ summary("microwaveprop.repo.query.total_time", unit: {:native, :millisecond}, description: "The sum of the other measurements" @@ -74,9 +78,12 @@ defmodule MicrowavepropWeb.Telemetry do summary("microwaveprop.repo.query.idle_time", unit: {:native, :millisecond}, description: "The time the connection spent waiting before being checked out for the query" - ), + ) + ] + end - # VM Metrics + defp vm_metrics do + [ summary("vm.memory.total", unit: {:byte, :kilobyte}), summary("vm.total_run_queue_lengths.total"), summary("vm.total_run_queue_lengths.cpu"), @@ -84,11 +91,192 @@ defmodule MicrowavepropWeb.Telemetry do ] end - defp periodic_measurements do + # Oban emits [:oban, :job, :start|:stop|:exception]. `stop` measurements + # include both `duration` (job runtime) and `queue_time` (available → + # executing latency). Tagging by worker + queue gives us per-worker + # distributions without per-worker metric registration. + defp oban_metrics do [ - # A module, function and arguments to be invoked periodically. - # This function must call :telemetry.execute/3 and a metric must be added above. - # {MicrowavepropWeb, :count_users, []} + summary("oban.job.stop.duration", + event_name: [:oban, :job, :stop], + measurement: :duration, + tags: [:worker, :queue, :state], + tag_values: &oban_tag_values/1, + unit: {:native, :millisecond} + ), + summary("oban.job.stop.queue_time", + event_name: [:oban, :job, :stop], + measurement: :queue_time, + tags: [:worker, :queue], + tag_values: &oban_tag_values/1, + unit: {:native, :millisecond} + ), + counter("oban.job.stop.count", + event_name: [:oban, :job, :stop], + tags: [:worker, :queue, :state], + tag_values: &oban_tag_values/1 + ), + counter("oban.job.exception.count", + event_name: [:oban, :job, :exception], + tags: [:worker, :queue], + tag_values: &oban_tag_values/1 + ), + last_value("microwaveprop.oban.queue.count", + event_name: [:microwaveprop, :oban, :queue, :depth], + measurement: :count, + tags: [:queue, :state], + description: "Oban job count per (queue, state) sampled every 10s" + ) ] end + + # HTTP client spans emitted via `Microwaveprop.Instrument`. Each client + # wraps its own public entry points so we get independent latency + # histograms per upstream. + defp http_client_metrics do + [ + # HRRR pressure-level fetch (batch-of-points via NOMADS/S3 + wgrib2) + summary("microwaveprop.hrrr.fetch_grid.stop.duration", + unit: {:native, :millisecond}, + tags: [:result] + ), + summary("microwaveprop.hrrr.fetch_profile.stop.duration", + unit: {:native, :millisecond}, + tags: [:result] + ), + summary("microwaveprop.hrrr.fetch_idx.stop.duration", + unit: {:native, :millisecond}, + tags: [:result] + ), + + # NEXRAD n0q frame (IEM archive PNG) + summary("microwaveprop.nexrad.fetch_frame.stop.duration", + unit: {:native, :millisecond}, + tags: [:result] + ), + summary("microwaveprop.nexrad.decode_png.stop.duration", + unit: {:native, :millisecond}, + tags: [:result] + ), + + # IEM Mesonet ASOS + RAOB + summary("microwaveprop.iem.fetch_asos.stop.duration", + unit: {:native, :millisecond}, + tags: [:result] + ), + summary("microwaveprop.iem.fetch_raob.stop.duration", + unit: {:native, :millisecond}, + tags: [:result] + ), + + # GEFS extended-horizon ensemble + summary("microwaveprop.gefs.fetch_grid_profiles.stop.duration", + unit: {:native, :millisecond}, + tags: [:result] + ), + + # NARR historical backfill + summary("microwaveprop.narr.fetch_profile.stop.duration", + unit: {:native, :millisecond}, + tags: [:result] + ), + + # University of Wyoming soundings (Canadian RAOB backfill) + summary("microwaveprop.uwyo.fetch_sounding.stop.duration", + unit: {:native, :millisecond}, + tags: [:result] + ), + + # Terrain elevation API (Open-Meteo or local SRTM) + summary("microwaveprop.elevation.fetch_profile.stop.duration", + unit: {:native, :millisecond}, + tags: [:result] + ) + ] + end + + defp db_batch_metrics do + [ + summary("microwaveprop.db.upsert_hrrr_profiles.stop.duration", + unit: {:native, :millisecond}, + tags: [:result] + ), + summary("microwaveprop.db.upsert_gefs_profiles.stop.duration", + unit: {:native, :millisecond}, + tags: [:result] + ), + summary("microwaveprop.db.upsert_surface_observations.stop.duration", + unit: {:native, :millisecond}, + tags: [:result] + ), + summary("microwaveprop.db.upsert_sounding.stop.duration", + unit: {:native, :millisecond}, + tags: [:result] + ), + summary("microwaveprop.db.replace_scores.stop.duration", + unit: {:native, :millisecond}, + tags: [:result] + ) + ] + end + + defp worker_phase_metrics do + [ + summary("microwaveprop.propagation_grid.score_band.stop.duration", + unit: {:native, :millisecond}, + tags: [:result] + ), + summary("microwaveprop.terrain.analyse.stop.duration", + unit: {:native, :millisecond}, + tags: [:result] + ), + summary("microwaveprop.radar.aggregate_stats.stop.duration", + unit: {:native, :millisecond}, + tags: [:result] + ) + ] + end + + defp oban_tag_values(%{job: %Oban.Job{worker: worker, queue: queue}} = meta) do + %{ + worker: worker || "unknown", + queue: queue || "unknown", + state: Map.get(meta, :state, :unknown) + } + end + + defp oban_tag_values(_), do: %{worker: "unknown", queue: "unknown", state: :unknown} + + defp periodic_measurements do + [ + {__MODULE__, :dispatch_oban_queue_depth, []} + ] + end + + @doc false + # Periodic poller: emit `microwaveprop.oban.queue.depth` gauges per + # (queue, state) tuple so LiveDashboard / Prometheus can chart queue + # saturation over time. Cheap aggregation query, runs every 10s. + def dispatch_oban_queue_depth do + import Ecto.Query + + query = + from(j in "oban_jobs", + group_by: [j.queue, j.state], + select: {j.queue, j.state, count(j.id)} + ) + + for {queue, state, count} <- Microwaveprop.Repo.all(query) do + :telemetry.execute( + [:microwaveprop, :oban, :queue, :depth], + %{count: count}, + %{queue: queue, state: state} + ) + end + + :ok + rescue + # Don't crash the poller if the DB blips — telemetry is best-effort. + _ -> :ok + end end