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
This commit is contained in:
parent
249e3fa675
commit
24796f98d4
7 changed files with 219 additions and 70 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue