prop/lib/microwaveprop/prom_ex/instrument_plugin.ex

defmodule Microwaveprop.PromEx.InstrumentPlugin do
  @moduledoc """
  PromEx plugin that registers every `Microwaveprop.Instrument` span as
  a Prometheus histogram. Keep the catalog in sync with the call sites
  in `HrrrClient`, `NexradClient`, `IemClient`, `GefsClient`,
  `NarrClient`, `UwyoSoundingClient`, `ElevationClient`, `Weather` batch
  upserts, `Propagation.replace_scores`, `PropagationGridWorker`,
  `TerrainAnalysis`, and `CommonVolumeRadarWorker`.
  """
  use PromEx.Plugin

  @impl true
  def event_metrics(_opts) do
    [
      http_client_events(),
      db_batch_events(),
      worker_phase_events(),
      oban_queue_depth_event()
    ]
  end

  # PromEx interprets `Telemetry.Metrics` specs the same way the
  # in-process telemetry_metrics module does, so we re-use the same
  # event_name / measurement / tags shape from MicrowavepropWeb.Telemetry.
  defp http_client_events do
    Event.build(
      :microwaveprop_http_client_events,
      [
        distribution("microwaveprop.hrrr.fetch_grid.duration.milliseconds",
          event_name: [:microwaveprop, :hrrr, :fetch_grid, :stop],
          measurement: :duration,
          unit: {:native, :millisecond},
          reporter_options: [buckets: default_buckets()]
        ),
        distribution("microwaveprop.hrrr.fetch_profile.duration.milliseconds",
          event_name: [:microwaveprop, :hrrr, :fetch_profile, :stop],
          measurement: :duration,
          unit: {:native, :millisecond},
          reporter_options: [buckets: default_buckets()]
        ),
        distribution("microwaveprop.hrrr.fetch_idx.duration.milliseconds",
          event_name: [:microwaveprop, :hrrr, :fetch_idx, :stop],
          measurement: :duration,
          unit: {:native, :millisecond},
          reporter_options: [buckets: default_buckets()]
        ),
        distribution("microwaveprop.nexrad.fetch_frame.duration.milliseconds",
          event_name: [:microwaveprop, :nexrad, :fetch_frame, :stop],
          measurement: :duration,
          unit: {:native, :millisecond},
          reporter_options: [buckets: default_buckets()]
        ),
        distribution("microwaveprop.nexrad.decode_png.duration.milliseconds",
          event_name: [:microwaveprop, :nexrad, :decode_png, :stop],
          measurement: :duration,
          unit: {:native, :millisecond},
          reporter_options: [buckets: default_buckets()]
        ),
        distribution("microwaveprop.iem.fetch_asos.duration.milliseconds",
          event_name: [:microwaveprop, :iem, :fetch_asos, :stop],
          measurement: :duration,
          unit: {:native, :millisecond},
          reporter_options: [buckets: default_buckets()]
        ),
        distribution("microwaveprop.iem.fetch_raob.duration.milliseconds",
          event_name: [:microwaveprop, :iem, :fetch_raob, :stop],
          measurement: :duration,
          unit: {:native, :millisecond},
          reporter_options: [buckets: default_buckets()]
        ),
        distribution("microwaveprop.gefs.fetch_grid_profiles.duration.milliseconds",
          event_name: [:microwaveprop, :gefs, :fetch_grid_profiles, :stop],
          measurement: :duration,
          unit: {:native, :millisecond},
          reporter_options: [buckets: default_buckets()]
        ),
        distribution("microwaveprop.narr.fetch_profile.duration.milliseconds",
          event_name: [:microwaveprop, :narr, :fetch_profile, :stop],
          measurement: :duration,
          unit: {:native, :millisecond},
          reporter_options: [buckets: default_buckets()]
        ),
        distribution("microwaveprop.uwyo.fetch_sounding.duration.milliseconds",
          event_name: [:microwaveprop, :uwyo, :fetch_sounding, :stop],
          measurement: :duration,
          unit: {:native, :millisecond},
          reporter_options: [buckets: default_buckets()]
        ),
        distribution("microwaveprop.elevation.fetch_profile.duration.milliseconds",
          event_name: [:microwaveprop, :elevation, :fetch_profile, :stop],
          measurement: :duration,
          unit: {:native, :millisecond},
          reporter_options: [buckets: default_buckets()]
        )
      ]
    )
  end

  defp db_batch_events do
    Event.build(
      :microwaveprop_db_batch_events,
      [
        distribution("microwaveprop.db.upsert_hrrr_profiles.duration.milliseconds",
          event_name: [:microwaveprop, :db, :upsert_hrrr_profiles, :stop],
          measurement: :duration,
          unit: {:native, :millisecond},
          reporter_options: [buckets: db_buckets()]
        ),
        distribution("microwaveprop.db.upsert_gefs_profiles.duration.milliseconds",
          event_name: [:microwaveprop, :db, :upsert_gefs_profiles, :stop],
          measurement: :duration,
          unit: {:native, :millisecond},
          reporter_options: [buckets: db_buckets()]
        ),
        distribution("microwaveprop.db.replace_scores.duration.milliseconds",
          event_name: [:microwaveprop, :db, :replace_scores, :stop],
          measurement: :duration,
          unit: {:native, :millisecond},
          reporter_options: [buckets: db_buckets()]
        )
      ]
    )
  end

  defp worker_phase_events do
    Event.build(
      :microwaveprop_worker_phase_events,
      [
        distribution("microwaveprop.propagation_grid.score_band.duration.milliseconds",
          event_name: [:microwaveprop, :propagation_grid, :score_band, :stop],
          measurement: :duration,
          unit: {:native, :millisecond},
          reporter_options: [buckets: long_buckets()]
        ),
        distribution("microwaveprop.terrain.analyse.duration.milliseconds",
          event_name: [:microwaveprop, :terrain, :analyse, :stop],
          measurement: :duration,
          unit: {:native, :millisecond},
          reporter_options: [buckets: default_buckets()]
        ),
        distribution("microwaveprop.radar.aggregate_stats.duration.milliseconds",
          event_name: [:microwaveprop, :radar, :aggregate_stats, :stop],
          measurement: :duration,
          unit: {:native, :millisecond},
          reporter_options: [buckets: default_buckets()]
        )
      ]
    )
  end

  defp oban_queue_depth_event do
    Event.build(
      :microwaveprop_oban_queue_depth,
      [
        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

  # Buckets in milliseconds. HTTP + compute phases: up to ~30s.
  defp default_buckets, do: [10, 50, 100, 250, 500, 1_000, 2_500, 5_000, 10_000, 30_000]

  # DB batch upserts: can take much longer on cold cache or full
  # forecast-hour writes — up to ~2 min.
  defp db_buckets, do: [50, 100, 250, 500, 1_000, 2_500, 5_000, 15_000, 60_000, 120_000]

  # Score computation for a full grid-point batch — up to ~5 min.
  defp long_buckets, do: [500, 1_000, 5_000, 15_000, 30_000, 60_000, 120_000, 300_000]
end