DevOpsLab/app/core/docker_client.py

"""
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)
            }