Graham McIntire
4421727e61
Removes unreachable catch-all clauses, drops unused `require Logger` lines, adds pin operators to bitstring size patterns, reorders function heads for correct dispatch, and replaces deprecated LoggerBackends calls.
231 lines
7.4 KiB
Elixir
231 lines
7.4 KiB
Elixir
defmodule Towerops.ConfigChanges.Correlator do
|
|
@moduledoc """
|
|
Performance correlation engine for config change events.
|
|
|
|
Compares QoE metrics from Preseem in a window around a config change:
|
|
2 hours before and 4 hours after. Detects significant degradation and
|
|
creates insights when correlation is found.
|
|
"""
|
|
|
|
import Ecto.Query
|
|
|
|
alias Towerops.ConfigChanges.ConfigChangeEvent
|
|
alias Towerops.Preseem.AccessPoint
|
|
alias Towerops.Preseem.Insights
|
|
alias Towerops.Preseem.SubscriberMetric
|
|
alias Towerops.Repo
|
|
|
|
@pre_window_hours 2
|
|
@post_window_hours 4
|
|
|
|
# Thresholds for "significant" impact
|
|
@latency_increase_threshold 0.20
|
|
@loss_increase_threshold 0.50
|
|
@throughput_drop_threshold 0.15
|
|
|
|
@doc """
|
|
Correlate a config change event with QoE metrics.
|
|
|
|
Returns `{:ok, :no_impact}`, `{:ok, :no_data}`, or `{:ok, insight}` when
|
|
significant degradation is detected.
|
|
"""
|
|
def correlate(%ConfigChangeEvent{} = event) do
|
|
# Find Preseem access points matched to this device
|
|
ap_ids = get_matched_ap_ids(event.device_id)
|
|
|
|
if Enum.empty?(ap_ids) do
|
|
{:ok, :no_data}
|
|
else
|
|
pre_start = DateTime.add(event.changed_at, -@pre_window_hours * 3600, :second)
|
|
post_end = DateTime.add(event.changed_at, @post_window_hours * 3600, :second)
|
|
|
|
pre_metrics = avg_metrics(ap_ids, pre_start, event.changed_at)
|
|
post_metrics = avg_metrics(ap_ids, event.changed_at, post_end)
|
|
|
|
case {pre_metrics, post_metrics} do
|
|
{nil, _} -> {:ok, :no_data}
|
|
{_, nil} -> {:ok, :no_data}
|
|
{pre, post} -> evaluate_impact(event, pre, post)
|
|
end
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Correlate all recent uncorrelated change events for an organization.
|
|
"""
|
|
def correlate_recent(organization_id, opts \\ []) do
|
|
hours_back = Keyword.get(opts, :hours_back, 24)
|
|
since = DateTime.add(DateTime.utc_now(), -hours_back * 3600, :second)
|
|
|
|
events =
|
|
ConfigChangeEvent
|
|
|> where(organization_id: ^organization_id)
|
|
|> where([e], e.changed_at >= ^since)
|
|
|> Repo.all()
|
|
|
|
results = Enum.map(events, &correlate/1)
|
|
{:ok, results}
|
|
end
|
|
|
|
# Find Preseem AP IDs matched to a device
|
|
defp get_matched_ap_ids(device_id) do
|
|
AccessPoint
|
|
|> where(device_id: ^device_id)
|
|
|> select([ap], ap.id)
|
|
|> Repo.all()
|
|
end
|
|
|
|
# Average metrics for given APs in a time window
|
|
defp avg_metrics(ap_ids, from, to) do
|
|
result =
|
|
SubscriberMetric
|
|
|> where([m], m.preseem_access_point_id in ^ap_ids)
|
|
|> where([m], m.recorded_at >= ^from and m.recorded_at < ^to)
|
|
|> select([m], %{
|
|
avg_latency: avg(m.avg_latency),
|
|
avg_jitter: avg(m.avg_jitter),
|
|
avg_loss: avg(m.avg_loss),
|
|
avg_throughput: avg(m.avg_throughput)
|
|
})
|
|
|> Repo.one()
|
|
|
|
if is_nil(result) || is_nil(result.avg_latency) do
|
|
nil
|
|
else
|
|
result
|
|
end
|
|
end
|
|
|
|
defp evaluate_impact(event, pre, post) do
|
|
deltas = %{
|
|
latency_delta: safe_pct_change(to_float(pre.avg_latency), to_float(post.avg_latency)),
|
|
jitter_delta: safe_pct_change(to_float(pre.avg_jitter), to_float(post.avg_jitter)),
|
|
loss_delta: safe_pct_change(to_float(pre.avg_loss), to_float(post.avg_loss)),
|
|
throughput_delta: safe_pct_change(to_float(pre.avg_throughput), to_float(post.avg_throughput))
|
|
}
|
|
|
|
significant? =
|
|
deltas.latency_delta > @latency_increase_threshold ||
|
|
deltas.loss_delta > @loss_increase_threshold ||
|
|
deltas.throughput_delta < -@throughput_drop_threshold
|
|
|
|
if significant? do
|
|
create_suspect_insight(event, deltas, pre, post)
|
|
else
|
|
{:ok, :no_impact}
|
|
end
|
|
end
|
|
|
|
defp create_suspect_insight(event, deltas, pre, post) do
|
|
severity = determine_severity(deltas)
|
|
|
|
attrs = %{
|
|
organization_id: event.organization_id,
|
|
device_id: event.device_id,
|
|
type: "suspect_config_change",
|
|
urgency: severity,
|
|
channel: "proactive",
|
|
source: "system",
|
|
title: "Config change may have degraded performance",
|
|
description: build_description(deltas, event),
|
|
metadata: %{
|
|
"change_event_id" => event.id,
|
|
"metric_deltas" => %{
|
|
"latency_pct" => Float.round(deltas.latency_delta * 100, 1),
|
|
"jitter_pct" => Float.round(deltas.jitter_delta * 100, 1),
|
|
"loss_pct" => Float.round(deltas.loss_delta * 100, 1),
|
|
"throughput_pct" => Float.round(deltas.throughput_delta * 100, 1)
|
|
},
|
|
"pre_metrics" => stringify_metrics(pre),
|
|
"post_metrics" => stringify_metrics(post),
|
|
"sections_changed" => event.sections_changed
|
|
}
|
|
}
|
|
|
|
Insights.insert_insight_if_new(attrs)
|
|
end
|
|
|
|
defp determine_severity(deltas) do
|
|
determine_severity_from_deltas(
|
|
deltas.latency_delta,
|
|
deltas.loss_delta,
|
|
deltas.throughput_delta
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Builds a human-readable description of a config change's impact.
|
|
|
|
Includes sections from `event.sections_changed` when present, and
|
|
includes metric deltas that exceed built-in reporting thresholds
|
|
(latency/loss > 10% increase, throughput > 5% drop).
|
|
"""
|
|
def build_description(deltas, event) do
|
|
parts =
|
|
[]
|
|
|> maybe_add(
|
|
"latency +#{round(deltas.latency_delta * 100)}%",
|
|
deltas.latency_delta > 0.10
|
|
)
|
|
|> maybe_add(
|
|
"loss +#{round(deltas.loss_delta * 100)}%",
|
|
deltas.loss_delta > 0.10
|
|
)
|
|
|> maybe_add(
|
|
"throughput #{round(deltas.throughput_delta * 100)}%",
|
|
deltas.throughput_delta < -0.05
|
|
)
|
|
|
|
"After config change#{format_sections(event)}: #{Enum.join(parts, ", ")}."
|
|
end
|
|
|
|
defp format_sections(%{sections_changed: sections}) when is_list(sections) and sections != [],
|
|
do: " (sections: #{Enum.join(sections, ", ")})"
|
|
|
|
defp format_sections(_event), do: ""
|
|
|
|
@doc """
|
|
Calculates percentage change between two values.
|
|
|
|
Returns `0.0` if the old value is zero or negative (avoids division by zero
|
|
and treats "no baseline" as "no change").
|
|
"""
|
|
@spec safe_pct_change(number(), number()) :: float()
|
|
def safe_pct_change(old, new_val) when old > 0.0, do: (new_val - old) / old
|
|
def safe_pct_change(_old, _new_val), do: 0.0
|
|
|
|
@doc """
|
|
Determines severity based on metric deltas expressed as ratios
|
|
(0.5 = 50% increase). Thresholds:
|
|
|
|
* `"critical"`: latency > 50% · loss > 100% · throughput < -30%
|
|
* `"warning"`: latency > 20% · loss > 50% · throughput < -15%
|
|
* `"info"`: otherwise
|
|
"""
|
|
@spec determine_severity_from_deltas(number(), number(), number()) :: String.t()
|
|
def determine_severity_from_deltas(latency_delta, loss_delta, throughput_delta)
|
|
when latency_delta > 0.50 or loss_delta > 1.0 or throughput_delta < -0.30, do: "critical"
|
|
|
|
def determine_severity_from_deltas(latency_delta, loss_delta, throughput_delta)
|
|
when latency_delta > 0.20 or loss_delta > 0.50 or throughput_delta < -0.15, do: "warning"
|
|
|
|
def determine_severity_from_deltas(_latency, _loss, _throughput), do: "info"
|
|
|
|
@doc false
|
|
def maybe_add(parts, text, true), do: parts ++ [text]
|
|
def maybe_add(parts, _text, false), do: parts
|
|
|
|
@doc false
|
|
def to_float(nil), do: 0.0
|
|
def to_float(v) when is_float(v), do: v
|
|
def to_float(v) when is_integer(v), do: v / 1
|
|
|
|
defp stringify_metrics(metrics) do
|
|
%{
|
|
"avg_latency" => metrics.avg_latency && Float.round(metrics.avg_latency, 2),
|
|
"avg_jitter" => metrics.avg_jitter && Float.round(metrics.avg_jitter, 2),
|
|
"avg_loss" => metrics.avg_loss && Float.round(metrics.avg_loss, 4),
|
|
"avg_throughput" => metrics.avg_throughput && Float.round(metrics.avg_throughput, 2)
|
|
}
|
|
end
|
|
end
|