diff --git a/lib/microwaveprop/geocoder.ex b/lib/microwaveprop/geocoder.ex index 5fb0338e..b84331b2 100644 --- a/lib/microwaveprop/geocoder.ex +++ b/lib/microwaveprop/geocoder.ex @@ -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 diff --git a/lib/microwaveprop/ionosphere/giro_client.ex b/lib/microwaveprop/ionosphere/giro_client.ex index 6fcc2803..c08d8f8e 100644 --- a/lib/microwaveprop/ionosphere/giro_client.ex +++ b/lib/microwaveprop/ionosphere/giro_client.ex @@ -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 """ diff --git a/lib/microwaveprop/prom_ex/instrument_plugin.ex b/lib/microwaveprop/prom_ex/instrument_plugin.ex index a5ca08a5..45cd56e4 100644 --- a/lib/microwaveprop/prom_ex/instrument_plugin.ex +++ b/lib/microwaveprop/prom_ex/instrument_plugin.ex @@ -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, diff --git a/lib/microwaveprop/propagation.ex b/lib/microwaveprop/propagation.ex index 3f33c74f..1f7f92b3 100644 --- a/lib/microwaveprop/propagation.ex +++ b/lib/microwaveprop/propagation.ex @@ -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) diff --git a/lib/microwaveprop/qrz/client.ex b/lib/microwaveprop/qrz/client.ex index d5e8c01e..3722113c 100644 --- a/lib/microwaveprop/qrz/client.ex +++ b/lib/microwaveprop/qrz/client.ex @@ -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 diff --git a/lib/microwaveprop/space_weather/swpc_client.ex b/lib/microwaveprop/space_weather/swpc_client.ex index 7a58a069..c9d8d396 100644 --- a/lib/microwaveprop/space_weather/swpc_client.ex +++ b/lib/microwaveprop/space_weather/swpc_client.ex @@ -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 diff --git a/lib/microwaveprop/terrain/srtm.ex b/lib/microwaveprop/terrain/srtm.ex index 54075dd5..432c7039 100644 --- a/lib/microwaveprop/terrain/srtm.ex +++ b/lib/microwaveprop/terrain/srtm.ex @@ -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" diff --git a/lib/microwaveprop/weather/grib2/wgrib2.ex b/lib/microwaveprop/weather/grib2/wgrib2.ex index 472acc04..8897e29c 100644 --- a/lib/microwaveprop/weather/grib2/wgrib2.ex +++ b/lib/microwaveprop/weather/grib2/wgrib2.ex @@ -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." diff --git a/lib/microwaveprop/weather/hrrr_client.ex b/lib/microwaveprop/weather/hrrr_client.ex index 9e8ff3c7..9a68a61e 100644 --- a/lib/microwaveprop/weather/hrrr_client.ex +++ b/lib/microwaveprop/weather/hrrr_client.ex @@ -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 diff --git a/lib/microwaveprop/weather/iem_client.ex b/lib/microwaveprop/weather/iem_client.ex index 9c7cac22..b7c40d87 100644 --- a/lib/microwaveprop/weather/iem_client.ex +++ b/lib/microwaveprop/weather/iem_client.ex @@ -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 diff --git a/lib/microwaveprop/weather/mrms_client.ex b/lib/microwaveprop/weather/mrms_client.ex index 0780bdca..553bcfde 100644 --- a/lib/microwaveprop/weather/mrms_client.ex +++ b/lib/microwaveprop/weather/mrms_client.ex @@ -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 diff --git a/lib/microwaveprop/weather/ncei_metar_client.ex b/lib/microwaveprop/weather/ncei_metar_client.ex index 7eb21039..6ca63d3e 100644 --- a/lib/microwaveprop/weather/ncei_metar_client.ex +++ b/lib/microwaveprop/weather/ncei_metar_client.ex @@ -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 """ diff --git a/lib/microwaveprop/weather/rtma_client.ex b/lib/microwaveprop/weather/rtma_client.ex index 6ebb3fd3..495c2af2 100644 --- a/lib/microwaveprop/weather/rtma_client.ex +++ b/lib/microwaveprop/weather/rtma_client.ex @@ -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 diff --git a/lib/microwaveprop/weather/solar_client.ex b/lib/microwaveprop/weather/solar_client.ex index d0500f9f..1fc42889 100644 --- a/lib/microwaveprop/weather/solar_client.ex +++ b/lib/microwaveprop/weather/solar_client.ex @@ -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()] diff --git a/lib/microwaveprop/workers/mechanism_classify_worker.ex b/lib/microwaveprop/workers/mechanism_classify_worker.ex index b30163cb..2c513a81 100644 --- a/lib/microwaveprop/workers/mechanism_classify_worker.ex +++ b/lib/microwaveprop/workers/mechanism_classify_worker.ex @@ -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 diff --git a/lib/microwaveprop/workers/mrms_fetch_worker.ex b/lib/microwaveprop/workers/mrms_fetch_worker.ex index 5212f1a0..ce2166ff 100644 --- a/lib/microwaveprop/workers/mrms_fetch_worker.ex +++ b/lib/microwaveprop/workers/mrms_fetch_worker.ex @@ -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 diff --git a/lib/microwaveprop/workers/terrain_profile_worker.ex b/lib/microwaveprop/workers/terrain_profile_worker.ex index 77406d11..9dbccd9d 100644 --- a/lib/microwaveprop/workers/terrain_profile_worker.ex +++ b/lib/microwaveprop/workers/terrain_profile_worker.ex @@ -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