컨테이너 기반 서비스를 운영하다 보면 단순한 CLI나 GUI를 넘어 프로그래밍 방식으로 컨테이너를 관리해야 할 때가 옵니다. 사용자별 컨테이너 할당, 자동 스케일링, 과금 시스템 연동 등 비즈니스 로직이 필요한 순간이죠. 이 글에서는 FastAPI를 활용해 Podman을 제어하는 RESTful API 서버를 구축하는 방법을 상세히 다룹니다. 보안, 인증, 비즈니스 로직 적용까지 프로덕션 레벨의 구성을 목표로 합니다.
왜 중간 레이어가 필요한가?
Podman API 직접 노출의 문제점
Podman은 Docker 호환 REST API를 기본 제공합니다. 그렇다면 프론트엔드에서 이 API를 직접 호출하면 되지 않을까요? 결론부터 말하면, 절대 그렇게 해서는 안 됩니다.
첫째, 보안 위험이 심각합니다. Podman 소켓에 접근할 수 있다는 것은 해당 시스템에서 임의의 컨테이너를 실행할 수 있다는 의미입니다. 악의적인 사용자가 호스트 파일시스템을 마운트한 컨테이너를 실행하면 서버 전체가 위험에 노출됩니다.
둘째, Podman API 자체에는 사용자 인증이나 권한 관리 기능이 없습니다. 누가 어떤 작업을 했는지 추적할 방법이 없고, 특정 사용자의 작업을 제한할 수도 없습니다.
셋째, 비즈니스 로직을 적용할 수 없습니다. 사용자당 컨테이너 수 제한, 특정 이미지만 허용, 리소스 쿼터 적용 같은 정책을 구현할 곳이 없습니다.
중간 레이어의 역할
FastAPI로 구축한 중간 레이어는 다음과 같은 역할을 수행합니다.
[클라이언트/프론트엔드]
│
▼
[FastAPI 백엔드] ─────────────────────────────────
│ • JWT/OAuth 인증
│ • 권한 검사 (RBAC)
│ • 입력 검증 및 새니타이징
│ • 비즈니스 로직 (제한, 정책)
│ • 감사 로깅
│ • 에러 핸들링 및 표준화된 응답
▼
[Podman 소켓] ──── 내부 네트워크에서만 접근 가능
│
▼
[컨테이너들]
시스템 요구사항 및 사전 준비
필요 환경
- Python 3.10 이상
- Podman 4.0 이상 (5.x 권장)
- Linux 서버 (RHEL, Ubuntu, Rocky Linux 등)
Podman 소켓 활성화
먼저 Podman API 소켓을 활성화해야 합니다.
# 사용자 레벨 소켓 활성화 (루트리스)
systemctl --user enable --now podman.socket
# 소켓 상태 확인
systemctl --user status podman.socket
# 소켓 경로 확인
echo $XDG_RUNTIME_DIR/podman/podman.sock
# 일반적으로 /run/user/1000/podman/podman.sock
시스템 레벨로 운영하려면 root 권한으로 설정합니다.
# 시스템 레벨 소켓 활성화
sudo systemctl enable --now podman.socket
# 소켓 경로: /run/podman/podman.sock
Python 환경 설정
# 프로젝트 디렉토리 생성
mkdir podman-api-server && cd podman-api-server
# 가상환경 생성 및 활성화
python -m venv venv
source venv/bin/activate
# 의존성 설치
pip install fastapi uvicorn podman python-jose[cryptography] passlib[bcrypt] python-multipart sqlalchemy aiosqlite pydantic-settings
프로젝트 구조
podman-api-server/
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI 앱 진입점
│ ├── config.py # 설정 관리
│ ├── dependencies.py # 의존성 주입
│ ├── models/
│ │ ├── __init__.py
│ │ ├── user.py # 사용자 모델
│ │ └── container.py # 컨테이너 관련 스키마
│ ├── routers/
│ │ ├── __init__.py
│ │ ├── auth.py # 인증 라우터
│ │ ├── containers.py # 컨테이너 관리
│ │ ├── images.py # 이미지 관리
│ │ └── system.py # 시스템 정보
│ ├── services/
│ │ ├── __init__.py
│ │ ├── podman_service.py # Podman 연동
│ │ └── audit_service.py # 감사 로깅
│ └── middleware/
│ ├── __init__.py
│ └── logging.py # 로깅 미들웨어
├── tests/
├── .env
├── requirements.txt
└── docker-compose.yml
핵심 구현
설정 관리 (config.py)
from pydantic_settings import BaseSettings
from typing import List
from functools import lru_cache
class Settings(BaseSettings):
# 앱 설정
app_name: str = "Podman API Server"
debug: bool = False
# Podman 설정
podman_socket: str = "unix:///run/user/1000/podman/podman.sock"
# 보안 설정
secret_key: str = "your-secret-key-change-in-production"
algorithm: str = "HS256"
access_token_expire_minutes: int = 30
# 비즈니스 정책
max_containers_per_user: int = 10
max_memory_per_container: str = "2g"
max_cpus_per_container: float = 2.0
allowed_images: List[str] = [
"nginx",
"python",
"node",
"postgres",
"redis",
"docker.io/library/nginx",
"docker.io/library/python",
]
# 데이터베이스
database_url: str = "sqlite+aiosqlite:///./podman_api.db"
class Config:
env_file = ".env"
@lru_cache()
def get_settings():
return Settings()
Podman 서비스 클래스 (services/podman_service.py)
from podman import PodmanClient
from podman.errors import NotFound, APIError
from typing import Optional, List, Dict, Any
from contextlib import contextmanager
import logging
from app.config import get_settings
logger = logging.getLogger(__name__)
settings = get_settings()
class PodmanService:
"""Podman API와 상호작용하는 서비스 클래스"""
def __init__(self):
self.socket_url = settings.podman_socket
@contextmanager
def get_client(self):
"""Podman 클라이언트 컨텍스트 매니저"""
client = PodmanClient(base_url=self.socket_url)
try:
yield client
finally:
client.close()
# ============ 컨테이너 관리 ============
def list_containers(
self,
all: bool = False,
filters: Optional[Dict] = None
) -> List[Dict[str, Any]]:
"""컨테이너 목록 조회"""
with self.get_client() as client:
containers = client.containers.list(all=all, filters=filters)
return [self._container_to_dict(c) for c in containers]
def get_container(self, container_id: str) -> Optional[Dict[str, Any]]:
"""특정 컨테이너 조회"""
with self.get_client() as client:
try:
container = client.containers.get(container_id)
return self._container_to_dict(container, detailed=True)
except NotFound:
return None
def create_container(
self,
image: str,
name: str,
owner: str,
environment: Optional[Dict[str, str]] = None,
ports: Optional[Dict[str, int]] = None,
volumes: Optional[List[str]] = None,
memory_limit: Optional[str] = None,
cpu_limit: Optional[float] = None,
) -> Dict[str, Any]:
"""컨테이너 생성"""
with self.get_client() as client:
# 라벨에 소유자 정보 추가
labels = {
"managed-by": "podman-api-server",
"owner": owner,
}
# 포트 매핑 변환
port_bindings = None
if ports:
port_bindings = {
f"{container_port}/tcp": host_port
for container_port, host_port in ports.items()
}
# 리소스 제한
mem_limit = memory_limit or settings.max_memory_per_container
cpus = cpu_limit or settings.max_cpus_per_container
container = client.containers.create(
image=image,
name=name,
labels=labels,
environment=environment or {},
ports=port_bindings,
mounts=volumes or [],
mem_limit=mem_limit,
cpus=cpus,
detach=True,
)
logger.info(f"Container created: {container.id} by {owner}")
return self._container_to_dict(container)
def start_container(self, container_id: str) -> bool:
"""컨테이너 시작"""
with self.get_client() as client:
try:
container = client.containers.get(container_id)
container.start()
logger.info(f"Container started: {container_id}")
return True
except NotFound:
return False
def stop_container(self, container_id: str, timeout: int = 10) -> bool:
"""컨테이너 중지"""
with self.get_client() as client:
try:
container = client.containers.get(container_id)
container.stop(timeout=timeout)
logger.info(f"Container stopped: {container_id}")
return True
except NotFound:
return False
def remove_container(self, container_id: str, force: bool = False) -> bool:
"""컨테이너 삭제"""
with self.get_client() as client:
try:
container = client.containers.get(container_id)
container.remove(force=force)
logger.info(f"Container removed: {container_id}")
return True
except NotFound:
return False
def get_container_logs(
self,
container_id: str,
tail: int = 100
) -> Optional[str]:
"""컨테이너 로그 조회"""
with self.get_client() as client:
try:
container = client.containers.get(container_id)
logs = container.logs(tail=tail, stdout=True, stderr=True)
return logs.decode('utf-8') if isinstance(logs, bytes) else str(logs)
except NotFound:
return None
def get_container_stats(self, container_id: str) -> Optional[Dict[str, Any]]:
"""컨테이너 리소스 사용량 조회"""
with self.get_client() as client:
try:
container = client.containers.get(container_id)
stats = container.stats(stream=False)
return stats
except NotFound:
return None
# ============ 이미지 관리 ============
def list_images(self) -> List[Dict[str, Any]]:
"""이미지 목록 조회"""
with self.get_client() as client:
images = client.images.list()
return [self._image_to_dict(img) for img in images]
def pull_image(self, image_name: str) -> Dict[str, Any]:
"""이미지 풀"""
with self.get_client() as client:
image = client.images.pull(image_name)
logger.info(f"Image pulled: {image_name}")
return self._image_to_dict(image)
def remove_image(self, image_id: str, force: bool = False) -> bool:
"""이미지 삭제"""
with self.get_client() as client:
try:
client.images.remove(image_id, force=force)
logger.info(f"Image removed: {image_id}")
return True
except NotFound:
return False
# ============ 시스템 정보 ============
def get_system_info(self) -> Dict[str, Any]:
"""시스템 정보 조회"""
with self.get_client() as client:
info = client.info()
return {
"version": client.version(),
"info": info,
}
def ping(self) -> bool:
"""Podman 연결 상태 확인"""
with self.get_client() as client:
try:
client.ping()
return True
except Exception:
return False
# ============ 유틸리티 ============
def _container_to_dict(
self,
container,
detailed: bool = False
) -> Dict[str, Any]:
"""컨테이너 객체를 딕셔너리로 변환"""
data = {
"id": container.id[:12],
"full_id": container.id,
"name": container.name,
"status": container.status,
"image": container.image.tags[0] if container.image.tags else "unknown",
"labels": container.labels,
"created": str(container.attrs.get("Created", "")),
}
if detailed:
data.update({
"ports": container.ports,
"mounts": container.attrs.get("Mounts", []),
"network_settings": container.attrs.get("NetworkSettings", {}),
})
return data
def _image_to_dict(self, image) -> Dict[str, Any]:
"""이미지 객체를 딕셔너리로 변환"""
return {
"id": image.id[:12],
"full_id": image.id,
"tags": image.tags,
"size": image.attrs.get("Size", 0),
"created": str(image.attrs.get("Created", "")),
}
def get_user_container_count(self, owner: str) -> int:
"""특정 사용자의 컨테이너 수 조회"""
containers = self.list_containers(
all=True,
filters={"label": f"owner={owner}"}
)
return len(containers)
def get_user_containers(self, owner: str) -> List[Dict[str, Any]]:
"""특정 사용자의 컨테이너 목록 조회"""
return self.list_containers(
all=True,
filters={"label": f"owner={owner}"}
)
# 싱글톤 인스턴스
podman_service = PodmanService()
Pydantic 스키마 (models/container.py)
from pydantic import BaseModel, Field, validator
from typing import Optional, Dict, List
from enum import Enum
from app.config import get_settings
settings = get_settings()
class ContainerStatus(str, Enum):
CREATED = "created"
RUNNING = "running"
PAUSED = "paused"
STOPPED = "stopped"
EXITED = "exited"
class ContainerCreate(BaseModel):
"""컨테이너 생성 요청"""
image: str = Field(..., description="사용할 이미지 이름")
name: str = Field(..., min_length=1, max_length=64, pattern=r'^[a-zA-Z0-9][a-zA-Z0-9_.-]*$')
environment: Optional[Dict[str, str]] = Field(default=None, description="환경 변수")
ports: Optional[Dict[int, int]] = Field(default=None, description="포트 매핑 (컨테이너:호스트)")
memory_limit: Optional[str] = Field(default=None, description="메모리 제한 (예: 512m, 1g)")
cpu_limit: Optional[float] = Field(default=None, ge=0.1, le=settings.max_cpus_per_container)
@validator('image')
def validate_image(cls, v):
# 이미지 화이트리스트 검증
allowed = settings.allowed_images
image_base = v.split(':')[0]
if not any(image_base.endswith(allowed_img.split(':')[0]) for allowed_img in allowed):
raise ValueError(f"이미지 '{v}'는 허용되지 않습니다. 허용된 이미지: {allowed}")
return v
@validator('memory_limit')
def validate_memory(cls, v):
if v is None:
return v
# 메모리 형식 검증 (예: 512m, 1g, 2g)
import re
if not re.match(r'^\d+[kmgKMG]?$', v):
raise ValueError("메모리 형식이 올바르지 않습니다 (예: 512m, 1g)")
return v
class ContainerResponse(BaseModel):
"""컨테이너 응답"""
id: str
full_id: str
name: str
status: str
image: str
labels: Dict[str, str]
created: str
class ContainerDetailResponse(ContainerResponse):
"""컨테이너 상세 응답"""
ports: Optional[Dict] = None
mounts: Optional[List] = None
network_settings: Optional[Dict] = None
class ContainerLogsResponse(BaseModel):
"""컨테이너 로그 응답"""
container_id: str
logs: str
class ContainerStatsResponse(BaseModel):
"""컨테이너 통계 응답"""
container_id: str
stats: Dict
class ContainerActionResponse(BaseModel):
"""컨테이너 작업 응답"""
success: bool
message: str
container_id: Optional[str] = None
인증 시스템 (routers/auth.py)
from datetime import datetime, timedelta
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel
from sqlalchemy.ext.asyncio import AsyncSession
from app.config import get_settings
from app.dependencies import get_db
router = APIRouter(prefix="/auth", tags=["인증"])
settings = get_settings()
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")
class Token(BaseModel):
access_token: str
token_type: str
class TokenData(BaseModel):
username: Optional[str] = None
class User(BaseModel):
username: str
email: Optional[str] = None
is_active: bool = True
is_admin: bool = False
class UserCreate(BaseModel):
username: str
email: str
password: str
class UserInDB(User):
hashed_password: str
# 임시 사용자 저장소 (실제로는 DB 사용)
fake_users_db = {
"admin": {
"username": "admin",
"email": "admin@example.com",
"hashed_password": pwd_context.hash("admin123"),
"is_active": True,
"is_admin": True,
},
"user": {
"username": "user",
"email": "user@example.com",
"hashed_password": pwd_context.hash("user123"),
"is_active": True,
"is_admin": False,
}
}
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
return pwd_context.hash(password)
def get_user(username: str) -> Optional[UserInDB]:
if username in fake_users_db:
user_dict = fake_users_db[username]
return UserInDB(**user_dict)
return None
def authenticate_user(username: str, password: str) -> Optional[UserInDB]:
user = get_user(username)
if not user:
return None
if not verify_password(password, user.hashed_password):
return None
return user
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, settings.secret_key, algorithm=settings.algorithm)
return encoded_jwt
async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="인증 정보를 확인할 수 없습니다",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
token_data = TokenData(username=username)
except JWTError:
raise credentials_exception
user = get_user(username=token_data.username)
if user is None:
raise credentials_exception
return user
async def get_current_active_user(current_user: User = Depends(get_current_user)) -> User:
if not current_user.is_active:
raise HTTPException(status_code=400, detail="비활성화된 사용자입니다")
return current_user
async def get_admin_user(current_user: User = Depends(get_current_active_user)) -> User:
if not current_user.is_admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="관리자 권한이 필요합니다"
)
return current_user
@router.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
"""로그인하여 액세스 토큰 발급"""
user = authenticate_user(form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="아이디 또는 비밀번호가 올바르지 않습니다",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=settings.access_token_expire_minutes)
access_token = create_access_token(
data={"sub": user.username}, expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}
@router.get("/me", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
"""현재 로그인한 사용자 정보 조회"""
return current_user
컨테이너 라우터 (routers/containers.py)
from fastapi import APIRouter, Depends, HTTPException, status, Query
from typing import List, Optional
import logging
from app.config import get_settings
from app.models.container import (
ContainerCreate,
ContainerResponse,
ContainerDetailResponse,
ContainerLogsResponse,
ContainerStatsResponse,
ContainerActionResponse,
)
from app.services.podman_service import podman_service
from app.routers.auth import get_current_active_user, get_admin_user, User
router = APIRouter(prefix="/containers", tags=["컨테이너"])
settings = get_settings()
logger = logging.getLogger(__name__)
@router.get("", response_model=List[ContainerResponse])
async def list_containers(
all: bool = Query(False, description="중지된 컨테이너 포함 여부"),
current_user: User = Depends(get_current_active_user)
):
"""
컨테이너 목록 조회
- 일반 사용자: 본인 소유 컨테이너만 조회
- 관리자: 모든 컨테이너 조회
"""
if current_user.is_admin:
containers = podman_service.list_containers(all=all)
else:
containers = podman_service.get_user_containers(current_user.username)
if not all:
containers = [c for c in containers if c["status"] == "running"]
return containers
@router.get("/{container_id}", response_model=ContainerDetailResponse)
async def get_container(
container_id: str,
current_user: User = Depends(get_current_active_user)
):
"""특정 컨테이너 상세 정보 조회"""
container = podman_service.get_container(container_id)
if not container:
raise HTTPException(status_code=404, detail="컨테이너를 찾을 수 없습니다")
# 권한 검사
if not current_user.is_admin:
if container.get("labels", {}).get("owner") != current_user.username:
raise HTTPException(status_code=403, detail="접근 권한이 없습니다")
return container
@router.post("", response_model=ContainerResponse, status_code=status.HTTP_201_CREATED)
async def create_container(
container_data: ContainerCreate,
current_user: User = Depends(get_current_active_user)
):
"""
새 컨테이너 생성
비즈니스 규칙:
- 사용자당 최대 컨테이너 수 제한
- 허용된 이미지만 사용 가능
- 리소스 제한 자동 적용
"""
# 사용자별 컨테이너 수 제한 검사
current_count = podman_service.get_user_container_count(current_user.username)
if current_count >= settings.max_containers_per_user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"컨테이너 제한 초과 (최대 {settings.max_containers_per_user}개)"
)
try:
container = podman_service.create_container(
image=container_data.image,
name=container_data.name,
owner=current_user.username,
environment=container_data.environment,
ports=container_data.ports,
memory_limit=container_data.memory_limit,
cpu_limit=container_data.cpu_limit,
)
logger.info(f"Container created: {container['id']} by {current_user.username}")
return container
except Exception as e:
logger.error(f"Container creation failed: {str(e)}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"컨테이너 생성 실패: {str(e)}"
)
@router.post("/{container_id}/start", response_model=ContainerActionResponse)
async def start_container(
container_id: str,
current_user: User = Depends(get_current_active_user)
):
"""컨테이너 시작"""
container = podman_service.get_container(container_id)
if not container:
raise HTTPException(status_code=404, detail="컨테이너를 찾을 수 없습니다")
# 권한 검사
if not current_user.is_admin:
if container.get("labels", {}).get("owner") != current_user.username:
raise HTTPException(status_code=403, detail="접근 권한이 없습니다")
success = podman_service.start_container(container_id)
if success:
logger.info(f"Container started: {container_id} by {current_user.username}")
return ContainerActionResponse(
success=True,
message="컨테이너가 시작되었습니다",
container_id=container_id
)
else:
raise HTTPException(status_code=500, detail="컨테이너 시작 실패")
@router.post("/{container_id}/stop", response_model=ContainerActionResponse)
async def stop_container(
container_id: str,
timeout: int = Query(10, ge=1, le=120),
current_user: User = Depends(get_current_active_user)
):
"""컨테이너 중지"""
container = podman_service.get_container(container_id)
if not container:
raise HTTPException(status_code=404, detail="컨테이너를 찾을 수 없습니다")
# 권한 검사
if not current_user.is_admin:
if container.get("labels", {}).get("owner") != current_user.username:
raise HTTPException(status_code=403, detail="접근 권한이 없습니다")
success = podman_service.stop_container(container_id, timeout=timeout)
if success:
logger.info(f"Container stopped: {container_id} by {current_user.username}")
return ContainerActionResponse(
success=True,
message="컨테이너가 중지되었습니다",
container_id=container_id
)
else:
raise HTTPException(status_code=500, detail="컨테이너 중지 실패")
@router.delete("/{container_id}", response_model=ContainerActionResponse)
async def remove_container(
container_id: str,
force: bool = Query(False, description="강제 삭제 여부"),
current_user: User = Depends(get_current_active_user)
):
"""컨테이너 삭제"""
container = podman_service.get_container(container_id)
if not container:
raise HTTPException(status_code=404, detail="컨테이너를 찾을 수 없습니다")
# 권한 검사
if not current_user.is_admin:
if container.get("labels", {}).get("owner") != current_user.username:
raise HTTPException(status_code=403, detail="접근 권한이 없습니다")
success = podman_service.remove_container(container_id, force=force)
if success:
logger.info(f"Container removed: {container_id} by {current_user.username}")
return ContainerActionResponse(
success=True,
message="컨테이너가 삭제되었습니다",
container_id=container_id
)
else:
raise HTTPException(status_code=500, detail="컨테이너 삭제 실패")
@router.get("/{container_id}/logs", response_model=ContainerLogsResponse)
async def get_container_logs(
container_id: str,
tail: int = Query(100, ge=1, le=10000),
current_user: User = Depends(get_current_active_user)
):
"""컨테이너 로그 조회"""
container = podman_service.get_container(container_id)
if not container:
raise HTTPException(status_code=404, detail="컨테이너를 찾을 수 없습니다")
# 권한 검사
if not current_user.is_admin:
if container.get("labels", {}).get("owner") != current_user.username:
raise HTTPException(status_code=403, detail="접근 권한이 없습니다")
logs = podman_service.get_container_logs(container_id, tail=tail)
return ContainerLogsResponse(container_id=container_id, logs=logs or "")
@router.get("/{container_id}/stats", response_model=ContainerStatsResponse)
async def get_container_stats(
container_id: str,
current_user: User = Depends(get_current_active_user)
):
"""컨테이너 리소스 사용량 조회"""
container = podman_service.get_container(container_id)
if not container:
raise HTTPException(status_code=404, detail="컨테이너를 찾을 수 없습니다")
# 권한 검사
if not current_user.is_admin:
if container.get("labels", {}).get("owner") != current_user.username:
raise HTTPException(status_code=403, detail="접근 권한이 없습니다")
stats = podman_service.get_container_stats(container_id)
if stats is None:
raise HTTPException(status_code=500, detail="통계 조회 실패")
return ContainerStatsResponse(container_id=container_id, stats=stats)
메인 애플리케이션 (main.py)
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
import logging
import time
from app.config import get_settings
from app.routers import auth, containers, images, system
# 로깅 설정
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
settings = get_settings()
app = FastAPI(
title=settings.app_name,
description="Podman 컨테이너를 관리하는 RESTful API 서버",
version="1.0.0",
docs_url="/docs",
redoc_url="/redoc",
)
# CORS 설정
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # 프로덕션에서는 특정 도메인만 허용
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 요청 로깅 미들웨어
@app.middleware("http")
async def log_requests(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
logger.info(
f"{request.method} {request.url.path} "
f"- Status: {response.status_code} "
f"- Time: {process_time:.3f}s"
)
return response
# 전역 예외 핸들러
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
logger.error(f"Unhandled exception: {str(exc)}", exc_info=True)
return JSONResponse(
status_code=500,
content={"detail": "내부 서버 오류가 발생했습니다"}
)
# 라우터 등록
app.include_router(auth.router)
app.include_router(containers.router)
# app.include_router(images.router)
# app.include_router(system.router)
@app.get("/")
async def root():
"""API 서버 상태 확인"""
return {
"name": settings.app_name,
"status": "running",
"docs": "/docs"
}
@app.get("/health")
async def health_check():
"""헬스 체크 엔드포인트"""
from app.services.podman_service import podman_service
podman_ok = podman_service.ping()
return {
"status": "healthy" if podman_ok else "degraded",
"podman": "connected" if podman_ok else "disconnected"
}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
실행 및 테스트
서버 실행
# 개발 모드 실행
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
# 프로덕션 실행
uvicorn app.main:app --workers 4 --host 0.0.0.0 --port 8000
API 테스트
# 1. 로그인하여 토큰 발급
TOKEN=$(curl -s -X POST "http://localhost:8000/auth/token" \
-H "Content-Type: application/x-www-form-urlencoded" \
-d "username=user&password=user123" | jq -r '.access_token')
echo $TOKEN
# 2. 컨테이너 목록 조회
curl -s -X GET "http://localhost:8000/containers" \
-H "Authorization: Bearer $TOKEN" | jq
# 3. 컨테이너 생성
curl -s -X POST "http://localhost:8000/containers" \
-H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-d '{
"image": "nginx",
"name": "my-nginx",
"ports": {"80": 8080}
}' | jq
# 4. 컨테이너 시작
curl -s -X POST "http://localhost:8000/containers/my-nginx/start" \
-H "Authorization: Bearer $TOKEN" | jq
# 5. 컨테이너 로그 조회
curl -s -X GET "http://localhost:8000/containers/my-nginx/logs?tail=50" \
-H "Authorization: Bearer $TOKEN" | jq
# 6. 컨테이너 중지 및 삭제
curl -s -X POST "http://localhost:8000/containers/my-nginx/stop" \
-H "Authorization: Bearer $TOKEN" | jq
curl -s -X DELETE "http://localhost:8000/containers/my-nginx" \
-H "Authorization: Bearer $TOKEN" | jq
Swagger UI 접속
브라우저에서 http://localhost:8000/docs로 접속하면 대화형 API 문서를 확인할 수 있습니다.
프로덕션 배포
systemd 서비스 등록
# /etc/systemd/system/podman-api-server.service
[Unit]
Description=Podman API Server
After=network.target podman.socket
[Service]
Type=simple
User=podman-api
Group=podman-api
WorkingDirectory=/opt/podman-api-server
Environment="PATH=/opt/podman-api-server/venv/bin"
ExecStart=/opt/podman-api-server/venv/bin/uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4
Restart=always
RestartSec=5
[Install]
WantedBy=multi-user.target
sudo systemctl daemon-reload
sudo systemctl enable --now podman-api-server
Nginx 리버스 프록시 (TLS 종단)
# /etc/nginx/conf.d/podman-api.conf
server {
listen 443 ssl http2;
server_name api.example.com;
ssl_certificate /etc/letsencrypt/live/api.example.com/fullchain.pem;
ssl_certificate_key /etc/letsencrypt/live/api.example.com/privkey.pem;
location / {
proxy_pass http://127.0.0.1:8000;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
}
}
확장 아이디어
이 기본 구조를 바탕으로 다음과 같은 기능을 추가할 수 있습니다.
- WebSocket 실시간 로그: 컨테이너 로그를 실시간 스트리밍
- 과금 시스템: 컨테이너 사용 시간 기반 과금
- 자동 스케일링: 부하에 따른 컨테이너 자동 확장/축소
- 백업/복원: 컨테이너 스냅샷 및 복원 기능
- 템플릿: 자주 사용하는 컨테이너 구성을 템플릿으로 저장
- 알림: Slack, Telegram, 이메일 연동
- 멀티 호스트: 여러 Podman 호스트 통합 관리
마무리
FastAPI와 Podman API를 결합하면 강력하고 안전한 컨테이너 관리 시스템을 구축할 수 있습니다. 핵심은 Podman 소켓을 직접 노출하지 않고, 중간 레이어에서 인증, 권한, 비즈니스 로직을 처리하는 것입니다. 이 가이드를 기반으로 자신만의 컨테이너 플랫폼을 구축해 보세요.