Autoscale worker pools based on the Pub/Sub queue volume

Create a Pub/Sub topic and subscription

To autoscale your worker, create a pull subscription for your consumer application using the following steps:

1. Create a Pub/Sub topic that represents a feed of messages:

    gcloud pubsub topics create $TOPIC_ID 
   

2. Create a pull subscription to consume messages from your Pub/Sub topic:

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

Create custom service accounts

This tutorial requires the following two service accounts with minimum permissions required to use the provisioned resources:

• Consumer service account: identity for the consumer worker pool that processes messages. Run the following command to create the consumer service account:

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

• CREMA service account: identity for the autoscaler. Run the following command to create the CREMA service account:

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

Grant additional permissions to your custom service accounts

To scale the worker pool, grant the following permissions on the custom service accounts:

1. Grant your CREMA service account permission to read from the 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. Grant your CREMA service account the permission to scale the worker pool:

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

3. Grant your CREMA service account the service account user role:

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

4. Grant your CREMA service account permission to view metrics:

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

5. Grant your CREMA service account permission to write metrics:

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

6. Grant your CREMA service account permission to view Pub/Sub messages:

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

7. Grant your consumer service account permission to pull messages from the subscription:

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

Deploy a Cloud Run worker pool

To deploy a worker pool that consumes messages from Pub/Sub subscriptions, follow these steps:

1. Create a folder named consumer and change the directory into it:

    mkdir  
   consumer cd 
     
   consumer 
   

2. Create a file named worker.py and add the following code:

     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 
    = 
     subscribe 
    
   r . 
     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. Create a Dockerfile and add the following code:

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

4. Deploy the consumer worker pool with 0 instances for CREMA to scale up:

    gcloud  
   beta  
   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 
    
   

Deploy the autoscaler CREMA service

Once you deploy the worker pool to consume messages from Pub/Sub, configure the CREMA autoscaler to provision worker instances based on the volume of messages.

Configure the autoscaler

This tutorial uses the Parameter Manager to store the YAML configuration file for CREMA.

1. Create a parameter in the Parameter Manager to store parameter versions for CREMA:

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

2. Navigate to the root directory of your project by running the following command:

     cd 
    
   

3. In your root directory, create a YAML file, my-crema-config.yaml to define the autoscaler configuration:

     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 
    
   

   Replace PROJECT_ID with the Google Cloud project ID.

4. Upload your local YAML file as a new parameter version:

     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. Run the following command to verify your parameter addition is successful:

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

   You should see the parameter path, such as projects/ PROJECT_ID /locations/global/parameters/crema-config/versions/1 .

Deploy the service to scale your workloads

To deploy the service to scale your worker pool, run the following command with a prebuilt container image:

  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  
beta  
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" 
 

Test your autoscaling service

Test your CREMA service by creating a script that generates 100 messages and pushes them to the Pub/Sub queue:

1. In your root directory, create a file named load-pubsub.sh , and add the following code:

     #!/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. Run the load test:

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

This command generates and pushes 100 messages to the Pub/Sub subscription.

Monitor Scaling

After the load-pubsub.sh script completes, wait three to four minutesbefore checking logs for the service, my-crema-service . The CREMA autoscaler service scales up the consumer worker instances from 0.

You should see the following logs:

Each log message is labeled with the component that emitted it.

 [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 ... 

Alternatively, run the following command to verify that the CREMA service recommends instances based on the queue depth:

 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 

To view the consumer logs consuming messages, run the following command:

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

You should see logs that follow the format, Done job-100 .