CloudRun Function Code Reference

"""
Cloud Function to track Cloud Run deployments.
Triggered by Eventarc on Cloud Run service updates.
Extracts image SHA256 digests and posts to configured endpoint.
"""
 
import functions_framework
import json
import os
from google.cloud import run_v2
from urllib import request, error
 
revisions_client = run_v2.RevisionsClient()
 
def get_image_with_digest(revision_name: str) -> tuple:
    """
    Get the full image URI with SHA256 digest from a Cloud Run revision.
    """
    print(f"Fetching revision: {revision_name}")
 
    try:
        revision = revisions_client.get_revision(name=revision_name)
 
        if hasattr(revision, 'containers') and revision.containers:
            full_image = revision.containers[0].image
            print(f"Full image URI from revision: {full_image}")
 
            if '@sha256:' in full_image:
                digest = full_image.split('@')[1]
                print(f"Extracted digest: {digest}")
                return full_image, digest
            else:
                print(f"WARNING: No digest in image URI: {full_image}")
                return full_image, None
 
        if hasattr(revision, 'template') and revision.template and revision.template.containers:
            full_image = revision.template.containers[0].image
            print(f"Full image URI from revision (via template): {full_image}")
 
            if '@sha256:' in full_image:
                digest = full_image.split('@')[1]
                print(f"Extracted digest: {digest}")
                return full_image, digest
            else:
                print(f"WARNING: No digest in image URI: {full_image}")
                return full_image, None
 
        print("ERROR: No containers in revision")
        return None, None
 
    except Exception as e:
        print(f"Error fetching revision: {e}")
        return None, None
 
def post_deployment(payload: dict) -> bool:
    """Post deployment info to configured endpoint."""
    endpoint = os.environ.get('DEPLOYMENT_ENDPOINT_URL')
    if not endpoint:
        print("WARNING: DEPLOYMENT_ENDPOINT_URL not set, skipping POST")
        return False
 
    try:
        headers = {
            'Content-Type': 'application/json',
            'User-Agent': 'CloudRun-Deployment-Tracker/1.0'
        }
 
        api_key = os.environ.get('API_KEY')
        if api_key:
            headers['Authorization'] = f'Bearer {api_key}'
 
        data = json.dumps(payload).encode('utf-8')
        req = request.Request(endpoint, data=data, headers=headers, method='POST')
 
        print(f"Posting to endpoint: {endpoint}")
 
        with request.urlopen(req, timeout=10) as response:
            status = response.getcode()
            body = response.read().decode('utf-8')
            print(f"POST response: status={status}, body={body}")
            return 200 <= status < 300
 
    except error.HTTPError as e:
        error_body = e.read().decode('utf-8')
        print(f"HTTP error posting deployment: status={e.code}, error={error_body}")
        return False
    except Exception as e:
        print(f"Error posting deployment: {e}")
        return False
 
@functions_framework.cloud_event
def cloudrun_deployment_tracker(cloud_event):
    """
    Cloud Function triggered by Eventarc on Cloud Run deployment events.
    """
    print(f"Received event: {cloud_event['type']}")
 
    event_data = cloud_event.data
    proto_payload = event_data.get('protoPayload', {})
    resource = event_data.get('resource', {})
 
    if not proto_payload:
        print("ERROR: No protoPayload in event")
        return
 
    method_name = proto_payload.get('methodName', '')
    print(f"Method: {method_name}")
 
    if 'ReplaceService' not in method_name and 'CreateService' not in method_name:
        print(f"Ignoring method: {method_name}")
        return
 
    request_data = proto_payload.get('request', {})
    service_spec = request_data.get('service', {})
 
    if not service_spec:
        print("ERROR: No service spec in request")
        return
 
    metadata = service_spec.get('metadata', {})
    status = service_spec.get('status', {})
 
    service_name = metadata.get('name')
    namespace = metadata.get('namespace')
 
    labels = resource.get('labels', {})
    project_id = labels.get('project_id', namespace)
    location = labels.get('location', 'us-central1')
 
    print(f"Service: {service_name}, Project: {project_id}, Location: {location}")
 
    revision_name = status.get('latestCreatedRevisionName', '')
    service_url = status.get('url', '')
 
    if not revision_name:
        print("ERROR: No revision name in status")
        return
 
    revision_resource_name = f"projects/{project_id}/locations/{location}/services/{service_name}/revisions/{revision_name}"
 
    full_image, digest = get_image_with_digest(revision_resource_name)
 
    if not full_image:
        print("ERROR: Could not fetch image from revision")
        return
 
    print(f"Image: {full_image}")
    print(f"Digest: {digest}")
 
    scope = f"{location}/{project_id}"
 
    deployment_payload = {
        "runtime": "cloudrun",
        "scope": scope,
        "workloads": [
            {
                "name": service_name,
                "properties": {
                    "revisionName": revision_name,
                    "image": full_image,
                    "project": project_id,
                    "location": location,
                    "serviceUrl": service_url,
                    "eventType": method_name,
                    "timestamp": event_data.get('timestamp')
                },
                "digests": [digest] if digest else []
            }
        ]
    }
 
    print(f"Deployment payload: {json.dumps(deployment_payload, indent=2)}")
 
    success = post_deployment(deployment_payload)
 
    if success:
        print("Successfully posted deployment data")
    else:
        print("Failed to post deployment data")
 
    return {"status": "ok" if success else "failed"}

functions-framework==3.*
google-cloud-run==0.10.*