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

Gather RabbitMQ connection details

1. Log on to the ApsaraMQ for RabbitMQ console. In the left navigation pane, click Instances.

2. Click the name of your instance. On the Instance Details page, click the Endpoint Information tab and record an endpoint.

   Warning

   A public endpoint exposes your instance to external access. This tutorial uses a public endpoint for demonstration only. If your ACK cluster and RabbitMQ instance are in the same VPC, use the VPC (internal) endpoint instead.

3. In the left navigation pane, click Users and Permissions. Record the Username and Password.

   If no user exists, see Users and Permissions to create one.

4. In the left navigation pane, click Vhosts. Record the Vhost value (for example, amq-test).

Create a Secret and TriggerAuthentication

1. Build a Base64-encoded connection string by running the following command. Replace the placeholders with the values you recorded.

   Placeholder	Description	Example
   <rabbitmq-username>	Username from the RabbitMQ console	Mjpt****
   <rabbitmq-password>	Password from the RabbitMQ console	QT****
   <endpoint>	Public or internal endpoint	amqp-cn-****.amqp-0.net.mq.amqp.aliyuncs.com
   <vhost>	Virtual host name	amq-test

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

2. Save the following YAML as secret.yaml. Replace the host value with the Base64 string from the previous step.

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

3. Save the following YAML as trigger-auth.yaml. This TriggerAuthentication tells KEDA how to authenticate with RabbitMQ by referencing the 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. Apply both files:

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

Create a ScaledObject

A ScaledObject links a Deployment to a KEDA trigger. The following configuration scales sample-app based on the queue-test queue length.

1. Save the following YAML as scaledobject.yaml:

   Parameter	Description
   scaleTargetRef.name	The Deployment to scale. Set this to the application created in Step 1.
   maxReplicaCount	Maximum number of replicas.
   minReplicaCount	Minimum number of replicas.
   protocol	Protocol used to connect to RabbitMQ. Valid values: auto, http, amqp.
   queueName	The RabbitMQ queue to monitor.
   mode	The metric to use for scaling. Set to QueueLength to scale based on the number of messages in the queue.
   value	The threshold that triggers a scale-out.
   metricName	The custom metric name reported to HPA. Typically set to the queue name.

      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. Apply the ScaledObject:

      kubectl apply -f scaledobject.yaml

Verify the configuration

1. Check the ScaledObject status: Expected output:

      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. Check the HPA status: Expected output: TARGETS: 0/20 (avg) shows the current queue length (0) against the threshold (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

Produce messages to trigger scale-out

1. Create a file named producer.go with the following code: Replace <username>, <password>, <endpoint>, and <vhost> with the values you recorded earlier. The URL format is 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. Initialize the Go module and download dependencies:

      go mod init producer
      go mod tidy

3. Run the producer:

      go run producer.go

4. In a separate terminal, watch the HPA scale out: Expected output: The replica count increases to the maximum (10) as messages accumulate in the queue.

      kubectl get hpa

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

Consume messages to observe scale-in

1. Stop the producer program. Create a file named consumer.go with the following code: Replace the url constant with your RabbitMQ connection string.

      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. Run the consumer:

      go run consumer.go

3. Monitor the HPA in real time: Expected output: After the consumer drains the queue, the target metric drops to 0. After a period of no data consumption, KEDA scales the Deployment back to the minimum replica count (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