08 · 告警中心：让平台半夜主动喊你

"""
alerts/models.py — 告警中心的存储模型
"""
from django.db import models
from django.utils import timezone


class AlertPolicy(models.Model):
    """告警策略：对某个注册表指标设定阈值条件

    metric_key 见 monitor/registry.py；custom.<name> 亦可告警。
    评估窗口默认 5 分钟（取窗口内均值与阈值比较），由 alerts.engine 周期执行。
    """

    OPERATORS = [('>', '大于'), ('<', '小于')]
    LEVELS = [
        ('P0', 'P0 紧急'),
        ('P1', 'P1 严重'),
        ('P2', 'P2 警告'),
        ('提示', '提示'),
    ]

    name = models.CharField('策略名称', max_length=80)
    metric_key = models.CharField('监控指标', max_length=64, db_index=True)
    operator = models.CharField('比较符', max_length=2, choices=OPERATORS, default='>')
    threshold = models.FloatField('阈值')
    level = models.CharField('告警级别', max_length=4, choices=LEVELS, default='P2')
    enabled = models.BooleanField('启用', default=True)
    note = models.CharField('备注', max_length=200, blank=True, default='')
    # 持续越限 N 分钟才触发（for-duration，防抖动误报）；0 表示立即触发
    for_minutes = models.FloatField('持续N分钟才触发', default=0)
    # 恢复迟滞（hysteresis）：配置后，指标回到"恢复阈值"的安全侧才算恢复，
    # 避免指标在阈值附近抖动导致事件反复触发/恢复；为空则条件解除即恢复
    resolve_threshold = models.FloatField('恢复阈值', null=True, blank=True)
    # 引擎工作字段：for-duration 观察期内首次越限时间（非用户输入）
    breach_since = models.DateTimeField('首次越限时间', null=True, blank=True)
    # 静默窗口：此时间之前评估引擎跳过该策略（发版/维护期间防轰炸）
    silenced_until = models.DateTimeField('静默至', null=True, blank=True)
    created_at = models.DateTimeField('创建时间', auto_now_add=True)

    class Meta:
        verbose_name = '告警策略'
        verbose_name_plural = verbose_name
        ordering = ['level', 'id']

    def __str__(self):
        return f'{self.name}（{self.metric_key} {self.operator} {self.threshold}）'

    def describe(self):
        op = ' > ' if self.operator == '>' else ' < '
        text = f'{self.metric_key}{op}{self.threshold}'
        if self.for_minutes:
            text += f' 持续{self.for_minutes:g}分钟'
        if self.resolve_threshold is not None:
            text += f' 恢复阈值{self.resolve_threshold:g}'
        return text


class AlertEvent(models.Model):
    """一次告警事件：触发(firing) -> 确认(ack) -> 恢复(resolved)；可关联故障单"""

    STATUS = [('firing', '触发中'), ('resolved', '已恢复')]

    policy = models.ForeignKey(AlertPolicy, on_delete=models.CASCADE, related_name='events',
                               verbose_name='所属策略')
    status = models.CharField('状态', max_length=10, choices=STATUS, default='firing')
    level = models.CharField('级别', max_length=4, default='P2')
    value = models.FloatField('触发时数值', default=0.0)
    summary = models.CharField('摘要', max_length=250, blank=True, default='')
    started_at = models.DateTimeField('触发时间', default=timezone.now, db_index=True)
    resolved_at = models.DateTimeField('恢复时间', null=True, blank=True)
    last_value_at = models.DateTimeField('最近评估', null=True, blank=True)
    # ---- 处置闭环 ----
    ack_by = models.CharField('确认人', max_length=60, blank=True, default='')
    ack_at = models.DateTimeField('确认时间', null=True, blank=True)
    handle_note = models.TextField('处理备注', blank=True, default='')
    # ---- 故障事件关联（ops.Incident，用字符串引用避免循环依赖） ----
    incident = models.ForeignKey('ops.Incident', on_delete=models.SET_NULL,
                                 related_name='alerts', null=True, blank=True,
                                 verbose_name='所属故障单')

    class Meta:
        verbose_name = '告警事件'
        verbose_name_plural = verbose_name
        ordering = ['-started_at']

    def __str__(self):
        return f'[{self.status}] {self.summary}'

    @property
    def duration_min(self):
        end = self.resolved_at or timezone.now()
        return round((end - self.started_at).total_seconds() / 60, 1)

    @property
    def ack_delay_min(self):
        if not self.ack_at:
            return None
        return round((self.ack_at - self.started_at).total_seconds() / 60, 1)


class NotificationRecord(models.Model):
    """通知记录：告警触发/恢复时各生成一条（站内信；webhook 为模拟演示）"""

    CHANNELS = [('站内信', '站内信'), ('webhook', 'Webhook(模拟)'), ('邮件(模拟)', '邮件(模拟)')]
    KINDS = [('fire', '触发'), ('resolve', '恢复')]

    event = models.ForeignKey(AlertEvent, on_delete=models.CASCADE, related_name='notifications',
                              verbose_name='关联事件')
    channel = models.CharField('渠道', max_length=16, choices=CHANNELS, default='站内信')
    kind = models.CharField('类型', max_length=10, choices=KINDS, default='fire')
    title = models.CharField('标题', max_length=150)
    content = models.TextField('内容')
    created_at = models.DateTimeField('时间', auto_now_add=True)

    class Meta:
        verbose_name = '通知记录'
        verbose_name_plural = verbose_name
        ordering = ['-created_at']

    def __str__(self):
        return self.title

def _notify(event, kind):
    """产生通知记录：站内信 + 模拟 Webhook + 真实渠道（邮件/企业微信/钉钉/通用）

    触发与恢复各生成一条记录（kind 参与去重键），恢复通知不会再被触发记录吞掉。
    """
    action = '触发' if kind == 'fire' else '恢复'
    title = f'[{event.level}] {event.summary}'
    content = (
        f'告警{action}通知\n'
        f'策略：{event.policy.name}\n'
        f'条件：{event.policy.describe()}\n'
        f'当前值：{_fmt_value(event.value)}\n'
        f'时间：{timezone.localtime(event.started_at if kind == "fire" else event.resolved_at):%Y-%m-%d %H:%M:%S}'
    )
    from .models import NotificationRecord
    NotificationRecord.objects.get_or_create(
        event=event, channel='站内信', kind=kind,
        defaults={'title': title[:150], 'content': content},
    )
    NotificationRecord.objects.get_or_create(
        event=event, channel='webhook', kind=kind,
        defaults={'title': title[:150], 'content': f'POST https://hooks.example.com/obs {content[:800]}'},
    )
    # 真实渠道（ops.NotifyConfig 启用后生效，10 篇接入）
    try:
        from ops.notify import send_notify
        for channel, status in send_notify(title, content):
            NotificationRecord.objects.get_or_create(
                event=event, channel=channel[:16], kind=kind,
                defaults={'title': f'{title[:120]} [{status[:20]}]', 'content': content[:1500]},
            )
    except Exception:
        logger.exception('外部通知渠道发送失败')

def _condition(value, policy):
    if value is None:
        return False
    return value > policy.threshold if policy.operator == '>' else value < policy.threshold


def _recovered(value, policy):
    """恢复判定：指标回到安全侧才算恢复（恢复迟滞）。

    未配置恢复阈值时，条件解除（breach=False）即恢复；
    配置后要求 value 严格越过 resolve_threshold 的安全侧——
    指标在触发阈值附近抖动时不会反复触发/恢复（防 flapping）。
    """
    if policy.resolve_threshold is None:
        return True
    if value is None:
        return True  # 数据中断视为解除，事件挂起无意义
    return (value < policy.resolve_threshold if policy.operator == '>'
            else value > policy.resolve_threshold)

def evaluate_once():
    """单轮评估：返回 (触发数, 恢复数)。

    进程内加锁：与"立即评估"按钮、多个后台线程互斥。
    已在评估中的调用直接返回 (0, 0)，不排队等待（引擎下一轮会重新评估）。
    """
    from ops.heal import process_heal_for_event          # 10 篇启用
    from ops.incidents import attach_or_create, maybe_resolve  # 10 篇启用

    from .models import AlertEvent, AlertPolicy

    if not evaluate_lock.acquire(blocking=False):
        return 0, 0
    try:
        window = MON['ALERT_WINDOW_MIN']
        fired = recovered = 0

        _merge_duplicate_firing_events()

        for policy in AlertPolicy.objects.filter(enabled=True):
            try:
                # 静默窗口：维护/发版期间跳过评估
                if policy.silenced_until and policy.silenced_until > timezone.now():
                    continue
                from monitor.registry import metric_value
                value = metric_value(policy.metric_key, minutes=window)
                breach = _condition(value, policy)
                active = AlertEvent.objects.filter(policy=policy, status='firing').first()
                now = timezone.now()

                if breach and not active:
                    # for-duration 防抖：首次越限记时，持续满 for_minutes 才真正触发
                    if policy.for_minutes > 0:
                        if policy.breach_since is None:
                            policy.breach_since = now
                            policy.save(update_fields=['breach_since'])
                            continue
                        if (now - policy.breach_since).total_seconds() < policy.for_minutes * 60:
                            continue  # 观察期内，暂不触发
                    event = AlertEvent.objects.create(
                        policy=policy, status='firing', level=policy.level,
                        value=value or 0,
                        summary=(
                            f'{policy.name}：{policy.metric_key} 当前 '
                            f'{_fmt_value(value)}（{"高于" if policy.operator == ">" else "低于"}阈值 '
                            f'{_fmt_value(policy.threshold)}'
                            f'{f"，已持续 {policy.for_minutes:g} 分钟" if policy.for_minutes else ""}）'
                        )[:250],
                    )
                    if policy.breach_since is not None:
                        policy.breach_since = None
                        policy.save(update_fields=['breach_since'])
                    attach_or_create(event)          # 并入/创建故障单（10 篇）
                    _notify(event, 'fire')
                    process_heal_for_event(event)    # 自愈动作（10 篇）
                    logger.warning('告警触发: %s (value=%s)', event.summary, value)
                    fired += 1
                elif not breach and active:
                    if not _recovered(value, policy):
                        # 恢复迟滞：指标未回到安全侧（在阈值附近抖动），保持 firing
                        active.last_value_at = now
                        if value is not None:
                            active.value = value
                        active.save(update_fields=['last_value_at', 'value'])
                        continue
                    if policy.breach_since is not None:
                        policy.breach_since = None
                        policy.save(update_fields=['breach_since'])
                    active.status = 'resolved'
                    active.resolved_at = now
                    active.value = value if value is not None else active.value
                    active.save(update_fields=['status', 'resolved_at', 'value'])
                    _notify(active, 'resolve')
                    if active.incident:
                        maybe_resolve(active.incident)
                    logger.info('告警恢复: %s', active.summary)
                    recovered += 1
                else:
                    if not breach and policy.breach_since is not None:
                        # 观察期内条件解除：重置计时（下一次越限重新观察）
                        policy.breach_since = None
                        policy.save(update_fields=['breach_since'])
                    if active:
                        active.last_value_at = now
                        if value is not None:
                            active.value = value
                        active.save(update_fields=['last_value_at', 'value'])
            except Exception:
                logger.exception('策略评估失败: %s', policy)
        return fired, recovered
    finally:
        evaluate_lock.release()

"""
alerts/engine.py — 自研告警评估引擎（后台线程）
…（文件头注释含并发设计说明，见仓库原文）…
"""
import logging
import threading
import time
from datetime import timedelta

from django.conf import settings
from django.utils import timezone

logger = logging.getLogger(__name__)

MON = settings.OBSERVABILITY

# 单轮评估互斥锁：后台引擎线程与页面"立即评估"共用
evaluate_lock = threading.Lock()


def _fmt_value(v):
    if v is None:
        return '-'
    return f'{v:g}'


def _merge_duplicate_firing_events():
    """自愈：多线程/多进程同时评估可能为同一策略产生多条触发中事件，合并保留最新一条"""
    from .models import AlertEvent
    for policy_id in AlertEvent.objects.filter(status='firing').values_list('policy_id', flat=True).distinct():
        dupes = list(AlertEvent.objects.filter(policy_id=policy_id, status='firing').order_by('-started_at'))
        for extra in dupes[1:]:
            extra.delete()


def _prune_model(model, deadline, batch=5000):
    """分批删除过期数据：SQLite 上一次大 DELETE 会长时间锁库，分批更平滑"""
    while True:
        pks = list(model.objects.filter(created_at__lt=deadline).values_list('pk', flat=True)[:batch])
        if not pks:
            break
        model.objects.filter(pk__in=pks).delete()
        if len(pks) < batch:
            break


def prune_old_data():
    """清理过期采集数据（请求指标 / 主机指标 / RUM 事件 / 日志 / 自定义指标）"""
    from loghub.models import LogEntry
    from hosts.models import HostMetric
    from monitor.models import CustomMetric, RequestMetric
    from rum.models import RumEvent

    deadline = timezone.now() - timedelta(days=MON['RETENTION_DAYS'])
    for model in (RequestMetric, HostMetric, RumEvent, LogEntry, CustomMetric):
        try:
            _prune_model(model, deadline)
        except Exception:
            logger.exception('清理 %s 过期数据失败', model.__name__)


_last_prune = [0.0]  # 清理节拍（可变容器，便于测试重置）


def alert_engine_round():
    """一轮：评估所有策略 + 每小时清理一次过期采集数据。

    顺序是"先评估再清理"，且清理单独兜异常——清理失败不能饿死告警评估。
    """
    evaluate_once()
    if time.time() - _last_prune[0] > 3600:
        _last_prune[0] = time.time()
        try:
            prune_old_data()
        except Exception:
            logger.exception('过期采集数据清理失败')


def alert_engine_loop():
    """告警评估线程主循环：只有持有租约的进程评估（多 worker 副本不会重复告警）"""
    from monitor.leadership import LeaseLoop

    LeaseLoop('alert-engine', alert_engine_round, MON['ALERT_INTERVAL_SEC']).run()

"""
alerts/views.py — 告警策略/事件/通知页面
"""
from django.contrib.auth.decorators import login_required
from django.http import JsonResponse
from django.shortcuts import get_object_or_404, redirect, render
from django.utils import timezone
from django.views.decorators.http import require_POST

from monitor.registry import catalog

from .engine import evaluate_once
from .models import AlertEvent, AlertPolicy, NotificationRecord


@login_required
def policies(request):
    items = list(AlertPolicy.objects.all())
    return render(request, 'alerts/policies.html',
                  {'items': items, 'metrics': catalog()})


@login_required
@require_POST
def policy_save(request):
    pk = request.POST.get('pk')
    params = {k: request.POST.get(k) for k in
              ('name', 'metric_key', 'operator', 'threshold', 'level')}
    if pk:
        policy = get_object_or_404(AlertPolicy, pk=pk)
        for k, v in params.items():
            setattr(policy, k, v)
        policy.save()
    else:
        policy = AlertPolicy.objects.create(**params)
    return redirect('alerts:policies')


@login_required
@require_POST
def policy_toggle(request, pk):
    policy = get_object_or_404(AlertPolicy, pk=pk)
    policy.enabled = not policy.enabled
    policy.save(update_fields=['enabled'])
    return redirect('alerts:policies')


@login_required
@require_POST
def policy_silence(request, pk):
    """静默 N 小时（发版/维护窗口用）"""
    policy = get_object_or_404(AlertPolicy, pk=pk)
    hours = float(request.POST.get('hours', 1))
    policy.silenced_until = timezone.now() + timezone.timedelta(hours=hours)
    policy.save(update_fields=['silenced_until'])
    return redirect('alerts:policies')


@login_required
@require_POST
def evaluate_now(request):
    """立即评估（与后台引擎共用一把锁，不会重复告警）"""
    fired, recovered = evaluate_once()
    return JsonResponse({'ok': True, 'fired': fired, 'recovered': recovered})


@login_required
def events(request):
    status = request.GET.get('status', 'firing')
    qs = AlertEvent.objects.select_related('policy')
    if status in ('firing', 'resolved'):
        qs = qs.filter(status=status)
    return render(request, 'alerts/events.html',
                  {'items': qs[:100], 'status': status})


@login_required
def notifications(request):
    return render(request, 'alerts/notifications.html',
                  {'items': NotificationRecord.objects.select_related('event')[:200]})

from django.urls import path

from . import views

urlpatterns = [
    path('', views.events, name='events'),
    path('policies/', views.policies, name='policies'),
    path('policies/save/', views.policy_save, name='policy_save'),
    path('policies/<int:pk>/toggle/', views.policy_toggle, name='policy_toggle'),
    path('policies/<int:pk>/silence/', views.policy_silence, name='policy_silence'),
    path('evaluate/', views.evaluate_now, name='evaluate_now'),
    path('notifications/', views.notifications, name='notifications'),
]

{% extends "base.html" %}
{% block title %}告警策略{% endblock %}
{% block content %}
<h1>告警策略</h1>
<form method="post" action="{% url 'alerts:policy_save' %}" style="background:#fff;padding:16px;">
  {% csrf_token %}
  <input name="name" placeholder="策略名（如 CPU 过高）" required>
  <select name="metric_key">
    {% for m in metrics %}<option value="{{ m.key }}">{{ m.label }}</option>{% endfor %}
  </select>
  <select name="operator"><option value=">">大于</option><option value="<">小于</option></select>
  <input name="threshold" type="number" step="any" placeholder="阈值" required style="width:100px">
  <select name="level">
    <option>P0</option><option>P1</option><option selected>P2</option><option>提示</option>
  </select>
  <button>保存策略</button>
</form>
<table border="1" cellpadding="6" style="border-collapse:collapse;background:#fff;margin-top:12px;">
  <tr><th>名称</th><th>条件</th><th>级别</th><th>状态</th><th>静默至</th><th>操作</th></tr>
  {% for p in items %}
  <tr>
    <td>{{ p.name }}</td>
    <td><code>{{ p.describe }}</code></td>
    <td>{{ p.level }}</td>
    <td>{{ p.enabled|yesno:"启用,停用" }}</td>
    <td>{{ p.silenced_until|date:"m-d H:i"|default:"—" }}</td>
    <td>
      <form method="post" action="{% url 'alerts:policy_toggle' p.pk %}" style="display:inline">
        {% csrf_token %}<button>{{ p.enabled|yesno:"停用,启用" }}</button></form>
      <form method="post" action="{% url 'alerts:policy_silence' p.pk %}" style="display:inline">
        {% csrf_token %}<input type="hidden" name="hours" value="1"><button>静默1h</button></form>
    </td>
  </tr>
  {% endfor %}
</table>
{% endblock %}

{% extends "base.html" %}
{% block title %}告警事件{% endblock %}
{% block content %}
<h1>告警事件</h1>
<p>
  <a href="?status=firing">触发中</a> ｜
  <a href="?status=resolved">已恢复</a> ｜
  <a href="?status=">全部</a>
  <button onclick="fetch('/alerts/evaluate/',{method:'POST',headers:{'X-CSRFToken':'{{ csrf_token }}'}}).then(r=>r.json()).then(()=>location.reload())">立即评估</button>
</p>
<table border="1" cellpadding="6" style="border-collapse:collapse;background:#fff;">
  <tr><th>时间</th><th>级别</th><th>摘要</th><th>状态</th><th>持续(分)</th></tr>
  {% for e in items %}
  <tr>
    <td>{{ e.started_at|date:"m-d H:i" }}</td>
    <td><b>{{ e.level }}</b></td>
    <td>{{ e.summary }}</td>
    <td>{{ e.get_status_display }}</td>
    <td>{{ e.duration_min }}</td>
  </tr>
  {% endfor %}
</table>
{% endblock %}

        from hosts.collector import host_collector_loop
        from alerts.engine import alert_engine_loop
        # from ops.workers import probe_loop, inspect_loop  # 10 篇启用

        threads = (
            ('obs-host-collector', host_collector_loop, 'host-collector'),
            ('obs-alert-engine', alert_engine_loop, 'alert-engine'),
        )

python manage.py makemigrations alerts
python manage.py migrate
python manage.py runserver 8014

# ---------------------------------------------------------------
# 故障演练：给观测平台制造真实的慢请求 / 错误 / 日志数据
# 演练端点一律 POST（防爬虫/预取链接误触发）+ 节流（15 秒最小间隔）
# ---------------------------------------------------------------
import threading as _threading  # noqa: E402

_chaos_lock = _threading.Lock()
_chaos_last = 0.0
_CHAOS_MIN_INTERVAL = 15


def _chaos_throttled():
    global _chaos_last
    now = time.monotonic()
    with _chaos_lock:
        if now - _chaos_last < _CHAOS_MIN_INTERVAL:
            return True
        _chaos_last = now
        return False


def chaos_page(request):
    """故障演练控制台页面"""
    drills = [
        {'title': '慢请求', 'url': '/forum/chaos/slow/',
         'desc': '请求人为 sleep 1.2 秒——在 APM 看到耗时尖刺。'},
        {'title': 'N+1 查询', 'url': '/forum/chaos/nplus1/',
         'desc': '30 条帖子逐条查作者与回复数——数据库分析看 SQL 次数抬升。'},
        {'title': '未捕获异常（500）', 'url': '/forum/chaos/error/',
         'desc': '错误率、ERROR 日志（django.request 自动落库）同时出现。'},
        {'title': '日志风暴', 'url': '/forum/chaos/log/',
         'desc': '批量写入 ERROR/WARNING——日志查询立即出现聚类模板。'},
        {'title': 'CPU 空转', 'url': '/forum/chaos/cpu/',
         'desc': '纯计算 0.8 秒——主机监控 CPU 曲线出现可见抬升。'},
        {'title': '批量制造（前 4 项）', 'url': '/forum/chaos/all/',
         'desc': '依次触发慢请求、N+1、500 错误、日志风暴各一次。'},
    ]
    return render(request, 'forum/chaos.html', {'drills': drills})


@require_POST
def chaos_slow(request):
    if _chaos_throttled():
        return JsonResponse({'chaos': 'slow', 'throttled': True}, status=429)
    time.sleep(1.2)
    return JsonResponse({'chaos': 'slow', 'sleeped_ms': 1200})


@require_POST
def chaos_error(request):
    if _chaos_throttled():
        return JsonResponse({'chaos': 'error', 'throttled': True}, status=429)
    raise RuntimeError('故障演练：人为抛出的未捕获异常')


@require_POST
def chaos_log_storm(request):
    if _chaos_throttled():
        return JsonResponse({'chaos': 'log_storm', 'throttled': True}, status=429)
    import logging
    logger = logging.getLogger('chaos')
    for i in range(30):
        logger.error('故障演练：订单 payment-%d 处理失败，退款队列积压 %d 条', i, i * 3)
        if i % 3 == 0:
            logger.warning('故障演练：缓存命中率下降至 %d%%', 40 + i)
    return JsonResponse({'chaos': 'log_storm', 'written': 40})


@require_POST
def chaos_all(request):
    if _chaos_throttled():
        return JsonResponse({'chaos': 'all', 'throttled': True}, status=429)
    time.sleep(1.2)
    import logging
    logging.getLogger('chaos').error('故障演练：模拟未捕获异常（批量演练中已内部捕获）')
    return JsonResponse({'chaos': 'all', 'done': ['slow', 'log', 'error(simulated)']})

    # 故障演练（给观测平台制造真实数据）
    path('chaos/', views.chaos_page, name='chaos'),
    path('chaos/slow/', views.chaos_slow, name='chaos_slow'),
    path('chaos/nplus1/', views.chaos_nplus1, name='chaos_nplus1'),
    path('chaos/error/', views.chaos_error, name='chaos_error'),
    path('chaos/log/', views.chaos_log_storm, name='chaos_log'),
    path('chaos/cpu/', views.chaos_cpu, name='chaos_cpu'),
    path('chaos/all/', views.chaos_all, name='chaos_all'),

{% extends "base.html" %}
{% block title %}故障演练{% endblock %}
{% block content %}
<h1>故障演练</h1>
<p>以下按钮各带 15 秒节流，全部 POST 触发。制造故障后去 APM/日志/告警三个页面看反应。</p>
{% for d in drills %}
<div style="background:#fff;padding:12px;margin:8px 0;">
  <b>{{ d.title }}</b> — {{ d.desc }}<br>
  <button onclick="fetch('{{ d.url }}',{method:'POST',headers:{'X-CSRFToken':'{{ csrf_token }}'}}).then(r=>r.json()).then(j=>alert(JSON.stringify(j))).catch(e=>alert('已触发（'+e+'）'))">触发</button>
</div>
{% endfor %}
{% endblock %}