Guide to integrate Google Cloud Workflows with other Google services.

4 min readJan 31, 2021

Why Workflows?

Server-less and hence easy to scale on demand.
Combine custom services, API’s, google cloud products etc into an automated workflow.
Inbuilt error handling mechanism making it easy to debug.
Easy to develop: accepts JSON/YAML formats.

Server less Orchestration through Google Cloud Workflows

In this article we will be covering some key services provided by the Google Could Platform, including the Secrets Manager , Cloud Functions, Big Query Data insertions, Pub/Sub , and all these services will be orchestrated and automated through the Google Cloud Workflows.

Google Cloud Workflows- Secrets Manager

Secret Manager is a secure and convenient storage system for API keys, passwords, certificates, and other sensitive data. Secret Manager provides a central place and single source of truth to manage, access, and audit secrets across Google Cloud, thus enabling us to build more secure applications. Secret keys and corresponding values are created manually from the Secrets Manager Cloud Console. The secret is encrypted by Google-managed key by default.

Problem Statement -1 : Accessing (any) API keys from Secrets manager through Workflows.

Assuming that we have a set of keys created up front in secrets manager, we access them from workflows as follows:

get_secret_cred :
    steps:
        - initVariables:
            assign:
            - project: ${sys.get_env("GOOGLE_CLOUD_PROJECT_NUMBER")}
            - secret_username: "api_username"
            - version_username: 1
            - secret_password: "api_password"
            - version_password: 1        - get_api_username:
            call: googleapis.secretmanager.v1.projects.secrets.versions.access
            args:
                name: ${"projects/" + project + "/secrets/" + secret_username + "/versions/" + string(version_username)}
            result: secretResult_username        - get_api_password:
            call: googleapis.secretmanager.v1.projects.secrets.versions.access
            args:
                name: ${"projects/" + project + "/secrets/" + secret_password + "/versions/" + string(version_password)}
            result: secretResult_password
        
        - storeSecret:
            assign:
            - api_username: ${text.decode(base64.decode(secretResult_username.payload.data))}
            - api_password: ${text.decode(base64.decode(secretResult_password.payload.data))}
            - credentials:
                username : ${api_username}
                password : ${api_password}       - returnResult:
            return: ${credentials}

Step initVariables: This is an assign step to initiate variable like the project number (Using environment variable), the API username and API password keys as set on secrets manager API and their respective versions.
Step get_api_username: This is a call step to invoke the secrets manager api .Takes an argument “name” as string and its value is set as /projects/<your project>/secrets/<key name as on secrets manager console>/versions/<version number of key as on SM console>. The result from the API call is saved in a variable secretResult_username. This result is base-64 encoded.
Step storeSecret: In order to use the secret values obtained , we need to decode the result and convert to text . For this , we will use APIs provided by workflows: text.decode and base64.decode. The final username , password are stored in a dictionary credentials, with keys as username and password and values as their respective values fetched from SM API.

Note: In order to access the secret manager, “Secret Manager Secret Accessor” role needs to be assigned to the accessing member.

Problem Statement 2 : Publishing JSON formatted data into Pub/Sub through Workflows

Here we have data in a JSON format , that we wish to publish to a particular pub/sub topic, through workflows. A pub/sub topic and its subscription has been created upfront. This is achieved with the following subworkflow:

publish_data:
    params: [finaljson]
    steps:
        - preparedata:
            assign:
                - strdata: ${base64.encode(text.encode(json.encode_to_string(finaljson)))}
        - publish_pubsub:
            call: http.post
            args:
                url: https://pubsub.googleapis.com/v1/projects/<project-name>/topics/<topic-name>:publish
                headers:
                    Content-type: "application/json"
                auth:
                    type: OAuth2
                body:
                    "messages": 
                        "data": ${strdata}
            result: resp
        - returnOutput:
            return: ${resp}

This sub workflow accepts as a parameter , the data to be published to our pub/sub topic, in JSON format.
Step prepareData: In order to publish JSON into a topic through workflows, we need to encode it to string → encode to text → encode to base64. This conversion is the only necessary masaging that need to be done to json formatted data. For this purpose also we will use the api’s provided by workflows.
Step publish_pubsub: This is a call step that invokes the pub/sub api, through an HTTP.Post call, while the encoded data is passed as request body under “messages” → “data”. The result of this call is stored in variable resp (which is basically the message id of the msg published.)
Step returnOutput: Simply returns the value of response stored in the resp variable to main workflow.

Problem Statement 3: Executing a workflow from another workflow (through API call)

We may come across use cases where execution of a workflow is required in a nested manner. Which means one workflow , calling another workflow iteratively. To resolve this , we used the following workflow:

test_args:
    params : [args]
    steps:
        - step1:
            assign:
                - i: 0       - step2:
            switch:
                - condition: ${len(args) > i}
                  next: step3
            next: step5
       - step3:
            call : subwf_1C
            args:
                pb1 : ${args[i]}
            result: sub_res
            next: step4
        - step4:
            assign:
                - i: ${i+1}
            next: step2
        - step5:
            return : "exit loop"subwf_1:
    params: [args]
    steps:
        - preparedata:-
            assign:
                - strdata: ${json.encode_to_string(args)}
 
        - call_next_wf:
            call: http.post
            args:
                url: https://workflowexecutions.googleapis.com/v1beta/projects/<your-project-name>/locations/<project-location>/workflows/<name-of-wf-tobe-called>/executions
                headers:
                    Content-type: "application/json"
                auth:
                    type: OAuth2
                body:
                    "argument": ${strdata}            result: resp

Main Workflow Execution:

* Step1: Assigns a variable i to 0.

* Step 2: Checks for the length of arguments . If length of argument array is greater than 0, step3 is called, else step 5.

* Step3 : Calls a sub workflow passing the i(th) element of the main workflow argument. Next step to be executed is step 4.

* Step4: Increments the variable i , and calls step 2 again.

* Step 5: Returns “exit loop” .

Sub- Workflow Execution(called from step3):

* Step preparedata: Converts the incoming JSON parameter to string and stores in variable “strdata”.

* Step call_next_wf: Calls a workflow execution using an http.post request. If arguments have to be passed to this wf, they are passed as request body as body → argument → data .(string format). Result from this wf is stored in variable resp.

Guide to integrate Google Cloud Workflows with other Google services.

Written by Megha Mohini