Door AI aangedreven databaseprobleemoplossing: intelligente monitoring, detectie van afwijkingen en automatisch herstel

Moderne productiedatabases genereren miljoenen statistieken per minuut: latentie van query's, vergrendelingsconflicten, replicatievertraging, hitratio's van bufferpools en uitputting van verbindingspools. Traditionele, op drempels gebaseerde waarschuwingen verdrinken teams in valse positieven, terwijl ze subtiele degradatiepatronen missen die aan catastrofale mislukkingen voorafgaan. AI en machinaal leren veranderen deze vergelijking fundamenteel door normaal gedrag te leren, afwijkingen te detecteren voordat ze in een stroomversnelling terechtkomen, zoekopdrachten automatisch te optimaliseren en herstel uit te voeren zonder menselijke tussenkomst. Deze handleiding behandelt het volledige spectrum van AI-aangedreven databaseprobleemoplossing in MySQL, PostgreSQL, MongoDB, Redis en Couchbase.

De AI-databasemonitoringpijplijn

Voordat we in specifieke technieken duiken, is het van cruciaal belang om de end-to-end architectuur van een AI-aangedreven databasemonitoringsysteem te begrijpen. De pijplijn verzamelt onbewerkte meetgegevens van elke database-engine, slaat deze op in een tijdreeksdatabase, voert ze door ML-modellen voor detectie van afwijkingen, routeert waarschuwingen via een intelligente waarschuwingsmanager en activeert automatische herstelacties wanneer aan de betrouwbaarheidsdrempels wordt voldaan.

AI/ML voor databasemonitoring en waarneembaarheid

Traditionele databasemonitoring is gebaseerd op statische drempels: waarschuwing wanneer CPU de 80 procent overschrijdt, wanneer de latentie van zoekopdrachten groter is dan 500 milliseconden, of wanneer het aantal verbindingen de 200 overschrijdt. Deze aanpak faalt catastrofaal in dynamische productieomgevingen waar normaal varieert per tijdstip, dag van de week, seizoenspatronen en implementatiegebeurtenissen. AI-gestuurde waarneembaarheid vervangt deze rigide drempels door aangeleerde basislijnen die zich voortdurend aanpassen.

Het verzamelen van de juiste statistieken

De basis van elk AI-monitoringsysteem is een uitgebreide verzameling van meetgegevens. Elke database-engine biedt unieke statistieken die van belang zijn voor de prestaties:

# 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)

Anomaliedetectie met tijdreeksanalyse

De kernwaardepropositie van AI bij databasemonitoring is het detecteren van afwijkingen: het identificeren van ongebruikelijke patronen die afwijken van aangeleerde basislijnen. Drie primaire algoritmen domineren deze ruimte: Facebook Prophet voor seizoensdecompositie, LSTM-netwerken voor complexe temporele patronen en Isolation Forest voor detectie van multivariate uitschieters.

Anomaliedetectie implementeren met scikit-learn en Prophet

De volgende Python-implementatie demonstreert een productieklare anomaliedetector die Isolation Forest voor multivariate detectie combineert met Prophet voor tijdreeksvoorspellingen. Deze dubbele aanpak vangt zowel plotselinge pieken als geleidelijke drift op.

# 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))

Voorspellende waarschuwingen versus op drempels gebaseerde waarschuwingen

Traditionele op drempels gebaseerde waarschuwingen hebben te kampen met twee tegengestelde faalmodi. Stel drempels te strak in en u verdrinkt in valse positieven tijdens normale belastingvariaties. Zet ze te los en je mist echte degradatie totdat het een volledige storing wordt. Voorspellende waarschuwingen lossen beide problemen op door te leren hoe 'normaal' er voor elke metriek op elk tijdstip uitziet.

LLM-integratie voor query's en optimalisatie van natuurlijke taaldatabases

Grote taalmodellen zoals GPT-4 en Claude kunnen dienen als intelligente database-assistenten, die vragen in natuurlijke taal vertalen naar SQL, EXPLAIN-plannen analyseren en optimalisaties voorstellen. Deze mogelijkheid transformeert de manier waarop DBA's en ontwikkelaars omgaan met databases: in plaats van het handmatig ontleden van uitvoeringsplannen, kunnen ze het probleem in gewoon Engels beschrijven en bruikbare aanbevelingen ontvangen.

Een LLM Query Optimizer bouwen

De volgende Python-implementatie creëert een door LLM aangedreven assistent voor query-optimalisatie die EXPLAIN-plannen analyseert en verbeteringen voorstelt. Het integreert met OpenAI's API en omvat schemabewuste contextopbouw.

# 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))

Werkstromen voor automatisch herstel

Bij automatisch herstel levert AI-aangedreven databasemonitoring de meest tastbare ROI op. In plaats van om drie uur 's nachts een DBA wakker te maken om een ​​op hol geslagen query te beëindigen of leesreplica's te schalen, handelt het systeem dit automatisch af met volledige audittrails en betrouwbaarheidsscores.

# 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}')

MySQL-specifieke AI-probleemoplossing

MySQL biedt unieke uitdagingen die enorm profiteren van AI-analyse. InnoDB-bufferpoolbeheer, deadlockdetectie, langzame querypatroonherkenning en replicatievertragingsvoorspelling vereisen elk gespecialiseerde ML-modellen die zijn getraind op MySQL-specifieke statistieken.

Langzame queryanalyse met ML

In plaats van het trage querylogboek handmatig te beoordelen, classificeert een ML-model query's op basis van hun impact op de prestaties en de hoofdoorzaak. Veelvoorkomende patronen zijn onder meer ontbrekende indexen, cartesiaanse joins, suboptimale WHERE-clausules met functies op geïndexeerde kolommen en SELECT * op brede tabellen.

Optimalisatie van de InnoDB-bufferpool

De hitratio van de bufferpool is de meest kritische maatstaf van MySQL. AI-modellen leren de relatie tussen werklastpatronen en de effectiviteit van de bufferpool, voorspellen wanneer de hitratio zal verslechteren en bevelen proactieve innodb_buffer_pool_size-aanpassingen aan. Een LSTM-model dat is getraind op metrische gegevens van de bufferpool kan de cachedruk 30 minuten voorspellen voordat deze de latentie van query's beïnvloedt.

Detectie en preventie van impasses

AI analyseert InnoDB-impassegrafieken om terugkerende patronen te identificeren. In plaats van impasses alleen maar te registreren nadat ze zich hebben voorgedaan, leert het systeem welke transactiereeksen tot impasses leiden en kan het de activiteiten opnieuw ordenen of isolatieniveaus preventief aanpassen.

PostgreSQL-specifieke AI-probleemoplossing

De MVCC-architectuur van PostgreSQL creëert unieke uitdagingen op het gebied van table bloat, vacuümplanning en WAL-beheer die profiteren van AI-gestuurde analyses.

Vacuümanalyse en zwellingsdetectie

AI-modellen volgen de relatie tussen transactiepercentages, accumulatie van dode tupels en de effectiviteit van autovacuüm. Door de groeisnelheid van elke tafel te leren kennen, voorspelt het systeem wanneer tafels problematische bloat-niveaus zullen bereiken en worden gerichte vacuümoperaties geactiveerd voordat de prestaties afnemen.

Indexaanbevelingen

Door pg_stat_user_indexes en pg_stat_statements samen te analyseren, worden indexgebruikspatronen onthuld. AI identificeert ongebruikte indexen die schijfruimte in beslag nemen en stelt nieuwe indexen voor op basis van querypatronen, waarbij rekening wordt gehouden met de schrijfversterkingskosten van extra indexen versus het leesprestatievoordeel.

Optimalisatie van verbindingspools

PostgreSQL verwerkt verbindingen anders dan MySQL, waarbij elke verbinding aanzienlijk meer geheugen verbruikt. AI-modellen analyseren gebruikspatronen van verbindingspools in PgBouncer om optimale poolgroottes te bepalen voor verschillende werklastprofielen (OLTP versus OLAP versus gemengd), waardoor zowel verbindingsgebrek als geheugenuitputting worden voorkomen.

MongoDB-specifieke AI-probleemoplossing

Het documentmodel en de gedistribueerde architectuur van MongoDB creëren een duidelijke reeks prestatie-uitdagingen die AI effectief kan aanpakken.

Indexsuggesties

AI-analyse van de MongoDB-queryprofiler identificeert query's die collectiescans uitvoeren (COLLSCAN) en beveelt samengestelde indexen aan op basis van queryveldcombinaties. Het model houdt rekening met selectiviteit, veldvolgorde en optimalisatie van gedekte zoekopdrachten om optimale indexspecificaties te genereren.

Sharding-optimalisatie

Voor gesharde clusters bewaakt AI de verdeling van segmenten, migratiesnelheden en routeringspatronen voor query's. Wanneer ongelijkmatig Shard-gebruik (hot shards) wordt gedetecteerd, worden er wijzigingen in de Shard-sleutel of pre-splitting-strategieën aanbevolen. ML-modellen voorspellen deelgroeipercentages om de gegevensdistributie proactief in evenwicht te brengen voordat prestatie-impact optreedt.

WiredTiger-cacheanalyse

WiredTiger-cache-verwijderingspatronen onthullen werklastkenmerken. AI-modellen leren wanneer de cachedruk wordt veroorzaakt door de groei van de werkset versus inefficiënte toegangspatronen, en bevelen een verhoging van de cachegrootte of wijzigingen op applicatieniveau aan, zoals het batchen van query's.

Redis-specifieke AI-probleemoplossing

Redis werkt onder andere beperkingen dan op schijven gebaseerde databases: geheugen is de kritische hulpbron en de latentievereisten bedragen vaak minder dan een milliseconde.

Geheugenanalyse

AI houdt geheugenfragmentatieverhoudingen, sleutelgrootteverdelingen en TTL-patronen bij. Wanneer de fragmentatie de gezonde drempelwaarden overschrijdt, bepaalt het systeem of een ACTIVEDEFRAG-aanpassing of een gecontroleerde herstart de betere oplossing is. ML-modellen voorspellen geheugengroeitrajecten om OOM-doden te voorkomen.

Sleutelpatroondetectie en hotspot-identificatie

Met behulp van MONITOR-sampling en OBJECT FREQ-analyse identificeert AI sneltoetsen die een ongelijkmatige verdeling van de belasting over clusterslots veroorzaken. Voor Redis Cluster-implementaties detecteert het systeem knelpunten bij de migratie van slots en worden wijzigingen in de sleutelnaam aanbevolen om de distributie van hash-slots te verbeteren.

Optimalisatie van het uitzettingsbeleid

Verschillende werklasten profiteren van verschillend uitzettingsbeleid (volatile-lru, allkeys-lfu, volatile-ttl). AI analyseert toegangspatronen om het optimale maximale geheugenbeleid aan te bevelen, waarbij de hitrate-impact van elk beleid wordt geprojecteerd op basis van de huidige sleuteltoegangsverdeling.

Couchbase-specifieke AI-probleemoplossing

Couchbase combineert documentopslag, sleutelwaarde en SQL-achtige (N1QL) querymogelijkheden, waardoor een uniek optimalisatielandschap ontstaat.

N1QL-queryoptimalisatie

AI analyseert N1QL-querypatronen en EXPLAIN-uitvoer om het maken van GSI (Global Secondary Index), gedekte indexstrategieën en het herschrijven van query's aan te bevelen. Het systeem leert welke N1QL-patronen consequent suboptimale plannen opleveren en stelt proactief alternatieven voor.

Indexadviseur-integratie

De ingebouwde indexadviseur van Couchbase geeft aanbevelingen, maar AI verbetert deze door rekening te houden met de mondiale werklast, waarbij de kosten voor het maken van indexen worden afgewogen tegen de queryvoordelen voor de gehele toegangspatronen van de applicatie, in plaats van individuele query's afzonderlijk.

Breng de planning opnieuw in evenwicht

Wanneer knooppunten worden toegevoegd of verwijderd, moet Couchbase de gegevens opnieuw in evenwicht brengen. AI voorspelt het opnieuw in evenwicht brengen van de duur, de impact van resources en optimale timingvensters op basis van historisch clustergedrag. Dit voorkomt dat het herbalanceren van activiteiten invloed heeft op het productieverkeer tijdens piekuren.

Architectuur voor observatie van AI met meerdere databases

De meeste productieomgevingen draaien meerdere database-engines. Een uniform AI-waarnemingsplatform moet de meetgegevens van verschillende motoren normaliseren, afwijkingen in de gegevenslaag correleren en een samenhangend beeld bieden aan operationele teams.

Een aangepaste AI-databaseassistent bouwen met ChatGPT en Claude

Door LLMs te integreren met uw database-infrastructuur ontstaat een interactieve DBA-assistent die vragen in natuurlijke taal beantwoordt, problemen diagnosticeert en herstelworkflows uitvoert. De assistent combineert ophaal-augmented generatie (RAG) met realtime metrische toegang.

# 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}'}

Prometheus + Grafana + ML-pijplijnconfiguratie

De observatiestapel vormt de ruggengraat van AI-databasemonitoring. Prometheus verzamelt statistieken van database-exporteurs, Grafana visualiseert ze en een ML-pijplijn verwerkt de tijdreeksgegevens voor detectie van afwijkingen.

Prometheus-configuratie voor monitoring van meerdere databases

# 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

Aangepaste Grafana-dashboardconfiguratie

# 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}
            ]
        }

PagerDuty en OpsGenie-integratie voor intelligente waarschuwingen

Intelligente waarschuwingen gaan verder dan eenvoudige webhookmeldingen. Met AI verrijkte waarschuwingen omvatten analyse van de hoofdoorzaak, historische context, voorgestelde runbooks en betrouwbaarheidsscores, waardoor technici op afroep de context krijgen die ze nodig hebben om problemen sneller op te lossen of te bevestigen dat automatisch herstel het probleem al heeft opgelost.

# 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
        )

Root Cause Analyse met AI

Wanneer afwijkingen worden gedetecteerd, is het vaststellen van de hoofdoorzaak de meest tijdrovende stap in de respons op incidenten. AI-gestuurde analyse van de hoofdoorzaak correleert meerdere signalen (metrische afwijkingen, logpatronen, traceergegevens en recente wijzigingen) om de waarschijnlijke oorzaak binnen enkele seconden in plaats van binnen enkele uren vast te stellen.

De aanpak werkt door het bijhouden van een kennisgrafiek van systeemafhankelijkheden en bekende faalwijzen. Wanneer een anomalie wordt geactiveerd, doorkruist de AI de grafiek om stroomopwaartse oorzaken te identificeren. Als de latentie van zoekopdrachten bijvoorbeeld op MySQL piekt, controleert het systeem: was er een recente implementatie? Is het aantal verbindingen veranderd? Is er sprake van replicatievertraging? Zijn schijf-IOPS verzadigd? Is er sprake van slotconflicten? Elk signaal draagt ​​bij aan een waarschijnlijkheidsscore voor verschillende hoofdoorzaken.

Capaciteitsplanning met ML-voorspellingen

ML-gestuurde capaciteitsplanning gaat verder dan reactief schalen naar voorspellend resourcebeheer. Door historische groeipatronen, seizoenscycli en geplande zakelijke gebeurtenissen te analyseren, voorspellen ML-modellen wanneer databases tegen hun resourcelimieten aanlopen.

Prophet blinkt uit in capaciteitsvoorspellingen omdat het ontbrekende gegevens, trendveranderingen en seizoenspatronen native verwerkt. Train het op 90 dagen aan gegevens over de dagelijkse opslaggroei en het produceert een voorspelling met betrouwbaarheidsintervallen die aangeven wanneer u extra opslag nodig heeft. LSTM-modellen zijn beter geschikt voor capaciteitsvoorspellingen op de korte termijn, waarbij wordt voorspeld dat het gebruik van de verbindingspool de komende 24 uur vooraf zal worden geschaald vóór pieken in het verkeer in de ochtend.

Cloudspecifieke AI-tools

AWS DevOps-goeroe voor RDS

AWS DevOps Guru biedt ML-aangedreven anomaliedetectie voor RDS-instanties. Het bewaakt automatisch de CloudWatch-statistieken en identificeert prestatieafwijkingen, en correleert deze met recente implementaties of configuratiewijzigingen. Voor integratie is het inschakelen van DevOps Guru op uw RDS-bronnen en het configureren van SNS-meldingen vereist.

Azure AI voor Azure SQL en Cosmos DB

Azure biedt Intelligent Insights voor Azure SQL Database, dat gebruikmaakt van een ingebouwd ML-model om prestatieregressies, blokkeerquery's en resourcelimieten te detecteren. Azure Cosmos DB bevat een geïntegreerde AI-adviseur voor optimalisatie van aanvraageenheden en selectie van partitiesleutels.

GCP Cloud Operations voor Cloud SQL en Firestore

Google Cloud Operations (voorheen Stackdriver) biedt intelligente waarschuwingen voor Cloud SQL. Het systeem leert metrische basislijnen en genereert alleen waarschuwingen wanneer het gedrag aanzienlijk afwijkt van aangeleerde patronen, waardoor het aantal valse positieven drastisch wordt verminderd in vergelijking met statische drempels.

Open source tools voor gegevenskwaliteit

Apache Griffioen

Apache Griffin biedt datakwaliteitsmeting voor grootschalige data-assets. Wanneer het wordt geïntegreerd met uw AI-monitoringpijplijn, detecteert het afwijkingen in de gegevenskwaliteit (ontbrekende waarden, schemaafwijkingen, distributiewijzigingen) die vaak voorafgaan aan problemen met de databaseprestaties.

Grote verwachtingen

Great Expectations maakt declaratieve gegevensvalidatie mogelijk. Door verwachtingen voor uw databasetabellen te definiëren (aantal rijen binnen bereik, kolomwaarden binnen de grenzen, referentiële integriteit), creëert u een monitoringlaag voor gegevenskwaliteit die AI-modellen kunnen gebruiken als aanvullende signalen voor de detectie van afwijkingen.

# 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)}

Compleet voorbeeld van pijplijnintegratie

Door alle componenten samen te brengen, verbindt de volgende Orchestrator het verzamelen van metrische gegevens, detectie van afwijkingen, LLM-analyse, waarschuwingen en automatisch herstel in één doorlopende pijplijn die alle database-engines in uw productieomgeving bewaakt.

# 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()

Belangrijke statistieken die u moet volgen voor het succes van AI-databasemonitoring

Beste praktijken en productieoverwegingen

• Begin met waarneembaarheid en voeg vervolgens intelligentie toe. Zorg ervoor dat er een uitgebreide verzameling van metrische gegevens aanwezig is voordat u ML-modellen implementeert. U kunt geen afwijkingen ontdekken in gegevens die u niet verzamelt.
• Gebruik betrouwbaarheidsdrempels voor herstel. Stel hoge betrouwbaarheidsbalken (95 procent of hoger) in voor destructieve acties zoals failover en lagere drempels (85 procent) voor niet-destructieve acties zoals schalen.
• Zorg voor menselijk toezicht. Automatisch herstel moet altijd acties registreren en mensen op de hoogte stellen. Kritieke acties zoals failover moeten een groter vertrouwen of expliciete menselijke goedkeuring vereisen.
• Train modellen regelmatig opnieuw. Databasewerklastpatronen evolueren met applicatiewijzigingen. Train de modellen voor anomaliedetectie minstens wekelijks opnieuw, of implementeer online leren dat zich voortdurend aanpast.
• Test eerst het herstel in fasering. Elke workflow voor automatisch herstel moet worden gevalideerd in een testomgeving met chaos-engineeringscenario's voordat deze in productie kan worden genomen.
• Combineer meerdere ML-benaderingen. Geen enkel algoritme verwerkt alle soorten afwijkingen. Gebruik ensemblemethoden die Prophet (seizoensgebonden), LSTM (sequentieel) en Isolation Forest (multivariate) combineren voor uitgebreide dekking.
• Veilige LLM-integraties. Wanneer u LLMs gebruikt voor query-analyse, verzend dan nooit daadwerkelijke gegevenswaarden; alleen schema-metagegevens en EXPLAIN-plannen. Gebruik speciale alleen-lezen databasereferenties voor AI-tools.
• Bouw feedbackloops. Volg fout-positieve en fout-negatieve percentages voor de detectie van afwijkingen. Gebruik menselijke feedback over de relevantie van waarschuwingen om de nauwkeurigheid van het model voortdurend te verbeteren.

Conclusie

Het op AI gebaseerde probleemoplossing voor databases vertegenwoordigt een fundamentele verschuiving van reactieve brandbestrijding naar proactieve, intelligente operaties. Door tijdreeksafwijkingsdetectie, door LLM aangestuurde query-optimalisatie, voorspellende waarschuwingen en geautomatiseerd herstel te combineren, kunnen teams detectie in minder dan een minuut realiseren, het aantal valse waarschuwingen dramatisch verminderen en de gemiddelde tijd tot oplossing aanzienlijk verbeteren. De sleutel is stapsgewijs opbouwen: begin met het verzamelen van gegevens en dashboards, voeg afwijkingendetectie toe en schakel vervolgens geleidelijk automatisch herstel in naarmate het vertrouwen in het systeem groeit. Of u nu MySQL, PostgreSQL, MongoDB, Redis of Couchbase beheert, de AI-gestuurde aanpak is universeel toepasbaar, past zich aan de unieke kenmerken van elke engine aan en biedt tegelijkertijd een uniforme observatie-ervaring over uw gehele gegevenslaag.