Add revision tag to newly created cloud run revision

[dahlbaek]

Hello!

• Vote on this issue by adding a 👍 reaction
• If you want to implement this feature, comment to let us know (we'll work with you on design, scheduling, etc.)

Issue details

We currently deploy our GCP Cloud Runs via gcloud run deploy with the --tag flag set to the version we are deploying https://cloud.google.com/sdk/gcloud/reference/run/deploy#--tag. This way, we are able to start using the new release in production, while allowing currently running tasks to use the old release until they finish.

We're currently in the process of moving to pulumi for such deployments, but I'm not finding an equivalent feature 🤔 How would I go about tagging a newly created revision, without having to list all previous revisions?

Affected area/feature

[squaremo]

Thanks for raising this @dahlbaek :star: It sounds like you use GCP; would you expect the desired feature to be general, or specific to GCP?

[squaremo]

Actually, I can kind of answer that -- I'm not sure there's a near analogue to Cloud Run services in many places, so it probably makes most sense that it's particular to GCP.

[squaremo]

Are LatestRevision or Tag useful? https://www.pulumi.com/registry/packages/gcp/api-docs/cloudrun/service/#servicetraffic

[rquitales]

@dahlbaek My understanding is that you would like to create custom tag names for a new deployment/revision for a Cloud Run Service. If that's the case, you should be able to use the above reference provided by @squaremo to tag a new revision.

[dahlbaek]

After a couple of iterations, we ended up on the following pattern

from collections.abc import Sequence

import pulumi
from pulumi_gcp import cloudrun, cloudrunv2

async def _get_service(project: str, location: str, name: str) -> cloudrun.GetServiceResult | None:
    try:
        coroutine = cloudrun.get_service(project=project, location=location, name=name)
        previous_service: cloudrun.GetServiceResult = await coroutine
        return previous_service
    except AttributeError as err:
        if "'NoneType' object has no attribute 'autogenerate_revision_name'" in str(err):
            return None
        else:
            raise RuntimeError from err

def get_previous_service(project: str, location: str, name: str) -> pulumi.Output[cloudrun.GetServiceResult | None]:
    previous_service = _get_service(project=project, location=location, name=name)
    return pulumi.Output.from_input(previous_service)

def deploy_cloudrun(
    *,
    project: str,
    name: str,
    location: str,
    revision: str,
    image: str,
) -> cloudrunv2.Service:
    def traffics_with_tag(
        previous_service: cloudrun.GetServiceResult | None,
    ) -> Sequence["cloudrunv2.outputs.ServiceTraffic"]:
        if previous_service is None:
            return [
                # traffic tags of a new service
            ]

        return [
            # take latest n tags from existing service and add one new tag
            # along with one service traffic that assigns 100% of traffic to latest tag.
        ]

    previous_service = get_previous_service(project=project, location=location, name=name)
    traffics = previous_service.apply(traffics_with_tag)

    return cloudrunv2.Service(
        resource_name=name,
        name=name,
        location=location,
        project=project,
        template=cloudrunv2.ServiceTemplateArgs(
            containers=[cloudrunv2.ServiceTemplateContainerArgs(image=image)],
            revision=revision,
        ),
        traffics=traffics,
    )

It feels a bit clunky that

• we combine pulumi_gcp.cloudrun with pulumi_gcp.cloudrunv2.
• we use an async function in there, but that was the only way I was able to gracefully handle that the service may or may not be deployed already.
• we check for a certain seemingly unrelated substring in the error message to deduce that the service was not deployed already.

Maybe I missed another obvious way to do it... any guidance would be much appreciated.