Ajusta automáticamente la escala de los grupos de trabajadores según el volumen de la cola de Pub/Sub

Crear un tema de Pub/Sub y una suscripción

Para ajustar automáticamente la escala de tu trabajador, crea una suscripción de extracción para tu aplicación de consumidor con los siguientes pasos:

1. Crea un tema de Pub/Sub que represente un feed de mensajes:

   gcloud pubsub topics create $TOPIC_ID
   

2. Crea una suscripción de extracción para consumir mensajes de tu tema de Pub/Sub:

   gcloud pubsub subscriptions create $SUBSCRIPTION_ID --topic=$TOPIC_ID
   

Crea cuentas de servicio personalizadas

En este instructivo, se requieren las siguientes dos cuentas de servicio con los permisos mínimos necesarios para usar los recursos aprovisionados:

• Cuenta de servicio del consumidor: Es la identidad del grupo de trabajadores del consumidor que procesa los mensajes. Ejecuta el siguiente comando para crear la cuenta de servicio del consumidor:

  gcloud iam service-accounts create $CONSUMER_SA_NAME \
    --display-name="Pub/Sub consumer service account"
  

• Cuenta de servicio de CREMA: Es la identidad del escalador automático. Ejecuta el siguiente comando para crear la cuenta de servicio de CREMA:

  gcloud iam service-accounts create $CREMA_SA_NAME \
    --display-name="CREMA service account"
  

Otorga permisos adicionales a tus cuentas de servicio personalizadas

Para escalar el grupo de trabajadores, otorga los siguientes permisos en las cuentas de servicio personalizadas:

1. Otorga permiso a tu cuenta de servicio de CREMA para leer desde el Parameter Manager:

   gcloud projects add-iam-policy-binding $PROJECT_ID \
     --member="serviceAccount:$CREMA_SA_NAME@$PROJECT_ID.iam.gserviceaccount.com" \
     --role="roles/parametermanager.parameterViewer"
   

2. Otorga a tu cuenta de servicio de CREMA el permiso para escalar el grupo de trabajadores:

   gcloud projects add-iam-policy-binding $PROJECT_ID \
     --member="serviceAccount:$CREMA_SA_NAME@$PROJECT_ID.iam.gserviceaccount.com" \
     --role="roles/run.developer"
   

3. Otorga a tu cuenta de servicio de CREMA el rol de usuario de la cuenta de servicio:

   gcloud projects add-iam-policy-binding $PROJECT_ID \
     --member="serviceAccount:$CREMA_SA_NAME@$PROJECT_ID.iam.gserviceaccount.com" \
     --role="roles/iam.serviceAccountUser"
   

4. Otorga permiso a tu cuenta de servicio de CREMA para ver las métricas:

    gcloud projects add-iam-policy-binding $PROJECT_ID \
      --member="serviceAccount:$CREMA_SA_NAME@$PROJECT_ID.iam.gserviceaccount.com" \
      --role="roles/monitoring.viewer"
   

5. Otorga permiso a tu cuenta de servicio de CREMA para escribir métricas:

    gcloud projects add-iam-policy-binding $PROJECT_ID \
      --member="serviceAccount:$CREMA_SA_NAME@$PROJECT_ID.iam.gserviceaccount.com" \
      --role="roles/monitoring.metricWriter"
   

6. Otorga permiso a tu cuenta de servicio de CREMA para ver mensajes de Pub/Sub:

   gcloud pubsub subscriptions add-iam-policy-binding $SUBSCRIPTION_ID \
     --member="serviceAccount:$CREMA_SA_NAME@$PROJECT_ID.iam.gserviceaccount.com" \
     --role="roles/pubsub.viewer"
   

7. Otorga permiso a tu cuenta de servicio de consumidor para extraer mensajes de la suscripción:

   gcloud pubsub subscriptions add-iam-policy-binding $SUBSCRIPTION_ID \
     --member="serviceAccount:$CONSUMER_SA_NAME@$PROJECT_ID.iam.gserviceaccount.com" \
     --role="roles/pubsub.subscriber"
   

Implementa un grupo de trabajadores de Cloud Run

Para implementar un grupo de trabajadores que consuma mensajes de suscripciones a Pub/Sub, sigue estos pasos:

1. Crea una carpeta llamada consumer y cambia el directorio a ella:

   mkdir consumer
   cd consumer
   

2. Crea un archivo llamado worker.py y agrega el siguiente código:

   import os
   import time
   from google.cloud import pubsub_v1
   from concurrent.futures import TimeoutError
   
   # Configuration
   PROJECT_ID = os.environ.get('PROJECT_ID')
   SUBSCRIPTION_ID = os.environ.get('SUBSCRIPTION_ID')
   
   subscription_path = f"projects/{PROJECT_ID}/subscriptions/{SUBSCRIPTION_ID}"
   
   print(f"Worker Pool instance starting. Watching {subscription_path}...")
   
   subscriber = pubsub_v1.SubscriberClient()
   
   def callback(message):
       try:
           data = message.data.decode("utf-8")
           print(f"Processing job: {data}")
           time.sleep(5)  # Simulate work
           print(f"Done {data}")
           message.ack()
       except Exception as e:
           print(f"Error processing message: {e}")
           message.nack()
   
   streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback)
   print(f"Listening for messages on {subscription_path}...")
   
   # Wrap subscriber in a 'with' block to automatically call close() when done.
   with subscriber:
       try:
           # When `timeout` is not set, result() will block indefinitely,
           # unless an exception is encountered first.
           streaming_pull_future.result()
       except TimeoutError:
           streaming_pull_future.cancel()  # Trigger the shutdown.
           streaming_pull_future.result()  # Block until the shutdown is complete.
       except Exception as e:
           print(f"Streaming pull failed: {e}")
   

3. Crea un Dockerfile y agrega el siguiente código:

   FROM python:3.12-slim
   RUN pip install google-cloud-pubsub
   COPY worker.py .
   CMD ["python", "-u", "worker.py"]
   

4. Implementa el grupo de trabajadores del consumidor con 0 instancias para que CREMA escale verticalmente:

   gcloud run worker-pools deploy $CONSUMER_WORKER_POOL_NAME \
     --source . \
     --region $REGION \
     --service-account="$CONSUMER_SA_NAME@$PROJECT_ID.iam.gserviceaccount.com" \
     --instances=0 \
     --set-env-vars PROJECT_ID=$PROJECT_ID,SUBSCRIPTION_ID=$SUBSCRIPTION_ID
   

Implementa el servicio CREMA del escalador automático

Una vez que implementes el grupo de trabajadores para consumir mensajes de Pub/Sub, configura el escalador automático de CREMA para aprovisionar instancias de trabajadores según el volumen de mensajes.

Configurar el escalador automático

En este instructivo, se usa Parameter Manager para almacenar el archivo de configuración YAML de CREMA.

1. Crea un parámetro en el Administrador de parámetros para almacenar versiones de parámetros de CREMA:

   PARAMETER_ID=crema-config
   PARAMETER_REGION=global
   gcloud parametermanager parameters create $PARAMETER_ID --location=$PARAMETER_REGION --parameter-format=YAML
   

2. Para navegar al directorio raíz de tu proyecto, ejecuta el siguiente comando:

   cd
   

3. En tu directorio raíz, crea un archivo YAML, my-crema-config.yaml, para definir la configuración del ajuste de escala automático:

   apiVersion: crema/v1
   kind: CremaConfig
   spec:
     pollingInterval: 30
     triggerAuthentications:
       - metadata:
           name: adc-trigger-auth
         spec:
           podIdentity:
             provider: gcp
     scaledObjects:
       - spec:
           scaleTargetRef:
             name: projects/PROJECT_ID/locations/us-central1/workerpools/worker-pool-consumer
           triggers:
             - type: gcp-pubsub
               metadata:
                 subscriptionName: "crema-subscription"
                 # Target number of undelivered messages per worker instance
                 value: "10"
                 mode: "SubscriptionSize"
               authenticationRef:
                 name: adc-trigger-auth
   

   Reemplaza PROJECT_ID por el ID del proyecto de Google Cloud .

4. Sube tu archivo YAML local como una nueva versión del parámetro:

   LOCAL_YAML_CONFIG_FILE=my-crema-config.yaml
   PARAMETER_VERSION=1
   
   gcloud parametermanager parameters versions create $PARAMETER_VERSION \
     --location=$PARAMETER_REGION \
     --parameter=$PARAMETER_ID \
     --payload-data-from-file=$LOCAL_YAML_CONFIG_FILE
   

5. Ejecuta el siguiente comando para verificar que se haya agregado el parámetro correctamente:

   gcloud parametermanager parameters versions list \
   --parameter=$PARAMETER_ID \
   --location=$PARAMETER_REGION
   

   Deberías ver la ruta del parámetro, como projects/PROJECT_ID/locations/global/parameters/crema-config/versions/1.

Implementa el servicio para escalar tus cargas de trabajo

Para implementar el servicio y escalar tu grupo de trabajadores, ejecuta el siguiente comando con una imagen de contenedor compilada previamente:

CREMA_CONFIG_PARAM_VERSION=projects/$PROJECT_ID/locations/$PARAMETER_REGION/parameters/$PARAMETER_ID/versions/$PARAMETER_VERSION
IMAGE=us-central1-docker.pkg.dev/cloud-run-oss-images/crema-v1/autoscaler:1.0

gcloud run deploy $CREMA_SERVICE_NAME \
  --image=${IMAGE} \
  --region=${REGION} \
  --service-account="${CREMA_SA_NAME}" \
  --no-allow-unauthenticated \
  --no-cpu-throttling \
  --base-image=us-central1-docker.pkg.dev/serverless-runtimes/google-24/runtimes/java25 \
  --labels=created-by=crema \
  --set-env-vars="CREMA_CONFIG=${CREMA_CONFIG_PARAM_VERSION},OUTPUT_SCALER_METRICS=True"

Prueba tu servicio de ajuste de escala automático

Para probar tu servicio de CREMA, crea una secuencia de comandos que genere 100 mensajes y los envíe a la cola de Pub/Sub:

1. En el directorio raíz, crea un archivo llamado load-pubsub.sh y agrega el siguiente código:

   #!/bin/bash
   
   TOPIC_ID=${TOPIC_ID}
   PROJECT_ID=${PROJECT_ID}
   NUM_MESSAGES=100
   
   echo "Publishing $NUM_MESSAGES messages to topic $TOPIC_ID..."
   
   for i in $(seq 1 $NUM_MESSAGES); do
     gcloud pubsub topics publish $TOPIC_ID --message="job-$i" --project=$PROJECT_ID &
     if (( $i % 10 == 0 )); then
       wait
       echo "Published $i messages..."
     fi
   done
   wait
   echo "Done. All messages published."
   

2. Ejecuta la prueba de carga:

   chmod +x load-pubsub.sh
   ./load-pubsub.sh
   

Este comando genera y envía 100 mensajes a la suscripción de Pub/Sub.

Supervisión del ajuste

Una vez que se complete la secuencia de comandos load-pubsub.sh, espera entre tres y cuatro minutos antes de revisar los registros del servicio, my-crema-service. El servicio de ajuste automático de CREMA aumenta la cantidad de instancias de trabajadores del consumidor a partir de 0.

Deberías ver los siguientes registros:

Cada mensaje de registro está etiquetado con el componente que lo emitió.

[INFO] [METRIC-PROVIDER] Starting metric collection cycle
[INFO] [METRIC-PROVIDER] Successfully fetched scaled object metrics ...
[INFO] [METRIC-PROVIDER] Sending scale request ...
[INFO] [SCALER] Received ScaleRequest ...
[INFO] [SCALER] Current instances ...
[INFO] [SCALER] Recommended instances ...

Como alternativa, ejecuta el siguiente comando para verificar que el servicio de CREMA recomiende instancias según la profundidad de la cola:

gcloud logging read "resource.type=cloud_run_revision AND resource.labels.service_name=$CREMA_SERVICE_NAME AND textPayload:SCALER" \
  --limit=20 \
  --format="value(textPayload)" \
  --freshness=5m

Para ver los registros de los consumidores que consumen mensajes, ejecuta el siguiente comando:

gcloud run worker-pools logs tail $CONSUMER_WORKER_POOL_NAME --region=$REGION

Deberías ver registros con el formato Done job-100.