Please update google cloud provider cloud functions api to support v2

[cbio-erik]

Description

We should parameterize this api_version argument to allow users to trigger v1 or v2 functions. Right now v2 functions just not reachable because this hard coded value. (https://github.com/apache/airflow/blob/main/airflow/providers/google/cloud/operators/functions.py#L453)

Use case/motivation

Able to trigger cloud functions v2 with larger machine types

Related issues

No response

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[nathadfield]

@cbio-erik I don't believe this is a hardcoded value but the default value. Your should be able to override this when using the operator.

test = CloudFunctionInvokeFunctionOperator(
        task_id='my-task',
        project_id='my-project',
        function_id='my-function',
        input_data={'key': 'value'},
        api_version='v2',
    )

[cbio-erik]

@nathadfield Seems like the google cloud functions v2 has to different base uri. I will close this ticket since CloudRunExecuteJobOperator is an alternative operator to run cloud function v2 jobs.

[houtanb]

@nathadfield NB that just providing api_version='v2', does not solve the issue.

@cbio-erik it's not clear to me how CloudRunExecuteJobOperator works for both v1 and v2 functions. And do we need to create jobs for each individual v2 function we want to call if we want to use CloudRunExecuteJobOperator? I have a lot of v1 and v2 functions that I would like to call, so I didn't see CloudRunExecuteJobOperator as an option, but perhaps I'm missing something in my setup...

It's unfortunate that CloudFunctionInvokeFunctionOperator hasn't been updated for v2 functions since they've been generally available since August 10, 2022.

But we don't have to rely on the Google library! We can just authenticate and call the v1 or v2 function directly...

@task
def cloud_function_task(endpoint):
    """Call cloud function represented by endpoint."""
    auth_req = google.auth.transport.requests.Request()
    id_token = google.oauth2.id_token.fetch_id_token(auth_req, endpoint)
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {id_token}",
    }

    response = requests.post(
        url=endpoint,
        headers=headers,
        data=None,
    )

    if response.status_code != 200:
        raise Exception(f"\nError invoking {endpoint}: {response.text}\n")

    return response.text

Your service account will need permissions to invoke the given cloud functions.

If running locally, you'll need to set the GOOGLE_APPLICATION_CREDENTIALS environment variable to the credentials file.