graham/prop
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(telemetry): broaden Instrument span coverage

Browse source 
Adds spans to 15 previously-unmeasured hot paths so every question we
might ask while tuning has a histogram to answer it:

External I/O:
- iem.fetch_iemre (gridded weather reanalysis)
- mrms.list_latest / mrms.download (precip radar)
- rtma.fetch_observation
- ncei.fetch_metar (historical 5-min METAR backfill)
- solar.fetch_indices (GFZ solar indices)
- swpc.fetch (SWPC Kp/F10.7/X-ray)
- giro.fetch (ionosonde)
- qrz.request, geocoder.geocode (callsign enrichment)
- srtm.download_tile (terrain tile download + gunzip)
- hrrr.download_grib_ranges (parallel byte-range fetch phase)

Subprocess:
- wgrib2.extract_grid / extract_grid_from_file / extract_grid_from_file_mapped

LiveView hot paths:
- propagation.scores_at (map score fetch + cache hit/miss counter)
- propagation.point_forecast (sparkline)
- propagation.point_detail (click-to-inspect)
- propagation.daily_outlook_at (/map outlook strip)

Worker-level end-to-end:
- worker.terrain_profile
- worker.mechanism_classify
- worker.mrms_fetch

Each event is registered in Microwaveprop.PromEx.InstrumentPlugin as
a Prometheus histogram (default / long buckets as appropriate) plus
a counter for the scores_at cache hit/miss ratio. Prometheus at
10.0.15.25 will start seeing the new series on the next scrape after
deploy.
This commit is contained in:
Graham McIntire 2026-04-18 17:25:32 -05:00
parent d9ffda43d3
commit 4e6c87eca2
No known key found for this signature in database
GPG key ID: F4ABF488E6029E59
17 changed files with 414 additions and 155 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

27
lib/microwaveprop/geocoder.ex
View file

@ -7,20 +7,25 @@ defmodule Microwaveprop.Geocoder do
@spec geocode(String.t()) :: {:ok, %{lat: float(), lon: float()}} | {:error, String.t()}
def geocode(address) do
case Req.get(client(), url: "/maps/api/geocode/json", params: [address: address, key: api_key()]) do
{:ok, %{status: 200, body: %{"status" => "OK", "results" => [first | _]}}} ->
%{"geometry" => %{"location" => %{"lat" => lat, "lng" => lon}}} = first
{:ok, %{lat: lat, lon: lon}}
Microwaveprop.Instrument.span([:geocoder, :geocode], %{}, fn ->
case Req.get(client(),
url: "/maps/api/geocode/json",
params: [address: address, key: api_key()]
) do
{:ok, %{status: 200, body: %{"status" => "OK", "results" => [first | _]}}} ->
%{"geometry" => %{"location" => %{"lat" => lat, "lng" => lon}}} = first
{:ok, %{lat: lat, lon: lon}}
{:ok, %{status: 200, body: %{"status" => "ZERO_RESULTS"}}} ->
{:error, "No results found"}
{:ok, %{status: 200, body: %{"status" => "ZERO_RESULTS"}}} ->
{:error, "No results found"}
{:ok, %{status: 200, body: %{"status" => status}}} ->
{:error, "Geocoding failed: #{status}"}
{:ok, %{status: 200, body: %{"status" => status}}} ->
{:error, "Geocoding failed: #{status}"}
{:error, reason} ->
{:error, "Geocoding request failed: #{inspect(reason)}"}
end
{:error, reason} ->
{:error, "Geocoding request failed: #{inspect(reason)}"}
end
end)
end
defp client do

20
lib/microwaveprop/ionosphere/giro_client.ex
View file

@ -53,18 +53,20 @@ defmodule Microwaveprop.Ionosphere.GiroClient do
@spec fetch(String.t(), DateTime.t(), DateTime.t()) ::
{:ok, [observation()]} | {:error, term()}
def fetch(ursi_code, %DateTime{} = from_dt, %DateTime{} = to_dt) do
url = build_url(ursi_code, from_dt, to_dt)
Microwaveprop.Instrument.span([:giro, :fetch], %{station: ursi_code}, fn ->
url = build_url(ursi_code, from_dt, to_dt)
case Req.get(url, [receive_timeout: @default_timeout_ms] ++ req_options()) do
{:ok, %{status: 200, body: body}} when is_binary(body) ->
parse_tabular(body)
case Req.get(url, [receive_timeout: @default_timeout_ms] ++ req_options()) do
{:ok, %{status: 200, body: body}} when is_binary(body) ->
parse_tabular(body)
{:ok, %{status: status, body: body}} ->
{:error, "GIRO HTTP #{status}: #{inspect(body)}"}
{:ok, %{status: status, body: body}} ->
{:error, "GIRO HTTP #{status}: #{inspect(body)}"}
{:error, reason} ->
{:error, "GIRO request failed: #{inspect(reason)}"}
end
{:error, reason} ->
{:error, "GIRO request failed: #{inspect(reason)}"}
end
end)
end
@doc """

182
lib/microwaveprop/prom_ex/instrument_plugin.ex
View file

@ -13,8 +13,13 @@ defmodule Microwaveprop.PromEx.InstrumentPlugin do
def event_metrics(_opts) do
[
http_client_events(),
more_http_client_events(),
subprocess_events(),
db_batch_events(),
worker_phase_events(),
top_level_worker_events(),
liveview_hot_path_events(),
cache_hit_events(),
oban_queue_depth_event()
]
end
@ -96,6 +101,183 @@ defmodule Microwaveprop.PromEx.InstrumentPlugin do
)
end
defp more_http_client_events do
Event.build(
:microwaveprop_more_http_client_events,
[
distribution("microwaveprop.iem.fetch_iemre.duration.milliseconds",
event_name: [:microwaveprop, :iem, :fetch_iemre, :stop],
measurement: :duration,
unit: {:native, :millisecond},
reporter_options: [buckets: default_buckets()]
),
distribution("microwaveprop.mrms.list_latest.duration.milliseconds",
event_name: [:microwaveprop, :mrms, :list_latest, :stop],
measurement: :duration,
unit: {:native, :millisecond},
reporter_options: [buckets: default_buckets()]
),
distribution("microwaveprop.mrms.download.duration.milliseconds",
event_name: [:microwaveprop, :mrms, :download, :stop],
measurement: :duration,
unit: {:native, :millisecond},
reporter_options: [buckets: default_buckets()]
),
distribution("microwaveprop.rtma.fetch_observation.duration.milliseconds",
event_name: [:microwaveprop, :rtma, :fetch_observation, :stop],
measurement: :duration,
unit: {:native, :millisecond},
reporter_options: [buckets: default_buckets()]
),
distribution("microwaveprop.ncei.fetch_metar.duration.milliseconds",
event_name: [:microwaveprop, :ncei, :fetch_metar, :stop],
measurement: :duration,
unit: {:native, :millisecond},
reporter_options: [buckets: default_buckets()]
),
distribution("microwaveprop.solar.fetch_indices.duration.milliseconds",
event_name: [:microwaveprop, :solar, :fetch_indices, :stop],
measurement: :duration,
unit: {:native, :millisecond},
reporter_options: [buckets: default_buckets()]
),
distribution("microwaveprop.swpc.fetch.duration.milliseconds",
event_name: [:microwaveprop, :swpc, :fetch, :stop],
measurement: :duration,
unit: {:native, :millisecond},
reporter_options: [buckets: default_buckets()]
),
distribution("microwaveprop.giro.fetch.duration.milliseconds",
event_name: [:microwaveprop, :giro, :fetch, :stop],
measurement: :duration,
unit: {:native, :millisecond},
reporter_options: [buckets: default_buckets()]
),
distribution("microwaveprop.qrz.request.duration.milliseconds",
event_name: [:microwaveprop, :qrz, :request, :stop],
measurement: :duration,
unit: {:native, :millisecond},
reporter_options: [buckets: default_buckets()]
),
distribution("microwaveprop.geocoder.geocode.duration.milliseconds",
event_name: [:microwaveprop, :geocoder, :geocode, :stop],
measurement: :duration,
unit: {:native, :millisecond},
reporter_options: [buckets: default_buckets()]
),
distribution("microwaveprop.srtm.download_tile.duration.milliseconds",
event_name: [:microwaveprop, :srtm, :download_tile, :stop],
measurement: :duration,
unit: {:native, :millisecond},
reporter_options: [buckets: default_buckets()]
),
distribution("microwaveprop.hrrr.download_grib_ranges.duration.milliseconds",
event_name: [:microwaveprop, :hrrr, :download_grib_ranges, :stop],
measurement: :duration,
unit: {:native, :millisecond},
reporter_options: [buckets: default_buckets()]
)
]
)
end
defp subprocess_events do
Event.build(
:microwaveprop_subprocess_events,
[
distribution("microwaveprop.wgrib2.extract_grid.duration.milliseconds",
event_name: [:microwaveprop, :wgrib2, :extract_grid, :stop],
measurement: :duration,
unit: {:native, :millisecond},
reporter_options: [buckets: long_buckets()]
),
distribution("microwaveprop.wgrib2.extract_grid_from_file.duration.milliseconds",
event_name: [:microwaveprop, :wgrib2, :extract_grid_from_file, :stop],
measurement: :duration,
unit: {:native, :millisecond},
reporter_options: [buckets: long_buckets()]
),
distribution("microwaveprop.wgrib2.extract_grid_from_file_mapped.duration.milliseconds",
event_name: [:microwaveprop, :wgrib2, :extract_grid_from_file_mapped, :stop],
measurement: :duration,
unit: {:native, :millisecond},
reporter_options: [buckets: long_buckets()]
)
]
)
end
defp top_level_worker_events do
Event.build(
:microwaveprop_top_level_worker_events,
[
distribution("microwaveprop.worker.terrain_profile.duration.milliseconds",
event_name: [:microwaveprop, :worker, :terrain_profile, :stop],
measurement: :duration,
unit: {:native, :millisecond},
reporter_options: [buckets: default_buckets()]
),
distribution("microwaveprop.worker.mechanism_classify.duration.milliseconds",
event_name: [:microwaveprop, :worker, :mechanism_classify, :stop],
measurement: :duration,
unit: {:native, :millisecond},
reporter_options: [buckets: default_buckets()]
),
distribution("microwaveprop.worker.mrms_fetch.duration.milliseconds",
event_name: [:microwaveprop, :worker, :mrms_fetch, :stop],
measurement: :duration,
unit: {:native, :millisecond},
reporter_options: [buckets: long_buckets()]
)
]
)
end
defp liveview_hot_path_events do
Event.build(
:microwaveprop_liveview_hot_path_events,
[
distribution("microwaveprop.propagation.scores_at.duration.milliseconds",
event_name: [:microwaveprop, :propagation, :scores_at, :stop],
measurement: :duration,
unit: {:native, :millisecond},
reporter_options: [buckets: default_buckets()]
),
distribution("microwaveprop.propagation.point_forecast.duration.milliseconds",
event_name: [:microwaveprop, :propagation, :point_forecast, :stop],
measurement: :duration,
unit: {:native, :millisecond},
reporter_options: [buckets: default_buckets()]
),
distribution("microwaveprop.propagation.point_detail.duration.milliseconds",
event_name: [:microwaveprop, :propagation, :point_detail, :stop],
measurement: :duration,
unit: {:native, :millisecond},
reporter_options: [buckets: default_buckets()]
),
distribution("microwaveprop.propagation.daily_outlook_at.duration.milliseconds",
event_name: [:microwaveprop, :propagation, :daily_outlook_at, :stop],
measurement: :duration,
unit: {:native, :millisecond},
reporter_options: [buckets: default_buckets()]
)
]
)
end
defp cache_hit_events do
Event.build(
:microwaveprop_cache_hit_events,
[
counter("microwaveprop.propagation.scores_at.cache.count",
event_name: [:microwaveprop, :propagation, :scores_at, :cache],
tags: [:hit],
description: "ScoreCache hit/miss for the map's scores_at fetch"
)
]
)
end
defp db_batch_events do
Event.build(
:microwaveprop_db_batch_events,

67
lib/microwaveprop/propagation.ex
View file

@ -249,23 +249,24 @@ defmodule Microwaveprop.Propagation do
@spec scores_at(non_neg_integer(), DateTime.t() | nil, map() | nil) ::
[%{lat: float(), lon: float(), score: non_neg_integer(), valid_time: DateTime.t()}]
def scores_at(band_mhz, valid_time, bounds \\ nil) do
time = valid_time || earliest_valid_time(band_mhz)
Microwaveprop.Instrument.span([:propagation, :scores_at], %{band_mhz: band_mhz}, fn ->
time = valid_time || earliest_valid_time(band_mhz)
case time do
nil ->
[]
_ ->
scores_at_fetch(band_mhz, time, bounds)
end
case time do
nil -> []
_ -> scores_at_fetch(band_mhz, time, bounds)
end
end)
end
defp scores_at_fetch(band_mhz, time, bounds) do
case ScoreCache.fetch_bounds(band_mhz, time, bounds) do
{:ok, scores} ->
:telemetry.execute([:microwaveprop, :propagation, :scores_at, :cache], %{}, %{hit: true})
Enum.map(scores, &Map.put(&1, :valid_time, time))
:miss ->
:telemetry.execute([:microwaveprop, :propagation, :scores_at, :cache], %{}, %{hit: false})
read_from_disk_and_cache(band_mhz, time, bounds)
end
end
@ -341,34 +342,40 @@ defmodule Microwaveprop.Propagation do
@spec daily_outlook_at(float(), float(), keyword()) ::
[%{date: Date.t(), peak_score: non_neg_integer()}]
def daily_outlook_at(lat, lon, opts \\ []) do
band_mhz = Keyword.get(opts, :band_mhz, 10_000)
days = Keyword.get(opts, :days, 7)
today = DateTime.to_date(DateTime.utc_now())
horizon = Date.add(today, days - 1)
Microwaveprop.Instrument.span([:propagation, :daily_outlook_at], %{}, fn ->
band_mhz = Keyword.get(opts, :band_mhz, 10_000)
days = Keyword.get(opts, :days, 7)
today = DateTime.to_date(DateTime.utc_now())
horizon = Date.add(today, days - 1)
band_mhz
|> point_forecast(lat, lon)
|> Enum.group_by(fn entry ->
DateTime.to_date(entry.valid_time)
band_mhz
|> point_forecast(lat, lon)
|> Enum.group_by(fn entry ->
DateTime.to_date(entry.valid_time)
end)
|> Enum.filter(fn {date, _} ->
Date.compare(date, today) != :lt and Date.compare(date, horizon) != :gt
end)
|> Enum.map(fn {date, entries} ->
%{date: date, peak_score: entries |> Enum.map(& &1.score) |> Enum.max()}
end)
|> Enum.sort_by(& &1.date, Date)
end)
|> Enum.filter(fn {date, _} -> Date.compare(date, today) != :lt and Date.compare(date, horizon) != :gt end)
|> Enum.map(fn {date, entries} ->
%{date: date, peak_score: entries |> Enum.map(& &1.score) |> Enum.max()}
end)
|> Enum.sort_by(& &1.date, Date)
end
@doc "Get scores across all forecast hours for a single grid point (for sparkline)."
@spec point_forecast(non_neg_integer(), float(), float()) ::
[%{valid_time: DateTime.t(), score: non_neg_integer()}]
def point_forecast(band_mhz, lat, lon) do
{snapped_lat, snapped_lon} = snap_to_grid(lat, lon)
now = DateTime.utc_now()
Microwaveprop.Instrument.span([:propagation, :point_forecast], %{band_mhz: band_mhz}, fn ->
{snapped_lat, snapped_lon} = snap_to_grid(lat, lon)
now = DateTime.utc_now()
case ScoreCache.valid_times(band_mhz) do
[] -> point_forecast_from_store(band_mhz, snapped_lat, snapped_lon, now)
cached -> point_forecast_from_cache(band_mhz, snapped_lat, snapped_lon, now, cached)
end
case ScoreCache.valid_times(band_mhz) do
[] -> point_forecast_from_store(band_mhz, snapped_lat, snapped_lon, now)
cached -> point_forecast_from_cache(band_mhz, snapped_lat, snapped_lon, now, cached)
end
end)
end
defp point_forecast_from_store(band_mhz, lat, lon, now) do
@ -428,6 +435,12 @@ defmodule Microwaveprop.Propagation do
}
| nil
def point_detail(band_mhz, lat, lon, valid_time \\ nil) do
Microwaveprop.Instrument.span([:propagation, :point_detail], %{band_mhz: band_mhz}, fn ->
do_point_detail(band_mhz, lat, lon, valid_time)
end)
end
defp do_point_detail(band_mhz, lat, lon, valid_time) do
{snapped_lat, snapped_lon} = snap_to_grid(lat, lon)
time = valid_time || latest_valid_time(band_mhz)

28
lib/microwaveprop/qrz/client.ex
View file

@ -82,23 +82,25 @@ defmodule Microwaveprop.Qrz.Client do
defp handle_lookup_result({:error, _} = error, _, _), do: error
defp do_request(params) do
config = Application.get_env(:microwaveprop, __MODULE__, [])
Microwaveprop.Instrument.span([:qrz, :request], %{}, fn ->
config = Application.get_env(:microwaveprop, __MODULE__, [])
req_opts =
[url: @url, params: params]
|> maybe_add_plug(config)
|> maybe_add_retry(config)
req_opts =
[url: @url, params: params]
|> maybe_add_plug(config)
|> maybe_add_retry(config)
case Req.get(req_opts) do
{:ok, %{status: 200, body: body}} ->
{:ok, body}
case Req.get(req_opts) do
{:ok, %{status: 200, body: body}} ->
{:ok, body}
{:ok, %{status: status}} ->
{:error, "HTTP #{status}"}
{:ok, %{status: status}} ->
{:error, "HTTP #{status}"}
{:error, reason} ->
{:error, "Request failed: #{inspect(reason)}"}
end
{:error, reason} ->
{:error, "Request failed: #{inspect(reason)}"}
end
end)
end
defp maybe_add_plug(opts, config) do

20
lib/microwaveprop/space_weather/swpc_client.ex
View file

@ -150,18 +150,20 @@ defmodule Microwaveprop.SpaceWeather.SwpcClient do
# ---------- Shared HTTP / parsing helpers ----------
defp fetch_and_parse(path, parser) do
url = @base_url <> path
Microwaveprop.Instrument.span([:swpc, :fetch], %{path: path}, fn ->
url = @base_url <> path
case Req.get(url, [receive_timeout: @default_timeout_ms] ++ req_options()) do
{:ok, %{status: 200, body: body}} ->
parser.(body)
case Req.get(url, [receive_timeout: @default_timeout_ms] ++ req_options()) do
{:ok, %{status: 200, body: body}} ->
parser.(body)
{:ok, %{status: status, body: body}} ->
{:error, "SWPC HTTP #{status}: #{inspect(body)}"}
{:ok, %{status: status, body: body}} ->
{:error, "SWPC HTTP #{status}: #{inspect(body)}"}
{:error, reason} ->
{:error, "SWPC request failed: #{inspect(reason)}"}
end
{:error, reason} ->
{:error, "SWPC request failed: #{inspect(reason)}"}
end
end)
end
defp decode_array(body) when is_binary(body) do

6
lib/microwaveprop/terrain/srtm.ex
View file

@ -23,6 +23,12 @@ defmodule Microwaveprop.Terrain.Srtm do
@spec download_tile(float(), float(), String.t()) :: {:ok, String.t()} | {:error, term()}
def download_tile(lat, lon, tiles_dir) do
Microwaveprop.Instrument.span([:srtm, :download_tile], %{lat: lat, lon: lon}, fn ->
do_download_tile(lat, lon, tiles_dir)
end)
end
defp do_download_tile(lat, lon, tiles_dir) do
filename = tile_filename(lat, lon)
lat_dir = String.slice(filename, 0, 3)
url = "#{@base_url}/#{lat_dir}/#{filename}.gz"

40
lib/microwaveprop/weather/grib2/wgrib2.ex
View file

@ -32,11 +32,17 @@ defmodule Microwaveprop.Weather.Grib2.Wgrib2 do
"""
@spec extract_grid(binary(), String.t(), grid_spec()) :: {:ok, point_grid()} | {:error, term()}
def extract_grid(grib_binary, match_pattern, grid_spec) do
if available?() do
extract_with_wgrib2(grib_binary, match_pattern, grid_spec)
else
{:error, :wgrib2_not_available}
end
Microwaveprop.Instrument.span(
[:wgrib2, :extract_grid],
%{bytes: byte_size(grib_binary)},
fn ->
if available?() do
extract_with_wgrib2(grib_binary, match_pattern, grid_spec)
else
{:error, :wgrib2_not_available}
end
end
)
end
@doc """
@ -47,11 +53,13 @@ defmodule Microwaveprop.Weather.Grib2.Wgrib2 do
@spec extract_grid_from_file(Path.t(), String.t(), grid_spec()) ::
{:ok, point_grid()} | {:error, term()}
def extract_grid_from_file(grib_path, match_pattern, grid_spec) do
if available?() do
extract_file_with_wgrib2(grib_path, match_pattern, grid_spec)
else
{:error, :wgrib2_not_available}
end
Microwaveprop.Instrument.span([:wgrib2, :extract_grid_from_file], %{}, fn ->
if available?() do
extract_file_with_wgrib2(grib_path, match_pattern, grid_spec)
else
{:error, :wgrib2_not_available}
end
end)
end
@doc """
@ -73,11 +81,13 @@ defmodule Microwaveprop.Weather.Grib2.Wgrib2 do
(%{String.t() => float()} -> term())
) :: {:ok, %{{float(), float()} => term()}} | {:error, term()}
def extract_grid_from_file_mapped(grib_path, match_pattern, grid_spec, cell_reducer) do
if available?() do
extract_file_mapped_with_wgrib2(grib_path, match_pattern, grid_spec, cell_reducer)
else
{:error, :wgrib2_not_available}
end
Microwaveprop.Instrument.span([:wgrib2, :extract_grid_from_file_mapped], %{}, fn ->
if available?() do
extract_file_mapped_with_wgrib2(grib_path, match_pattern, grid_spec, cell_reducer)
else
{:error, :wgrib2_not_available}
end
end)
end
@doc "Check if wgrib2 is available on the system."

27
lib/microwaveprop/weather/hrrr_client.ex
View file

@ -417,7 +417,14 @@ defmodule Microwaveprop.Weather.HrrrClient do
def download_grib_ranges(_url, []), do: {:ok, <<>>}
def download_grib_ranges(url, ranges) do
# Check local cache first (dev only)
Microwaveprop.Instrument.span(
[:hrrr, :download_grib_ranges],
%{range_count: length(ranges)},
fn -> do_download_grib_ranges(url, ranges) end
)
end
defp do_download_grib_ranges(url, ranges) do
cache_key = url_to_cache_key(url, ranges)
case read_cache(cache_key) do
@ -426,14 +433,18 @@ defmodule Microwaveprop.Weather.HrrrClient do
{:ok, binary}
:miss ->
case download_grib_ranges_remote(url, ranges) do
{:ok, binary} = ok ->
write_cache(cache_key, binary)
ok
download_and_cache_grib_ranges(url, ranges, cache_key)
end
end
error ->
error
end
defp download_and_cache_grib_ranges(url, ranges, cache_key) do
case download_grib_ranges_remote(url, ranges) do
{:ok, binary} = ok ->
write_cache(cache_key, binary)
ok
error ->
error
end
end

20
lib/microwaveprop/weather/iem_client.ex
View file

@ -90,18 +90,20 @@ defmodule Microwaveprop.Weather.IemClient do
@spec fetch_iemre(float(), float(), Date.t()) :: {:ok, [map()]} | {:error, term()}
def fetch_iemre(lat, lon, date) do
url = iemre_url(lat, lon, date)
Microwaveprop.Instrument.span([:iem, :fetch_iemre], %{}, fn ->
url = iemre_url(lat, lon, date)
case Req.get(url, req_options()) do
{:ok, %{status: 200, body: body}} ->
{:ok, parse_iemre_json(body)}
case Req.get(url, req_options()) do
{:ok, %{status: 200, body: body}} ->
{:ok, parse_iemre_json(body)}
{:ok, %{status: status}} ->
{:error, "IEM IEMRE HTTP #{status}"}
{:ok, %{status: status}} ->
{:error, "IEM IEMRE HTTP #{status}"}
{:error, reason} ->
{:error, reason}
end
{:error, reason} ->
{:error, reason}
end
end)
end
defp req_options do

22
lib/microwaveprop/weather/mrms_client.ex
View file

@ -29,6 +29,10 @@ defmodule Microwaveprop.Weather.MrmsClient do
"""
@spec list_latest(non_neg_integer()) :: {:ok, [listing_entry()]} | {:error, term()}
def list_latest(limit \\ 10) do
Microwaveprop.Instrument.span([:mrms, :list_latest], %{}, fn -> do_list_latest(limit) end)
end
defp do_list_latest(limit) do
case Req.get(@base_url, req_options()) do
{:ok, %{status: 200, body: body}} ->
entries =
@ -130,16 +134,18 @@ defmodule Microwaveprop.Weather.MrmsClient do
# the URL ends in .gz. No gunzip step needed.
url = @base_url <> filename
case Req.get(url, req_options()) do
{:ok, %{status: 200, body: body}} when is_binary(body) ->
{:ok, body}
Microwaveprop.Instrument.span([:mrms, :download], %{}, fn ->
case Req.get(url, req_options()) do
{:ok, %{status: 200, body: body}} when is_binary(body) ->
{:ok, body}
{:ok, %{status: status}} ->
{:error, {:http_status, status}}
{:ok, %{status: status}} ->
{:error, {:http_status, status}}
{:error, reason} ->
{:error, reason}
end
{:error, reason} ->
{:error, reason}
end
end)
end
defp write_temp(grib_bytes) do

22
lib/microwaveprop/weather/ncei_metar_client.ex
View file

@ -23,19 +23,21 @@ defmodule Microwaveprop.Weather.NceiMetarClient do
"""
@spec fetch(String.t(), pos_integer(), pos_integer()) :: {:ok, [map()]} | {:error, term()}
def fetch(icao, year, month) do
month_str = month |> Integer.to_string() |> String.pad_leading(2, "0")
url = "#{@base_url}/#{year}/#{month_str}/asos-5min-#{icao}-#{year}#{month_str}.dat"
Microwaveprop.Instrument.span([:ncei, :fetch_metar], %{station: icao}, fn ->
month_str = month |> Integer.to_string() |> String.pad_leading(2, "0")
url = "#{@base_url}/#{year}/#{month_str}/asos-5min-#{icao}-#{year}#{month_str}.dat"
case Req.get(url, receive_timeout: 60_000) do
{:ok, %{status: 200, body: body}} ->
{:ok, parse(body)}
case Req.get(url, receive_timeout: 60_000) do
{:ok, %{status: 200, body: body}} ->
{:ok, parse(body)}
{:ok, %{status: status}} ->
{:error, "NCEI 5-min HTTP #{status}"}
{:ok, %{status: status}} ->
{:error, "NCEI 5-min HTTP #{status}"}
{:error, reason} ->
{:error, reason}
end
{:error, reason} ->
{:error, reason}
end
end)
end
@doc """

8
lib/microwaveprop/weather/rtma_client.ex
View file

@ -35,6 +35,14 @@ defmodule Microwaveprop.Weather.RtmaClient do
"""
@spec fetch_observation(float(), float(), DateTime.t()) :: {:ok, map()} | {:error, term()}
def fetch_observation(lat, lon, timestamp) do
Microwaveprop.Instrument.span(
[:rtma, :fetch_observation],
%{},
fn -> do_fetch_observation(lat, lon, timestamp) end
)
end
defp do_fetch_observation(lat, lon, timestamp) do
# RTMA runs every hour with 15-min updates; round to nearest hour
valid_time = %{DateTime.truncate(timestamp, :second) | minute: 0, second: 0}
rlat = Float.round(lat * 40) / 40

20
lib/microwaveprop/weather/solar_client.ex
View file

@ -8,18 +8,20 @@ defmodule Microwaveprop.Weather.SolarClient do
@spec fetch_solar_indices() :: {:ok, [map()]} | {:error, term()}
def fetch_solar_indices do
req_options = Application.get_env(:microwaveprop, :solar_req_options, [])
Microwaveprop.Instrument.span([:solar, :fetch_indices], %{}, fn ->
req_options = Application.get_env(:microwaveprop, :solar_req_options, [])
case Req.get(@gfz_url, [receive_timeout: 120_000, retry: false] ++ req_options) do
{:ok, %{status: 200, body: body}} ->
{:ok, parse_gfz_file(body)}
case Req.get(@gfz_url, [receive_timeout: 120_000, retry: false] ++ req_options) do
{:ok, %{status: 200, body: body}} ->
{:ok, parse_gfz_file(body)}
{:ok, %{status: status}} ->
{:error, "HTTP #{status}"}
{:ok, %{status: status}} ->
{:error, "HTTP #{status}"}
{:error, reason} ->
{:error, reason}
end
{:error, reason} ->
{:error, reason}
end
end)
end
@spec filter_since([map()], Date.t()) :: [map()]

20
lib/microwaveprop/workers/mechanism_classify_worker.ex
View file

@ -42,17 +42,19 @@ defmodule Microwaveprop.Workers.MechanismClassifyWorker do
@impl Oban.Worker
def perform(%Oban.Job{args: %{"contact_id" => contact_id}}) do
case Repo.get(Contact, contact_id) do
nil ->
:ok
Microwaveprop.Instrument.span([:worker, :mechanism_classify], %{contact_id: contact_id}, fn ->
case Repo.get(Contact, contact_id) do
nil ->
:ok
%Contact{pos1: p1, pos2: p2} = contact when is_map(p1) and is_map(p2) ->
classify_and_persist(contact)
%Contact{pos1: p1, pos2: p2} = contact when is_map(p1) and is_map(p2) ->
classify_and_persist(contact)
contact ->
mark_status(contact, :unavailable, nil, nil)
:ok
end
contact ->
mark_status(contact, :unavailable, nil, nil)
:ok
end
end)
end
defp classify_and_persist(%Contact{} = contact) do

26
lib/microwaveprop/workers/mrms_fetch_worker.ex
View file

@ -17,22 +17,24 @@ defmodule Microwaveprop.Workers.MrmsFetchWorker do
@impl Oban.Worker
def perform(%Oban.Job{}) do
case MrmsClient.fetch_latest(MrmsCache.valid_time()) do
{:ok, valid_time, grid} ->
MrmsCache.broadcast_put(valid_time, grid)
Microwaveprop.Instrument.span([:worker, :mrms_fetch], %{}, fn ->
case MrmsClient.fetch_latest(MrmsCache.valid_time()) do
{:ok, valid_time, grid} ->
MrmsCache.broadcast_put(valid_time, grid)
Logger.info("MrmsFetch: cached #{map_size(grid)} cells at #{valid_time} (#{count_rainy(grid)} with rain)")
Logger.info("MrmsFetch: cached #{map_size(grid)} cells at #{valid_time} (#{count_rainy(grid)} with rain)")
:ok
:ok
{:up_to_date, valid_time} ->
Logger.debug("MrmsFetch: cache already has #{valid_time}")
:ok
{:up_to_date, valid_time} ->
Logger.debug("MrmsFetch: cache already has #{valid_time}")
:ok
{:error, reason} ->
Logger.warning("MrmsFetch: skipped — #{inspect(reason)}")
:ok
end
{:error, reason} ->
Logger.warning("MrmsFetch: skipped — #{inspect(reason)}")
:ok
end
end)
end
defp count_rainy(grid) do

14
lib/microwaveprop/workers/terrain_profile_worker.ex
View file

@ -18,12 +18,14 @@ defmodule Microwaveprop.Workers.TerrainProfileWorker do
@impl Oban.Worker
def perform(%Oban.Job{args: %{"contact_id" => contact_id}}) do
if Terrain.has_terrain_profile?(contact_id) do
Radio.set_enrichment_status!([contact_id], :terrain_status, :complete)
:ok
else
analyse_terrain(contact_id)
end
Microwaveprop.Instrument.span([:worker, :terrain_profile], %{contact_id: contact_id}, fn ->
if Terrain.has_terrain_profile?(contact_id) do
Radio.set_enrichment_status!([contact_id], :terrain_status, :complete)
:ok
else
analyse_terrain(contact_id)
end
end)
end
defp analyse_terrain(contact_id) do