Google: forward Cloud Run container logs to Airflow log via verbose flag

[aaron-y-chen]

closes: #36963

How

• Use the Cloud Logging library to fetch container logs.
• Support both normal and deferrable mode.

How to reproduce

1. Use gcloud to create a job template.

   export PROJECT_ID="<put-your-project-id>"
   export REGION="us-central1"
   export JOB_NAME="issue-36963-hello-job"
   
   gcloud run jobs create "${JOB_NAME}" \
       --image="us-docker.pkg.dev/cloudrun/container/job:latest" \
       --region="${REGION}" \
       --project="${PROJECT_ID}" \
       --labels="created-by=airflow-36963-repro"

2. Set up the GCP connection.
3. Run the reproduction Dag.

   with DAG(
       dag_id="issue_36963_cloud_run_logs_repro",
       start_date=datetime(2026, 1, 1),
       schedule=None,
       catchup=False,
       tags=["repro", "issue-36963", "cloud-run"],
   ) as dag:
       execute_job = CloudRunExecuteJobOperator(
           task_id="execute_cloud_run_job",
           project_id=PROJECT_ID,
           region=REGION,
           job_name=JOB_NAME,
           deferrable=False,
       )
   
       execute_job_verbose = CloudRunExecuteJobOperator(
           task_id="execute_cloud_run_job_verbose",
           project_id=PROJECT_ID,
           region=REGION,
           job_name=JOB_NAME,
           deferrable=False,
           verbose=True,
       )
   
       execute_job_deferrable = CloudRunExecuteJobOperator(
           task_id="execute_cloud_run_job_deferrable",
           project_id=PROJECT_ID,
           region=REGION,
           job_name=JOB_NAME,
           deferrable=True,
       )
   
       execute_job_deferrable_verbose = CloudRunExecuteJobOperator(
           task_id="execute_cloud_run_job_deferrable_verbose",
           project_id=PROJECT_ID,
           region=REGION,
           job_name=JOB_NAME,
           deferrable=True,
           verbose=True,
       )
   
       execute_job_failing_verbose = CloudRunExecuteJobOperator(
           task_id="execute_cloud_run_job_failing_verbose",
           project_id=PROJECT_ID,
           region=REGION,
           job_name=FAILING_JOB_NAME,
           verbose=True,
       )
   
       (
           execute_job
           >> execute_job_verbose
           >> execute_job_deferrable
           >> execute_job_deferrable_verbose
           >> execute_job_failing_verbose
       )

What

Before the fix

• Cloud Run has container logs such as Starting Task #0, Attempt #0 ....

• However, these logs are not shown in the Airflow UI.

After the fix

• Cloud Run container logs are shown in the Airflow UI as expected.

• Failed jobs also show their container logs before the task failure is raised.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Codex 5.5 following the guidelines

[shahar1]

Looks promising!
Few comments before approval :)

[aaron-y-chen]

Thanks for the review! I believe this version is better than the original one.

Please let me know if there's anything that can be improved :)

[shahar1]

Thanks for the review! I believe this version is better than the original one.

Please let me know if there's anything that can be improved :)

Thanks for fixing!
Hopefully I'll get to it in the next few days :)