Système d’analyse de logs en quasi temps réel avec suivi de fichier et agrégations sur fenêtres temporelles

import threading
import time
import re
from datetime import datetime, timedelta
import pandas as pd

# Configurations
LOG_FILE_PATH = 'app.log'  # Chemin du fichier de log à surveiller
WINDOW_SIZE_SEC = 60       # Durée de la fenêtre temporelle en secondes
AGGREGATION_INTERVAL_SEC = 10  # Intervalle de mise à jour de l'agrégation

# Regex pour le parsing des lignes de log : exemple de format '[2024-06-01 12:00:00] [INFO] Message'
LOG_LINE_REGEX = re.compile(r'\[(?P<timestamp>[^\]]+)\]\s+\[(?P<level>DEBUG|INFO|WARN|ERROR)\]\s+(?P<message>.*)')

# Format du timestamp dans le log
TIMESTAMP_FORMAT = '%Y-%m-%d %H:%M:%S'

# Buffer pour stocker les lignes de log parsées sous forme de dictionnaires
log_entries = []

# Verrou pour synchroniser l'accès à log_entries
log_lock = threading.Lock()


def tail_log_file(file_path):
    """
    Fonction qui effectue le tailing non bloquant du fichier de log.
    Lit uniquement les nouvelles lignes ajoutées au fichier.
    """
    with open(file_path, 'r', encoding='utf-8') as f:
        # Se positionner à la fin du fichier
        f.seek(0, 2)
        while True:
            line = f.readline()
            if not line:
                # Pas de nouvelle ligne, attendre un peu
                time.sleep(0.1)
                continue
            # Parsing de la ligne
            m = LOG_LINE_REGEX.match(line.strip())
            if m:
                try:
                    timestamp_str = m.group('timestamp')
                    level = m.group('level')
                    message = m.group('message')
                    timestamp = datetime.strptime(timestamp_str, TIMESTAMP_FORMAT)
                    entry = {'timestamp': timestamp, 'level': level, 'message': message}
                    # Ajout thread-safe
                    with log_lock:
                        log_entries.append(entry)
                except Exception as e:
                    # Ignorer les lignes mal formées
                    print(f'Erreur de parsing de ligne de log : {e}')
            else:
                # Ligne non conforme au format attendu
                print(f'Ligne de log non conforme : {line.strip()}')


def aggregate_and_print():
    """
    Fonction qui agrège périodiquement les données dans la fenêtre temporelle actuelle
    et affiche les statistiques.
    """
    while True:
        time.sleep(AGGREGATION_INTERVAL_SEC)
        now = datetime.now()
        window_start = now - timedelta(seconds=WINDOW_SIZE_SEC)

        with log_lock:
            # Filtrer uniquement les événements dans la fenêtre temporelle
            recent_entries = [e for e in log_entries if e['timestamp'] >= window_start]
            # Mettre à jour le buffer en conservant uniquement les événements récents
            log_entries[:] = recent_entries

        if not recent_entries:
            print(f'[{now.strftime(TIMESTAMP_FORMAT)}] Aucun événement dans la fenêtre temporelle.')
            continue

        # Création du DataFrame pour l'agrégation
        df = pd.DataFrame(recent_entries)

        # Comptage des événements par niveau
        counts = df['level'].value_counts().to_dict()

        # Affichage des statistiques
        print(f'[{now.strftime(TIMESTAMP_FORMAT)}] Statistiques des {WINDOW_SIZE_SEC} dernières secondes :')
        for level in ['DEBUG', 'INFO', 'WARN', 'ERROR']:
            count = counts.get(level, 0)
            print(f'  {level} : {count}')
        print('-' * 40)


def main():
    # Démarrage du thread pour le tailing du log
    tail_thread = threading.Thread(target=tail_log_file, args=(LOG_FILE_PATH,), daemon=True)
    tail_thread.start()

    # Démarrage de l'agrégation périodique
    aggregate_and_print()


if __name__ == '__main__':
    main()