poyee-data-warehouse/bin/cdc-monitor.py

#!/usr/bin/env /usr/bin/python3
# -*- coding:utf-8 -*-
"""
CDC 链路监控：源库复制槽 + Flink 采集作业存活。

两种模式，同一份检查逻辑与配置：
  - 缺省（DS 每 10 分钟调一次）：只在检测到异常时推企微。无状态，每次独立判断，
    连续收到就是还没恢复，不再收到就是已恢复。
  - -report（DS 手动触发，上线核对用）：无条件推当前状态，正常也发。

检查两类：
  1. 复制槽——源库**必须是主库**，从库上 pg_current_wal_lsn() 报
     "recovery is in progress"，且 pg_replication_slots 只能看到本节点的槽
     - cdc_ 前缀的槽 active = false → 采集作业可能已停
     - 任意槽滞后超阈值 → 消费跟不上
  2. Flink session 集群上 CDC 作业缺失，或状态非 RUNNING → 作业挂了

连库失败 / Flink 查询失败本身也计入告警，不静默跳过——查不到状态和状态异常同样需要人介入。

两个 check 都返回 (facts, alerts)：facts 是当前值，供 -report 无条件列出；
alerts 是判定结果，供缺省模式决定发不发。判定口径两种模式共用一份。

Alerter 在开跑前构造：配置缺失或 key 未替换立刻暴露，不留到真出事才发现告警链路是哑的。

退出码：检测到异常仍返回 0（DS 任务不标红，异常靠企微通知）；
只有告警推送失败或脚本自身异常才非 0——告警发不出去必须让调度看见。

CLI:
  python3 bin/cdc-monitor.py [-ds postgresql/prd-poyee-aliyun-cdc] [-channel default]
                             [-jm cdhmaster02:8081] [-jobs st-cdc-large,st-cdc-small]
                             [-lag-gb 5] [-report]

参数:
  -ds       主库 datasource ref，解析项目同级 ../datasource/{ds}.ini
  -channel  告警通道名，对应 conf/alerter.ini [channels] 段
  -jm       Flink JobManager REST 地址
  -jobs     CDC 作业名，逗号分隔，逐个在 Flink 作业列表里找
  -lag-gb   槽滞后告警阈值（GB），正常应为 0
  -report   无条件推当前状态，正常也发；不带则只在异常时告警
"""
import argparse
import json
import os
import sys
import urllib.request
from datetime import datetime

project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(project_root)

from dw_base.alerter.alerter import Alerter
from dw_base.io.db import postgresql as pgdb

# 不加 slot_name 过滤：顺带覆盖财务库那条 PG→PG 的槽
SLOT_SQL = ('SELECT slot_name, active, '
            'pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn) AS lag_bytes '
            'FROM pg_replication_slots')

_FLINK_TIMEOUT_SEC = 20


def check_slots(ds_ref, lag_crit_bytes):
    """查复制槽，返回 (facts, alerts)。

    facts: [(槽名, active, 滞后字节)]，连库失败时为空。
    alerts: [(项, 说明)]，连库失败本身也算一项。
    """
    facts = []
    alerts = []
    conn = None
    try:
        conn = pgdb.connect(ds_ref)
        for slot_name, active, lag_bytes in pgdb.query(conn, SLOT_SQL):
            lag = int(lag_bytes or 0)
            facts.append((slot_name, active, lag))
            # active 只对 cdc_ 槽判：PG→PG 那条 apply 卡住时会反复重连，会误报
            if slot_name.startswith('cdc_') and not active:
                alerts.append(('槽 ' + slot_name, '空闲，采集作业可能已停'))
            if lag > lag_crit_bytes:
                alerts.append(('槽 ' + slot_name, '滞后 %.1f GB' % (lag / 1024.0 ** 3)))
    except Exception as e:
        alerts.append(('主库', '复制槽查询失败：%s' % e))
    finally:
        if conn is not None:
            conn.close()
    return facts, alerts


def check_flink(jm, jobs):
    """查 Flink session 集群的作业状态，返回 (facts, alerts)。

    facts: [(作业名, state)]，按 -jobs 顺序；集群上没有的 state 为 None。
    只报 -jobs 点名的作业——这是 CDC 链路状态，不是集群状态。

    必须按 state 过滤：作业取消/失败后仍留在 /jobs/overview 列表里，
    只判名字存在会漏报——重启死循环时状态是 RESTARTING，名字一直在。
    """
    try:
        with urllib.request.urlopen('http://%s/jobs/overview' % jm,
                                    timeout=_FLINK_TIMEOUT_SEC) as resp:
            overview = json.loads(resp.read().decode('utf-8'))
    except Exception as e:
        return [], [('Flink 集群', '查询失败：%s' % e)]

    # 同名作业会留下历史记录，RUNNING 优先
    states = {}
    for j in overview.get('jobs', []):
        if j['name'] not in states or j['state'] == 'RUNNING':
            states[j['name']] = j['state']

    facts = []
    alerts = []
    for job in jobs:
        state = states.get(job)
        facts.append((job, state))
        if state is None:
            alerts.append(('作业 ' + job, '不在 Flink 集群上'))
        elif state != 'RUNNING':
            alerts.append(('作业 ' + job, '状态 %s' % state))
    return facts, alerts


def _line(key, value, is_bad):
    """一行正文：字段名默认色，只给值上色（对齐 kb/41 §3.5）。"""
    return '> %s：<font color="%s">%s</font>' % (key, 'warning' if is_bad else 'info', value)


def _head(title, color, now):
    """标题 + 检测时间。时间用服务器本地时区（与 DS 告警消息同格式，同群里看着一致）。"""
    return ['### <font color="%s">%s</font>' % (color, title),
            '> 检测时间：%s' % (now or datetime.now()).strftime('%Y-%m-%d %H:%M:%S')]


def render(alerts, now=None):
    """拼告警 markdown：只列异常项。now 留给单测注入固定值。"""
    lines = _head('CDC 监控告警', 'warning', now)
    lines.extend(_line(k, v, True) for k, v in alerts)
    return '\n'.join(lines)


def render_status(slot_facts, flink_facts, alerts, now=None):
    """拼状态报告 markdown：无条件列出每一项当前值。

    异常项的值上 warning，正常项上 info；标题按有无异常切色。
    查询失败时没有事实可列，把失败原因单独补在末尾，否则报告会是空的。
    """
    bad = set(k for k, _ in alerts)
    lines = _head('CDC 链路状态', 'warning' if alerts else 'info', now)

    fact_keys = set()
    for slot_name, active, lag in slot_facts:
        key = '槽 ' + slot_name
        fact_keys.add(key)
        lines.append(_line(key, '%s，滞后 %.1f GB' % (
            '活跃' if active else '空闲', lag / 1024.0 ** 3), key in bad))
    for job, state in flink_facts:
        key = '作业 ' + job
        fact_keys.add(key)
        lines.append(_line(key, state or '不在集群上', key in bad))

    lines.extend(_line(k, v, True) for k, v in alerts if k not in fact_keys)
    return '\n'.join(lines)


def main():
    parser = argparse.ArgumentParser(description='CDC 链路监控（复制槽 + Flink 作业存活）')
    parser.add_argument('-ds', default='postgresql/prd-poyee-aliyun-cdc',
                        help='主库 datasource ref（默认 postgresql/prd-poyee-aliyun-cdc）')
    parser.add_argument('-channel', default='default',
                        help='告警通道名，对应 conf/alerter.ini [channels]（默认 default）')
    parser.add_argument('-jm', default='cdhmaster02:8081',
                        help='Flink JobManager REST 地址（默认 cdhmaster02:8081）')
    parser.add_argument('-jobs', default='st-cdc-large,st-cdc-small',
                        help='CDC 作业名，逗号分隔（默认 st-cdc-large,st-cdc-small）')
    parser.add_argument('-lag-gb', type=float, default=5, dest='lag_gb',
                        help='槽滞后告警阈值 GB（默认 5，正常应为 0）')
    parser.add_argument('-report', action='store_true',
                        help='无条件推当前状态，正常也发；不带则只在异常时告警')
    args = parser.parse_args()

    alerter = Alerter(channel=args.channel)
    jobs = [j.strip() for j in args.jobs.split(',') if j.strip()]

    slot_facts, slot_alerts = check_slots(args.ds, int(args.lag_gb * 1024 ** 3))
    flink_facts, flink_alerts = check_flink(args.jm, jobs)
    alerts = slot_alerts + flink_alerts

    for key, desc in alerts:
        print('%s: %s' % (key, desc))  # 进 DS 任务日志

    if args.report:
        content = render_status(slot_facts, flink_facts, alerts)
        print(content)                 # 发了什么，DS 任务日志里留一份
        alerter.send_markdown(content)
    elif alerts:
        alerter.send_markdown(render(alerts))


if __name__ == '__main__':
    main()