#!/usr/bin/env python3
"""Cloud cost data collector for multiple providers."""

import json
import requests
from datetime import datetime, timedelta
from dataclasses import dataclass, asdict
from typing import Optional


@dataclass
class CostRecord:
    timestamp: str
    provider: str
    service: str
    amount: float
    currency: str = "USD"
    metadata: dict = None

    def __post_init__(self):
        if self.metadata is None:
            self.metadata = {}


class CostCollector:
    """Unified cost data collector supporting multiple cloud providers."""

    def __init__(self):
        self.records: list[CostRecord] = []

    def collect_digitalocean(self, token: str, days: int = 7) -> list[CostRecord]:
        """Collect usage data from DigitalOcean API."""
        records = []
        start = (datetime.utcnow() - timedelta(days=days)).strftime("%Y-%m-%dT00:00:00Z")
        end = datetime.utcnow().strftime("%Y-%m-%dT00:00:00Z")

        resp = requests.get(
            "https://api.digitalocean.com/v2/billing/history",
            headers={"Authorization": f"Bearer {token}"},
            params={"start_date": start, "end_date": end},
        )
        resp.raise_for_status()

        for item in resp.json().get("billing_history", []):
            record = CostRecord(
                timestamp=item["timestamp"],
                provider="digitalocean",
                service=item.get("description", "unknown"),
                amount=abs(float(item.get("amount", 0))),
                currency=item.get("currency", "USD"),
                metadata={"type": item.get("type", "")},
            )
            records.append(record)

        return records

    def collect_linode(self, token: str, days: int = 7) -> list[CostRecord]:
        """Collect usage data from Linode API."""
        records = []
        start = (datetime.utcnow() - timedelta(days=days)).strftime("%Y-%m-%dT00:00:00Z")
        end = datetime.utcnow().strftime("%Y-%m-%dT00:00:00Z")

        resp = requests.get(
            "https://api.linode.com/v4/account/billing",
            headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
        )
        resp.raise_for_status()

        for item in resp.json().get("data", []):
            record = CostRecord(
                timestamp=item.get("dtstart", ""),
                provider="linode",
                service=item.get("label", "unknown"),
                amount=abs(float(item.get("total", 0))),
                currency="USD",
            )
            records.append(record)

        return records

    def save_to_database(self, records: list[CostRecord]):
        """Save collected records to local SQLite database."""
        import sqlite3

        conn = sqlite3.connect("/opt/cost-monitor/costs.db")
        conn.execute("""
            CREATE TABLE IF NOT EXISTS cost_records (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp TEXT NOT NULL,
                provider TEXT NOT NULL,
                service TEXT,
                amount REAL,
                currency TEXT DEFAULT 'USD',
                metadata TEXT
            )
        """)

        for r in records:
            conn.execute(
                "INSERT INTO cost_records (timestamp, provider, service, amount, currency, metadata) VALUES (?, ?, ?, ?, ?, ?)",
                (r.timestamp, r.provider, r.service, r.amount, r.currency, json.dumps(r.metadata)),
            )

        conn.commit()
        conn.close()


if __name__ == "__main__":
    collector = CostCollector()

    # Collect from DigitalOcean
    do_records = collector.collect_digitalocean(token="your_do_token")
    collector.save_to_database(do_records)
    print(f"Collected {len(do_records)} DigitalOcean records")

    # Collect from Linode
    linode_records = collector.collect_linode(token="your_linode_token")
    collector.save_to_database(linode_records)
    print(f"Collected {len(linode_records)} Linode records")

# prometheus.yml - 自定义成本相关指标
global:
  scrape_interval: 60s

scrape_configs:
  - job_name: 'vps_resources'
    static_configs:
      - targets: ['localhost:9100']  # node_exporter

  - job_name: 'container_costs'
    metrics_path: '/metrics'
    static_configs:
      - targets: ['cost-exporter:8080']  # 自定义导出器

#!/usr/bin/env python3
"""Prometheus exporter that converts resource usage to cost estimates."""

from prometheus_client import start_http_server, Gauge
import time

# CPU cost gauge (USD/hour based on usage)
cpu_cost_gauge = Gauge(
    'vps_cpu_cost_usd_per_hour',
    'Estimated CPU cost per hour based on current usage'
)

# Memory cost gauge
memory_cost_gauge = Gauge(
    'vps_memory_cost_usd_per_gb_hour',
    'Memory cost per GB per hour'
)

# Network cost gauge
network_cost_gauge = Gauge(
    'vps_network_cost_usd_per_gb',
    'Network transfer cost per GB'
)

# Disk cost gauge
disk_cost_gauge = Gauge(
    'vps_disk_cost_usd_per_gb_month',
    'Disk storage cost per GB per month'
)

# Pricing configuration (adjust per provider)
PRICING = {
    'cpu_per_core_hour': 0.025,      # $0.025 per vCPU hour
    'memory_per_gb_hour': 0.0035,    # $0.0035 per GB memory hour
    'network_per_gb': 0.01,           # $0.01 per GB outbound
    'disk_per_gb_month': 0.10,        # $0.10 per GB/month
}


def get_resource_usage():
    """Simulate fetching resource usage from system or API."""
    # In production, read from /proc/stat, cgroups, or cloud API
    return {
        'cpu_cores_used': 2.5,
        'memory_gb_used': 4.2,
        'network_gb_out': 15.3,
        'disk_gb_used': 50.0,
    }


def update_cost_metrics():
    """Calculate and expose cost metrics."""
    usage = get_resource_usage()

    cpu_cost = usage['cpu_cores_used'] * PRICING['cpu_per_core_hour']
    mem_cost = usage['memory_gb_used'] * PRICING['memory_per_gb_hour']
    net_cost = usage['network_gb_out'] * PRICING['network_per_gb']
    disk_cost = usage['disk_gb_used'] * PRICING['disk_per_gb_month'] / 730  # convert to hourly

    cpu_cost_gauge.set(cpu_cost)
    memory_cost_gauge.set(mem_cost)
    network_cost_gauge.set(net_cost)
    disk_cost_gauge.set(disk_cost)

    total_hourly = cpu_cost + mem_cost + net_cost + disk_cost
    print(f"Estimated hourly cost: ${total_hourly:.4f}")


if __name__ == "__main__":
    start_http_server(8080)
    print("Cost exporter running on :8080")
    while True:
        update_cost_metrics()
        time.sleep(60)

#!/usr/bin/env python3
"""Statistical anomaly detection engine for VPS cost monitoring."""

import sqlite3
import numpy as np
from datetime import datetime, timedelta
from dataclasses import dataclass
from enum import Enum


class Severity(Enum):
    INFO = "info"
    WARNING = "warning"
    CRITICAL = "critical"


@dataclass
class AnomalyAlert:
    timestamp: str
    severity: str
    metric: str
    current_value: float
    expected_value: float
    deviation_pct: float
    description: str


class CostAnomalyDetector:
    """Multi-method statistical anomaly detector for cloud costs."""

    def __init__(self, db_path: str = "/opt/cost-monitor/costs.db"):
        self.db_path = db_path

    def _get_daily_totals(self, days: int = 30) -> list[float]:
        """Get daily cost totals from the database."""
        conn = sqlite3.connect(self.db_path)
        query = """
            SELECT DATE(timestamp) as day, SUM(amount) as total
            FROM cost_records
            WHERE timestamp >= datetime('now', '-' || ? || ' days')
            GROUP BY day
            ORDER BY day
        """
        rows = conn.execute(query, (days,)).fetchall()
        conn.close()
        return [r[1] for r in rows] if rows else []

    def detect_zscore_anomalies(self, threshold: float = 2.0) -> list[AnomalyAlert]:
        """Detect anomalies using Z-score method."""
        alerts = []
        daily_totals = self._get_daily_totals()

        if len(daily_totals) < 7:
            return alerts  # Need minimum data points

        values = np.array(daily_totals[-7:])  # Last 7 days
        mean = np.mean(values)
        std = np.std(values)

        if std == 0:
            return alerts

        for i, val in enumerate(values):
            z_score = abs(val - mean) / std
            if z_score > threshold:
                deviation_pct = ((val - mean) / mean) * 100 if mean != 0 else 0

                if deviation_pct > 100:
                    severity = Severity.CRITICAL.value
                elif deviation_pct > 50:
                    severity = Severity.WARNING.value
                else:
                    severity = Severity.INFO.value

                alert = AnomalyAlert(
                    timestamp=datetime.utcnow().isoformat(),
                    severity=severity,
                    metric="daily_cost",
                    current_value=float(val),
                    expected_value=float(mean),
                    deviation_pct=float(deviation_pct),
                    description=f"Daily cost ${val:.2f} deviates {deviation_pct:.1f}% from 7-day average (${mean:.2f})",
                )
                alerts.append(alert)

        return alerts

    def detect_moving_average_deviation(self, short_window: int = 3,
                                        long_window: int = 14) -> list[AnomalyAlert]:
        """Detect anomalies using moving average deviation."""
        alerts = []
        daily_totals = self._get_daily_totals()

        if len(daily_totals) < long_window:
            return alerts

        recent = np.array(daily_totals[-short_window:])
        baseline = np.array(daily_totals[-long_window:-short_window]) if long_window > short_window else baseline

        recent_avg = np.mean(recent)
        baseline_avg = np.mean(baseline)

        if baseline_avg == 0:
            return alerts

        deviation_pct = ((recent_avg - baseline_avg) / baseline_avg) * 100

        if abs(deviation_pct) > 30:  # 30% threshold
            severity = Severity.CRITICAL.value if abs(deviation_pct) > 100 else Severity.WARNING.value

            alert = AnomalyAlert(
                timestamp=datetime.utcnow().isoformat(),
                severity=severity,
                metric="moving_average_deviation",
                current_value=float(recent_avg),
                expected_value=float(baseline_avg),
                deviation_pct=float(deviation_pct),
                description=f"Recent {short_window}-day avg (${recent_avg:.2f}) deviates {deviation_pct:.1f}% from {long_window}-day baseline (${baseline_avg:.2f})",
            )
            alerts.append(alert)

        return alerts

    def detect_seasonal_anomalies(self) -> list[AnomalyAlert]:
        """Detect anomalies based on day-of-week seasonal patterns."""
        alerts = []
        conn = sqlite3.connect(self.db_path)

        # Get costs grouped by day of week
        query = """
            SELECT CAST(strftime('%w', timestamp) AS INTEGER) as dow,
                   AVG(amount) as avg_cost, COUNT(*) as cnt
            FROM cost_records
            WHERE timestamp >= datetime('now', '-90 days')
            GROUP BY dow
        """
        rows = conn.execute(query).fetchall()
        conn.close()

        if not rows:
            return alerts

        # Build weekday/weekend averages
        weekdays = [r for r in rows if r[0] < 5]
        weekends = [r for r in rows if r[0] >= 5]

        if not weekdays or not weekends:
            return alerts

        weekday_avg = np.mean([r[1] for r in weekdays])
        weekend_avg = np.mean([r[1] for r in weekends])

        # Check if today's pattern breaks the seasonal norm
        today_dow = datetime.now().weekday()
        today_cost_query = """
            SELECT COALESCE(SUM(amount), 0)
            FROM cost_records
            WHERE strftime('%w', timestamp) = ?
              AND DATE(timestamp) = DATE('now')
        """
        conn = sqlite3.connect(self.db_path)
        today_cost = conn.execute(today_cost_query, (str(today_dow),)).fetchone()[0]
        conn.close()

        expected_avg = weekday_avg if today_dow < 5 else weekend_avg

        if expected_avg > 0:
            deviation = ((today_cost - expected_avg) / expected_avg) * 100
            if abs(deviation) > 50:
                severity = Severity.WARNING.value if abs(deviation) > 100 else Severity.INFO.value
                day_name = "工作日" if today_dow < 5 else "周末"
                alert = AnomalyAlert(
                    timestamp=datetime.utcnow().isoformat(),
                    severity=severity,
                    metric="seasonal_pattern",
                    current_value=float(today_cost),
                    expected_value=float(expected_avg),
                    deviation_pct=float(deviation),
                    description=f"{day_name} cost ${today_cost:.2f} deviates {deviation:.1f}% from typical {day_name} average (${expected_avg:.2f})",
                )
                alerts.append(alert)

        return alerts

    def run_full_detection(self) -> list[AnomalyAlert]:
        """Run all detection methods and return combined alerts."""
        all_alerts = []
        all_alerts.extend(self.detect_zscore_anomalies())
        all_alerts.extend(self.detect_moving_average_deviation())
        all_alerts.extend(self.detect_seasonal_anomalies())

        # Deduplicate by severity + metric
        seen = set()
        unique_alerts = []
        for alert in all_alerts:
            key = (alert.severity, alert.metric)
            if key not in seen:
                seen.add(key)
                unique_alerts.append(alert)

        return unique_alerts


if __name__ == "__main__":
    detector = CostAnomalyDetector()
    alerts = detector.run_full_detection()

    if alerts:
        print(f"\n🚨 {len(alerts)} anomaly alert(s) detected:\n")
        for alert in alerts:
            emoji = {"critical": "🔴", "warning": "🟡", "info": "🔵"}.get(alert.severity, "⚪")
            print(f"  {emoji} [{alert.severity.upper()}] {alert.description}")
    else:
        print("\n✅ All cost metrics within normal range.")

#!/usr/bin/env python3
"""LLM-powered cost anomaly analyzer for VPS infrastructure."""

import json
import os
import sqlite3
from datetime import datetime
from typing import Optional


class CostAnomalyAnalyzer:
    """Uses LLM to analyze cost anomalies and generate actionable reports."""

    def __init__(self, api_key: Optional[str] = None, model: str = "gpt-4o-mini"):
        self.api_key = api_key or os.environ.get("OPENAI_API_KEY")
        self.model = model
        self.system_prompt = """你是一个专业的 FinOps 分析师和云成本优化专家。
你的职责是分析云基础设施的成本异常，找出根本原因，并提供具体的优化建议。

分析原则：
1. 关注数据中的模式和趋势，而非孤立的数据点
2. 考虑多种可能的根因（资源浪费、配置错误、安全事件、业务增长）
3. 提供具体、可执行的优化建议
4. 区分紧急问题和长期优化机会
5. 用简洁清晰的语言表达，避免过度技术化"""

    def _build_analysis_context(self, alerts) -> dict:
        """Build rich context for LLM analysis from alerts and metrics."""
        conn = sqlite3.connect("/opt/cost-monitor/costs.db")

        # Recent cost history
        cost_history = conn.execute("""
            SELECT DATE(timestamp) as day, SUM(amount) as total, COUNT(*) as tx_count
            FROM cost_records
            WHERE timestamp >= datetime('now', '-30 days')
            GROUP BY day
            ORDER BY day
        """).fetchall()

        # Provider breakdown
        provider_breakdown = conn.execute("""
            SELECT provider, SUM(amount) as total, COUNT(*) as tx_count
            FROM cost_records
            WHERE timestamp >= datetime('now', '-7 days')
            GROUP BY provider
        """).fetchall()

        # Service breakdown
        service_breakdown = conn.execute("""
            SELECT service, SUM(amount) as total
            FROM cost_records
            WHERE timestamp >= datetime('now', '-7 days')
            GROUP BY service
            ORDER BY total DESC
            LIMIT 10
        """).fetchall()

        conn.close()

        return {
            "alerts": [
                {
                    "severity": a.severity,
                    "metric": a.metric,
                    "current_value": a.current_value,
                    "expected_value": a.expected_value,
                    "deviation_pct": a.deviation_pct,
                    "description": a.description,
                }
                for a in alerts
            ],
            "cost_history_30d": [
                {"day": r[0], "total": r[1], "transactions": r[2]}
                for r in cost_history
            ],
            "provider_breakdown": [
                {"provider": r[0], "total": r[1], "transactions": r[2]}
                for r in provider_breakdown
            ],
            "top_services": [
                {"service": r[0], "total": r[1]}
                for r in service_breakdown
            ],
            "analysis_time": datetime.utcnow().isoformat(),
        }

    def analyze(self, alerts) -> dict:
        """Run LLM analysis on detected anomalies."""
        if not alerts:
            return {
                "status": "normal",
                "summary": "All cost metrics are within normal parameters.",
                "recommendations": [],
                "risk_level": "low",
            }

        context = self._build_analysis_context(alerts)

        # Build the user prompt for the LLM
        user_prompt = f"""请分析以下云成本异常数据并生成分析报告：

## 检测到的异常
{json.dumps(context['alerts'], indent=2, ensure_ascii=False)}

## 过去30天成本趋势
{json.dumps(context['cost_history_30d'][-7:], indent=2, ensure_ascii=False)}

## 提供商费用分布
{json.dumps(context['provider_breakdown'], indent=2, ensure_ascii=False)}

## Top 消费服务
{json.dumps(context['top_services'], indent=2, unused_ascii=False)}

## 分析要求
1. 识别最可能的根因（按可能性排序）
2. 评估严重程度和影响范围
3. 提供立即采取的行动建议
4. 给出长期优化建议
5. 预估潜在节省金额

请用中文回答，结构清晰，重点突出。"""

        # Call LLM API
        analysis_result = self._call_llm(user_prompt)

        return analysis_result

    def _call_llm(self, user_prompt: str) -> dict:
        """Call the LLM API and parse the response."""
        try:
            from openai import OpenAI
            client = OpenAI(api_key=self.api_key)

            response = client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": self.system_prompt},
                    {"role": "user", "content": user_prompt},
                ],
                temperature=0.3,
                max_tokens=2000,
            )

            content = response.choices[0].message.content

            # Parse structured output
            return {
                "status": "anomaly_detected",
                "llm_report": content,
                "model": self.model,
                "generated_at": datetime.utcnow().isoformat(),
            }

        except ImportError:
            # Fallback: generate a template analysis without LLM
            return self._generate_template_analysis(alerts)

    def _generate_template_analysis(self, alerts) -> dict:
        """Generate a template analysis when LLM is unavailable."""
        critical_count = sum(1 for a in alerts if a.severity == "critical")
        warning_count = sum(1 for a in alerts if a.severity == "warning")

        return {
            "status": "anomaly_detected",
            "llm_report": f"检测到 {len(alerts)} 个异常（{critical_count} 个严重，{warning_count} 个警告）。LLM 分析服务不可用，已回退到模板分析。建议检查最近的资源变更和计费记录。",
            "model": "template-fallback",
            "generated_at": datetime.utcnow().isoformat(),
        }


if __name__ == "__main__":
    from cost_anomaly_detector import CostAnomalyDetector, AnomalyAlert

    # Run detection
    detector = CostAnomalyDetector()
    alerts = detector.run_full_detection()

    # Analyze with LLM
    analyzer = CostAnomalyAnalyzer()
    result = analyzer.analyze(alerts)

    print(f"\n{'='*60}")
    print(f"📊 Cost Anomaly Analysis Report")
    print(f"{'='*60}")
    print(f"Status: {result['status']}")
    print(f"Model: {result['model']}")
    print(f"\nReport:\n{result['llm_report']}")

# 使用 Ollama 部署轻量级模型
curl -fsSL https://ollama.com/install.sh | sh
ollama pull llama3.2:3b

# 或者使用 vLLM 部署更强大的模型
pip install vllm
vllm serve meta-llama/Llama-3.2-3B-Instruct --port 8000

def _call_local_llm(self, user_prompt: str) -> dict:
    """Call a locally hosted LLM via Ollama API."""
    import requests

    response = requests.post(
        "http://localhost:11434/api/generate",
        json={
            "model": "llama3.2:3b",
            "prompt": user_prompt,
            "stream": False,
            "options": {
                "temperature": 0.3,
                "num_predict": 2000,
            },
        },
        timeout=120,
    )
    response.raise_for_status()
    return response.json()

#!/usr/bin/env python3
"""Alert dispatcher for cost anomaly notifications."""

import os
import requests
from datetime import datetime


class AlertDispatcher:
    """Dispatches alerts through multiple channels."""

    def __init__(self):
        self.telegram_token = os.environ.get("TELEGRAM_BOT_TOKEN")
        self.telegram_chat_id = os.environ.get("TELEGRAM_CHAT_ID")
        self.webhook_url = os.environ.get("ALERT_WEBHOOK_URL")

    def send_telegram(self, message: str, severity: str = "info"):
        """Send alert via Telegram Bot."""
        if not self.telegram_token or not self.telegram_chat_id:
            return

        emoji = {"critical": "🔴", "warning": "🟡", "info": "🔵"}.get(severity, "ℹ️")
        formatted = f"{emoji} *[{severity.upper()}]*\n\n{message}"

        url = f"https://api.telegram.org/bot{self.telegram_token}/sendMessage"
        requests.post(url, json={
            "chat_id": self.telegram_chat_id,
            "text": formatted,
            "parse_mode": "Markdown",
        })

    def send_webhook(self, payload: dict):
        """Send alert via webhook (Slack, Discord, etc.)."""
        if not self.webhook_url:
            return

        requests.post(self.webhook_url, json=payload)

    def dispatch(self, result: dict):
        """Dispatch all alerts from analysis result."""
        if result['status'] == 'normal':
            return

        report = result.get('llm_report', 'Unknown issue detected.')
        severity = "critical" if "严重" in report or "CRITICAL" in report else "warning"

        # Send Telegram notification
        self.send_telegram(report[:1000], severity)  # Telegram has 4096 char limit

        # Send webhook if configured
        self.send_webhook({
            "text": f"Cost Anomaly Alert: {report[:200]}",
            "level": severity,
            "timestamp": datetime.utcnow().isoformat(),
            "details": report,
        })


if __name__ == "__main__":
    dispatcher = AlertDispatcher()
    print("Alert dispatcher ready. Configure TELEGRAM_BOT_TOKEN and TELEGRAM_CHAT_ID.")

#!/usr/bin/env python3
"""Automated remediation actions for known cost anomaly patterns."""

import subprocess
import json


class CostRemediator:
    """Executes automated remediation actions for common cost anomalies."""

    ACTIONS = {
        "high_cpu_cost": {
            "description": "Kill runaway processes consuming excessive CPU",
            "command": "ps aux --sort=-%cpu | head -5",
        },
        "high_network_cost": {
            "description": "Identify top bandwidth consumers",
            "command": "nethogs -t 5 || echo 'Install nethogs for detailed analysis'",
        },
        "storage_growth_spike": {
            "description": "Find recently modified large files",
            "command": "find / -type f -mtime -1 -size +100M 2>/dev/null | head -20",
        },
    }

    def execute(self, anomaly_type: str, dry_run: bool = True) -> dict:
        """Execute remediation action for a given anomaly type."""
        action = self.ACTIONS.get(anomaly_type)
        if not action:
            return {"status": "skipped", "reason": f"No action defined for {anomaly_type}"}

        if dry_run:
            return {
                "status": "dry_run",
                "action": action["description"],
                "command": action["command"],
                "note": "Dry run mode - no action executed",
            }

        try:
            result = subprocess.run(
                action["command"],
                shell=True,
                capture_output=True,
                text=True,
                timeout=30,
            )
            return {
                "status": "executed",
                "action": action["description"],
                "output": result.stdout[:500],
                "error": result.stderr[:500] if result.returncode != 0 else None,
            }
        except subprocess.TimeoutExpired:
            return {"status": "timeout", "action": action["description"]}
        except Exception as e:
            return {"status": "error", "action": action["description"], "error": str(e)}


if __name__ == "__main__":
    remediator = CostRemediator()

    # Dry run all actions
    for anomaly_type in remediator.ACTIONS:
        result = remediator.execute(anomaly_type, dry_run=True)
        print(f"\n🔧 {anomaly_type}:")
        print(f"   Status: {result['status']}")
        print(f"   Description: {result.get('action', 'N/A')}")

# 创建虚拟环境
python3 -m venv /opt/cost-monitor/venv
source /opt/cost-monitor/venv/bin/activate

# 安装 Python 依赖
pip install requests numpy openai prometheus-client

# 安装系统工具
apt-get update && apt-get install -y \
    curl \
    jq \
    nethogs \
    btop

# /etc/systemd/system/cost-monitor.service
[Unit]
Description=VPS Cost Anomaly Detection System
After=network.target prometheus.service

[Service]
Type=simple
User=root
WorkingDirectory=/opt/cost-monitor
Environment=PATH=/opt/cost-monitor/venv/bin:/usr/bin:/bin
Environment=OPENAI_API_KEY=${OPENAI_API_KEY}
Environment=TELEGRAM_BOT_TOKEN=${TELEGRAM_BOT_TOKEN}
Environment=TELEGRAM_CHAT_ID=${TELEGRAM_CHAT_ID}
ExecStart=/opt/cost-monitor/venv/bin/python3 /opt/cost-monitor/main.py
Restart=always
RestartSec=30

[Install]
WantedBy=multi-user.target

#!/usr/bin/env python3
"""Main scheduler for the cost anomaly detection system."""

import schedule
import time
import logging
from datetime import datetime

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    handlers=[
        logging.FileHandler('/opt/cost-monitor/monitor.log'),
        logging.StreamHandler(),
    ]
)
logger = logging.getLogger(__name__)


def run_monitoring_cycle():
    """Execute one complete monitoring cycle."""
    logger.info("=" * 50)
    logger.info("Starting cost monitoring cycle at %s", datetime.utcnow().isoformat())

    try:
        # Step 1: Collect cost data
        from cost_collector import CostCollector
        collector = CostCollector()
        records = collector.collect_digitalocean(token=os.environ.get("DO_TOKEN"))
        if records:
            collector.save_to_database(records)
            logger.info(f"Collected {len(records)} cost records")

        # Step 2: Detect anomalies
        from cost_anomaly_detector import CostAnomalyDetector
        detector = CostAnomalyDetector()
        alerts = detector.run_full_detection()

        if alerts:
            logger.warning(f"Detected {len(alerts)} anomaly alerts")

            # Step 3: LLM analysis
            from cost_anomaly_analyzer import CostAnomalyAnalyzer
            analyzer = CostAnomalyAnalyzer()
            result = analyzer.analyze(alerts)

            # Step 4: Dispatch alerts
            from alert_dispatcher import AlertDispatcher
            dispatcher = AlertDispatcher()
            dispatcher.dispatch(result)

            # Step 5: Execute remediation for critical issues
            if any(a.severity == "critical" for a in alerts):
                from cost_remediator import CostRemediator
                remediator = CostRemediator()
                for anomaly_type in ["high_cpu_cost", "high_network_cost", "storage_growth_spike"]:
                    remed_result = remediator.execute(anomaly_type, dry_run=True)
                    logger.info(f"Remediation {anomaly_type}: {remed_result['status']}")
        else:
            logger.info("No anomalies detected - all costs within normal range")

    except Exception as e:
        logger.error(f"Monitoring cycle failed: {e}", exc_info=True)

    logger.info("Monitoring cycle completed")


if __name__ == "__main__":
    import os
    # Run every 15 minutes during business hours, every hour off-hours
    schedule.every(15).minutes.do(run_monitoring_cycle)
    # Also run immediately on startup
    run_monitoring_cycle()

    logger.info("Cost monitor started. Checking every 15 minutes.")
    while True:
        schedule.run_pending()
        time.sleep(30)

# Edit crontab
crontab -e

# Add these entries:
# Collect cost data every 30 minutes
*/30 * * * * /opt/cost-monitor/venv/bin/python3 /opt/cost-monitor/collect_costs.py >> /opt/cost-monitor/logs/collect.log 2>&1

# Run full monitoring cycle every 15 minutes
*/15 * * * * /opt/cost-monitor/venv/bin/python3 /opt/cost-monitor/run_monitor.py >> /opt/cost-monitor/logs/monitor.log 2>&1

# Daily summary report at 9 AM
0 9 * * * /opt/cost-monitor/venv/bin/python3 /opt/cost-monitor/daily_summary.py >> /opt/cost-monitor/logs/summary.log 2>&1

# 找出异常进程
top -b -n 1 | head -20

# 找到并终止挖矿进程
kill -9 $(pgrep -f xmrig)

# 检查并清理可疑容器
docker ps -a | grep -i suspicious
docker rm -f suspicious_container

# 更新安全组规则，限制出站连接
iptables -A OUTPUT -p tcp --dport 3333 -j DROP