DEV Community: Ezequiel Lopes

Como construir uma aplicação escalável com Terraform e AWS

Ezequiel Lopes — Mon, 01 Jul 2024 11:15:31 +0000

Neste artigo, mostrarei como construir um sistema escalável e resiliente na AWS utilizando Terraform. O sistema será escalável horizontalmente e distribuído em diferentes zonas de disponibilidade. Além disso, terá um health check para verificar a saúde das instâncias.

Resiliência

Um sistema resiliente é capaz de se adaptar a condições anormais e manter seu funcionamento total ou parcial. Para isso, existem várias estratégias tanto para a infraestrutura quanto para o software, como políticas de retry, mensageria, sistema distribuído, health check, redundância, entre outros.

O que acontece se o volume de clientes acessando nosso site for maior do que o planejado? O sistema suportaria esses usuários? O que acontece se a zona de disponibilidade em que nosso sistema está localizado ficar fora do ar? Ou se recebermos um ataque de negação de serviço? Esses eventos não são nossa culpa, mas precisamos ter estratégias para nos proteger quando esse tipo de evento acontecer.

Construiremos uma estratégia que minimiza os riscos, mas não garante resiliência total. Para alcançar resiliência total do seu sistema, prepare-se para criar um servidor no espaço, como mencionado no livro "Designing Data-Intensive Applications".

Terraform

Terraform é uma ferramenta para provisionamento de infraestrutura como código. Facilitando muito todo o processo de criação, edição e deleção de diversos recursos, incluindo redes virtuais, máquinas e sistemas de armazenamento, entre outros. Não se limitando apenas à AWS, com Terraform é possível integrar-se a diversos provedores de nuvem, como Azure, GCP, Oracle e outros, além de ser possível importar projetos existentes neles.

Auto Scaling

Auto Scaling é a capacidade de aumentar ou reduzir o número de workloads de acordo com as necessidades a qualquer momento. A AWS oferece um serviço de Auto Scaling que podemos integrar ao nosso sistema. Permitindo criar mais máquinas com base em métricas como utilização de CPU, memória ou número de requisições. Quando a demanda diminui, as máquinas podem ser removidas, garantindo que os recursos sejam utilizados de forma eficiente e econômica.

Load Balancer

Com várias máquinas trabalhando em conjunto, o Load Balancer tem o papel de decidir para onde cada requisição deve ser direcionada. Existem várias estratégias de balanceamento, desde as mais simples, onde cada requisição é distribuída igualmente entre as máquinas, até as mais complexas, que consideram a carga atual de cada instância.

Bora lá

Nosso sistema funcionará da seguinte maneira:

Um load balancer na frente que distribuirá as requisições,
Dois security groups, um para o load balancer e outro para as instâncias EC2,
Auto Scaling que será responsável por criar as máquinas,
Um template que terá as configurações das instâncias.

Primeiramente, vamos executar o comando para iniciar nosso projeto Terraform.

terraform init

Vamos criar o arquivo de variáveis onde definiremos valores como o nome da aplicação, a região que utilizaremos, o tipo da instância, os valores de saúde, entre outros.

// variables.tf
variable "aws_region" {
  type        = string
  description = "AWS region where the resources will be deployed."
  default     = "us-east-2"
}

variable "environment" {
  type        = string
  description = "Name of the deployment environment (e.g., dev, stage, prod)."
  default     = "dev"
}

variable "service_name" {
  type        = string
  description = "Name of the service/application to be deployed."
  default     = "app-go"
}

variable "instance_config" {
  description = "Configuration for the EC2 instances."
  type = object({
    ami      = string
    type     = string
    key_name = optional(string, null)
  })
  default = {
    ami  = "ami-09040d770ffe2224f"
    type = "t2.medium"
  }
}

variable "alb_health_check_config" {
  description = "Configuration for the Application Load Balancer (ALB) health checks."
  nullable    = true
  default     = {}
  type = object({
    enabled             = optional(bool, true)
    healthy_threshold   = optional(number, 3)
    interval            = optional(number, 30)
    matcher             = optional(string, "200")
    path                = optional(string, "/healthz")
    port                = optional(number, 80)
    protocol            = optional(string, "HTTP")
    timeout             = optional(number, 5)
    unhealthy_threshold = optional(number, 3)
  })
}

variable "autoscaling_group_config" {
  description = "Configuration for the Auto Scaling group."
  default     = {}
  type = object({
    desired_capacity          = optional(number, 2)
    min_size                  = optional(number, 1)
    max_size                  = optional(number, 5)
    health_check_grace_period = optional(number, 320) // Grace period for health checks (seconds).
    health_check_type         = optional(string, "ELB")
    force_delete              = optional(bool, false)
  })
}

variable "autoscaling_policy_cpu" {
  description = "Configuration for the CPU utilization auto scaling policy."
  nullable    = true
  default     = {}
  type = object({
    enabled          = optional(bool, true)
    name             = optional(string, "CPU utilization")
    disable_scale_in = optional(bool, false)
    target_value     = optional(number, 40)
  })
}

variable "autoscaling_policy_alb" {
  description = "Configuration for the ALB request rate auto scaling policy."
  nullable    = true
  default     = {}
  type = object({
    enabled          = optional(bool, true)
    name             = optional(string, "Load balancer request per minute")
    disable_scale_in = optional(bool, false)
    target_value     = optional(number, 40)
  })
}

No security group, iremos configurar da seguinte forma: no EC2, iremos permitir apenas tráfego vindo do load balancer na porta 80, e no load balancer também iremos permitir apenas a porta 80 vinda da internet.

// security_groups.tf
resource "aws_security_group" "alb" {
  name        = "${local.namespaced_service_name}-alb"
  description = "Allow HTTP internet traffic"
  vpc_id      = aws_vpc.this.id

  ingress {
    from_port   = 80
    to_port     = 80
    protocol    = "tcp"
    cidr_blocks = local.internet_cidr_block
  }

  egress {
    from_port   = 0
    to_port     = 0
    protocol    = "-1"
    cidr_blocks = local.internet_cidr_block
  }

  tags = {
    "Name" = "${local.namespaced_service_name}-alb"
  }
}

resource "aws_security_group" "autoscaling_group" {
  name        = "${local.namespaced_service_name}-autoscaling-group"
  description = "Allow HTTP traffic only through the load balancer"
  vpc_id      = aws_vpc.this.id

  ingress {
    from_port       = 80
    to_port         = 80
    protocol        = "tcp"
    security_groups = [aws_security_group.alb.id]

  }

  egress {
    from_port   = 0
    to_port     = 0
    protocol    = "-1"
    cidr_blocks = local.internet_cidr_block
    description = "Allow all traffic"
  }

  tags = {
    "Name" = "${local.namespaced_service_name}-autoscaling-group"
  }
}

Aqui está a configuração do template que utilizaremos nas instâncias. Temos também um script app-setup.sh de inicialização, que será responsável por configurar e iniciar nosso projeto.

// template.tf
resource "aws_launch_template" "this" {
  name_prefix   = local.namespaced_service_name
  image_id      = var.instance_config.ami
  instance_type = var.instance_config.type
  user_data     = filebase64("app-setup.sh")

  monitoring {
    enabled = true
  }

  network_interfaces {
    associate_public_ip_address = true
    security_groups             = [aws_security_group.autoscaling_group.id]
  }

  tag_specifications {
    resource_type = "instance"
    tags = {
      "Name" = "${local.namespaced_service_name}-server"
    }
  }
}

O app-setup.sh contém as configurações necessárias para iniciar a aplicação de exemplo em Golang. Ele instala o Golang, clona o repositório do projeto de exemplo (https://github.com/infezek/app-alb-terraform.git) e inicia a aplicação.

O servidor de exemplo possui duas rotas: a rota / retorna uma mensagem de "ok" e a zona de disponibilidade em que a instância está rodando, enquanto a rota /healthz retorna uma mensagem de "ok" nos primeiros 60 segundos e depois retorna um erro, fazendo com que o health check a identifique e reinicie a instância.

#!/bin/bash
sudo apt update -y
sudo apt upgrade -y

cd /var/tmp/
wget https://dl.google.com/go/go1.22.4.linux-amd64.tar.gz
sudo rm -rf /usr/local/go 
sudo tar -C /usr/local -xzf go1.22.4.linux-amd64.tar.gz

git clone https://github.com/infezek/app-alb-terraform.git alb-app
cd /var/tmp/alb-app
GOCACHE=/tmp/gocache GOOS=linux GOARCH=amd64 /usr/local/go/bin/go build -o /home/ubuntu/serve -buildvcs=false

sudo iptables -t nat -A PREROUTING -p tcp --dport 80 -j REDIRECT --to-ports 8080
sudo cp alb-http.service /etc/systemd/system/
sudo systemctl daemon-reload
sudo systemctl enable alb-http
sudo systemctl restart alb-http

Agora, definiremos o autoscaling, especificando os valores mínimos e máximos das instâncias, juntamente com as políticas de escalonamento. Utilizaremos duas métricas principais: requisições por minuto e a média de uso da CPU das instâncias. Além disso, configuraremos a VPC para o autoscaling.

// autoscaling.tf
resource "aws_autoscaling_group" "this" {
  name                      = local.namespaced_service_name
  desired_capacity          = var.autoscaling_group_config.desired_capacity
  min_size                  = var.autoscaling_group_config.min_size
  max_size                  = var.autoscaling_group_config.max_size
  health_check_grace_period = var.autoscaling_group_config.health_check_grace_period
  health_check_type         = var.autoscaling_group_config.health_check_type
  force_delete              = var.autoscaling_group_config.force_delete

  target_group_arns   = [aws_alb_target_group.http.id]
  vpc_zone_identifier = local.public_subnet_ids

  launch_template {
    id      = aws_launch_template.this.id
    version = aws_launch_template.this.latest_version
  }
}

resource "aws_autoscaling_policy" "rpm_policy" {
  name                   = var.autoscaling_policy_alb.name
  enabled                = var.autoscaling_policy_alb.enabled
  autoscaling_group_name = aws_autoscaling_group.this.name
  policy_type            = "TargetTrackingScaling"

  target_tracking_configuration {
    disable_scale_in = var.autoscaling_policy_alb.disable_scale_in
    target_value     = var.autoscaling_policy_alb.target_value
    predefined_metric_specification {
      predefined_metric_type = "ALBRequestCountPerTarget"
      resource_label         = "${aws_alb.this.arn_suffix}/${aws_alb_target_group.http.arn_suffix}"
    }
  }
}

resource "aws_autoscaling_policy" "cpu_policy" {
  name                   = var.autoscaling_policy_cpu.name
  enabled                = var.autoscaling_policy_cpu.enabled
  autoscaling_group_name = aws_autoscaling_group.this.name
  policy_type            = "TargetTrackingScaling"

  target_tracking_configuration {
    disable_scale_in = var.autoscaling_policy_cpu.disable_scale_in
    target_value     = var.autoscaling_policy_cpu.target_value
    predefined_metric_specification {
      predefined_metric_type = "ASGAverageCPUUtilization"
    }
  }
}

No nosso load balancer, definiremos as propriedades de redirecionamento, porta, health check, políticas de erro e outros parâmetros. Neste exemplo, utilizaremos a política de redirecionamento round robin, que distribuirá igualmente as requisições entre as instâncias.

// load_balance.tf
resource "aws_alb" "this" {
  name               = local.namespaced_service_name
  internal           = false
  load_balancer_type = "application"
  security_groups    = [aws_security_group.alb.id]
  subnets            = local.public_subnet_ids

  tags = {
    "Name" = local.namespaced_service_name
  }
}

resource "aws_alb_target_group" "http" {
  name     = local.namespaced_service_name
  port     = 80
  protocol = "HTTP"
  vpc_id   = aws_vpc.this.id

  health_check {
    enabled             = var.alb_health_check_config.enabled
    port                = var.alb_health_check_config.port
    timeout             = var.alb_health_check_config.timeout
    protocol            = var.alb_health_check_config.protocol
    unhealthy_threshold = var.alb_health_check_config.unhealthy_threshold
    interval            = var.alb_health_check_config.interval
    matcher             = var.alb_health_check_config.matcher
    path                = var.alb_health_check_config.path
    healthy_threshold   = var.alb_health_check_config.healthy_threshold
  }
}

resource "aws_alb_listener" "http" {
  load_balancer_arn = aws_alb.this.arn
  port              = 80
  protocol          = "HTTP"

  default_action {
    type             = "forward"
    target_group_arn = aws_alb_target_group.http.arn
  }
}

Por fim, podemos executar o comando plan para verificar quais serão as mudanças e, em seguida, utilizar o comando apply para aplicá-las.

terraform plan
terraform apply

Caso queira desfazer tudo que foi feito, basta executar o comando destroy.

terraform destroy

Considerações finais:

Criar um sistema escalável e resiliente é um trabalho difícil e requer planejamento, uma estratégia sólida e compreensão do que pode acontecer e como lidar com esses cenários.

Distribuir suas instâncias em diferentes regiões garante uma maior resiliência, pois as zonas de disponibilidade possuem requisitos como distância mínima e máxima entre elas e sistemas de rede elétrica e internet diferentes. Isso faz com que uma falha em uma zona não afete outra, tornando muito mais difícil a ocorrência de indisponibilidade em todas as zonas da mesma região.

Outra vantagem desse sistema é que a AWS oferece uma variedade de serviços e maneiras de configuração, como security groups, AZs, regiões, VPCs, EC2, entre outros. Mesmo para uma aplicação pequena, a configuração manual pelo painel pode se tornar repetitiva, um pouco chata e complexa. Ter um sistema de infraestrutura como código (IaC) automatizado facilita o dia a dia e minimiza erros humanos. Além disso, IaC é um passo importante para uma infraestrutura moderna e eficiente.

LinkedIn
Repositório

Como construir um operador HPA baseado na quantidade de mensagens na fila do RabbitMQ no Kubernetes

Ezequiel Lopes — Tue, 11 Jun 2024 00:57:00 +0000

Os operadores Kubernetes são extensões projetadas para gerenciar aplicações complexas e seus componentes no Kubernetes. Utilizando recursos personalizados, eles automatizam tarefas que normalmente seriam realizadas por operadores humanos, como implantação, configuração, escalonamento e recuperação de aplicações. Ao encapsular conhecimento específico de domínio, os operadores permitem que o Kubernetes gerencie aplicações de forma eficaz, algo que seria desafiador usando apenas as capacidades nativas do Kubernetes (Red Hat Developer) (Redhat).

Custom Resources:

Custom Resources (CRs) no Kubernetes são extensões da API do Kubernetes que permitem aos usuários criar e gerenciar seus próprios tipos de recursos personalizados, além dos recursos padrão fornecidos pelo Kubernetes. Eles são utilizados para adicionar funcionalidades específicas que não estão disponíveis na instalação padrão do Kubernetes, permitindo a customização do cluster conforme as necessidades dos aplicativos. (CNCF) (sokube)

Custom Resource Definitions:

Custom Resource Definition (CRD) é um recurso que permite aos usuários definir recursos personalizados além dos tipos de recursos nativos fornecidos pelo Kubernetes, como Pods, Services e Deployments. Com CRDs, os usuários podem estender a funcionalidade do Kubernetes de acordo com suas necessidades específicas.

Role-Based Access Control:

Role-Based Access Control (RBAC) é um sistema de controle de acesso no Kubernetes que gerencia permissões de acesso a recursos do cluster. Ele permite atribuir permissões específicas a usuários e grupos, garantindo um controle granular sobre quem pode executar ações como criar, listar e atualizar Pods, Deployments e Services. (k8s).

Problema:

O Golang consome poucos recursos e quando você tem uma aplicação feita nele que consome uma fila do RabbitMQ, pode ser um pouco difícil escalar as réplicas com base no consumo de recursos como memória e CPU. Uma opção interessante seria manter a quantidade de réplicas pelo número de mensagens na fila.

Neste artigo, vamos construir um operador que monitora uma fila do RabbitMQ e, de acordo com a quantidade de mensagens na fila, altera a quantidade de réplicas no deployment. O objetivo deste artigo é criar um operador experimental apenas para explorar os conceitos desse tipo de recurso no Kubernetes.

Configuração Inicial

Para começar, precisamos definir um recurso personalizado que descreva a configuração desejada para o RabbitMQ e o comportamento de escalonamento. Isso envolve a criação de um CRD e um controlador que monitora o estado atual da fila e ajusta o número de réplicas conforme necessário.

Monitoramento da Fila

O operador deve ser capaz de consultar a fila do RabbitMQ para obter a quantidade de mensagens prontas para processamento. Utilizando uma chamada HTTP, podemos obter essas informações e usá-las para decidir se é necessário escalar o número de réplicas da aplicação.

Auto Scaling

Com base nas informações obtidas da fila, o operador ajustará automaticamente o número de réplicas do deployment do Kubernetes. Isso garante que a aplicação possa lidar com variações na carga de trabalho sem intervenção manual.

Pré-requisitos

Antes de começar, certifique-se que atende os requisitos:

Instalar o Kubebuilder:
- O Kubebuilder é uma ferramenta para construir APIs Kubernetes usando CRDs. Siga as instruções de instalação disponíveis no repositório do GitHub do Kubebuilder.
Instalar o kubectl:
- O kubectl é a ferramenta de linha de comando para interagir com o Kubernetes. Siga as instruções de instalação específicas para seu sistema operacional no site oficial do Kubernetes.
Instalar e Configurar o RabbitMQ:
- Instale o RabbitMQ e configure um usuário. Para este experimento, utilizaremos admin como usuário e admin como senha.
Ter um Cluster Kubernetes:
- Certifique-se de ter um cluster Kubernetes funcionando. Pode ser um cluster local (como Minikube) ou um cluster na nuvem.

Para iniciar o projeto utilizaremos o kubebuilder.

kubebuilder init --domain my.domain --repo github.com/infezek/rabbitmq-operator --owner "Seu nome"
kubebuilder create api --group apps --version v1 --kind RabbitHPA

make manifests
make generate # aceite com y para criar os modulos

make docker-build # para testar o build

Para começar, vamos definir o Spec, que é a estrutura principal do manifesto do nosso operador. Aqui, podemos definir os requisitos de configuração, nome e tipo. Os arquivos de manifesto são tipicamente criados no formato .yaml devido à sua legibilidade facilitada pela indentação, embora sigam o mesmo princípio do json.

No Spec, vamos separar em dois tipos:

Rabbit: Onde ficarão as propriedades da fila que será monitorada.
HPA: Onde ficarão os requisitos de escalabilidade e o deployment.

// api/v1/rabbithpa_types.go
package v1

import (
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
// 
type RabbitHPASpec struct {
    Rabbit RabbitSpec `json:"rabbit"`
    HPA    HPASpec    `json:"hpa"`
}
// Estrutura das Configurações do RabbitMQ
type RabbitSpec struct {
    Namespace string `json:"namespace"`
    Host      string `json:"host"`
    Username  string `json:"username"`
    Password  string `json:"password"`
    Queue     string `json:"queue"`
}
// Estrutura das Configurações do HPA
type HPASpec struct {
    Label                string `json:"label"`
    Namespace            string `json:"namespace"`
    ReplicasetPerMessage int32  `json:"replicasetPerMessage"`
    MinReplicas          int32  `json:"minReplicas"`
    MaxReplicas          int32  `json:"maxReplicas"`
}

type RabbitHPAStatus struct {
    CurrentReplicas int32 `json:"currentReplicas"`
}

type RabbitHPA struct {
    metav1.TypeMeta        `json:",inline"`
    metav1.ObjectMeta      `json:"metadata,omitempty"`
    Spec   RabbitHPASpec   `json:"spec,omitempty"`
    Status RabbitHPAStatus `json:"status,omitempty"`
}

type RabbitHPAList struct {
    metav1.TypeMeta             `json:",inline"`
    metav1.ListMeta             `json:"metadata,omitempty"`
    Items           []RabbitHPA `json:"items"`
}

func init() {
    SchemeBuilder.Register(&RabbitHPA{}, &RabbitHPAList{})
}

O método CalculateReplicas será responsável pela regra de réplicas que deverá ser aplicada no deployment. Sendo um método do RabbitHPA, podemos chamar essa função passando a quantidade de mensagens, e ele devolverá a quantidade de réplicas que deve ter.

// api/v1/rabbit_hpa.go
package v1

func (r *RabbitHPA) CalculateReplicas(currentMessage int32) (target int32) {
    target = currentMessage / r.Spec.HPA.ReplicasetPerMessage
    if target > r.Spec.HPA.MaxReplicas {
        return r.Spec.HPA.MaxReplicas
    }
    if target < r.Spec.HPA.MinReplicas {
        return r.Spec.HPA.MinReplicas
    }
    return target
}

Implementando o Reconcile no Controlador

O método Reconcile é responsável pela lógica do operador, onde podemos verificar a nossa estratégia e atualizar os componentes conforme necessário. Para o nosso projeto, podemos definir a lógica em quatro partes principais:

Verificar os requisitos do Spec: Obter e validar as especificações definidas pelo usuário.
Consultar o RabbitMQ: Obter o número de mensagens na fila do RabbitMQ.
Calcular o número de réplicas que deve ter: Determinar a quantidade ideal de réplicas com base nas mensagens na fila.
Atualizar o Deployment: Se o número atual de réplicas for diferente do esperado, atualizar o deployment para refletir a quantidade correta de réplicas.

// internal/controller/rabbithpa_controller.go
package controller

import (
    "context"
    "time"
    apiv1 "github.com/infezek/rabbitmq-operator/api/v1"
    "github.com/sirupsen/logrus"
    apps "k8s.io/api/apps/v1"
    "k8s.io/apimachinery/pkg/api/errors"
    "k8s.io/apimachinery/pkg/runtime"
    "k8s.io/apimachinery/pkg/types"
    ctrl "sigs.k8s.io/controller-runtime"
    "sigs.k8s.io/controller-runtime/pkg/client"
)

type RabbitHPAReconciler struct {
    client.Client
    Scheme *runtime.Scheme
}

func (r *RabbitHPAReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    log := logrus.WithField("Type", "[RECONCILE]")
    log.Infof("Starting")
    defer log.Infof("Finished")
    // Buscar as especificações do RabbitHPA
    request := &apiv1.RabbitHPA{}
    if err := r.Get(ctx, req.NamespacedName, request); err != nil {
        if errors.IsNotFound(err) {
            log.Info("RabbitHPA resource not found. Ignoring since object must be deleted")
            return ctrl.Result{}, nil
        }
        return ctrl.Result{}, err
    }
    // Realizar a busca do deployment
    deployment := &apps.Deployment{}
    err := r.Get(ctx, types.NamespacedName{
        Namespace: request.Spec.HPA.Namespace,
        Name:      request.Spec.HPA.Label,
    }, deployment)
    if err != nil {
        log.Error(err.Error())
        return ctrl.Result{}, err
    }
    rabbitProps := request.Spec.Rabbit
    // Buscar as propriedades da fila
    gateway := NewRabbitMQGateway(rabbitProps.Host, rabbitProps.Namespace, rabbitProps.Username, rabbitProps.Password)
    queueProps, err := gateway.RequestQuery(rabbitProps.Queue)
    if err != nil {
        log.Errorf(err.Error())
        return ctrl.Result{}, err
    }
    // Calcular a quantidade de réplicas que precisa ter
    target := request.CalculateReplicas(queueProps.MessagesReady)
    log.WithFields(logrus.Fields{
        "current": *deployment.Spec.Replicas,
        "target":  target,
    }).Infof("Current ReplicaSet Deployment")
    if *deployment.Spec.Replicas != target {
        log.Info("Updating deployment replicas")
        // Atualizar o deployment
        deployment.Spec.Replicas = &target
        err = r.Update(ctx, deployment)
        if err != nil {
            log.Errorf(err.Error())
            return ctrl.Result{}, err
        }
    }
    // Tempo para uma nova reconciliação
    return ctrl.Result{RequeueAfter: 5 * time.Second}, nil
}

func (r *RabbitHPAReconciler) SetupWithManager(mgr ctrl.Manager) error {
    return ctrl.NewControllerManagedBy(mgr).
        For(&appsv1.RabbitHPA{}).
        Complete(r)
}

Consultar da fila

É possível consultar detalhes da fila do RabbitMQ por meio de uma chamada HTTP. Basta realizar a chamada conforme o exemplo abaixo. Apenas um detalhe: se estiver dentro do Kubernetes e em outro namespace, é necessário especificar o namespace na URL.

# my é o nome da nossa fila
curl -u admin:admin -s -w "\n%{http_code}\n"  GET http://localhost:15672/api/queues/%2F/my

Esta chamada retornará várias informações relevantes sobre a fila, sendo de especial interesse para o nosso operador o campo messages_ready, que indica a quantidade de mensagens prontas para leitura na fila.

{
...
    "messages": "20"
    "messages_ready": "15"
...
}

A função RequestQuery será responsável por consultar os detalhes da fila atual no RabbitMQ com Golang. Vamos criar a URL para a chamada, passar as credenciais de autenticação no cabeçalho e decodificar a resposta. A função também inclui vários logs para acompanhar e entender o que está acontecendo.

// internal/controller/gateway.go
package controller

import (
    "encoding/base64"
    "encoding/json"
    "fmt"
    "io"
    "net/http"
    "github.com/sirupsen/logrus"
)

type RabbitMQGateway struct {
    host      string
    namespace string
    username  string
    password  string
}

type RabbitOutput struct {
    Messages      int32 `json:"messages"`
    MessagesReady int32 `json:"messages_ready"`
}

func NewRabbitMQGateway(host, namespace, user, pass string) *RabbitMQGateway {
    return &RabbitMQGateway{host, namespace, user, pass}
}

func (r *RabbitMQGateway) RequestQuery(queue string) (*RabbitOutput, error) {
    log := logrus.WithField("Type", "[RABBITMQ_GATEWAY]")
    log.Info("Requesting RabbitMQ")
    defer log.Info("Finished")
    url := fmt.Sprintf("http://%s.%s:15672/api/queues/%s/%s", r.host, r.namespace, "%2f", queue)
    req, err := http.NewRequest(http.MethodGet, url, nil)
    if err != nil {
        log.WithFields(logrus.Fields{"error": err}).Error("Failed to create request")
        return nil, err
    }
    auth := fmt.Sprintf("%s:%s", r.username, r.password)
    authHeader := "Basic " + base64.StdEncoding.EncodeToString([]byte(auth))
    req.Header.Add("Authorization", authHeader)
    client := http.Client{}
    resp, err := client.Do(req)
    if err != nil {
        log.WithFields(logrus.Fields{"error": err}).Error("Failed to request RabbitMQ")
        return nil, err
    }
    defer resp.Body.Close()
    log.Infof("Status code: %d", resp.StatusCode)
    if resp.StatusCode == http.StatusUnauthorized {
        return nil, fmt.Errorf("unauthorized")
    }
    if resp.StatusCode == http.StatusNotFound {
        return nil, fmt.Errorf("not found queue")
    }
    if resp.StatusCode != http.StatusOK {
        return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
    }
    responseBty, err := io.ReadAll(resp.Body)
    if err != nil {
        log.WithFields(logrus.Fields{"error": err}).Error("Failed to read response")
        return nil, err
    }
    output := &RabbitOutput{}
    if err := json.Unmarshal(responseBty, output); err != nil {
        return nil, err
    }
    return output, nil
}

Agora vamos realizar o deploy do operador no Kubernetes. Para isso, o próprio Kubebuilder já fornece uma maneira muito simples de buildar e realizar o deploy com o Makefile. Basta rodar o comando:

# IMG=usuario do docker hub ou do Register Container / nome da imagem
make docker-build docker-push IMG=infectiionz/rabbitmq-operator:beta1
make deploy IMG=infectiionz/rabbitmq-operator:beta1

Aplicando o Manifesto do Operador

Com o manifesto do operador definido, podemos aplicá-lo ao cluster Kubernetes. Em seguida, vamos definir o deployment que será modificado pelo nosso operador.

apiVersion: apps/v1
kind: Deployment
metadata:
  name: ubuntu-deployment
  namespace: ubuntu
spec:
  replicas: 1
  selector:
    matchLabels:
      app: ubuntu
  template:
    metadata:
      labels:
        app: ubuntu
    spec:
      containers:
        - name: ubuntu
          image: ubuntu:latest
          command: ["/bin/bash", "-c", "--"]
          args: ["while true; do sleep 30; done;"]

kubectl apply -f deployment.yaml

Este é o manifesto do nosso operador. Conforme explicado anteriormente, ele contém as especificações sobre nosso RabbitMQ e o HPA.

apiVersion: apps.my.domain/v1
kind: RabbitHPA
metadata:
  name: rabbitmq-hpa
  namespace: default
spec:
  rabbit:
    namespace: "rabbitmq"
    host: "rabbitmq"
    username: "admin"
    password: "admin"
    queue: "my"
  hpa:
    label: "ubuntu-deployment"
    namespace: "ubuntu"
    replicasetPerMessage: 2
    minReplicas: 1
    maxReplicas: 5

kubectl apply -f rabbit-operator.yaml

Podemos verificar os logs do operador.

# Verifique o nome correto do seu operador e o seu namespace e altere no comando abaixo
kubectl logs rabbitmq-operator-controller-manager-767f8947-xptj6 -n rabbitmq-operator-system --tail=30

Teremos um erro parecido com este aqui.

E0609 1 reflector.go:150] pkg/mod/k8s.io/client-go@v0.30.0/tools/cache/reflector.go:232: Failed to watch *v1.Deployment: failed to list *v1.Deployment: deployments.apps is forbidden: User "system:serviceaccount:rabbitmq-operator-system:rabbitmq-operator-controller-manager" cannot list resource "deployments" in API group "apps" at the cluster scope

Esse erro está relacionado à falta de permissão para que o operador execute determinadas tarefas no cluster, especialmente no deployment. Para solucionar este problema, é necessário criar um ClusterRole com as permissões necessárias e um ClusterRoleBinding para atribuir essa role ao nosso operador.

Criando um ClusterRole e ClusterRoleBinding

Um ClusterRole define permissões em nível de cluster. Aqui está um exemplo de um ClusterRole que concede permissões para listar, obter e atualizar deployments:

apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: rabbit-hpa-manager
rules:
  - apiGroups: ["apps"]
    resources: ["deployments"]
    verbs: ["list", "watch", "update"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: rabbit-hpa-manager-binding
subjects:
  - kind: ServiceAccount
    name: rabbitmq-operator-controller-manager
    namespace: rabbitmq-operator-system
roleRef:
  kind: ClusterRole
  name: rabbit-hpa-manager
  apiGroup: rbac.authorization.k8s.io

kubectl apply -f role.yaml

Agora que nosso operador tem as permissões para executar a tarefa, basta apagar o operador para recriá-lo com as permissões corretas.

kubectl delete pod rabbitmq-operator-controller-manager-767f8947-xptj6 -n rabbitmq-operator-system

Agora que o operador está configurado, podemos testar e verificar seu funcionamento, observando a criação e terminação dos pods de acordo com o número de mensagens prontas na fila do RabbitMQ. Lembre-se das condições definidas: mínimo de um pod, máximo de cinco pods, e a cada duas mensagens, um novo pod será criado.

watch -n 2 "kubectl get pods -A | grep ubuntu"

Com 8 mensagens na fila

Na próxima vez que nosso operador rodar e verificar que a quantidade de réplicas desejada não está correta, ele atualizará o deployment para um novo número de réplicas. Dessa forma, o Kubernetes criará ou removerá os pods conforme necessário.

A aplicação possui vários logs para facilitar o entendimento do que está acontecendo. Portanto, basta verificar os logs do operador para acompanhar seu funcionamento e diagnosticar problemas.

Conclusão

Os operadores Kubernetes simplificam a gestão de aplicações complexas, automatizando tarefas críticas e permitindo um gerenciamento eficiente. Ao integrar o RabbitMQ com um operador personalizado, podemos explorar novas maneiras de escalar aplicações de forma inteligente e responsiva, aumentando a eficiência e a confiabilidade do sistema.

Este artigo tem como objetivo demonstrar um pouco do poder dos operadores, abordando conceitos teóricos e criando um projeto prático para um problema específico que podemos ter. Com essa base, será possível criar um operador específico para o seu problema.

Linkedin
GitHub