Copy page

Cloud Security

Google Pub/Sub

Ingests messages from a Google Cloud Pub/Sub subscription.

Sync Type: Streaming

Overview

The Google Cloud Pub/Sub input allows you to consume streaming messages from a Google Cloud Pub/Sub subscription. This input is ideal for ingesting real-time event data, logs, and messages published to a Pub/Sub topic by Google Cloud services or your own applications.

Key Features

• Real-time Message Consumption: Pulls messages as they arrive in the subscription
• Automatic Acknowledgment: Messages are acknowledged as soon as they are handed to the pipeline
• Pub/Sub redelivery: Messages that were not acknowledged are redelivered by Pub/Sub after the acknowledgement deadline; the input keeps no checkpoint state of its own
• Configurable Subscription: Specify any Pub/Sub subscription within your GCP project

How It Works

The input continuously pulls messages from the specified Pub/Sub subscription:

1. Connection: Connects to the subscription using GCP credentials
2. Message Consumption: Pulls messages as they arrive using the Pub/Sub client defaults (up to 1000 outstanding messages)
3. Processing: The message data is emitted as the record; the message ID and publish time become record metadata and message attributes are not included
4. Acknowledgment: Each message is acknowledged as soon as the input hands it to the pipeline

Requirements

Before setting up the Google Cloud Pub/Sub input, you need to:

1. Have a Google Cloud Platform account with access to the desired project
2. Have an existing Pub/Sub subscription created and ready to receive messages
3. Create a service account with the necessary permissions
4. Generate a JSON key for the service account

Details

Authentication

The input uses Google Cloud service account credentials for authentication. You must provide a JSON service account key that has permissions to pull messages from your Pub/Sub subscription.

Message Delivery

• Messages are pulled from the subscription using the Pub/Sub client defaults (up to 1000 outstanding messages)
• The record is the message data; the message ID and publish time are carried as record metadata
• Message attributes are not included in the record
• Messages are acknowledged as soon as they are handed to the pipeline

Checkpointing

The input keeps no checkpoint state of its own. If the connection is interrupted, Pub/Sub redelivers every message that was not acknowledged. A message is acknowledged as soon as the input hands it to the pipeline, so a failure downstream of the input after that point is not redelivered.

Prerequisites

Service Account Setup

1. Navigate to the Google Cloud Console
2. Select your project from the dropdown at the top
3. Go to IAM & Admin > Service Accounts
4. Click Create Service Account
5. Enter a name (e.g., "monad-pubsub-input") and click Create
6. Assign the following role to the service account:
   • Pub/Sub Subscriber - Required to pull messages from subscriptions
7. Click Continue and then Done
8. Find the service account in the list and click on it
9. Go to the Keys tab and click Add Key > Create new key
10. Choose JSON as the key type and click Create
11. The JSON key file will be downloaded - store it securely

Subscription Requirements

Ensure that:

1. You have an existing Pub/Sub subscription created
2. The service account has the "Pub/Sub Subscriber" role on the project
3. The subscription is active and can receive messages
4. You know the subscription's name (format: typically just the name, like "my-subscription")

Configuration

Settings

Setting	Type	Required	Default	Description
project_id	string	Yes	-	The GCP project ID that contains the Pub/Sub subscription.
subscription_id	string	Yes	-	The Pub/Sub subscription name to pull messages from.

Secrets

Secret	Type	Required	Description
credentials_json	secret	No	JSON credentials to authenticate with Google Cloud (required for service account auth).

Setting up the Input

1. In the Monad UI, go to the Inputs section
2. Click Add Input and select Google Cloud Pub/Sub
3. Configure the input settings:
   • Project ID: Your Google Cloud project ID (e.g., my-project-123)
   • Subscription ID: The name of your Pub/Sub subscription (e.g., my-subscription)
4. In the Secrets section, paste the entire contents of your service account JSON key file into the Credentials JSON field
5. Click Save to configure the input

Recommended path for high-volume Google Cloud logs

For high-volume or no-loss log collection, this input is the recommended way to pull Google Cloud Logging data: Cloud Audit Logs, GKE audit logs, Looker Core audit logs, or any other type.

The direct-read connectors (Cloud Logs, GKE Audit Logs, and Looker Core Audit Logs) read straight from the Cloud Logging API. That is simpler to set up, but reads are capped at 60,000 entries per minute per project (a shared, non-raisable quota) with no Google-managed checkpoint, so at high volume entries can lag or drop. Google's recommended way to export logs to a SIEM or third party is to route them through a Log Router sink into Pub/Sub, then consume that topic with this input:

Code
Cloud Logging  ->  Log Router sink  ->  Pub/Sub topic  ->  subscription  ->  Monad Pub/Sub input

The subscription's ack cursor is a Google-managed checkpoint, so this path has no self-managed watermark, no read-API cap, and no silent drops, and it resumes cleanly after a restart. Late or backdated entries flow through like any other message, and one aggregated sink can cover a whole folder or organization.

Setup

1. Create a Pub/Sub topic (e.g. monad-logs) and a pull subscription on it (e.g. monad-logs-sub) in the Google Cloud Console.
2. Create a Log Router sink to the topic: go to Logging > Log Router, click Create Sink, then:
   • In Sink Service, select Cloud Pub/Sub topic and choose the topic from step 1.
   • Under Choose logs to include in sink, optionally set an inclusion filter to route a subset (see Inclusion filter recipes below), or leave it empty to route all logs.
   • Click Create Sink.
3. Grant the sink's writer identity the Pub/Sub Publisher (roles/pubsub.publisher) role on the topic. Cloud Logging assigns this identity when the sink is created and shows it in the sink details.
4. Grant Monad's service account the Pub/Sub Subscriber (roles/pubsub.subscriber) role on the subscription (the same service account whose JSON key you provide, per Prerequisites).
5. Configure this input with the project ID, the subscription ID from step 1, and the service account credentials, as described in Setting up the Input above.

The equivalent gcloud commands, for reference (replace PROJECT_ID and the resource names):

Code
# 1. Topic + subscription
gcloud pubsub topics create monad-logs
gcloud pubsub subscriptions create monad-logs-sub --topic monad-logs

# 2. Log Router sink. --log-filter is optional: omit it to route ALL logs, or
#    set it to one of the recipes in the table below to route a subset.
gcloud logging sinks create monad-logs-sink \
  pubsub.googleapis.com/projects/PROJECT_ID/topics/monad-logs \
  --log-filter='logName:"cloudaudit.googleapis.com"'

# 3. Grant the sink's writer identity publisher on the topic
#    (get WRITER_IDENTITY from: gcloud logging sinks describe monad-logs-sink)
gcloud pubsub topics add-iam-policy-binding monad-logs \
  --member=WRITER_IDENTITY --role=roles/pubsub.publisher

# 4. Grant Monad's service account subscriber on the subscription
gcloud pubsub subscriptions add-iam-policy-binding monad-logs-sub \
  --member=serviceAccount:MONAD_SERVICE_ACCOUNT --role=roles/pubsub.subscriber

Inclusion filter recipes

The inclusion filter is optional: leave it empty to route every log entry (as Google's built-in _Default sink does), or set one of the filters below to route a subset. These use the Logging query language; the audit-log filter is Google's own "all audit logs" query, and the GKE and Looker filters match what the direct-read connectors apply:

Logs to route	Inclusion filter
Everything	Leave the inclusion filter empty (no filter)
All Cloud Audit Logs (activity, data access, system event, policy)	logName:"cloudaudit.googleapis.com"
GKE audit logs	resource.type="k8s_cluster"
Looker Core audit logs	resource.type="audited_resource" AND protoPayload.serviceName="looker.googleapis.com"

Combine with additional conditions as needed, for example resource.type="k8s_cluster" AND resource.labels.cluster_name="my-cluster" to scope to one GKE cluster, or severity>=WARNING to reduce volume.

Delivered payload shape

The Log Router publishes each log entry as raw LogEntry JSON in the Pub/Sub message body, and this input emits that JSON as the record. This is close to the direct-read connectors' output (both are LogEntry-shaped) but not identical: the direct-read connectors proto-parse protoPayload from the read API, while this path delivers the entry as Cloud Logging routes it. If you already have transforms or detections built against the direct-read output (for example an OCSF transform written against the Cloud Logs input), validate them against the routed shape before switching a pipeline over.

Cost

Routing to Pub/Sub adds Pub/Sub charges on top of Cloud Logging costs, and Data Access audit logs can be high volume. Use the inclusion filter to route only what you need, and review Pub/Sub pricing and Cloud Logging pricing.

Rate Limits

Google Cloud Pub/Sub has the following rate limits and considerations:

Scope	Limit	Notes
Pull Requests	1,000 requests per second per subscription	Aggregated across all connections
Message Rate	Based on subscription quota	Configured in your GCP project
Concurrent Connections	Configurable per project	Default allows multiple concurrent subscribers

Related Articles

• Google Cloud Pub/Sub Documentation
• Creating and Managing Subscriptions
• Subscribing to Topics
• Access Control with IAM
• Routing and storage overview (Log Router)
• Configure and manage sinks
• Stream logs to Pub/Sub for third-party integrations
• Cloud Logging quotas and limits

Sync frequency

This input consumes from the source queue or subscription, ingesting records at whatever rate the source delivers them rather than on a fixed poll. See Input Sync Frequency for details.