Dynamic Pod Scaling with KEDA for Event-Driven Architectures

# thumbnail-worker-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: thumbnail-generator-worker
  namespace: media-processing
  labels:
    app: thumbnail-generator-worker
spec:
  replicas: 0 # Start with zero, KEDA will scale it up
  selector:
    matchLabels:
      app: thumbnail-generator-worker
  template:
    metadata:
      labels:
        app: thumbnail-generator-worker
    spec:
      containers:
      - name: worker
        image: your-repo/thumbnail-generator:1.2.0
        env:
        - name: RABBITMQ_URI
          valueFrom:
            secretKeyRef:
              name: rabbitmq-conn-secret
              key: connectionString
---
# rabbitmq-secret.yaml
apiVersion: v1
kind: Secret
metadata:
  name: rabbitmq-conn-secret
  namespace: media-processing
type: Opaque
data:
  # amqp://user:[email protected]:5672
  connectionString: YW1xcDovL3VzZXI6cGFzc3dvcmRAcmFiYml0bXEtc2VydmljZS5kZWZhdWx0LnN2Yy5jbHVzdGVyLmxvY2FsOjU2NzI=

# keda-rabbitmq-auth.yaml
apiVersion: keda.sh/v1alpha1
kind: TriggerAuthentication
metadata:
  name: rabbitmq-trigger-auth
  namespace: media-processing
spec:
  secretTargetRef:
    - parameter: host # The parameter name KEDA's RabbitMQ scaler expects
      name: rabbitmq-conn-secret # The secret we created
      key: connectionString # The key within the secret

# keda-scaledobject-rabbitmq.yaml
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: rabbitmq-thumbnail-scaler
  namespace: media-processing
spec:
  scaleTargetRef:
    name: thumbnail-generator-worker # Name of the Deployment to scale
  pollingInterval: 15  # How often to check the queue (in seconds). Default is 30.
  cooldownPeriod:  180 # Period to wait before scaling down to 0. Default is 300.
  minReplicaCount: 0   # Allow scaling down to zero
  maxReplicaCount: 50  # Safety cap to avoid overwhelming downstream systems
  triggers:
  - type: rabbitmq
    metadata:
      # Required
      queueName: image-jobs
      # Optional - defaults to 1 if not set. This is the target value.
      queueLength: "10"
      # Optional - vhost, defaults to "/"
      vhost: "/media"
    authenticationRef:
      name: rabbitmq-trigger-auth # Reference our TriggerAuthentication object

// main.go
package main

import (
	"log"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/streadway/amqp"
)

func main() {
	uri := os.Getenv("RABBITMQ_URI")
	if uri == "" {
		log.Fatal("RABBITMQ_URI environment variable not set")
	}

	conn, err := connectToRabbitMQ(uri)
	if err != nil {
		log.Fatalf("Failed to connect to RabbitMQ: %s", err)
	}
	defer conn.Close()

	ch, err := conn.Channel()
	if err != nil {
		log.Fatalf("Failed to open a channel: %s", err)
	}
	defer ch.Close()

	q, err := ch.QueueDeclare(
		"image-jobs", // name
		true,         // durable
		false,        // delete when unused
		false,        // exclusive
		false,        // no-wait
		nil,          // arguments
	)
	if err != nil {
		log.Fatalf("Failed to declare a queue: %s", err)
	}

	// Set Quality of Service: only fetch one message at a time.
	// This ensures messages are distributed evenly among workers.
	err = ch.Qos(1, 0, false)
	if err != nil {
		log.Fatalf("Failed to set QoS: %s", err)
	}

	msgs, err := ch.Consume(
		q.Name,
		"thumbnail-worker", // consumer tag
		false,            // auto-ack is false. We will manually ack.
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		log.Fatalf("Failed to register a consumer: %s", err)
	}

	// Graceful shutdown channel
	done := make(chan bool)

	go func() {
		for d := range msgs {
			log.Printf("Received a message: %s", d.Body)
			// Simulate work (e.g., image processing)
			time.Sleep(2 * time.Second)
			log.Printf("Done processing message")
			d.Ack(false) // Acknowledge the message
		}
	}()

	log.Printf("[*] Waiting for messages. To exit press CTRL+C")

	// Handle SIGTERM for graceful shutdown
	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
	<-sigChan

	log.Println("Shutdown signal received, exiting gracefully.")
}

func connectToRabbitMQ(uri string) (*amqp.Connection, error) {
	var conn *amqp.Connection
	var err error
	for i := 0; i < 5; i++ {
		conn, err = amqp.Dial(uri)
		if err == nil {
			return conn, nil
		}
		log.Printf("Failed to connect, retrying in 5 seconds... (%v)", err)
		time.Sleep(5 * time.Second)
	}
	return nil, err
}

# audit-processor-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: audit-log-processor
  namespace: security
  labels:
    app: audit-log-processor
spec:
  replicas: 0
  selector:
    matchLabels:
      app: audit-log-processor
  template:
    metadata:
      labels:
        app: audit-log-processor
    spec:
      containers:
      - name: processor
        image: your-repo/audit-log-processor:2.5.1
        env:
        - name: KAFKA_BROKERS
          value: "kafka-service.kafka.svc.cluster.local:9092"
        - name: KAFKA_TOPIC
          value: "platform-events"
        - name: KAFKA_CONSUMER_GROUP
          value: "audit-log-processors"
        # ... other env vars for SASL username/password if needed

# keda-kafka-auth.yaml
apiVersion: keda.sh/v1alpha1
kind: TriggerAuthentication
metadata:
  name: kafka-trigger-auth
  namespace: security
spec:
  secretTargetRef:
    - parameter: sasl # KEDA's Kafka scaler expects this parameter for SASL type
      name: kafka-sasl-secret
      key: sasl_type
    - parameter: username
      name: kafka-sasl-secret
      key: username
    - parameter: password
      name: kafka-sasl-secret
      key: password
---
# kafka-sasl-secret.yaml
apiVersion: v1
kind: Secret
metadata:
  name: kafka-sasl-secret
  namespace: security
type: Opaque
data:
  sasl_type: "SCRAM-SHA-512" # Base64 encoded
  username: "..."
  password: "..."

# keda-scaledobject-kafka.yaml
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: kafka-audit-log-scaler
  namespace: security
spec:
  scaleTargetRef:
    name: audit-log-processor
  pollingInterval: 20
  cooldownPeriod:  300
  minReplicaCount: 1   # Keep at least one replica for faster response
  maxReplicaCount: 20
  triggers:
  - type: kafka
    metadata:
      # Required
      bootstrapServers: kafka-service.kafka.svc.cluster.local:9092
      topic: platform-events
      consumerGroup: audit-log-processors

      # The target value for scaling
      lagThreshold: "500"

      # Optional: For topics with few partitions, allow multiple pods per partition
      # allowIdleConsumers: "true"

      # Optional: Which offset to start with if no offset is committed
      offsetResetPolicy: "latest"

      # Optional: Version of Kafka broker
      version: "2.8.1"
    authenticationRef:
      name: kafka-trigger-auth

# keda-scaledobject-multi-trigger.yaml
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: video-transcoder-scaler
  namespace: media-processing
spec:
  scaleTargetRef:
    name: video-transcoder-worker
  minReplicaCount: 1
  maxReplicaCount: 100
  triggers:
  # Trigger 1: Scale based on RabbitMQ queue length
  - type: rabbitmq
    metadata:
      queueName: video-transcode-jobs
      queueLength: "5"
    authenticationRef:
      name: rabbitmq-trigger-auth

  # Trigger 2: Scale based on CPU utilization
  - type: cpu
    metadata:
      type: Utilization
      value: "75"