Container Service for Kubernetes:Horizontal pod autoscaling based on ApsaraMQ for RabbitMQ metrics

Coletar detalhes de conexão do RabbitMQ

1. Acesse o console do ApsaraMQ for RabbitMQ. No painel de navegação à esquerda, clique em Instances.

2. Clique no nome da sua instância. Na página Instance Details, clique em aba Endpoint Information e anote um endpoint.

   Aviso

   Um endpoint público expõe sua instância ao acesso externo. Este tutorial usa um endpoint público apenas para fins de demonstração. Se o seu cluster ACK e a instância do RabbitMQ estiverem na mesma vpc, use o endpoint interno da vpc.

3. Ainda no menu lateral, clique em Users and Permissions. Anote o Username e a Password.

   Se não houver nenhum usuário, consulte Users and Permissions para criar um.

4. No painel lateral, clique em Vhosts. Anote o valor de Vhost (por exemplo, amq-test).

Criar um Secret e um TriggerAuthentication

1. Crie uma string de conexão codificada em Base64 com o seguinte comando. Substitua os placeholders pelos valores que você anotou.

   Placeholder	Descrição	Exemplo
   <rabbitmq-username>	Nome de usuário do console do RabbitMQ	Mjpt****
   <rabbitmq-password>	Senha do console do RabbitMQ	QT****
   <endpoint>	Endpoint público ou interno	amqp-cn-****.amqp-0.net.mq.amqp.aliyuncs.com
   <vhost>	Nome do host virtual	amq-test

      echo -n "amqp://<rabbitmq-username>:<rabbitmq-password>@<endpoint>:5672/<vhost>" | base64

2. Salve o YAML a seguir como secret.yaml. Substitua o valor de host pela string Base64 da etapa anterior.

      apiVersion: v1
      kind: Secret
      metadata:
        name: keda-rabbitmq-secret
      data:
        host: YW1x******  # Your Base64-encoded connection string.

3. Salve o código YAML abaixo no arquivo trigger-auth.yaml. Este TriggerAuthentication informa ao KEDA como se autenticar no RabbitMQ ao referenciar o Secret.

      apiVersion: keda.sh/v1alpha1
      kind: TriggerAuthentication
      metadata:
        name: keda-trigger-auth-rabbitmq-conn
        namespace: default
      spec:
        secretTargetRef:
          - parameter: host
            name: keda-rabbitmq-secret
            key: host

4. Aplique ambos os arquivos:

      kubectl apply -f secret.yaml
      kubectl apply -f trigger-auth.yaml

Criar um ScaledObject

Um ScaledObject vincula um Deployment a um gatilho do KEDA. A configuração a seguir dimensiona o Deployment sample-app com base no comprimento da fila queue-test.

1. Salve o seguinte YAML como scaledobject.yaml:

   Parâmetro	Descrição
   scaleTargetRef.name	O Deployment alvo do dimensionamento. defina como a aplicação criada na Etapa 1.
   maxReplicaCount	Número máximo de réplicas.
   minReplicaCount	Número mínimo de réplicas.
   protocol	Protocolo de conexão com o RabbitMQ. Valores válidos: auto, http, amqp.
   queueName	A fila do RabbitMQ monitorada.
   mode	A métrica usada para o dimensionamento. defina como QueueLength para dimensionar com base no número de mensagens na fila.
   value	O limiar que aciona um scale-out.
   metricName	O nome da métrica personalizada enviada ao HPA. Geralmente defina como o nome da fila.

      apiVersion: keda.sh/v1alpha1
      kind: ScaledObject
      metadata:
        name: rabbitmq-scaledobject
        namespace: default
      spec:
        scaleTargetRef:
          name: sample-app
        maxReplicaCount: 10
        minReplicaCount: 1
        triggers:
        - type: rabbitmq
          metadata:
            protocol: amqp
            queueName: queue-test
            mode: QueueLength
            value: "20"
            metricName: queue-test
          authenticationRef:
            name: keda-trigger-auth-rabbitmq-conn

2. Aplique o ScaledObject:

      kubectl apply -f scaledobject.yaml

verifique a configuração

1. verifique o status do ScaledObject. Saída esperada:

      kubectl get ScaledObject

      NAME                    SCALETARGETKIND      SCALETARGETNAME   MIN   MAX   TRIGGERS   AUTHENTICATION                    READY   ACTIVE   FALLBACK   AGE
      rabbitmq-scaledobject   apps/v1.Deployment   sample-app        1     10   rabbitmq   keda-trigger-auth-rabbitmq-conn    True    False    False      17s

2. verifique o status do HPA. Saída esperada: TARGETS: 0/20 (avg) mostra o comprimento atual da fila (0) comparado ao limiar (20).

      kubectl get hpa

      NAME                             REFERENCE               TARGETS      MINPODS   MAXPODS   REPLICAS   AGE
      keda-hpa-rabbitmq-scaledobject   Deployment/sample-app   0/20 (avg)   1         10        2          2m35s

Produzir mensagens para acionar o scale-out

1. Crie um arquivo chamado producer.go com o código a seguir. Substitua <username>, <password>, <endpoint> e <vhost> pelos valores que você anotou anteriormente. O formato da URL é amqp://username:password@endpoint:5672/vhost.

      package main
   
      import (
      	"fmt"
      	"log"
      	"time"
      	"github.com/streadway/amqp"
      )
   
      const (
      	queueName = "queue-test"    // Must match the queueName in your ScaledObject.
      	numMsgs   = 10000
      	pauseTime = 10 * time.Millisecond
      	url = "amqp://<username>:<password>@<endpoint>:5672/<vhost>"
      )
   
      func main() {
      	conn, err := amqp.Dial(url)
      	failOnError(err, "Failed to connect to RabbitMQ")
      	defer conn.Close()
   
      	ch, err := conn.Channel()
      	failOnError(err, "Failed to open a channel")
      	defer ch.Close()
   
      	q, err := ch.QueueDeclare(
      		queueName,
      		true,
      		false,
      		false,
      		false,
      		nil,
      	)
      	failOnError(err, "Failed to declare a queue")
      	go produce(ch, q)
      	select {}
      }
   
      // produce publishes messages to the queue.
      func produce(ch *amqp.Channel, q amqp.Queue) {
      	for i := 0; i < numMsgs; i++ {
      		msg := fmt.Sprintf("Message %d", i)
      		err := ch.Publish(
      			"",
      			q.Name,
      			false,
      			false,
      			amqp.Publishing{
      				ContentType: "text/plain",
      				Body:        []byte(msg),
      			},
      		)
      		failOnError(err, "Failed to publish a message")
      		log.Printf("Published message: %s", msg)
      		time.Sleep(pauseTime)
      	}
      }
   
      func failOnError(err error, msg string) {
      	if err != nil {
      		log.Fatalf("%s: %s", msg, err)
      	}
      }

2. Inicialize o módulo Go e baixe as dependências:

      go mod init producer
      go mod tidy

3. execute o producer:

      go run producer.go

4. Em um terminal separado, acompanhe o scale-out do HPA. Saída esperada: A contagem de réplicas aumenta até o máximo (10) conforme as mensagens se acumulam na fila.

      kubectl get hpa

      NAME                             REFERENCE               TARGETS           MINPODS   MAXPODS   REPLICAS   AGE
      keda-hpa-rabbitmq-scaledobject   Deployment/sample-app   443000m/20 (avg)   1         10        10         9m15s

Consumir mensagens para observar o scale-in

1. Interrompa o programa producer. Crie o arquivo consumer.go com o código a seguir. Substitua a constante url pela sua string de conexão do RabbitMQ.

      package main
   
      import (
      	"fmt"
      	"log"
      	"github.com/streadway/amqp"
      )
   
      const (
      	queueName = "queue-test"
      	url = "amqp://<username>:<password>@<endpoint>:5672/<vhost>"
      )
   
      func main() {
      	conn, err := amqp.Dial(url)
      	failOnError(err, "Failed to connect to RabbitMQ")
      	defer conn.Close()
   
      	ch, err := conn.Channel()
      	failOnError(err, "Failed to open a channel")
      	defer ch.Close()
   
      	_, err = ch.QueueDeclare(
      		queueName,
      		true,   // durable
      		false,  // exclusive
      		false,  // no-wait
      		false,  // auto-delete
      		nil,
      	)
      	failOnError(err, "Failed to declare a queue")
   
      	msgs, err := ch.Consume(
      		queueName,
      		"",     // consumer tag
      		true,   // auto-ack
      		false,  // exclusive
      		false,  // no-local
      		false,  // no-wait
      		nil,
      	)
      	failOnError(err, "Failed to register a consumer")
   
      	go func() {
      		for msg := range msgs {
      			log.Printf("Received: %s", msg.Body)
      			fmt.Printf("Processed: %s\n", msg.Body)
      		}
      	}()
   
      	select {}
      }
   
      func failOnError(err error, msg string) {
      	if err != nil {
      		log.Fatalf("%s: %s", msg, err)
      	}
      }

2. execute o consumer:

      go run consumer.go

3. Monitore o HPA em tempo real. Saída esperada: Após o consumer esvaziar a fila, a métrica alvo cai para 0. Depois de um período sem consumo de dados, o KEDA reduz a escala do Deployment de volta para a contagem mínima de réplicas (1).

      kubectl get hpa -w

      NAME                             REFERENCE               TARGETS            MINPODS   MAXPODS   REPLICAS   AGE
      keda-hpa-rabbitmq-scaledobject   Deployment/sample-app   443000m/20 (avg)   1         10        10         9m15s
      keda-hpa-rabbitmq-scaledobject   Deployment/sample-app   235000m/20 (avg)   1         10        10         9m51s
      keda-hpa-rabbitmq-scaledobject   Deployment/sample-app   0/20 (avg)         1         10        10         10m
      keda-hpa-rabbitmq-scaledobject   Deployment/sample-app   0/20 (avg)         1         10        1          15m