システム概況と設計方針
LivePortraitは高精度な顔写真アニメーション生成パイプラインを提供するアプリケーションであり、本番環境では大規模な推論タスクを安定的に捌く必要がある。サービスの中断を最小化し、予期せぬインフラ障害やソフトウェア異常に対して迅速に対応するため、体系的な災害復旧（DR）および事業継続計画（BCP）が必須である。以下の節では、重要コンポーネントの管理、想定される障害タイプ別の対応手順、バックアップアーキテクチャ、自動復旧フロー、および定期的な訓練体制について詳述する。
依存関係と重要リソース管理
| コンポーネント | 推奨バージョン | Criticality Level | 管理方針 |
|---|---|---|---|
| PyTorch | 2.3.0以降 | Critical | スナップショット取得とレジストリミラーリング |
| OpenCV | 4.10.0.84 | High | バージョン固定パッケージング |
| FFmpeg | OS標準 | High | イメージ内バンドルまたはホストレベル検証 |
| 学習済みモデル | 特定リリース | Critical Asset | 地理的に分散したストレージへの複製 |
| 設定ファイル | プロジェクト固有 | Moderate | Git追跡と差分コミット |
障害シナリオ分類と緩和策
ハードウェア層の異常検知
import torch
import logging
logger = logging.getLogger(__name__)
def assess_gpu_state() -> dict:
"""CUDAデバイス状態の評価を行う"""
result = {"status": "unavailable", "detail": ""}
if not torch.cuda.is_available():
result["detail"] = "CUDAドライバー未検出"
logger.warning(result)
return result
try:
# 軽量マトリックス演算による機能テスト
dummy_tensor = torch.empty((512, 512), device="cuda")
dummy_tensor.sum() # アクティベーション強制
result["status"] = "operational"
result["detail"] = f"利用可能GPU: {torch.cuda.get_device_name(0)}"
logger.info(f"GPUステータス: {result['detail']}")
except RuntimeError as err:
result["status"] = "fault"
result["detail"] = str(err)
logger.error("GPU初期化失敗", exc_info=True)
return resultストレージ階層の保護戦略
- 重みファイル群: 每日差分スナップショットと週次完全バックアップを併用
- 設定リポジトリ: Git分散追跡＋オフサイト暗号化転送
- 出力メディア: オブジェクトストレージへのリアルタイム同期配置
ランタイム依存関係の復元
# 仮想環境内のパッケージ状態をダンプ
import sys
import subprocess
subprocess.run([sys.executable, "-m", "pip", "freeze"], capture_output=True, text=True, stdout=open("dep_snapshot_v1.lock", "w"))# 環境リストからの再構築
subprocess.check_call(["pip", "install", "-r", "dep_snapshot_v1.lock", "--force-reinstall"])システムレヴェルの外部ツールの稼働確認には、非同期サブプロセス呼び出しを採用する。
import asyncio
import shlex
from pathlib import Path
async def probe_system_bins(binary_names: list[str]) -> dict:
"""指定されたCLIツールがパス上に存在し、--versionを実行できるか確認"""
status_map = {}
for bin_name in binary_names:
try:
proc = await asyncio.create_subprocess_exec(shlex.split(bin_name))
await proc.wait()
status_map[bin_name] = "ready" if proc.returncode == 0 else "missing_flag"
except FileNotFoundError:
status_map[bin_name] = "not_found"
return status_mapアセット整合性検証の最適化
import hashlib
from typing import Tuple
from pathlib import Path
BLOCK_SIZE = 8192
def compute_file_checksum(target_path: str, algorithm: str = "sha256") -> str:
"""ファイルストリームから逐次ハッシュ値を算出"""
h = hashlib.new(algorithm)
with open(target_path, "rb") as archive:
while chunk := archive.read(BLOCK_SIZE):
h.update(chunk)
return h.hexdigest()
def validate_artifact(filepath: str, expected_digest: str) -> Tuple[bool, str]:
"""期待値との照合を実行"""
if not Path(filepath).is_file():
return False, "対象パスが存在しません"
current_hash = compute_file_checksum(filepath)
match = current_hash == expected_digest
return match, f"照合結果: {'一致' if match else '不一致'} (計算値={current_hash[:12]}...)"バックアップポリシーと自動化実装
| データ種類 | 実施間隔 | 保持期間 | 保管先 |
|---|---|---|---|
| システムイメージ | 週1回 | 4週間 | NAS + 冷たいクラウドストレージ |
| 変更データ | 日次 | 30日分 | S3互換オブジェクト |
| インフラ設定 | イベント駆動 | 履歴全保存 | Gitワークフロー |
| コアモデル | 月次 | 永久 | リージョン横断レプリケーション |
シェルベースの実態収集スクリプト
#!/usr/bin/env bash
set -euo pipefail
BACKUP_ROOT="/mnt/disks/backup/liveportrait_archives"
ARCHIVE_NAME="snapshot_$(date +%Y%m%d_%H%M%S)"
TARGET_DIR="${BACKUP_ROOT}/${ARCHIVE_NAME}"
mkdir -p "${TARGET_DIR}"
echo "[+] アーカイブ作成を開始..."
# モデルパラメータ群
tar cf "${TARGET_DIR}/model_weights.tar" ./pretrained_weights/
# 設定ドキュメント一式
tar cf "${TARGET_DIR}/project_config.tar" ./src/config/
# Python環境スナップショット
python3 -m pip freeze > "${TARGET_DIR}/env_deps_${ARCHIVE_NAME}.txt"
# ハードウェア情報キャプチャ
nvidia-smi --query-gpu=name,memory.used,memory.total --format=csv,noheader > "${TARGET_DIR}/gpu_state_${ARCHIVE_NAME}.csv"
python3 -c "import torch, json; print(json.dumps({'version': torch.__version__, 'cuda': torch.cuda.is_available()}))" > "${TARGET_DIR}/framework_info.json"
echo "[+] バックアップ完了: ${TARGET_DIR}"稼働状況監視とアラート閾値
FastAPI連携ヘルスエンドポイント
from fastapi import FastAPI, APIRouter
from pydantic import BaseModel
import cv2
import torch
import os
app = FastAPI(title="LivePortrait Monitor")
class ServiceStatus(BaseModel):
service: str
component: str
state: str
latency_ms: int | None = None
router = APIRouter()
@router.get("/v1/status/ping", response_model=list[ServiceStatus])
async def fetch_component_health() -> list[ServiceStatus]:
"""各ミドルウェアとAI推論エンジンの現在状態を取得"""
reports = []
# PyTorch CUDAプール評価
if torch.cuda.is_available():
mem_alloc = torch.cuda.memory_allocated(0)
mem_total = torch.cuda.mem_get_info(0)[1]
reports.append(ServiceStatus(service="ai_core", component="pytorch_cuda", state="ok", latency_ms=int((mem_alloc/mem_total)*100)))
else:
reports.append(ServiceStatus(service="ai_core", component="pytorch_cuda", state="down"))
# OpenCVバッチ処理能力確認
reports.append(ServiceStatus(service="vision", component="opencv", state="ok" if cv2.__version__ else "error", latency_ms=None))
# ローカルモデルキャッシュの有効性チェック
cache_paths = [
"assets/checkpoints/driver_extractor.pth",
"assets/checkpoints/appearance_net.pth"
]
missing_count = sum(1 for p in cache_paths if not os.path.isfile(p))
reports.append(ServiceStatus(service="ml_assets", component="checkpoint_cache", state="degraded" if missing_count > 0 else "healthy", latency_ms=None))
return reportsパフォーマンス監視メトリクス定義表
| 計測項目 | 正常域 | 警告閾値 | サンプリング周期 |
|---|---|---|---|
| GPUメモリ占有率 | 70〜85% | 92%超を300秒維持 | 60秒毎 |
| ホストRAM利用率 | 60〜80% | 88%超 | 60秒毎 |
| モデルマッピング時間 | <15s | >45s | 起動時 |
| エンドツーエンド推論レイテンシ | <600ms | >1.2s | リクエスト単位 |
復旧目標指標と実施手順
| インシデント種別 | RTO（最大停止時間） | RPO（許容データ損失） | 優先順位 |
|---|---|---|---|
| ノード物理障害 | 4時間以内 | 直近1時間分 | P1 |
| コンテナ/Runtime障害 | 2時間以内 | 15分以内 | P1 |
| パーミッション/設定破損 | 8時間以内 | 24時間分 | P2 |
| ネットワーク分断 | 1時間以内 | 連続転送 | P1 |
段階的復旧オーケストレーション
フェーズ1：基盤再構築
# condaベースのクリーン環境準備
import subprocess
subprocess.run(["conda", "create", "-y", "-n", "liveportrait_dr", "python=3.10"], check=True)
# 指定フレームワークインストール
subprocess.run(["pip", "install", "torch==2.3.0+cu118", "-f", "https://download.pytorch.org/whl/torch_stable.html"], check=True)
subprocess.run(["pip", "install", "-r", "requirements_inference.txt"], check=True)フェーズ2：アセット復旧と整合性確認
import tarfile
from pathlib import Path
import hashlib
def restore_and_validate(source_archive: str, dest_dir: str) -> bool:
"""アーカイブ展開後、必要なファイルサイズとハッシュをチェック"""
try:
with tarfile.open(source_archive, "r:gz") as tf:
tf.extractall(dest_dir)
required_hashes = {
"driver_extractor.pth": "expected_a1b2...",
"appearance_net.pth": "expected_c3d4..."
}
base_path = Path(dest_dir) / "pretrained_weights"
for fname, target_hash in required_hashes.items():
full_path = base_path / fname
if not full_path.exists():
raise FileNotFoundError(f"{fname}が見つかりません")
calc_hash = compute_file_checksum(str(full_path))
if calc_hash != target_hash:
print(f"警告: {fname} の整合性が崩れています")
return False
except Exception as e:
print(f"復旧失敗: {e}")
return False
return Trueフェーズ3：サービス再起動テスト
def run_smoke_tests() -> dict:
"""回復後の主要フローをサクラテストで検証"""
test_suite = {
"config_load": lambda: load_yaml("./src/config/test.yaml"),
"model_warmup": lambda: init_inference_engine(cache_init=True),
"dummy_inference": lambda: generate_dummy_video(out_path="./output/smoke.mp4")
}
outcomes = {}
for name, task_fn in test_suite.items():
try:
task_fn()
outcomes[name] = "success"
except BaseException as ex:
outcomes[name] = f"failed ({type(ex).__name__})"
return outcomes事業継続性とエグレス制御
マルチゾーン冗長構成と切り替えフロー
- プローブ系がプライマリ環境の応答遅延を検知
- ロードバランサ経由でセカンダリーゾーンへルーティング切替
- オブジェクトストレージ間の差分レプリケーション完走待ち
- 自動デプロイパイプラインにより最新コードを適用
- ユーザー向けメンテナンスウェルカムページへの誘導
リソース逼迫時の graceful degradation 設定例
DEGRADATION_MODES = {
"conservative": {
"input_resolution_limit": 512,
"use_fp16_computation": True,
"motion_scale_factor": 0.75,
"enable_face_retargeting": False
},
"critical_throttle": {
"input_resolution_limit": 256,
"stitching_algorithm_disabled": True,
"background_pasteback_disabled": True,
"render_fps_cap": 12
}
}訓練運用とガバナンス体制
定期訓練スケジュール表
| 訓練種目 | 実施頻度 | 担当組織 | 検証目的 |
|---|---|---|---|
| 全体フェイルオーバー演習 | 年4回 | SREチーム全体 | RTO/RPO目標達成率の測定 |
| 単体コンポーネント障害注入 | 月1回 | インフラ担当 | 孤立した異常ケースへの耐性確認 |
| スクリプト自動実行テスト | 週1回 | CICDシステム | 復旧コマンドの互換性担保 |
訓練成果スコアリングロジック
class DRExerciseEvaluator:
def __init__(self, target_rto_hours: float = 4.0, target_rpo_hours: float = 1.0):
self.rto_target = target_rto_hours * 3600
self.rpo_target = target_rpo_hours * 3600
def calculate_performance_score(self, elapsed_seconds: float, lost_records_hours: float, operations_list: list[bool]) -> int:
"""復旧所要時間、データ損失量、作業成功率から総合得点を算出"""
points = 0
# RTO基準加点
if elapsed_seconds <= self.rto_target:
points += 30
elif elapsed_seconds <= self.rto_target * 1.5:
points += 15
# RPO基準加点
if lost_records_hours <= self.rpo_target:
points += 30
elif lost_records_hours <= self.rpo_target * 2:
points += 15
# 成功オペレーション加重平均
if operations_list:
success_ratio = sum(operations_list) / len(operations_list)
points += int(success_ratio * 40)
return min(points, 100)ドキュメントライフサイクル管理
- マスタードキュメント: Gitリポジトリ管理。マージ前にレビュー委員会による技術的精査を実施
- パラメータテンプレート: アプリケーションコード同梱のCIフックで自動差分検知
- 復旧ユーティリティ: バージョンタグ付けによる個別管理。ステージング環境での事前適合試験を義務付け
変更適応ワークフロー
- 影響範囲を含む詳細設計書を提出
- 復旧インパクト分析シートへの記載と承認
- 分離されたQA環境での模擬切断テスト
- 関連付録とスクリプトコメントの即時更新
- 関係者への技術通達メール配信