Publish and route custom events

Deploy an event receiver to Cloud Run

Using a prebuilt image, gcr.io/cloudrun/hello, deploy a Cloud Run service that will receive and log events. For more information, see Deploying container images.

gcloud run deploy helloworld-service \
    --image=gcr.io/cloudrun/hello \
    --allow-unauthenticated \
    --region=us-central1

When you see the service URL, the deployment is complete.

Create a channel

A channel is a Google Cloud resource that is used as a pipeline to deliver custom events from producers to consumers. Custom events are published to a channel and an Eventarc trigger subscribes to those events.

Create a channel to support the integration between producers and consumers of custom events:

gcloud eventarc channels create CHANNEL_ID

Replace CHANNEL_ID with the resource name for the channel. For example: helloworld-channel.

Retrieve channel properties

After creating a channel, you can retrieve the channel properties and confirm that it is active:

gcloud eventarc channels describe CHANNEL_ID

The output is similar to the following:

createTime: '2022-04-26T20:46:06.113533183Z'
name: projects/your-project-123/locations/us-central1/channels/helloworld-channel
pubsubTopic: projects/your-project-123/topics/eventarc-channel-us-central1-helloworld-channel-465
state: ACTIVE
uid: 86391a0b-a264-4172-a3b5-a893179f1d1a
updateTime: '2022-04-26T20:46:10.106938915Z'

Create a trigger using the gcloud CLI

An Eventarc trigger declares your interest in a certain event or set of events. You can configure event routing by specifying filters for the trigger, including the channel and a destination.

Eventarc supports filtering on any CloudEvents attribute specified by the custom event producer.

You can create a trigger by running a gcloud command along with required and optional flags. Requests to the follow destinations are triggered according to the filters.

gcloud eventarc triggers create TRIGGER \
    --channel=CHANNEL_ID \
    --destination-run-service=DESTINATION_RUN_SERVICE \
    --destination-run-region=DESTINATION_RUN_REGION \
    --event-filters=type=EVENT_TYPE \
    --service-account=$PROJECT_NUMBER-compute@developer.gserviceaccount.com

gcloud eventarc triggers create helloworld-trigger \
    --channel=helloworld-channel \
    --destination-run-service=helloworld-service \
    --destination-run-region=us-central1 \
    --event-filters=type=mycompany.myorg.myproject.v1.myevent \
    --event-filters=someAttr=foo \
    --service-account=$PROJECT_NUMBER-compute@developer.gserviceaccount.com

gcloud eventarc triggers create TRIGGER \
    --channel=CHANNEL_ID \
    --destination-gke-cluster=DESTINATION_GKE_CLUSTER \
    --destination-gke-location=DESTINATION_GKE_LOCATION \
    --destination-gke-namespace=DESTINATION_GKE_NAMESPACE \
    --destination-gke-service=DESTINATION_GKE_SERVICE \
    --destination-gke-path=DESTINATION_GKE_PATH \
    --event-filters=type=EVENT_TYPE \
    --service-account=$PROJECT_NUMBER-compute@developer.gserviceaccount.com

gcloud eventarc triggers create helloworld-trigger \
    --channel=helloworld-channel \
    --destination-gke-cluster=gke-events-cluster \
    --destination-gke-location=us-central1-a \
    --destination-gke-namespace=default \
    --destination-gke-service=helloworld-service \
    --destination-gke-path=/ \
    --event-filters=type=mycompany.myorg.myproject.v1.myevent \
    --event-filters=someAttr=foo \
    --service-account=$PROJECT_NUMBER-compute@developer.gserviceaccount.com

gcloud eventarc triggers create TRIGGER \
    --channel=CHANNEL_ID \
    --destination-workflow=DESTINATION_WORKFLOW \
    --destination-workflow-location=DESTINATION_WORKFLOW_LOCATION \
    --event-filters=type=EVENT_TYPE \
    --service-account=$PROJECT_NUMBER-compute@developer.gserviceaccount.com

gcloud eventarc triggers create helloworld-trigger \
    --channel=helloworld-channel \
    --destination-workflow=helloworld-workflow \
    --destination-workflow-location=us-central1 \
    --event-filters=type=mycompany.myorg.myproject.v1.myevent \
    --event-filters=someAttr=foo \
    --service-account=$PROJECT_NUMBER-compute@developer.gserviceaccount.com

Publish events

Once a channel exists, custom events can be published directly to that channel by the events producer using the Eventarc Publishing API. There are several ways that you can do this.

Publish using the gcloud CLI

You can publish custom events to a channel using the gcloud CLI.

gcloud eventarc channels publish CHANNEL_ID \
    --event-id=EVENT_ID \
    --event-type=CUSTOM_EVENT_TYPE \
    --event-attributes=CUSTOM_EVENT_ATTRIBUTE \
    --event-source=gcloud \
    --event-data="{ EVENT_DATA }"

Replace the following:

• EVENT_ID: the event ID
• CUSTOM_EVENT_TYPE: the event type
• CUSTOM_EVENT_ATTRIBUTE: event attributes
• CUSTOM_EVENT_DATA: any arbitrary, valid JSON data

Example:

gcloud eventarc channels publish helloworld-channel \
    --event-id=12345 \
    --event-type=mycompany.myorg.myproject.v1.myevent \
    --event-attributes=someAttr=foo \
    --event-source=gcloud \
    --event-data="{\"message\" : \"Hello world from gcloud\"}"

You should see an Event published successfully message.

Publish using curl

You can publish custom events to a channel using curl commands. The following example uses an access token generated by the gcloud CLI.

curl -H "Authorization: Bearer $(gcloud auth print-access-token)" \
  -H "Content-Type: application/json" \
  -H "X-Goog-User-Project: $PROJECT_ID" \
  -X POST \
  -d '{
       "events": [
             {
              "@type": "type.googleapis.com/io.cloudevents.v1.CloudEvent",
              "attributes": {
                "datacontenttype": {"ceString": "application/json"},
                "someattribute": {"ceString": "somevalue"},
                "time": {"ceTimestamp": "2022-03-19T21:29:13.899-04:00"}
              },
              "id": "12345",
              "source": "curl",
              "specVersion": "1.0",
              "textData": "{\"message\": \"Hello world from curl\"}",
              "type": "mycompany.myorg.myproject.v1.myevent"
            }
          ]
      }' \
  https://eventarcpublishing.googleapis.com/v1/projects/$(gcloud config get-value project)/locations/us-central1/channels/helloworld-channel:publishEvents

Publish using client libraries

For C# and Java, the CloudEvents SDK can be used to generate an event, convert it to a protobuf schema, and then publish it to the API.

During this preview, for other languages, you must define an event using a compiled CloudEvents proto file or by compiling your own based on the CloudEvents spec; you can then publish custom events to a channel using client libraries. (Note that the CloudEvents proto file must be included.) Protobuf schema is the preferred method of publishing events unless the schema is not supported.

1. Take a look at the samples.

   C#

   You can publish events in the protobuf or JSON schema.

   using Google.Cloud.Eventarc.Publishing.V1;
   using CloudNative.CloudEvents;
   using CloudNative.CloudEvents.Protobuf;
   using CloudNative.CloudEvents.SystemTextJson;
   using System.Net.Mime;
   using Newtonsoft.Json;
   using Google.Protobuf.WellKnownTypes;
   
   var commandArgs = Environment.GetCommandLineArgs();
   var ProjectId = commandArgs[1];
   var Region = commandArgs[2];
   var Channel = commandArgs[3];
   // Controls the format of events sent to Eventarc.
   // 'true' for using text format.
   // 'false' for proto (preferred) format.
   bool UseTextEvent = commandArgs.Length > 4 ? bool.TryParse(commandArgs[4], out UseTextEvent) : false;
   
   var FullChannelName = $"projects/{ProjectId}/locations/{Region}/channels/{Channel}";
   Console.WriteLine($"Channel: {FullChannelName}");
   Console.WriteLine($"UseTextEvent: {UseTextEvent}");
   
   var publisherClient = await PublisherClient.CreateAsync();
   
   //Construct the CloudEvent and set necessary attributes.
   var cloudEventAttributes = new[]
   {
       CloudEventAttribute.CreateExtension("someattribute", CloudEventAttributeType.String)
   };
   
   var cloudEvent = new CloudEvent(cloudEventAttributes)
   {
       Id = Guid.NewGuid().ToString(),
       // Note: Type has to match with the trigger!
       Type = "mycompany.myorg.myproject.v1.myevent",
       Source = new Uri("urn:csharp/client/library"),
       DataContentType = MediaTypeNames.Application.Json,
       Data = JsonConvert.SerializeObject(new { Message = "Hello World from C#"}),
       Time = DateTimeOffset.UtcNow,
       // Note: someattribute and somevalue have to match with the trigger!
       ["someattribute"] = "somevalue",
   };
   
   PublishEventsRequest request;
   
   if (UseTextEvent)
   {
       // Convert the CloudEvent to JSON
       var formatter = new JsonEventFormatter();
       var cloudEventJson = formatter.ConvertToJsonElement(cloudEvent).ToString();
       Console.WriteLine($"Sending CloudEvent: {cloudEventJson}");
       request = new PublishEventsRequest
       {
           Channel = FullChannelName,
           TextEvents = { cloudEventJson }
       };
   }
   else
   {
       // Convert the CloudEvent to Proto
       var formatter = new ProtobufEventFormatter();
       var cloudEventProto = formatter.ConvertToProto(cloudEvent);
       Console.WriteLine($"Sending CloudEvent: {cloudEventProto}");
       request = new PublishEventsRequest
       {
           Channel = FullChannelName,
           Events = { Any.Pack(cloudEventProto) }
       };
   }
   
   var response = await publisherClient.PublishEventsAsync(request);
   Console.WriteLine("Event published!");
   

   Java

   You can publish events in the protobuf or JSON schema.

   package com.google.cloud.eventarc.publishing.example;
   
   import com.fasterxml.jackson.databind.ObjectMapper;
   import com.google.cloud.eventarc.publishing.v1.PublishEventsRequest;
   import com.google.cloud.eventarc.publishing.v1.PublishEventsResponse;
   import com.google.cloud.eventarc.publishing.v1.PublisherClient;
   import com.google.protobuf.Any;
   
   import io.cloudevents.CloudEvent;
   import io.cloudevents.core.provider.EventFormatProvider;
   import io.cloudevents.core.v1.CloudEventBuilder;
   import io.cloudevents.jackson.JsonCloudEventData;
   import io.cloudevents.jackson.JsonFormat;
   import io.cloudevents.protobuf.ProtobufFormat;
   
   import java.net.URI;
   import java.time.OffsetDateTime;
   import java.util.UUID;
   import java.util.logging.Level;
   import java.util.logging.Logger;
   
   public class PublishEventsExample {
   
     static Logger LOGGER = Logger.getLogger(PublishEventsExample.class.getName());
   
     /**
      * CustomMessage represents a payload delivered as a content of the CloudEvent.
      */
     class CustomMessage {
       public CustomMessage(String message) {
         this.message = message;
       }
   
       public String message;
     }
   
     private PublishEventsRequest GetPublishEventsRequestWithTextFormat(String channelName, CloudEvent event) {
   
       byte[] serializedEvent = EventFormatProvider.getInstance()
           .resolveFormat(JsonFormat.CONTENT_TYPE)
           .serialize(event);
       String textEvent = new String(serializedEvent);
   
       PublishEventsRequest request = PublishEventsRequest.newBuilder()
           .setChannel(channelName)
           .addTextEvents(textEvent)
           .build();
   
       return request;
     }
   
     private PublishEventsRequest GetPublishEventsRequestWithProtoFormat(String channelName, CloudEvent event) throws Exception {
   
       byte[] serializedEvent = EventFormatProvider.getInstance()
           .resolveFormat(ProtobufFormat.PROTO_CONTENT_TYPE)
           .serialize(event);
   
       io.cloudevents.v1.proto.CloudEvent protoEvent = io.cloudevents.v1.proto.CloudEvent.parseFrom(serializedEvent);
       Any wrappedEvent = Any.pack(protoEvent);
   
       PublishEventsRequest request = PublishEventsRequest.newBuilder()
           .setChannel(channelName)
           .addEvents(wrappedEvent)
           .build();
   
       return request;
     }
   
     public void PublishEvent(String channelName, boolean useTextEvent) throws Exception {
   
       CustomMessage eventData = new CustomMessage("Hello world from Java");
   
       LOGGER.log(Level.INFO, "Building CloudEvent");
   
       ObjectMapper objectMapper = new ObjectMapper();
       CloudEvent event = new CloudEventBuilder()
           .withId(UUID.randomUUID().toString())
           .withSource(URI.create("//custom/from/java"))
           // Note: Type has to match with the trigger!
           .withType("mycompany.myorg.myproject.v1.myevent")
           .withTime(OffsetDateTime.now())
           // Note: someattribute and somevalue have to match with the trigger!
           .withExtension("someattribute", "somevalue")
           .withExtension("extsourcelang", "java")
           .withData("application/json",
               JsonCloudEventData.wrap(objectMapper.valueToTree(eventData)))
           .build();
   
   
       PublishEventsRequest request = useTextEvent ?
         GetPublishEventsRequestWithTextFormat(channelName, event) :
         GetPublishEventsRequestWithProtoFormat(channelName, event);
   
       LOGGER.log(Level.INFO, "Publishing message to Eventarc");
   
       try {
         PublisherClient client = PublisherClient.create();
         PublishEventsResponse response = client.publishEvents(request);
         LOGGER.log(Level.INFO, String.format("Message published successfully.\nReceived response: %s",
             response.toString()));
       } catch (Exception ex) {
         LOGGER.log(Level.SEVERE, "An exception occurred while publishing", ex);
       }
     }
   
     public static void main(String[] args) throws Exception {
       String projectId = args[0];
       String region = args[1];
       String channel = args[2];
       // Controls the format of events sent to Eventarc.
       // 'true' for using text format.
       // 'false' for proto (preferred) format.
       boolean useTextEvent = args.length > 3 ? Boolean.parseBoolean(args[3]) : false;
   
       String channelName = "projects/" + projectId + "/locations/" + region + "/channels/" + channel;
       System.out.println("Channel: " + channelName);
       System.out.println("useTextEvent: " + useTextEvent);
   
       new PublishEventsExample().PublishEvent(channelName, useTextEvent);
     }
   }

   Node.js

   You can publish events only in the JSON schema.

   const { CloudEvent } = require('cloudevents');
   const { PublisherClient } = require('@google-cloud/eventarc-publishing');
   
   /**
    * Builds a CloudEvent to publish to an Eventarc channel.
    *
    * @returns Fully constructed CloudEvent
    */
   const BuildCloudEvent = () => {
      return new CloudEvent({
        type: "mycompany.myorg.myproject.v1.myevent",
        source: "//event/from/nodejs",
        data: {
           message: "Hello World from Node.js"
        },
        datacontenttype: "application/json",
        time: new Date().toISOString(),
        // Note: someattribute and somevalue have to match with the trigger!
        someattribute: 'somevalue'
        });
    }
   
   /**
    * Publish event to the channel with the Eventarc publisher client.
    *
    * @param {string} channel
    * @param {CloudEvent} event
    */
   const publishEventToChannel = async (channel, event) => {
       // Instantiates a client with default credentials and options.
       const publishingClient = new PublisherClient();
   
       // Construct publish request.
       const request = {
           channel: channel,
           // Prepare text event rather than proto representation.
           // Since NodeJS CloudEvents SDK doesn't have method to transform
           // the object to protobuf it's easier to send the text representation.
           textEvents: [
               JSON.stringify(event)
           ]
       };
       console.log("Constructed the request with the event: ", request);
   
       // Publish event
       try {
           const response = await publishingClient.publishEvents(request);
           console.log("Received response: ", response);
       } catch (e) {
           console.error("Received error from publishing API: ", e);
       }
   }
   
   const arguments = process.argv.slice(2);
   const channel = arguments[0];
   if (!channel) {
       console.error("Missing channel, please pass it in the argument in the form of projects/$PROJECT_ID/locations/$REGION/channels/$CHANNEL_NAME")
       return;
   }
   
   const event = BuildCloudEvent();
   
   publishEventToChannel(channel, event);
   

   Python

   You can publish events only in the JSON schema.

   import sys
   import argparse
   import logging
   
   from cloudevents.http import CloudEvent
   from cloudevents.conversion import to_json
   from google.cloud.eventarc_publishing_v1.types import publisher
   from google.cloud.eventarc_publishing_v1.services.publisher import client
   
   parser = argparse.ArgumentParser(description="Arguments to publish an event with Eventarc Publisher API")
   parser.add_argument('--channel', type=str, help='Name of the channel on which the event should be published')
   parser.add_argument('--log', type=str, choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'], default='ERROR')
   
   logger = logging.getLogger()
   console_handler = logging.StreamHandler(sys.stdout)
   logger.addHandler(console_handler)
   
   def main():
     args = parser.parse_args()
     logger.setLevel(args.log)
   
     # Build a valid CloudEvent
     # The CloudEvent "id" and "time" are auto-generated if omitted and "specversion" defaults to "1.0".
     attributes = {
         "type": "mycompany.myorg.myproject.v1.myevent",
         "source": "//event/from/python",
         "datacontenttype": "application/json",
         # Note: someattribute and somevalue have to match with the trigger!
         "someattribute": "somevalue"
     }
     data = {"message": "Hello World from Python"}
     event = CloudEvent(attributes, data)
   
     logger.debug('Prepared CloudEvent:')
     logger.debug(event)
   
     request = publisher.PublishEventsRequest()
     request.channel = args.channel
     request.text_events.append(to_json(event))
   
     logger.debug('Prepared publishing request:')
     logger.debug(request);
   
     publisher_client = client.PublisherClient()
     response = publisher_client.publish_events(request)
     logger.info('Event published')
     logger.debug(f'Result: {response}')
   
   
   if __name__ == '__main__':
     main()

2. Invoke the samples using the JSON schema.

   C#

   PROJECT_ID=$(gcloud config get-value project)
   REGION=us-central1
   CHANNEL_ID=helloworld-channel
   
   dotnet run $PROJECT_ID $REGION $CHANNEL_ID true
   

   Java

   PROJECT_ID=$(gcloud config get-value project)
   REGION=us-central1
   CHANNEL_ID=helloworld-channel
   
   ./gradlew run --args="$PROJECT_ID $REGION $CHANNEL_NAME true"
   

   Node.js

   PROJECT_ID=$(gcloud config get-value project)
   REGION=us-central1
   CHANNEL_ID=helloworld-channel
   
   npm run invoke projects/$PROJECT_ID/locations/$REGION/channels/$CHANNEL_NAME
   

   Python

   PROJECT_ID=$(gcloud config get-value project)
   REGION=us-central1
   CHANNEL_ID=helloworld-channel
   
   python3 publish.py \
       --channel projects/$PROJECT_ID/locations/$REGION/channels/$CHANNEL_NAME \
       --log=DEBUG
   

View the custom event in the logs

Publishing a custom event to the Eventarc channel triggers the Cloud Run service. You can view the event-related log entries created by your Cloud Run service:

gcloud logging read "resource.type=cloud_run_revision AND resource.labels.service_name=SERVICE_NAME" --limit 5

Replace SERVICE_NAME with the name of your Cloud Run service.

The log entry should be similar to the following (Hello world from gcloud, Hello world from curl, Hello world from Python, and so on):

jsonPayload:
  event:
    data:
      message: Hello world from gcloud
    datacontenttype: application/json
    id: '12345'
    someattr: foo
    source: gcloud
    specversion: '1.0'
    time: '2022-04-27T14:04:28.586Z'
    type: mycompany.myorg.myproject.v1.myevent
  eventType: mycompany.myorg.myproject.v1.myevent
  message: 'Received event of type mycompany.myorg.myproject.v1.myevent. Event data:
    {"message" : "Hello world from gcloud"}'

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