Google Cloud Run Operators¶
Cloud Run is used to build and deploy scalable containerized apps written in any language (including Go, Python, Java, Node.js, .NET, and Ruby) on a fully managed platform.
For more information about the service visit Google Cloud Run documentation.
Prerequisite Tasks¶
To use these operators, you must do a few things:
Select or create a Cloud Platform project using the Cloud Console.
Enable billing for your project, as described in the Google Cloud documentation.
Enable the API, as described in the Cloud Console documentation.
Install API libraries via pip.
pip install 'apache-airflow[google]'Detailed information is available for Installation.
Create a job¶
Before you create a job in Cloud Run, you need to define it. For more information about the Job object fields, visit Google Cloud Run Job description
A simple job configuration can be created with a Job object:
def _create_job_instance() -> Job:
"""
Create a Cloud Run job configuration with google.cloud.run_v2.Job object.
As a minimum the configuration must contain a container image name in its template.
The rest of the configuration parameters are optional and will be populated with default values if not set.
"""
job = Job()
container = k8s_min.Container()
container.image = "us-docker.pkg.dev/cloudrun/container/job:latest"
container.resources.limits = {"cpu": "2", "memory": "1Gi"}
job.template.template.containers.append(container)
return job
or with a Python dictionary:
def _create_job_dict() -> dict:
"""
Create a Cloud Run job configuration with a Python dict.
As a minimum the configuration must contain a container image name in its template.
"""
return {
"template": {
"template": {
"containers": [
{
"image": "us-docker.pkg.dev/cloudrun/container/job:latest",
"resources": {
"limits": {"cpu": "1", "memory": "512Mi"},
"cpu_idle": False,
"startup_cpu_boost": False,
},
"name": "",
"command": [],
"args": [],
"env": [],
"ports": [],
"volume_mounts": [],
"working_dir": "",
"depends_on": [],
}
],
"volumes": [],
"execution_environment": 0,
"encryption_key": "",
},
"labels": {},
"annotations": {},
"parallelism": 0,
"task_count": 0,
},
"name": "",
"uid": "",
"generation": "0",
"labels": {},
"annotations": {},
"creator": "",
"last_modifier": "",
"client": "",
"client_version": "",
"launch_stage": 0,
"observed_generation": "0",
"conditions": [],
"execution_count": 0,
"reconciling": False,
"satisfies_pzs": False,
"etag": "",
}
You can create a Cloud Run Job with any of these configurations :
CloudRunCreateJobOperator
create1 = CloudRunCreateJobOperator(
task_id=create1_task_name,
project_id=PROJECT_ID,
region=region,
job_name=job1_name,
job=_create_job_instance(),
dag=dag,
)
Note that this operator only creates the job without executing it. The Job’s dictionary representation is pushed to XCom.
Create a service¶
Before you create a service in Cloud Run, you need to define it. For more information about the Service object fields, visit Google Cloud Run Service description
A simple service configuration can look as follows:
def _create_service():
service = Service()
container = k8s_min.Container()
container.image = "us-docker.pkg.dev/cloudrun/container/placeholder:latest"
service.template.containers.append(container)
return service
With this configuration we can create the service:
CloudRunCreateServiceOperator
create_cloud_run_service = CloudRunCreateServiceOperator(
task_id="create-cloud-run-service",
project_id=PROJECT_ID,
region="us-central1",
service=_create_service(),
service_name="cloudrun-system-test-service",
)
Note that this operator only creates the service without executing it. The Service’s dictionary representation is pushed to XCom.
Delete a service¶
With this configuration we can delete the service:
CloudRunDeleteServiceOperator
delete_cloud_run_service = CloudRunDeleteServiceOperator(
task_id="delete-cloud-run-service",
project_id=PROJECT_ID,
region="us-central1",
service_name="cloudrun-system-test-service",
dag=dag,
)
Note this operator waits for the service to be deleted, and the deleted Service’s dictionary representation is pushed to XCom.
Execute a job¶
To execute a job, you can use:
execute1 = CloudRunExecuteJobOperator(
task_id=execute1_task_name,
project_id=PROJECT_ID,
region=region,
job_name=job1_name,
dag=dag,
deferrable=False,
)
or you can define the same operator in the deferrable mode:
execute2 = CloudRunExecuteJobOperator(
task_id=execute2_task_name,
project_id=PROJECT_ID,
region=region,
job_name=job2_name,
dag=dag,
deferrable=True,
)
Transport¶
The CloudRunExecuteJobOperator accepts an optional transport
parameter to choose the underlying API transport.
transport="grpc"(default): use gRPC transport. Iftransportis not set, gRPC is used.transport="rest": use REST/HTTP transport.
In deferrable mode, when using gRPC (explicitly or by default), the trigger uses an async gRPC client internally; for non-deferrable execution, the operator uses the regular (synchronous) gRPC client.
In general, it is better to use gRPC (or leave transport unset) unless there is a specific reason you must use REST (for example,
if gRPC is not available or fails in your environment).
Deferrable mode considerations
When using deferrable mode, the operator defers to an async trigger that polls the long-running operation status.
- With gRPC (explicitly or by default), the trigger uses the native async gRPC client internally. The
grpc_asynciotransport is an implementation detail of the Google client library and is not a user-facing
transportvalue.
- With gRPC (explicitly or by default), the trigger uses the native async gRPC client internally. The
- With REST, the REST transport is synchronous-only in the Google Cloud library. To remain compatible with deferrable mode, the
trigger performs REST calls using the synchronous client wrapped in a background thread.
REST can be used with deferrable mode, but it may be less efficient than gRPC and is generally best reserved for cases where gRPC cannot be used.
You can also specify overrides that allow you to give a new entrypoint command to the job and more:
overrides = {
"container_overrides": [
{
"name": "job",
"args": ["python", "main.py"],
"env": [{"name": "ENV_VAR", "value": "value"}],
"clear_args": False,
}
],
"task_count": 1,
"timeout": "60s",
}
execute3 = CloudRunExecuteJobOperator(
task_id=execute3_task_name,
project_id=PROJECT_ID,
region=region,
overrides=overrides,
job_name=job3_name,
dag=dag,
deferrable=False,
)
Update a job¶
To update a job, you can use:
update_job1 = CloudRunUpdateJobOperator(
task_id=update_job1_task_name,
project_id=PROJECT_ID,
region=region,
job_name=job1_name,
job=_create_job_instance_with_label(),
dag=dag,
)
The Job’s dictionary representation is pushed to XCom.
List jobs¶
To list the jobs, you can use:
list_jobs = CloudRunListJobsOperator(
task_id=list_jobs_task_name, project_id=PROJECT_ID, region=region, dag=dag
)
The operator takes two optional parameters: “limit” to limit the number of tasks returned, and “show_deleted” to include deleted jobs in the result.
Delete a job¶
To delete a job you can use:
delete_job1 = CloudRunDeleteJobOperator(
task_id="delete-job1",
project_id=PROJECT_ID,
region=region,
job_name=job1_name,
dag=dag,
trigger_rule=TriggerRule.ALL_DONE,
)
Note this operator waits for the job to be deleted, and the deleted Job’s dictionary representation is pushed to XCom.