Skip to main content

GCP Pub/Sub triggers

Windmill can connect to Google Cloud Pub/Sub and trigger runnables (scripts, flows) when messages are published on topics.
You can configure Windmill to either pull messages from subscriptions or receive pushed messages via auto-generated endpoints.

Google Cloud Pub/Sub triggers is a self-hosted Enterprise feature.

How to use​

Configure GCP connection​

Windmill can authenticate to Google Cloud in two ways, selected with a toggle in the trigger editor:

  • Service account: select an existing GCP resource (service account credentials) or create a new one.
  • Application default credentials: authenticate as the Windmill server itself, using the credentials of its environment. Selecting this mode requires being a workspace admin. See Application default credentials.

You can also set a Project ID: the GCP project in which topics and subscriptions are listed and created. Leave it empty to use the project of the credentials. Topic and subscription IDs given as fully qualified names (projects/<project>/topics/<id>, projects/<project>/subscriptions/<id>) are reached whatever it is set to, so a subscription can live in a different project than its topic.

The identity used (service account or application default credentials) must have enough permissions for Windmill to fully manage Pub/Sub resources. Specifically:

  • Pub/Sub Viewer (roles/pubsub.viewer): to check if topics or subscriptions exist, list them.
  • Pub/Sub Subscriber (roles/pubsub.subscriber): to attach to subscriptions and consume messages.
  • Pub/Sub Editor (roles/pubsub.editor): needed to create or update subscriptions, and to optionally delete the subscription in the cloud when deleting the associated trigger if the user chooses to do so.

If you prefer not to assign these three individually, you can simply grant the Pub/Sub Admin role (roles/pubsub.admin).

Additionally, if you want to create authenticated push delivery subscriptions, the service account must also have Service Account User (roles/iam.serviceAccountUser) permission. See Authenticate Push Subscriptions for more details.

Application default credentials​

Instead of handing Windmill a downloadable service account key through a resource, a trigger can use the server's own application default credentials (ADC). This resolves credentials from the server's environment: GKE Workload Identity, the GCE metadata server, the gcloud well-known file, or a credentials file pointed to by the GOOGLE_APPLICATION_CREDENTIALS environment variable on the server.

Because these credentials are the server's identity and no workspace ACL covers them (unlike a gcloud resource, whose ACL decides who may use it), only workspace admins can select this mode. Saving any change to such a trigger also requires workspace admin, since saving re-provisions the subscription with those credentials, and so does enabling or disabling it. Non-admins can still view a trigger that already uses application default credentials.

Workload Identity Federation credential files are supported with the file, url and aws credential sources; the executable source is not supported.

Some ADC files (Workload Identity Federation in particular) carry no project, in which case you should fill in the Project ID field.

Subscription setup​

Select topic and subscription​

  • Choose a topic from your GCP project. You can refresh the list if needed.
  • Decide how to set up your subscription:
    • Create or update a subscription: Windmill will create a new subscription or update an existing one.
    • Use an existing subscription: Link an existing subscription from your GCP project.
When creating/updating a subscription:​
  • Specify a Subscription ID, or leave it empty to auto-generate one.
  • Choose the delivery type:
    • Pull: Windmill sets the subscription as a Pull subscription.
    • Push: Windmill sets the subscription as a Push subscription.
      • For push delivery, Windmill sets the subscription's push endpoint URL to match the path of the trigger.
        The format is:
        {base_endpoint}/api/gcp/w/{workspace_id}/{trigger_path}
      • Example: if the trigger path is u/test/fabulous_trigger, the endpoint will be:
        {base_endpoint}/api/gcp/w/myworkspace/u/test/fabulous_trigger
      • When creating or updating a push subscription, Windmill allows you to configure:
        • Whether authentication is enabled or disabled.

Refer to Google Cloud Pub/Sub - Managing Subscriptions for more details about delivery types.

When using an existing subscription:​
  • Select an existing subscription ID among the subscriptions fetched from the selected topic.
  • Windmill will automatically detect the subscription's delivery type based on the cloud configuration.
  • If the subscription is of push delivery type:
    • The subscription's endpoint URL must match the path of the trigger that will be bound to it.
    • The expected format is:
      {base_endpoint}/api/gcp/w/{workspace_id}/{trigger_path}

Note: You must not have multiple subscriptions pointing to the same trigger URL (for example, two subscriptions targeting {base_endpoint}/api/gcp/w/myworkspace/u/test/fabulous_trigger).

Choose the runnable​

  • Select the script or flow to trigger when Pub/Sub messages are received.

Implementation examples​

Below are examples for handling GCP Pub/Sub messages in Windmill.

Windmill provides the Pub/Sub message as the argument payload (a base64-encoded string) to your runnable.

Basic script​

export async function main(payload: string) {
  const decoded = new TextDecoder().decode(Uint8Array.from(atob(payload), c => c.charCodeAt(0)));

  try {
    const jsonData = JSON.parse(decoded);
    console.log("Received JSON data:", jsonData);
    // Process structured data
  } catch (e) {
    console.log("Received plain text:", decoded);
    // Process raw text
  }

  return { processed: true };
}

Using a preprocessor​

If you configure a preprocessor, you can extract fields before they reach the main function.

Windmill provides the Pub/Sub message as the argument payload (a base64-encoded string) to the preprocessor.

GCP Pub/Sub trigger object​

  • subscription: Subscription ID
  • topic: Topic ID
  • message_id: Unique message ID
  • publish_time: Publish timestamp (RFC 3339 format with Z, e.g., "2024-04-07T12:34:56Z")
  • attributes: Key-value metadata
  • delivery_type: "push" or "pull" (the type of delivery)
  • ordering_key: Ordering key (optional, if message ordering is enabled)
  • headers: HTTP headers for push delivery (only present for push)

Example preprocessor:

export async function preprocessor(
  event: {
    kind: 'gcp',
    payload: string, // base64 encoded payload
    message_id: string,
    subscription: string,
    ordering_key?: string,
    attributes?: Record<string, string>,
    delivery_type: "push" | "pull",
    headers?: Record<string, string>,
    publish_time?: string,
  }
) {
  if (event.kind === 'gcp') {
    const decodedString = atob(event.payload);

    const attributes = event.attributes || {};
    const contentType = attributes['content-type'] || attributes['Content-Type'];
    const isJson = contentType === 'application/json';

    let parsedMessage: any = decodedString;
    if (isJson) {
      try {
        parsedMessage = JSON.parse(decodedString);
      } catch (err) {
        throw new Error(`Invalid JSON payload: ${err}`);
      }
    }

    return {
      messageAsDecodedString: decodedString,
      contentType,
      parsedMessage,
      attributes
    };
  }

  throw new Error(`Expected gcp trigger kind got: ${event.kind}`);
}

Then your main function can simply receive the extracted arguments:

export async function main(
  messageAsDecodedString: string,
  contentType?: string,
  parsedMessage?: any,
  attributes?: Record<string, string>,
) {
  console.log("Decoded String:", messageAsDecodedString);
  console.log("Content-Type:", contentType);
  console.log("Parsed Message:", parsedMessage);
  console.log("Attributes:", attributes);
}

Troubleshooting​

  • Permission issues: Verify the service account has required Pub/Sub permissions. If the correct permissions are set but you still encounter unauthorized or permission denied errors, it might indicate that Google has updated required permissions. Please contact Windmill support so we can investigate and assist.
  • Push delivery failures: If using existing subscription ensure the push endpoint URL matches the required format ({base_endpoint}/api/gcp/w/{workspace_id}/{trigger_path}) and is unique across the workspace.
  • Topic or subscription not found: Refresh the list to fetch the latest available resources.

Error handling​

GCP triggers support local error handlers that override workspace error handlers for specific triggers. See the error handling documentation for configuration details and examples.