استكشاف أخطاء قاعدة البيانات المدعومة بالذكاء الاصطناعي وإصلاحها: المراقبة الذكية والكشف عن الحالات الشاذة والمعالجة التلقائية

تولد قواعد بيانات الإنتاج الحديثة ملايين المقاييس في الدقيقة — زمن استجابة الاستعلام، وتنافس القفل، وتأخر النسخ المتماثل، ونسب عدد مرات دخول تجمع المخزن المؤقت، واستنفاد تجمع الاتصال. يغرق التنبيه التقليدي القائم على العتبة الفرق في نتائج إيجابية كاذبة بينما يغيب عن أنماط التدهور الدقيقة التي تسبق حالات الفشل الكارثية. يغير الذكاء الاصطناعي والتعلم الآلي هذه المعادلة بشكل أساسي من خلال تعلم السلوك الطبيعي، واكتشاف الحالات الشاذة قبل أن تتكرر، وتحسين الاستعلامات تلقائيًا، وتنفيذ العلاج دون تدخل بشري. يغطي هذا الدليل النطاق الكامل لاستكشاف أخطاء قواعد البيانات التي تعمل بالذكاء الاصطناعي وإصلاحها عبر MySQL وPostgreSQL وMongoDB وRedis وCouchbase.

خط أنابيب مراقبة قاعدة بيانات الذكاء الاصطناعي

قبل التعمق في تقنيات محددة، من المهم فهم البنية الشاملة لنظام مراقبة قاعدة البيانات الذي يعمل بالذكاء الاصطناعي. يجمع المسار المقاييس الأولية من كل محرك قاعدة بيانات، ويخزنها في قاعدة بيانات متسلسلة زمنية، ويغذيها من خلال نماذج تعلم الآلة لاكتشاف الحالات الشاذة، ويوجه التنبيهات من خلال مدير تنبيه ذكي، ويطلق إجراءات المعالجة التلقائية عند استيفاء حدود الثقة.

الذكاء الاصطناعي/التعلم الآلي لرصد قواعد البيانات وإمكانية ملاحظتها

تعتمد مراقبة قواعد البيانات التقليدية على عتبات ثابتة: يتم التنبيه عندما يتجاوز CPU 80 بالمائة، أو عندما يتجاوز زمن استجابة الاستعلام 500 مللي ثانية، أو عندما يتجاوز عدد الاتصال 200. يفشل هذا النهج بشكل كارثي في ​​بيئات الإنتاج الديناميكية حيث يختلف الوضع الطبيعي حسب الوقت من اليوم، ويوم من الأسبوع، والأنماط الموسمية، وأحداث النشر. تستبدل إمكانية المراقبة المعتمدة على الذكاء الاصطناعي هذه العتبات الصارمة بخطوط أساسية مكتسبة تتكيف بشكل مستمر.

جمع المقاييس الصحيحة

أساس أي نظام مراقبة للذكاء الاصطناعي هو جمع المقاييس الشاملة. يعرض كل محرك قاعدة بيانات مقاييس فريدة مهمة للأداء:

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

الكشف عن الشذوذ مع تحليل السلاسل الزمنية

تتمثل القيمة الأساسية للذكاء الاصطناعي في مراقبة قواعد البيانات في الكشف عن الحالات الشاذة - وتحديد الأنماط غير العادية التي تنحرف عن خطوط الأساس المستفادة. تهيمن ثلاث خوارزميات أساسية على هذا الفضاء: Facebook Prophet للتحليل الموسمي، وشبكات LSTM للأنماط الزمنية المعقدة، وIsolation Forest للكشف عن المتغيرات الخارجية المتعددة.

تنفيذ الكشف عن الشذوذ مع scikit-Learn و Prophet

يوضح تطبيق Python التالي كاشف الشذوذ الجاهز للإنتاج والذي يجمع بين Isolation Forest للكشف عن المتغيرات المتعددة مع Prophet للتنبؤ بالسلاسل الزمنية. يلتقط هذا النهج المزدوج الارتفاعات المفاجئة والانجراف التدريجي.

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

التنبيه التنبؤي مقابل التنبيه القائم على العتبة

يعاني التنبيه التقليدي القائم على العتبة من وضعين متعارضين للفشل. قم بتعيين عتبات ضيقة جدًا وستغرق في نتائج إيجابية كاذبة أثناء تغيرات الحمل العادية. إذا قمت بضبطها بشكل فضفاض جدًا، فسوف تفوتك التدهور الحقيقي حتى يصبح انقطاعًا كاملاً. يعمل التنبيه التنبؤي على حل كلتا المشكلتين من خلال معرفة الشكل "العادي" لكل مقياس في كل نقطة زمنية.

تكامل LLM لاستعلامات قاعدة بيانات اللغة الطبيعية وتحسينها

يمكن لنماذج اللغات الكبيرة مثل GPT-4 وClaude أن تعمل كمساعدين أذكياء لقواعد البيانات، حيث تترجم أسئلة اللغة الطبيعية إلى SQL، وتحلّل خطط EXPLAIN، وتقترح التحسينات. تعمل هذه الإمكانية على تحويل كيفية تفاعل مسؤولي قواعد البيانات والمطورين مع قواعد البيانات - فبدلاً من تحليل خطط التنفيذ يدويًا، يمكنهم وصف المشكلة بلغة إنجليزية بسيطة وتلقي توصيات قابلة للتنفيذ.

بناء مُحسِّن الاستعلام LLM

يقوم تطبيق Python التالي بإنشاء مساعد تحسين الاستعلام الذي يعمل بنظام LLM والذي يقوم بتحليل خطط EXPLAIN واقتراح التحسينات. إنه يتكامل مع API الخاص بـ OpenAI ويتضمن إنشاء سياق مدرك للمخطط.

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

سير عمل المعالجة التلقائية

المعالجة التلقائية هي المكان الذي توفر فيه مراقبة قاعدة البيانات المدعومة بالذكاء الاصطناعي عائد الاستثمار الملموس. بدلاً من تنشيط DBA في الساعة 3 صباحًا لإنهاء الاستعلام الجامح أو قياس النسخ المتماثلة للقراءة، يتعامل النظام مع الأمر تلقائيًا من خلال مسارات التدقيق الكاملة وتسجيل الثقة.

# 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

يقدم MySQL تحديات فريدة تستفيد بشكل كبير من تحليل الذكاء الاصطناعي. تتطلب إدارة تجمع المخزن المؤقت InnoDB، واكتشاف حالة التوقف التام، والتعرف البطيء على نمط الاستعلام، والتنبؤ بتأخر النسخ المتماثل، نماذج ML متخصصة مدربة على مقاييس خاصة بـ MySQL.

تحليل الاستعلام البطيء باستخدام ML

بدلاً من مراجعة سجل الاستعلام البطيء يدويًا، يقوم نموذج تعلم الآلة بتصنيف الاستعلامات حسب تأثير أدائها والسبب الجذري لها. تتضمن الأنماط الشائعة الفهارس المفقودة، والصلات الديكارتية، وجمل WHERE دون المستوى الأمثل مع الوظائف في الأعمدة المفهرسة، وSELECT * في الجداول العريضة.

تحسين تجمع المخزن المؤقت InnoDB

تعد نسبة نجاح تجمع المخزن المؤقت هي المقياس الأكثر أهمية لـ MySQL. تتعلم نماذج الذكاء الاصطناعي العلاقة بين أنماط عبء العمل وفعالية تجمع المخزن المؤقت، وتتنبأ بموعد انخفاض نسبة الدخول والتوصية بتعديلات innodb_buffer_pool_size الاستباقية. يمكن لنموذج LSTM الذي تم تدريبه على مقاييس تجمع المخزن المؤقت التنبؤ بضغط ذاكرة التخزين المؤقت قبل 30 دقيقة من تأثيره على زمن استجابة الاستعلام.

كشف الجمود والوقاية

يقوم الذكاء الاصطناعي بتحليل الرسوم البيانية للتوقف التام في InnoDB لتحديد الأنماط المتكررة. بدلاً من مجرد تسجيل حالات التوقف التام بعد حدوثها، يتعرف النظام على تسلسلات المعاملات التي تؤدي إلى حالات توقف تام ويمكنه إعادة ترتيب العمليات أو ضبط مستويات العزل بشكل استباقي.

استكشاف أخطاء الذكاء الاصطناعي وإصلاحها الخاصة بـ PostgreSQL

تخلق بنية MVCC الخاصة بـ PostgreSQL تحديات فريدة حول انتفاخ الطاولة وجدولة الفراغ وإدارة WAL التي تستفيد من التحليل القائم على الذكاء الاصطناعي.

تحليل الفراغ والكشف عن الانتفاخ

تتتبع نماذج الذكاء الاصطناعي العلاقة بين معدلات المعاملات، وتراكم الصف الميت، وفعالية الفراغ التلقائي. من خلال تعلم معدل نمو الانتفاخ لكل جدول، يتنبأ النظام بالوقت الذي ستصل فيه الجداول إلى مستويات الانتفاخ التي تنطوي على مشكلات، ويطلق عمليات التفريغ المستهدفة قبل أن يتدهور الأداء.

توصيات الفهرس

يكشف تحليل pg_stat_user_indexes وpg_stat_statements معًا عن أنماط استخدام الفهرس. يحدد الذكاء الاصطناعي الفهارس غير المستخدمة التي تستهلك مساحة القرص ويقترح فهارس جديدة بناءً على أنماط الاستعلام، مع الأخذ في الاعتبار تكلفة تضخيم الكتابة للفهارس الإضافية مقابل فائدة أداء القراءة.

تحسين تجمع الاتصال

يتعامل PostgreSQL مع الاتصالات بشكل مختلف عن MySQL، حيث يستهلك كل اتصال ذاكرة أكبر بكثير. تقوم نماذج الذكاء الاصطناعي بتحليل أنماط استخدام تجمع الاتصال عبر PgBouncer لتحديد أحجام التجمع المثالية لملفات تعريف أحمال العمل المختلفة (OLTP مقابل OLAP مقابل المختلط)، مما يمنع انقطاع الاتصال واستنفاد الذاكرة.

استكشاف أخطاء الذكاء الاصطناعي وإصلاحها الخاصة بـ MongoDB

ينشئ نموذج مستند MongoDB والبنية الموزعة مجموعة متميزة من تحديات الأداء التي يمكن للذكاء الاصطناعي معالجتها بفعالية.

اقتراحات الفهرس

يحدد تحليل الذكاء الاصطناعي لمعرف استعلام MongoDB الاستعلامات التي تجري عمليات فحص المجموعة (COLLSCAN) ويوصي بالفهارس المركبة بناءً على مجموعات حقول الاستعلام. يأخذ النموذج في الاعتبار الانتقائية وترتيب الحقل وتحسين الاستعلام المغطى لإنشاء مواصفات الفهرس المثالية.

تحسين المشاركة

بالنسبة للمجموعات المقسمة، يقوم الذكاء الاصطناعي بمراقبة توزيع الأجزاء ومعدلات الترحيل وأنماط توجيه الاستعلام. عندما يكتشف الاستخدام غير المتساوي للجزء (الأجزاء الساخنة)، فإنه يوصي بتغييرات مفتاح الجزء أو استراتيجيات التقسيم المسبق. تتنبأ نماذج ML بمعدلات نمو القطع لموازنة توزيع البيانات بشكل استباقي قبل حدوث تأثيرات الأداء.

تحليل ذاكرة التخزين المؤقت WiredTiger

تكشف أنماط إخلاء ذاكرة التخزين المؤقت لـ WiredTiger عن خصائص عبء العمل. تتعلم نماذج الذكاء الاصطناعي عندما يكون ضغط ذاكرة التخزين المؤقت ناتجًا عن نمو مجموعة العمل مقابل أنماط الوصول غير الفعالة، وتوصي إما بزيادة حجم ذاكرة التخزين المؤقت أو تغييرات على مستوى التطبيق مثل تجميع الاستعلامات.

استكشاف أخطاء الذكاء الاصطناعي وإصلاحها الخاصة بـ Redis

يعمل Redis في ظل قيود مختلفة عن قواعد البيانات المستندة إلى القرص — فالذاكرة هي المورد المهم، وغالبًا ما تكون متطلبات زمن الوصول أقل من مللي ثانية.

تحليل الذاكرة

يتتبع الذكاء الاصطناعي نسب تجزئة الذاكرة وتوزيعات حجم المفتاح وأنماط TTL. عندما يتجاوز التجزئة الحدود السليمة، يحدد النظام ما إذا كان تعديل ACTIVEDEFRAG أو إعادة التشغيل الخاضعة للرقابة هو العلاج الأفضل. تتنبأ نماذج ML بمسارات نمو الذاكرة لمنع عمليات قتل OOM.

كشف النمط الرئيسي وتحديد نقطة الاتصال

باستخدام أخذ عينات MONITOR وتحليل OBJECT FREQ، يحدد الذكاء الاصطناعي المفاتيح الساخنة التي تسبب توزيعًا غير متساوٍ للحمل عبر فتحات المجموعة. بالنسبة لعمليات نشر Redis Cluster، يكتشف النظام اختناقات ترحيل الفتحات ويوصي بتغييرات في تسمية المفاتيح لتحسين توزيع فتحات التجزئة.

تحسين سياسة الإخلاء

تستفيد أحمال العمل المختلفة من سياسات الإخلاء المختلفة (volatile-lru، allkeys-lfu، volatile-ttl). يقوم الذكاء الاصطناعي بتحليل أنماط الوصول للتوصية بسياسة الحد الأقصى الأمثل للذاكرة، مع توقع تأثير معدل الدخول لكل سياسة بناءً على توزيع الوصول الرئيسي الحالي.

استكشاف أخطاء الذكاء الاصطناعي وإصلاحها الخاصة بـ Couchbase

تجمع Couchbase بين إمكانات الاستعلام عن مخزن المستندات وقيمة المفتاح وإمكانات الاستعلام المشابهة لـ SQL (N1QL)، مما يؤدي إلى إنشاء مشهد تحسين فريد من نوعه.

تحسين استعلام N1QL

يقوم الذكاء الاصطناعي بتحليل أنماط استعلام N1QL وشرح المخرجات للتوصية بإنشاء GSI (الفهرس الثانوي العالمي)، واستراتيجيات الفهرس المغطاة، وإعادة كتابة الاستعلام. يتعرف النظام على أنماط N1QL التي تنتج باستمرار خططًا دون المستوى الأمثل ويقترح البدائل بشكل استباقي.

تكامل مستشار الفهرس

يوفر مستشار الفهرس المدمج في Couchbase توصيات، لكن الذكاء الاصطناعي يعززها من خلال النظر في عبء العمل العالمي - موازنة تكاليف إنشاء الفهرس مقابل فوائد الاستعلام عبر أنماط الوصول للتطبيق بالكامل بدلاً من الاستعلامات الفردية بمعزل عن غيرها.

تخطيط إعادة التوازن

عند إضافة العقد أو إزالتها، يجب على Couchbase إعادة توازن البيانات. يتنبأ الذكاء الاصطناعي بمدة إعادة التوازن، وتأثير الموارد، ونوافذ التوقيت الأمثل بناءً على سلوك المجموعة التاريخي. ويمنع هذا عمليات إعادة التوازن من التأثير على حركة الإنتاج خلال ساعات الذروة.

بنية مراقبة الذكاء الاصطناعي لقواعد البيانات المتعددة

تقوم معظم بيئات الإنتاج بتشغيل محركات قواعد بيانات متعددة. يجب أن تقوم منصة موحدة لرصد الذكاء الاصطناعي بتطبيع المقاييس عبر المحركات، وربط الحالات الشاذة عبر طبقة البيانات، وتقديم رؤية متماسكة لفرق العمليات.

إنشاء مساعد قاعدة بيانات AI مخصص باستخدام ChatGPT وClaude

يؤدي دمج LLMs مع البنية الأساسية لقاعدة البيانات الخاصة بك إلى إنشاء مساعد DBA تفاعلي يجيب على أسئلة اللغة الطبيعية، ويشخص المشكلات، وينفذ سير عمل المعالجة. يجمع المساعد بين توليد الاسترجاع المعزز (RAG) والوصول المتري في الوقت الفعلي.

# 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

تشكل مجموعة إمكانية المراقبة العمود الفقري لمراقبة قاعدة بيانات الذكاء الاصطناعي. يقوم Prometheus بجمع المقاييس من مصدري قواعد البيانات، ويقوم Grafana بتصورها، ويقوم خط ML بمعالجة بيانات السلاسل الزمنية للكشف عن الحالات الشاذة.

تكوين بروميثيوس لمراقبة قواعد البيانات المتعددة

# 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

تكوين لوحة معلومات 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}
            ]
        }

تكامل PagerDuty وOpsGenie للتنبيه الذكي

يتجاوز التنبيه الذكي مجرد إشعارات خطاف الويب البسيطة. تتضمن التنبيهات المعززة بالذكاء الاصطناعي تحليل السبب الجذري والسياق التاريخي وسجلات التشغيل المقترحة ونتائج الثقة، مما يمنح المهندسين تحت الطلب السياق الذي يحتاجون إليه لحل المشكلات بشكل أسرع أو التأكيد على أن المعالجة التلقائية قد عالجت المشكلة بالفعل.

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

تحليل السبب الجذري باستخدام الذكاء الاصطناعي

عند اكتشاف حالات شاذة، فإن تحديد السبب الجذري هو الخطوة الأكثر استهلاكًا للوقت في الاستجابة للحوادث. يربط تحليل السبب الجذري المعتمد على الذكاء الاصطناعي إشارات متعددة - الشذوذات المترية، وأنماط السجل، وبيانات التتبع، والتغييرات الأخيرة - لتحديد السبب المحتمل في غضون ثوانٍ بدلاً من ساعات.

يعمل هذا النهج من خلال الحفاظ على رسم بياني معرفي لتبعيات النظام وأنماط الفشل المعروفة. عند حدوث حالة شاذة، يجتاز الذكاء الاصطناعي الرسم البياني لتحديد الأسباب الأولية. على سبيل المثال، إذا ارتفع زمن استجابة الاستعلام على MySQL، فسيقوم النظام بالتحقق مما يلي: هل كانت هناك عملية نشر حديثة؟ هل تغير عدد الاتصال؟ هل هناك تأخر النسخ؟ هل القرص IOPS مشبع؟ هل هناك منافسة على القفل؟ تساهم كل إشارة في الحصول على درجة احتمالية لأسباب جذرية مختلفة.

تخطيط القدرات مع توقعات تعلم الآلة

ينتقل تخطيط السعة المستند إلى التعلم الآلي إلى ما هو أبعد من القياس التفاعلي إلى الإدارة التنبؤية للموارد. من خلال تحليل أنماط النمو التاريخية، والدورات الموسمية، وأحداث الأعمال المخططة، تتنبأ نماذج تعلم الآلة بالوقت الذي ستصل فيه قواعد البيانات إلى حدود الموارد.

يتفوق Prophet في التنبؤ بالسعة لأنه يتعامل مع البيانات المفقودة وتغيرات الاتجاه والأنماط الموسمية محليًا. قم بتدريبه على 90 يومًا من بيانات نمو التخزين اليومي وينتج تنبؤات بفواصل ثقة توضح متى ستحتاج إلى توفير مساحة تخزين إضافية. تعد نماذج LSTM أكثر ملاءمة للتنبؤ بالسعة على المدى القصير - للتنبؤ بالـ 24 ساعة القادمة من استخدام تجمع الاتصال للقياس المسبق قبل ارتفاع حركة المرور في الصباح.

أدوات الذكاء الاصطناعي الخاصة بالسحابة

AWS DevOps Guru لـ RDS

يوفر AWS DevOps Guru اكتشافًا شاذًا مدعومًا بالتعلم الآلي لمثيلات RDS. فهو يقوم تلقائيًا بمراقبة مقاييس CloudWatch وتحديد الحالات الشاذة في الأداء، وربطها بعمليات النشر الأخيرة أو تغييرات التكوين. يتطلب التكامل تمكين DevOps Guru على موارد RDS وتكوين إشعارات SNS.

Azure AI لـ Azure SQL وCosmos DB

يوفر Azure رؤى ذكية لقاعدة بيانات Azure SQL، والتي تستخدم نموذج ML مدمج لاكتشاف تراجعات الأداء وحظر الاستعلامات وحدود الموارد. يتضمن Azure Cosmos DB مستشار الذكاء الاصطناعي المتكامل لتحسين وحدة الطلب واختيار مفتاح القسم.

العمليات السحابية لـ GCP لـ Cloud SQL وFirestore

توفر Google Cloud Operations (المعروفة سابقًا باسم Stackdriver) تنبيهًا ذكيًا لـ Cloud SQL. يتعلم النظام الخطوط الأساسية المترية ويقوم بإنشاء تنبيهات فقط عندما ينحرف السلوك بشكل كبير عن الأنماط المستفادة، مما يقلل بشكل كبير من الإيجابيات الخاطئة مقارنة بالعتبات الثابتة.

أدوات جودة البيانات مفتوحة المصدر

أباتشي غريفين

يوفر Apache Griffin قياس جودة البيانات لأصول البيانات واسعة النطاق. عند دمجه مع مسار مراقبة الذكاء الاصطناعي لديك، فإنه يكتشف الحالات الشاذة في جودة البيانات - القيم المفقودة، وانحراف المخطط، وتغييرات التوزيع - التي غالبًا ما تسبق مشكلات أداء قاعدة البيانات.

توقعات عظيمة

تتيح التوقعات العظيمة إمكانية التحقق من صحة البيانات التصريحية. من خلال تحديد التوقعات لجداول قاعدة البيانات الخاصة بك (عدد الصفوف ضمن النطاق، وقيم الأعمدة ضمن الحدود، والتكامل المرجعي)، يمكنك إنشاء طبقة مراقبة جودة البيانات التي يمكن أن تستهلكها نماذج الذكاء الاصطناعي كإشارات إضافية للكشف عن الحالات الشاذة.

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

مثال كامل لتكامل خطوط الأنابيب

من خلال جمع جميع المكونات معًا، يربط المنسق التالي مجموعة المقاييس واكتشاف الحالات الشاذة وتحليل LLM والتنبيه والمعالجة التلقائية في مسار واحد مستمر يراقب جميع محركات قاعدة البيانات في بيئة الإنتاج الخاصة بك.

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

المقاييس الرئيسية التي يجب تتبعها لنجاح مراقبة قاعدة بيانات الذكاء الاصطناعي

أفضل الممارسات واعتبارات الإنتاج

• ابدأ بالملاحظة، ثم أضف الذكاء. تأكد من وجود مجموعة قياس شاملة قبل نشر نماذج تعلم الآلة. لا يمكنك اكتشاف الحالات الشاذة في البيانات التي لا تجمعها.
• استخدم حدود الثقة للمعالجة. قم بتعيين أشرطة ثقة عالية (95 بالمائة أو أعلى) للإجراءات التدميرية مثل تجاوز الفشل والحدود الأدنى (85 بالمائة) للإجراءات غير التدميرية مثل القياس.
• الحفاظ على الرقابة البشرية. يجب أن تقوم المعالجة التلقائية دائمًا بتسجيل الإجراءات وإخطار البشر. يجب أن تتطلب الإجراءات الحاسمة مثل تجاوز الفشل ثقة عالية أو موافقة بشرية صريحة.
• إعادة تدريب النماذج بانتظام. تتطور أنماط عبء عمل قاعدة البيانات مع تغييرات التطبيق. أعد تدريب نماذج الكشف عن الحالات الشاذة أسبوعيًا على الأقل، أو قم بتنفيذ التعلم عبر الإنترنت الذي يتكيف بشكل مستمر.
• اختبار العلاج في التدريج أولا. يجب التحقق من صحة كل سير عمل للمعالجة التلقائية في بيئة مرحلية تتضمن سيناريوهات هندسة الفوضى قبل تمكينها في الإنتاج.
• الجمع بين أساليب تعلم الآلة المتعددة. لا توجد خوارزمية واحدة تعالج جميع أنواع الحالات الشاذة. استخدم أساليب المجموعة التي تجمع بين النبي (الموسمي)، وLSTM (التسلسلي)، والغابات المعزولة (متعددة المتغيرات) لتغطية شاملة.
• تكاملات LLM الآمنة. عند استخدام LLMs لتحليل الاستعلام، لا ترسل أبدًا قيم البيانات الفعلية - فقط بيانات تعريف المخطط وخطط الشرح. استخدم بيانات اعتماد قاعدة البيانات المخصصة للقراءة فقط لأدوات الذكاء الاصطناعي.
• بناء حلقات ردود الفعل. تتبع المعدلات الإيجابية والسلبية الكاذبة للكشف عن الحالات الشاذة. استخدم التعليقات البشرية بشأن أهمية التنبيه لتحسين دقة النموذج بشكل مستمر.

خاتمة

يمثل استكشاف أخطاء قاعدة البيانات وإصلاحها باستخدام الذكاء الاصطناعي تحولًا أساسيًا من مكافحة الحرائق التفاعلية إلى العمليات الذكية والاستباقية. من خلال الجمع بين الكشف عن الحالات الشاذة في السلاسل الزمنية، وتحسين الاستعلامات المدعومة بـ LLM، والتنبيه التنبؤي، والمعالجة الآلية، يمكن للفرق تحقيق اكتشاف أقل من دقيقة، وتخفيضات كبيرة في التنبيهات الكاذبة، وتحسينات كبيرة في متوسط ​​الوقت اللازم للحل. المفتاح هو البناء بشكل تدريجي — البدء بجمع المقاييس ولوحات المعلومات، ثم طبقة الكشف عن الحالات الشاذة، ثم تمكين المعالجة التلقائية تدريجيًا مع نمو الثقة في النظام. سواء كنت تدير MySQL، أو PostgreSQL، أو MongoDB، أو Redis، أو Couchbase، فإن النهج القائم على الذكاء الاصطناعي ينطبق عالميًا، ويتكيف مع الخصائص الفريدة لكل محرك مع توفير تجربة مراقبة موحدة عبر طبقة البيانات بأكملها.