如何在Cloud Function的Storage触发器中用Python调用GKE外部API？

Hey there! I see you've got a Cloud Function set up to trigger on new files in your recorded-videos-audios GCS bucket, and now you need to call an API hosted on GKE. Let's break this down into actionable steps with code examples that fit right into your existing function.

1. First, confirm your GKE API's accessibility #

Before writing code, you need to know how your GKE service is exposed—this dictates how your Cloud Function can reach it:

• LoadBalancer: If your service uses this type, it’ll have an external IP (grab it with kubectl get services). You can call this IP directly, just make sure your cluster’s firewall allows traffic from Cloud Function’s IP range.
• ClusterIP: If it’s an internal-only service, you’ll need to either:
  • Deploy your Cloud Function in the same VPC as your GKE cluster, or
  • Set up VPC peering between your Cloud Function’s VPC and the GKE cluster’s VPC.
• NodePort: This exposes the service on a static port across cluster nodes. You can use a node’s external IP plus the NodePort to access it, but this is less ideal for production.

2. Use Python's requests library for API calls #

The simplest way to make HTTP requests in Python is with the requests library. Cloud Functions usually have this pre-installed, but if not, add it to your requirements.txt file:

requests>=2.31.0

3. Add API call logic to your existing function #

Here’s how to modify your hello_gcs function to include the API call, with examples for both unauthenticated and secured scenarios:

Example 1: Basic unauthenticated API call #

def hello_gcs(data, context):
    from google.cloud import storage
    import re
    import requests

    # Your existing bucket listing logic
    list_object = []
    list_jpg = []
    list_flac = []
    list_json = []
    storage_client = storage.Client()
    blobs = storage_client.list_blobs('recorded-videos-audios')
    for blob in blobs:
        list_object.append(blob.name)
    print(list_object)

    # Call your GKE-hosted API
    gke_api_url = "http://<YOUR_GKE_SERVICE_IP>:<PORT>/your-api-endpoint"  # Replace with your actual URL
    try:
        # Adjust method (GET/PUT/POST) and payload to match your API's requirements
        response = requests.post(
            gke_api_url,
            json={"new_video_files": list_object}  # Send relevant data to the API
        )
        response.raise_for_status()  # Trigger error for HTTP status codes >=400
        print(f"API call succeeded! Response: {response.json()}")
    except requests.exceptions.RequestException as e:
        print(f"API call failed: {str(e)}")

Example 2: Authenticated API call (using service account) #

If your GKE API requires OAuth2 authentication (common for secure services), you can use the Cloud Function’s default service account to fetch an access token:

def hello_gcs(data, context):
    from google.cloud import storage
    import re
    import requests
    from google.oauth2 import id_token
    from google.auth.transport.requests import Request

    # Your existing bucket logic here...
    list_object = []
    storage_client = storage.Client()
    blobs = storage_client.list_blobs('recorded-videos-audios')
    for blob in blobs:
        list_object.append(blob.name)
    print(list_object)

    # Fetch an access token for your API's audience
    gke_api_audience = "https://<YOUR_GKE_API_AUDIENCE>"  # Replace with your API's audience value
    request_instance = Request()
    token = id_token.fetch_id_token(request_instance, gke_api_audience)

    # Call the secured API
    gke_api_url = "https://<YOUR_GKE_SERVICE_IP>:<PORT>/your-secure-endpoint"
    try:
        response = requests.post(
            gke_api_url,
            json={"new_video_files": list_object},
            headers={"Authorization": f"Bearer {token}"}
        )
        response.raise_for_status()
        print(f"Authenticated API call succeeded! Response: {response.json()}")
    except requests.exceptions.RequestException as e:
        print(f"Authenticated API call failed: {str(e)}")

4. Production-ready considerations #

• VPC Setup: For internal ClusterIP services, ensure your Cloud Function shares the same VPC as your GKE cluster, or set up VPC peering. You can configure this during Cloud Function deployment via the GCP Console or gcloud CLI.
• Firewall Rules: Create a firewall rule on your GKE cluster to allow incoming traffic on your API’s port from your Cloud Function’s subnet or IP range.
• Error Handling: Add retry logic for transient failures using libraries like tenacity if your API is prone to temporary downtime.
• Environment Variables: Store sensitive values (like API URLs or audiences) in Cloud Function environment variables instead of hardcoding them—this keeps your code secure and flexible.

内容的提问来源于stack exchange，提问作者Ankita Shinde