GCP Pub Sub

This guide shows you how to invoke a Fission function from a message in a GCP Pub/Sub topic, using Fission’s Keda-based message queue trigger. GCP Pub/Sub is a global messaging system for event-driven systems and streaming analytics.

Prerequisites

We’ll assume you have Fission and Kubernetes installed. If not, please head over to the Fission install guide.

To enable KEDA integration, set the flag mqt_keda.enabled to true while installing Fission with helm chart.

You need to install Keda Helm Chart in your cluster for Fission Keda GCP Pubsub trigger to work.

Google Cloud Platform - PubSub

You will need an active Google Cloud account for this demo. If you don’t have one, you can register for a Google Cloud Account.

Creating a Project and Topics

To start with this demo, we will create a project followed by three topics

1. Create a New Project by providing a name.
2. In the search bar, search for PubSub.
3. Click on Create Topic button and give a Topic ID to create a topic.
4. Create three topics: request-topic, response-topic & error-topic.

Setting up Authentication

Authentication ensures that only your Fission function can send and receive messages on the queue. Follow GCP’s Steps to Setup Authentication for Pub/Sub. This downloads a json file with the credentials, which you’ll use below.

Overview

Before we dive into details, let us walk through the overall flow of events and functions involved.

1. A Python producer function (producer) that drops a message in a GCP PubSub queue named request-topic.
2. Fission GCP PubSub trigger that activates upon message arrival in request-topic and invokes another function (consumer) with message received from producer.
3. The consumer function (consumer) gets body of message and returns a response.
4. Fission GCP PubSub trigger takes the response of consumer function (consumer) and drops the message in a response queue named response-topic. If there is an error, the message is dropped in error queue named error-topic.

Sample App

You can get the source code for the sample app explained in this document in our Keda GCP PubSub Trigger Repo

Building the app

Secret

We’ll create a Kubernetes secret holding the credentials our function needs to connect to the GCP queue. This demo requires the GoogleApplicationCredentials env variable to be set. The following command creates the secret:

kubectl create secret generic pubsub-secret --from-file=GoogleApplicationCredentials=filename.json --from-literal=PROJECT_ID=project_id

Replace the filename and project_id with your values

Publisher Function

The publisher function is a Python program which creates a message with a number and drops into a queue request-topic.

from google.cloud import pubsub_v1
import os

path = os.path.dirname(os.path.realpath(__file__))

# GCP Credentials
os.environ["GOOGLE_APPLICATION_CREDENTIALS"]=path+"/filename.json"

# Project and Topic Ids
project_id = "projectid"
topic_id = "request-topic"

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(project_id, topic_id)

def main():

    for n in range(1, 100):
        data = f"Message number {n}"
        # Data must be a bytestring
        data = data.encode("utf-8")
        # When you publish a message, the client returns a future.
        future = publisher.publish(topic_path, data)
        print(future.result())

    print(f"Published messages to {topic_path}.")

    return "Messages Published!"

Create a zip archive of publisher folder using zip -j pub.zip pub/*

Let’s create the environment, package and function:

fission env create --name python-gcp --image ghcr.io/fission/python-env --builder ghcr.io/fission/python-builder
fission fn create --name producer --env python-gcp --src pub/pub.py  --entrypoint main --src pub/requirements.txt

Consumer Function

The consumer function is a NodeJS program which takes the body of the request, process the message and drops it in the response-queue

module.exports = async function (context) {
    console.log(context.request.body);
    let obj = context.request.body;
    return {
        status: 200,
        body: "Consumer Response "+ JSON.stringify(obj)
    };
}

Creating the consumer function:

fission fn create --name consumer --env nodeenv --code consumer.js

Connecting via trigger

Now that we have both Publisher and Consumer functions ready, let’s create the Trigger. This message queue trigger will invoke the consumer function every time there is a message in request-topic queue. The response will be sent to response-topic queue and in case of consumer function invocation fails, the error is written to error-topic queue.

fission mqt create --name gcptest --function consumer --mqtype gcp-pubsub --mqtkind keda \
    --topic request-topic-sub --resptopic response-topic --errortopic error-topic \
    --maxretries 3 --cooldownperiod=30 --pollinginterval=5 --metadata subscriptionName=request-topic-sub \
    --metadata credentialsFromEnv=GoogleApplicationCredentials --secret pubsub-secret

Parameter list:

• subscriptionName - Name of the subscription for the request queue for which the trigger is created.
• credentialsFromEnv - GCP credentials for authentication.

Specs

You can also use the following Fission spec. Read our guide on how to use Fission spec.

fission spec init
fission env create --name python-gcp --image ghcr.io/fission/python-env --builder ghcr.io/fission/python-builder --spec
fission fn create --name producer --env python-gcp --src pub/pub.py  --entrypoint main --src pub/requirements.txt --spec
fission env create --name nodeenv --image ghcr.io/fission/node-env --spec
fission fn create --name consumer --env nodeenv --code consumer.js --spec
fission mqt create --name gcptest --function consumer --mqtype gcp-pubsub --mqtkind keda \
    --topic request-topic-sub --resptopic response-topic --errortopic error-topic --maxretries 3 \
    --cooldownperiod=30 --pollinginterval=5 --metadata subscriptionName=request-topic-sub \
    --metadata credentialsFromEnv=GoogleApplicationCredentials --secret pubsub-secret --spec
fission spec apply

Testing it out

Let’s invoke the producer function so that the queue request-topic gets some messages and we can see the consumer function in action.

$ fission fn test --name producer
3845235698474295
3845233690419265
3845239134812172
3845232823384352
3845239513065630
Published messages to projects/projectid/topics/request-topic.
Messages Published!

To add authentication to your function calls, refer to our Fission Authentication guide.

To verify if the messages were successfully sent, navigate to the Google Cloud console and observe the request-topic-sub subscription and verify the count of messages sent.

If you’ve followed the tutorial correctly, the message queue trigger will be triggered and our consumer function will be invoked.

When the consumer function is invoked, it will process the message in the request-topic and push them to the response-topic. We can verify the same by checking the messages in the response-topic on the GCP console.

Debugging

For debugging, check the logs of the pods created in the fission and default namespace.

Function pods are typically created in the default namespace, named based on the environment. Check both the consumer and producer function logs.

Try out the Sample app to see it in action.