/* Enter a unique ExecutionPlan */

@Plan:name('APIMAnalytics-HealthAvailabilityPerMinAlert-HealthAvailabilityPerMin-realtime1')

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

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


@Plan:trace('false')

@Plan:statistics('false')

@Import('org.wso2.apimgt.statistics.perMinuteRequest:1.0.0')
define stream requestPerMinStream (meta_clientType string, year int, month int, day int, hour int, minute int, consumerKey string, context string,
api_version string, api string, version string, requestTime long, userId string, hostName string, apiPublisher string,
total_request_count long, resourceTemplate string, method string, applicationName string, tenantDomain string,
userAgent string, resourcePath string, request int, applicationId string, tier string, throttledOut bool, clientIp string,
applicationOwner string, _timestamp long);

@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);

@Import('org.wso2.apimgt.statistics.response:1.1.0')
define stream responseStream (meta_clientType string, consumerKey string, context string, api_version string, api string, resourcePath string,
resourceTemplate string, method string, version string, response int, responseTime long, serviceTime long, backendTime long,
username string, eventTime long, tenantDomain string, hostName string, apiPublisher string, applicationName string,
applicationId string, cacheHit bool, responseSize long, protocol string, responseCode int, destination string);


/* First define response/request count streams */

/*@Export('org.wso2.analytics.apim.healthAvailabilityAlertStream:1.0.0')
define stream minStatusTestStream (api_version string, status bool, reason string);*/

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

@Export('org.wso2.analytics.apim.apiHealthMonitorAlertStream:1.0.0')
define stream apiHealthMonitorAlertStream (api_version string, apiPublisher string, tenantDomain string, msg string, severity int, alertTimestamp long);

/*  -- Event tables -- */

/* Lower percentiles */
@from(eventtable = 'analytics.table' , table.name = 'ORG_WSO2_ANALYTICS_APIM_REQUESTPERAPIPERCENTILE',  primary.keys = 'api_version, tenantDomain', caching = 'true', cache.timeout.seconds='300', wait.for.indexing = 'false')
define table requestPerApiPercentileTable (requestsPerMinPerApiLowerPercentile double, api_version string, tenantDomain string);

@from(eventtable = 'analytics.table' , table.name = 'ORG_WSO2_ANALYTICS_APIM_RESPONSEPERAPIPERCENTILE',  primary.keys = 'api_version, tenantDomain', caching = 'true', cache.timeout.seconds='300', wait.for.indexing = 'false')
define table responsePerApiPercentileTable (api_version string, tenantDomain string, responsesPerMinPerApiLowerPercentile double);

/* Then define response time streams */
@from(eventtable = 'analytics.table' , table.name = 'ORG_WSO2_ANALYTICS_APIM_RESPONSETIMEPERAPIPERCENTILE',  primary.keys = 'api_version, tenantDomain', caching = 'true', cache.timeout.seconds='300', wait.for.indexing = 'false')
define table responseTimePerApiPercentileTable (api_version string, tenantDomain string, responseTimePerApiUpperPercentile double);

/* API status availability table */
@from(eventtable = 'analytics.table' , table.name = 'ORG_WSO2_ANALYTICS_APIM_APIAVAILABILITY',  primary.keys = 'api_version, tenantDomain', caching='true', cache.timeout.seconds='300', wait.for.indexing = 'false')
define table apiAvailabilityTable (api_version string, tenantDomain string, status string);

/* APIHEALTHMONITORALERTSTREAM TABLE */
@from(eventtable = 'analytics.table' , table.name = 'ORG_WSO2_ANALYTICS_APIM_APIHEALTHMONITORALERTSTREAM', caching='true', cache.timeout.seconds='300', wait.for.indexing = 'false')
define table apiHealthMonitorAlertStreamTable (api_version string, apiPublisher string, tenantDomain string, msg string, alertTimestamp long);

@from(eventtable = 'analytics.table', table.name = 'ORG_WSO2_ANALYTICS_APIM_APIHEALTHMONITORCURRENTALERTS', primary.keys = 'api_version, tenantDomain', caching='true', cache.timeout.seconds='300', wait.for.indexing = 'false')
define table apiHealthMonitorCurrentAlertTable (api_version string, apiPublisher string, tenantDomain string, msg string, alertTimestamp long);

@from(eventtable = 'analytics.table' , table.name = 'ORG_WSO2_ANALYTICS_APIM_APIHEALTHALERTSUMMARYTABLE', primary.keys='api_version, apiPublisher, tenantDomain, msg' , indices = 'api_version, apiPublisher, tenantDomain, msg,lastAlertedTimestamp', wait.for.indexing = 'false')
define table ApiHealthAlertSummaryTable(api_version string, apiPublisher string, tenantDomain string, msg string, lastAlertedTimestamp long);

@IndexBy('tenantApiVersion')
define table alertTableResponseCode (tenantApiVersion string, count int);

@IndexBy('tenantApiVersion')
define table alertTableResponseInfo (tenantApiVersion string, count int);

@IndexBy('tenantApiVersion')
define table alertTableRequestResponse (tenantApiVersion string, count int);

/* timeBatch used because Minutely updates are needed */
@info(name = 'query1')
from requestPerMinStream
select  api_version, apiPublisher, tenantDomain, total_request_count as  requestsPerMinPerApi
group by api_version, tenantDomain
insert into requestPerApiCountStream;

@info(name = 'query2')
from responseSummaryStream
select  api_version, tenantDomain, total_response_count as  responsesPerMinPerApi
group by api_version, tenantDomain
insert into responsePerApiCountStream;

/* Creating requests stream with percentiles*/
@info(name = 'query3')
from requestPerApiCountStream join requestPerApiPercentileTable on (requestPerApiCountStream.api_version == requestPerApiPercentileTable.api_version AND requestPerApiCountStream.tenantDomain == requestPerApiPercentileTable.tenantDomain)
select requestPerApiCountStream.api_version as api_version, requestPerApiCountStream.tenantDomain as tenantDomain, requestPerApiCountStream.apiPublisher as apiPublisher, requestsPerMinPerApi as requestsPerMin, requestsPerMinPerApiLowerPercentile as requestsPerMinLowerPercentile
insert into requestsPercentileStream;

/* Creating responses stream with percentiles  */
@info(name = 'query4')
from responsePerApiCountStream join responsePerApiPercentileTable on (responsePerApiCountStream.api_version == responsePerApiPercentileTable.api_version AND responsePerApiCountStream.tenantDomain == responsePerApiPercentileTable.tenantDomain)
select responsePerApiCountStream.api_version as api_version, responsePerApiCountStream.tenantDomain as tenantDomain,responsesPerMinPerApi as responsesPerMin, responsesPerMinPerApiLowerPercentile as responsesPerMinLowerPercentile
insert into responsesPercentileStream;

/* Combine*/
@info(name = 'query5')
from every e1=requestsPercentileStream, e2=responsesPercentileStream[e1.api_version == e2.api_version AND e1.tenantDomain == e2.tenantDomain]
within 1 min
select e1.api_version as api_version, e1.apiPublisher as apiPublisher, e1.requestsPerMin as requestsPerMin,e1.requestsPerMinLowerPercentile as requestsPerMinLowerPercentile,
ifThenElse(e2.responsesPerMin is null, convert(0, 'long'), e2.responsesPerMin) as responsesPerMin, e2.tenantDomain ,ifThenElse(e2.responsesPerMinLowerPercentile is null, 1.0, e2.responsesPerMinLowerPercentile) as responsesPerMinLowerPercentile
insert into requestResponseSummaryStream;

----------------------------------------------------------------------------------------------------------------------------------------------------------
@info(name = 'requestResponse1')
from requestResponseSummaryStream
select str:concat(tenantDomain, "_", api_version) as tenantApiVersion, api_version, tenantDomain, apiPublisher, requestsPerMin, requestsPerMinLowerPercentile, responsesPerMin, responsesPerMinLowerPercentile
insert into tmpRequestResponseSummaryStream;

@info(name = 'requestResponse2')
from tmpRequestResponseSummaryStream[(requestsPerMin > requestsPerMinLowerPercentile AND responsesPerMin < responsesPerMinLowerPercentile) and not (alertTableRequestResponse.tenantApiVersion == tenantApiVersion in alertTableRequestResponse)]
select tenantApiVersion, 1 as count
insert into alertTableRequestResponse;

@info(name = 'requestResponse3')
from tmpRequestResponseSummaryStream[not(requestsPerMin > requestsPerMinLowerPercentile AND responsesPerMin < responsesPerMinLowerPercentile)]
delete alertTableRequestResponse
    on alertTableRequestResponse.tenantApiVersion == tenantApiVersion;

@info(name = 'requestResponse4')
from tmpRequestResponseSummaryStream[(requestsPerMin > requestsPerMinLowerPercentile AND responsesPerMin < responsesPerMinLowerPercentile) and alertTableRequestResponse.tenantApiVersion == tenantApiVersion in alertTableRequestResponse]
insert into updateRequestReponseStream;

@info(name = 'requestResponse5')
from updateRequestReponseStream#window.length(1) join alertTableRequestResponse
	on alertTableRequestResponse.tenantApiVersion == updateRequestReponseStream.tenantApiVersion
select alertTableRequestResponse.tenantApiVersion, (alertTableRequestResponse.count + 1) as count
insert into alertTableRequestResponse;

@info(name = 'requestResponse6')
from tmpRequestResponseSummaryStream#window.length(1) join alertTableRequestResponse
	on alertTableRequestResponse.tenantApiVersion == tmpRequestResponseSummaryStream.tenantApiVersion
select alertTableRequestResponse.tenantApiVersion as tenantApiVersion, alertTableRequestResponse.count as count, api_version, tenantDomain, apiPublisher, requestsPerMin, requestsPerMinLowerPercentile, responsesPerMin, responsesPerMinLowerPercentile
insert into intermRequestReponseSummaryStream;

@info(name = 'requestResponse7')
from intermRequestReponseSummaryStream[count > 5]
select api_version , ifThenElse(tenantDomain == 'carbon.super', str:concat(apiPublisher, "@carbon.super"), apiPublisher) as apiPublisher, tenantDomain, str:concat('Response count is lower than ' , responsesPerMinLowerPercentile, ' continuously for 5 or more minutes.')  as msg, 2 as severity, (time:timestampInMilliseconds()) as alertTimestamp
insert into apiHealthMonitorAlertStreamTemp;
----------------------------------------------------------------------------------------------------------------------------------------------------------

@info(name = 'indexing')
from apiHealthMonitorAlertStream
select api_version, apiPublisher, tenantDomain, msg, alertTimestamp
insert into apiHealthMonitorCurrentAlertTable;

/* Now start response time queries */

@info(name = 'query7')
from responseSummaryStream
select api_version, apiPublisher, responseTime, tenantDomain
insert into responseTimeSummaryStream;

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

@info(name = 'query8')
from responseTimeSummaryStream#window.timeBatch(1 min) join responseTimePerApiPercentileTable
on (responseTimeSummaryStream.api_version==responseTimePerApiPercentileTable.api_version AND responseTimeSummaryStream.tenantDomain==responseTimePerApiPercentileTable.tenantDomain)
select responseTimeSummaryStream.api_version, responseTimeSummaryStream.apiPublisher, responseTimeSummaryStream.tenantDomain, responseTimeSummaryStream.responseTime, responseTimePerApiPercentileTable.responseTimePerApiUpperPercentile
insert into responseInfoStream;

----------------------------------------------------------------------------------------------------------------------------------------------------------
@info(name = 'responseInfo1')
from responseInfoStream
select str:concat(tenantDomain, "_", api_version) as tenantApiVersion, api_version, tenantDomain, apiPublisher, responseTime, responseTimePerApiUpperPercentile
insert into tmpResponseInfoSummaryStream;

@info(name = 'responseInfo2')
from tmpResponseInfoSummaryStream[(responseTime>responseTimePerApiUpperPercentile) and not (alertTableResponseInfo.tenantApiVersion == tenantApiVersion in alertTableResponseInfo)]
select tenantApiVersion, 1 as count
insert into alertTableResponseInfo;

@info(name = 'responseInfo3')
from tmpResponseInfoSummaryStream[not(responseTime>responseTimePerApiUpperPercentile)]
delete alertTableResponseInfo
    on alertTableResponseInfo.tenantApiVersion == tenantApiVersion;

@info(name = 'responseInfo4')
from tmpResponseInfoSummaryStream[(responseTime>responseTimePerApiUpperPercentile) and alertTableResponseInfo.tenantApiVersion == tenantApiVersion in alertTableResponseInfo]
insert into updateResponseInfoStream;

@info(name = 'responseInfo5')
from updateResponseInfoStream#window.length(1) join alertTableResponseInfo
	on alertTableResponseInfo.tenantApiVersion == updateResponseInfoStream.tenantApiVersion
select alertTableResponseInfo.tenantApiVersion, (alertTableResponseInfo.count + 1) as count
insert into alertTableResponseInfo;

@info(name = 'responseInfo6')
from tmpResponseInfoSummaryStream#window.length(1) join alertTableResponseInfo
	on alertTableResponseInfo.tenantApiVersion == tmpResponseInfoSummaryStream.tenantApiVersion
select alertTableResponseInfo.tenantApiVersion as tenantApiVersion, alertTableResponseInfo.count as count, api_version, tenantDomain, apiPublisher, responseTime, responseTimePerApiUpperPercentile
insert into intermResponseInfoSummaryStream;

@info(name = 'responseInfo7')
from intermResponseInfoSummaryStream[count > 5]
select api_version, ifThenElse(tenantDomain == 'carbon.super', str:concat(apiPublisher, "@carbon.super"), apiPublisher) as apiPublisher, tenantDomain, str:concat('Response time is higher than ' , responseTimePerApiUpperPercentile, ' ms continuously for 5 or more responses.') as msg, 2 as severity, (time:timestampInMilliseconds()) as alertTimestamp
insert into apiHealthMonitorAlertStreamTemp;
----------------------------------------------------------------------------------------------------------------------------------------------------------

@info(name = 'query10')
from responseStream
select api_version, apiPublisher,  responseCode, tenantDomain
insert into responseCodeSummaryStream;

-----------------------------------------------------------------------------------------------------------------------------
@info(name = 'responseCode1')
from responseCodeSummaryStream
select str:concat(tenantDomain, "_", api_version) as tenantApiVersion, api_version, tenantDomain, apiPublisher, responseCode
insert into tmpResponseCodeSummaryStream;

@info(name = 'responseCode2')
from tmpResponseCodeSummaryStream[(responseCode > 499 AND responseCode < 600) and not (alertTableResponseCode.tenantApiVersion == tenantApiVersion in alertTableResponseCode)]
select tenantApiVersion, 1 as count
insert into alertTableResponseCode;

@info(name = 'responseCode3')
from tmpResponseCodeSummaryStream[not(responseCode > 499 AND responseCode < 600)]
delete alertTableResponseCode
    on alertTableResponseCode.tenantApiVersion == tenantApiVersion;

@info(name = 'responseCode4')
from tmpResponseCodeSummaryStream[(responseCode > 499 AND responseCode < 600) and alertTableResponseCode.tenantApiVersion == tenantApiVersion in alertTableResponseCode]
insert into updateStream;

@info(name = 'responseCode5')
from updateStream#window.length(1) join alertTableResponseCode
	on alertTableResponseCode.tenantApiVersion == updateStream.tenantApiVersion
select alertTableResponseCode.tenantApiVersion, (alertTableResponseCode.count + 1) as count
insert into alertTableResponseCode;

@info(name = 'responseCode6')
from tmpResponseCodeSummaryStream#window.length(1) join alertTableResponseCode
	on alertTableResponseCode.tenantApiVersion == tmpResponseCodeSummaryStream.tenantApiVersion
select alertTableResponseCode.tenantApiVersion as tenantApiVersion, alertTableResponseCode.count as count, api_version, tenantDomain, apiPublisher, responseCode
insert into intermResponseCodeSummaryStream;

@info(name = 'responseCode7')
from intermResponseCodeSummaryStream[count > 5]
select api_version, ifThenElse(tenantDomain == 'carbon.super', str:concat(apiPublisher, "@carbon.super"), apiPublisher) as apiPublisher, tenantDomain, 'Server error occurred continuously for 5 or more times.'  as msg, 2 as severity, (time:timestampInMilliseconds()) as alertTimestamp
insert into apiHealthMonitorAlertStreamTemp;

/* suppression logic for response code */

@info(name = 'suppression1')
from apiHealthMonitorAlertStreamTemp as r left outer join ApiHealthAlertSummaryTable
on (api_version ==ApiHealthAlertSummaryTable.api_version and apiPublisher == ApiHealthAlertSummaryTable.apiPublisher and tenantDomain == ApiHealthAlertSummaryTable.tenantDomain and msg == ApiHealthAlertSummaryTable.msg)
select r.api_version, r.apiPublisher , r.tenantDomain , r.msg, r.severity, (time:timestampInMilliseconds()) as alertTimestamp, ifThenElse(lastAlertedTimestamp is null, 0l , lastAlertedTimestamp) as lastAlertedTimestamp
insert into ApiHealthAlertSummaryTableModified;

@info(name='suppression2')
from ApiHealthAlertSummaryTableModified[(alertTimestamp - lastAlertedTimestamp) > 30*60*1000]
select api_version, apiPublisher, tenantDomain, msg, severity, alertTimestamp
insert into apiHealthMonitorAlertStream;

@info(name='suppression3')
from apiHealthMonitorAlertStream[(api_version == ApiHealthAlertSummaryTable.api_version   and apiPublisher == ApiHealthAlertSummaryTable.apiPublisher and tenantDomain == ApiHealthAlertSummaryTable.tenantDomain and msg == ApiHealthAlertSummaryTable.msg ) in ApiHealthAlertSummaryTable]
select  api_version, apiPublisher, tenantDomain, msg, alertTimestamp as lastAlertedTimestamp
update ApiHealthAlertSummaryTable
on (api_version == ApiHealthAlertSummaryTable.api_version   and apiPublisher == ApiHealthAlertSummaryTable.apiPublisher and tenantDomain == ApiHealthAlertSummaryTable.tenantDomain and msg == ApiHealthAlertSummaryTable.msg);

@info(name = 'suppression4')
from apiHealthMonitorAlertStream[not ((api_version == ApiHealthAlertSummaryTable.api_version and apiPublisher == ApiHealthAlertSummaryTable.apiPublisher and tenantDomain == ApiHealthAlertSummaryTable.tenantDomain and msg == ApiHealthAlertSummaryTable.msg) in ApiHealthAlertSummaryTable)]
select  api_version, apiPublisher, tenantDomain, msg,  alertTimestamp as lastAlertedTimestamp
insert into ApiHealthAlertSummaryTable;


@info(name='query12')
from ApiHealthAlertSummaryTableModified[(alertTimestamp - lastAlertedTimestamp) > 30*60*1000]
select 'healthAvailabilityPerMin' as type, tenantDomain, str:concat('API:', api_version, '-', msg) as msg , severity, (time:timestampInMilliseconds()) as alertTimestamp
insert into allApimAlertStream;

/* update apiAvailability Table */


@info(name='query13')
from apiHealthMonitorAlertStream[((api_version == apiAvailabilityTable.api_version AND tenantDomain == apiAvailabilityTable.tenantDomain) in apiAvailabilityTable)]
select api_version, tenantDomain,msg as status
update apiAvailabilityTable
on (api_version == apiAvailabilityTable.api_version AND tenantDomain == apiAvailabilityTable.tenantDomain);

@info(name='query14')
from apiHealthMonitorAlertStream[ not ((api_version == apiAvailabilityTable.api_version AND tenantDomain == apiAvailabilityTable.tenantDomain) in apiAvailabilityTable)]
select api_version, tenantDomain, msg as status
insert into apiAvailabilityTable;

@info(name='query15')
from requestPerMinStream[(not ((api_version == apiHealthMonitorAlertStreamTable.api_version AND tenantDomain == apiHealthMonitorAlertStreamTable.tenantDomain) in apiHealthMonitorAlertStreamTable)) and (not ((api_version == apiAvailabilityTable.api_version AND tenantDomain == apiAvailabilityTable.tenantDomain) in apiAvailabilityTable))]
select api_version , tenantDomain, 'Available' as status
insert into apiAvailabilityTable;

@info(name='query16')
from requestPerMinStream as rpm join apiHealthMonitorCurrentAlertTable as latestAlertTable
on (rpm.api_version == latestAlertTable.api_version AND rpm.tenantDomain == latestAlertTable.tenantDomain)
select rpm.api_version, rpm.tenantDomain, latestAlertTable.alertTimestamp
insert into tempStream;


@info(name='query17')
from tempStream[(time:timestampInMilliseconds() - alertTimestamp) > 300000]
select api_version , tenantDomain, 'Available' as status
update apiAvailabilityTable
on (api_version == apiAvailabilityTable.api_version AND tenantDomain == apiAvailabilityTable.tenantDomain);