Introduction 

There are a few use cases that we want to support with this feature. The first is to support a use case where the pipeline is a geofencing use case, where the pipeline is processing user locations, and wants to send notifications alerts when a user enters or leaves a geo fence. Similar use cases include pipelines that are reading user biometric data that need to send notifications if alerts if user heart rate exceeds a certain value, or pipelines that read machine cpu usage that need to send notification if alerts if cpu usage exceeds a certain value. In these use cases, some uncommon event occurs that must be acted on by a separate system. The events that trigger these notifications still alerts still need to be processed by the rest of the pipeline, there is just some side effect that is triggered once they are observed.

Goals

The goal is to allow stages in a pipeline to emit notificationsalerts, which can be configured to be published to TMS or to Kafka.

User Stories 

• As a pipeline developer, I want to be able to create a pipeline where notifications are alerts are published to Kafka or TMS when some condition is met
• As a pipeline developer, I want to be able to create some pipelines that publish notifications and alerts and some that do not, even when the conditions are met
• As a pipeline developer, I want to be able to configure which topic notifications are alerts are published to
• As a pipeline developer, I want to be able to tell which plugins can emit notifications and alerts and which plugins cannot
• As a cluster administrator, I want to be able to see how many notifications were alerts were published for a pipeline run
• As a plugin developer, I want to be able to write arbitrary logic to control when to publish notificationspublish alerts
• As a plugin developer, I want to be able to indicate which plugins can emit notificationsemit alerts

Design

At a high level, we would like each existing plugin type (except sinks) to be able to emit notificationsalerts. A notification An alert is not an arbitrary record, but must conform to a specific schema. Each plugin will indicate whether it can emit notifications or not, which can be reflected in the UI by an additional 'port'. When a pipeline stage is connected to a new 'NotificationPublisherAlertPublisher' plugin type, any notifications emitted by that stage will be sent to the NotificationPublisher for AlertPublisher for actual publishing.

When the actual notifications alerts are published is left up to the pipeline. If a NotificationPublisher plugin AlertPublisher plugin is not attached to a stage, any notifications alerts emitted by the stage will be dropped.

Approach

Pipeline Config

There will be no changes to the structure of the pipeline config. If a NotificationPublisher plugin AlertPublisher plugin type is connected to a stage, any notifications alerts emitted by the stage will be sent to the publisher. The pipeline depicted in the image above would look something like:

{
  "stages": [
    { "name": "source", "plugin": { "type": "batchsource", ... } },
    { "name": "transform", "plugin": { "type": "transform", ... } },
    { "name": "sink", "plugin": { "type": "batchsink", ... } },
    { "name": "notificationemitter", "plugin": { "type": "notificationpublisheralertpublisher", ... } }
  ],
  "connections": [
    { "from": "source", "to": "transform" },
    { "from": "transform", "to": "sink" },
    { "from": "transform", "to": "notificationemitter" }
  ]
}

Emitting Notifications

The Emitter interface will be enhanced to allow emitting a notificationan alert:

public interface Emitter<T> {

  // existing method
  void emit(T value);

  // existing method
  void emitError(InvalidEntry<T> invalidEntry);
 
  void getNotificationEmittergetAlertEmitter(String topic);
 
  void getNotificationEmittergetAlertEmitter(String namespace, String topic);
}
 
public interface NotificationEmitterAlertEmitter {

  void emit(Charset charset, String...payloads);

  void emit(Charset charset, Iterator<String> payloads);

  void emit(byte[]...payloads);

  void emit(Iterator<byte[]> payloads);
}

Publishing

Alerts

Notifications Alerts can be published using a new NotificationPublisher AlertPublisher plugin type:

public abstract class NotificationPublisherAlertPublisher extends PipelineConfigurable, implements StageLifecycle<MessagingContext> {
  private MessagingContext context;
  
  @Override
  public void initialize(MessagingContext context) throws Exception {
    this.context = context;
  }

  @Override
  public void publish(Iterator<Notification>Iterator<Alert> notificationalerts) throws Exception {
  }

  @Override
  public void destroy() {
    // no-op
  }
}
 
public class NotificationAlert {
  String namespace;
  String topic;
  byte[] payloads;
}

The publish method receives an Iterator of notifications Alerts in case it wants to do some dedup or aggregation logic before actually publishing anything.

Plugin

Alert Port Indication

Though every plugin will be able to emit notifications through the programmatic API, most plugins will not make use of this functionality. Users should be given some indication of which plugins make use of it and which do not. One way to do this is to annotate the plugin:

@Plugin(type = Transform.PLUGIN_TYPE)
@Name("NotifierAlert")
@NotificationsEnabled@AlertsEnabled
public class NotifierTransformAlertTransform extends Transform {
 
}

The NotificationsEnabled will be an annotation only in the cdap-etl-api, as it is specific to pipelines. It is not a CDAP annotation. In order for the UI to see it, it must be somehow exposed through the RESTful API in a generic way. One way to do this is to add an 'annotations' field to the PluginClass object:

GET /v3/namespaces/<namespace-id>/artifacts/<artifact-name>/versions/<artifact-version>/extensions/<plugin-type>/plugins/<plugin-name>
[
  {
    "name": "JavaScript",
    "type": "transform",
    "annotations": [
      "NotificationsEnabledAlertsEnabled"
    ],
    "description": "Executes user-provided JavaScript that transforms one record into zero or more records.",
    "className": "co.cask.hydrator.plugin.transform.JavaScriptTransform",
    "artifact": { ... }, 
    "properties": { ... }
  }
  ...
]

This would still make CDAP generic, as it knows nothing about what the custom annotation means, but allows the UI and data pipeline app to do special logic with the annotation.

Implementation

Notifications Alerts will be collected in the Spark Driver in memory (through the .collect() method), before being sent to a NotificationPublisheran AlertPublisher. This is because we are not able to publish within a Spark closure, and in case the publisher wants to do some dedup logic or aggregation logic. In MapReduce, tasks are also unable to publish to TMS. The program will need to write all notifications alerts to a local FileSet, then read from the FileSet and publish everything at the end.

API changes

New Programmatic APIs

Emitter AlertEmitter and NotificationPublisher AlertPublisher are new APIs.

Deprecated Programmatic APIs

Modified REST APIs

Deprecated REST API

None

CLI Impact or Changes

• None

UI Impact or Changes

• UI must be able to detect which plugins can emit notifications, and display a corresponding port
• UI must display metrics for notifications emitted

Security Impact 

This feature will use TMS, so any authorization added to TMS will affect this feature.

Impact on Infrastructure Outages 

None

Test Scenarios

Releases

Release X.Y.Z

Release X.Y.Z

Related Work

• Work #1
• Work #2
• Work #3

Future work