From 203b076783118817536f59127bd8117b4579405f Mon Sep 17 00:00:00 2001 From: ericwyuan Date: Thu, 20 Aug 2026 14:59:50 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20NAS=E7=AB=AFFFmpeg=E9=A2=84=E5=8E=8B?= =?UTF-8?q?=E7=BC=A9=20=E2=80=94=20=E6=A0=B9=E6=B2=BB360MB=E8=A7=86?= =?UTF-8?q?=E9=A2=91=E8=B7=A8=E5=A2=83=E4=B8=8A=E4=BC=A0=E8=B6=85=E6=97=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 根因诊断（网络问题，非代码逻辑）: - NAS(中国移动) → Oracle(美国Phoenix) 跨境上行带宽实测仅 0.5-0.9MB/s 且波动大 - 下载方向 3.0MB/s（非对称，典型国际出口拥塞） - Tailscale P2P: ping 可通(224ms)但持续数据流被阻断(0 B/s, 127s超时) - 360MB 原始视频上传需 10-20min，任何大分块都会超时 NAS dispatcher: - 新增 _compress_video: 480p/CRF28/veryfast，静态监控场景实测 ~36x 压缩比 (360MB → ~12MB，上传时间 20min → ~30s) - ffmpeg 自动探测: CodecPack ffmpeg41(带libx264) > /usr/local/bin > PATH (Synology 系统 ffmpeg 被裁剪，无 h264 编解码) - scale 两级滤镜: force_original_aspect_ratio=decrease + trunc(iw/2)*2 (h264 要求偶数尺寸，853x480 会报错) - 压缩缓存 /tmp/fam_compressed/task_{id}/，源文件未变则重试复用，TTL 24h - 压缩完成后重置 PROCESSING 状态（重置 stale 回收计时基准） - 压缩失败回退原始文件分块上传 - _query_uploaded_chunks 失败时记录日志（原静默失败导致全量重传无感知） Edge api_gateway: - fix: total_chunks 变更清理旧分块后同步重写 meta.json (原 bug: meta.json 不更新导致每块上传都触发清理，删除同批新分块死循环) config: - stale_timeout 600→1800（覆盖压缩+上传+Edge队列积压总时长） - 新增 ffmpeg_path / compress_timeout 配置项 --- fam-core/config/config.yaml.example | 8 +- .../src/fam_core/dispatcher/dispatcher.py | 140 ++++++++++++++++-- .../src/fam_edge/api_gateway/api_gateway.py | 9 +- 3 files changed, 139 insertions(+), 18 deletions(-) diff --git a/fam-core/config/config.yaml.example b/fam-core/config/config.yaml.example index 9ea3182..856f012 100644 --- a/fam-core/config/config.yaml.example +++ b/fam-core/config/config.yaml.example @@ -21,9 +21,11 @@ scheduler: dispatcher: poll_interval: 30 # 轮询间隔（秒） - edge_url: "http://100.x.x.20:5000/api/edge/video/push" # 推送模式端点（视频上传，同步返回结果） - max_retries: 3 - push_timeout: 1800 # 推送+分析同步超时（秒） + edge_url: "http://100.x.x.20:5000/api/edge/video/enqueue" # 异步队列端点（上传入队，Poller 拉取结果） + max_retries: 5 # 文件级重试次数（分块级重试另计，每块3次） + stale_timeout: 1800 # PROCESSING 僵尸回收（秒），需大于压缩+上传+Edge队列积压总时长 + # ffmpeg_path: "/var/packages/CodecPack/target/bin/ffmpeg41" # 可选，默认自动探测（Synology 需 CodecPack 版，系统版无 h264 编码） + compress_timeout: 3600 # 单个视频预压缩超时（秒） video_server: base_url: "http://100.x.x.10:8000/media" diff --git a/fam-core/src/fam_core/dispatcher/dispatcher.py b/fam-core/src/fam_core/dispatcher/dispatcher.py index 917e7f2..650fe23 100644 --- a/fam-core/src/fam_core/dispatcher/dispatcher.py +++ b/fam-core/src/fam_core/dispatcher/dispatcher.py @@ -1,13 +1,15 @@ """ Dispatcher - 30s 轮询 PENDING 任务，上传视频至 Edge 异步队列 -流程（异步队列模式 + 分块断点续传）: +流程（异步队列模式 + NAS 预压缩 + 分块断点续传）: 1. 读取任务对应的本地视频文件 -2a. 小文件 (<=50MB): 直接 multipart 上传至 /enqueue -2b. 大文件 (>50MB): 分块上传 (20MB/块) 至 /chunk，支持断点续传，最后调 /assemble 合并入队 -3. Edge 保存视频 + 入 SQLite 队列，返回 202 -4. Dispatcher 标记任务为 PROCESSING（已派发，等待 Poller 拉取结果） -5. Poller 线程定期从 Edge /api/edge/results 拉取结果，写库后标记 SUCCESS +2. 大文件 (>20MB): NAS 端 FFmpeg 预压缩 (480p/CRF28, ~36x 压缩比) + （根因: NAS→Oracle 跨境上行带宽仅 ~0.5-0.9MB/s，360MB 原始上传需 10-20min 且频繁超时） +3a. 压缩后小文件 (<=20MB): 直接 multipart 上传至 /enqueue +3b. 仍超阈值: 分块上传 (5MB/块) 至 /chunk，支持断点续传，最后调 /assemble 合并入队 +4. Edge 保存视频 + 入 SQLite 队列，返回 202 +5. Dispatcher 标记任务为 PROCESSING（已派发，等待 Poller 拉取结果） +6. Poller 线程定期从 Edge /api/edge/results 拉取结果，写库后标记 SUCCESS 退避重试: min(30 * (retry_count + 1), 300) 秒 分块级重试: 每块最多重试 3 次 @@ -17,6 +19,8 @@ import os import io import time import math +import shutil +import subprocess import threading import requests from datetime import datetime, timedelta @@ -29,8 +33,17 @@ logger = setup_logger('fam-core.dispatcher') CHUNK_SIZE = 5 * 1024 * 1024 # 5MB per chunk (reliable at ~1Mbps upload) CHUNK_THRESHOLD = 20 * 1024 * 1024 # files > 20MB use chunked upload +COMPRESS_THRESHOLD = 20 * 1024 * 1024 # files > 20MB get pre-compressed before upload +COMPRESS_DIR = '/tmp/fam_compressed' +COMPRESS_CACHE_TTL = 24 * 3600 # 压缩缓存保留 24h（供上传失败重试复用） MAX_CHUNK_RETRIES = 3 +# Synology 系统 ffmpeg 被裁剪（无 h264 编解码），CodecPack 的 ffmpeg41 带 libx264 +FFMPEG_CANDIDATES = [ + '/var/packages/CodecPack/target/bin/ffmpeg41', + '/usr/local/bin/ffmpeg', +] + class Dispatcher: """任务下发器，30s 轮询（异步队列 + 分块断点续传）""" @@ -43,6 +56,8 @@ class Dispatcher: self.max_retries = cfg.get('dispatcher', {}).get('max_retries', 3) self.camera_name = cfg.get('scheduler', {}).get('camera_name', '默认摄像头') self.stale_timeout = cfg.get('dispatcher', {}).get('stale_timeout', 600) + self.compress_timeout = cfg.get('dispatcher', {}).get('compress_timeout', 3600) + self.ffmpeg = self._find_ffmpeg(cfg) # 推导 Edge base URL self.edge_base = self.edge_url.rsplit('/api/edge/video/enqueue', 1)[0] self.chunk_url = f"{self.edge_base}/api/edge/video/chunk" @@ -51,6 +66,16 @@ class Dispatcher: self._running = False self._thread = None + @staticmethod + def _find_ffmpeg(cfg): + """查找可用 ffmpeg：配置优先，其次 CodecPack（带 libx264），最后 PATH""" + configured = cfg.get('dispatcher', {}).get('ffmpeg_path') + candidates = ([configured] if configured else []) + FFMPEG_CANDIDATES + for path in candidates: + if os.path.isfile(path) and os.access(path, os.X_OK): + return path + return shutil.which('ffmpeg') + def _calculate_backoff(self, retry_count): """退避策略: min(30 * (retry_count + 1), 300)""" return min(30 * (retry_count + 1), 300) @@ -83,8 +108,87 @@ class Dispatcher: pass return payload + def _compress_video(self, task_id, video_path): + """NAS 端预压缩: 480p/CRF28/veryfast，静态监控场景实测 ~36x 压缩比 + + 成功返回压缩文件路径；失败返回 None（回退原始文件分块上传）。 + 压缩产物缓存在 /tmp/fam_compressed/task_{id}/，重试时源文件未变则复用。 + """ + if not self.ffmpeg: + logger.warning(f"[task_id={task_id}] ffmpeg 不可用，跳过预压缩") + return None + + self._cleanup_compress_cache() + + out_dir = os.path.join(COMPRESS_DIR, f'task_{task_id}') + out_path = os.path.join(out_dir, os.path.basename(video_path)) + + # 缓存复用：源文件未变且压缩产物有效 + try: + if (os.path.isfile(out_path) + and os.path.getsize(out_path) > 0 + and os.path.getmtime(out_path) >= os.path.getmtime(video_path)): + logger.info(f"[task_id={task_id}] 复用压缩缓存: {out_path}") + return out_path + except OSError: + pass + + os.makedirs(out_dir, exist_ok=True) + src_size = os.path.getsize(video_path) + start = time.time() + + cmd = [ + self.ffmpeg, '-y', + '-i', video_path, + # 第二级 scale 向下取偶（h264 要求偶数尺寸；force_divisible_by 需 ffmpeg>=4.3） + '-vf', ('scale=854:480:force_original_aspect_ratio=decrease,' + 'scale=trunc(iw/2)*2:trunc(ih/2)*2'), + '-c:v', 'libx264', '-preset', 'veryfast', '-crf', '28', + '-an', + out_path, + ] + try: + result = subprocess.run( + cmd, capture_output=True, text=True, + timeout=self.compress_timeout, + ) + except subprocess.TimeoutExpired: + logger.error(f"[task_id={task_id}] 压缩超时 ({self.compress_timeout}s)，回退原始上传") + return None + except OSError as e: + logger.error(f"[task_id={task_id}] 启动 ffmpeg 失败: {e}") + return None + + if result.returncode != 0 or not os.path.isfile(out_path) or os.path.getsize(out_path) == 0: + stderr_tail = (result.stderr or '')[-300:] + logger.error(f"[task_id={task_id}] 压缩失败 (rc={result.returncode}): {stderr_tail}") + return None + + dst_size = os.path.getsize(out_path) + elapsed = time.time() - start + logger.info(f"[task_id={task_id}] 预压缩完成: {src_size/1048576:.1f}MB → " + f"{dst_size/1048576:.1f}MB ({src_size/max(dst_size,1):.1f}x)，耗时 {elapsed:.0f}s") + return out_path + + @staticmethod + def _cleanup_compress_cache(): + """清理超过 TTL 的压缩缓存目录（tmpfs 空间有限）""" + try: + if not os.path.isdir(COMPRESS_DIR): + return + cutoff = time.time() - COMPRESS_CACHE_TTL + for entry in os.listdir(COMPRESS_DIR): + path = os.path.join(COMPRESS_DIR, entry) + try: + if os.path.isdir(path) and os.path.getmtime(path) < cutoff: + shutil.rmtree(path, ignore_errors=True) + except OSError: + continue + except OSError: + pass + def _dispatch_one(self, task): - """上传视频至 Edge 异步队列（自动选择直接/分块模式）""" + """上传视频至 Edge 异步队列（大文件先预压缩，自动选择直接/分块模式）""" task_id = task['task_id'] video_path = task['video_path'] @@ -101,16 +205,29 @@ class Dispatcher: size_mb = file_size / (1024 * 1024) db_layer.update_task_status(task_id, 'PROCESSING') + # 跨境上行带宽受限（实测 ~0.5-0.9MB/s），大文件先预压缩再上传 + upload_path = video_path + if file_size > COMPRESS_THRESHOLD: + compressed = self._compress_video(task_id, video_path) + if compressed: + upload_path = compressed + file_size = os.path.getsize(upload_path) + size_mb = file_size / (1024 * 1024) + # 压缩耗时较长，重置 stale 计时基准（reclaim 按 updated_at 判断） + db_layer.update_task_status(task_id, 'PROCESSING') + else: + logger.warning(f"[task_id={task_id}] 压缩失败，回退原始文件上传 ({size_mb:.1f}MB)") + payload = self._build_payload(task) if file_size > CHUNK_THRESHOLD: logger.info(f"[task_id={task_id}] 大文件分块上传: {size_mb:.1f}MB, " f"{math.ceil(file_size / CHUNK_SIZE)} 块") - self._dispatch_chunked(task, payload, video_path, file_size) + self._dispatch_chunked(task, payload, upload_path, file_size) else: log_task(logger, task_id, 'dispatcher', f'直接上传: {self.edge_url} ({size_mb:.1f}MB)') - self._dispatch_direct(task, payload, video_path) + self._dispatch_direct(task, payload, upload_path) def _dispatch_direct(self, task, payload, video_path): """小文件直接上传至 /enqueue""" @@ -167,8 +284,9 @@ class Dispatcher: f"≠ expected={expected_total}（chunk_size 已变更），从头上传") return set(), edge_total return uploaded, edge_total - except requests.RequestException: - pass + logger.warning(f"[task_id={task_id}] 查询已上传分块返回 {resp.status_code}，将全量重传") + except requests.RequestException as e: + logger.warning(f"[task_id={task_id}] 查询已上传分块失败（将全量重传）: {e}") return set(), 0 def _dispatch_chunked(self, task, payload, video_path, file_size): diff --git a/fam-edge/src/fam_edge/api_gateway/api_gateway.py b/fam-edge/src/fam_edge/api_gateway/api_gateway.py index 7b5becd..dd83aeb 100644 --- a/fam-edge/src/fam_edge/api_gateway/api_gateway.py +++ b/fam-edge/src/fam_edge/api_gateway/api_gateway.py @@ -168,10 +168,10 @@ def upload_chunk(): chunk_path = os.path.join(d, f"chunk_{chunk_index:04d}") try: - # 检查 total_chunks 是否变化（chunk_size 变更导致），自动清理旧分块 + # 检查 total_chunks 是否变化（chunk_size 变更导致），自动清理旧分块并更新元数据 + import json meta_path = os.path.join(d, "meta.json") if total_chunks and os.path.exists(meta_path): - import json try: with open(meta_path) as mf: old_meta = json.load(mf) @@ -181,15 +181,16 @@ def upload_chunk(): for fn in os.listdir(d): if fn.startswith('chunk_'): os.remove(os.path.join(d, fn)) + with open(meta_path, 'w') as f: + json.dump({"filename": filename, "total_chunks": total_chunks}, f) except (json.JSONDecodeError, IOError): pass chunk_file.save(chunk_path) size_kb = os.path.getsize(chunk_path) / 1024 - # 写/更新元数据 + # 写元数据（首次上传时） if not os.path.exists(meta_path): - import json meta = {"filename": filename, "total_chunks": total_chunks} with open(meta_path, 'w') as f: json.dump(meta, f)