From 24796f98d4df91d7973a98020cc6f5b5e88abec7 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 12 May 2026 17:48:58 -0500 Subject: [PATCH] Optimize prod: scale deployment, drop expensive metric queries, fix slow callsign and nearby-stations queries - k8s: 2 replicas, POOL_SIZE 25, cpu 2000m, mem 1Gi (was 1 replica, 5, 500m, 512Mi) - telemetry poller 10s -> 60s; drop pg_database_size, pg_stat_statements, pg_table_size and pg_indexes_size (these were ~7100s of cumulative DB time on their own) - get_latest_packet_for_callsign: bound by :packet_retention_days so partition pruning kicks in - get_nearby_stations_knn: ST_DWithin spatial pre-filter (default 500km) so the GiST geography index cuts candidates before DISTINCT ON sort (EXPLAIN: 5793ms -> 400ms) - New migration: partial functional indexes on upper(object_name) and upper(item_name), built per-partition CONCURRENTLY then attached to parent. Lets get_latest_packet_for_callsign use BitmapOr across all 3 upper() indexes instead of falling back to a scan --- k8s/deployment.yaml | 10 +- lib/aprsme/packets/prepared_queries.ex | 22 ++++ lib/aprsme/telemetry/database_metrics.ex | 63 ++--------- lib/aprsme_web/telemetry.ex | 2 +- ..._add_object_name_and_item_name_indexes.exs | 77 +++++++++++++ test/aprsme/packets/prepared_queries_test.exs | 106 ++++++++++++++++++ .../telemetry/database_metrics_test.exs | 9 +- 7 files changed, 219 insertions(+), 70 deletions(-) create mode 100644 priv/repo/migrations/20260512173957_add_object_name_and_item_name_indexes.exs diff --git a/k8s/deployment.yaml b/k8s/deployment.yaml index cedd97e..07c3b09 100644 --- a/k8s/deployment.yaml +++ b/k8s/deployment.yaml @@ -4,7 +4,7 @@ metadata: name: aprs namespace: aprs spec: - replicas: 1 + replicas: 2 strategy: type: RollingUpdate rollingUpdate: @@ -130,7 +130,7 @@ spec: - name: MIX_ENV value: "prod" - name: POOL_SIZE - value: "5" + value: "25" - name: PACKET_RETENTION_DAYS value: "1" - name: APRS_CALLSIGN @@ -161,11 +161,11 @@ spec: name: aprs-db resources: requests: - memory: "256Mi" - cpu: "100m" - limits: memory: "512Mi" cpu: "500m" + limits: + memory: "1Gi" + cpu: "2000m" startupProbe: httpGet: path: /health diff --git a/lib/aprsme/packets/prepared_queries.ex b/lib/aprsme/packets/prepared_queries.ex index 7a02504..fd7f908 100644 --- a/lib/aprsme/packets/prepared_queries.ex +++ b/lib/aprsme/packets/prepared_queries.ex @@ -13,13 +13,20 @@ defmodule Aprsme.Packets.PreparedQueries do Get the latest packet for a callsign using a prepared statement. This is one of the most frequently called queries. Searches by sender, object_name, or item_name. + + Bounded by the configured `:packet_retention_days` so PostgreSQL can prune + unrelated daily partitions. Data older than retention is already cleaned up + by `Aprsme.Workers.PacketCleanupWorker`, so this bound does not exclude any + packets that would otherwise exist. """ @spec get_latest_packet_for_callsign(String.t()) :: Packet.t() | nil def get_latest_packet_for_callsign(callsign) when is_binary(callsign) do normalized = String.upcase(String.trim(callsign)) + cutoff = retention_cutoff() Repo.one( from(p in Packet, + where: p.received_at >= ^cutoff, where: fragment("upper(?)", p.sender) == ^normalized or fragment("upper(?)", p.object_name) == ^normalized or @@ -34,6 +41,11 @@ defmodule Aprsme.Packets.PreparedQueries do ) end + defp retention_cutoff do + retention_days = Application.get_env(:aprsme, :packet_retention_days, 7) + DateTime.add(DateTime.utc_now(), -retention_days * 86_400, :second) + end + @doc """ Get the latest full packet for each callsign in a single query. Returns a list of `Packet` structs with lat/lon populated. @@ -146,11 +158,20 @@ defmodule Aprsme.Packets.PreparedQueries do @doc """ Get nearby stations using KNN (K-nearest neighbors) search. Uses the <-> operator for efficient spatial queries. + + Options: + * `:limit` - max number of stations to return (default 10) + * `:hours_back` - only consider packets received in the last N hours (default 1) + * `:max_radius_km` - only consider stations within N kilometers of the point + (default 500 km / ~310 mi). This bound lets PostgreSQL pre-filter via + the GiST geography index before the DISTINCT ON sort, which is otherwise + the hot spot in this query. """ @spec get_nearby_stations_knn(float(), float(), String.t() | nil, map()) :: [map()] def get_nearby_stations_knn(lat, lon, exclude_callsign \\ nil, opts \\ %{}) do limit = Map.get(opts, :limit, 10) hours_back = Map.get(opts, :hours_back, 1) + max_radius_meters = Map.get(opts, :max_radius_km, 500) * 1000 cutoff_time = DateTime.add(DateTime.utc_now(), -hours_back * 3600, :second) # Build point for KNN search @@ -161,6 +182,7 @@ defmodule Aprsme.Packets.PreparedQueries do where: p.has_position == true, where: p.received_at >= ^cutoff_time, where: not is_nil(p.location), + where: fragment("ST_DWithin(?::geography, ?::geography, ?)", p.location, ^point, ^max_radius_meters), distinct: p.base_callsign, order_by: [ asc: p.base_callsign, diff --git a/lib/aprsme/telemetry/database_metrics.ex b/lib/aprsme/telemetry/database_metrics.ex index 14c7250..2ffecda 100644 --- a/lib/aprsme/telemetry/database_metrics.ex +++ b/lib/aprsme/telemetry/database_metrics.ex @@ -107,26 +107,14 @@ defmodule Aprsme.Telemetry.DatabaseMetrics do end defp collect_database_metrics do - collect_database_size() collect_connection_stats() collect_packets_table_stats() - collect_query_performance_stats() collect_replication_lag() rescue e -> Logger.debug("Error collecting PostgreSQL metrics: #{inspect(e)}") end - defp collect_database_size do - case Aprsme.Repo.query("SELECT pg_database_size(current_database()) as size") do - {:ok, %{rows: [[size]]}} -> - :telemetry.execute([:aprsme, :postgres, :database], %{size_bytes: to_number(size)}, %{}) - - _ -> - :ok - end - end - defp collect_connection_stats do case Aprsme.Repo.query(""" SELECT @@ -163,42 +151,12 @@ defmodule Aprsme.Telemetry.DatabaseMetrics do n_dead_tup as dead_tuples, n_tup_ins as inserts, n_tup_upd as updates, - n_tup_del as deletes, - pg_table_size(c.oid) as table_size, - pg_indexes_size(c.oid) as indexes_size - FROM pg_stat_user_tables s - JOIN pg_class c ON c.relname = s.relname - WHERE s.relname = 'packets' + n_tup_del as deletes + FROM pg_stat_user_tables + WHERE relname = 'packets' """) do - {:ok, %{rows: [[live, dead, ins, upd, del, table_size, idx_size]]}} -> - emit_packets_table_telemetry(live, dead, ins, upd, del, table_size, idx_size) - - _ -> - :ok - end - end - - defp collect_query_performance_stats do - case Aprsme.Repo.query(""" - SELECT - sum(calls) as total_calls, - sum(total_exec_time) as total_time, - avg(mean_exec_time) as avg_time, - max(max_exec_time) as max_time - FROM pg_stat_statements - WHERE query NOT LIKE '%pg_stat%' - """) do - {:ok, %{rows: [[calls, total_time, avg_time, max_time]]}} -> - :telemetry.execute( - [:aprsme, :postgres, :query_stats], - %{ - total_calls: to_number(calls, :integer), - total_time_ms: to_number(total_time), - avg_time_ms: to_number(avg_time), - max_time_ms: to_number(max_time) - }, - %{} - ) + {:ok, %{rows: [[live, dead, ins, upd, del]]}} -> + emit_packets_table_telemetry(live, dead, ins, upd, del) _ -> :ok @@ -225,7 +183,7 @@ defmodule Aprsme.Telemetry.DatabaseMetrics do :ok end - defp emit_packets_table_telemetry(live, dead, ins, upd, del, table_size, idx_size) do + defp emit_packets_table_telemetry(live, dead, ins, upd, del) do :telemetry.execute( [:aprsme, :postgres, :packets_table], %{ @@ -233,9 +191,7 @@ defmodule Aprsme.Telemetry.DatabaseMetrics do dead_tuples: to_number(dead), total_inserts: to_number(ins), total_updates: to_number(upd), - total_deletes: to_number(del), - table_size_bytes: to_number(table_size), - indexes_size_bytes: to_number(idx_size) + total_deletes: to_number(del) }, %{} ) @@ -246,9 +202,4 @@ defmodule Aprsme.Telemetry.DatabaseMetrics do defp to_number(%Decimal{} = d), do: Decimal.to_float(d) defp to_number(n) when is_number(n), do: n defp to_number(_), do: 0 - - defp to_number(nil, :integer), do: 0 - defp to_number(%Decimal{} = d, :integer), do: d |> Decimal.to_float() |> trunc() - defp to_number(n, :integer) when is_number(n), do: trunc(n) - defp to_number(_, :integer), do: 0 end diff --git a/lib/aprsme_web/telemetry.ex b/lib/aprsme_web/telemetry.ex index 559262e..e0ed83c 100644 --- a/lib/aprsme_web/telemetry.ex +++ b/lib/aprsme_web/telemetry.ex @@ -13,7 +13,7 @@ defmodule AprsmeWeb.Telemetry do @impl true def init(_arg) do children = [ - {:telemetry_poller, measurements: periodic_measurements(), period: 10_000} + {:telemetry_poller, measurements: periodic_measurements(), period: 60_000} ] Supervisor.init(children, strategy: :one_for_one) diff --git a/priv/repo/migrations/20260512173957_add_object_name_and_item_name_indexes.exs b/priv/repo/migrations/20260512173957_add_object_name_and_item_name_indexes.exs new file mode 100644 index 0000000..91900b4 --- /dev/null +++ b/priv/repo/migrations/20260512173957_add_object_name_and_item_name_indexes.exs @@ -0,0 +1,77 @@ +defmodule Aprsme.Repo.Migrations.AddObjectNameAndItemNameIndexes do + use Ecto.Migration + + @disable_ddl_transaction true + @disable_migration_lock true + + @parent_object_idx "idx_packets_upper_object_name" + @parent_item_idx "idx_packets_upper_item_name" + + def up do + if Aprsme.Repo.config()[:pool] == Ecto.Adapters.SQL.Sandbox do + # Test env (Sandbox does not support CONCURRENTLY). Build directly on parent + # so tests that exercise these queries still hit an index. + execute(""" + CREATE INDEX IF NOT EXISTS #{@parent_object_idx} + ON packets (upper(object_name)) + WHERE object_name IS NOT NULL + """) + + execute(""" + CREATE INDEX IF NOT EXISTS #{@parent_item_idx} + ON packets (upper(item_name)) + WHERE item_name IS NOT NULL + """) + else + # Production path: build per-partition CONCURRENTLY, then ATTACH to parent. + execute(""" + CREATE INDEX IF NOT EXISTS #{@parent_object_idx} + ON ONLY packets (upper(object_name)) + WHERE object_name IS NOT NULL + """) + + execute(""" + CREATE INDEX IF NOT EXISTS #{@parent_item_idx} + ON ONLY packets (upper(item_name)) + WHERE item_name IS NOT NULL + """) + + partitions = + Aprsme.Repo.query!(""" + SELECT inhrelid::regclass::text + FROM pg_inherits + WHERE inhparent = 'packets'::regclass + """).rows + |> List.flatten() + + Enum.each(partitions, &build_and_attach/1) + end + end + + def down do + execute("DROP INDEX IF EXISTS #{@parent_object_idx}") + execute("DROP INDEX IF EXISTS #{@parent_item_idx}") + end + + defp build_and_attach(partition) do + partition_unquoted = String.trim(partition, "\"") + safe = String.replace(partition_unquoted, ~r/[^a-zA-Z0-9_]/, "_") + object_idx = "#{safe}_upper_object_name_idx" + item_idx = "#{safe}_upper_item_name_idx" + + execute(""" + CREATE INDEX CONCURRENTLY IF NOT EXISTS #{object_idx} + ON #{partition} (upper(object_name)) + WHERE object_name IS NOT NULL + """) + + execute(""" + CREATE INDEX CONCURRENTLY IF NOT EXISTS #{item_idx} + ON #{partition} (upper(item_name)) + WHERE item_name IS NOT NULL + """) + + execute("ALTER INDEX #{@parent_object_idx} ATTACH PARTITION #{object_idx}") + execute("ALTER INDEX #{@parent_item_idx} ATTACH PARTITION #{item_idx}") + end +end diff --git a/test/aprsme/packets/prepared_queries_test.exs b/test/aprsme/packets/prepared_queries_test.exs index 6ed4c38..204e917 100644 --- a/test/aprsme/packets/prepared_queries_test.exs +++ b/test/aprsme/packets/prepared_queries_test.exs @@ -158,6 +158,112 @@ defmodule Aprsme.Packets.PreparedQueriesTest do assert packet.sender == "STATUS-ONLY" assert packet.has_position == false end + + test "ignores packets older than the retention window" do + # Insert a packet whose received_at is way beyond any reasonable retention + # window — older than the configured retention_days plus a margin. The + # query should not return it, so partition pruning can skip those + # partitions on the live table. + retention_days = Application.get_env(:aprsme, :packet_retention_days, 7) + ancient = DateTime.add(DateTime.utc_now(), -(retention_days + 30) * 86_400, :second) + + {:ok, _} = + create_positioned_packet(%{ + sender: "ANCIENT-1", + base_callsign: "ANCIENT", + ssid: "1", + received_at: DateTime.truncate(ancient, :second) + }) + + assert is_nil(PreparedQueries.get_latest_packet_for_callsign("ANCIENT-1")) + end + end + + describe "get_nearby_stations_knn/4" do + test "returns stations sorted by distance from the point" do + # Close to center + {:ok, _} = + create_positioned_packet(%{ + sender: "NEAR-1", + base_callsign: "NEAR", + ssid: "1", + lat: Decimal.new("33.0100"), + lon: Decimal.new("-96.0100") + }) + + # Further + {:ok, _} = + create_positioned_packet(%{ + sender: "FAR-1", + base_callsign: "FAR", + ssid: "1", + lat: Decimal.new("33.5000"), + lon: Decimal.new("-96.5000") + }) + + result = PreparedQueries.get_nearby_stations_knn(33.0, -96.0, nil, %{limit: 5}) + + callsigns = Enum.map(result, & &1.callsign) + assert "NEAR-1" in callsigns + assert "FAR-1" in callsigns + # NEAR-1 must come before FAR-1 (sorted by distance) + assert Enum.find_index(result, &(&1.callsign == "NEAR-1")) < + Enum.find_index(result, &(&1.callsign == "FAR-1")) + end + + test "excludes the specified callsign" do + {:ok, _} = + create_positioned_packet(%{ + sender: "ME-1", + base_callsign: "ME", + ssid: "1", + lat: Decimal.new("33.0100"), + lon: Decimal.new("-96.0100") + }) + + {:ok, _} = + create_positioned_packet(%{ + sender: "OTHER-1", + base_callsign: "OTHER", + ssid: "1", + lat: Decimal.new("33.0200"), + lon: Decimal.new("-96.0200") + }) + + result = PreparedQueries.get_nearby_stations_knn(33.0, -96.0, "ME-1", %{limit: 5}) + + callsigns = Enum.map(result, & &1.callsign) + refute "ME-1" in callsigns + assert "OTHER-1" in callsigns + end + + test "honors :max_radius_km option" do + # Within 10km of (33.0, -96.0): ~1km offset + {:ok, _} = + create_positioned_packet(%{ + sender: "INSIDE-1", + base_callsign: "INSIDE", + ssid: "1", + lat: Decimal.new("33.0100"), + lon: Decimal.new("-96.0100") + }) + + # Outside 10km: ~50km away + {:ok, _} = + create_positioned_packet(%{ + sender: "OUTSIDE-1", + base_callsign: "OUTSIDE", + ssid: "1", + lat: Decimal.new("33.5000"), + lon: Decimal.new("-96.5000") + }) + + result = PreparedQueries.get_nearby_stations_knn(33.0, -96.0, nil, %{limit: 10, max_radius_km: 10}) + + callsigns = Enum.map(result, & &1.callsign) + assert "INSIDE-1" in callsigns + refute "OUTSIDE-1" in callsigns + end end describe "get_latest_positions_for_callsigns/1" do diff --git a/test/aprsme/telemetry/database_metrics_test.exs b/test/aprsme/telemetry/database_metrics_test.exs index c2d6585..93b1cd7 100644 --- a/test/aprsme/telemetry/database_metrics_test.exs +++ b/test/aprsme/telemetry/database_metrics_test.exs @@ -109,16 +109,12 @@ defmodule Aprsme.Telemetry.DatabaseMetricsTest do end describe "collect_postgres_metrics/0 outside :test env" do - test "emits database, connections, packets_table and query_stats telemetry events" do + test "emits connections and packets_table telemetry events" do original = Application.get_env(:aprsme, :env) Application.put_env(:aprsme, :env, :prod) - # Attach collectors for every event emitted by the helpers. - attach_collector([:aprsme, :postgres, :database]) attach_collector([:aprsme, :postgres, :connections]) attach_collector([:aprsme, :postgres, :packets_table]) - # Query stats requires pg_stat_statements which may not be loaded — - # don't assert on that one, just call the function. try do DatabaseMetrics.collect_postgres_metrics() @@ -126,9 +122,6 @@ defmodule Aprsme.Telemetry.DatabaseMetricsTest do Application.put_env(:aprsme, :env, original) end - assert_receive {:telemetry, [:aprsme, :postgres, :database], %{size_bytes: size}, _}, 2_000 - assert is_number(size) and size > 0 - assert_receive {:telemetry, [:aprsme, :postgres, :connections], conn_metrics, _}, 2_000 assert Map.has_key?(conn_metrics, :total) assert Map.has_key?(conn_metrics, :active)