feat(alerter): 加企微告警封装 + CDC 链路监控脚本与单测

Alerter 读 conf/alerter.ini 按通道取 key，配置缺失 / 通道未注册 / key 占位 /
errcode 非 0 四类失败均抛异常。cdc-monitor 查复制槽 active + 滞后与 YARN 作业
存活，消息带检测时间，复用 dw_base.io.db.postgresql 的 ds_ref 解析；单测 29 例
PG / YARN / urlopen 全走替身。
kb/00 配置表原指向已删的 alerter_constants.py，一并改掉。

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

bin/cdc-monitor.py

@@ -0,0 +1,121 @@
+#!/usr/bin/env /usr/bin/python3
+# -*- coding:utf-8 -*-
+"""
+CDC 链路监控：源库复制槽 + YARN 采集作业存活。
+
+DS 每 5 分钟调一次，无状态：每次独立判断。连续收到就是还没恢复，不再收到就是已恢复。
+
+检查两类：
+  1. 复制槽——源库**必须是主库**，从库上 pg_current_wal_lsn() 报
+     "recovery is in progress"，且 pg_replication_slots 只能看到本节点的槽
+     - cdc_ 前缀的槽 active = false → 采集作业可能已停
+     - 任意槽滞后超阈值 → 消费跟不上
+  2. YARN RUNNING 列表里缺 CDC 作业 → 作业挂了
+
+连库失败 / YARN 查询失败本身也计入告警，不静默跳过——查不到状态和状态异常同样需要人介入。
+
+Alerter 在开跑前构造：配置缺失或 key 未替换立刻暴露，不留到真出事才发现告警链路是哑的。
+
+退出码：检测到异常仍返回 0（DS 任务不标红，异常靠企微通知）；
+只有告警推送失败或脚本自身异常才非 0——告警发不出去必须让调度看见。
+
+CLI:
+  python3 bin/cdc-monitor.py [-ds postgresql/prd-poyee-aliyun-cdc] [-channel default]
+                             [-jobs st-cdc-large,st-cdc-small] [-lag-gb 5]
+
+参数:
+  -ds       主库 datasource ref，解析项目同级 ../datasource/{ds}.ini
+  -channel  告警通道名，对应 conf/alerter.ini [channels] 段
+  -jobs     CDC 作业名，逗号分隔，逐个在 YARN RUNNING 列表里找
+  -lag-gb   槽滞后告警阈值（GB），正常应为 0
+"""
+import argparse
+import os
+import subprocess
+import sys
+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')
+
+_YARN_TIMEOUT_SEC = 60
+
+
+def check_slots(ds_ref, lag_crit_bytes):
+    """查复制槽，返回 [(项, 说明)]。连库失败本身也算一项。"""
+    alerts = []
+    conn = None
+    try:
+        conn = pgdb.connect(ds_ref)
+        for slot_name, active, lag_bytes in pgdb.query(conn, SLOT_SQL):
+            # active 只对 cdc_ 槽判：PG→PG 那条 apply 卡住时会反复重连，会误报
+            if slot_name.startswith('cdc_') and not active:
+                alerts.append(('槽 ' + slot_name, '空闲，采集作业可能已停'))
+            lag = int(lag_bytes or 0)
+            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 alerts
+
+
+def check_yarn(jobs):
+    """查 YARN RUNNING 列表，返回 [(项, 说明)]。"""
+    try:
+        running = subprocess.check_output(
+            ['yarn', 'application', '-list', '-appStates', 'RUNNING'],
+            stderr=subprocess.STDOUT, timeout=_YARN_TIMEOUT_SEC).decode('utf-8')
+    except Exception as e:
+        return [('YARN', '查询失败：%s' % e)]
+    return [('作业 ' + job, '不在 YARN RUNNING 列表')
+            for job in jobs if job not in running]
+
+
+def render(alerts, now=None):
+    """拼企微 markdown。
+
+    检测时间用服务器本地时区（与 DS 告警消息同格式，同群里看着一致）。
+    now 留给单测注入固定值，默认取当前时间。
+    """
+    lines = ['### <font color="warning">CDC 监控告警</font>',
+             '> 检测时间：%s' % (now or datetime.now()).strftime('%Y-%m-%d %H:%M:%S')]
+    lines.extend('> %s：<font color="warning">%s</font>' % (k, v) for k, v in alerts)
+    return '\n'.join(lines)
+
+
+def main():
+    parser = argparse.ArgumentParser(description='CDC 链路监控（复制槽 + YARN 作业存活）')
+    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('-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）')
+    args = parser.parse_args()
+
+    alerter = Alerter(channel=args.channel)
+    jobs = [j.strip() for j in args.jobs.split(',') if j.strip()]
+
+    alerts = check_slots(args.ds, int(args.lag_gb * 1024 ** 3)) + check_yarn(jobs)
+
+    for key, desc in alerts:
+        print('%s: %s' % (key, desc))  # 进 DS 任务日志
+    if alerts:
+        alerter.send_markdown(render(alerts))
+
+
+if __name__ == '__main__':
+    main()

conf/alerter.ini

@@ -14,4 +14,4 @@
 base_url = https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=
 
 [channels]
-default = <替换为企微机器人 key>
+default = 09344dbd-616a-4bc1-8add-b8c255e6f644

dw_base/alerter/alerter.py

@@ -0,0 +1,51 @@
+# -*- coding:utf-8 -*-
+"""
+企业微信告警发送。
+
+职责：
+- 读 conf/alerter.ini 拿 base_url + 按通道名取机器人 key
+- send_markdown 发企微 markdown 消息
+
+失败一律抛异常，不静默：告警发不出去必须让调度看见，否则问题被双重掩盖。
+覆盖四种情况——配置文件缺失、通道未注册、key 还是占位符、企微返回 errcode != 0。
+（ConfigParser.read 对缺失文件静默返回，故先显式校验存在性。）
+
+消息拼装由调用方负责：不同 monitor 的标题和字段各不相同，塞进来只会变成参数堆。
+"""
+import json
+import os
+import urllib.request
+from configparser import ConfigParser
+
+_PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+_DEFAULT_CONF_PATH = os.path.join(_PROJECT_ROOT, 'conf', 'alerter.ini')
+_TIMEOUT_SEC = 20
+
+
+class Alerter:
+    def __init__(self, channel='default', conf_path=None):
+        # type: (str, str) -> None
+        path = conf_path or _DEFAULT_CONF_PATH
+        if not os.path.isfile(path):
+            raise RuntimeError('告警配置文件不存在：' + path)
+        cp = ConfigParser()
+        cp.read(path, encoding='utf-8')
+        if not cp.has_option('channels', channel):
+            raise RuntimeError(
+                "告警通道未注册：'{0}'；{1} [channels] 段加 '{0} = <key>'".format(channel, path))
+        key = cp.get('channels', channel).strip()
+        if key.startswith('<'):
+            raise RuntimeError(
+                "告警通道 '{0}' 的 key 仍是占位符，填真实机器人 key：{1}".format(channel, path))
+        self.webhook = cp.get('wechat_work', 'base_url').strip() + key
+
+    def send_markdown(self, content):
+        # type: (str) -> None
+        """发 markdown 消息。企微 markdown 单条上限 4096 字节，超长由调用方自行截断。"""
+        payload = json.dumps({'msgtype': 'markdown', 'markdown': {'content': content}},
+                             ensure_ascii=False).encode('utf-8')
+        req = urllib.request.Request(
+            self.webhook, data=payload, headers={'Content-Type': 'application/json'})
+        resp = urllib.request.urlopen(req, timeout=_TIMEOUT_SEC).read().decode('utf-8')
+        if json.loads(resp).get('errcode') != 0:
+            raise RuntimeError('企微发送失败：' + resp)

kb/00-项目导览.md

@@ -44,6 +44,7 @@ poyee-data-warehouse/              # 项目根目录（仓库名 = 部署名）
 │   │   ├── common/                #     通用 UDF
 │   │   └── business/              #     业务专用 UDF
 │   ├── utils/                     #   通用工具（参数解析、日期、文件、日志、SQL 解析等）
+│   ├── alerter/                   #   告警发送（企微 webhook，配置见 conf/alerter.ini）
 │   ├── io/                        #   （占位）I/O 边界：跨进程读写
 │   │   ├── db/
 │   │   ├── file/
@@ -133,7 +134,7 @@ poyee-data-warehouse/              # 项目根目录（仓库名 = 部署名）
 | Spark 默认参数 | `conf/spark-defaults.conf`（行为/开关）+ `conf/spark-tuning.conf`（资源/调优） | 是 | 开发   |
 | Spark 单作业覆盖 | 对应 `jobs/*.sql` 文件内 `SET spark.x.y=z` | 是 | 开发   |
 | 环境变量 / 路径 | `conf/env.sh`（`dw_base/utils/env_loader.py` 解析 + `bootstrap_env` 注入；运行时 env 变量明细见 `01-运行环境.md` §4） | 是 | 开发   |
-| 告警 Webhook | `dw_base/common/alerter_constants.py` | 是 | 开发   |
+| 告警 Webhook | `conf/alerter.ini`（`dw_base/alerter/alerter.py` 解析；按通道名取 key） | 是 | 开发   |
 
 ## 4. DataX 入口使用说明（待重构后完善）
 

tests/unit/alerter/test_alerter.py

@@ -0,0 +1,116 @@
+# -*- coding:utf-8 -*-
+"""dw_base.alerter.alerter 配置加载 + 企微推送单测（不发真消息，urlopen 走替身）。"""
+import json
+from unittest.mock import MagicMock
+
+import pytest
+
+from dw_base.alerter import alerter as alerter_mod
+from dw_base.alerter.alerter import Alerter
+
+BASE_URL = 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key='
+
+
+def _write_conf(tmp_path, channels=None):
+    """造一份临时 alerter.ini，返回路径。"""
+    if channels is None:
+        channels = {'default': 'real-key-0001'}
+    lines = ['[wechat_work]', 'base_url = ' + BASE_URL, '', '[channels]']
+    lines.extend('{0} = {1}'.format(name, key) for name, key in channels.items())
+    path = tmp_path / 'alerter.ini'
+    path.write_text('\n'.join(lines) + '\n', encoding='utf-8')
+    return str(path)
+
+
+def _stub_urlopen(captured, resp='{"errcode":0,"errmsg":"ok"}'):
+    """替身 urlopen：把 Request 存进 captured，返回可 read 的响应对象。"""
+    def _open(req, timeout=None):
+        captured['req'] = req
+        captured['timeout'] = timeout
+        fake_resp = MagicMock()
+        fake_resp.read.return_value = resp.encode('utf-8')
+        return fake_resp
+    return _open
+
+
+# ---------- 构造期的四类失败 ----------
+
+def test_conf_file_missing_raises(tmp_path):
+    with pytest.raises(RuntimeError, match='告警配置文件不存在'):
+        Alerter(conf_path=str(tmp_path / 'nope.ini'))
+
+
+def test_channel_not_registered_raises(tmp_path):
+    conf = _write_conf(tmp_path, {'default': 'real-key-0001'})
+    with pytest.raises(RuntimeError, match='告警通道未注册'):
+        Alerter(channel='realtime', conf_path=conf)
+
+
+def test_placeholder_key_raises(tmp_path):
+    conf = _write_conf(tmp_path, {'default': '<替换为企微机器人 key>'})
+    with pytest.raises(RuntimeError, match='仍是占位符'):
+        Alerter(conf_path=conf)
+
+
+def test_webhook_composed_from_base_and_key(tmp_path):
+    conf = _write_conf(tmp_path, {'default': 'abc-123'})
+    assert Alerter(conf_path=conf).webhook == BASE_URL + 'abc-123'
+
+
+def test_custom_channel_picks_its_own_key(tmp_path):
+    conf = _write_conf(tmp_path, {'default': 'key-d', 'realtime': 'key-r'})
+    assert Alerter(channel='realtime', conf_path=conf).webhook == BASE_URL + 'key-r'
+
+
+def test_key_whitespace_stripped(tmp_path):
+    path = tmp_path / 'alerter.ini'
+    path.write_text(
+        '[wechat_work]\nbase_url = {0}\n\n[channels]\ndefault =   abc-123  \n'.format(BASE_URL),
+        encoding='utf-8')
+    assert Alerter(conf_path=str(path)).webhook == BASE_URL + 'abc-123'
+
+
+# ---------- 发送 ----------
+
+def test_send_markdown_posts_expected_payload(tmp_path, monkeypatch):
+    conf = _write_conf(tmp_path, {'default': 'abc-123'})
+    captured = {}
+    monkeypatch.setattr(alerter_mod.urllib.request, 'urlopen', _stub_urlopen(captured))
+
+    Alerter(conf_path=conf).send_markdown('### 标题\n> 一行')
+
+    req = captured['req']
+    assert req.full_url == BASE_URL + 'abc-123'
+    assert req.headers['Content-type'] == 'application/json'
+    assert captured['timeout'] == alerter_mod._TIMEOUT_SEC
+    body = json.loads(req.data.decode('utf-8'))
+    assert body == {'msgtype': 'markdown', 'markdown': {'content': '### 标题\n> 一行'}}
+
+
+def test_send_markdown_keeps_chinese_unescaped(tmp_path, monkeypatch):
+    """ensure_ascii=False：中文按 UTF-8 原样进 body，不是 \\uXXXX。"""
+    conf = _write_conf(tmp_path)
+    captured = {}
+    monkeypatch.setattr(alerter_mod.urllib.request, 'urlopen', _stub_urlopen(captured))
+
+    Alerter(conf_path=conf).send_markdown('槽空闲')
+
+    raw = captured['req'].data
+    assert '槽空闲'.encode('utf-8') in raw
+    assert b'\\u' not in raw
+
+
+def test_send_markdown_raises_on_nonzero_errcode(tmp_path, monkeypatch):
+    conf = _write_conf(tmp_path)
+    monkeypatch.setattr(
+        alerter_mod.urllib.request, 'urlopen',
+        _stub_urlopen({}, resp='{"errcode":93000,"errmsg":"invalid webhook url"}'))
+
+    with pytest.raises(RuntimeError, match='企微发送失败'):
+        Alerter(conf_path=conf).send_markdown('x')
+
+
+def test_send_markdown_ok_on_zero_errcode(tmp_path, monkeypatch):
+    conf = _write_conf(tmp_path)
+    monkeypatch.setattr(alerter_mod.urllib.request, 'urlopen', _stub_urlopen({}))
+    Alerter(conf_path=conf).send_markdown('x')  # 不抛即通过

tests/unit/alerter/test_cdc_monitor.py

@@ -0,0 +1,223 @@
+# -*- coding:utf-8 -*-
+"""
+bin/cdc-monitor.py 单测：复制槽判定 / YARN 存活 / 消息渲染 / main 串联。
+
+不连真 PG（pgdb.connect + query 走替身）、不跑真 yarn（subprocess 走替身）、
+不发真消息（Alerter 走替身）。
+脚本路径含连字符，用 importlib.util 动态加载为模块。
+"""
+import importlib.util
+import os
+import re
+import sys
+from datetime import datetime
+from unittest.mock import MagicMock
+
+import pytest
+
+PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
+SCRIPT_PATH = os.path.join(PROJECT_ROOT, 'bin', 'cdc-monitor.py')
+
+GB = 1024 ** 3
+LAG_CRIT = 5 * GB
+
+
+def _load_script():
+    spec = importlib.util.spec_from_file_location('cdc_monitor', SCRIPT_PATH)
+    mod = importlib.util.module_from_spec(spec)
+    sys.modules['cdc_monitor'] = mod
+    spec.loader.exec_module(mod)
+    return mod
+
+
+MON = _load_script()
+
+
+def _stub_pg(monkeypatch, rows, conn=None):
+    """替身 PG：connect 返回 conn，query 返回 rows。返回该 conn 供断言 close。"""
+    conn = conn or MagicMock()
+    monkeypatch.setattr(MON.pgdb, 'connect', lambda ds_ref: conn)
+    monkeypatch.setattr(MON.pgdb, 'query', lambda c, sql: rows)
+    return conn
+
+
+# ---------- 复制槽 ----------
+
+def test_cdc_slot_inactive_alerts(monkeypatch):
+    _stub_pg(monkeypatch, [('cdc_large', False, 0)])
+    assert MON.check_slots('postgresql/x', LAG_CRIT) == [
+        ('槽 cdc_large', '空闲，采集作业可能已停')]
+
+
+def test_non_cdc_slot_inactive_is_ignored(monkeypatch):
+    """PG→PG 那条槽 apply 卡住时会反复重连，active 抖动不算告警。"""
+    _stub_pg(monkeypatch, [('finance_pg2pg', False, 0)])
+    assert MON.check_slots('postgresql/x', LAG_CRIT) == []
+
+
+def test_slot_lag_over_threshold_alerts(monkeypatch):
+    _stub_pg(monkeypatch, [('cdc_small', True, 6 * GB)])
+    alerts = MON.check_slots('postgresql/x', LAG_CRIT)
+    assert alerts == [('槽 cdc_small', '滞后 6.0 GB')]
+
+
+def test_slot_lag_at_threshold_not_alerted(monkeypatch):
+    """阈值是 >，正好等于不报。"""
+    _stub_pg(monkeypatch, [('cdc_small', True, LAG_CRIT)])
+    assert MON.check_slots('postgresql/x', LAG_CRIT) == []
+
+
+def test_lag_applies_to_non_cdc_slot_too(monkeypatch):
+    """active 只对 cdc_ 判，滞后对所有槽都判。"""
+    _stub_pg(monkeypatch, [('finance_pg2pg', True, 9 * GB)])
+    assert MON.check_slots('postgresql/x', LAG_CRIT) == [
+        ('槽 finance_pg2pg', '滞后 9.0 GB')]
+
+
+def test_null_lag_does_not_crash(monkeypatch):
+    """confirmed_flush_lsn 为 NULL 时 lag 是 None，当 0 处理。"""
+    _stub_pg(monkeypatch, [('cdc_large', True, None)])
+    assert MON.check_slots('postgresql/x', LAG_CRIT) == []
+
+
+def test_inactive_and_lagging_yields_two_alerts(monkeypatch):
+    _stub_pg(monkeypatch, [('cdc_large', False, 7 * GB)])
+    assert MON.check_slots('postgresql/x', LAG_CRIT) == [
+        ('槽 cdc_large', '空闲，采集作业可能已停'),
+        ('槽 cdc_large', '滞后 7.0 GB'),
+    ]
+
+
+def test_connection_closed_after_query(monkeypatch):
+    conn = _stub_pg(monkeypatch, [])
+    MON.check_slots('postgresql/x', LAG_CRIT)
+    conn.close.assert_called_once()
+
+
+def test_db_failure_becomes_alert(monkeypatch):
+    def _boom(ds_ref):
+        raise RuntimeError('数据源 ini 不存在')
+    monkeypatch.setattr(MON.pgdb, 'connect', _boom)
+    alerts = MON.check_slots('postgresql/x', LAG_CRIT)
+    assert len(alerts) == 1
+    assert alerts[0][0] == '主库'
+    assert '数据源 ini 不存在' in alerts[0][1]
+
+
+# ---------- YARN ----------
+
+def test_all_jobs_running_no_alert(monkeypatch):
+    monkeypatch.setattr(MON.subprocess, 'check_output',
+                        lambda *a, **kw: b'app-1 st-cdc-large RUNNING\napp-2 st-cdc-small RUNNING')
+    assert MON.check_yarn(['st-cdc-large', 'st-cdc-small']) == []
+
+
+def test_missing_job_alerts(monkeypatch):
+    monkeypatch.setattr(MON.subprocess, 'check_output',
+                        lambda *a, **kw: b'app-1 st-cdc-large RUNNING')
+    assert MON.check_yarn(['st-cdc-large', 'st-cdc-small']) == [
+        ('作业 st-cdc-small', '不在 YARN RUNNING 列表')]
+
+
+def test_yarn_failure_becomes_single_alert(monkeypatch):
+    def _boom(*a, **kw):
+        raise OSError('yarn: command not found')
+    monkeypatch.setattr(MON.subprocess, 'check_output', _boom)
+    alerts = MON.check_yarn(['st-cdc-large', 'st-cdc-small'])
+    assert len(alerts) == 1          # 查不到就是一条，不按作业数量刷屏
+    assert alerts[0][0] == 'YARN'
+    assert 'command not found' in alerts[0][1]
+
+
+# ---------- 渲染：走一圈全部五类告警 ----------
+
+FIXED_NOW = datetime(2026, 8, 26, 15, 21, 27)
+
+
+def test_render_single_alert():
+    assert MON.render([('槽 cdc_large', '空闲，采集作业可能已停')], now=FIXED_NOW) == (
+        '### <font color="warning">CDC 监控告警</font>\n'
+        '> 检测时间：2026-08-26 15:21:27\n'
+        '> 槽 cdc_large：<font color="warning">空闲，采集作业可能已停</font>')
+
+
+def test_render_defaults_to_current_time(monkeypatch):
+    """不传 now 时取当前时间，格式 yyyy-MM-dd HH:mm:ss。"""
+    out = MON.render([('主库', '连不上')])
+    assert re.match(r'^> 检测时间：\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$', out.split('\n')[1])
+
+
+def test_render_all_alert_kinds():
+    alerts = [
+        ('槽 cdc_large', '空闲，采集作业可能已停'),
+        ('槽 cdc_small', '滞后 6.0 GB'),
+        ('作业 st-cdc-large', '不在 YARN RUNNING 列表'),
+        ('主库', '复制槽查询失败：connection refused'),
+        ('YARN', '查询失败：timeout'),
+    ]
+    out = MON.render(alerts, now=FIXED_NOW)
+    lines = out.split('\n')
+    assert lines[0] == '### <font color="warning">CDC 监控告警</font>'
+    assert lines[1] == '> 检测时间：2026-08-26 15:21:27'
+    assert len(lines) == 7
+    for (key, desc), line in zip(alerts, lines[2:]):
+        assert line == '> {0}：<font color="warning">{1}</font>'.format(key, desc)
+
+
+# ---------- main 串联 ----------
+
+def _run_main(monkeypatch, slot_alerts, yarn_alerts):
+    """跑 main，返回替身 alerter 实例供断言。"""
+    monkeypatch.setattr(sys, 'argv', ['cdc-monitor.py'])
+    monkeypatch.setattr(MON, 'check_slots', lambda ds, lag: list(slot_alerts))
+    monkeypatch.setattr(MON, 'check_yarn', lambda jobs: list(yarn_alerts))
+    instance = MagicMock()
+    monkeypatch.setattr(MON, 'Alerter', MagicMock(return_value=instance))
+    MON.main()
+    return instance
+
+
+def test_main_sends_nothing_when_healthy(monkeypatch):
+    instance = _run_main(monkeypatch, [], [])
+    instance.send_markdown.assert_not_called()
+
+
+def test_main_sends_once_with_all_alerts(monkeypatch):
+    instance = _run_main(
+        monkeypatch,
+        [('槽 cdc_large', '空闲，采集作业可能已停')],
+        [('作业 st-cdc-small', '不在 YARN RUNNING 列表')])
+    instance.send_markdown.assert_called_once()
+    content = instance.send_markdown.call_args[0][0]
+    assert '槽 cdc_large' in content
+    assert '作业 st-cdc-small' in content
+
+
+def test_main_builds_alerter_before_checks(monkeypatch):
+    """健康时也要构造 Alerter：配置坏了要立刻暴露，不能等真出事才发现发不出去。"""
+    monkeypatch.setattr(sys, 'argv', ['cdc-monitor.py'])
+    monkeypatch.setattr(MON, 'check_slots', lambda ds, lag: [])
+    monkeypatch.setattr(MON, 'check_yarn', lambda jobs: [])
+    fake_cls = MagicMock()
+    monkeypatch.setattr(MON, 'Alerter', fake_cls)
+    MON.main()
+    fake_cls.assert_called_once_with(channel='default')
+
+
+def test_main_passes_cli_overrides(monkeypatch):
+    seen = {}
+    monkeypatch.setattr(sys, 'argv', [
+        'cdc-monitor.py', '-ds', 'postgresql/other', '-channel', 'realtime',
+        '-jobs', 'a, b ,', '-lag-gb', '2'])
+    monkeypatch.setattr(MON, 'check_slots',
+                        lambda ds, lag: seen.update(ds=ds, lag=lag) or [])
+    monkeypatch.setattr(MON, 'check_yarn', lambda jobs: seen.update(jobs=jobs) or [])
+    fake_cls = MagicMock()
+    monkeypatch.setattr(MON, 'Alerter', fake_cls)
+
+    MON.main()
+
+    assert seen['ds'] == 'postgresql/other'
+    assert seen['lag'] == 2 * GB
+    assert seen['jobs'] == ['a', 'b']      # 逗号分隔去空白、丢空项
+    fake_cls.assert_called_once_with(channel='realtime')