Add Cloud Run Operator

[mattsax]

This PR add a CloudRunOperator along with CloudRunHook.
This new Operator allows to make authenticated call against Cloud Run (HTTP request with proper authentication header)

CloudRunOperator extends the SimpleHttpOperator: this add a dependency to HTTP provider.
The authentication token needed in HTTP header (for proper Cloud Run authentication) is an ID Token.
The credential_provider.py had to be modified with a new function get_id_token_credentials to provide "ID Token like" credentials.

---

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst)
Here are some useful points:

• Pay attention to the quality of your code (flake8, mypy and type annotations). Our pre-commits will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[mik-laj]

I wonder how much this code has to do with the this code:
https://github.com/apache/airflow/blob/0fd129f33792a95b06c4f4ad1f7ee83dc2f5cef5/airflow/providers/google/common/utils/id_token_credentials.py

[mattsax]

The two seems indeed related. But id_token_credentials.py provide the credentials from Google application default credentials.
Whereas the credential_provider.py already contains all the code to infer credentials from an Airflow Connection

[potiuk]

Static checks failing.

[mattsax]

I fixed the doc Build.
But for the static checks, the failure comes from airflow/providers/neo4j/hooks/neo4j.py. It doesn't seem related to this PR.

[potiuk]

Conflicts need to be solved.

[potiuk]

And errors to fix.

[eladkal]

I rebased the PR. lets see if test passes

[steren]

Hi, I am the Product Manager for Cloud Run. Thanks for working on a Cloud Run Operator for Airflow!

Please feel free to ask any Cloud Run related question here.

I looked at the PR, it seems that it is using an HTTP call to a Cloud Run service. I would have believed that a Cloud Run job would be more suited for Airflow, as it doesn't require the container to expose any HTTP server, it can be executed via a simple API call. Have you evaluated a job?

[matthieuvilatte]

Hi @steren
Depending on the use case we use both Cloud Run Service and Cloud Run Jobs.
(An other PR has been submited to add a Cloud Run Job operator )
I deployed dbt using Cloud Run. It is pretty convient to be abble to send parameters though a POST HTTP request: start_date and end_date or even --full-refresh parameter when building an incremental model.

[steren]

@matthieuvilatte Ah I might have misunderstood this Operator. It doesn't deploy a Cloud Run service but calls an existing Cloud Run service, is that right?

Maybe the Operator name should make this clearer (i.e. Cloud Run hook)? To differentiate it with the other "Cloud Run Job" Operator

[matthieuvilatte]

@steren
This Operator doesn't deploy a new Cloud Run Service, but can be used to make an authenticated call to an exising Cloud Run Service using a Service Account stored in Airflow Connections.

This Operator name is indeed confusing, but I don't think we should rename it Cloud Run hook. I might be wrong but when describing Airflow DAGs, we always use Operators. An Operator holds the execute method which is called when executing the task and uses a hook to deal with the connection.

May be Cloud Run Service Call Operator?

[eladkal]

@matthieuvilatte operators usually design to do a specific action thus we like/prefer when operator has a verb in it's name indicate the action for example: SqlExecuteOperator, EC2StartInstanceOperator, WasbDeleteBlobOperator

[eladkal]

Tests are failing.
Could you please fix it?

[github-actions]

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.