208 lines
11 KiB
Python
208 lines
11 KiB
Python
"""The metrics of the router in PostgreSQL, for the metrics service and the web service.
|
|
|
|
The metrics service writes one detail row per minute to the four detail tables, of TABLES:
|
|
metrics_system (CPU, memory, disk and load averages), metrics_network (bytes per second
|
|
received and sent per interface), metrics_vpn (clients per kind) and metrics_site_to_site (per
|
|
site-to-site connection: up, 1 or 0, bytes and packets per second received and sent, and the
|
|
seconds its IKE SA has been up). Each detail table has four
|
|
summary tables, of LEVELS: <table>_hourly from the details, <table>_daily from the hourly,
|
|
<table>_monthly from the daily and <table>_yearly from the monthly rows. A summary row has the
|
|
start of its period, in UTC, and the average and the maximum of each value over the period: of
|
|
the details for the hourly rows, else the average of the averages and the maximum of the maxima
|
|
below. A period is summarized once it is complete; summarize() catches up every complete period
|
|
after the last summarized one, so a stopped service loses no summary of what was written.
|
|
|
|
collect_garbage() deletes the rows older than the retention of their level, metrics.keep_*;
|
|
the service runs it on start and once a day. history() answers the views of the Metrics tab.
|
|
|
|
The tables are in the database of the refresh tokens (vpnrouter_auth.connection_string) and are
|
|
created when missing. Every failure of the database raises DatabaseError.
|
|
"""
|
|
|
|
import contextlib
|
|
|
|
import psycopg
|
|
|
|
import vpnrouter_auth
|
|
|
|
#: The detail tables: their key columns besides time, and their value columns.
|
|
TABLES = {
|
|
'metrics_system': ((), ('cpu', 'memory_used', 'memory_total', 'disk_free', 'disk_total', 'load1', 'load5', 'load15')),
|
|
'metrics_network': (('interface',), ('rx', 'tx')),
|
|
'metrics_vpn': (('kind',), ('clients',)),
|
|
'metrics_site_to_site': (('connection',), ('up', 'rx', 'tx', 'packets_rx', 'packets_tx', 'uptime')),
|
|
}
|
|
|
|
#: The summary levels, from the finest: the suffix of the tables, the period of date_trunc and the retention setting.
|
|
LEVELS = (
|
|
('hourly', 'hour', 'keep_hourly'),
|
|
('daily', 'day', 'keep_daily'),
|
|
('monthly', 'month', 'keep_monthly'),
|
|
('yearly', 'year', 'keep_yearly'),
|
|
)
|
|
|
|
#: The kinds of VPN clients of metrics_vpn.
|
|
VPN_KINDS = ('p2s', 'openvpn', 'wireguard')
|
|
|
|
#: The time slots of the site-to-site availability of each view of history(): the summary level read (None: the
|
|
#: details), the slot as PostgreSQL starts it from time, and its length.
|
|
UP_SLOTS = {
|
|
'hour': (None, "date_bin('5 minutes', time, timestamptz '2000-01-01 00:00:00+00')", '5 minutes'),
|
|
'day': (None, "date_trunc('hour', time, 'UTC')", '1 hour'),
|
|
'month': ('hourly', "date_trunc('day', time, 'UTC')", '1 day'),
|
|
'year': ('monthly', "date_trunc('month', time, 'UTC')", '1 month'),
|
|
}
|
|
|
|
#: The views of history(): the summary level read (None: the details), the bucket and the span.
|
|
RANGES = {
|
|
'hour': (None, '1 minute', '1 hour'),
|
|
'day': (None, '10 minutes', '1 day'),
|
|
'month': ('hourly', '6 hours', '1 month'),
|
|
'year': ('monthly', '1 month', '1 year'),
|
|
}
|
|
|
|
|
|
class DatabaseError(Exception):
|
|
"""The metrics database failed."""
|
|
|
|
|
|
def schema():
|
|
"""The statements that create the detail and the summary tables."""
|
|
statements = []
|
|
for table, (keys, values) in TABLES.items():
|
|
key_columns = ''.join(f', {key} text NOT NULL' for key in keys)
|
|
primary = ', '.join(('time', *keys))
|
|
detail = ''.join(f', {value} double precision' for value in values)
|
|
statements.append(f'CREATE TABLE IF NOT EXISTS {table} (time timestamptz NOT NULL{key_columns}{detail}, '
|
|
f'PRIMARY KEY ({primary}))')
|
|
summary = ''.join(f', {value}_avg double precision, {value}_max double precision' for value in values)
|
|
for suffix, _, _ in LEVELS:
|
|
statements.append(f'CREATE TABLE IF NOT EXISTS {table}_{suffix} (time timestamptz NOT NULL{key_columns}{summary}, '
|
|
f'PRIMARY KEY ({primary}))')
|
|
return statements
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def database(cfg):
|
|
"""A connection to the metrics database, with its tables, closed at the end; DatabaseError on any failure."""
|
|
try:
|
|
with psycopg.connect(vpnrouter_auth.connection_string(cfg), autocommit=True) as connection:
|
|
for statement in schema():
|
|
connection.execute(statement)
|
|
yield connection
|
|
except (psycopg.Error, OSError) as exc:
|
|
raise DatabaseError(f'the metrics database failed: {exc}') from exc
|
|
|
|
|
|
def write(connection, system, network, vpn, site_to_site):
|
|
"""Write a sample as the detail rows of the current minute.
|
|
|
|
system maps the values of metrics_system to numbers; network maps an interface to its rx and
|
|
tx; vpn maps a kind of VPN_KINDS to its clients; site_to_site maps a connection to the values
|
|
of metrics_site_to_site. A value that is None is written as NULL.
|
|
"""
|
|
with connection.transaction():
|
|
minute = connection.execute("SELECT date_trunc('minute', now(), 'UTC')").fetchone()[0]
|
|
names = TABLES['metrics_system'][1]
|
|
connection.execute(f"INSERT INTO metrics_system (time, {', '.join(names)}) "
|
|
f"VALUES (%s{', %s' * len(names)}) ON CONFLICT DO NOTHING",
|
|
(minute, *(system.get(name) for name in names)))
|
|
for interface, (rx, tx) in network.items():
|
|
connection.execute('INSERT INTO metrics_network (time, interface, rx, tx) VALUES (%s, %s, %s, %s) '
|
|
'ON CONFLICT DO NOTHING', (minute, interface, rx, tx))
|
|
for kind, clients in vpn.items():
|
|
connection.execute('INSERT INTO metrics_vpn (time, kind, clients) VALUES (%s, %s, %s) ON CONFLICT DO NOTHING',
|
|
(minute, kind, clients))
|
|
names = TABLES['metrics_site_to_site'][1]
|
|
for name, values in site_to_site.items():
|
|
connection.execute(f"INSERT INTO metrics_site_to_site (time, connection, {', '.join(names)}) "
|
|
f"VALUES (%s, %s{', %s' * len(names)}) ON CONFLICT DO NOTHING",
|
|
(minute, name, *(values.get(value) for value in names)))
|
|
|
|
|
|
def summarize(connection, level):
|
|
"""Summarize the complete periods of a level of LEVELS that are not summarized yet, for every detail table."""
|
|
index = [suffix for suffix, _, _ in LEVELS].index(level)
|
|
_, period, _ = LEVELS[index]
|
|
for table, (keys, values) in TABLES.items():
|
|
target = f'{table}_{level}'
|
|
source = table if index == 0 else f'{table}_{LEVELS[index - 1][0]}'
|
|
if index == 0:
|
|
aggregates = ', '.join(f'avg({value}), max({value})' for value in values)
|
|
else:
|
|
aggregates = ', '.join(f'avg({value}_avg), max({value}_max)' for value in values)
|
|
columns = ', '.join(('time', *keys, *(f'{value}_{kind}' for value in values for kind in ('avg', 'max'))))
|
|
groups = ''.join(f', {key}' for key in keys)
|
|
connection.execute(
|
|
f"INSERT INTO {target} ({columns}) "
|
|
f"SELECT date_trunc('{period}', time, 'UTC') AS period{groups}, {aggregates} FROM {source} "
|
|
f"WHERE time < date_trunc('{period}', now(), 'UTC') "
|
|
f"AND time >= coalesce((SELECT max(time) FROM {target}) + interval '1 {period}', '-infinity') "
|
|
f"GROUP BY period{groups} ON CONFLICT DO NOTHING")
|
|
|
|
|
|
def collect_garbage(connection, cfg):
|
|
"""Delete the rows older than the retention of their level: metrics.keep_details, keep_hourly and so on."""
|
|
metrics = cfg['metrics']
|
|
for table in TABLES:
|
|
connection.execute(f'DELETE FROM {table} WHERE time < now() - %s::interval', (metrics['keep_details'],))
|
|
for suffix, _, keep in LEVELS:
|
|
connection.execute(f'DELETE FROM {table}_{suffix} WHERE time < now() - %s::interval', (metrics[keep],))
|
|
|
|
|
|
def history(connection, view):
|
|
"""The points of a view of RANGES, oldest first, each with the start of its bucket in Unix seconds.
|
|
|
|
A point has the average and the maximum of each value of metrics_system, network, which maps
|
|
each interface to its rx and tx, clients, which maps each kind of VPN client to its clients,
|
|
and site_to_site, which maps each site-to-site connection to the values of
|
|
metrics_site_to_site; a value without data is None.
|
|
|
|
site_to_site_up is the availability of each site-to-site connection in the time slots of UP_SLOTS, oldest first:
|
|
for each slot its start in Unix seconds and, per connection, the percentage of its samples in which the connection
|
|
was up, 0 to 100; the summaries' averages of up give it for the month and the year.
|
|
"""
|
|
level, bucket, span = RANGES[view]
|
|
up_level, up_start, up_slot = UP_SLOTS[view]
|
|
|
|
def rows(table):
|
|
keys, values = TABLES[table]
|
|
source = table if level is None else f'{table}_{level}'
|
|
if level is None:
|
|
aggregates = ', '.join(f'avg({value}), max({value})' for value in values)
|
|
else:
|
|
aggregates = ', '.join(f'avg({value}_avg), max({value}_max)' for value in values)
|
|
start = (f"date_bin('{bucket}', time, timestamptz '2000-01-01 00:00:00+00')" if view != 'year'
|
|
else "date_trunc('month', time, 'UTC')")
|
|
groups = ''.join(f', {key}' for key in keys)
|
|
return connection.execute(
|
|
f"SELECT extract(epoch FROM {start})::bigint AS bucket{groups}, {aggregates} FROM {source} "
|
|
f"WHERE time >= now() - interval '{span}' GROUP BY bucket{groups} ORDER BY bucket").fetchall()
|
|
|
|
def pairs(row, values):
|
|
return {value: {'avg': row[2 * index], 'max': row[2 * index + 1]} for index, value in enumerate(values)}
|
|
|
|
points = {}
|
|
|
|
def point(start):
|
|
return points.setdefault(start, {'time': start, 'network': {}, 'clients': {}, 'site_to_site': {}})
|
|
|
|
for row in rows('metrics_system'):
|
|
point(row[0]).update(pairs(row[1:], TABLES['metrics_system'][1]))
|
|
for row in rows('metrics_network'):
|
|
point(row[0])['network'][row[1]] = pairs(row[2:], TABLES['metrics_network'][1])
|
|
for row in rows('metrics_vpn'):
|
|
point(row[0])['clients'][row[1]] = pairs(row[2:], TABLES['metrics_vpn'][1])['clients']
|
|
for row in rows('metrics_site_to_site'):
|
|
point(row[0])['site_to_site'][row[1]] = pairs(row[2:], TABLES['metrics_site_to_site'][1])
|
|
empty = {value: {'avg': None, 'max': None} for value in TABLES['metrics_system'][1]}
|
|
|
|
slots = {}
|
|
for start, name, percent in connection.execute(
|
|
f"SELECT extract(epoch FROM {up_start})::bigint AS slot, connection, avg({'up' if up_level is None else 'up_avg'}) * 100 "
|
|
f"FROM {'metrics_site_to_site' if up_level is None else f'metrics_site_to_site_{up_level}'} "
|
|
f"WHERE time >= now() - interval '{span}' GROUP BY slot, connection ORDER BY slot").fetchall():
|
|
slots.setdefault(start, {'time': start, 'connections': {}})['connections'][name] = percent
|
|
return {'range': view, 'bucket': bucket, 'points': [{**empty, **points[start]} for start in sorted(points)],
|
|
'site_to_site_up': {'slot': up_slot, 'points': [slots[start] for start in sorted(slots)]}}
|