Using Pub/Sub with Knative serving

---

This tutorial shows how to write, deploy, and call a Knative serving service from a Pub/Sub push subscription.

Objectives

• Write, build, and deploy a service to Knative serving
• Call the service by publishing a message to a Pub/Sub topic.

Costs

In this document, you use the following billable components of Google Cloud:

• Cloud Build
• Container Registry
• Pub/Sub
• Knative serving

To generate a cost estimate based on your projected usage, use the pricing calculator. New Google Cloud users might be eligible for a free trial.

Before you begin

1. Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.

2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

3. Make sure that billing is enabled for your Google Cloud project.

4. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

5. Make sure that billing is enabled for your Google Cloud project.

6. Enable the Knative serving API
7. Install and initialize the gcloud CLI.
8. Install the kubectl component:

   gcloud components install kubectl

9. Update components:

   gcloud components update

10. If you are using Knative serving, create a new cluster using the instructions in Setting up Knative serving.

Setting up gcloud defaults

To configure gcloud with defaults for your Knative serving service:

1. Set your default project:

   gcloud config set project PROJECT_ID

   Replace PROJECT_ID with the name of the project you use for this tutorial.

2. Configure gcloud for your cluster:

   gcloud config set run/platform gke
   gcloud config set run/cluster CLUSTER-NAME
   gcloud config set run/cluster_location REGION

   Replace:

   • CLUSTER-NAME with the name you used for your cluster,
   • REGION with the supported cluster location of your choice.

Creating a Pub/Sub topic

The sample service is triggered by messages published to a Pub/Sub topic, so you'll need to create a topic in Pub/Sub.

To create a new Pub/Sub topic, use the command:

gcloud pubsub topics create myRunTopic

You can use myRunTopic or replace with a topic name unique within your Google Cloud project.

Retrieving the code sample

To retrieve the code sample for use:

1. Clone the sample app repository to your local machine:

   Node.js

   git clone https://github.com/GoogleCloudPlatform/nodejs-docs-samples.git

   Alternatively, you can download the sample as a zip file and extract it.

   Python

   git clone https://github.com/GoogleCloudPlatform/python-docs-samples.git

   Alternatively, you can download the sample as a zip file and extract it.

   Go

   git clone https://github.com/GoogleCloudPlatform/golang-samples.git

   Alternatively, you can download the sample as a zip file and extract it.

   Java

   git clone https://github.com/GoogleCloudPlatform/java-docs-samples.git

   Alternatively, you can download the sample as a zip file and extract it.

2. Change to the directory that contains the Knative serving sample code:

   Node.js

   cd nodejs-docs-samples/run/pubsub/

   Python

   cd python-docs-samples/run/pubsub/

   Go

   cd golang-samples/run/pubsub/

   Java

   cd java-docs-samples/run/pubsub/

Looking at the code

The code for this tutorial consists of the following:

• A server that handles incoming requests.

  Node.js

  To keep the Node.js service easy to test, the server configuration is separate from the server startup.

  The Node.js web server is set up in app.js.

  const express = require('express');
  const app = express();
  
  // This middleware is available in Express v4.16.0 onwards
  app.use(express.json());

  The web server is started in index.js:

  const app = require('./app.js');
  const PORT = parseInt(parseInt(process.env.PORT)) || 8080;
  
  app.listen(PORT, () =>
    console.log(`nodejs-pubsub-tutorial listening on port ${PORT}`)
  );

  Python

  import base64
  
  from flask import Flask, request
  
  
  app = Flask(__name__)
  

  Go

  
  // Sample run-pubsub is a Cloud Run service which handles Pub/Sub messages.
  package main
  
  import (
  	"encoding/json"
  	"io"
  	"log"
  	"net/http"
  	"os"
  )
  
  func main() {
  	http.HandleFunc("/", HelloPubSub)
  	// Determine port for HTTP service.
  	port := os.Getenv("PORT")
  	if port == "" {
  		port = "8080"
  		log.Printf("Defaulting to port %s", port)
  	}
  	// Start HTTP server.
  	log.Printf("Listening on port %s", port)
  	if err := http.ListenAndServe(":"+port, nil); err != nil {
  		log.Fatal(err)
  	}
  }
  

  Java

  import org.springframework.boot.SpringApplication;
  import org.springframework.boot.autoconfigure.SpringBootApplication;
  
  @SpringBootApplication
  public class PubSubApplication {
    public static void main(String[] args) {
      SpringApplication.run(PubSubApplication.class, args);
    }
  }

• A handler that processes the Pub/Sub message and logs a greeting.

  Node.js

  app.post('/', (req, res) => {
    if (!req.body) {
      const msg = 'no Pub/Sub message received';
      console.error(`error: ${msg}`);
      res.status(400).send(`Bad Request: ${msg}`);
      return;
    }
    if (!req.body.message) {
      const msg = 'invalid Pub/Sub message format';
      console.error(`error: ${msg}`);
      res.status(400).send(`Bad Request: ${msg}`);
      return;
    }
  
    const pubSubMessage = req.body.message;
    const name = pubSubMessage.data
      ? Buffer.from(pubSubMessage.data, 'base64').toString().trim()
      : 'World';
  
    console.log(`Hello ${name}!`);
    res.status(204).send();
  });

  Python

  @app.route("/", methods=["POST"])
  def index():
      """Receive and parse Pub/Sub messages."""
      envelope = request.get_json()
      if not envelope:
          msg = "no Pub/Sub message received"
          print(f"error: {msg}")
          return f"Bad Request: {msg}", 400
  
      if not isinstance(envelope, dict) or "message" not in envelope:
          msg = "invalid Pub/Sub message format"
          print(f"error: {msg}")
          return f"Bad Request: {msg}", 400
  
      pubsub_message = envelope["message"]
  
      name = "World"
      if isinstance(pubsub_message, dict) and "data" in pubsub_message:
          name = base64.b64decode(pubsub_message["data"]).decode("utf-8").strip()
  
      print(f"Hello {name}!")
  
      return ("", 204)
  
  

  Go

  
  // PubSubMessage is the payload of a Pub/Sub event.
  // See the documentation for more details:
  // https://cloud.google.com/pubsub/docs/reference/rest/v1/PubsubMessage
  type PubSubMessage struct {
  	Message struct {
  		Data []byte `json:"data,omitempty"`
  		ID   string `json:"id"`
  	} `json:"message"`
  	Subscription string `json:"subscription"`
  }
  
  // HelloPubSub receives and processes a Pub/Sub push message.
  func HelloPubSub(w http.ResponseWriter, r *http.Request) {
  	var m PubSubMessage
  	body, err := io.ReadAll(r.Body)
  	defer r.Body.Close()
  	if err != nil {
  		log.Printf("io.ReadAll: %v", err)
  		http.Error(w, "Bad Request", http.StatusBadRequest)
  		return
  	}
  	// byte slice unmarshalling handles base64 decoding.
  	if err := json.Unmarshal(body, &m); err != nil {
  		log.Printf("json.Unmarshal: %v", err)
  		http.Error(w, "Bad Request", http.StatusBadRequest)
  		return
  	}
  
  	name := string(m.Message.Data)
  	if name == "" {
  		name = "World"
  	}
  	log.Printf("Hello %s!", name)
  }
  

  Java

  import com.example.cloudrun.Body;
  import java.util.Base64;
  import org.apache.commons.lang3.StringUtils;
  import org.springframework.http.HttpStatus;
  import org.springframework.http.ResponseEntity;
  import org.springframework.web.bind.annotation.RequestBody;
  import org.springframework.web.bind.annotation.RequestMapping;
  import org.springframework.web.bind.annotation.RequestMethod;
  import org.springframework.web.bind.annotation.RestController;
  
  // PubsubController consumes a Pub/Sub message.
  @RestController
  public class PubSubController {
    @RequestMapping(value = "/", method = RequestMethod.POST)
    public ResponseEntity<String> receiveMessage(@RequestBody Body body) {
      // Get PubSub message from request body.
      Body.Message message = body.getMessage();
      if (message == null) {
        String msg = "Bad Request: invalid Pub/Sub message format";
        System.out.println(msg);
        return new ResponseEntity<>(msg, HttpStatus.BAD_REQUEST);
      }
  
      String data = message.getData();
      String target =
          !StringUtils.isEmpty(data) ? new String(Base64.getDecoder().decode(data)) : "World";
      String msg = "Hello " + target + "!";
  
      System.out.println(msg);
      return new ResponseEntity<>(msg, HttpStatus.OK);
    }
  }

  You must code the service to return an accurate HTTP response code. Success codes, such as HTTP 200 or 204, acknowledge complete processing of the Pub/Sub message. Error codes, such as HTTP 400 or 500, indicate the message will be retried, as described in Receiving messages using Push guide.

• A Dockerfile that defines the operating environment for the service. The contents of the Dockerfile vary by language.

  Node.js

  
  # Use the official lightweight Node.js image.
  # https://hub.docker.com/_/node
  FROM node:20-slim
  # Create and change to the app directory.
  WORKDIR /usr/src/app
  
  # Copy application dependency manifests to the container image.
  # A wildcard is used to ensure both package.json AND package-lock.json are copied.
  # Copying this separately prevents re-running npm install on every code change.
  COPY package*.json ./
  
  # Install dependencies.
  # if you need a deterministic and repeatable build create a
  # package-lock.json file and use npm ci:
  # RUN npm ci --omit=dev
  # if you need to include development dependencies during development
  # of your application, use:
  # RUN npm install --dev
  
  RUN npm install --omit=dev
  
  # Copy local code to the container image.
  COPY . .
  
  # Run the web service on container startup.
  CMD [ "npm", "start" ]
  

  Python

  
  # Use the official Python image.
  # https://hub.docker.com/_/python
  FROM python:3.11
  
  # Allow statements and log messages to immediately appear in the Cloud Run logs
  ENV PYTHONUNBUFFERED True
  
  # Copy application dependency manifests to the container image.
  # Copying this separately prevents re-running pip install on every code change.
  COPY requirements.txt ./
  
  # Install production dependencies.
  RUN pip install -r requirements.txt
  
  # Copy local code to the container image.
  ENV APP_HOME /app
  WORKDIR $APP_HOME
  COPY . ./
  
  # Run the web service on container startup.
  # Use gunicorn webserver with one worker process and 8 threads.
  # For environments with multiple CPU cores, increase the number of workers
  # to be equal to the cores available.
  # Timeout is set to 0 to disable the timeouts of the workers to allow Cloud Run to handle instance scaling.
  CMD exec gunicorn --bind :$PORT --workers 1 --threads 8 --timeout 0 main:app
  

  Go

  
  # Use the offical golang image to create a binary.
  # This is based on Debian and sets the GOPATH to /go.
  # https://hub.docker.com/_/golang
  FROM golang:1.21-bookworm as builder
  
  # Create and change to the app directory.
  WORKDIR /app
  
  # Retrieve application dependencies.
  # This allows the container build to reuse cached dependencies.
  # Expecting to copy go.mod and if present go.sum.
  COPY go.* ./
  RUN go mod download
  
  # Copy local code to the container image.
  COPY . ./
  
  # Build the binary.
  RUN go build -v -o server
  
  # Use the official Debian slim image for a lean production container.
  # https://hub.docker.com/_/debian
  # https://docs.docker.com/develop/develop-images/multistage-build/#use-multi-stage-builds
  FROM debian:bookworm-slim
  RUN set -x && apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y \
      ca-certificates && \
      rm -rf /var/lib/apt/lists/*
  
  # Copy the binary to the production image from the builder stage.
  COPY --from=builder /app/server /server
  
  # Run the web service on container startup.
  CMD ["/server"]
  

  Java

  This sample uses Jib to build Docker images using common Java tools. Jib optimizes container builds without the need for a Dockerfile or having Docker installed. Learn more about building Java containers with Jib.

  <plugin>
    <groupId>com.google.cloud.tools</groupId>
    <artifactId>jib-maven-plugin</artifactId>
    <version>3.4.0</version>
    <configuration>
      <to>
        <image>gcr.io/PROJECT_ID/pubsub</image>
      </to>
    </configuration>
  </plugin>
  

For details on how to authenticate the origin of Pub/Sub requests, read the section below on Integrating with Pub/Sub.

Shipping the code

Shipping code consists of three steps: building a container image with Cloud Build, uploading the container image to Container Registry, and deploying the container image to Knative serving.

To ship your code:

1. Build your container and publish on Container Registry:

   Node.js

   gcloud builds submit --tag gcr.io/PROJECT_ID/pubsub

   Where PROJECT_ID is your Google Cloud project ID, and pubsub is the name you want to give your service.

   Upon success, you should see a SUCCESS message containing the ID, creation time, and image name. The image is stored in Container Registry and can be re-used if desired.

   Python

   gcloud builds submit --tag gcr.io/PROJECT_ID/pubsub

   Where PROJECT_ID is your Google Cloud project ID, and pubsub is the name you want to give your service.

   Upon success, you should see a SUCCESS message containing the ID, creation time, and image name. The image is stored in Container Registry and can be re-used if desired.

   Go

   gcloud builds submit --tag gcr.io/PROJECT_ID/pubsub

   Where PROJECT_ID is your Google Cloud project ID, and pubsub is the name you want to give your service.

   Upon success, you should see a SUCCESS message containing the ID, creation time, and image name. The image is stored in Container Registry and can be re-used if desired.

   Java

   mvn compile jib:build -Dimage=gcr.io/PROJECT_ID/pubsub

   Where PROJECT_ID is your Google Cloud project ID, and pubsub is the name you want to give your service.

   Upon success, you should see a BUILD SUCCESS message. The image is stored in Container Registry and can be re-used if desired.

2. Run the following command to deploy your app:

   gcloud run deploy pubsub-tutorial --image gcr.io/PROJECT_ID/pubsub

   Replace PROJECT_ID with your Google Cloud project ID. pubsub is the container name and pubsub-tutorial is the name of the service. Notice that the container image is deployed to the service and cluster that you configured previously under Setting up gcloud

   Wait until the deployment is complete: this can take about half a minute. On success, the command line displays the service URL. This URL is used to configure a Pub/Sub subscription.

3. If you want to deploy a code update to the service, repeat the previous steps. Each deployment to a service creates a new revision and automatically starts serving traffic when ready.

Integrating with Pub/Sub

Now that we have deployed our Knative serving service, we will configure Pub/Sub to push messages to it.

To integrate the service with Pub/Sub:

1. Enable Pub/Sub to create authentication tokens in your project:

   gcloud projects add-iam-policy-binding PROJECT_ID \
        --member=serviceAccount:[e-mail] \
        --role=roles/iam.serviceAccountTokenCreator

   Replace

   • PROJECT_ID with your Google Cloud project ID.
   • PROJECT-NUMBER with your Google Cloud project number.

2. Create or select a service account to represent the Pub/Sub subscription identity.

   gcloud iam service-accounts create cloud-run-pubsub-invoker \
        --display-name "Cloud Run for Anthos Pub/Sub Invoker"

   You can use cloud-run-pubsub-invoker or replace with a name unique within your Google Cloud project.

3. Create a Pub/Sub subscription with the service account:

   1. Enable auto TLS and HTTPS for your cluster and add a domain mapping to your service.

   2. Register domain ownership for Pub/Sub.

   3. Add code to validate the authentication token attached to Pub/Sub messages. Sample code is provided in Authentication and authorization by the push endpoint.

      The authentication must ensure that the token is valid and associated with the expected service account. Unlike Cloud Run, Knative serving has no built-in authorization check that the token is valid or that the service account has authorization to invoke the Knative serving service.

   4. Create a Pub/Sub subscription with the service account:

      gcloud pubsub subscriptions create myRunSubscription --topic myRunTopic \
           --push-endpoint=SERVICE-URL/ \
           --push-auth-service-account=cloud-run-pubsub-invoker@PROJECT_ID.iam.gserviceaccount.com

      Replace

      • myRunTopic with the topic you previously created.
      • SERVICE-URL with your custom service URL. Specify https as the protocol.
      • PROJECT_ID with your Google Cloud project ID.

      The --push-auth-service-account flag activates the Pub/Sub push functionality for Authentication and authorization.

Your service is now fully integrated with Pub/Sub.

Trying it out

To test the end-to-end solution:

1. Send a Pub/Sub message to the topic:

   gcloud pubsub topics publish myRunTopic --message "Runner"

   You can also publish messages programmatically instead of using the command-line as shown in this tutorial. For more information, see Publishing messages.

2. Navigate to the service logs:

   1. Navigate to the Knative serving page in the Google Cloud console:

      Go to Knative serving

   2. Click the pubsub-tutorial service.

   3. Select the Logs tab.

      Logs might take a few moments to appear. If you don't see them immediately, check again after a few moments.

3. Look for the "Hello Runner!" message.

Clean up

To walk through a more in-depth use case of using Knative serving with Pub/Sub, skip cleanup for now and continue with the Image processing tutorial.

If you created a new project for this tutorial, delete the project. If you used an existing project and wish to keep it without the changes added in this tutorial, delete resources created for the tutorial.

Deleting the project

The easiest way to eliminate billing is to delete the project that you created for the tutorial.

To delete the project:

Caution: Deleting a project has the following effects:

• Everything in the project is deleted. If you used an existing project for the tasks in this document, when you delete it, you also delete any other work you've done in the project.
• Custom project IDs are lost. When you created this project, you might have created a custom project ID that you want to use in the future. To preserve the URLs that use the project ID, such as an appspot.com URL, delete selected resources inside the project instead of deleting the whole project.

If you plan to explore multiple architectures, tutorials, or quickstarts, reusing projects can help you avoid exceeding project quota limits.

1. In the Google Cloud console, go to the Manage resources page.

   Go to Manage resources

2. In the project list, select the project that you want to delete, and then click Delete.
3. In the dialog, type the project ID, and then click Shut down to delete the project.

Deleting tutorial resources

1. Delete the Knative serving service you deployed in this tutorial:

   gcloud run services delete SERVICE-NAME

   Where SERVICE-NAME is your chosen service name.

   You can also delete Knative serving services from the Google Cloud console:

   Go to Knative serving

2. Remove the gcloud default configurations you added during the tutorial setup:

    gcloud config unset run/platform
    gcloud config unset run/cluster
    gcloud config unset run/cluster_location
   

3. Remove the project configuration:

    gcloud config unset project
   

4. Delete other Google Cloud resources created in this tutorial:

   • Delete the Pub/Sub topic myRunTopic
   • Delete the Pub/Sub subscription myRunSubscription
   • Delete the container image named gcr.io/PROJECT_ID/pubsub from Container Registry.
   • Delete the invoker service account cloud-run-pubsub-invoker@PROJECT_ID.iam.gserviceaccount.com
   • If you created a cluster for this tutorial, delete the cluster

What's next

• Expand the sample service deployed in this tutorial to add image processing functionality that modifies images uploaded to Cloud Storage.
• Learn more about how topics fit into Pub/Sub architecture and how to manage topics
• Learn more about Pub/Sub subscriptions in managing subscriptions.
• Explore reference architectures, diagrams, and best practices about Google Cloud. Take a look at our Cloud Architecture Center.