From 07b2016f33eefde14b6926e18567d357cc1783de Mon Sep 17 00:00:00 2001 From: publ Date: Tue, 23 Jun 2026 11:24:53 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20ETCD=20=E5=A4=87=E4=BB=BD=20+=20?= =?UTF-8?q?=E9=82=AE=E4=BB=B6/=E9=A3=9E=E4=B9=A6=E9=80=9A=E7=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- backend/Dockerfile | 11 + .../alembic/versions/0002_add_notify_on.py | 30 +++ backend/app/api/jobs.py | 6 + backend/app/api/router.py | 3 +- backend/app/api/settings.py | 154 +++++++++++ backend/app/core/backup/etcd.py | 117 ++++++++ backend/app/core/executor.py | 60 +++++ backend/app/core/notify.py | 217 +++++++++++++++ backend/app/models/job.py | 3 + backend/app/schemas/job.py | 18 +- frontend/src/api/settings.ts | 16 ++ frontend/src/pages/Jobs/JobForm.tsx | 74 +++++- frontend/src/pages/Jobs/JobList.tsx | 14 + frontend/src/pages/Settings.tsx | 250 +++++++++++++++++- frontend/src/types/index.ts | 37 ++- 15 files changed, 996 insertions(+), 14 deletions(-) create mode 100644 backend/alembic/versions/0002_add_notify_on.py create mode 100644 backend/app/api/settings.py create mode 100644 backend/app/core/backup/etcd.py create mode 100644 backend/app/core/notify.py create mode 100644 frontend/src/api/settings.ts diff --git a/backend/Dockerfile b/backend/Dockerfile index 9c6f2af..a7aa9d1 100644 --- a/backend/Dockerfile +++ b/backend/Dockerfile @@ -27,6 +27,17 @@ RUN set -eux; \ rm -rf /tmp/influx.tar.gz /tmp/influx; \ influx version +# 安装 etcdctl（直接下载二进制） +RUN set -eux; \ + ETCD_VERSION=v3.5.17; \ + curl -fsSL "https://github.com/etcd-io/etcd/releases/download/${ETCD_VERSION}/etcd-${ETCD_VERSION}-linux-amd64.tar.gz" \ + -o /tmp/etcd.tar.gz; \ + tar -xzf /tmp/etcd.tar.gz -C /tmp; \ + mv /tmp/etcd-${ETCD_VERSION}-linux-amd64/etcdctl /usr/local/bin/etcdctl; \ + chmod +x /usr/local/bin/etcdctl; \ + rm -rf /tmp/etcd.tar.gz /tmp/etcd-${ETCD_VERSION}-linux-amd64; \ + etcdctl version + WORKDIR /app # 先安装依赖以利用 Docker 缓存 diff --git a/backend/alembic/versions/0002_add_notify_on.py b/backend/alembic/versions/0002_add_notify_on.py new file mode 100644 index 0000000..63fce5e --- /dev/null +++ b/backend/alembic/versions/0002_add_notify_on.py @@ -0,0 +1,30 @@ +"""add notify_on to jobs + +Revision ID: 0002_add_notify_on +Revises: 0001_initial +Create Date: 2026-06-23 11:30:00 +""" +from alembic import op +import sqlalchemy as sa + + +revision = "0002_add_notify_on" +down_revision = "0001_initial" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column( + "jobs", + sa.Column( + "notify_on", + sa.String(16), + nullable=False, + server_default="none", + ), + ) + + +def downgrade() -> None: + op.drop_column("jobs", "notify_on") diff --git a/backend/app/api/jobs.py b/backend/app/api/jobs.py index 3df82ed..44ad08f 100644 --- a/backend/app/api/jobs.py +++ b/backend/app/api/jobs.py @@ -12,6 +12,7 @@ from app.models.run import Run from app.models.storage import Storage from app.schemas.job import ( DirectorySourceConfig, + EtcdSourceConfig, InfluxDBSourceConfigV1, InfluxDBSourceConfigV2, JobCreate, @@ -55,6 +56,8 @@ def _validate_source(type_: str, cfg: dict) -> dict: if ver == "2": return InfluxDBSourceConfigV2(**cfg).model_dump() raise HTTPException(status_code=400, detail="InfluxDB config 缺少 version (1 或 2)") + if type_ == "etcd": + return EtcdSourceConfig(**cfg).model_dump() raise HTTPException(status_code=400, detail=f"Unsupported job type: {type_}") @@ -96,6 +99,7 @@ def create_job(payload: JobCreate, db: DBSession, _: AdminUser) -> JobOut: storage_id=payload.storage_id, cron_expression=payload.cron_expression, enabled=payload.enabled, + notify_on=payload.notify_on, retention_count=payload.retention_count, retention_days=payload.retention_days, description=payload.description, @@ -147,6 +151,8 @@ def update_job(job_id: int, payload: JobUpdate, db: DBSession, _: AdminUser) -> job.cron_expression = payload.cron_expression or None if payload.enabled is not None: job.enabled = payload.enabled + if payload.notify_on is not None: + job.notify_on = payload.notify_on if payload.retention_count is not None: job.retention_count = payload.retention_count if payload.retention_days is not None: diff --git a/backend/app/api/router.py b/backend/app/api/router.py index 9b9018d..8f42fee 100644 --- a/backend/app/api/router.py +++ b/backend/app/api/router.py @@ -1,7 +1,7 @@ """API 总路由。""" from fastapi import APIRouter -from app.api import auth, jobs, restore, runs, storages +from app.api import auth, jobs, restore, runs, settings, storages api_router = APIRouter() api_router.include_router(auth.router) @@ -9,3 +9,4 @@ api_router.include_router(storages.router) api_router.include_router(jobs.router) api_router.include_router(runs.router) api_router.include_router(restore.router) +api_router.include_router(settings.router) diff --git a/backend/app/api/settings.py b/backend/app/api/settings.py new file mode 100644 index 0000000..cdf7e75 --- /dev/null +++ b/backend/app/api/settings.py @@ -0,0 +1,154 @@ +"""系统设置 API：通知配置（SMTP + Feishu webhook）。""" +from typing import Any + +from fastapi import APIRouter, HTTPException +from pydantic import BaseModel, EmailStr, Field + +from app.deps import AdminUser, DBSession +from app.models.setting import Setting +from app.utils.crypto import decrypt_dict, encrypt_dict + +router = APIRouter(prefix="/settings", tags=["settings"]) + +# Settings key +_NOTIFY_KEY = "notifications" + +# 脱敏字段 +_SENSITIVE = {"password", "feishu_webhook_url"} + + +def _redact(d: dict) -> dict: + return {k: ("***" if k in _SENSITIVE and v else v) for k, v in d.items()} + + +def _unredact(stored: dict, incoming: dict) -> dict: + """编辑时：若 incoming 中脱敏字段为 '***'，保留 stored 原值。""" + for k in _SENSITIVE: + if incoming.get(k) == "***" and k in stored: + incoming[k] = stored[k] + return incoming + + +def _load(db, key: str) -> dict: + row = db.get(Setting, key) + if not row: + return {} + try: + return decrypt_dict(row.value) + except Exception: + return {} + + +def _save(db, key: str, value: str) -> None: + row = db.get(Setting, key) + if row: + row.value = value + else: + db.add(Setting(key=key, value=value)) + db.commit() + + +class SmtpConfig(BaseModel): + host: str = "" + port: int = 587 + user: str = "" + password: str = "" + from_addr: str = "" + use_tls: bool = True + + +class NotificationsConfig(BaseModel): + email_enabled: bool = False + smtp: SmtpConfig = Field(default_factory=SmtpConfig) + email_to: list[str] = Field(default_factory=list) + feishu_enabled: bool = False + feishu_webhook_url: str = "" + + +class TestRequest(BaseModel): + channel: str = Field(..., pattern="^(email|feishu)$") + to: str | None = None # 仅 email 通道有效 + + +class TestResponse(BaseModel): + ok: bool + message: str + + +def _public_view(stored: dict) -> dict: + """返回给前端：密码脱敏。""" + out = dict(stored) + if "smtp" in out and isinstance(out["smtp"], dict): + out["smtp"] = _redact(out["smtp"]) + if _SENSITIVE.intersection(out.keys()): + out = _redact(out) + return out + + +@router.get("/notifications", response_model=dict) +def get_notifications(db: DBSession, _: AdminUser) -> dict: + stored = _load(db, _NOTIFY_KEY) + return _public_view(stored) + + +@router.put("/notifications", response_model=dict) +def put_notifications(payload: NotificationsConfig, db: DBSession, _: AdminUser) -> dict: + stored = _load(db, _NOTIFY_KEY) + incoming = payload.model_dump() + + # 处理 smtp.password 脱敏占位 + smtp_in = incoming.get("smtp") or {} + smtp_stored = stored.get("smtp") or {} + if smtp_in.get("password") == "***" and smtp_stored.get("password"): + smtp_in["password"] = smtp_stored["password"] + incoming["smtp"] = smtp_in + + # 处理 webhook_url 脱敏 + if incoming.get("feishu_webhook_url") == "***": + incoming["feishu_webhook_url"] = stored.get("feishu_webhook_url", "") + + _save(db, _NOTIFY_KEY, encrypt_dict(incoming)) + return _public_view(incoming) + + +@router.post("/notifications/test", response_model=TestResponse) +async def test_notification(payload: TestRequest, db: DBSession, _: AdminUser) -> TestResponse: + from app.core.notify import NotificationContext, send_email, send_feishu + + stored = _load(db, _NOTIFY_KEY) + if not stored: + raise HTTPException(status_code=400, detail="通知配置未设置") + + ctx = NotificationContext( + job_name="test-job", + job_type="mysql", + status="success", + run_id=0, + artifact_size=12345, + duration_seconds=3, + started_at="2026-06-22T10:00:00Z", + finished_at="2026-06-22T10:00:03Z", + ) + + if payload.channel == "email": + if not stored.get("email_enabled"): + raise HTTPException(status_code=400, detail="邮件通知未启用") + to_list = [payload.to] if payload.to else stored.get("email_to") or [] + ok, msg = await send_email( + stored.get("smtp") or {}, + to_list, + "[备份系统] 测试邮件", + "这是一封来自备份管理系统的测试邮件。", + ) + return TestResponse(ok=ok, message=msg) + else: # feishu + if not stored.get("feishu_enabled"): + raise HTTPException(status_code=400, detail="飞书通知未启用") + ok, msg = await send_feishu(stored.get("feishu_webhook_url", ""), ctx) + return TestResponse(ok=ok, message=msg) + + +@router.get("", response_model=list[dict[str, Any]]) +def list_all(db: DBSession, _: AdminUser) -> list[dict[str, Any]]: + rows = db.query(Setting).order_by(Setting.key).all() + return [{"key": r.key, "value": r.value[:200] + ("..." if len(r.value) > 200 else "")} for r in rows] diff --git a/backend/app/core/backup/etcd.py b/backend/app/core/backup/etcd.py new file mode 100644 index 0000000..de93d0a --- /dev/null +++ b/backend/app/core/backup/etcd.py @@ -0,0 +1,117 @@ +"""ETCD 备份：通过 etcdctl snapshot save 生成快照文件，再流式上传。 + +配置： +- endpoints：列表，如 ["http://10.0.0.1:2379", "http://10.0.0.2:2379"] +- username/password：可选 +- cacert/cert/key：可选 TLS +""" +import asyncio +import shutil +import tarfile +import tempfile +from datetime import datetime +from pathlib import Path +from typing import AsyncIterator + +from app.core.backup.base import BackupMetadata, BaseBackup +from app.utils.logging import get_logger + +log = get_logger(__name__) + + +class EtcdBackup(BaseBackup): + def __init__(self, source_config: dict, job_name: str): + super().__init__(source_config, job_name) + # endpoints 接收 string（逗号分隔）或 list + eps = source_config.get("endpoints", "") + if isinstance(eps, str): + self.endpoints: list[str] = [e.strip() for e in eps.split(",") if e.strip()] + else: + self.endpoints = list(eps) + if not self.endpoints: + raise ValueError("ETCD endpoints 不能为空") + self.username: str = source_config.get("username", "") + self.password: str = source_config.get("password", "") + self.cacert: str = source_config.get("cacert", "") + self.cert: str = source_config.get("cert", "") + self.key: str = source_config.get("key", "") + + def metadata(self) -> BackupMetadata: + ts = datetime.now().strftime("%Y%m%d-%H%M%S") + return BackupMetadata( + suggested_filename=f"{self.job_name}-etcd-{ts}.tar.gz", + content_type="application/gzip", + extra={ + "endpoints": self.endpoints, + "has_tls": bool(self.cacert or self.cert or self.key), + "has_auth": bool(self.username), + }, + ) + + async def produce(self) -> AsyncIterator[bytes]: + tmpdir = Path(tempfile.mkdtemp(prefix=f"etcdbk_{self.job_name}_")) + snap_path = tmpdir / "etcd.snap" + + try: + cmd = ["etcdctl", "snapshot", "save", str(snap_path)] + # 多次 --endpoints 让 etcdctl 选择健康节点 + for ep in self.endpoints: + cmd.extend(["--endpoints", ep]) + if self.username: + cmd.extend(["--user", f"{self.username}:{self.password}"]) + if self.cacert: + cmd.extend(["--cacert", self.cacert]) + if self.cert: + cmd.extend(["--cert", self.cert]) + if self.key: + cmd.extend(["--key", self.key]) + + log.info("Running: etcdctl snapshot save (endpoints=%s)", self.endpoints) + + proc = await asyncio.create_subprocess_exec( + *cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + stdout, stderr = await proc.communicate() + + if proc.returncode != 0: + raise RuntimeError( + f"etcdctl snapshot save failed (exit={proc.returncode}): " + f"{(stderr or stdout).decode(errors='replace')[:2000]}" + ) + + if not snap_path.exists() or snap_path.stat().st_size == 0: + raise RuntimeError("etcd snapshot 文件为空，请检查 ETCD 连接") + + # 写入 metadata 文件并 tar.gz 打包 + log.info("Snapshot size: %s bytes; packing...", snap_path.stat().st_size) + meta_path = tmpdir / "META.txt" + meta_path.write_text( + f"etcd_endpoints={','.join(self.endpoints)}\n" + f"backup_time={datetime.utcnow().isoformat()}Z\n" + f"snapshot_size={snap_path.stat().st_size}\n" + f"restore_hint=etcdctl snapshot restore " + f"--name --initial-cluster --initial-advertise-peer-urls \n", + encoding="utf-8", + ) + + tar_path = tmpdir / "backup.tar.gz" + loop = asyncio.get_event_loop() + + def _tar() -> None: + with tarfile.open(str(tar_path), mode="w:gz") as tar: + tar.add(str(snap_path), arcname="etcd.snap") + tar.add(str(meta_path), arcname="META.txt") + + await loop.run_in_executor(None, _tar) + + # 流式产出 + with open(tar_path, "rb") as f: + while True: + chunk = await loop.run_in_executor(None, f.read, 64 * 1024) + if not chunk: + break + yield chunk + finally: + shutil.rmtree(tmpdir, ignore_errors=True) diff --git a/backend/app/core/executor.py b/backend/app/core/executor.py index 7f4516f..34a2fdf 100644 --- a/backend/app/core/executor.py +++ b/backend/app/core/executor.py @@ -14,6 +14,7 @@ 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 @@ -35,6 +36,8 @@ def _make_backup(job: Job) -> BaseBackup: 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}") @@ -127,6 +130,63 @@ async def run_backup(run_id: int, trigger: str = "manual") -> None: 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() diff --git a/backend/app/core/notify.py b/backend/app/core/notify.py new file mode 100644 index 0000000..5fa0f32 --- /dev/null +++ b/backend/app/core/notify.py @@ -0,0 +1,217 @@ +"""通知模块：邮件 (SMTP) + 飞书群机器人 (Webhook)。 + +设计：纯函数式 + 异步，调用方传入配置 dict 与上下文。 +不依赖数据库或 settings 表（settings 读取由调用方负责）。 +""" +import asyncio +import smtplib +from dataclasses import dataclass +from email.mime.multipart import MIMEMultipart +from email.mime.text import MIMEText +from typing import Iterable, Optional + +import httpx + +from app.utils.logging import get_logger + +log = get_logger(__name__) + + +@dataclass +class NotificationContext: + job_name: str + job_type: str + status: str # "success" / "failed" + run_id: int + artifact_size: Optional[int] = None + duration_seconds: Optional[int] = None + error_message: Optional[str] = None + artifact_path: Optional[str] = None + started_at: Optional[str] = None + finished_at: Optional[str] = None + + +def _human_size(n: Optional[int]) -> str: + if n is None: + return "-" + if n < 1024: + return f"{n} B" + if n < 1024 * 1024: + return f"{n / 1024:.1f} KB" + if n < 1024 * 1024 * 1024: + return f"{n / 1024 / 1024:.1f} MB" + return f"{n / 1024 / 1024 / 1024:.2f} GB" + + +def _build_text(ctx: NotificationContext) -> str: + lines = [ + f"[备份系统] {ctx.job_name}", + f"状态: {ctx.status.upper()}", + f"类型: {ctx.job_type}", + f"Run ID: {ctx.run_id}", + ] + if ctx.started_at: + lines.append(f"开始: {ctx.started_at}") + if ctx.finished_at: + lines.append(f"结束: {ctx.finished_at}") + if ctx.duration_seconds is not None: + lines.append(f"耗时: {ctx.duration_seconds}s") + if ctx.artifact_size is not None: + lines.append(f"大小: {_human_size(ctx.artifact_size)}") + if ctx.artifact_path: + lines.append(f"路径: {ctx.artifact_path}") + if ctx.error_message: + lines.append(f"错误: {ctx.error_message[:1000]}") + return "\n".join(lines) + + +def _build_feishu_card(ctx: NotificationContext) -> dict: + color = "green" if ctx.status == "success" else "red" + fields = [ + {"is_short": True, "text": {"tag": "lark_md", "content": f"**类型**\n{ctx.job_type}"}}, + {"is_short": True, "text": {"tag": "lark_md", "content": f"**Run ID**\n{ctx.run_id}"}}, + ] + if ctx.duration_seconds is not None: + fields.append({"is_short": True, "text": {"tag": "lark_md", "content": f"**耗时**\n{ctx.duration_seconds}s"}}) + if ctx.artifact_size is not None: + fields.append({"is_short": True, "text": {"tag": "lark_md", "content": f"**大小**\n{_human_size(ctx.artifact_size)}"}}) + if ctx.error_message: + fields.append({"is_short": False, "text": {"tag": "lark_md", "content": f"**错误**\n{ctx.error_message[:1000]}"}}) + + return { + "msg_type": "interactive", + "card": { + "config": {"wide_screen_mode": True}, + "header": { + "title": {"tag": "plain_text", "content": f"[备份] {ctx.job_name} — {ctx.status.upper()}"}, + "template": color, + }, + "elements": [ + {"tag": "div", "fields": fields}, + {"tag": "hr"}, + { + "tag": "note", + "elements": [ + {"tag": "plain_text", "content": f"Backup System • {ctx.started_at or ''}"} + ], + }, + ], + }, + } + + +# ===== Email ===== + +async def send_email( + smtp_config: dict, + to_addrs: Iterable[str], + subject: str, + body: str, +) -> tuple[bool, str]: + """通过 SMTP 发送邮件。smtp_config 必填字段： + host, port, user, password, from_addr, use_tls (bool) + """ + host = smtp_config.get("host") + if not host: + return False, "SMTP host 未配置" + port = int(smtp_config.get("port") or 587) + user = smtp_config.get("user", "") + password = smtp_config.get("password", "") + from_addr = smtp_config.get("from_addr") or user + use_tls = bool(smtp_config.get("use_tls", True)) + + to_list = [a.strip() for a in to_addrs if a and a.strip()] + if not to_list: + return False, "收件人列表为空" + + msg = MIMEMultipart() + msg["From"] = from_addr + msg["To"] = ", ".join(to_list) + msg["Subject"] = subject + msg.attach(MIMEText(body, "plain", "utf-8")) + + def _send() -> tuple[bool, str]: + try: + if use_tls: + client = smtplib.SMTP(host, port, timeout=15) + client.ehlo() + client.starttls() + client.ehlo() + else: + client = smtplib.SMTP_SSL(host, port, timeout=15) + client.ehlo() + if user: + client.login(user, password) + client.sendmail(from_addr, to_list, msg.as_string()) + client.quit() + return True, f"sent to {len(to_list)} recipient(s)" + except Exception as e: + return False, f"{type(e).__name__}: {e}" + + ok, msg_text = await asyncio.get_event_loop().run_in_executor(None, _send) + return ok, msg_text + + +# ===== Feishu Webhook ===== + +async def send_feishu(webhook_url: str, ctx: NotificationContext) -> tuple[bool, str]: + """飞书自定义机器人：发送 interactive card。""" + if not webhook_url: + return False, "webhook URL 未配置" + payload = _build_feishu_card(ctx) + try: + async with httpx.AsyncClient(timeout=15) as client: + resp = await client.post(webhook_url, json=payload) + if resp.status_code != 200: + return False, f"HTTP {resp.status_code}: {resp.text[:300]}" + data = resp.json() + if data.get("code") not in (0, None): + return False, f"feishu err: {data.get('msg') or data}" + return True, "ok" + except Exception as e: + return False, f"{type(e).__name__}: {e}" + + +# ===== Aggregator ===== + +@dataclass +class NotifyResult: + email_ok: bool = True + email_msg: str = "" + feishu_ok: bool = True + feishu_msg: str = "" + + +async def send_notifications( + notify_settings: dict, + ctx: NotificationContext, +) -> NotifyResult: + """根据 settings 发送通知。settings 结构： + + { + "email_enabled": true, + "smtp": {"host", "port", "user", "password", "from_addr", "use_tls"}, + "email_to": ["ops@example.com", ...], + "feishu_enabled": true, + "feishu_webhook_url": "https://open.feishu.cn/open-apis/bot/v2/hook/...", + } + """ + res = NotifyResult() + + if notify_settings.get("email_enabled"): + smtp = notify_settings.get("smtp") or {} + to_list = notify_settings.get("email_to") or [] + subject = f"[{'成功' if ctx.status == 'success' else '失败'}] 备份任务 {ctx.job_name}" + ok, msg = await send_email(smtp, to_list, subject, _build_text(ctx)) + res.email_ok = ok + res.email_msg = msg + log.info("Email notify: ok=%s msg=%s", ok, msg) + + if notify_settings.get("feishu_enabled"): + url = notify_settings.get("feishu_webhook_url", "") + ok, msg = await send_feishu(url, ctx) + res.feishu_ok = ok + res.feishu_msg = msg + log.info("Feishu notify: ok=%s msg=%s", ok, msg) + + return res diff --git a/backend/app/models/job.py b/backend/app/models/job.py index e9091f3..3adde36 100644 --- a/backend/app/models/job.py +++ b/backend/app/models/job.py @@ -25,6 +25,9 @@ class Job(Base): ) cron_expression: Mapped[str | None] = mapped_column(String(64), nullable=True) enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) + notify_on: Mapped[str] = mapped_column( + String(16), nullable=False, default="none" + ) # 'none' / 'failure' / 'all' retention_count: Mapped[int] = mapped_column(Integer, nullable=False, default=7) retention_days: Mapped[int | None] = mapped_column(Integer, nullable=True) description: Mapped[str | None] = mapped_column(String(255), nullable=True) diff --git a/backend/app/schemas/job.py b/backend/app/schemas/job.py index 86f0ba4..0224f39 100644 --- a/backend/app/schemas/job.py +++ b/backend/app/schemas/job.py @@ -41,13 +41,27 @@ class InfluxDBSourceConfigV2(BaseModel): token: str = Field(..., min_length=1) +class EtcdSourceConfig(BaseModel): + endpoints: str = Field( + ..., + min_length=1, + description="ETCD endpoints，逗号分隔，如 http://10.0.0.1:2379,http://10.0.0.2:2379", + ) + username: str = Field("", description="可选 basic auth 用户名") + password: str = Field("") + cacert: str = Field("", description="CA 证书路径（容器内）") + cert: str = Field("", description="客户端证书路径") + key: str = Field("", description="客户端私钥路径") + + class JobBase(BaseModel): name: str = Field(..., min_length=1, max_length=64) - type: Literal["mysql", "directory", "influxdb"] + type: Literal["mysql", "directory", "influxdb", "etcd"] source_config: dict storage_id: int cron_expression: Optional[str] = Field(None, description="标准 5 段 cron 表达式，空表示手动") enabled: bool = True + notify_on: Literal["none", "failure", "all"] = Field("none", description="何时发通知") retention_count: int = Field(7, ge=1, le=365) retention_days: Optional[int] = Field(None, ge=1, le=3650) description: Optional[str] = Field(None, max_length=255) @@ -74,6 +88,7 @@ class JobUpdate(BaseModel): storage_id: Optional[int] = None cron_expression: Optional[str] = None enabled: Optional[bool] = None + notify_on: Optional[Literal["none", "failure", "all"]] = None retention_count: Optional[int] = Field(None, ge=1, le=365) retention_days: Optional[int] = Field(None, ge=1, le=3650) description: Optional[str] = None @@ -86,6 +101,7 @@ class JobOut(BaseModel): type: str cron_expression: Optional[str] enabled: bool + notify_on: str retention_count: int retention_days: Optional[int] description: Optional[str] diff --git a/frontend/src/api/settings.ts b/frontend/src/api/settings.ts new file mode 100644 index 0000000..96c7035 --- /dev/null +++ b/frontend/src/api/settings.ts @@ -0,0 +1,16 @@ +import { api } from './client' +import type { NotificationsConfig } from '../types' + +export const settingsApi = { + getNotifications: () => + api.get('/settings/notifications').then((r) => r.data), + updateNotifications: (payload: NotificationsConfig) => + api.put('/settings/notifications', payload).then((r) => r.data), + test: (channel: 'email' | 'feishu', to?: string) => + api + .post<{ ok: boolean; message: string }>('/settings/notifications/test', { + channel, + to, + }) + .then((r) => r.data), +} diff --git a/frontend/src/pages/Jobs/JobForm.tsx b/frontend/src/pages/Jobs/JobForm.tsx index 98ea05e..510166c 100644 --- a/frontend/src/pages/Jobs/JobForm.tsx +++ b/frontend/src/pages/Jobs/JobForm.tsx @@ -22,10 +22,11 @@ const { Title } = Typography interface FormValues { name: string - type: 'mysql' | 'directory' | 'influxdb' + type: 'mysql' | 'directory' | 'influxdb' | 'etcd' storage_id: number enabled: boolean cron_expression?: string + notify_on?: 'none' | 'failure' | 'all' retention_count: number retention_days?: number description?: string @@ -50,6 +51,13 @@ interface FormValues { influx_org?: string influx_bucket?: string influx_token?: string + // ETCD + etcd_endpoints?: string + etcd_username?: string + etcd_password?: string + etcd_cacert?: string + etcd_cert?: string + etcd_key?: string } function toPayload(values: FormValues): JobCreate { @@ -71,8 +79,7 @@ function toPayload(values: FormValues): JobCreate { ? values.dir_exclude_patterns.split(',').map((s) => s.trim()).filter(Boolean) : [], } - } else { - // influxdb + } else if (values.type === 'influxdb') { if (values.influx_version === '1') { source_config = { version: '1', @@ -92,6 +99,16 @@ function toPayload(values: FormValues): JobCreate { token: values.influx_token!, } } + } else { + // etcd + source_config = { + endpoints: values.etcd_endpoints!, + username: values.etcd_username || '', + password: values.etcd_password || '', + cacert: values.etcd_cacert || '', + cert: values.etcd_cert || '', + key: values.etcd_key || '', + } } return { @@ -100,6 +117,7 @@ function toPayload(values: FormValues): JobCreate { storage_id: values.storage_id, source_config, enabled: values.enabled, + notify_on: values.notify_on || 'none', cron_expression: values.cron_expression?.trim() || undefined, retention_count: values.retention_count, retention_days: values.retention_days, @@ -167,6 +185,16 @@ export function JobForm({ mode }: { mode: 'create' | 'edit' }) { influx_bucket: cfg.bucket, influx_token: cfg.token, } + } else if (j.type === 'etcd') { + flat = { + ...flat, + etcd_endpoints: cfg.endpoints, + etcd_username: cfg.username, + etcd_password: cfg.password, + etcd_cacert: cfg.cacert, + etcd_cert: cfg.cert, + etcd_key: cfg.key, + } } form.setFieldsValue(flat as FormValues) }) @@ -204,6 +232,7 @@ export function JobForm({ mode }: { mode: 'create' | 'edit' }) { initialValues={{ type: 'mysql', enabled: true, + notify_on: 'none', retention_count: 7, mysql_port: 3306, }} @@ -224,6 +253,7 @@ export function JobForm({ mode }: { mode: 'create' | 'edit' }) { { value: 'mysql', label: 'MySQL 数据库' }, { value: 'directory', label: '服务器目录' }, { value: 'influxdb', label: 'InfluxDB 数据库' }, + { value: 'etcd', label: 'ETCD' }, ]} /> @@ -326,6 +356,30 @@ export function JobForm({ mode }: { mode: 'create' | 'edit' }) { )} + {jobType === 'etcd' && ( + <> + ETCD 数据源 + + + + + + + + + + + + + + + + + + + + )} + 存储与调度 @@ -347,6 +401,20 @@ export function JobForm({ mode }: { mode: 'create' | 'edit' }) { + + + + + + + + + + + + + + + + + + + + + + + + + + ) : null + } + + + ), + }, + { + key: 'feishu', + label: '飞书群机器人', + children: ( + <> + + + + + {() => + ((notifyForm.getFieldValue('feishu_enabled') as boolean) || false) ? ( + <> + + + + + + ) : null + } + + + ), + }, + ]} + /> + + + + + + ) } diff --git a/frontend/src/types/index.ts b/frontend/src/types/index.ts index b56161e..59aa40a 100644 --- a/frontend/src/types/index.ts +++ b/frontend/src/types/index.ts @@ -97,6 +97,32 @@ export interface InfluxDBSourceConfigV2 { export type InfluxDBSourceConfig = InfluxDBSourceConfigV1 | InfluxDBSourceConfigV2 +export interface EtcdSourceConfig { + endpoints: string + username?: string + password?: string + cacert?: string + cert?: string + key?: string +} + +export interface SmtpConfig { + host: string + port: number + user: string + password?: string + from_addr: string + use_tls: boolean +} + +export interface NotificationsConfig { + email_enabled: boolean + smtp: SmtpConfig + email_to: string[] + feishu_enabled: boolean + feishu_webhook_url: string +} + export interface RunSummary { id: number status: string @@ -111,9 +137,10 @@ export interface RunSummary { export interface JobOut { id: number name: string - type: 'mysql' | 'directory' | 'influxdb' + type: 'mysql' | 'directory' | 'influxdb' | 'etcd' cron_expression: string | null enabled: boolean + notify_on: 'none' | 'failure' | 'all' retention_count: number retention_days: number | null description: string | null @@ -129,11 +156,12 @@ export interface JobWithLastRun extends JobOut { export interface JobCreate { name: string - type: 'mysql' | 'directory' | 'influxdb' - source_config: MySQLSourceConfig | DirectorySourceConfig | InfluxDBSourceConfig + type: 'mysql' | 'directory' | 'influxdb' | 'etcd' + source_config: MySQLSourceConfig | DirectorySourceConfig | InfluxDBSourceConfig | EtcdSourceConfig storage_id: number cron_expression?: string enabled?: boolean + notify_on?: 'none' | 'failure' | 'all' retention_count?: number retention_days?: number description?: string @@ -141,10 +169,11 @@ export interface JobCreate { export interface JobUpdate { name?: string - source_config?: MySQLSourceConfig | DirectorySourceConfig | InfluxDBSourceConfig + source_config?: MySQLSourceConfig | DirectorySourceConfig | InfluxDBSourceConfig | EtcdSourceConfig storage_id?: number cron_expression?: string | null enabled?: boolean + notify_on?: 'none' | 'failure' | 'all' retention_count?: number retention_days?: number | null description?: string | null