Implementando Keda com ServiceBus e tópicos

Depois de passar um tempo quebrando a cabeça tentando achar material para fazer essa implementação finalmente consegui termina-la e gostaria de compartilhar quais foram as questões que tive com essa tarefa, então vamos por partes.

O que é o Keda?

Keda é uma abreviação para Kubernetes Event-driven Autoscaling, uma aplicação open-source e com apoio da CNCF, ele vem a ser um objeto que aumenta o nível de gerenciamento para o HPA (Horizontal Pod Autoscaling), um objeto do Kubernetes responsável por escalar pods com base em cpu e memória, porém, o Keda vai mais longe e utiliza-se de eventos para poder realizar esse trabalho e o que são esses eventos? Eles podem ser requisições vindo de um Kafka, RabbitMQ, Amazon SQS e também de um Azure Service Bus que é o serviço do qual falaremos nesse artigo.

O que é o Azure Service Bus?

Ele é um serviço de mensageria fornecido pela Microsoft, ele é baseado em Kafka mas com a opção de também poder utiliza-lo como um RabbitMQ, tudo isso devido a duas opções dele que são Queues e Topics, essas informações podemos ver mais aqui.

Implementando o Keda

Para implementar o Keda podemos utilizar o Helm para realizar seu deploy, para isso utilizaremos os seguintes comandos:

• Adicionar o repositório do Keda

helm repo add kedacore https://kedacore.github.io/charts

• Atualizar esse repositório

helm repo update

• Instalar o Keda pelo Helm Chart

kubectl create namespace keda
helm install keda kedacore/keda --namespace keda

Essa instalação criará um Custom Resource Definition chamado keda.sh onde será criado novos objetos, mais especificamente 4:

• ClusterTriggerAuthentication
• ScaledJob
• ScaledObject
• TriggerAuthentication

Para esse tutorial iremos utilizar dois desses objetos, ScaledObject e TriggerAuthentication, irei explicar sobre eles:

• ScaledObject
  • Ele é o objeto do Kubernetes que faz a criação do HPA mas utilizando um evento como métrica e também seu gerenciamento.
• TriggerAuthentication
  • É o objeto que consultará a secret onde contém a Key de acesso do ServiceBus e caso esteja certo irá validar, o ScaledObject irá consultar esse objeto para realizar a autenticação corretamente.

Junto disso tudo também precisaremos criar mais dois objetos:

• Deployment
  • O deployment será quem irá consumir as mensagens do ServiceBus para que o Keda possa escala-lo.
• Secret
  • Será onde irá conter a Key de acesso do ServiceBus.

Vamos a mão na massa:

• Iremos começar criando a secret, para isso você irá no Azure Service Bus crie um tópico, nesse tópico crie a Subscription e cadastre a Access Key. Após isso precisamos converter essa string para base64 para poder armazenar na secret então em um linux iremos utilizar o seguinte comando:

echo '<access_key>' | base64

e com a saída iremos configurar em:

apiVersion: v1
kind: Secret
metadata:
  name: secret-name
type: Opaque
data:
  connection: <saída_do_comando_echo>

Salve esse arquivo e agora iremos subi-lo no cluster Kubernetes:

kubectl apply -f <nome_do_arquivo> --namespace <opcional>

• Iremos subir o TriggerAuthentication que consultará o Secret e será consultado pelo ScaledObject.

apiVersion: keda.sh/v1alpha1
kind: TriggerAuthentication
metadata:
  name: azure-servicebus-auth
spec:
  secretTargetRef:
    - key: connection
      name: secret-name # nome da secret
      parameter: connection # chave da secret

• Nesse momento iremos subir o Deployment que será referenciado posteriormente no ScaledObject, para isso utilizei um código em Python que consome mensagens do tópico, que é o seguinte:

import os
from azure.servicebus import ServiceBusClient


CONNECTION_STR = "<access_key_service_bus>"
TOPIC_NAME = "<nome_do_topico>"
SUBSCRIPTION_NAME = "<nome_da_subscription>"

servicebus_client = ServiceBusClient.from_connection_string(conn_str=CONNECTION_STR)

with servicebus_client:
    # get the Subscription Receiver object for the subscription    
    receiver = servicebus_client.get_subscription_receiver(topic_name=TOPIC_NAME, subscription_name=SUBSCRIPTION_NAME, max_wait_time=5)
    with receiver:
        for msg in receiver:
            print("Received: " + str(msg))
            # complete the message so that the message is removed from the subscription
            receiver.complete_message(msg)

Essa aplicação irá subir no Kubernetes e começar a consumir as mensagens da fila e deleta-las.

• Por último iremos subir o ScaledObject

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: azure-servicebus-topic-scaledobject
  namespace: namespace-scaledobject
spec:
  scaleTargetRef:
    name: deployment-referencia
  pollingInterval: 30
  cooldownPeriod: 60
  minReplicaCount: 1
  maxReplicaCount: 4

  triggers:
  - type: azure-servicebus
    metadata:
      topicName: <topic_name>
      subscriptionName: <subscription_name>
      namespace: <servicebus_namespace_name>
      messageCount: "5"
    authenticationRef:
        name: <nome_do_authentication_trigger>

Dessa forma, irá funcionar assim:

Scaled Object irá escalar o -> Deployment mas ele só funcionará pois ele estará validando o acesso ao Service Bus ao checar o Trigger Authentication que estará checando a -> Secret.

Depois de todos esses passos teremos nossa aplicação escalando através de eventos graças ao Keda.

Sendo assim, conseguiremos criar um cluster com escalabilidade inteligente e aplicações com maior resiliência.

Referências:
https://keda.sh/
https://keda.sh/docs/2.7/scalers/azure-service-bus/
https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-messaging-overview
https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-queues-topics-subscriptions
https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/servicebus/azure-servicebus/samples/sync_samples/receive_subscription.py