""" Docker клиент для управления контейнерами Автор: Сергей Антропов Сайт: https://devops.org.ru """ import docker from docker import APIClient import os from typing import List, Dict, Optional from app.core.config import settings import logging logger = logging.getLogger(__name__) class DockerClient: """Клиент для работы с Docker API""" def __init__(self): """Инициализация Docker клиента (ленивая инициализация)""" self._client = None @property def client(self): """Ленивая инициализация Docker клиента""" if self._client is None: try: # Временно удаляем DOCKER_HOST из окружения, если он установлен # Это необходимо, так как Docker SDK может неправильно парсить его original_docker_host = os.environ.pop("DOCKER_HOST", None) try: # Пробуем docker.from_env() без DOCKER_HOST logger.info("Trying docker.from_env() without DOCKER_HOST") self._client = docker.from_env() self._client.ping() logger.info("Docker client initialized successfully with docker.from_env()") except Exception as e1: logger.warning(f"docker.from_env() failed: {e1}") # Если from_env не работает, пробуем прямой base_url try: # Используем прямой путь к Docker socket base_url = "unix:///var/run/docker.sock" logger.info(f"Trying direct base_url: {base_url}") self._client = docker.DockerClient(base_url=base_url) self._client.ping() logger.info("Docker client initialized successfully with direct base_url") except Exception as e2: logger.error(f"Direct base_url also failed: {e2}") # Последняя попытка - используем APIClient try: logger.info("Trying APIClient as last resort") api_client = APIClient(base_url="unix:///var/run/docker.sock") api_client.version() # Если APIClient работает, создаем DockerClient self._client = docker.DockerClient(base_url="unix:///var/run/docker.sock") self._client.ping() logger.info("Docker client initialized successfully with APIClient") except Exception as e3: logger.error(f"All methods failed. Last error: {e3}") raise finally: # Восстанавливаем DOCKER_HOST, если он был установлен if original_docker_host: os.environ["DOCKER_HOST"] = original_docker_host except Exception as e: logger.error(f"Failed to initialize Docker client: {e}") logger.error(f"DOCKER_HOST env: {os.getenv('DOCKER_HOST', 'not set')}") logger.error(f"Settings DOCKER_HOST: {settings.DOCKER_HOST}") import traceback logger.error(traceback.format_exc()) raise return self._client def list_images(self) -> List[Dict]: """Список всех Docker образов""" try: images = self.client.images.list() return [ { "id": img.id, "tags": img.tags, "created": img.attrs.get("Created"), "size": img.attrs.get("Size", 0) } for img in images ] except Exception as e: logger.error(f"Error listing images: {e}") return [] def list_containers(self, all: bool = False) -> List[Dict]: """Список контейнеров""" try: containers = self.client.containers.list(all=all) return [ { "id": container.id, "name": container.name, "status": container.status, "image": container.image.tags[0] if container.image.tags else container.image.id, "ports": container.ports, "created": container.attrs.get("Created") } for container in containers ] except Exception as e: logger.error(f"Error listing containers: {e}") return [] def get_container_ip(self, container_name: str) -> Optional[str]: """Получение IP адреса контейнера""" try: container = self.client.containers.get(container_name) network_settings = container.attrs.get("NetworkSettings", {}) networks = network_settings.get("Networks", {}) # Ищем IP в первой доступной сети for network_name, network_info in networks.items(): ip = network_info.get("IPAddress") if ip: return ip return None except Exception as e: logger.error(f"Error getting container IP for {container_name}: {e}") return None def create_container( self, image: str, name: str, command: Optional[str] = None, environment: Optional[Dict] = None, volumes: Optional[Dict] = None, network: Optional[str] = None, privileged: bool = False ) -> Dict: """Создание контейнера""" try: container = self.client.containers.run( image=image, name=name, command=command, environment=environment or {}, volumes=volumes or {}, network=network, privileged=privileged, detach=True, remove=False ) return { "success": True, "id": container.id, "name": container.name, "status": container.status } except Exception as e: logger.error(f"Error creating container: {e}") return { "success": False, "error": str(e) } def stop_container(self, container_name: str) -> bool: """Остановка контейнера""" try: container = self.client.containers.get(container_name) container.stop() return True except Exception as e: logger.error(f"Error stopping container {container_name}: {e}") return False def remove_container(self, container_name: str, force: bool = False) -> bool: """Удаление контейнера""" try: container = self.client.containers.get(container_name) container.remove(force=force) return True except Exception as e: logger.error(f"Error removing container {container_name}: {e}") return False def get_container_logs(self, container_name: str, tail: int = 100) -> str: """Получение логов контейнера""" try: container = self.client.containers.get(container_name) logs = container.logs(tail=tail, timestamps=True) return logs.decode('utf-8', errors='replace') except Exception as e: logger.error(f"Error getting logs for {container_name}: {e}") return "" def exec_command(self, container_name: str, command: str) -> Dict: """Выполнение команды в контейнере""" try: container = self.client.containers.get(container_name) result = container.exec_run(command) return { "success": result.exit_code == 0, "output": result.output.decode('utf-8', errors='replace'), "exit_code": result.exit_code } except Exception as e: logger.error(f"Error executing command in {container_name}: {e}") return { "success": False, "error": str(e) }