Container Service for Kubernetes:Mise à l'échelle horizontale des pods basée sur les métriques ApsaraMQ for RabbitMQ

Recueillir les détails de connexion RabbitMQ

1. Connectez-vous à la console ApsaraMQ for RabbitMQ. Dans le volet de navigation de gauche, cliquez sur Instances.

2. Cliquez sur le nom de votre instance. Sur la page Instance Details, cliquez sur l'onglet Endpoint Information et notez un endpoint.

   Avertissement

   Un endpoint public expose votre instance à un accès externe. Ce tutoriel utilise un endpoint public à des fins de démonstration uniquement. Si votre cluster ACK et votre instance RabbitMQ se trouvent dans le même VPC, utilisez plutôt l'endpoint VPC (interne).

3. Dans le volet de navigation de gauche, cliquez sur Users and Permissions. Notez le Username et le Password.

   Si aucun utilisateur n'existe, consultez Users and Permissions pour en créer un.

4. Dans le volet de navigation de gauche, cliquez sur Vhosts. Notez la valeur Vhost (par exemple, amq-test).

Créer un Secret et un TriggerAuthentication

1. Générez une chaîne de connexion encodée en Base64 en exécutant la commande suivante. Remplacez les espaces réservés par les valeurs que vous avez notées.

   Espace réservé	Description	Exemple
   <rabbitmq-username>	Nom d'utilisateur issu de la console RabbitMQ	Mjpt****
   <rabbitmq-password>	Mot de passe issu de la console RabbitMQ	QT****
   <endpoint>	Endpoint public ou interne	amqp-cn-****.amqp-0.net.mq.amqp.aliyuncs.com
   <vhost>	Nom de l'hôte virtuel	amq-test

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

2. Enregistrez le code YAML suivant sous le nom secret.yaml. Remplacez la valeur host par la chaîne Base64 obtenue à l'étape précédente.

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

3. Enregistrez le code YAML suivant sous le nom trigger-auth.yaml. Ce TriggerAuthentication indique à KEDA comment s'authentifier auprès de RabbitMQ en faisant référence au 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. Appliquez les deux fichiers :

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

Créer un ScaledObject

Un ScaledObject lie un Deployment à un déclencheur KEDA. La configuration suivante met à l'échelle sample-app en fonction de la longueur de la file d'attente queue-test.

1. Enregistrez le code YAML suivant sous le nom scaledobject.yaml :

   Paramètre	Description
   scaleTargetRef.name	Le Deployment à mettre à l'échelle. Définissez cette valeur sur l'application créée à l'étape 1.
   maxReplicaCount	Nombre maximal de réplicas.
   minReplicaCount	Nombre minimal de réplicas.
   protocol	Protocole utilisé pour se connecter à RabbitMQ. Valeurs valides : auto, http, amqp.
   queueName	La file d'attente RabbitMQ à surveiller.
   mode	La métrique à utiliser pour la mise à l'échelle. Définissez sur QueueLength pour mettre à l'échelle en fonction du nombre de messages dans la file d'attente.
   value	Le seuil qui déclenche une mise à l'échelle ascendante.
   metricName	Le nom de la métrique personnalisée rapporté au HPA. Généralement défini sur le nom de la file d'attente.

      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. Appliquez le ScaledObject :

      kubectl apply -f scaledobject.yaml

Vérifier la configuration

1. Vérifiez l'état du ScaledObject. Sortie attendue :

      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. Vérifiez l'état du HPA. Sortie attendue : TARGETS: 0/20 (avg) affiche la longueur actuelle de la file d'attente (0) par rapport au seuil (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

Produire des messages pour déclencher la mise à l'échelle ascendante

1. Créez un fichier nommé producer.go avec le code suivant. Remplacez <username>, <password>, <endpoint> et <vhost> par les valeurs que vous avez notées précédemment. Le format de l'URL est 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. Initialisez le module Go et téléchargez les dépendances :

      go mod init producer
      go mod tidy

3. Exécutez le producteur :

      go run producer.go

4. Dans un terminal séparé, observez la mise à l'échelle ascendante du HPA. Sortie attendue : Le nombre de réplicas augmente jusqu'au maximum (10) à mesure que les messages s'accumulent dans la file d'attente.

      kubectl get hpa

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

Consommer des messages pour observer la mise à l'échelle descendante

1. Arrêtez le programme producteur. Créez un fichier nommé consumer.go avec le code suivant. Remplacez la constante url par votre chaîne de connexion 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. Exécutez le consommateur :

      go run consumer.go

3. Surveillez le HPA en temps réel. Sortie attendue : Une fois que le consommateur a vidé la file d'attente, la métrique cible tombe à 0. Après une période sans consommation de données, KEDA ramène le Deployment au nombre minimal 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

kubectl delete -f scaledobject.yaml
kubectl delete -f trigger-auth.yaml
kubectl delete -f secret.yaml
kubectl delete deployment sample-app