/* Enter a unique ExecutionPlan */
@Plan:name('APIMAnalytics-AbnormalResponseAndBackendTimeDetection-AbnormalResponseAndBackendTimeDetection-realtime1')

/* Enter a unique description for ExecutionPlan */
-- @Plan:description('ExecutionPlan')

/* define streams/tables and write queries here ... */

@Import('org.wso2.apimgt.statistics.perMinuteResponse:1.0.0')
define stream responseSummaryStream (meta_clientType string, year int, month int, day int, hour int, minute int, context string, api_version string, api string, resourceTemplate string, version string, tenantDomain string, hostName string, apiPublisher string, destination string, consumerKey string, resourcePath string, method string, response int, responseTime long, serviceTime long, backendTime long, username string, eventTime long, applicationName string, applicationId string, cacheHit bool, responseSize long, protocol string, responseCode int, total_response_count long, _timestamp long);

@Export('org.wso2.analytics.apim.abnormalBackendTimeAlertStream:1.0.0')
define stream abnormalBackendTimeAlertStream (api string, apiPublisher string, tenantDomain string, resourceTemplate string, method string, backendTime long, backendPercentile double, msg string, severity int, alertTimestamp long);

@Export('org.wso2.analytics.apim.abnormalResponseTimeAlertStream:1.0.0')
define stream abnormalResponseTimeAlertStream (api string, apiPublisher string, tenantDomain string, resourceTemplate string, method string, responseTime long, responsePercentile double, msg string, severity int, alertTimestamp long);

@Export('org.wso2.analytics.apim.allApimAlertsStream:1.0.0')
define stream allApimAlertStream (type string, tenantDomain string, msg string, severity int, alertTimestamp long);

@from(eventtable = 'analytics.table' , table.name = 'ORG_WSO2_ANALYTICS_APIM_RESPONSEPERCENTILE',  primary.keys = 'api_version,tenantDomain,resourceTemplate,method', caching = 'true', cache.timeout.seconds='300', wait.for.indexing = 'false')
define table percentileTable (api_version string, tenantDomain string, resourceTemplate string, method string, responsePercentile double, backendPercentile double );

/*spark script calculates percentile of responseTime and backendTime  and stores in percentileTable */

@info(name = 'query1')
from responseSummaryStream join percentileTable
on (responseSummaryStream.api_version==percentileTable.api_version and responseSummaryStream.tenantDomain==percentileTable.tenantDomain and responseSummaryStream.resourceTemplate==percentileTable.resourceTemplate and responseSummaryStream.method==percentileTable.method)
select responseSummaryStream.api as api_name, responseSummaryStream.api_version, ifThenElse(responseSummaryStream.tenantDomain == 'carbon.super', str:concat(responseSummaryStream.apiPublisher, "@carbon.super"), responseSummaryStream.apiPublisher) as apiPublisher, responseSummaryStream.tenantDomain, responseSummaryStream.resourceTemplate, responseSummaryStream.method, responseSummaryStream.responseTime, percentileTable.responsePercentile , responseSummaryStream.backendTime,  percentileTable.backendPercentile
insert into responseInfoStream;

@info(name = 'query2')
from responseInfoStream[responseTime > responsePercentile]
select api_version as api, apiPublisher, tenantDomain, resourceTemplate, method, responseTime, responsePercentile , 'Abnormal response time detected.' as msg, 2 as severity, (time:timestampInMilliseconds()) as alertTimestamp
insert into abnormalResponseTimeAlertStreamTemp;

@info(name = 'suppression1')
from abnormalResponseTimeAlertStreamTemp#window.length(1) as a left outer join abnormalResponseTimeAlertStreamTemp#window.time(10 minute) as b
on (a.api== b.api and a.apiPublisher== b.apiPublisher and a.tenantDomain==b.tenantDomain and a.resourceTemplate==b.resourceTemplate and a.method== b.method and
a.msg== b.msg)
select a.api, a.apiPublisher, a.tenantDomain, a.resourceTemplate, a.method, a.responseTime, a.responsePercentile, a.msg as msg,
a.severity, a.alertTimestamp
having b.api  is null
insert into abnormalResponseTimeAlertStream;

@info(name = 'query3')
from responseInfoStream[backendTime > backendPercentile]
select api_version as api, apiPublisher, tenantDomain, resourceTemplate, method, backendTime, backendPercentile, 'Abnormal backend response time detected.' as msg, 3 as severity, (time:timestampInMilliseconds()) as alertTimestamp
insert into abnormalBackendTimeAlertStreamTemp;

@info(name = 'suppression2')
from abnormalBackendTimeAlertStreamTemp#window.length(1) as a left outer join abnormalBackendTimeAlertStreamTemp#window.time(10 minute) as b
on (a.api== b.api and a.apiPublisher== b.apiPublisher and a.tenantDomain==b.tenantDomain and a.resourceTemplate==b.resourceTemplate and a.method== b.method and
a.msg== b.msg)
select a.api, a.apiPublisher, a.tenantDomain, a.resourceTemplate, a.method, a.backendTime, a.backendPercentile, a.msg as msg, a.severity, a.alertTimestamp
having b.api  is null
insert into abnormalBackendTimeAlertStream;

@info(name = 'query4')
from abnormalResponseTimeAlertStream
select 'abnormalResponseTime' as type, tenantDomain, str:concat('Abnormal response time detected for http ',method,' method of resource template:',resourceTemplate,' in api:', api,' of tenant domain:',tenantDomain,', threshold value:',responsePercentile, 'ms.') as msg, severity, (time:timestampInMilliseconds()) as alertTimestamp
insert into allApimAlertStream;

@info(name = 'query5')
from abnormalBackendTimeAlertStream
select 'abnormalBackendTime' as type, tenantDomain, str:concat('Abnormal backend response time detected for http ',method,' method of resource template:',resourceTemplate,' in api:', api,' of tenant domain:',tenantDomain,', threshold value:',backendPercentile, 'ms.') as msg , severity, (time:timestampInMilliseconds()) as alertTimestamp
insert into allApimAlertStream;