Código fonte para asgard.backends.mesos.impl

from typing import List, Optional

from asgard.backends.base import Orchestrator, AgentsBackend, Interval
from asgard.backends.mesos.models.agent import MesosAgent
from asgard.backends.mesos.models.converters import MesosAgentConverter
from asgard.clients.mesos.client import MesosClient
from asgard.conf import settings
from asgard.models.account import Account
from asgard.models.agent import Agent
from asgard.models.app import App, AppStats
from asgard.models.user import User
from hollowman import log


[documentos]async def populate_apps(agent: MesosAgent): try: agent.applications = await agent.apps() agent.total_apps = len(agent.applications) except Exception as e: agent.add_error(field_name="total_apps", error_msg="INDISPONIVEL") log.logger.exception( { "event": "Erro buscando agent applications", "agent": agent.id, "hostname": agent.hostname, } )
[documentos]class MesosAgentsBackend(AgentsBackend):
[documentos] async def get_agents( # type: ignore self, user: User, account: Account ) -> List[MesosAgent]: async with MesosClient(*settings.MESOS_API_URLS) as mesos: filtered_agents: List[MesosAgent] = [] client_agents = await mesos.get_agents() for client_agent in client_agents: agent = MesosAgentConverter.to_asgard_model(client_agent) if not agent.attr_has_value("owner", account.owner): continue await populate_apps(agent) await agent.calculate_stats() filtered_agents.append(agent) return filtered_agents
[documentos] async def get_by_id( self, agent_id: str, user: User, account: Account ) -> Optional[MesosAgent]: async with MesosClient(*settings.MESOS_API_URLS) as mesos: client_agent = await mesos.get_agent_by_id(agent_id=agent_id) if client_agent: agent = MesosAgentConverter.to_asgard_model(client_agent) if agent.attr_has_value("owner", account.owner): await populate_apps(agent) await agent.calculate_stats() return agent return None
[documentos] async def get_apps_running(self, user: User, agent: Agent) -> List[App]: if agent: return agent.applications return []
[documentos]class MesosOrchestrator(Orchestrator):
[documentos] async def get_agents(self, user: User, account: Account) -> List[Agent]: return await self.agents_backend.get_agents(user, account)
[documentos] async def get_agent_by_id( self, agent_id: str, user: User, account: Account ) -> Optional[Agent]: return await self.agents_backend.get_by_id(agent_id, user, account)
[documentos] async def get_apps_running_for_agent( self, user: User, agent: Agent ) -> List[App]: return await self.agents_backend.get_apps_running(user, agent)
[documentos] async def get_app_stats( self, app: App, interval: Interval, user: User, account: Account ) -> AppStats: return await self.apps_backend.get_app_stats( app, interval, user, account )