"""自动化异常台账:`autoexc_exceptions` + `autoexc_occurrences` + 三条 cron。 平台已经有五份「失败记录」——建号任务的 `failed`、通知出箱的 `dead`、定时任务运行的 `failed`、身份同步条目的 `failed`/`blocked`、集成的 `last_test_status`。**缺的不是第六份 日志,是它们上面的那一层**:*这一个问题,自周二起出现 47 次,在这个集成上,这样修*。 那才是人会去分诊、指派、修复、关闭的对象。 两张表: * `autoexc_exceptions` —— 归并后的异常。身份是 `(source, scope_key, signature_hash)`, 唯一约束兜底,防止两个并发采集器建出双胞胎。**`scope_key` 是身份的一部分**:同一句 厂商报错来自两个集成就是两个问题、两种修法。 * `autoexc_occurrences` —— 每一条失败的源行。`(source, source_ref_type, source_ref_id)` 唯一,这是所有采集器的幂等键,也是「每 30 分钟重扫一遍整个窗口」得以成立的原因。 父行的计数列(`occurrence_count` / `open_occurrence_count` / `affected_person_count`) **一律从子行重算,绝不累加**。累加更快,但第一次 tick 叠 tick 就会漂;而这个平台确实 会叠(部署打断一次运行、手动触发撞上定时触发)。 `scope_key` 用空字符串而不是 NULL:MySQL 的唯一约束里 NULL 不等于 NULL,可空的话 重复簇会静默堆积。 三条 cron: * `integrations.health_check`(每日 01:20 UTC)—— **本次真正补上的盲区**。此前 `integrations.last_test_status` 只在管理员手点「测试连接」时才被写过,所以凭证过期 在集成这一层永远不会被发现,只表现为下游五个域各自报出的一堆看似无关的失败。 * `automation_exception.collect`(每 30 分钟)—— 走五个采集器,归并、登记、判定恢复。 * `automation_exception.reconcile`(每日 04:10 UTC)—— 把每一条未关闭的异常从它的发生 记录重算一遍;采集只碰本次报到的簇,所以源头彻底停止产出的簇需要这条兜底。 三条都以 `enabled=1` 播种:它们只读源域、只写本域两张表(体检那条只写 integrations 自己的三个 `last_test_*` 列),不发通知、不改任何业务数据。 """ from __future__ import annotations import uuid import sqlalchemy as sa from alembic import op from app.core.migration_helpers import id_type_for, table_kw_for revision = "618" # The real head on ``origin/main`` at push time, re-checked immediately # before pushing — NOT simply the previous number. Migration numbers can # have gaps and it makes no difference to alembic; ``down_revision`` is # the only thing that decides the chain, so pointing it at a stale head # is how two branches land two heads and red every test shard. down_revision = "617" branch_labels = None depends_on = None EXC = "autoexc_exceptions" OCC = "autoexc_occurrences" JOBS = "scheduled_jobs" _JOBS_TO_SEED: tuple[tuple[str, str, str, str], ...] = ( ( "integrations.health_check", "集成 · 每日连接体检", "20 1 * * *", "每日 01:20 UTC 对每个**启用**的 integrations 行调用驱动的 test_connection()," "把结果写回 last_tested_at / last_test_status / last_test_message。" "此前这三列只在管理员手点「测试连接」时被写过,凭证过期因此永远不会在集成这层被" "发现,只会表现为 provisioning / identity_sync / 通知出箱各自报出的一堆失败。" "只确立事实,不发通知(告警与派工统一由自动化异常台承担)。每个探测独立 SAVEPOINT," "探测之间间隔 1 秒;停用的集成不探测。", ), ( "automation_exception.collect", "自动化异常 · 采集与归并", "*/30 * * * *", "每 30 分钟走一遍五个来源采集器(集成体检 / 建号任务 / 身份同步条目 / 通知出箱死信 / " "定时任务失败),按 (来源, 作用域, 错误签名) 归并成异常对象,每条失败源行登记为一次" "发生。幂等:发生记录按 (来源, 源行类型, 源行 id) 唯一,父行计数从子行重算。" "恢复由各采集器表态——完整扫描的来源可以用「这次没报」证明恢复,历史事件类的" "(定时任务运行 / 同步条目)必须给出「之后成功过」的显式证据。采集器自身失败时" "结果标 partial,引擎不做任何基于缺席的关闭。只登记不通知、不改源域数据。", ), ( "automation_exception.reconcile", "自动化异常 · 每日对账", "10 4 * * *", "每日 04:10 UTC 把每一条未关闭的异常从它的发生记录重算:总次数 / 未恢复次数 / " "受影响人数 / 严重度 / 状态。采集只碰本次报到的簇,所以源头彻底停止产出的簇" "(任务被删、集成被移除)需要这条兜底,它同时也是计数漂移时的修复路径。", ), ) def _has_table(bind, name: str) -> bool: return bool(bind.execute( sa.text( "SELECT 1 FROM information_schema.tables " "WHERE table_schema = DATABASE() AND table_name = :t LIMIT 1", ), {"t": name}, ).scalar()) def upgrade() -> None: bind = op.get_bind() # ``sa.Uuid()`` renders CHAR(32) on MySQL, while the ORM's ``Uuid`` is # ``app.core.sql_types.GUID`` = CHAR(36). A migration that reaches for # the real ``sa.Uuid`` builds a column the models can never match and # an FK the parent rejects outright (errno 3780). ``id_type_for`` # reads the parent column and pins its collation too. Migration 612 # shipped this bug and froze the production chain at 611. pid = id_type_for("identity_persons") integ_id = id_type_for("integrations") kw = table_kw_for("identity_persons") if not _has_table(bind, EXC): op.create_table( EXC, sa.Column("id", pid, nullable=False), sa.Column("source", sa.String(length=32), nullable=False), # Empty string, never NULL — NULL never compares equal to NULL # in a MySQL unique constraint, so a nullable scope would let # duplicate clusters accumulate silently. sa.Column( "scope_key", sa.String(length=160), nullable=False, server_default="", ), sa.Column("signature_hash", sa.String(length=32), nullable=False), sa.Column("signature", sa.String(length=255), nullable=False), sa.Column( "status", sa.String(length=16), nullable=False, server_default="open", ), sa.Column( "severity", sa.String(length=16), nullable=False, server_default="normal", ), sa.Column( "root_cause", sa.String(length=32), nullable=False, server_default="unknown", ), sa.Column( "cause_source", sa.String(length=16), nullable=False, server_default="none", ), sa.Column("integration_id", integ_id, nullable=True), sa.Column("owner_person_id", pid, nullable=True), sa.Column( "owner_source", sa.String(length=24), nullable=False, server_default="none", ), sa.Column( "occurrence_count", sa.Integer(), nullable=False, server_default="0", ), sa.Column( "open_occurrence_count", sa.Integer(), nullable=False, server_default="0", ), sa.Column( "affected_person_count", sa.Integer(), nullable=False, server_default="0", ), sa.Column( "recurrence_count", sa.Integer(), nullable=False, server_default="0", ), sa.Column("first_seen_at", sa.DateTime(timezone=True), nullable=False), sa.Column("last_seen_at", sa.DateTime(timezone=True), nullable=False), sa.Column("sample_error", sa.Text(), nullable=True), sa.Column("context", sa.JSON(), nullable=True), sa.Column("advice", sa.Text(), nullable=True), sa.Column("advice_refs", sa.JSON(), nullable=True), sa.Column( "advice_generated_at", sa.DateTime(timezone=True), nullable=True, ), sa.Column("advice_model", sa.String(length=64), nullable=True), sa.Column("acknowledged_at", sa.DateTime(timezone=True), nullable=True), sa.Column("acknowledged_by_id", pid, nullable=True), sa.Column("resolved_at", sa.DateTime(timezone=True), nullable=True), sa.Column("resolved_by_id", pid, nullable=True), sa.Column("resolution_kind", sa.String(length=16), nullable=True), sa.Column("resolution_note", sa.Text(), nullable=True), sa.Column("subject_notified_at", sa.DateTime(timezone=True), nullable=True), sa.Column("escalated_at", sa.DateTime(timezone=True), nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False, ), sa.Column( "updated_at", sa.DateTime(timezone=True), server_default=sa.text("CURRENT_TIMESTAMP"), server_onupdate=sa.text("CURRENT_TIMESTAMP"), nullable=False, ), sa.ForeignKeyConstraint( ["integration_id"], ["integrations.id"], ondelete="SET NULL", ), sa.ForeignKeyConstraint( ["owner_person_id"], ["identity_persons.id"], ondelete="SET NULL", ), sa.ForeignKeyConstraint( ["acknowledged_by_id"], ["identity_persons.id"], ondelete="SET NULL", ), sa.ForeignKeyConstraint( ["resolved_by_id"], ["identity_persons.id"], ondelete="SET NULL", ), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint( "source", "scope_key", "signature_hash", name="uq_autoexc_source_scope_signature", ), **kw, ) op.create_index( "ix_autoexc_status_last_seen", EXC, ["status", "last_seen_at"], ) op.create_index( "ix_autoexc_integration_status", EXC, ["integration_id", "status"], ) op.create_index("ix_autoexc_source_status", EXC, ["source", "status"]) op.create_index( "ix_autoexc_owner_status", EXC, ["owner_person_id", "status"], ) op.create_index("ix_autoexc_resolved_at", EXC, ["resolved_at"]) if not _has_table(bind, OCC): op.create_table( OCC, sa.Column("id", pid, nullable=False), sa.Column("exception_id", pid, nullable=False), sa.Column("source", sa.String(length=32), nullable=False), sa.Column("source_ref_type", sa.String(length=48), nullable=False), sa.Column("source_ref_id", sa.String(length=64), nullable=False), sa.Column("person_id", pid, nullable=True), sa.Column("occurred_at", sa.DateTime(timezone=True), nullable=False), sa.Column("error_text", sa.Text(), nullable=True), sa.Column("detail", sa.JSON(), nullable=True), sa.Column("recovered_at", sa.DateTime(timezone=True), nullable=True), sa.Column("recovery_kind", sa.String(length=16), nullable=True), sa.Column( "replay_attempt_count", sa.Integer(), nullable=False, server_default="0", ), sa.Column("last_replay_at", sa.DateTime(timezone=True), nullable=True), sa.Column("last_replay_error", sa.Text(), nullable=True), sa.Column( "replayable", sa.Boolean(), nullable=False, server_default="0", ), sa.Column( "created_at", sa.DateTime(timezone=True), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False, ), sa.Column( "updated_at", sa.DateTime(timezone=True), server_default=sa.text("CURRENT_TIMESTAMP"), server_onupdate=sa.text("CURRENT_TIMESTAMP"), nullable=False, ), sa.ForeignKeyConstraint( ["exception_id"], [f"{EXC}.id"], ondelete="CASCADE", ), sa.ForeignKeyConstraint( ["person_id"], ["identity_persons.id"], ondelete="SET NULL", ), sa.PrimaryKeyConstraint("id"), # The idempotency key every collector relies on. sa.UniqueConstraint( "source", "source_ref_type", "source_ref_id", name="uq_autoexc_occurrence_source_ref", ), **kw, ) op.create_index( "ix_autoexc_occ_exception_recovered", OCC, ["exception_id", "recovered_at"], ) op.create_index( "ix_autoexc_occ_person_recovered", OCC, ["person_id", "recovered_at"], ) # Recovery runs "every still-open occurrence of this source" on # each collector tick; no other index leads with ``source``. op.create_index( "ix_autoexc_occ_source_recovered", OCC, ["source", "recovered_at"], ) op.create_index("ix_autoexc_occ_occurred_at", OCC, ["occurred_at"]) if _has_table(bind, JOBS): for handler, display, cron, desc in _JOBS_TO_SEED: bind.execute( sa.text( f"INSERT INTO {JOBS} " "(id, name, display_name, description, handler_key, " " trigger_type, cron_expr, timezone, enabled, config, " " run_count, failure_count, created_at, updated_at) " "SELECT :id, :name, :display, :desc, :handler, 'cron', " ":cron, 'UTC', 1, :config, 0, 0, NOW(), NOW() " "FROM DUAL WHERE NOT EXISTS " f"(SELECT 1 FROM {JOBS} WHERE handler_key = :handler)", ), { "id": str(uuid.uuid4()), "name": handler, "display": display, "desc": desc, "handler": handler, "cron": cron, "config": "{}", }, ) def downgrade() -> None: bind = op.get_bind() if _has_table(bind, JOBS): for handler, _display, _cron, _desc in _JOBS_TO_SEED: bind.execute( sa.text(f"DELETE FROM {JOBS} WHERE handler_key = :h"), {"h": handler}, ) if _has_table(bind, OCC): op.drop_table(OCC) if _has_table(bind, EXC): op.drop_table(EXC)