Notification Plugin should provide interface which accepts the Eagle Alert Entity and return Status of the same.
Today in Eagle there is no Notification Plugin Framework where users can implement the Notification Plugin Interface and upload their own library. This Plugin allows user to implement their own and also provides below default implementation
- Sending Email Alert
- Persisting Message to Kafka
- Writing Entities to Eagle Service
Notification Interface:
At the time of Eagle Topology starts , Code should scan and register the Notification Type ( Custom Notification / Email / Kafka Message etc.. ). Since Eagle allows users to implement their own Notification Implementation , Our Topology Init code should automatically detects and register which type of Notification needs to be used .
public class SECAlerts implements NotificationPlugin {
}
When users deploys their code with Eagle Service , Our Topology Initializer have to detect SECAlerts and Register it automatically.
How to select Notification When defining Policy ?
Eagle should allow users to select the Notification Type at the time of Policy Creation.. For that we need to persist the detected policies in HBase and Provide API on Top of It to Query.
To ensure the consistency always delete the Notification Table and Recreate with detected Notifications.
Persisting Message to Kafka:
package org.apache.eagle.notification; import org.apache.eagle.alert.entity.AlertAPIEntity; import org.apache.eagle.common.config.EagleConfig; import org.apache.eagle.common.config.EagleConfigFactory; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; @SuppressWarnings({ "rawtypes", "unchecked" }) public class PersistAlertToKafkaTopic implements NotificationPlugin { private boolean isNotified; private static EagleConfig config = EagleConfigFactory.load(); @Override public void onAlert(AlertAPIEntity alertEntity) { processAlertEntity(alertEntity); isNotified = true; } public void processAlertEntity( AlertAPIEntity alertEntity ) { KafkaProducer producer = KafkaProducerSingleton.INSTANCE.getProducer(); producer.send(createRecord(alertEntity)); } public ProducerRecord createRecord(AlertAPIEntity entity ){ ProducerRecord record = new ProducerRecord(config.getConfig().getString(NotificationConstants.NOTIFICATION_ALERT_KAFKA_TOPIC), entity); return record; } @Override public NotificationStatus getStatus() { NotificationStatus status = new NotificationStatus(); status.setNotificationSuccess(isNotified); return status; } }