How to autoscale pods based on RabbitMQ metrics - Container Service for Kubernetes

The ack-keda component provides event-driven scaling by periodically polling data from an event source. When messages accumulate, it triggers scaling actions within seconds. You can use KEDA with ApsaraMQ for RabbitMQ to monitor metrics such as queue length and message rate. This topic describes how to use KEDA to autoscale an application based on the queue length of an ApsaraMQ for RabbitMQ instance.

Prerequisites

You have deployed the ack-keda component. For more information, see event-driven scaling.
You have created an ApsaraMQ for RabbitMQ instance and its related resources. For more information, see Create resources.
You have connected to your Kubernetes cluster by using kubectl. For more information, see Connect to a cluster using kubectl.
You have installed a Go runtime environment.

Step 1: Deploy a workload

Log on to the ACK console. In the left navigation pane, click Clusters.
On the Clusters page, click the name of your cluster. In the left navigation pane, click Workloads > Deployments.

On the Deployments page, click Create from YAML. Select Sample Template for Custom, use the following YAML to create an application named sample-app, and then click Create.

apiVersion: apps/v1
kind: Deployment
metadata:
  name: sample-app
  namespace: default
  labels:
    app: sample-deploy
spec:
  replicas: 2
  selector:
    matchLabels:
      app: sample-app
  template:
    metadata:
      labels:
        app: sample-app
    spec:
      containers:
      - name: consumer
        image: consumer  # Replace this with the image of your actual RabbitMQ consumer.
        resources:
          limits:
            cpu: "500m"

Step 2: Deploy a scaling example

In this section, you will get your ApsaraMQ for RabbitMQ instance details and use them to configure KEDA to autoscale pods.

Get your ApsaraMQ for RabbitMQ instance details.
1. Log on to the ApsaraMQ for RabbitMQ console. In the left-side navigation pane, click Instances.
2. Click the name of the target instance to open the Instance Details page. On the Endpoint Information tab, find and record the Endpoint for the public endpoint.
  Using a public endpoint increases security risks, such as external attacks or unauthorized access. This topic uses a public endpoint for demonstration purposes only. If your application runs within an Alibaba Cloud Virtual Private Cloud (VPC) and does not require external access, we recommend that you do not enable public access.
3. In the left-side navigation pane, click Users and Permissions. View and record the Username and Password.
  If you have not created a user, see Users and Permissions to create one.
4. In the left-side navigation pane, click Vhosts. Find and record the Vhost value, such as amq-test.
Run the following command to create and record the connection string.
```
echo -n "amqp://rabbitmq-username:rabbitmq-password@localhost:5672/vhost" | base64
```
Where rabbitmq-username and rabbitmq-password are the previously obtained Username and Password respectively, localhost is the recorded Endpoint, and vhost is the name recorded in the previous step.
Note
If you use the static AccessKey credential of a RAM user to access ApsaraMQ for RabbitMQ and send or receive messages, make sure that the RAM user is granted the required permissions. For more information, see RAM permission policies.

Create a file named secret.yaml with the following content to define a Secret.

apiVersion: v1
kind: Secret
metadata:
  name: keda-rabbitmq-secret
data:
  host: YW1x****** # The connection string that you created.

Use the following YAML to create a file named rabbitmq-trigger-auth.yaml that defines a TriggerAuthentication object.

apiVersion: keda.sh/v1alpha1
kind: TriggerAuthentication
metadata:
  name: keda-trigger-auth-rabbitmq-conn
  namespace: default
spec:
  secretTargetRef:
    - parameter: host
      name: keda-rabbitmq-secret # keda-rabbitmq-secret is the Secret created in the preceding step.
      key: host

Run the following commands to deploy the YAML files to the cluster.

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

After you deploy these resources, KEDA can use the TriggerAuthentication object to connect to RabbitMQ and retrieve metrics.

Create a YAML file named ScaledObject.yaml with the following content.

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

Parameter	Description
`scaleTargetRef`	The application to scale, which in this example is sample-app from Step 1: Deploy a workload to create an application.
`maxReplicaCount`	The maximum replica count.
`minReplicaCount`	The minimum replica count.
`protocol`	The protocol for communication between KEDA and RabbitMQ. Valid values: auto, http, and amqp.
`queueName`	The name of the queue to read messages from.
`value`	The threshold that triggers a scale-out.
`metricName`	The name of the custom metric for the HPA to use when scaling. In this example, queue-test is used to obtain the queue length.

Apply the configuration and check the status of the ScaledObject and HPA resources.

Run the following command to create the resource.

kubectl apply -f ScaledObject.yaml

Run the following command to check the status of the ScaledObject.

kubectl get ScaledObject

Expected output:

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

Run the following command to check the HPA status.

kubectl get hpa

Expected output:

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

Step 3: Produce and consume messages

The following producer and consumer code demonstrates how queue length in ApsaraMQ for RabbitMQ drives pod autoscaling.

Use the following Go code to produce messages to the RabbitMQ queue.

package main

import (
	"fmt"
	"log"
	"time"
	"github.com/streadway/amqp"
)

const (
	queueName = "queue-test"    // Replace this with the name of the queue to read messages from.
	numMsgs   = 10000
	pauseTime = 10 * time.Millisecond
        url = "amqp://Mjpt****:QT****@amqp-cn-zxux009.cn-beijing.amqp-0.net.mq.amqp.aliyuncs.com:5672/amq-test"   // The URL for accessing RabbitMQ. Format: amqp://username:password@your-rabbitmq-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 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("Succeeded to publish a message: %s", msg)
		time.Sleep(pauseTime)
	}
}

func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
	}
}

Initialize the Go environment and download dependencies.
```
go mod init producer
go mod tidy
```

View the HPA details.

Run the producer code to produce messages.

go run producer.go

Run the following command to view HPA details.

kubectl get hpa

Expected output:

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

The output shows that sample-app has scaled out to the maximum replica count defined in the ScaledObject.

Stop the producer program and run the following consumer program to consume messages from the queue.

package main

import (
	"fmt"
	"log"
	"github.com/streadway/amqp"
)

const (
	queueName = "queue-test"
	url = "amqp://Mjpt****:QT****@amqp-cn-zxux009.cn-beijing.amqp-0.net.mq.amqp.aliyuncs.com:5672/amq-test"   // The URL for accessing RabbitMQ. Format: amqp://username:password@your-rabbitmq-endpoint:5672/vhost
)

func main() {
	// Connect to RabbitMQ.
	conn, err := amqp.Dial(url)
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	// Create a channel.
	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

	// Declare a queue to ensure that it exists.
	_, err = ch.QueueDeclare(
		queueName, // Queue name
		true,      // durable
		false,     // exclusive
		false,     // no-wait
		false,     // auto-delete
		nil,       // extra attributes
	)
	failOnError(err, "Failed to declare a queue")

	// Get the message channel.
	msgs, err := ch.Consume(
		queueName, // Queue name
		"",        // consumer tag, can be left empty
		true,      // auto-ack
		false,     // exclusive
		false,     // no-local
		false,     // no-wait
		nil,       // extra arguments
	)
	failOnError(err, "Failed to register a consumer")

	// Create a goroutine to process the received messages.
	go func() {
		for msg := range msgs {
			log.Printf("Received a message: %s", msg.Body)
			// You can process the received message here, for example, by adding your business logic.
			fmt.Printf("Processed message: %s\n", msg.Body)
		}
	}()

	// Block the main program to keep it running.
	select {}
}

// Simple error handler.
func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
	}
}

Run the consumer code to consume messages.

go run consumer.go

Run the following command to monitor the HPA scale-in.

kubectl get hpa -w

Expected output:

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

The output shows that after a cooldown period, sample-app scales in to the minimum replica count defined in the ScaledObject.

Container Service for Kubernetes:Autoscale pods based on RabbitMQ metrics

Prerequisites

Step 1: Deploy a workload

Step 2: Deploy a scaling example

Step 3: Produce and consume messages

Related documents