eagle-topology-dsl.sql

# Stream Config
CONFIG {
    "config.resource" -> "application.conf"
    "default.flow.engine" -> "siddhi"
}
 
FLOW {
    # Stream Definition
    define stream metricStream_1 (metricType: string,site: string,host: string,component: string,value: double);
    define stream metricStream_2 (metricType: string,site: string,host: string,component: string,value: double);
  
    # Stream Aggregation (native siddhi syntax)
 
 
    from metricStream_1[component=='dn' and metricType==''RpcActivityForPort50020.RpcQueueTimeNumOps"].time[3600]
    select sum(value) group by host output every 1 hour
    insert into aggregatedMetricStream_1;
     
    from metricStream_2[component=='dn' and metricType==''RpcActivityForPort50020.RpcQueueTimeNumOps"].time[3600]
    select sum(value) group by host output every 1 hour
    insert into aggregatedMetricStream_2;
 
    # Stream Alert
    from metricStream_1[component=='dn' and metricType==''RpcActivityForPort50020.RpcQueueTimeNumOps"].time[3600]
    select sum(value) group by host output every 1 hour
    insert into globalAlertStream;
 
 
    from metricStream_2[component=='dn' and metricType==''RpcActivityForPort50020.RpcQueueTimeNumOps"].time[3600]
 
    select sum(value) group by host output every 1 hour
    insert into globalAlertStream;
}
 
# Stream Source 
SOURCE {
    metricStream_1 <- Kafka("source_1","localhost:2181");
    metricStream_1 <- Kafka("source_1","localhost:2181");
}
# Stream Sink
SINK {
    aggregatedMetricStream_1 -> druid(datasource="",broker=[""]);
    aggregatedMetricStream_2 -> kafka (topic="stream_2",broker=[""]);
    globalAlertStream -> eagle (service="AlertService",host="localhost",port="9099");
    globalAlertStream -> notification (protocol="SMTP",server="localhost:25");
    globalAlertStream -> kafka (topic="alert_topic");
    globalAlertStream -> org.apache.eagle.OtherCustomizedAlertReceiver(topic="alert_topic");
 
    globalAlertStream -> [
        eagle (service="AlertService",host="localhost",port="9099"),
        notification (protocol="SMTP",server="localhost:25"),
        kafka (topic="alert_topic"),
        org.apache.eagle.OtherCustomizedAlertReceiver(topic="alert_topic")
    ]
}