Writing an event source using Javascript

This tutorial provides instructions to build an event source in Javascript and implement it with a ContainerSource or SinkBinding.

  • Using a ContainerSource is a simple way to turn any dispatcher container into a Knative event source.
  • Using SinkBinding provides a framework for injecting environment variables into any Kubernetes resource that has a spec.template and is PodSpecable.

ContainerSource and SinkBinding both work by injecting environment variables to an application. Injected environment variables at minimum contain the URL of a sink that will receive events.

Bootstrapping

Create the project and add the dependencies:

  1. npm init
  2. npm install cloudevents-sdk@2.0.1 --save

NOTE: Due to this bug, you must use version 2.0.1 of the Javascript SDK or newer.

Using ContainerSource

A ContainerSource creates a container for your event source image and manages this container.

The sink URL to post the events will be made available to the application through the K_SINK environment variable by the ContainerSource.

Example

The following example event source emits an event to the sink every 1000 milliseconds:

  1. // File - index.js
  3. const { CloudEvent, HTTPEmitter } = require("cloudevents-sdk");
  5. let sinkUrl = process.env['K_SINK'];
  7. console.log("Sink URL is " + sinkUrl);
  9. let emitter = new HTTPEmitter({
  10.     url: sinkUrl
  11. });
  13. let eventIndex = 0;
  14. setInterval(function () {
  15.     console.log("Emitting event #" + ++eventIndex);
  17.     let myevent = new CloudEvent({
  18.         source: "urn:event:from:my-api/resource/123",
  19.         type: "your.event.source.type",
  20.         id: "your-event-id",
  21.         dataContentType: "application/json",
  22.         data: {"hello": "World " + eventIndex},
  23.     });
  25.     // Emit the event
  26.     emitter.send(myevent)
  27.         .then(response => {
  28.             // Treat the response
  29.             console.log("Event posted successfully");
  30.             console.log(response.data);
  31.         })
  32.         .catch(err => {
  33.             // Deal with errors
  34.             console.log("Error during event post");
  35.             console.error(err);
  36.         });
  37. }, 1000);
  1. # File - Dockerfile
  3. FROM node:10
  4. WORKDIR /usr/src/app
  5. COPY package*.json ./
  6. RUN npm install
  7. COPY . .
  8. EXPOSE 8080
  9. CMD [ "node", "index.js" ]

The example code uses Binary mode for CloudEvents. To employ structured code, change let binding = new v1.BinaryHTTPEmitter(config); to let binding = new v1.StructuredHTTPEmitter(config);.

Binary mode is used in most cases because:

  • It is faster in terms of serialization and deserialization.
  • It works better with CloudEvent-aware proxies, such as Knative Channels, and can simply check the header instead of parsing the payload.

Procedure

  1. Build and push the image:

    1. docker build . -t path/to/image/registry/node-knative-heartbeat-source:v1
    2. docker push path/to/image/registry/node-knative-heartbeat-source:v1

  2. Create the event display service which logs any CloudEvents posted to it:

    1. apiVersion: serving.knative.dev/v1
    2. kind: Service
    3. metadata:
    4.   name: event-display
    5. spec:
    6.   template:
    7.     spec:
    8.       containers:
    9.         - image: docker.io/aliok/event_display-864884f202126ec3150c5fcef437d90c@sha256:93cb4dcda8fee80a1f68662ae6bf20301471b046ede628f3c3f94f39752fbe08

  3. Create the ContainerSource object:

    1. apiVersion: sources.knative.dev/v1
    2. kind: ContainerSource
    3. metadata:
    4.   name: test-heartbeats
    5. spec:
    6.   template:
    7.     spec:
    8.       containers:
    9.         - image: path/to/image/registry/node-knative-heartbeat-source:v1
    10.           name: heartbeats
    11.   sink:
    12.     ref:
    13.       apiVersion: serving.knative.dev/v1
    14.       kind: Service
    15.       name: event-display

  4. Check the logs of the event display service. You will see a new message is pushed every second:

    1. $ kubectl logs -l serving.knative.dev/service=event-display -c user-container
    3. ☁️  cloudevents.Event
    4. Validation: valid
    5. Context Attributes,
    6.   specversion: 1.0
    7.   type: your.event.source.type
    8.   source: urn:event:from:your-api/resource/123
    9.   id: your-event-id
    10.   datacontenttype: application/json
    11. Data,
    12.   {
    13.     "hello": "World 1"
    14.   }

  5. Optional: If you are interested in seeing what is injected into the event source as a K_SINK, you can check the logs:

    1. $ kubectl logs test-heartbeats-deployment-7575c888c7-85w5t
    3. Sink URL is http://event-display.default.svc.cluster.local
    4. Emitting event #1
    5. Emitting event #2
    6. Event posted successfully
    7. Event posted successfully

Using SinkBinding

SinkBinding does not create any containers. It injects the sink information to an already existing Kubernetes resources. This is a flexible approach as you can use any Kubernetes PodSpecable object as an event source, such as Deployment, Job, or Knative services.

Procedure

  1. Create an event display service:

    1. apiVersion: serving.knative.dev/v1
    2. kind: Service
    3. metadata:
    4.   name: event-display
    5. spec:
    6.   template:
    7.     spec:
    8.       containers:
    9.         - image: docker.io/aliok/event_display-864884f202126ec3150c5fcef437d90c@sha256:93cb4dcda8fee80a1f68662ae6bf20301471b046ede628f3c3f94f39752fbe08

  2. Create a Kubernetes deployment that runs the event source:

    1. apiVersion: apps/v1
    2. kind: Deployment
    3. metadata:
    4.   name: node-heartbeats-deployment
    5.   labels:
    6.     app: node-heartbeats
    7. spec:
    8.   replicas: 2
    9.   selector:
    10.     matchLabels:
    11.       app: node-heartbeats
    12.   template:
    13.     metadata:
    14.       labels:
    15.         app: node-heartbeats
    16.     spec:
    17.       containers:
    18.       - name: node-heartbeats
    19.         image: path/to/image/registry/node-knative-heartbeat-source:v1
    20.         ports:
    21.         - containerPort: 8080

  3. Because the SinkBinding has not yet been created, you will see an error message, because the K_SINK environment variable is not yet injected:

    1. $ kubectl logs node-heartbeats-deployment-9ffbb644b-llkzk
    3. Sink URL is undefined
    4. Emitting event #1
    5. Error during event post
    6. TypeError [ERR_INVALID_ARG_TYPE]: The "url" argument must be of type string. Received type undefined

  4. Create the SinkBinding object:

    1. apiVersion: sources.knative.dev/v1
    2. kind: SinkBinding
    3. metadata:
    4.   name: bind-node-heartbeat
    5. spec:
    6.   subject:
    7.     apiVersion: apps/v1
    8.     kind: Deployment
    9.     selector:
    10.       matchLabels:
    11.         app: node-heartbeats
    12.   sink:
    13.     ref:
    14.       apiVersion: serving.knative.dev/v1
    15.       kind: Service
    16.       name: event-display

    You will see the pods are recreated and this time the K_SINK environment variable is injected.

    Also note that since the replicas is set to 2, there will be 2 pods that are posting events to the sink.

    1. $ kubectl logs event-display-dpplv-deployment-67c9949cf9-bvjvk -c user-container
    3. ☁️  cloudevents.Event
    4. Validation: valid
    5. Context Attributes,
    6.   specversion: 1.0
    7.   type: your.event.source.type
    8.   source: urn:event:from:your-api/resource/123
    9.   id: your-event-id
    10.   datacontenttype: application/json
    11. Data,
    12.   {
    13.     "hello": "World 1"
    14.   }
    16. ☁️  cloudevents.Event
    17. Validation: valid
    18. Context Attributes,
    19.   specversion: 1.0
    20.   type: your.event.source.type
    21.   source: urn:event:from:your-api/resource/123
    22.   id: your-event-id
    23.   datacontenttype: application/json
    24. Data,
    25.   {
    26.     "hello": "World 1"
    27.   }