Escrevendo uma implementação de um novo Orquestrador

Aqui descreveremos como implementar o suporte a um novo orquestrador, quais classes abstratas temos que implementar, com organizar o código de forma que fique análogo ao que já existe e por isso fique mais fácil de compreender, à medida que o codebase for crescendo.

A implementação do suporte a um novo Orquestrador começa na classe asgard.backends.base.Orchestrator.

class asgard.backends.base.Orchestrator(agents_backend: asgard.backends.base.AgentsBackend, apps_backend: asgard.backends.base.AppsBackend)[código fonte]

Classe abstrata que mapeia todas as ações que um orquestrador pode excutar. As depdenências injetadas aqui são implementações que efetivamente falam com cada um dos backends suportados.

__init__(agents_backend: asgard.backends.base.AgentsBackend, apps_backend: asgard.backends.base.AppsBackend) → None[código fonte]

Initialize self. See help(type(self)) for accurate signature.

get_agents(user: asgard.models.user.User, account: asgard.models.account.Account) → List[asgard.models.agent.Agent][código fonte]

Perceba que o __init__() já recebe dois parametros, que são dois backends. Um para Apps e outro para Agents. Como o Orchestrator é uma classse abstrata, precisamos implementar todos os métodos. Cada método tem relação com algum backend, nesse caso a implementação vai usar esse backend específico para poder obter as informações.

Vejamos os métodos de cada backend.

Agents backend

class asgard.backends.base.AgentsBackend[código fonte]
get_agents(user: asgard.models.user.User, account: asgard.models.account.Account) → List[asgard.models.agent.Agent][código fonte]

Retorna todos os Agents da conta account.

Vamos pegar como exemplo o método get_agents(). Esse método é quem deve retornar a lista de agents desse novo orquestrador. Então uma possível implementação, considerando um backend fictício chamado K8S, poderia ser:

Modelo:

from asgard.models.agent import Agent


class K8SAgent(Agent):
    type = "K8S"

    def is_from_account(self, account) -> bool:
        return self.attributes["owner"] == account.owner

Código do AgentsBackend:

from asgard.backend.k8s.models.agent import K8SAgent
from asgard.backends.base import AgentsBackend
from asgard.conf import settings
from asgard.http.client import http_client
from asgard.models.account import Account
from asgard.models.user import User


class K8SAgentBackend(AgentsBackend):

  async def get_agents(sefl, user: User, account: account) -> List[K8SAgent]:
    async with http_client.get(settings.K8S_API_URL) as response:
      agents: List[K8SAgent] = []
      data = await response.json()
      for agent in data["objects"]:
        new_agent = K8SAgent(id=agent["id"], attributes=agent["labels"], ...)
        if new_agent.is_from_account(account):
          agents.append(new_agent)
      return agents

Código do K8SOrchestrator:

from asgard.backend.k8s.models.agent import K8SAgent
from asgard.backens.base import Orchestrator
from asgard.models.account import Account
from asgard.models.user import User


class K8SOrchestrator(Orquestrador):
    async def get_agents(self, user, account) -> List[K8SAgent]:
        return self.agents_backend.get_agents(user, account)

Com uma implementação nessa linha, seria possível listar todos os agents desse novo Orchestrador dessa forma:

from asgard.backends.k8s.agent import K8SAgentBackend
from asgard.backends.k8s.impl import K8SOrchestrator

orchestrator = K8SOrchestrator(agents_backend=K8SAgentBackend(), ...)
agents = await orchestrator.get_agents(user, account)

Pensando nos endpoints HTTP, que são todos autenticados, o valor de user (asgard.models.user.User) e account (asgard.models.account.Account) são descobertos assim que a view HTTP começa a rodar.