AIOps: ปัญญาประดิษฐ์
เพื่อการดำเนินงาน IT

global:
  scrape_interval: 15s
  evaluation_interval: 15s

alerting:
  alertmanagers:
    - static_configs:
        - targets: []

rule_files: []

scrape_configs:
  - job_name: 'prometheus'
    static_configs:
      - targets: ['localhost:9090']

  - job_name: 'node-exporter'
    static_configs:
      - targets: ['node-exporter:9100']

  - job_name: 'kubernetes-pods'
    kubernetes_sd_configs:
      - role: pod
    relabel_configs:
      - source_labels: [__meta_kubernetes_pod_annotation_prometheus_io_scrape]
        action: keep
        regex: true

# ตรวจจับ CPU Usage ที่ผิดปกติ
# ใช้ z-score เพื่อหาค่าที่ห่างจากค่าเฉลี่ยมากกว่า 2 standard deviations

avg_over_time(
  rate(container_cpu_usage_seconds_total[5m])[1h]
) > 
avg_over_time(
  rate(container_cpu_usage_seconds_total[5m])[7d]
) + 2 * stddev_over_time(
  rate(container_cpu_usage_seconds_total[5m])[7d]
)

# ตรวจจับ Memory Leak
# หาก Memory ใช้เพิ่มขึ้นเรื่อยๆ อย่างต่อเนื่อง

deriv(
  container_memory_usage_bytes[1h]
) > 0
and
  predict_linear(
    container_memory_usage_bytes[1h],
    3600
  ) > container_spec_memory_limit_bytes * 0.9

# ตรวจจับ Response Time ที่สูงผิดปกติ
histogram_quantile(0.95, 
  sum by (le) (
    rate(http_request_duration_seconds_bucket[5m])
  )
) > 
avg_over_time(
  histogram_quantile(0.95, 
    sum by (le) (
      rate(http_request_duration_seconds_bucket[5m])
    )
  )[7d]
) * 2

workflow:
  id: aiops-alert-correlation
  name: AIOps Alert Correlation
  description: รวมและกรอง Alert โดยอัตโนมัติ
  
  triggers:
    - type: alert
      filters:
        - key: severity
          value: critical|warning

  steps:
    - name: check_duplicate
      type: dedupe
      with:
        time_window: 300  # 5 นาที
        group_by:
          - service
          - alert_name

    - name: correlate_events
      type: correlate
      with:
        algorithm: similarity
        threshold: 0.8
        lookback: 900  # 15 นาที

    - name: enrich_with_context
      type: enrich
      with:
        add_silence_info: true
        add_runbook_link: true
        add_service_dependency: true

    - name: calculate_priority
      type: ml_score
      with:
        model: priority_classifier
        features:
          - severity
          - frequency
          - business_impact

  actions:
    - name: notify_slack
      condition: steps.calculate_priority.score > 0.7
      type: slack
      with:
        channel: '#ops-alerts'
        message: |
          🚨 Alert: {{ alert.name }}
          Service: {{ alert.service }}
          Priority: {{ steps.calculate_priority.score }}
          Runbook: {{ steps.enrich_with_context.runbook }}

    - name: auto_remediate
      condition: steps.calculate_priority.score > 0.9
      type: webhook
      with:
        url: http://auto-remediation:8080/trigger
        payload:
          alert_id: {{ alert.id }}
          service: {{ alert.service }}

import numpy as np
import pandas as pd
from sklearn.ensemble import IsolationForest
from prometheus_api_client import PrometheusConnect
import joblib
from datetime import datetime, timedelta

class AIOpsAnomalyDetector:
    """
    AIOps Anomaly Detector ใช้ Isolation Forest
    สำหรับตรวจจับความผิดปกติใน Metrics
    """
    
    def __init__(self, prometheus_url: str, contamination: float = 0.05):
        """
        Args:
            prometheus_url: URL ของ Prometheus server
            contamination: อัตราส่วนของ outliers ที่คาดว่าจะพบ (default: 5%)
        """
        self.prom = PrometheusConnect(url=prometheus_url, disable_ssl=True)
        self.model = IsolationForest(
            n_estimators=100,
            contamination=contamination,
            random_state=42,
            n_jobs=-1
        )
        self.feature_columns = None
        
    def fetch_metrics(self, 
                      metric_name: str,
                      hours_back: int = 24) -> pd.DataFrame:
        """
        ดึง Metrics จาก Prometheus
        
        Args:
            metric_name: ชื่อ metric (เช่น 'node_cpu_seconds_total')
            hours_back: จำนวนชั่วโมงที่ย้อนหลัง
        """
        end_time = datetime.now()
        start_time = end_time - timedelta(hours=hours_back)
        
        metric_data = self.prom.custom_query_range(
            query=metric_name,
            start_time=start_time,
            end_time=end_time,
            step="1m"
        )
        
        # แปลงเป็น DataFrame
        records = []
        for item in metric_data:
            labels = item['metric']
            for value in item['values']:
                records.append({
                    'timestamp': pd.to_datetime(value[0], unit='s'),
                    'value': float(value[1]),
                    **labels
                })
        
        return pd.DataFrame(records)
    
    def prepare_features(self, df: pd.DataFrame) -> np.ndarray:
        """
        เตรียม Features สำหรับ ML Model
        """
        # สร้าง lag features
        df = df.copy()
        df['value_lag_1'] = df['value'].shift(1)
        df['value_lag_5'] = df['value'].shift(5)
        df['value_lag_15'] = df['value'].shift(15)
        
        # สร้าง rolling statistics
        df['rolling_mean_5'] = df['value'].rolling(window=5).mean()
        df['rolling_std_5'] = df['value'].rolling(window=5).std()
        df['rolling_mean_15'] = df['value'].rolling(window=15).mean()
        
        # สร้าง rate of change
        df['rate_of_change'] = df['value'].pct_change()
        
        # Drop NaN
        df = df.dropna()
        
        self.feature_columns = [
            'value', 'value_lag_1', 'value_lag_5', 'value_lag_15',
            'rolling_mean_5', 'rolling_std_5', 'rolling_mean_15',
            'rate_of_change'
        ]
        
        return df[self.feature_columns].values
    
    def train(self, X: np.ndarray):
        """
        Train Isolation Forest Model
        """
        print(f"Training Isolation Forest with {len(X)} samples...")
        self.model.fit(X)
        print("Training complete!")
        
    def predict(self, X: np.ndarray) -> np.ndarray:
        """
        ทำนายว่าเป็น Anomaly หรือไม่
        
        Returns:
            -1 = Anomaly, 1 = Normal
        """
        return self.model.predict(X)
    
    def get_anomaly_score(self, X: np.ndarray) -> np.ndarray:
        """
        คำนวณ Anomaly Score (ยิ่งต่ำยิ่งผิดปกติ)
        """
        return self.model.score_samples(X)
    
    def save_model(self, filepath: str):
        """บันทึก Model"""
        joblib.dump({
            'model': self.model,
            'feature_columns': self.feature_columns
        }, filepath)
        print(f"Model saved to {filepath}")
    
    def load_model(self, filepath: str):
        """โหลด Model"""
        data = joblib.load(filepath)
        self.model = data['model']
        self.feature_columns = data['feature_columns']
        print(f"Model loaded from {filepath}")


# ตัวอย่างการใช้งาน
if __name__ == "__main__":
    # สร้าง Detector
    detector = AIOpsAnomalyDetector(
        prometheus_url="http://localhost:9090",
        contamination=0.05  # คาดว่าจะมี 5% outliers
    )
    
    # ดึงข้อมูล CPU usage
    df = detector.fetch_metrics(
        metric_name='rate(node_cpu_seconds_total[5m])',
        hours_back=168  # 7 วัน
    )
    
    # เตรียม Features
    X = detector.prepare_features(df)
    
    # Train Model
    detector.train(X)
    
    # บันทึก Model
    detector.save_model("anomaly_model.pkl")
    
    # ทำนายข้อมูลใหม่
    predictions = detector.predict(X[-100:])  # 100 ข้อมูลล่าสุด
    scores = detector.get_anomaly_score(X[-100:])
    
    # หา Anomalies
    anomaly_indices = np.where(predictions == -1)[0]
    print(f"Found {len(anomaly_indices)} anomalies in last 100 samples")
    
    # Alert ถ้าพบ Anomaly
    if len(anomaly_indices) > 5:
        print("⚠️ ALERT: High number of anomalies detected!")
        print("Sending notification to Slack...")

apiVersion: machinelearning.seldon.io/v1
kind: SeldonDeployment
metadata:
  name: aiops-anomaly-detector
  namespace: aiops
spec:
  predictors:
  - name: default
    replicas: 2
    graph:
      name: anomaly-model
      implementation: SKLEARN_SERVER
      modelUri: s3://ml-models/anomaly-detector
      envSecretRefName: s3-credentials
      parameters:
        - name: method
          type: STRING
          value: predict_proba
    componentSpecs:
    - spec:
        containers:
        - name: anomaly-model
          resources:
            requests:
              memory: "1Gi"
              cpu: "500m"
            limits:
              memory: "2Gi"
              cpu: "1000m"
---
# Service สำหรับเรียก Model
apiVersion: v1
kind: Service
metadata:
  name: anomaly-detector
  namespace: aiops
spec:
  selector:
    app: aiops-anomaly-detector
  ports:
  - port: 8000
    targetPort: 8000
  type: ClusterIP

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.slack.operators.slack_webhook import SlackWebhookOperator
from datetime import datetime, timedelta
import requests
import pandas as pd

default_args = {
    'owner': 'aiops',
    'depends_on_past': False,
    'start_date': datetime(2025, 1, 1),
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'aiops_anomaly_detection_pipeline',
    default_args=default_args,
    schedule_interval='*/5 * * * *',  # ทุก 5 นาที
    catchup=False,
    tags=['aiops', 'monitoring', 'ml']
) as dag:

    def fetch_prometheus_metrics(**context):
        """ดึง Metrics จาก Prometheus"""
        prometheus_url = "http://prometheus:9090"
        query = 'rate(container_cpu_usage_seconds_total[5m])'
        
        response = requests.get(
            f"{prometheus_url}/api/v1/query",
            params={'query': query}
        )
        
        data = response.json()['data']['result']
        metrics_df = pd.DataFrame([
            {
                'metric': item['metric'],
                'value': float(item['value'][1]),
                'timestamp': item['value'][0]
            }
            for item in data
        ])
        
        context['ti'].xcom_push(key='metrics', value=metrics_df.to_dict())
        return metrics_df

    def detect_anomalies(**context):
        """เรียก Seldon Model ตรวจจับ Anomaly"""
        metrics = context['ti'].xcom_pull(key='metrics', task_ids='fetch_metrics')
        
        # เรียก Seldon Model
        seldon_url = "http://anomaly-detector:8000/api/v1.0/predictions"
        
        payload = {
            "data": {
                "ndarray": [[m['value']] for m in metrics.values()]
            }
        }
        
        response = requests.post(seldon_url, json=payload)
        predictions = response.json()['data']['ndarray']
        
        # หา anomalies (ค่า -1)
        anomalies = [
            list(metrics.keys())[i] 
            for i, pred in enumerate(predictions) 
            if pred[0] == -1
        ]
        
        context['ti'].xcom_push(key='anomalies', value=anomalies)
        return anomalies

    def check_and_alert(**context):
        """ตรวจสอบและส่ง Alert"""
        anomalies = context['ti'].xcom_pull(key='anomalies', task_ids='detect_anomalies')
        
        if len(anomalies) > 0:
            # ส่งต่อไป Keep สำหรับ Correlation
            keep_url = "http://keep:8080/api/alerts"
            alert_payload = {
                "source": "airflow-aiops",
                "severity": "warning",
                "message": f"Detected {len(anomalies)} anomalies",
                "details": anomalies
            }
            requests.post(keep_url, json=alert_payload)
        
        return len(anomalies)

    # Task Definitions
    fetch_metrics = PythonOperator(
        task_id='fetch_metrics',
        python_callable=fetch_prometheus_metrics,
    )

    detect_anomalies = PythonOperator(
        task_id='detect_anomalies',
        python_callable=detect_anomalies,
    )

    check_alert = PythonOperator(
        task_id='check_and_alert',
        python_callable=check_and_alert,
    )

    # Pipeline
    fetch_metrics >> detect_anomalies >> check_alert