logboard/client/app/main.py

#!/usr/bin/env python3
"""
Клиент для отправки логов на удаленный сервер LogBoard
Автор: Сергей Антропов
Сайт: https://devops.org.ru
"""

import asyncio
import json
import logging
import os
import sys
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional

import aiofiles
import aiohttp
import docker
from docker.errors import DockerException

# Настройка логирования
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.StreamHandler(sys.stdout),
        logging.FileHandler('/var/log/logboard-client.log')
    ]
)
logger = logging.getLogger(__name__)


class LogBoardClient:
    """Клиент для отправки логов в LogBoard сервер"""
    
    def __init__(self, server_url: str, api_key: str, hostname: str):
        """
        Инициализация клиента
        
        Args:
            server_url: URL сервера LogBoard
            api_key: API ключ для аутентификации
            hostname: Имя хоста для идентификации
        """
        self.server_url = server_url.rstrip('/')
        self.api_key = api_key
        self.hostname = hostname
        self.session: Optional[aiohttp.ClientSession] = None
        try:
            # Используем тот же способ, что и в основном сервисе
            self.docker_client = docker.from_env()
            # Проверяем подключение
            self.docker_client.ping()
            logger.info("Docker клиент успешно инициализирован")
        except Exception as e:
            logger.error(f"Критическая ошибка Docker клиента: {e}")
            raise
        
    async def __aenter__(self):
        """Асинхронный контекстный менеджер - вход"""
        self.session = aiohttp.ClientSession(
            headers={
                'Authorization': f'Bearer {self.api_key}',
                'Content-Type': 'application/json',
                'User-Agent': 'LogBoard-Client/1.0'
            }
        )
        return self
        
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Асинхронный контекстный менеджер - выход"""
        if self.session:
            await self.session.close()
            
    async def send_logs(self, container_name: str, logs: List[str]) -> bool:
        """
        Отправка логов на сервер
        
        Args:
            container_name: Имя контейнера
            logs: Список строк логов
            
        Returns:
            bool: True если отправка успешна, False в противном случае
        """
        if not self.session:
            logger.error("Сессия не инициализирована")
            return False
            
        payload = {
            "hostname": self.hostname,
            "container_name": container_name,
            "logs": logs,
            "timestamp": datetime.utcnow().isoformat()
        }
        
        try:
            async with self.session.post(
                f"{self.server_url}/api/logs/remote",
                json=payload,
                timeout=aiohttp.ClientTimeout(total=30)
            ) as response:
                if response.status == 200:
                    logger.info(f"Логи контейнера {container_name} успешно отправлены")
                    return True
                else:
                    logger.error(f"Ошибка отправки логов: {response.status} - {await response.text()}")
                    return False
        except Exception as e:
            logger.error(f"Ошибка при отправке логов: {e}")
            return False
            
    def get_containers(self) -> List[Dict]:
        """
        Получение списка контейнеров Docker
        
        Returns:
            List[Dict]: Список контейнеров с информацией
        """
        try:
            containers = []
            for container in self.docker_client.containers.list():
                containers.append({
                    "id": container.id,
                    "name": container.name,
                    "status": container.status,
                    "image": container.image.tags[0] if container.image.tags else container.image.id,
                    "created": container.attrs['Created']
                })
            return containers
        except DockerException as e:
            logger.error(f"Ошибка при получении списка контейнеров: {e}")
            return []
            
    async def collect_container_logs(self, container_name: str, lines: int = 100) -> List[str]:
        """
        Сбор логов контейнера
        
        Args:
            container_name: Имя контейнера
            lines: Количество строк логов для сбора
            
        Returns:
            List[str]: Список строк логов
        """
        try:
            container = self.docker_client.containers.get(container_name)
            logs = container.logs(
                stdout=True,
                stderr=True,
                tail=lines,
                timestamps=True
            ).decode('utf-8')
            
            return logs.splitlines() if logs else []
        except DockerException as e:
            logger.error(f"Ошибка при получении логов контейнера {container_name}: {e}")
            return []
            
    async def monitor_containers(self, interval: int = 60):
        """
        Мониторинг контейнеров и отправка логов
        
        Args:
            interval: Интервал мониторинга в секундах
        """
        logger.info(f"Запуск мониторинга контейнеров с интервалом {interval} секунд")
        
        while True:
            try:
                containers = self.get_containers()
                logger.info(f"Найдено {len(containers)} контейнеров")
                
                for container in containers:
                    container_name = container['name']
                    if container['status'] == 'running':
                        logs = await self.collect_container_logs(container_name)
                        if logs:
                            await self.send_logs(container_name, logs)
                            
            except Exception as e:
                logger.error(f"Ошибка в мониторинге: {e}")
                
            await asyncio.sleep(interval)


async def main():
    """Основная функция"""
    # Получение переменных окружения
    server_url = os.getenv('LOGBOARD_SERVER_URL', 'http://localhost:8000')
    api_key = os.getenv('LOGBOARD_API_KEY')
    hostname = os.getenv('HOSTNAME', os.uname().nodename)
    interval = int(os.getenv('LOGBOARD_INTERVAL', '60'))
    
    if not api_key:
        logger.error("LOGBOARD_API_KEY не установлен")
        sys.exit(1)
        
    logger.info(f"Запуск LogBoard клиента для хоста: {hostname}")
    logger.info(f"Подключение к серверу: {server_url}")
    
    async with LogBoardClient(server_url, api_key, hostname) as client:
        await client.monitor_containers(interval)


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logger.info("Получен сигнал прерывания, завершение работы")
    except Exception as e:
        logger.error(f"Критическая ошибка: {e}")
        sys.exit(1)