Dépannage des bases de données basées sur l'IA : surveillance intelligente, détection des anomalies et correction automatique

Les bases de données de production modernes génèrent des millions de métriques par minute : latences de requêtes, conflits de verrouillage, décalage de réplication, taux d'accès aux pools de mémoire tampon et épuisement des pools de connexions. Les alertes traditionnelles basées sur des seuils noient les équipes sous des faux positifs tout en passant à côté des modèles de dégradation subtils qui précèdent les pannes catastrophiques. L'IA et l'apprentissage automatique modifient fondamentalement cette équation en apprenant le comportement normal, en détectant les anomalies avant qu'elles ne se produisent, en optimisant automatiquement les requêtes et en exécutant des mesures correctives sans intervention humaine. Ce guide couvre l'ensemble du dépannage des bases de données basées sur l'IA sur MySQL, PostgreSQL, MongoDB, Redis et Couchbase.

Le pipeline de surveillance de la base de données IA

Avant de plonger dans des techniques spécifiques, il est essentiel de comprendre l’architecture de bout en bout d’un système de surveillance de base de données alimenté par l’IA. Le pipeline collecte des métriques brutes de chaque moteur de base de données, les stocke dans une base de données de séries chronologiques, les alimente via des modèles ML pour la détection des anomalies, achemine les alertes via un gestionnaire d'alertes intelligent et déclenche des actions de correction automatique lorsque les seuils de confiance sont atteints.

IA/ML pour la surveillance et l'observabilité des bases de données

La surveillance traditionnelle des bases de données repose sur des seuils statiques : alerte lorsque CPU dépasse 80 %, lorsque la latence des requêtes dépasse 500 millisecondes ou lorsque le nombre de connexions dépasse 200. Cette approche échoue de manière catastrophique dans les environnements de production dynamiques où la normale varie en fonction de l'heure de la journée, du jour de la semaine, des modèles saisonniers et des événements de déploiement. L'observabilité basée sur l'IA remplace ces seuils rigides par des références apprises qui s'adaptent en permanence.

Collecter les bonnes mesures

La base de tout système de surveillance de l’IA est une collecte complète de métriques. Chaque moteur de base de données expose des métriques uniques importantes pour les performances :

# prometheus_db_collector.py — Unified metric collector for multi-DB environments
import prometheus_client as prom
import mysql.connector
import psycopg2
import pymongo
import redis
from couchbase.cluster import Cluster
from couchbase.options import ClusterOptions
from couchbase.auth import PasswordAuthenticator
import time
import logging

logger = logging.getLogger(__name__)

# MySQL metrics
mysql_slow_queries = prom.Gauge('mysql_slow_queries_total', 'Total slow queries')
mysql_buffer_pool_hit = prom.Gauge('mysql_innodb_buffer_pool_hit_ratio', 'Buffer pool hit ratio')
mysql_deadlocks = prom.Counter('mysql_deadlocks_total', 'Total deadlocks detected')
mysql_repl_lag = prom.Gauge('mysql_replication_lag_seconds', 'Replication lag in seconds')
mysql_active_connections = prom.Gauge('mysql_active_connections', 'Current active connections')
mysql_threads_running = prom.Gauge('mysql_threads_running', 'Currently running threads')

# PostgreSQL metrics
pg_bloat_ratio = prom.Gauge('pg_table_bloat_ratio', 'Table bloat ratio', ['table_name'])
pg_vacuum_age = prom.Gauge('pg_vacuum_age_seconds', 'Seconds since last vacuum', ['table_name'])
pg_index_hit_ratio = prom.Gauge('pg_index_hit_ratio', 'Index hit ratio')
pg_wal_rate = prom.Gauge('pg_wal_bytes_per_second', 'WAL generation rate')
pg_active_locks = prom.Gauge('pg_active_locks', 'Number of active locks', ['lock_type'])

# MongoDB metrics
mongo_opcounters = prom.Gauge('mongo_opcounters', 'Operation counters', ['op_type'])
mongo_wiredtiger_cache = prom.Gauge('mongo_wiredtiger_cache_usage_pct', 'WiredTiger cache usage')
mongo_repl_lag = prom.Gauge('mongo_replication_lag_seconds', 'Replica set lag')

# Redis metrics
redis_memory_frag = prom.Gauge('redis_memory_fragmentation_ratio', 'Memory fragmentation ratio')
redis_evicted_keys = prom.Counter('redis_evicted_keys_total', 'Total evicted keys')
redis_keyspace_hitrate = prom.Gauge('redis_keyspace_hit_ratio', 'Keyspace hit ratio')


class UnifiedDBCollector:
    def __init__(self, config):
        self.config = config
        self.connections = {}

    def collect_mysql(self):
        conn = mysql.connector.connect(**self.config['mysql'])
        cursor = conn.cursor(dictionary=True)

        cursor.execute("SHOW GLOBAL STATUS LIKE 'Slow_queries'")
        row = cursor.fetchone()
        mysql_slow_queries.set(int(row['Value']))

        cursor.execute("""
            SELECT
                (1 - (Innodb_buffer_pool_reads / Innodb_buffer_pool_read_requests)) * 100
            AS hit_ratio FROM (
                SELECT
                    VARIABLE_VALUE AS Innodb_buffer_pool_reads
                FROM performance_schema.global_status
                WHERE VARIABLE_NAME = 'Innodb_buffer_pool_reads'
            ) a, (
                SELECT
                    VARIABLE_VALUE AS Innodb_buffer_pool_read_requests
                FROM performance_schema.global_status
                WHERE VARIABLE_NAME = 'Innodb_buffer_pool_read_requests'
            ) b
        """)
        result = cursor.fetchone()
        mysql_buffer_pool_hit.set(float(result['hit_ratio']))

        cursor.execute("SHOW GLOBAL STATUS LIKE 'Innodb_deadlocks'")
        row = cursor.fetchone()
        mysql_deadlocks.inc(int(row['Value']))

        cursor.execute("SHOW SLAVE STATUS")
        slave = cursor.fetchone()
        if slave and slave.get('Seconds_Behind_Master') is not None:
            mysql_repl_lag.set(float(slave['Seconds_Behind_Master']))

        cursor.execute("SHOW GLOBAL STATUS LIKE 'Threads_connected'")
        row = cursor.fetchone()
        mysql_active_connections.set(int(row['Value']))

        cursor.close()
        conn.close()

    def collect_postgresql(self):
        conn = psycopg2.connect(**self.config['postgresql'])
        cursor = conn.cursor()

        cursor.execute("""
            SELECT schemaname, tablename,
                   pg_total_relation_size(schemaname || '.' || tablename) as total_size,
                   pg_relation_size(schemaname || '.' || tablename) as table_size
            FROM pg_tables
            WHERE schemaname = 'public'
        """)
        for row in cursor.fetchall():
            if row[3] > 0:
                bloat = (row[2] - row[3]) / row[2]
                pg_bloat_ratio.labels(table_name=row[1]).set(bloat)

        cursor.execute("""
            SELECT relname, extract(epoch from now() - last_vacuum) as vacuum_age
            FROM pg_stat_user_tables
            WHERE last_vacuum IS NOT NULL
        """)
        for row in cursor.fetchall():
            pg_vacuum_age.labels(table_name=row[0]).set(row[1])

        cursor.execute("""
            SELECT sum(heap_blks_hit) / nullif(sum(heap_blks_hit) + sum(heap_blks_read), 0)
            FROM pg_statio_user_tables
        """)
        result = cursor.fetchone()
        if result[0]:
            pg_index_hit_ratio.set(float(result[0]))

        cursor.close()
        conn.close()

    def collect_mongodb(self):
        client = pymongo.MongoClient(self.config['mongodb']['uri'])
        status = client.admin.command('serverStatus')

        for op in ['insert', 'query', 'update', 'delete']:
            mongo_opcounters.labels(op_type=op).set(status['opcounters'][op])

        cache = status['wiredTiger']['cache']
        cache_used = cache['bytes currently in the cache']
        cache_max = cache['maximum bytes configured']
        mongo_wiredtiger_cache.set((cache_used / cache_max) * 100)

        client.close()

    def collect_redis(self):
        r = redis.Redis(**self.config['redis'])
        info = r.info()

        redis_memory_frag.set(info.get('mem_fragmentation_ratio', 0))
        redis_evicted_keys.inc(info.get('evicted_keys', 0))

        hits = info.get('keyspace_hits', 0)
        misses = info.get('keyspace_misses', 0)
        if hits + misses > 0:
            redis_keyspace_hitrate.set(hits / (hits + misses))

        r.close()

    def run(self, interval=15):
        prom.start_http_server(9100)
        logger.info('Metric collector started on :9100')
        while True:
            try:
                self.collect_mysql()
                self.collect_postgresql()
                self.collect_mongodb()
                self.collect_redis()
            except Exception as e:
                logger.error(f'Collection error: {e}')
            time.sleep(interval)

Détection d'anomalies avec analyse de séries chronologiques

La principale proposition de valeur de l’IA dans la surveillance des bases de données est la détection des anomalies, c’est-à-dire l’identification de modèles inhabituels qui s’écartent des références apprises. Trois algorithmes principaux dominent cet espace : Facebook Prophet pour la décomposition saisonnière, les réseaux LSTM pour les modèles temporels complexes et Isolation Forest pour la détection des valeurs aberrantes multivariées.

Implémentation de la détection des anomalies avec scikit-learn et Prophet

L'implémentation Python suivante présente un détecteur d'anomalies prêt pour la production qui combine Isolation Forest pour la détection multivariée avec Prophet pour la prévision de séries chronologiques. Cette double approche détecte à la fois les pics soudains et la dérive progressive.

# anomaly_detector.py — Production anomaly detection for database metrics
import numpy as np
import pandas as pd
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
from prophet import Prophet
from prometheus_api_client import PrometheusConnect
from datetime import datetime, timedelta
import warnings
import json
import logging

warnings.filterwarnings('ignore')
logger = logging.getLogger(__name__)


class DatabaseAnomalyDetector:
    def __init__(self, prometheus_url, contamination=0.05):
        self.prom = PrometheusConnect(url=prometheus_url, disable_ssl=True)
        self.scaler = StandardScaler()
        self.isolation_forest = IsolationForest(
            contamination=contamination,
            n_estimators=200,
            max_samples='auto',
            random_state=42,
            n_jobs=-1
        )
        self.prophet_models = {}
        self.baseline_stats = {}

    def fetch_metrics(self, query, hours=168):
        """Fetch metric data from Prometheus for the given time window."""
        end_time = datetime.now()
        start_time = end_time - timedelta(hours=hours)
        result = self.prom.custom_query_range(
            query=query,
            start_time=start_time,
            end_time=end_time,
            step='60s'
        )
        if not result:
            return pd.DataFrame()

        timestamps, values = [], []
        for point in result[0]['values']:
            timestamps.append(datetime.fromtimestamp(float(point[0])))
            values.append(float(point[1]))

        return pd.DataFrame({'timestamp': timestamps, 'value': values})

    def train_isolation_forest(self, metrics_dict):
        """Train Isolation Forest on multiple metric dimensions."""
        frames = []
        for name, df in metrics_dict.items():
            if not df.empty:
                series = df.set_index('timestamp')['value'].rename(name)
                frames.append(series)

        if not frames:
            raise ValueError('No metric data available for training')

        combined = pd.concat(frames, axis=1).dropna()
        scaled = self.scaler.fit_transform(combined)
        self.isolation_forest.fit(scaled)

        self.baseline_stats = {
            col: {'mean': combined[col].mean(), 'std': combined[col].std()}
            for col in combined.columns
        }
        logger.info(f'Isolation Forest trained on {len(combined)} samples, {len(frames)} features')
        return combined

    def train_prophet(self, metric_name, df):
        """Train a Prophet model for seasonal time-series forecasting."""
        if df.empty:
            return
        prophet_df = df.rename(columns={'timestamp': 'ds', 'value': 'y'})
        model = Prophet(
            changepoint_prior_scale=0.05,
            seasonality_prior_scale=10,
            holidays_prior_scale=10,
            daily_seasonality=True,
            weekly_seasonality=True,
            yearly_seasonality=False,
            interval_width=0.95
        )
        model.fit(prophet_df)
        self.prophet_models[metric_name] = model
        logger.info(f'Prophet model trained for {metric_name}')

    def detect_anomalies_multivariate(self, current_metrics):
        """Detect anomalies using Isolation Forest across multiple metrics."""
        scaled = self.scaler.transform(current_metrics)
        predictions = self.isolation_forest.predict(scaled)
        scores = self.isolation_forest.decision_function(scaled)

        anomalies = []
        for i, (pred, score) in enumerate(zip(predictions, scores)):
            if pred == -1:
                anomaly_score = max(0, min(1, 0.5 - score))
                anomalies.append({
                    'index': i,
                    'score': round(anomaly_score, 4),
                    'severity': 'critical' if anomaly_score > 0.8 else 'warning',
                    'values': current_metrics.iloc[i].to_dict()
                })
        return anomalies

    def detect_anomalies_timeseries(self, metric_name, df):
        """Detect anomalies using Prophet forecast bounds."""
        model = self.prophet_models.get(metric_name)
        if not model or df.empty:
            return []

        prophet_df = df.rename(columns={'timestamp': 'ds', 'value': 'y'})
        forecast = model.predict(prophet_df[['ds']])
        merged = prophet_df.merge(forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']], on='ds')

        anomalies = []
        for _, row in merged.iterrows():
            if row['y'] < row['yhat_lower'] or row['y'] > row['yhat_upper']:
                deviation = abs(row['y'] - row['yhat'])
                band = row['yhat_upper'] - row['yhat_lower']
                severity_score = min(1.0, deviation / band) if band > 0 else 0.5
                anomalies.append({
                    'timestamp': str(row['ds']),
                    'actual': round(row['y'], 4),
                    'predicted': round(row['yhat'], 4),
                    'lower': round(row['yhat_lower'], 4),
                    'upper': round(row['yhat_upper'], 4),
                    'score': round(severity_score, 4),
                    'severity': 'critical' if severity_score > 0.8 else 'warning'
                })
        return anomalies

    def run_full_analysis(self, db_type='mysql'):
        """Run complete anomaly detection pipeline for a database type."""
        metric_queries = {
            'mysql': {
                'cpu': 'rate(process_cpu_seconds_total{job="mysql"}[5m])',
                'connections': 'mysql_global_status_threads_connected',
                'slow_queries': 'rate(mysql_global_status_slow_queries[5m])',
                'buffer_pool_hit': 'mysql_global_status_innodb_buffer_pool_hit_ratio',
                'repl_lag': 'mysql_slave_status_seconds_behind_master'
            },
            'postgresql': {
                'cpu': 'rate(process_cpu_seconds_total{job="postgres"}[5m])',
                'connections': 'pg_stat_activity_count',
                'cache_hit': 'pg_stat_database_blks_hit / (pg_stat_database_blks_hit + pg_stat_database_blks_read)',
                'deadlocks': 'rate(pg_stat_database_deadlocks[5m])',
                'wal_rate': 'rate(pg_wal_lsn_diff[5m])'
            }
        }

        queries = metric_queries.get(db_type, metric_queries['mysql'])
        metrics = {}
        for name, query in queries.items():
            metrics[name] = self.fetch_metrics(query)

        self.train_isolation_forest(metrics)
        for name, df in metrics.items():
            self.train_prophet(name, df)

        results = {'db_type': db_type, 'anomalies': [], 'summary': {}}
        for name, df in metrics.items():
            ts_anomalies = self.detect_anomalies_timeseries(name, df)
            if ts_anomalies:
                results['anomalies'].extend([
                    {**a, 'metric': name} for a in ts_anomalies
                ])

        results['summary'] = {
            'total_anomalies': len(results['anomalies']),
            'critical': sum(1 for a in results['anomalies'] if a['severity'] == 'critical'),
            'warning': sum(1 for a in results['anomalies'] if a['severity'] == 'warning')
        }
        return results


if __name__ == '__main__':
    detector = DatabaseAnomalyDetector('http://prometheus:9090')
    results = detector.run_full_analysis('mysql')
    print(json.dumps(results, indent=2))

Alertes prédictives et alertes basées sur des seuils

Les alertes traditionnelles basées sur des seuils souffrent de deux modes de défaillance opposés. Fixez des seuils trop serrés et vous vous noierez sous les faux positifs lors des variations normales de charge. Si vous les relâchez trop, vous passerez à côté d'une véritable dégradation jusqu'à ce qu'elle devienne une panne totale. Les alertes prédictives résolvent ces deux problèmes en apprenant à quoi ressemble la « normalité » pour chaque métrique à chaque instant.

Intégration LLM pour les requêtes et l'optimisation de bases de données en langage naturel

Les grands modèles de langage comme GPT-4 et Claude peuvent servir d'assistants de base de données intelligents, traduisant des questions en langage naturel en SQL, analysant les plans EXPLAIN et suggérant des optimisations. Cette fonctionnalité transforme la façon dont les administrateurs de base de données et les développeurs interagissent avec les bases de données : au lieu de disséquer manuellement les plans d'exécution, ils peuvent décrire le problème dans un anglais simple et recevoir des recommandations concrètes.

Création d'un optimiseur de requêtes LLM

L'implémentation Python suivante crée un assistant d'optimisation de requêtes basé sur LLM qui analyse les plans EXPLAIN et suggère des améliorations. Il s'intègre à API d'OpenAI et inclut la création de contexte tenant compte des schémas.

# llm_query_optimizer.py — AI-powered database query optimization
import openai
import json
import mysql.connector
import psycopg2
import logging
from dataclasses import dataclass
from typing import Optional

logger = logging.getLogger(__name__)


@dataclass
class QueryAnalysis:
    original_query: str
    explain_plan: dict
    schema_context: str
    suggestions: list
    optimized_query: Optional[str]
    estimated_improvement: str


class LLMQueryOptimizer:
    def __init__(self, api_key, db_config, db_type='mysql', model='gpt-4'):
        self.client = openai.OpenAI(api_key=api_key)
        self.db_config = db_config
        self.db_type = db_type
        self.model = model

    def get_explain_plan(self, query):
        """Execute EXPLAIN ANALYZE and return the plan."""
        if self.db_type == 'mysql':
            conn = mysql.connector.connect(**self.db_config)
            cursor = conn.cursor(dictionary=True)
            cursor.execute(f'EXPLAIN FORMAT=JSON {query}')
            plan = cursor.fetchone()
            cursor.close()
            conn.close()
            return json.loads(plan['EXPLAIN'])
        elif self.db_type == 'postgresql':
            conn = psycopg2.connect(**self.db_config)
            cursor = conn.cursor()
            cursor.execute(f'EXPLAIN (FORMAT JSON, ANALYZE, BUFFERS) {query}')
            plan = cursor.fetchone()[0]
            cursor.close()
            conn.close()
            return plan

    def get_schema_context(self, tables):
        """Extract schema DDL and statistics for context."""
        context_parts = []
        if self.db_type == 'mysql':
            conn = mysql.connector.connect(**self.db_config)
            cursor = conn.cursor()
            for table in tables:
                cursor.execute(f'SHOW CREATE TABLE {table}')
                row = cursor.fetchone()
                context_parts.append(f'-- Table: {table}\n{row[1]}')

                cursor.execute(f'SHOW INDEX FROM {table}')
                indexes = cursor.fetchall()
                idx_info = '\n'.join([f'  Index: {idx[2]}, Column: {idx[4]}, Cardinality: {idx[6]}' for idx in indexes])
                context_parts.append(f'-- Indexes for {table}:\n{idx_info}')

                cursor.execute(f"SELECT table_rows, data_length, index_length FROM information_schema.tables WHERE table_name = '{table}'")
                stats = cursor.fetchone()
                if stats:
                    context_parts.append(f'-- Stats: rows={stats[0]}, data_size={stats[1]}, index_size={stats[2]}')
            cursor.close()
            conn.close()

        return '\n\n'.join(context_parts)

    def analyze_query(self, query, tables):
        """Full LLM analysis of a slow query."""
        explain_plan = self.get_explain_plan(query)
        schema_context = self.get_schema_context(tables)

        prompt = f"""You are an expert database administrator specializing in {self.db_type} performance tuning.

Analyze the following slow query, its EXPLAIN plan, and the schema context. Provide:
1. Root cause of poor performance
2. Specific index recommendations (with CREATE INDEX statements)
3. Query rewrite suggestions (with the rewritten SQL)
4. Estimated performance improvement
5. Any schema changes that would help

## Original Query
```sql
{query}
```

## EXPLAIN Plan
```json
{json.dumps(explain_plan, indent=2)}
```

## Schema Context
```
{schema_context}
```

Respond in JSON format:
{{
  "root_cause": "...",
  "index_recommendations": ["CREATE INDEX ...", ...],
  "rewritten_query": "SELECT ...",
  "estimated_improvement": "Nx faster",
  "schema_changes": ["..."],
  "explanation": "..."
}}"""

        response = self.client.chat.completions.create(
            model=self.model,
            messages=[
                {'role': 'system', 'content': 'You are an expert DBA. Return valid JSON only.'},
                {'role': 'user', 'content': prompt}
            ],
            temperature=0.1,
            response_format={'type': 'json_object'}
        )

        result = json.loads(response.choices[0].message.content)

        return QueryAnalysis(
            original_query=query,
            explain_plan=explain_plan,
            schema_context=schema_context,
            suggestions=result.get('index_recommendations', []),
            optimized_query=result.get('rewritten_query'),
            estimated_improvement=result.get('estimated_improvement', 'Unknown')
        )

    def batch_optimize(self, slow_query_log_path, top_n=20):
        """Parse slow query log and optimize the top N most impactful queries."""
        queries = self._parse_slow_log(slow_query_log_path)
        sorted_queries = sorted(queries, key=lambda q: q['total_time'], reverse=True)[:top_n]

        results = []
        for q in sorted_queries:
            try:
                tables = self._extract_tables(q['query'])
                analysis = self.analyze_query(q['query'], tables)
                results.append({
                    'query': q['query'],
                    'frequency': q['count'],
                    'total_time': q['total_time'],
                    'analysis': analysis
                })
                logger.info(f'Optimized query (est. {analysis.estimated_improvement}): {q["query"][:80]}')
            except Exception as e:
                logger.error(f'Failed to analyze query: {e}')
        return results

    def _parse_slow_log(self, path):
        queries = {}
        current_query = []
        current_time = 0
        with open(path) as f:
            for line in f:
                if line.startswith('# Query_time:'):
                    parts = line.split()
                    current_time = float(parts[2])
                elif line.startswith('SET timestamp') or line.startswith('#'):
                    continue
                elif line.strip().endswith(';'):
                    current_query.append(line.strip())
                    full_query = ' '.join(current_query)
                    if full_query not in queries:
                        queries[full_query] = {'query': full_query, 'count': 0, 'total_time': 0}
                    queries[full_query]['count'] += 1
                    queries[full_query]['total_time'] += current_time
                    current_query = []
                else:
                    current_query.append(line.strip())
        return list(queries.values())

    def _extract_tables(self, query):
        import re
        tables = set()
        for match in re.finditer(r'(?:FROM|JOIN|INTO|UPDATE)\s+[`"]?(\w+)[`"]?', query, re.IGNORECASE):
            tables.add(match.group(1))
        return list(tables)


if __name__ == '__main__':
    import os
    optimizer = LLMQueryOptimizer(
        api_key=os.environ['OPENAI_API_KEY'],
        db_config={'host': 'localhost', 'user': 'root', 'password': '', 'database': 'app_db'},
        db_type='mysql'
    )
    analysis = optimizer.analyze_query(
        'SELECT * FROM orders o JOIN users u ON o.user_id = u.id WHERE o.status = "pending" AND o.created_at > "2026-01-01" ORDER BY o.created_at DESC LIMIT 100',
        ['orders', 'users']
    )
    print(json.dumps(analysis.__dict__, indent=2, default=str))

Workflows de correction automatique

La correction automatique est le domaine dans lequel la surveillance des bases de données basée sur l'IA offre le retour sur investissement le plus tangible. Au lieu de réveiller un administrateur de base de données à 3 heures du matin pour supprimer une requête incontrôlable ou mettre à l'échelle les réplicas en lecture, le système la gère automatiquement avec des pistes d'audit complètes et un score de confiance.

# auto_remediation.py — Automated database issue remediation
import subprocess
import mysql.connector
import psycopg2
import pymongo
import redis
import logging
import json
from datetime import datetime
from enum import Enum

logger = logging.getLogger(__name__)


class Severity(Enum):
    LOW = 'low'
    MEDIUM = 'medium'
    HIGH = 'high'
    CRITICAL = 'critical'


class RemediationAction:
    def __init__(self, name, description, severity_threshold, confidence_threshold=0.9):
        self.name = name
        self.description = description
        self.severity_threshold = severity_threshold
        self.confidence_threshold = confidence_threshold


class AutoRemediator:
    def __init__(self, db_configs, notification_webhook=None):
        self.db_configs = db_configs
        self.webhook = notification_webhook
        self.action_log = []

    def _log_action(self, action, target, result, confidence):
        entry = {
            'timestamp': datetime.utcnow().isoformat(),
            'action': action,
            'target': target,
            'result': result,
            'confidence': confidence
        }
        self.action_log.append(entry)
        logger.info(f'Remediation: {json.dumps(entry)}')
        if self.webhook:
            self._notify(entry)

    def kill_long_running_queries(self, db_type='mysql', max_duration_seconds=300, confidence=0.95):
        """Kill queries exceeding duration threshold."""
        if confidence < 0.9:
            logger.warning(f'Low confidence ({confidence}), skipping kill action')
            return []

        killed = []
        if db_type == 'mysql':
            conn = mysql.connector.connect(**self.db_configs['mysql'])
            cursor = conn.cursor(dictionary=True)
            cursor.execute("""
                SELECT id, user, host, db, time, state, info
                FROM information_schema.processlist
                WHERE command != 'Sleep'
                  AND time > %s
                  AND user != 'system user'
                ORDER BY time DESC
            """, (max_duration_seconds,))

            for proc in cursor.fetchall():
                try:
                    cursor.execute(f'KILL {proc["id"]}')
                    killed.append(proc)
                    self._log_action('kill_query', f'mysql:{proc["id"]}', 'success', confidence)
                except Exception as e:
                    self._log_action('kill_query', f'mysql:{proc["id"]}', f'failed: {e}', confidence)

            cursor.close()
            conn.close()

        elif db_type == 'postgresql':
            conn = psycopg2.connect(**self.db_configs['postgresql'])
            cursor = conn.cursor()
            cursor.execute("""
                SELECT pid, usename, application_name, state,
                       extract(epoch from now() - query_start) as duration, query
                FROM pg_stat_activity
                WHERE state = 'active'
                  AND extract(epoch from now() - query_start) > %s
                  AND usename != 'postgres'
            """, (max_duration_seconds,))

            for row in cursor.fetchall():
                try:
                    cursor.execute('SELECT pg_terminate_backend(%s)', (row[0],))
                    conn.commit()
                    killed.append({'pid': row[0], 'user': row[1], 'duration': row[4]})
                    self._log_action('kill_query', f'pg:{row[0]}', 'success', confidence)
                except Exception as e:
                    self._log_action('kill_query', f'pg:{row[0]}', f'failed: {e}', confidence)

            cursor.close()
            conn.close()

        return killed

    def scale_read_replicas(self, platform='kubernetes', target_replicas=None, confidence=0.92):
        """Scale database read replicas based on load prediction."""
        if confidence < 0.85:
            logger.warning('Insufficient confidence for scaling action')
            return None

        if platform == 'kubernetes':
            cmd = f'kubectl scale statefulset mysql-read --replicas={target_replicas}'
            result = subprocess.run(cmd.split(), capture_output=True, text=True)
            self._log_action('scale_replicas', f'k8s:mysql-read:{target_replicas}', result.stdout.strip(), confidence)
            return result.stdout
        elif platform == 'aws':
            import boto3
            rds = boto3.client('rds')
            response = rds.create_db_instance_read_replica(
                DBInstanceIdentifier=f'read-replica-{datetime.now().strftime("%Y%m%d%H%M")}',
                SourceDBInstanceIdentifier='production-primary'
            )
            self._log_action('create_replica', 'aws:rds', response['DBInstance']['DBInstanceIdentifier'], confidence)
            return response

    def trigger_failover(self, db_type='mysql', confidence=0.98):
        """Initiate database failover when primary is unhealthy."""
        if confidence < 0.95:
            logger.critical(f'Failover requires confidence >= 0.95, got {confidence}. Escalating to human.')
            self._notify({'action': 'failover_escalation', 'confidence': confidence})
            return None

        self._log_action('failover_initiated', db_type, 'starting', confidence)

        if db_type == 'mysql':
            result = subprocess.run(
                ['mysqlsh', '--', 'dba', 'switchToSecondary'],
                capture_output=True, text=True
            )
            self._log_action('failover', 'mysql:innodb_cluster', result.stdout.strip(), confidence)
        elif db_type == 'postgresql':
            result = subprocess.run(
                ['patronictl', 'failover', '--force'],
                capture_output=True, text=True
            )
            self._log_action('failover', 'pg:patroni', result.stdout.strip(), confidence)

    def flush_redis_hotspot(self, pattern, confidence=0.9):
        """Identify and handle Redis key hotspots."""
        r = redis.Redis(**self.db_configs['redis'])
        cursor = 0
        hot_keys = []
        while True:
            cursor, keys = r.scan(cursor, match=pattern, count=1000)
            for key in keys:
                idle = r.object('idletime', key)
                if idle is not None and idle < 5:
                    hot_keys.append(key.decode())
            if cursor == 0:
                break

        if hot_keys:
            self._log_action('hotspot_detected', f'redis:{pattern}', f'{len(hot_keys)} hot keys', confidence)
        return hot_keys

    def run_pg_vacuum(self, table, confidence=0.92):
        """Force VACUUM ANALYZE on bloated PostgreSQL tables."""
        conn = psycopg2.connect(**self.db_configs['postgresql'])
        conn.autocommit = True
        cursor = conn.cursor()
        cursor.execute(f'VACUUM (VERBOSE, ANALYZE) {table}')
        self._log_action('vacuum', f'pg:{table}', 'completed', confidence)
        cursor.close()
        conn.close()

    def _notify(self, payload):
        import requests
        try:
            requests.post(self.webhook, json=payload, timeout=5)
        except Exception as e:
            logger.error(f'Notification failed: {e}')

Dépannage de l'IA spécifique à MySQL

MySQL présente des défis uniques qui bénéficient énormément de l'analyse de l'IA. La gestion du pool de tampons InnoDB, la détection des blocages, la reconnaissance des modèles de requêtes lentes et la prédiction du décalage de réplication nécessitent chacune des modèles de ML spécialisés formés sur des métriques spécifiques à MySQL.

Analyse lente des requêtes avec ML

Au lieu d'examiner manuellement le journal des requêtes lentes, un modèle ML classe les requêtes en fonction de leur impact sur les performances et de leur cause première. Les modèles courants incluent les index manquants, les jointures cartésiennes, les clauses WHERE sous-optimales avec des fonctions sur les colonnes indexées et SELECT * sur les tables larges.

Optimisation du pool de tampons InnoDB

Le taux de réussite du pool de mémoire tampon est la mesure la plus critique de MySQL. Les modèles d'IA apprennent la relation entre les modèles de charge de travail et l'efficacité du pool de mémoire tampon, prédisant le moment où le taux de réussite se dégradera et recommandant des ajustements proactifs de innodb_buffer_pool_size. Un modèle LSTM formé sur les métriques du pool de mémoire tampon peut prédire la pression du cache 30 minutes avant qu'elle n'impacte la latence des requêtes.

Détection et prévention des blocages

L'IA analyse les graphiques de blocage d'InnoDB pour identifier les modèles récurrents. Plutôt que de simplement enregistrer les blocages après leur apparition, le système apprend quelles séquences de transactions conduisent à des blocages et peut réorganiser les opérations ou ajuster les niveaux d'isolement de manière préventive.

Dépannage de l'IA spécifique au PostgreSQL

L'architecture MVCC de PostgreSQL crée des défis uniques en matière de gonflement des tables, de planification sous vide et de gestion des WAL qui bénéficient d'une analyse basée sur l'IA.

Analyse du vide et détection des ballonnements

Les modèles d'IA suivent la relation entre les taux de transaction, l'accumulation de tuples morts et l'efficacité de l'autovide. En apprenant le taux de croissance du ballonnement pour chaque table, le système prédit quand les tables atteindront des niveaux de ballonnement problématiques et déclenchera des opérations de vide ciblées avant que les performances ne se dégradent.

Recommandations d'index

L'analyse conjointe de pg_stat_user_indexes et de pg_stat_statements révèle des modèles d'utilisation des index. L'IA identifie les index inutilisés consommant de l'espace disque et suggère de nouveaux index basés sur des modèles de requête, en tenant compte du coût d'amplification en écriture des index supplémentaires par rapport aux avantages en termes de performances de lecture.

Optimisation du pool de connexions

PostgreSQL gère les connexions différemment de MySQL, chaque connexion consommant beaucoup plus de mémoire. Les modèles d'IA analysent les modèles d'utilisation du pool de connexions sur PgBouncer pour déterminer les tailles de pool optimales pour différents profils de charge de travail (OLTP vs OLAP vs mixte), évitant ainsi à la fois le manque de connexion et l'épuisement de la mémoire.

Dépannage de l'IA spécifique à MongoDB

Le modèle de document et l'architecture distribuée de MongoDB créent un ensemble distinct de défis de performances que l'IA peut résoudre efficacement.

Suggestions d'index

L'analyse IA du profileur de requêtes MongoDB identifie les requêtes effectuant des analyses de collection (COLLSCAN) et recommande des index composés basés sur des combinaisons de champs de requête. Le modèle prend en compte la sélectivité, l'ordre des champs et l'optimisation des requêtes couvertes pour générer des spécifications d'index optimales.

Optimisation du partage

Pour les clusters partitionnés, l'IA surveille la distribution des fragments, les taux de migration et les modèles de routage des requêtes. Lorsqu'il détecte une utilisation inégale des fragments (fragments chauds), il recommande des changements de clé de fragment ou des stratégies de pré-fractionnement. Les modèles ML prédisent les taux de croissance des fragments pour équilibrer de manière proactive la distribution des données avant que des impacts sur les performances ne se produisent.

Analyse du cache WiredTiger

Les modèles d'expulsion du cache WiredTiger révèlent les caractéristiques de la charge de travail. Les modèles d'IA apprennent quand la pression du cache est causée par la croissance des ensembles de travail ou par des modèles d'accès inefficaces, recommandant soit une augmentation de la taille du cache, soit des modifications au niveau de l'application, comme le traitement par lots de requêtes.

Dépannage de l'IA spécifique à Redis

Redis fonctionne sous des contraintes différentes de celles des bases de données sur disque : la mémoire est la ressource critique et les exigences de latence sont souvent inférieures à la milliseconde.

Analyse de la mémoire

L'IA suit les taux de fragmentation de la mémoire, les distributions de tailles de clés et les modèles TTL. Lorsque la fragmentation dépasse les seuils sains, le système détermine si un ajustement ACTIVEDEFRAG ou un redémarrage contrôlé constitue la meilleure solution. Les modèles ML prédisent les trajectoires de croissance de la mémoire pour éviter les suppressions de MOO.

Détection des modèles clés et identification des points chauds

Grâce à l'échantillonnage MONITOR et à l'analyse OBJECT FREQ, l'IA identifie les touches de raccourci provoquant une répartition inégale de la charge entre les emplacements du cluster. Pour les déploiements de cluster Redis, le système détecte les goulots d'étranglement lors de la migration des emplacements et recommande des modifications de dénomination des clés pour améliorer la distribution des emplacements de hachage.

Optimisation de la politique d'expulsion

Différentes charges de travail bénéficient de différentes politiques d'expulsion (volatile-lru, allkeys-lfu, volatile-ttl). L'IA analyse les modèles d'accès pour recommander la politique de mémoire maximale optimale, en projetant l'impact du taux de réussite de chaque politique en fonction de la distribution actuelle des accès aux clés.

Dépannage de l'IA spécifique à Couchbase

Couchbase combine des capacités de stockage de documents, de valeurs-clés et de requêtes de type SQL (N1QL), créant ainsi un paysage d'optimisation unique.

Optimisation des requêtes N1QL

L'IA analyse les modèles de requête N1QL et la sortie EXPLAIN pour recommander la création de GSI (Global Secondaire Index), les stratégies d'index couvertes et les réécritures de requêtes. Le système apprend quels modèles N1QL produisent systématiquement des plans sous-optimaux et suggère de manière proactive des alternatives.

Intégration du conseiller en indices

Le conseiller d'index intégré de Couchbase fournit des recommandations, mais l'IA les améliore en prenant en compte la charge de travail globale, en équilibrant les coûts de création d'index et les avantages des requêtes sur l'ensemble des modèles d'accès de l'application plutôt que des requêtes individuelles isolément.

Planification du rééquilibrage

Lorsque des nœuds sont ajoutés ou supprimés, Couchbase doit rééquilibrer les données. L'IA prédit la durée du rééquilibrage, l'impact sur les ressources et les fenêtres temporelles optimales en fonction du comportement historique du cluster. Cela évite que les opérations de rééquilibrage n’aient un impact sur le trafic de production pendant les heures de pointe.

Architecture d'observabilité de l'IA multi-bases de données

La plupart des environnements de production exécutent plusieurs moteurs de base de données. Une plate-forme d'observabilité d'IA unifiée doit normaliser les métriques entre les moteurs, corréler les anomalies au sein de la couche de données et présenter une vue cohérente aux équipes opérationnelles.

Création d'un assistant de base de données IA personnalisé avec ChatGPT et Claude

L'intégration des LLM à votre infrastructure de base de données crée un assistant DBA interactif qui répond aux questions en langage naturel, diagnostique les problèmes et exécute les workflows de correction. L'assistant combine la génération augmentée par récupération (RAG) avec un accès aux métriques en temps réel.

# ai_dba_assistant.py — Custom AI DBA assistant with tool integration
import openai
import json
import os
from datetime import datetime


class AIDBAssistant:
    def __init__(self, db_connections, prometheus_url):
        self.client = openai.OpenAI(api_key=os.environ['OPENAI_API_KEY'])
        self.db_conns = db_connections
        self.prom_url = prometheus_url
        self.conversation_history = []
        self.tools = [
            {
                'type': 'function',
                'function': {
                    'name': 'query_prometheus',
                    'description': 'Execute a PromQL query to fetch database metrics',
                    'parameters': {
                        'type': 'object',
                        'properties': {
                            'query': {'type': 'string', 'description': 'PromQL query'},
                            'duration': {'type': 'string', 'description': 'Time range (e.g. 1h, 24h)'}
                        },
                        'required': ['query']
                    }
                }
            },
            {
                'type': 'function',
                'function': {
                    'name': 'run_explain',
                    'description': 'Run EXPLAIN on a SQL query',
                    'parameters': {
                        'type': 'object',
                        'properties': {
                            'query': {'type': 'string'},
                            'db_type': {'type': 'string', 'enum': ['mysql', 'postgresql']}
                        },
                        'required': ['query', 'db_type']
                    }
                }
            },
            {
                'type': 'function',
                'function': {
                    'name': 'get_active_queries',
                    'description': 'List currently running database queries',
                    'parameters': {
                        'type': 'object',
                        'properties': {
                            'db_type': {'type': 'string', 'enum': ['mysql', 'postgresql', 'mongodb']},
                            'min_duration_seconds': {'type': 'integer', 'default': 0}
                        },
                        'required': ['db_type']
                    }
                }
            },
            {
                'type': 'function',
                'function': {
                    'name': 'kill_query',
                    'description': 'Terminate a running database query by ID',
                    'parameters': {
                        'type': 'object',
                        'properties': {
                            'db_type': {'type': 'string'},
                            'process_id': {'type': 'integer'}
                        },
                        'required': ['db_type', 'process_id']
                    }
                }
            }
        ]

    def chat(self, user_message):
        self.conversation_history.append({'role': 'user', 'content': user_message})

        system_prompt = """You are an expert DBA assistant with access to real-time database monitoring tools.
You can query Prometheus metrics, analyze EXPLAIN plans, view active queries, and kill problematic queries.
Always ground your answers in actual data by using the available tools.
When diagnosing issues, follow this methodology:
1. Check current metrics for anomalies
2. Identify root cause
3. Suggest specific remediation steps
4. Execute remediation if the user approves"""

        messages = [{'role': 'system', 'content': system_prompt}] + self.conversation_history

        response = self.client.chat.completions.create(
            model='gpt-4',
            messages=messages,
            tools=self.tools,
            tool_choice='auto'
        )

        message = response.choices[0].message

        if message.tool_calls:
            for tool_call in message.tool_calls:
                fn_name = tool_call.function.name
                fn_args = json.loads(tool_call.function.arguments)
                result = self._execute_tool(fn_name, fn_args)
                self.conversation_history.append(message)
                self.conversation_history.append({
                    'role': 'tool',
                    'tool_call_id': tool_call.id,
                    'content': json.dumps(result)
                })

            follow_up = self.client.chat.completions.create(
                model='gpt-4',
                messages=[{'role': 'system', 'content': system_prompt}] + self.conversation_history
            )
            assistant_reply = follow_up.choices[0].message.content
        else:
            assistant_reply = message.content

        self.conversation_history.append({'role': 'assistant', 'content': assistant_reply})
        return assistant_reply

    def _execute_tool(self, name, args):
        if name == 'query_prometheus':
            from prometheus_api_client import PrometheusConnect
            prom = PrometheusConnect(url=self.prom_url)
            return prom.custom_query(args['query'])
        elif name == 'run_explain':
            return {'plan': 'EXPLAIN output here'}
        elif name == 'get_active_queries':
            return {'queries': []}
        elif name == 'kill_query':
            return {'status': 'killed', 'process_id': args['process_id']}
        return {'error': f'Unknown tool: {name}'}

Configuration du pipeline Prometheus + Grafana + ML

La pile d’observabilité constitue l’épine dorsale de la surveillance des bases de données d’IA. Prometheus récupère les métriques des exportateurs de bases de données, Grafana les visualise et un pipeline ML traite les données de séries chronologiques pour la détection des anomalies.

Configuration Prometheus pour la surveillance multi-DB

# prometheus.yml — Multi-database monitoring configuration
global:
  scrape_interval: 15s
  evaluation_interval: 15s

rule_files:
  - /etc/prometheus/rules/db_anomaly_rules.yml

alerting:
  alertmanagers:
    - static_configs:
        - targets: ['alertmanager:9093']

scrape_configs:
  - job_name: 'mysql'
    static_configs:
      - targets: ['mysql-exporter:9104']
    metrics_path: /metrics
    scrape_interval: 10s

  - job_name: 'postgresql'
    static_configs:
      - targets: ['postgres-exporter:9187']
    scrape_interval: 10s

  - job_name: 'mongodb'
    static_configs:
      - targets: ['mongodb-exporter:9216']
    scrape_interval: 15s

  - job_name: 'redis'
    static_configs:
      - targets: ['redis-exporter:9121']
    scrape_interval: 10s

  - job_name: 'couchbase'
    static_configs:
      - targets: ['couchbase-exporter:9420']
    scrape_interval: 15s

remote_write:
  - url: http://victoriametrics:8428/api/v1/write

Configuration personnalisée du tableau de bord Grafana

# grafana_dashboard_generator.py — Auto-generate AI-powered Grafana dashboards
import json
import requests


class GrafanaDashboardGenerator:
    def __init__(self, grafana_url, api_key):
        self.url = grafana_url
        self.headers = {'Authorization': f'Bearer {api_key}', 'Content-Type': 'application/json'}

    def create_db_overview_dashboard(self):
        dashboard = {
            'dashboard': {
                'title': 'AI Database Health Overview',
                'tags': ['database', 'ai', 'monitoring'],
                'timezone': 'browser',
                'panels': [
                    self._anomaly_score_panel(grid_pos={'x': 0, 'y': 0, 'w': 12, 'h': 8}),
                    self._query_latency_panel(grid_pos={'x': 12, 'y': 0, 'w': 12, 'h': 8}),
                    self._connection_pool_panel(grid_pos={'x': 0, 'y': 8, 'w': 8, 'h': 8}),
                    self._replication_lag_panel(grid_pos={'x': 8, 'y': 8, 'w': 8, 'h': 8}),
                    self._buffer_cache_panel(grid_pos={'x': 16, 'y': 8, 'w': 8, 'h': 8}),
                    self._remediation_log_panel(grid_pos={'x': 0, 'y': 16, 'w': 24, 'h': 6})
                ],
                'refresh': '10s'
            },
            'overwrite': True
        }
        resp = requests.post(f'{self.url}/api/dashboards/db', headers=self.headers, json=dashboard)
        return resp.json()

    def _anomaly_score_panel(self, grid_pos):
        return {
            'title': 'AI Anomaly Score (All Databases)',
            'type': 'timeseries',
            'gridPos': grid_pos,
            'targets': [
                {'expr': 'db_anomaly_score{db_type="mysql"}', 'legendFormat': 'MySQL'},
                {'expr': 'db_anomaly_score{db_type="postgresql"}', 'legendFormat': 'PostgreSQL'},
                {'expr': 'db_anomaly_score{db_type="mongodb"}', 'legendFormat': 'MongoDB'},
                {'expr': 'db_anomaly_score{db_type="redis"}', 'legendFormat': 'Redis'},
                {'expr': 'db_anomaly_score{db_type="couchbase"}', 'legendFormat': 'Couchbase'}
            ],
            'fieldConfig': {
                'defaults': {
                    'thresholds': {
                        'steps': [
                            {'value': 0, 'color': 'green'},
                            {'value': 0.5, 'color': 'yellow'},
                            {'value': 0.8, 'color': 'red'}
                        ]
                    },
                    'max': 1, 'min': 0
                }
            }
        }

    def _query_latency_panel(self, grid_pos):
        return {
            'title': 'Query Latency P95 with AI Prediction',
            'type': 'timeseries',
            'gridPos': grid_pos,
            'targets': [
                {'expr': 'histogram_quantile(0.95, rate(db_query_duration_seconds_bucket[5m]))', 'legendFormat': 'Actual P95'},
                {'expr': 'db_query_latency_predicted_p95', 'legendFormat': 'AI Predicted P95'}
            ]
        }

    def _connection_pool_panel(self, grid_pos):
        return {
            'title': 'Connection Pool Utilization',
            'type': 'gauge',
            'gridPos': grid_pos,
            'targets': [
                {'expr': 'db_connections_active / db_connections_max * 100', 'legendFormat': '{{db_type}}'}
            ]
        }

    def _replication_lag_panel(self, grid_pos):
        return {
            'title': 'Replication Lag (seconds)',
            'type': 'timeseries',
            'gridPos': grid_pos,
            'targets': [
                {'expr': 'mysql_slave_status_seconds_behind_master', 'legendFormat': 'MySQL'},
                {'expr': 'pg_replication_lag_seconds', 'legendFormat': 'PostgreSQL'},
                {'expr': 'mongodb_replset_member_replication_lag', 'legendFormat': 'MongoDB'}
            ]
        }

    def _buffer_cache_panel(self, grid_pos):
        return {
            'title': 'Buffer/Cache Hit Ratio',
            'type': 'stat',
            'gridPos': grid_pos,
            'targets': [
                {'expr': 'mysql_global_status_innodb_buffer_pool_hit_ratio', 'legendFormat': 'MySQL InnoDB'},
                {'expr': 'pg_stat_database_blks_hit / (pg_stat_database_blks_hit + pg_stat_database_blks_read)', 'legendFormat': 'PostgreSQL'},
                {'expr': 'redis_keyspace_hit_ratio', 'legendFormat': 'Redis'}
            ]
        }

    def _remediation_log_panel(self, grid_pos):
        return {
            'title': 'Auto-Remediation Action Log',
            'type': 'table',
            'gridPos': grid_pos,
            'targets': [
                {'expr': 'db_remediation_actions_total', 'format': 'table', 'instant': True}
            ]
        }

Intégration de PagerDuty et OpsGenie pour des alertes intelligentes

Les alertes intelligentes vont au-delà des simples notifications de webhook. Les alertes enrichies par l'IA incluent l'analyse des causes profondes, le contexte historique, les runbooks suggérés et les scores de confiance, donnant ainsi aux ingénieurs de garde le contexte dont ils ont besoin pour résoudre les problèmes plus rapidement ou confirmant que la remédiation automatique a déjà résolu le problème.

# intelligent_alerting.py — AI-enriched alerting for PagerDuty and OpsGenie
import requests
import json
from datetime import datetime


class IntelligentAlertManager:
    def __init__(self, pagerduty_key=None, opsgenie_key=None):
        self.pd_key = pagerduty_key
        self.og_key = opsgenie_key

    def send_enriched_alert(self, anomaly, ai_analysis):
        severity = anomaly.get('severity', 'warning')
        pd_severity = {'critical': 'critical', 'warning': 'warning', 'info': 'info'}.get(severity, 'warning')

        details = {
            'anomaly_score': anomaly.get('score', 0),
            'metric': anomaly.get('metric', 'unknown'),
            'root_cause': ai_analysis.get('root_cause', 'Under investigation'),
            'suggested_actions': ai_analysis.get('actions', []),
            'auto_remediation_status': ai_analysis.get('remediation_status', 'pending'),
            'similar_incidents': ai_analysis.get('similar_past_incidents', []),
            'estimated_impact': ai_analysis.get('impact', 'Unknown'),
            'confidence': ai_analysis.get('confidence', 0)
        }

        if self.pd_key:
            self._send_pagerduty(pd_severity, anomaly, details)
        if self.og_key:
            self._send_opsgenie(severity, anomaly, details)

    def _send_pagerduty(self, severity, anomaly, details):
        payload = {
            'routing_key': self.pd_key,
            'event_action': 'trigger',
            'payload': {
                'summary': f'[AI] Database anomaly: {anomaly["metric"]} (score: {anomaly["score"]})',
                'severity': severity,
                'source': 'ai-db-monitor',
                'component': anomaly.get('db_type', 'database'),
                'custom_details': details
            }
        }
        requests.post('https://events.pagerduty.com/v2/enqueue', json=payload)

    def _send_opsgenie(self, severity, anomaly, details):
        payload = {
            'message': f'[AI] Database anomaly: {anomaly["metric"]} (score: {anomaly["score"]})',
            'priority': {'critical': 'P1', 'warning': 'P3', 'info': 'P5'}.get(severity, 'P3'),
            'details': details,
            'tags': ['ai-monitoring', anomaly.get('db_type', 'database')]
        }
        requests.post(
            'https://api.opsgenie.com/v2/alerts',
            headers={'Authorization': f'GenieKey {self.og_key}'},
            json=payload
        )

Analyse des causes profondes avec l'IA

Lorsque des anomalies sont détectées, la détermination de la cause première est l’étape la plus longue de la réponse aux incidents. L'analyse des causes profondes basée sur l'IA met en corrélation plusieurs signaux (anomalies métriques, modèles de journaux, données de trace et changements récents) pour identifier la cause probable en quelques secondes plutôt qu'en quelques heures.

L'approche fonctionne en maintenant un graphe de connaissances des dépendances du système et des modes de défaillance connus. Lorsqu'une anomalie se déclenche, l'IA parcourt le graphique pour identifier les causes en amont. Par exemple, si la latence des requêtes augmente sur MySQL, le système vérifie : y a-t-il eu un déploiement récent ? Le nombre de connexions a-t-il changé ? Y a-t-il un décalage de réplication ? Les IOPS du disque sont-elles saturées ? Y a-t-il un conflit de verrouillage ? Chaque signal contribue à un score de probabilité pour différentes causes profondes.

Planification des capacités avec les prédictions ML

La planification des capacités basée sur le ML va au-delà de la mise à l'échelle réactive vers la gestion prédictive des ressources. En analysant les modèles de croissance historiques, les cycles saisonniers et les événements commerciaux planifiés, les modèles ML prévoient le moment où les bases de données atteindront les limites de ressources.

Prophet excelle dans la prévision de capacité car il gère de manière native les données manquantes, les changements de tendances et les modèles saisonniers. Entraînez-le sur 90 jours de données quotidiennes de croissance du stockage et il produit une prévision avec des intervalles de confiance indiquant quand vous aurez besoin de provisionner du stockage supplémentaire. Les modèles LSTM sont mieux adaptés à la prévision de capacité à court terme, en prévoyant les prochaines 24 heures d'utilisation du pool de connexions pour une pré-évolution avant les pics de trafic matinaux.

Outils d'IA spécifiques au cloud

AWS DevOps Guru pour RDS

AWS DevOps Guru fournit une détection des anomalies basée sur le ML pour les instances RDS. Il surveille automatiquement les métriques CloudWatch et identifie les anomalies de performances, en les corrélant avec les déploiements récents ou les modifications de configuration. L'intégration nécessite l'activation de DevOps Guru sur vos ressources RDS et la configuration des notifications SNS.

Azure AI pour Azure SQL et Cosmos DB

Azure fournit Intelligent Insights pour la base de données Azure SQL, qui utilise un modèle ML intégré pour détecter les régressions de performances, les requêtes bloquantes et les limites de ressources. Azure Cosmos DB comprend un conseiller IA intégré pour l'optimisation des unités de requête et la sélection des clés de partition.

Opérations cloud GCP pour Cloud SQL et Firestore

Google Cloud Operations (anciennement Stackdriver) propose des alertes intelligentes pour Cloud SQL. Le système apprend les références métriques et génère des alertes uniquement lorsque le comportement s'écarte considérablement des modèles appris, réduisant ainsi considérablement les faux positifs par rapport aux seuils statiques.

Outils de qualité des données open source

Apache Griffon

Apache Griffin permet de mesurer la qualité des données pour les actifs de données à grande échelle. Lorsqu'il est intégré à votre pipeline de surveillance de l'IA, il détecte les anomalies de qualité des données (valeurs manquantes, dérive de schéma, modifications de distribution) qui précèdent souvent les problèmes de performances de la base de données.

De grandes attentes

Great Expectations permet la validation déclarative des données. En définissant les attentes pour vos tables de base de données (nombre de lignes dans la plage, valeurs de colonnes dans les limites, intégrité référentielle), vous créez une couche de surveillance de la qualité des données que les modèles d'IA peuvent utiliser comme signaux supplémentaires pour la détection des anomalies.

# data_quality_check.py — Great Expectations integration for DB quality monitoring
import great_expectations as gx


def run_database_quality_checks(connection_string, suite_name='db_health'):
    context = gx.get_context()

    datasource = context.data_sources.add_sql(
        name='production_db',
        connection_string=connection_string
    )

    orders_asset = datasource.add_table_asset(name='orders', table_name='orders')
    batch = orders_asset.add_batch_definition_whole_table('full_table').get_batch()

    suite = context.suites.add(
        gx.ExpectationSuite(name=suite_name)
    )

    suite.add_expectation(
        gx.expectations.ExpectTableRowCountToBeBetween(min_value=1000, max_value=10000000)
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToNotBeNull(column='user_id')
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeUnique(column='order_number')
    )

    validation_result = batch.validate(suite)

    if not validation_result.success:
        failed = [r for r in validation_result.results if not r.success]
        return {
            'status': 'failed',
            'failed_checks': len(failed),
            'details': [{
                'expectation': str(r.expectation_config),
                'observed': r.result
            } for r in failed]
        }
    return {'status': 'passed', 'checks_run': len(validation_result.results)}

Exemple complet d'intégration de pipeline

Réunissant tous les composants, l'orchestrateur suivant relie la collecte de métriques, la détection des anomalies, l'analyse LLM, les alertes et la correction automatique dans un seul pipeline continu qui surveille tous les moteurs de base de données de votre environnement de production.

# pipeline_orchestrator.py — Full AI database monitoring pipeline
import schedule
import time
import logging
from anomaly_detector import DatabaseAnomalyDetector
from auto_remediation import AutoRemediator
from intelligent_alerting import IntelligentAlertManager
from llm_query_optimizer import LLMQueryOptimizer
import json
import os

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class AIDatabasePipeline:
    def __init__(self):
        self.detector = DatabaseAnomalyDetector(
            prometheus_url=os.environ['PROMETHEUS_URL']
        )
        self.remediator = AutoRemediator(
            db_configs={
                'mysql': {'host': os.environ['MYSQL_HOST'], 'user': 'monitor', 'password': os.environ['MYSQL_PASS'], 'database': 'production'},
                'postgresql': {'host': os.environ['PG_HOST'], 'user': 'monitor', 'password': os.environ['PG_PASS'], 'dbname': 'production'},
                'redis': {'host': os.environ['REDIS_HOST'], 'port': 6379}
            },
            notification_webhook=os.environ.get('SLACK_WEBHOOK')
        )
        self.alerter = IntelligentAlertManager(
            pagerduty_key=os.environ.get('PAGERDUTY_KEY'),
            opsgenie_key=os.environ.get('OPSGENIE_KEY')
        )
        self.optimizer = LLMQueryOptimizer(
            api_key=os.environ['OPENAI_API_KEY'],
            db_config={'host': os.environ['MYSQL_HOST'], 'user': 'root', 'password': os.environ['MYSQL_PASS'], 'database': 'production'},
            db_type='mysql'
        )

    def run_anomaly_detection_cycle(self):
        """Main detection cycle — runs every minute."""
        for db_type in ['mysql', 'postgresql']:
            try:
                results = self.detector.run_full_analysis(db_type)
                logger.info(f'{db_type}: {results["summary"]["total_anomalies"]} anomalies found')

                for anomaly in results['anomalies']:
                    if anomaly['severity'] == 'critical':
                        ai_analysis = self._analyze_anomaly(anomaly, db_type)
                        self.alerter.send_enriched_alert(anomaly, ai_analysis)

                        if ai_analysis.get('confidence', 0) > 0.95:
                            self._auto_remediate(anomaly, db_type, ai_analysis)
            except Exception as e:
                logger.error(f'Detection cycle failed for {db_type}: {e}')

    def run_query_optimization_cycle(self):
        """Batch query optimization — runs daily."""
        try:
            results = self.optimizer.batch_optimize('/var/log/mysql/slow.log', top_n=10)
            for r in results:
                logger.info(f'Query optimized: {r["analysis"].estimated_improvement}')
        except Exception as e:
            logger.error(f'Query optimization failed: {e}')

    def _analyze_anomaly(self, anomaly, db_type):
        return {
            'root_cause': f'Anomaly in {anomaly["metric"]} for {db_type}',
            'confidence': anomaly.get('score', 0.5),
            'actions': ['investigate', 'scale_if_needed'],
            'remediation_status': 'pending'
        }

    def _auto_remediate(self, anomaly, db_type, analysis):
        metric = anomaly.get('metric', '')
        confidence = analysis.get('confidence', 0)

        if 'slow_queries' in metric or 'query_latency' in metric:
            self.remediator.kill_long_running_queries(db_type=db_type, confidence=confidence)
        elif 'connections' in metric:
            self.remediator.scale_read_replicas(target_replicas=5, confidence=confidence)
        elif 'repl_lag' in metric and confidence > 0.98:
            self.remediator.trigger_failover(db_type=db_type, confidence=confidence)

        logger.info(f'Auto-remediation executed for {metric} on {db_type}')

    def start(self):
        logger.info('AI Database Pipeline started')
        schedule.every(1).minutes.do(self.run_anomaly_detection_cycle)
        schedule.every(1).day.at('02:00').do(self.run_query_optimization_cycle)

        while True:
            schedule.run_pending()
            time.sleep(10)


if __name__ == '__main__':
    pipeline = AIDatabasePipeline()
    pipeline.start()

Indicateurs clés à suivre pour le succès de la surveillance des bases de données IA

Meilleures pratiques et considérations de production

• Commencez par l'observabilité, puis ajoutez l'intelligence. Assurez-vous qu’une collecte complète de métriques est en place avant de déployer des modèles ML. Vous ne pouvez pas détecter d'anomalies dans les données que vous ne collectez pas.
• Utilisez des seuils de confiance pour la correction. Définissez des barres de confiance élevées (95 % ou plus) pour les actions destructrices telles que le basculement et des seuils inférieurs (85 %) pour les actions non destructives telles que la mise à l'échelle.
• Maintenir une surveillance humaine. La correction automatique doit toujours enregistrer les actions et avertir les humains. Les actions critiques telles que le basculement doivent nécessiter une confiance élevée ou une approbation humaine explicite.
• Recycler régulièrement les modèles. Les modèles de charge de travail des bases de données évoluent avec les modifications des applications. Recycler les modèles de détection d'anomalies au moins une fois par semaine ou mettre en œuvre un apprentissage en ligne qui s'adapte en permanence.
• Testez d’abord la remédiation lors de la préparation. Chaque flux de travail de correction automatique doit être validé dans un environnement de test avec des scénarios d'ingénierie du chaos avant d'être activé en production.
• Combinez plusieurs approches de ML. Aucun algorithme ne gère à lui seul tous les types d’anomalies. Utilisez des méthodes d’ensemble combinant Prophet (saisonnier), LSTM (séquentiel) et Isolation Forest (multivarié) pour une couverture complète.
• Intégrations LLM sécurisées. Lorsque vous utilisez des LLM pour l'analyse de requêtes, n'envoyez jamais de valeurs de données réelles : uniquement des métadonnées de schéma et des plans EXPLAIN. Utilisez des informations d'identification de base de données dédiées en lecture seule pour les outils d'IA.
• Créez des boucles de rétroaction. Suivez les taux de faux positifs et de faux négatifs pour la détection des anomalies. Utilisez les commentaires humains sur la pertinence des alertes pour améliorer continuellement la précision du modèle.

Conclusion

Le dépannage des bases de données basé sur l’IA représente un changement fondamental de la lutte réactive contre les incendies vers des opérations proactives et intelligentes. En combinant la détection des anomalies de séries chronologiques, l'optimisation des requêtes basée sur LLM, les alertes prédictives et la correction automatisée, les équipes peuvent réaliser une détection en moins d'une minute, une réduction spectaculaire des fausses alertes et des améliorations significatives du temps moyen de résolution. La clé est de construire progressivement : commencez par la collecte de métriques et de tableaux de bord, ajoutez la détection des anomalies, puis activez progressivement la correction automatique à mesure que la confiance dans le système augmente. Que vous gériez MySQL, PostgreSQL, MongoDB, Redis ou Couchbase, l'approche basée sur l'IA s'applique universellement, s'adaptant aux caractéristiques uniques de chaque moteur tout en offrant une expérience d'observabilité unifiée sur l'ensemble de votre couche de données.