엣지 AI 배포 파이프라인

# Dockerfile for Edge AI Application (Jetson)

# Base image: NVIDIA L4T with TensorRT
FROM nvcr.io/nvidia/l4t-tensorrt:r8.5.2-runtime

# Set environment variables
ENV DEBIAN_FRONTEND=noninteractive
ENV PYTHONUNBUFFERED=1
ENV MODEL_PATH=/app/models
ENV CONFIG_PATH=/app/config

# Install system dependencies
RUN apt-get update && apt-get install -y --no-install-recommends \
    python3-pip \
    python3-dev \
    libopencv-dev \
    libgstreamer1.0-dev \
    && rm -rf /var/lib/apt/lists/*

# Install Python dependencies
COPY requirements.txt /tmp/
RUN pip3 install --no-cache-dir -r /tmp/requirements.txt

# Copy application code
WORKDIR /app
COPY src/ ./src/
COPY models/ ./models/
COPY config/ ./config/

# Create non-root user for security
RUN useradd -m -u 1000 edgeai && \
    chown -R edgeai:edgeai /app
USER edgeai

# Health check
HEALTHCHECK --interval=30s --timeout=10s --retries=3 \
    CMD python3 -c "import requests; requests.get('http://localhost:8080/health')"

# Entrypoint
ENTRYPOINT ["python3", "src/main.py"]
CMD ["--config", "/app/config/production.yaml"]

# docker-compose.yml for Edge AI Stack

version: '3.8'

services:
  edge-ai-inference:
    image: registry.example.com/edge-ai:${VERSION:-latest}
    runtime: nvidia
    environment:
      - NVIDIA_VISIBLE_DEVICES=all
      - MODEL_VERSION=${MODEL_VERSION}
      - LOG_LEVEL=INFO
    volumes:
      - ./models:/app/models:ro
      - ./config:/app/config:ro
      - ./logs:/app/logs
      - /dev/video0:/dev/video0  # Camera access
    devices:
      - /dev/video0
    ports:
      - "8080:8080"
    restart: unless-stopped
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: all
              capabilities: [gpu]

  edge-gateway:
    image: registry.example.com/edge-gateway:${VERSION:-latest}
    environment:
      - MQTT_BROKER=${MQTT_BROKER}
      - CLOUD_API_URL=${CLOUD_API_URL}
    volumes:
      - ./data:/app/data
    ports:
      - "1883:1883"
    depends_on:
      - edge-ai-inference

  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
    ports:
      - "9090:9090"

import asyncio
import hashlib
import subprocess
from dataclasses import dataclass
from enum import Enum
from typing import Optional, Callable
import aiohttp
import docker

class UpdateState(Enum):
    IDLE = "idle"
    DOWNLOADING = "downloading"
    VERIFYING = "verifying"
    INSTALLING = "installing"
    VALIDATING = "validating"
    COMPLETED = "completed"
    FAILED = "failed"
    ROLLING_BACK = "rolling_back"

@dataclass
class OTAUpdate:
    version: str
    image_url: str
    checksum: str
    signature: str
    release_notes: str
    min_version: Optional[str] = None

class OTAUpdateManager:
    """OTA 업데이트 관리자"""

    def __init__(
        self,
        device_id: str,
        current_version: str,
        update_server_url: str,
        validation_fn: Optional[Callable] = None
    ):
        self.device_id = device_id
        self.current_version = current_version
        self.server_url = update_server_url
        self.validation_fn = validation_fn

        self.state = UpdateState.IDLE
        self.docker_client = docker.from_env()
        self.previous_image: Optional[str] = None

    async def check_for_updates(self) -> Optional[OTAUpdate]:
        """서버에서 업데이트 확인"""
        async with aiohttp.ClientSession() as session:
            params = {
                "device_id": self.device_id,
                "current_version": self.current_version
            }
            async with session.get(
                f"{self.server_url}/updates/check",
                params=params
            ) as resp:
                if resp.status == 200:
                    data = await resp.json()
                    return OTAUpdate(**data) if data else None
        return None

    async def apply_update(self, update: OTAUpdate) -> bool:
        """업데이트 적용 (원자적)"""
        try:
            # 1. 이미지 다운로드
            self.state = UpdateState.DOWNLOADING
            await self._pull_image(update.image_url)

            # 2. 체크섬 검증
            self.state = UpdateState.VERIFYING
            if not self._verify_checksum(update.image_url, update.checksum):
                raise ValueError("Checksum mismatch")

            # 3. 이전 버전 백업
            self.previous_image = self._get_current_image()

            # 4. 새 버전 설치
            self.state = UpdateState.INSTALLING
            await self._deploy_container(update.image_url)

            # 5. 헬스 체크
            self.state = UpdateState.VALIDATING
            if not await self._validate_deployment():
                raise RuntimeError("Validation failed")

            # 6. 완료
            self.state = UpdateState.COMPLETED
            self.current_version = update.version
            return True

        except Exception as e:
            self.state = UpdateState.FAILED
            await self.rollback()
            return False

    async def rollback(self) -> bool:
        """이전 버전으로 롤백"""
        if not self.previous_image:
            return False

        self.state = UpdateState.ROLLING_BACK
        try:
            await self._deploy_container(self.previous_image)
            self.state = UpdateState.IDLE
            return True
        except:
            self.state = UpdateState.FAILED
            return False

    async def _pull_image(self, image_url: str):
        """Docker 이미지 풀"""
        self.docker_client.images.pull(image_url)

    async def _deploy_container(self, image: str):
        """컨테이너 배포 (무중단)"""
        # 기존 컨테이너 중지
        try:
            old = self.docker_client.containers.get("edge-ai")
            old.stop(timeout=30)
            old.remove()
        except docker.errors.NotFound:
            pass

        # 새 컨테이너 시작
        self.docker_client.containers.run(
            image,
            name="edge-ai",
            runtime="nvidia",
            detach=True,
            restart_policy={"Name": "unless-stopped"},
            environment={"NVIDIA_VISIBLE_DEVICES": "all"},
            ports={"8080/tcp": 8080}
        )

    async def _validate_deployment(self) -> bool:
        """배포 검증 (헬스 체크 + 커스텀 검증)"""
        # 헬스 체크 대기
        for _ in range(30):
            try:
                async with aiohttp.ClientSession() as session:
                    async with session.get(
                        "http://localhost:8080/health"
                    ) as resp:
                        if resp.status == 200:
                            break
            except:
                pass
            await asyncio.sleep(1)
        else:
            return False

        # 커스텀 검증 (정확도 체크 등)
        if self.validation_fn:
            return self.validation_fn()

        return True

class RollbackPolicy:
    """롤백 정책 관리"""

    def __init__(
        self,
        health_check_timeout: int = 30,
        accuracy_threshold: float = 0.95,
        max_cpu_percent: float = 80,
        max_memory_mb: int = 4096,
        observation_period_hours: int = 24
    ):
        self.health_check_timeout = health_check_timeout
        self.accuracy_threshold = accuracy_threshold
        self.max_cpu = max_cpu_percent
        self.max_memory = max_memory_mb
        self.observation_period = observation_period_hours

        self.deployed_at: Optional[datetime] = None
        self.metrics_history: List[dict] = []

    def should_rollback(self, current_metrics: dict) -> tuple:
        """롤백 필요 여부 판단"""
        reasons = []

        # 정확도 체크
        if current_metrics.get("accuracy", 1.0) < self.accuracy_threshold:
            reasons.append(f"Accuracy below threshold: {current_metrics['accuracy']}")

        # 리소스 체크
        if current_metrics.get("cpu_percent", 0) > self.max_cpu:
            reasons.append(f"CPU usage too high: {current_metrics['cpu_percent']}%")

        if current_metrics.get("memory_mb", 0) > self.max_memory:
            reasons.append(f"Memory usage too high: {current_metrics['memory_mb']}MB")

        # 추론 지연 체크
        if current_metrics.get("p99_latency_ms", 0) > 100:
            reasons.append(f"Latency too high: {current_metrics['p99_latency_ms']}ms")

        return len(reasons) > 0, reasons

    def in_observation_period(self) -> bool:
        """관찰 기간 내 여부"""
        if self.deployed_at is None:
            return False
        elapsed = datetime.now() - self.deployed_at
        return elapsed < timedelta(hours=self.observation_period)

    def record_metrics(self, metrics: dict):
        """메트릭 기록 (롤백 판단용)"""
        self.metrics_history.append({
            "timestamp": datetime.now(),
            **metrics
        })
        # 최근 1시간만 유지
        cutoff = datetime.now() - timedelta(hours=1)
        self.metrics_history = [
            m for m in self.metrics_history
            if m["timestamp"] > cutoff
        ]

# GitHub Actions workflow for multi-arch build
name: Build and Push Edge AI Image

on:
  push:
    tags:
      - 'v*'

jobs:
  build:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout
        uses: actions/checkout@v4

      - name: Set up QEMU
        uses: docker/setup-qemu-action@v3

      - name: Set up Docker Buildx
        uses: docker/setup-buildx-action@v3

      - name: Login to Registry
        uses: docker/login-action@v3
        with:
          registry: ${{ secrets.REGISTRY_URL }}
          username: ${{ secrets.REGISTRY_USER }}
          password: ${{ secrets.REGISTRY_PASSWORD }}

      - name: Build and Push
        uses: docker/build-push-action@v5
        with:
          context: .
          platforms: linux/amd64,linux/arm64
          push: true
          tags: |
            ${{ secrets.REGISTRY_URL }}/edge-ai:${{ github.ref_name }}
            ${{ secrets.REGISTRY_URL }}/edge-ai:latest
          cache-from: type=gha
          cache-to: type=gha,mode=max

      - name: Notify OTA Server
        run: |
          curl -X POST ${{ secrets.OTA_SERVER }}/releases \
            -H "Authorization: Bearer ${{ secrets.OTA_TOKEN }}" \
            -d '{"version":"${{ github.ref_name }}","image":"edge-ai:${{ github.ref_name }}"}'

from dataclasses import dataclass
from typing import List, Dict
import asyncio

@dataclass
class DeploymentTarget:
    device_id: str
    device_group: str
    current_version: str
    status: str

class FleetDeploymentOrchestrator:
    """대규모 엣지 플릿 배포 오케스트레이터"""

    def __init__(self, api_endpoint: str):
        self.api = api_endpoint
        self.deployment_status: Dict[str, str] = {}

    async def staged_rollout(
        self,
        targets: List[DeploymentTarget],
        new_version: str,
        stages: List[float] = [0.05, 0.25, 0.5, 1.0],  # 5%, 25%, 50%, 100%
        validation_wait_sec: int = 300
    ):
        """단계적 롤아웃"""
        total = len(targets)

        for stage_idx, stage_percent in enumerate(stages):
            target_count = int(total * stage_percent)
            batch = targets[:target_count]

            print(f"Stage {stage_idx + 1}: Deploying to {len(batch)} devices ({stage_percent*100}%)")

            # 배치 배포
            results = await asyncio.gather(*[
                self._deploy_single(t.device_id, new_version)
                for t in batch
            ])

            # 결과 확인
            failed = sum(1 for r in results if not r)
            if failed / len(batch) > 0.1:  # 10% 이상 실패 시 중단
                print(f"Too many failures ({failed}), aborting rollout")
                await self._rollback_batch(batch)
                return False

            # 안정화 대기
            print(f"Waiting {validation_wait_sec}s for validation...")
            await asyncio.sleep(validation_wait_sec)

            # 메트릭 기반 검증
            if not await self._validate_stage(batch):
                print("Stage validation failed, rolling back")
                await self._rollback_batch(batch)
                return False

            print(f"Stage {stage_idx + 1} completed successfully")

        return True

    async def _validate_stage(self, targets: List[DeploymentTarget]) -> bool:
        """스테이지 검증 (메트릭 수집 및 분석)"""
        # 각 디바이스의 메트릭 수집
        metrics = []
        for target in targets:
            m = await self._get_device_metrics(target.device_id)
            metrics.append(m)

        # 이상 탐지
        avg_fps = statistics.mean([m["fps"] for m in metrics])
        avg_latency = statistics.mean([m["latency_ms"] for m in metrics])
        error_rate = sum(1 for m in metrics if m["has_error"]) / len(metrics)

        return avg_fps > 30 and avg_latency < 50 and error_rate < 0.05