publ
07b2016f33
ETCD 备份（新增类型）： - 后端 core/backup/etcd.py：调用 etcdctl snapshot save（v3.5.17 二进制） 生成快照文件，与 META.txt 一起 tar.gz 流式打包上传。 - 配置支持 endpoints（逗号分隔多节点）、basic auth、TLS（cacert/cert/key）。 - Dockerfile 下载 etcdctl 二进制（绕过 apt 签名问题）。 通知模块： - 核心 app/core/notify.py： - 邮件：smtplib + MIMEMultipart（线程池里跑避免阻塞） - 飞书：httpx 调自定义机器人 webhook，发 interactive card - settings API： - GET/PUT /api/settings/notifications（密码脱敏 *** 编辑占位） - POST /api/settings/notifications/test 测试通道 - Job 增加 notify_on 字段（none/failure/all），Alembic 0002 迁移。 - executor 在 run 完成后根据 job.notify_on 与 run.status 异步触发通知， 重新加载最新 run 信息生成上下文。 前端： - JobForm 加 ETCD 表单字段与 notify_on 选择器；type 选项加 etcd。 - JobList 显示通知策略徽标。 - Settings 页加通知配置卡片：邮件（折叠面板 + 测试按钮）+ 飞书 + 保存。 - 新增 api/settings.ts。 - 类型扩展：EtcdSourceConfig、NotificationsConfig、SmtpConfig。 Co-Authored-By: Claude <noreply@anthropic.com>
215 lines
7.4 KiB
Python
215 lines
7.4 KiB
Python
"""备份执行编排：解密配置 → 流式产出 → 流式上传 → 应用保留策略 → 写日志。"""
|
||
import asyncio
|
||
import hashlib
|
||
import io
|
||
import logging
|
||
import os
|
||
import tempfile
|
||
import time
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Optional
|
||
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.backup.base import BaseBackup
|
||
from app.core.backup.directory import DirectoryBackup
|
||
from app.core.backup.etcd import EtcdBackup
|
||
from app.core.backup.influxdb import InfluxDBBackup
|
||
from app.core.backup.mysql import MySQLBackup
|
||
from app.core.retention import apply_retention
|
||
from app.core.storage.factory import create_storage
|
||
from app.database import SessionLocal
|
||
from app.models.job import Job
|
||
from app.models.run import Run, RunLog
|
||
from app.utils.crypto import decrypt_dict
|
||
from app.utils.logging import get_logger
|
||
|
||
log = get_logger(__name__)
|
||
|
||
|
||
def _make_backup(job: Job) -> BaseBackup:
|
||
src = decrypt_dict(job.source_config_json)
|
||
if job.type == "mysql":
|
||
return MySQLBackup(src, job.name)
|
||
if job.type == "directory":
|
||
return DirectoryBackup(src, job.name)
|
||
if job.type == "influxdb":
|
||
return InfluxDBBackup(src, job.name)
|
||
if job.type == "etcd":
|
||
return EtcdBackup(src, job.name)
|
||
raise ValueError(f"Unsupported job type: {job.type}")
|
||
|
||
|
||
async def run_backup(run_id: int, trigger: str = "manual") -> None:
|
||
"""执行一次备份。所有副作用都在自己 new 的 Session 上完成。"""
|
||
started = datetime.utcnow()
|
||
db: Session = SessionLocal()
|
||
try:
|
||
run: Run | None = db.get(Run, run_id)
|
||
if not run:
|
||
log.error("run_id=%s not found", run_id)
|
||
return
|
||
|
||
job: Job | None = db.get(Job, run.job_id)
|
||
if not job:
|
||
run.status = "failed"
|
||
run.error_message = f"Job id={run.job_id} disappeared"
|
||
run.finished_at = datetime.utcnow()
|
||
db.commit()
|
||
return
|
||
|
||
log_buffer = io.StringIO()
|
||
tee = _TeeLogger(log, log_buffer)
|
||
|
||
run.status = "running"
|
||
run.trigger = trigger
|
||
run.started_at = started
|
||
db.commit()
|
||
|
||
tee.info("Backup started: run_id=%s job=%s trigger=%s", run_id, job.name, trigger)
|
||
|
||
try:
|
||
backup = _make_backup(job)
|
||
meta = backup.metadata()
|
||
tee.info("Metadata: filename=%s content_type=%s", meta.suggested_filename, meta.content_type)
|
||
|
||
storage = create_storage(job.storage)
|
||
ok, msg = await storage.test_connection()
|
||
tee.info("Storage test: ok=%s msg=%s", ok, msg)
|
||
if not ok:
|
||
raise RuntimeError(f"Storage not reachable: {msg}")
|
||
|
||
# 关键路径：备份文件名以 {job_name}/{YYYY-MM-DD}/{filename} 组织
|
||
date_prefix = datetime.utcnow().strftime("%Y-%m-%d")
|
||
object_key = f"{job.name}/{date_prefix}/{meta.suggested_filename}"
|
||
|
||
tee.info("Streaming backup to storage key=%s ...", object_key)
|
||
|
||
async def _consume_with_progress():
|
||
async for chunk in backup.produce():
|
||
yield chunk
|
||
|
||
t0 = time.monotonic()
|
||
size, sha = await storage.upload_stream(
|
||
_consume_with_progress(),
|
||
object_key,
|
||
content_type=meta.content_type,
|
||
)
|
||
elapsed = int(time.monotonic() - t0)
|
||
|
||
run.artifact_path = object_key
|
||
run.artifact_size = size
|
||
run.artifact_checksum = sha
|
||
run.duration_seconds = elapsed
|
||
run.status = "success"
|
||
run.finished_at = datetime.utcnow()
|
||
tee.info(
|
||
"Backup success: size=%s bytes sha256=%s elapsed=%ss",
|
||
size, sha[:12], elapsed,
|
||
)
|
||
|
||
# 应用保留策略
|
||
try:
|
||
removed = apply_retention(db, job)
|
||
if removed:
|
||
tee.info("Retention cleanup removed %s old runs", removed)
|
||
except Exception as e:
|
||
tee.warning("Retention cleanup failed: %s", e)
|
||
|
||
except Exception as e:
|
||
log.exception("Backup run_id=%s failed", run_id)
|
||
tee.error("Backup failed: %s: %s", type(e).__name__, e)
|
||
run.status = "failed"
|
||
run.error_message = f"{type(e).__name__}: {e}"[:2000]
|
||
run.finished_at = datetime.utcnow()
|
||
finally:
|
||
log_text = log_buffer.getvalue()
|
||
run.log_excerpt = log_text[-4000:] if len(log_text) > 4000 else log_text
|
||
run_log = RunLog(run_id=run.id, content=log_text)
|
||
db.add(run_log)
|
||
db.commit()
|
||
log.info("Backup run_id=%s finished status=%s", run_id, run.status)
|
||
|
||
# 触发通知（在事务外，避免阻塞 executor；fire and forget）
|
||
try:
|
||
should_notify = (
|
||
job.notify_on == "all"
|
||
or (job.notify_on == "failure" and run.status == "failed")
|
||
)
|
||
if should_notify:
|
||
asyncio.create_task(_dispatch_notification(job.id, run.id))
|
||
except Exception as e:
|
||
log.warning("Failed to schedule notification: %s", e)
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
async def _dispatch_notification(job_id: int, run_id: int) -> None:
|
||
"""从 DB 重新加载最新 run 状态后发通知。"""
|
||
from app.core.notify import NotificationContext, send_notifications
|
||
from app.models.setting import Setting
|
||
from app.utils.crypto import decrypt_dict
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
job = db.get(Job, job_id)
|
||
run = db.get(Run, run_id)
|
||
if not job or not run:
|
||
return
|
||
|
||
notify_row = db.get(Setting, "notifications")
|
||
if not notify_row or not notify_row.value:
|
||
return
|
||
try:
|
||
settings = decrypt_dict(notify_row.value)
|
||
except Exception:
|
||
return
|
||
|
||
if not (settings.get("email_enabled") or settings.get("feishu_enabled")):
|
||
return
|
||
|
||
ctx = NotificationContext(
|
||
job_name=job.name,
|
||
job_type=job.type,
|
||
status=run.status,
|
||
run_id=run.id,
|
||
artifact_size=run.artifact_size,
|
||
duration_seconds=run.duration_seconds,
|
||
error_message=run.error_message,
|
||
artifact_path=run.artifact_path,
|
||
started_at=run.started_at.isoformat() if run.started_at else None,
|
||
finished_at=run.finished_at.isoformat() if run.finished_at else None,
|
||
)
|
||
|
||
result = await send_notifications(settings, ctx)
|
||
log.info(
|
||
"Notification dispatched for run_id=%s: email_ok=%s feishu_ok=%s",
|
||
run.id, result.email_ok, result.feishu_ok,
|
||
)
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
class _TeeLogger:
|
||
"""同时写标准 logging 和内存 StringIO（用于存数据库）。"""
|
||
|
||
def __init__(self, real: logging.Logger, buf: io.StringIO):
|
||
self._real = real
|
||
self._buf = buf
|
||
|
||
def _emit(self, level: int, msg: str) -> None:
|
||
ts = datetime.utcnow().strftime("%H:%M:%S")
|
||
line = f"[{ts}] {msg}"
|
||
self._real.log(level, line)
|
||
self._buf.write(line + "\n")
|
||
|
||
def info(self, msg: str, *args) -> None:
|
||
self._emit(logging.INFO, msg % args if args else msg)
|
||
|
||
def warning(self, msg: str, *args) -> None:
|
||
self._emit(logging.WARNING, msg % args if args else msg)
|
||
|
||
def error(self, msg: str, *args) -> None:
|
||
self._emit(logging.ERROR, msg % args if args else msg)
|