new calculation approach with unique sessions, new API endpoint to get list of active sessions, fix for UNDEF user, UI and Back to support certificate management still under development
This commit is contained in:
Антон
@@ -109,6 +109,10 @@ class OpenVPNDataGatherer:
|
||||
# Инициализация модуля агрегации
|
||||
# Передаем ссылку на метод подключения к БД
|
||||
self.ts_aggregator = TimeSeriesAggregator(self.db_manager.get_connection)
|
||||
|
||||
# In-Memory Cache для отслеживания сессий (CN, RealAddress) -> {last_bytes...}
|
||||
# Используется для корректного расчета инкрементов при множественных сессиях одного CN
|
||||
self.session_cache = {}
|
||||
|
||||
def load_config(self, config_file):
|
||||
"""Загрузка конфигурации или создание дефолтной со сложной структурой"""
|
||||
@@ -284,6 +288,10 @@ class OpenVPNDataGatherer:
|
||||
# 6: Bytes Sent
|
||||
|
||||
if len(parts) >= 8 and parts[1] != 'Common Name':
|
||||
# SKIPPING 'UNDEF' CLIENTS
|
||||
if parts[1].strip() == 'UNDEF':
|
||||
continue
|
||||
|
||||
try:
|
||||
client = {
|
||||
'common_name': parts[1].strip(),
|
||||
@@ -304,51 +312,130 @@ class OpenVPNDataGatherer:
|
||||
return clients
|
||||
|
||||
def update_client_status_and_bytes(self, active_clients):
|
||||
"""Обновление статусов и расчет инкрементов трафика"""
|
||||
"""
|
||||
Обновление статусов и расчет инкрементов трафика.
|
||||
Использует In-Memory Cache (self.session_cache) для корректной обработки
|
||||
множественных сессий одного пользователя (Ping-Pong effect fix).
|
||||
"""
|
||||
conn = self.db_manager.get_connection()
|
||||
cursor = conn.cursor()
|
||||
|
||||
try:
|
||||
# Загружаем текущее состояние всех клиентов
|
||||
cursor.execute('SELECT id, common_name, status, last_bytes_received, last_bytes_sent FROM clients')
|
||||
db_clients = {}
|
||||
for row in cursor.fetchall():
|
||||
db_clients[row[1]] = {
|
||||
'id': row[0],
|
||||
'status': row[2],
|
||||
'last_bytes_received': row[3],
|
||||
'last_bytes_sent': row[4]
|
||||
}
|
||||
# 1. Загружаем текущее состояние CNs из БД для обновления статусов
|
||||
cursor.execute('SELECT id, common_name, status, total_bytes_received, total_bytes_sent FROM clients')
|
||||
db_clients = {row[1]: {'id': row[0], 'status': row[2]} for row in cursor.fetchall()}
|
||||
|
||||
active_names = set()
|
||||
# Структура для агрегации инкрементов по Common Name перед записью в БД
|
||||
# cn -> { 'inc_rx': 0, 'inc_tx': 0, 'curr_rx': 0, 'curr_tx': 0, 'real_address': '...'}
|
||||
cn_updates = {}
|
||||
|
||||
# Множество активных ключей сессий (CN, RealAddr) для очистки кэша
|
||||
active_session_keys = set()
|
||||
active_cns = set()
|
||||
|
||||
# 2. Обрабатываем каждую активную сессию
|
||||
for client in active_clients:
|
||||
name = client['common_name']
|
||||
active_names.add(name)
|
||||
|
||||
real_addr = client['real_address']
|
||||
curr_recv = client['bytes_received']
|
||||
curr_sent = client['bytes_sent']
|
||||
|
||||
if name in db_clients:
|
||||
# Клиент существует в базе
|
||||
db_client = db_clients[name]
|
||||
client['db_id'] = db_client['id'] # ID для агрегатора и истории
|
||||
# Уникальный ключ сессии
|
||||
session_key = (name, real_addr)
|
||||
active_session_keys.add(session_key)
|
||||
active_cns.add(name)
|
||||
|
||||
# --- ЛОГИКА РАСЧЕТА ДЕЛЬТЫ (In-Memory) ---
|
||||
if session_key in self.session_cache:
|
||||
prev_state = self.session_cache[session_key]
|
||||
prev_recv = prev_state['bytes_received']
|
||||
prev_sent = prev_state['bytes_sent']
|
||||
|
||||
# Проверка на рестарт сервера/сессии (сброс счетчиков)
|
||||
# Если текущее значение меньше сохраненного, значит был сброс -> берем всё текущее значение как дельту
|
||||
if curr_recv < db_client['last_bytes_received']:
|
||||
# Расчет RX
|
||||
if curr_recv < prev_recv:
|
||||
# Рестарт сессии (счетчик сбросился)
|
||||
inc_recv = curr_recv
|
||||
self.logger.info(f"Counter reset detected for {name} (Recv)")
|
||||
self.logger.info(f"Session reset detected for {session_key} (Recv)")
|
||||
else:
|
||||
inc_recv = curr_recv - db_client['last_bytes_received']
|
||||
|
||||
if curr_sent < db_client['last_bytes_sent']:
|
||||
inc_recv = curr_recv - prev_recv
|
||||
|
||||
# Расчет TX
|
||||
if curr_sent < prev_sent:
|
||||
inc_sent = curr_sent
|
||||
self.logger.info(f"Counter reset detected for {name} (Sent)")
|
||||
self.logger.info(f"Session reset detected for {session_key} (Sent)")
|
||||
else:
|
||||
inc_sent = curr_sent - db_client['last_bytes_sent']
|
||||
inc_sent = curr_sent - prev_sent
|
||||
|
||||
else:
|
||||
# Новая сессия (или после рестарта сервиса)
|
||||
# Если сервиса только запустился, мы не знаем предыдущего состояния.
|
||||
# Чтобы избежать спайков, считаем инкремент = 0 для первого появления,
|
||||
# если это похоже на продолжающуюся сессию (большие числа).
|
||||
# Если числа маленькие (<10MB), считаем как новую.
|
||||
|
||||
# Обновляем клиента
|
||||
# 10 MB threshold
|
||||
threshold = 10 * 1024 * 1024
|
||||
|
||||
if curr_recv < threshold and curr_sent < threshold:
|
||||
inc_recv = curr_recv
|
||||
inc_sent = curr_sent
|
||||
else:
|
||||
# Скорее всего рестарт сервиса, пропускаем первый тик
|
||||
inc_recv = 0
|
||||
inc_sent = 0
|
||||
self.logger.debug(f"New session tracking started for {session_key}. Initializing baseline.")
|
||||
|
||||
# Обновляем кэш
|
||||
if session_key not in self.session_cache:
|
||||
# New session
|
||||
self.session_cache[session_key] = {
|
||||
'bytes_received': curr_recv,
|
||||
'bytes_sent': curr_sent,
|
||||
'last_seen': datetime.now(),
|
||||
'connected_since': datetime.now() # Track start time
|
||||
}
|
||||
else:
|
||||
# Update existing
|
||||
self.session_cache[session_key]['bytes_received'] = curr_recv
|
||||
self.session_cache[session_key]['bytes_sent'] = curr_sent
|
||||
self.session_cache[session_key]['last_seen'] = datetime.now()
|
||||
|
||||
# Добавляем в клиентский объект (для истории/графиков)
|
||||
# Важно: это инкремент конкретной сессии
|
||||
client['bytes_received_inc'] = inc_recv
|
||||
client['bytes_sent_inc'] = inc_sent
|
||||
# Ensure db_id is available for active_sessions later (populated in step 4 or from cache)
|
||||
# We defer writing to active_sessions until we have DB IDs
|
||||
client['session_key'] = session_key
|
||||
|
||||
# --- АГРЕГАЦИЯ ДЛЯ БД (по Common Name) ---
|
||||
if name not in cn_updates:
|
||||
cn_updates[name] = {
|
||||
'inc_recv': 0, 'inc_tx': 0,
|
||||
'max_rx': 0, 'max_tx': 0, # Для last_bytes в БД сохраним текущие счетчики самой большой сессии (примерно)
|
||||
'real_address': real_addr # Берем последний адрес
|
||||
}
|
||||
|
||||
cn_updates[name]['inc_recv'] += inc_recv
|
||||
cn_updates[name]['inc_tx'] += inc_sent
|
||||
# Сохраняем "текущее" значение как максимальное из сессий, чтобы в БД last_bytes было хоть что-то осмысленное
|
||||
# (хотя при in-memory подходе поле last_bytes в БД теряет смысл для логики, но нужно для UI)
|
||||
cn_updates[name]['max_rx'] = max(cn_updates[name]['max_rx'], curr_recv)
|
||||
cn_updates[name]['max_tx'] = max(cn_updates[name]['max_tx'], curr_sent)
|
||||
cn_updates[name]['real_address'] = real_addr
|
||||
|
||||
# 3. Очистка кэша от мертвых сессий
|
||||
# Создаем список ключей для удаления, чтобы не менять словарь во время итерации
|
||||
dead_sessions = [k for k in self.session_cache if k not in active_session_keys]
|
||||
for k in dead_sessions:
|
||||
del self.session_cache[k]
|
||||
self.logger.debug(f"Removed inactive session from cache: {k}")
|
||||
|
||||
# 4. Обновление БД (Upsert Clients)
|
||||
for name, data in cn_updates.items():
|
||||
if name in db_clients:
|
||||
# UPDATE
|
||||
db_id = db_clients[name]['id']
|
||||
cursor.execute('''
|
||||
UPDATE clients
|
||||
SET status = 'Active',
|
||||
@@ -361,47 +448,72 @@ class OpenVPNDataGatherer:
|
||||
last_activity = CURRENT_TIMESTAMP
|
||||
WHERE id = ?
|
||||
''', (
|
||||
client['real_address'],
|
||||
inc_recv,
|
||||
inc_sent,
|
||||
curr_recv,
|
||||
curr_sent,
|
||||
db_client['id']
|
||||
data['real_address'],
|
||||
data['inc_recv'],
|
||||
data['inc_tx'],
|
||||
data['max_rx'],
|
||||
data['max_tx'],
|
||||
db_id
|
||||
))
|
||||
|
||||
client['bytes_received_inc'] = inc_recv
|
||||
client['bytes_sent_inc'] = inc_sent
|
||||
|
||||
# Прокидываем DB ID обратно в объекты клиентов (для TSDB)
|
||||
# Так как active_clients - это список сессий, ищем все сессии этого юзера
|
||||
for client in active_clients:
|
||||
if client['common_name'] == name:
|
||||
client['db_id'] = db_id
|
||||
|
||||
else:
|
||||
# Новый клиент
|
||||
# INSERT (New Client)
|
||||
cursor.execute('''
|
||||
INSERT INTO clients
|
||||
(common_name, real_address, status, total_bytes_received, total_bytes_sent, last_bytes_received, last_bytes_sent)
|
||||
VALUES (?, ?, 'Active', 0, 0, ?, ?)
|
||||
''', (
|
||||
name,
|
||||
client['real_address'],
|
||||
curr_recv,
|
||||
curr_sent
|
||||
data['real_address'],
|
||||
data['max_rx'],
|
||||
data['max_tx']
|
||||
))
|
||||
|
||||
new_id = cursor.lastrowid
|
||||
client['db_id'] = new_id
|
||||
# Для первой записи считаем инкремент 0 (или можно считать весь трафик)
|
||||
client['bytes_received_inc'] = 0
|
||||
client['bytes_sent_inc'] = 0
|
||||
self.logger.info(f"New client added: {name}")
|
||||
# Прокидываем ID
|
||||
for client in active_clients:
|
||||
if client['common_name'] == name:
|
||||
client['db_id'] = new_id
|
||||
|
||||
# Помечаем отключенных
|
||||
for name, db_client in db_clients.items():
|
||||
if name not in active_names and db_client['status'] == 'Active':
|
||||
# 5. Помечаем отключенных
|
||||
for name, db_info in db_clients.items():
|
||||
if name not in active_cns and db_info['status'] == 'Active':
|
||||
cursor.execute('''
|
||||
UPDATE clients
|
||||
SET status = 'Disconnected', updated_at = CURRENT_TIMESTAMP
|
||||
WHERE id = ?
|
||||
''', (db_client['id'],))
|
||||
''', (db_info['id'],))
|
||||
self.logger.info(f"Client disconnected: {name}")
|
||||
|
||||
# 6. SYNC ACTIVE SESSIONS TO DB (Snapshot)
|
||||
# Clear old state
|
||||
cursor.execute('DELETE FROM active_sessions')
|
||||
|
||||
# Insert current state
|
||||
for client in active_clients:
|
||||
# client['db_id'] should be populated by now (from step 4)
|
||||
if 'db_id' in client and 'session_key' in client:
|
||||
sess_data = self.session_cache.get(client['session_key'])
|
||||
if sess_data:
|
||||
cursor.execute('''
|
||||
INSERT INTO active_sessions
|
||||
(client_id, common_name, real_address, bytes_received, bytes_sent, connected_since, last_seen)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?)
|
||||
''', (
|
||||
client['db_id'],
|
||||
client['common_name'],
|
||||
client['real_address'],
|
||||
client['bytes_received'],
|
||||
client['bytes_sent'],
|
||||
sess_data.get('connected_since', datetime.now()),
|
||||
sess_data.get('last_seen', datetime.now())
|
||||
))
|
||||
|
||||
conn.commit()
|
||||
|
||||
except Exception as e:
|
||||
|
||||
Reference in New Issue
Block a user