Long Running Operations on GCP with Cloud Batch and Cloud Workflow

Introduction

Long Running Operations (from now on LRO) are computational tasks or processes that take a significant amount of time to complete, often extending beyond a few seconds or minutes.

They may vary between several use cases, such as data processing, complex calculations, file transfers, machine learning model training and so on.

A good strategy to handle them is using an asynchronous pattern, where their execution is run in the background, in order to avoid blocking main processes.

To do that there is generally an orchestration tool that manages the lifecycle of a LRO, from scheduling to observing its state until its termination.

Of course some good practices can also be considered to better handle LRO, such as fault tolerance, retry logic, split into smaller subtasks and many more.

But we’re not here to deep dive on LRO. This was just a small introduction just to put on the table the key points on what we are going to talk about:

• Long Running Operation task: we will treat a LRO as a simple single task, without any particular retry logic or fault tolerance mechanism to make thing easier for the sake of the article.
• Orchestration: how we can manage our LRO lifecycle without adding too much complexity to the solution.

In this article we will see how we can implement a simple architecture to do all of that using Google Cloud Platform (GCP).

Let’s move on.

Possible solutions on GCP (and drawbacks)

I will just point out some valid solutions, but enforcing the drawbacks we may encounter while choosing them and that could bring us to looking for alternative approaches.

Cloud Run Jobs

Using a serverless option like Cloud Run Jobs might be the best way to go in most situations, especially now that the timeout for the duration of a single task has been extended to 24 hours. I won’t repeat all the benefits of using a serverless approach in general (we know them all too well I suppose), but I just want to list a couple of limitations that persist in the case of LROs:

• Execution limits: if more than 24 hours of execution are necessary, we would be forced to change the logic of our LRO, for example introducing checkpoints or dividing the task into shorter segments. And maybe we could not be able to do that.
• Complexity in debugging: in a serverless environment it becomes more difficult to debug in case of problems. A trivial example could be the inability to SSH into the environment (even when containerized) during execution to intercept the problem.

Google Kubernetes Engine

The “old” good Kubernetes Jobs (or CronJobs if the periodicity of the task is fixed) certainly do not have (or almost) the disadvantages indicated for Cloud Run Jobs, but in this case we must deal with the following considerations:

• Costs: we have lower cost optimization opportunities, due to the fact that we are paying for control planes and a minimum set of resources to run system pods, even when we are not running any jobs.
• Complexity: we must also introduce into our solution the complexity due to knowing (at least a minimum) how to use Kubernetes to manage job deployments, cluster updates… you got the idea.

Cloud Composer

Speaking of the orchestration platforms, Cloud Composer shines for several reasons, but let’s just consider the same drawbacks we listed for GKE to get the idea and proceed.

Proposed Solution

The two main characters of the proposed solution will be:

Cloud Batch

Fully managed service to schedule and run jobs leveraging Google Compute Engine VMs, also with containerized images. The VMs are provisioned when a job is scheduled and started, and are automatically removed when the job ends (the main process generates an exit code).

So in a way we’re getting the same PAYG nature of Cloud Run Jobs (we are only paying for the time our job is processing), but without the limitations indicated above.

I suggest you read an overview on how the service works.

https://cloud.google.com/batch/docs/get-started

Cloud Workflow

Fully serverless managed service, where we can define with a configuration file (called workflow) an execution order of orchestration steps, such as invoking a job, getting its status, terminate it, handle success or failure logic and so on.

Since it is serverless, we don’t need to manage any infrastructure and we don’t pay anything while idle (just for each step execution), making it a very simple and flexible solution.

Again, you should get familiar with the key concepts on how the service works.

https://cloud.google.com/workflows/docs/overview

Solution Architecture

The end-to-end proposed architecture is shown in the figure below:

To better explain how the proposed solution works, we can just jump to the implementation details, to understand and build everything piece by piece.

Implementation Details

We will proceed to the implementation of the architecture proposed step by step to further explain what we are doing and why.

You can follow along the steps provided (code available here) if you have the following in place:

• Google Cloud Project linked to an active Billing Account (expenses will be a few cents if you run the solution for just a few hours)
• Google Account with enough privileges on the GCP Project (let’s assume it is a personal project on which you are Owner)
• gcloud SDK installed on your local computer, or you can just use (for free) the Cloud Shell on the GCP Console

Environment Variables

Just setup a few environment variables to run more smoothly the provisioning steps:

PROJECT_ID=<PROJECT_ID>

VPC=lro-vpc
SUBNET=lro-subnet
REGION=europe-west1
SUBNET_RANGE=10.10.0.0/24
WORKFLOW_SA=lro-workflow
BATCH_SA=lro-vm

AR_REPO=primegen
WORKFLOW_NAME=primegen

PROJECT_NUMBER=$(gcloud projects describe $PROJECT_ID --format='value(projectNumber)')

gcloud config set project $PROJECT_ID

Activate APIs

We will activate the following GCP APIs required for using the resources indicated in the solution:

gcloud services enable artifactregistry.googleapis.com \
  cloudbuild.googleapis.com \
  compute.googleapis.com \
  workflowexecutions.googleapis.com \
  batch.googleapis.com \
  workflows.googleapis.com

Network

Batch requires that you specify a VPC and subnet on which the VM running the LRO will be placed.

If you don’t specify anything, behind the scene the default VPC (and relative subnet in the choosen region) will be used.

But to provide a more flexible solution we will explicitly declare a custom VPC and subnet which we will create with the following commands:

gcloud compute networks create $VPC \
 --project=$PROJECT_ID \
 --subnet-mode=custom

gcloud compute networks subnets create $SUBNET \
 --project=$PROJECT_ID \
 --range=$SUBNET_RANGE \
 --network=$VPC \
 --region=$REGION \
 --enable-private-ip-google-access

Notice that we are enabling Private Google Access, since we will need to pull the container image from Artifact Registry, and we don’t want to use a public IP interface for the Batch VM.

Private Google Access allows us to do exactly this, feel free to deep dive on the argument here.

Prepare Containerized Job

In this scenario I’m using a simple bash script which acts as a prime number generator. It needs a single numeric environment variable (PRIME_NUMBER_LIMIT) and it will print all the prime numbers up to PRIME_NUMBER_LIMIT.

The only thing you should make sure using a custom containerized job is that you must have in place a proper logic to handle exit codes (0 for success, anything else otherwise), or Workflow will provide false positives/negatives regarding job results.

That said, you can test locally the provided code and push it to a dedicated GCP Artifact Registry, ready to be pulled from the Batch VM.

# Local test
docker build -t primegen primegen/
docker run --rm --name primegen -e PRIME_NUMBER_TARGET=4242 primegen

# Build and Push to Artifact Registry
gcloud artifacts repositories create $AR_REPO \
    --repository-format=docker \
    --location=$REGION

gcloud builds submit \
  -t $REGION-docker.pkg.dev/$PROJECT_ID/$AR_REPO/primegen:v1 primegen/

Service Account and Permissions

Now we will create 2 different service accounts with related IAM Policies, but why do we need to do that?

Again, if we don’t specify anything, Batch and Workflow will rely on the default Service Account to call other GCP APIs, which is not a good practice due to the excessive privileges this service account has on the project.

Using separate service accounts (one for each scope) with least privilege access is a best practice which should always be followed from the beginning (otherwise technical debt will haunt us).

So, practically speaking, we will need a service account used by Batch VM to perform the following actions:

• write logs to Cloud Logging
• write status results to Batch
• pull images from Artifact Registry repository

gcloud iam service-accounts create $BATCH_SA

gcloud projects add-iam-policy-binding ${PROJECT_ID} \
  --member=serviceAccount:${BATCH_SA}@${PROJECT_ID}.iam.gserviceaccount.com \
  --role=roles/logging.logWriter

gcloud projects add-iam-policy-binding ${PROJECT_ID} \
  --member=serviceAccount:${BATCH_SA}@${PROJECT_ID}.iam.gserviceaccount.com \
  --role=roles/batch.agentReporter

gcloud artifacts repositories add-iam-policy-binding $AR_REPO \
  --member=serviceAccount:${BATCH_SA}@${PROJECT_ID}.iam.gserviceaccount.com \
  --role=roles/artifactregistry.reader \
  --location $REGION

And a service account used by Workflow to perform the following actions:

• create and delete batch jobs
• write logs to Cloud Logging
• assign the Batch service account to the created VM

gcloud iam service-accounts create $WORKFLOW_SA

gcloud projects add-iam-policy-binding ${PROJECT_ID} \
    --member=serviceAccount:${WORKFLOW_SA}@${PROJECT_ID}.iam.gserviceaccount.com \
    --role=roles/batch.jobsEditor

gcloud projects add-iam-policy-binding ${PROJECT_ID} \
    --member=serviceAccount:${WORKFLOW_SA}@${PROJECT_ID}.iam.gserviceaccount.com \
    --role=roles/logging.logWriter

gcloud iam service-accounts add-iam-policy-binding \
  ${BATCH_SA}@${PROJECT_ID}.iam.gserviceaccount.com \
  --member=serviceAccount:${WORKFLOW_SA}@${PROJECT_ID}.iam.gserviceaccount.com \
  --role=roles/iam.serviceAccountUser

Workflow Deployment

I will breakdown here the Workflow steps to better understand how the job lifecycle is handled.

The first step sets up a set of arguments that include VM configuration (machine size, network, disk, service account used, image to pull..) and variables or arguments to pass to the job execution.

- init:
    assign:
      - projectId: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
      - batchServiceAccount: ${args.batchServiceAccount}
      - region: ${args.region}
      - jobParent: ${"projects/" + projectId + "/locations/" + region}
      - network: ${args.network}
      - subnetwork: ${args.subnetwork}
      - machineType: ${args.machineType}
      - diskSizeGb: ${args.diskSizeGb}
      - diskType: ${args.diskType}
      - imageUri: ${args.imageUri}
      - jobId: ${args.jobName + "-" + uuid.generate()} # This way we avoid duplicates in JobName 
      - getJobResult: null
      - logsUrl: ${"https://console.cloud.google.com/logs/query;query=" + jobId + ";?project=" + projectId}
      - runtimeVariables: ${args} # in this way you can pass any number of environment variables

The second step creates and starts the LRO invoking a Batch Job.

Notice in particular the usage of Workflow built-in connectors to pause Workflow after a LRO starts, and periodically polling the job to wait for it to finish. You can use a timeout up to 1 year.

- create_and_start_lro:
    try:
      call: googleapis.batch.v1.projects.locations.jobs.create
      args:
        parent: ${jobParent}
        jobId: ${jobId}
        body:
          labels:
            job_id: ${jobId} 
          taskGroups:
            taskSpec:
              runnables:
                - container:
                    imageUri: ${imageUri}
                  environment:
                    # Pass any number of runtime variables as environment variables
                    variables: ${runtimeVariables}
            taskCount: 1
            parallelism: 1
            permissiveSsh: false
          allocationPolicy:
            instances:
            - policy:
                provisioningModel: STANDARD
                machineType: ${machineType}
                bootDisk:
                  image: batch-cos
                  sizeGb: ${diskSizeGb}
                  type: ${diskType}
            serviceAccount:
              email: ${batchServiceAccount}
              scopes:
                - https://www.googleapis.com/auth/cloud-platform
            network:
              networkInterfaces:
              # Full link is used for providing easy configuration if Shared VPC architectures are used
              - network: ${"projects/" + projectId + "/global/networks/" + network}
                subnetwork: ${"projects/" + projectId + "/regions/" + region + "/subnetworks/" + subnetwork}
                noExternalIpAddress: true # We usually don't want a public IP address on VM
          logsPolicy:
            destination: CLOUD_LOGGING
        # Used to configure the polling mechanism on which workflow periodically
        # checks for job completion status.
        # In this way no url callback mechanism is needed and logic is simpler
        connector_params:
          timeout: 25920000 #300 days. Max 365
          polling_policy:
            initial_delay: 60.0
            multiplier: 1.1
            max_delay: 300
          skip_polling: False
      # Save job result status in the variable Job
      result: job
    # Catch error an if any, save status in the variable job
    except:
      as: e
      steps:
        - log_error:
            call: sys.log
            args:
              data: ${"Error " + json.encode_to_string(e.operation)}
        - get_job_id:
            call: googleapis.batch.v1.projects.locations.jobs.get
            args:
              name: ${e.operation.name}
            result: job

The other steps are really easy to understand, in brief:

• Job state is logged to Cloud Logging
• Job is deleted from Batch History (just to keep it clean)
• Job result (success/failed) is saved into a variable to determine Workflow final outcome
• Workflow returns an ok or error status based on Batch Job result

# Just print job state
- log_job_state:
    call: sys.log
    args:
      data: ${"Current job state " + job.status.state}
# Batch job is delete to keep history of Batch clean.
# You could also comment this step, but you could incur in limits
# regarding the number of Jobs per project
- delete_batch_job:
    call: googleapis.batch.v1.projects.locations.jobs.delete
    args:
      name: ${job.name}
# We check the job state to determine if print a success info message
# (and closing the worklow execution with success) or raising a workflow
# error and providing the Cloud Logging URL for further inspection
- check_job_result:
    switch:
      - condition: ${job.status.state == "SUCCEEDED"}
        next: return_result
      - condition: ${job.status.state == "FAILED"}
        next: fail_execution
- return_result:
    return: ${"The batch job " + job.name + " completed successfully"}
- fail_execution:
    raise:
      message: ${"The batch job " + job.name + " failed. See GCP logs for further details"}

The Workflow can be deployed in this way:

gcloud workflows deploy $WORKFLOW_NAME \
  --source workflow-lro.yaml \
  --service-account=${WORKFLOW_SA}@${PROJECT_ID}.iam.gserviceaccount.com \
  --location=$REGION

Workflow Execution

Now let’s test the workflow execution by calling it with the required arguments and variables.

gcloud workflows execute $WORKFLOW_NAME --location $REGION --data \
"{
  \"batchServiceAccount\": \"${BATCH_SA}@${PROJECT_ID}.iam.gserviceaccount.com\",
  \"region\": \"$REGION\",
  \"network\" : \"$VPC\",
  \"subnetwork\": \"$SUBNET\",
  \"machineType\": \"e2-medium\",
  \"diskType\": \"pd-balanced\",
  \"diskSizeGb\": \"30\",
  \"imageUri\" : \"$REGION-docker.pkg.dev/$PROJECT_ID/$AR_REPO/primegen:v1\",
  \"jobName\" : \"test-lro\",
  \"PRIME_NUMBER_LIMIT\": \"100000\"
}"  

We can verify during the execution the workflow status:

as well as the Batch Job being provisioned.

When the job is running we can check its logs written to Cloud Logging.

For debug purposes we have the possibility to SSH into the VM (you may need to add a firewall rule to allow ingress traffic using IAP).

Once the job finishes (hopefully correctly) it will be deleted by Workflow and finally Workflow will end with a success state and according logs.

In case of a job failure (in this case I’m passing a non-numeric argument) we can see the job failed status (due to an exit code != 0):

and Workflow negative feedback.

Bonus Point: notifications using Webhook (Google Chat)

Well, this is straightforward: who doesn’t want a notification system that sends us a message when the job is done, without having us to manually polling check on the status?

We can simply add two steps in our workflow, the first one notifying us when a job is started, and a second one to notify us when the job is finished with an overview of the status details and a quick link to access logs on GCP.

# Before job start
- send_start_notification:
    call: http.post
    args:
      url: ${webhookUrl}
      headers:
        Content-Type: "application/json; charset=UTF-8"
      body:
        text: ${"Job " + jobId + "is starting.. See logs at " + logsUrl}
# After job finishes
- send_finish_notification:
    call: http.post      
    args:
      url: ${webhookUrl}
      headers:
        Content-Type: "application/json; charset=UTF-8"
      body:
        text: ${"Job " + jobId + " finished with state " + job.status.state + ". See logs at " + logsUrl}

In the example provided the notification is sent via a Webhook URL (in my case using Google Chat) with a raw POST request, but you can easily adapt it based on your needs.

This is what I get for a job lifecycle:

Conclusion

Ok, maybe that was too much talking for a solution that is pretty simple and straightforward, but I hope that by reading all the considerations I made (and please correct me if I am getting anything wrong) throughout writing this post you can find some ideas for your own use cases or find new approaches starting from this one.