Container Service for Kubernetes：Penyesuaian Otomatis Pod Horizontal berdasarkan metrik ApsaraMQ for RabbitMQ

Komponen ack-keda menyediakan penskalaan berbasis peristiwa dengan mengonsumsi data secara berkala dari sumber peristiwa. Saat terjadi akumulasi pesan, ack-keda memicu penskalaan untuk sejumlah tugas offline dalam hitungan detik. Anda dapat menggunakan ApsaraMQ for RabbitMQ bersama KEDA untuk memantau metrik seperti panjang antrian dan laju pesan. Topik ini menjelaskan cara menggunakan metrik panjang antrian ApsaraMQ for RabbitMQ dan penyekala otomatis berbasis peristiwa KEDA untuk melakukan penskalaan otomatis pada aplikasi.

Prasyarat

• Komponen ack-keda telah diterapkan. Untuk informasi selengkapnya, lihat Penskalaan berbasis peristiwa.

• Sebuah instans ApsaraMQ for RabbitMQ dan sumber daya terkait telah dibuat. Untuk informasi selengkapnya, lihat Buat sumber daya.

• Anda telah terhubung ke kluster Kubernetes menggunakan kubectl. Untuk informasi selengkapnya, lihat Terhubung ke kluster menggunakan kubectl.

• Lingkungan runtime Go telah diinstal.

Langkah 1: Terapkan beban kerja untuk membuat aplikasi

1. Masuk ke Konsol ACK. Di panel navigasi kiri, klik Clusters.

2. Di halaman Clusters, temukan kluster yang diinginkan lalu klik namanya. Di panel kiri, pilih Workloads > Deployments.

3. Di halaman Deployments, klik Create from YAML. Atur Template menjadi Custom. Gunakan YAML berikut untuk membuat aplikasi bernama sample-app, lalu klik Create.

   apiVersion: apps/v1
   kind: Deployment
   metadata:
     name: sample-app
     namespace: default
     labels:
       app: sample-app
   spec:
     replicas: 2
     selector:
       matchLabels:
         app: sample-app
     template:
       metadata:
         labels:
           app: sample-app
       spec:
         containers:
         - name: consumer
           image: consumer  # Ganti ini dengan citra aktual konsumen RabbitMQ Anda.
           resources:
             limits:
               cpu: "500m"

Langkah 2: Terapkan contoh penskalaan berdasarkan metrik RabbitMQ

Langkah-langkah berikut menunjukkan cara mendapatkan informasi instans ApsaraMQ for RabbitMQ Anda dan mengonfigurasi Horizontal Pod Autoscaler (HPA) agar secara otomatis menskalakan aplikasi Anda berdasarkan metrik RabbitMQ.

1. Dapatkan informasi instans ApsaraMQ for RabbitMQ.

   1. Masuk ke Konsol ApsaraMQ for RabbitMQ. Di panel navigasi kiri, pilih Instances.

   2. Klik nama instans target untuk membuka halaman Instance Details. Di tab Endpoints, lihat dan catat Endpoint untuk Public Endpoint.

      Menggunakan titik akhir publik menimbulkan risiko keamanan, seperti serangan eksternal atau akses tidak sah. Contoh ini menggunakan titik akhir publik hanya untuk demonstrasi. Jika aplikasi Anda berjalan di dalam VPC Alibaba Cloud dan tidak memerlukan akses eksternal, jangan aktifkan akses publik.

   3. Di panel navigasi kiri, klik Users and Permissions. Lihat dan catat Username dan Password.

      Jika Anda belum membuat pengguna, lihat Users and Permissions untuk membuatnya.

   4. Di panel navigasi kiri, klik Vhosts. Lihat dan catat nilai Vhost, misalnya, amq-test.

2. Jalankan perintah berikut untuk membuat dan mencatat string autentikasi koneksi.

   echo -n "amqp://rabbitmq-username:rabbitmq-password@localhost:5672/vhost" | base64

   Ganti rabbitmq-username dan rabbitmq-password dengan Username dan Password yang telah Anda peroleh. Ganti localhost dengan titik akhir yang telah dicatat. Ganti vhost dengan nama yang dicatat pada langkah sebelumnya.

3. Gunakan YAML berikut untuk membuat Secret.

   apiVersion: v1
   kind: Secret
   metadata:
     name: keda-rabbitmq-secret
   data:
     host: YW1x****** # String autentikasi koneksi yang telah Anda buat.

4. Gunakan YAML berikut untuk membuat objek TriggerAuthentication dan terapkan file YAML ke kluster.

   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 adalah Secret yang dibuat pada langkah sebelumnya.
         key: host

   Jalankan perintah berikut untuk menerapkan file YAML ke kluster.

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

   Setelah penerapan selesai, Anda dapat menggunakan pemicu RabbitMQ di KEDA. Anda juga dapat mereferensikan objek TriggerAuthentication untuk terhubung ke RabbitMQ dan mengambil data metrik.

5. Buat file YAML bernama ScaledObject.yaml dengan konten berikut.

   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	Deskripsi
   scaleTargetRef	Objek yang akan diskalakan. Dalam contoh ini, diatur ke sample-app, yaitu aplikasi yang dibuat di Langkah 1: Terapkan beban kerja untuk membuat aplikasi.
   maxReplicaCount	Jumlah maksimum replika.
   minReplicaCount	Jumlah minimum replika.
   protocol	Protokol komunikasi antara komponen Keda dan RabbitMQ. Nilai yang valid: auto, http, dan amqp.
   queueName	Nama antrian tempat membaca pesan.
   value	Ambang batas yang memicu skala keluar.
   metricName	Nama metrik kustom. Ini memberi tahu HPA metrik mana yang digunakan untuk pengambilan keputusan penskalaan. Di sini, ini adalah nama antrian RabbitMQ, queue-test, yang digunakan untuk mendapatkan panjang antrian.

6. Terapkan konfigurasi dan periksa status sumber daya ScaledObject dan HPA.

   Jalankan perintah berikut untuk membuat sumber daya.

   kubectl apply -f ScaledObject.yaml  

   Jalankan perintah berikut untuk memeriksa status konfigurasi penskalaan.

   kubectl get ScaledObject

   Keluaran yang diharapkan:

   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

   Jalankan perintah berikut untuk memeriksa status HPA.

   kubectl get hpa

   Keluaran yang diharapkan:

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

Langkah 3: Hasilkan dan konsumsi data untuk memicu penskalaan

Kode produsen dan konsumen berikut untuk antrian ApsaraMQ for RabbitMQ menunjukkan cara menggunakan panjang antrian untuk menskalakan kontainer.

1. Gunakan kode produsen berikut untuk menghasilkan pesan RabbitMQ.

   package main
   
   import (
   	"fmt"
   	"log"
   	"time"
   	"github.com/streadway/amqp"
   )
   
   const (
   	queueName = "queue-test"    // Ganti ini dengan nama antrian tempat membaca pesan.
   	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"   // URL untuk mengakses RabbitMQ. Format: amqp://username:password@localhost: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.
   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("Successed to publish a message: %s", msg)
   		time.Sleep(pauseTime)
   	}
   }
   
   func failOnError(err error, msg string) {
   	if err != nil {
   		log.Fatalf("%s: %s", msg, err)
   	}
   }

2. Inisialisasi lingkungan Go dan unduh dependensi.

   go mod init producer
   go mod tidy

3. Periksa detail HPA.

   Jalankan kode produsen untuk menghasilkan pesan ke antrian.

   go run producer.go

   Jalankan perintah berikut untuk memeriksa detail HPA.

   kubectl get hpa

   Keluaran yang diharapkan:

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

   Keluaran yang diharapkan menunjukkan bahwa sample-app telah diskalakan keluar hingga jumlah maksimum replika yang ditetapkan dalam komponen KEDA.

4. Hentikan program produsen dan jalankan program konsumen berikut untuk mengonsumsi pesan dari antrian.

   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"   // URL untuk mengakses RabbitMQ. Format: amqp://username:password@localhost: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 the queue to make sure it exists.
   	_, err = ch.QueueDeclare(
   		queueName, // Queue name
   		true,      // durable
   		false,     // exclusive
   		false,     // no-wait
   		false,     // auto-delete
   		nil,       // arguments
   	)
   	failOnError(err, "Failed to declare a queue")
   
   	// Get the message channel.
   	msgs, err := ch.Consume(
   		queueName, // Queue name
   		"",        // Consumer name. Leave it empty if not needed.
   		true,      // auto-ack
   		false,     // exclusive
   		false,     // no-local
   		false,     // no-wait
   		nil,       // arguments
   	)
   	failOnError(err, "Failed to register a consumer")
   
   	// Create a goroutine to process received messages.
   	go func() {
   		for msg := range msgs {
   			log.Printf("Received a message: %s", msg.Body)
   			// You can process the received messages here, for example, by implementing business logic.
   			fmt.Printf("Processed message: %s\n", msg.Body)
   		}
   	}()
   
   	// Block the main program to keep it running.
   	select {}
   }
   
   // Simple error handling function.
   func failOnError(err error, msg string) {
   	if err != nil {
   		log.Fatalf("%s: %s", msg, err)
   	}
   }

   Jalankan kode konsumen untuk mengonsumsi pesan.

   go run consumer.go

   Untuk memantau skala-masuk HPA, jalankan perintah berikut.

   kubectl get hpa -w

   Keluaran yang diharapkan:

   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

   Keluaran yang diharapkan menunjukkan bahwa setelah periode tanpa konsumsi data, sample-app diskalakan masuk ke jumlah minimum replika yang ditetapkan dalam komponen KEDA.

Referensi

Untuk informasi tentang cara mengonfigurasi HPA berdasarkan metrik akumulasi pesan RocketMQ kustom guna penskalaan yang lebih fleksibel, lihat KEDA berdasarkan metrik RocketMQ.