<?xml version="1.0" encoding="UTF-8"?>

<!--
  ~ Copyright (c) 2016, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
  ~
  ~ Licensed under the Apache License, Version 2.0 (the "License");
  ~ you may not use this file except in compliance with the License.
  ~ You may obtain a copy of the License at
  ~
  ~     http://www.apache.org/licenses/LICENSE-2.0
  ~
  ~ Unless required by applicable law or agreed to in writing, software
  ~ distributed under the License is distributed on an "AS IS" BASIS,
  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  ~ See the License for the specific language governing permissions and
  ~ limitations under the License.
  -->
<domain name="APIMAnalytics">
    <description>Real-time and batch API Analytics.</description>
    <scenarios>
        <scenario type="RequestSummarizer">
            <description>Responsible for summarizing incoming traffic.</description>
            <templates>
                <template type="realtime">
                    <![CDATA[
/* Enter a unique ExecutionPlan */
@Plan:name('APIMAnalytics-RequestSummarizer-RequestSummarizer-realtime1')

/* Enter a unique description for ExecutionPlan */

-- @Plan:description('ExecutionPlan')

@Plan:statistics('false')

@Import('org.wso2.apimgt.statistics.request:1.1.0')
define stream requests (meta_clientType string, consumerKey string, context string, api_version string, api string, resourcePath string, resourceTemplate string, method string, version string, request int, requestTime long, userId string, tenantDomain string, hostName string, apiPublisher string, applicationName string, applicationId string, userAgent string, tier string, throttledOut bool, clientIp string, applicationOwner string);

@Import('org.wso2.apimgt.statistics.response:1.1.0')
define stream responses (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);

@Import('org.wso2.apimgt.statistics.execution.time:1.0.0')
define stream executionTimes (meta_clientType string, api string, api_version string, tenantDomain string, apiPublisher string, apiResponseTime long, context string, securityLatency long, throttlingLatency long, requestMediationLatency long, responseMediationLatency long, backendLatency long, otherLatency long, eventTime long);

@Export('org.wso2.analytics.apim.stateStream:1.0.0')
define stream stateStream (context string, method string, resourceTemplate string, tenantDomain string, state string, userId string, requestTimestamp string, consumerKey string, applicationName string, applicationOwner string);

@Export('org.wso2.apimgt.statistics.perMinuteRequest:1.0.0')
define stream perMinuteRequests (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);

@Export('org.wso2.apimgt.statistics.perMinuteResponse:1.0.0')
define stream perMinuteResponses (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.apimgt.statistics.perMinuteExecutionTimes:1.0.0')
define stream perMinuteExecutionTimes (meta_clientType string, year int, month int, day int, hour int, minute int, context string, api_version string, api string, tenantDomain string, apiPublisher string, apiResponseTime double, securityLatency double, throttlingLatency double, requestMediationLatency double, responseMediationLatency double, backendLatency double, otherLatency double, eventTime long, _timestamp long);

@Export('org.wso2.analytics.apim.responsePerMinPerApiStream:1.0.0')
define stream responsePerMinPerApiStream (api_version string, tenantDomain string, responsesPerMinPerApi long);

@Export('org.wso2.analytics.apim.requestPerMinPerApiStream:1.0.0')
define stream requestPerMinPerApiStream (api_version string, tenantDomain string, requestsPerMinPerApi long);

@Export('org.wso2.apimgt.statistics.requestsPerMinPerKeyStream:1.0.0')
define stream requestsPerMinPerKeyStream (consumerKey string, requestsPerMinPerKey long);


-- requests

-- second
-- global
from requests#window.externalTimeBatch(requestTime, 1 sec, 0, $schedulerTimeInterval sec, true)
select requestTime as startTime, meta_clientType, consumerKey, context, api_version, api, version,
max(requestTime) as requestTime, userId, hostName, apiPublisher, count() as total_request_count, 
resourceTemplate, method, applicationName, tenantDomain, userAgent, resourcePath, request, applicationId, 
tier, throttledOut, clientIp, applicationOwner
group by consumerKey, context, api_version, userId, hostName, apiPublisher, resourceTemplate, method, userAgent, clientIp  
insert into perSecondRequestsTemp;

from perSecondRequestsTemp
select meta_clientType, (startTime - (startTime % 1000)) as startTime, convert((startTime - (startTime % 1000)), 'string') as facetStartTime,
time:extract(startTime, 'year') as year, time:extract(startTime, 'month') as month, 
time:extract(startTime, 'day') as day, time:extract(startTime, 'hour') as hour, 
time:extract(startTime, 'minute') as minute, time:extract(startTime, 'second') as second,
consumerKey, context, api_version, api, version, 
requestTime, userId, hostName, apiPublisher, total_request_count, 
resourceTemplate, method, applicationName, tenantDomain, userAgent, resourcePath, request, applicationId, 
tier, throttledOut, clientIp, applicationOwner
insert into perSecondRequests;

--api_version
from requests#window.externalTimeBatch(requestTime, 1 sec, 0, $schedulerTimeInterval sec, true)
select requestTime as startTime, api_version, tenantDomain, count() as total_request_count
group by api_version, tenantDomain 
insert into perSecondPerApiRequestsTemp;

from perSecondPerApiRequestsTemp
select (startTime - (startTime % 1000)) as startTime, convert((startTime - (startTime % 1000)), 'string') as facetStartTime,
api_version, tenantDomain, total_request_count
insert into perSecondPerApiRequests;

-- consumer key
from requests#window.externalTimeBatch(requestTime, 1 sec, 0, 10 sec)
select requestTime as startTime, consumerKey, count() as total_request_count
group by consumerKey 
insert into perSecondPerKeyRequestsTemp;

from perSecondPerKeyRequestsTemp
select (startTime - (startTime % 1000)) as startTime, convert((startTime - (startTime % 1000)), 'string') as facetStartTime,
consumerKey, total_request_count
insert into perSecondPerKeyRequests;

-- minute
-- global
from perSecondRequests
select startTime, meta_clientType, str:concat(facetStartTime, '_', consumerKey, '_', context, '_', 
  api_version, '_', userId, '_', hostName, '_', apiPublisher, '_', resourceTemplate, '_', method, '_', userAgent, '_', clientIp) 
as uniqueId, year, month, day, hour, minute, second, consumerKey, context, api_version, api, version, 
requestTime, userId, hostName, apiPublisher, total_request_count, resourceTemplate, method, applicationName, 
tenantDomain, userAgent, resourcePath, request, applicationId, tier, throttledOut, clientIp, applicationOwner
insert into perSecondRequestsWithUniqueId;

from perSecondRequestsWithUniqueId#window.uniqueExternalTimeBatch(uniqueId, startTime, $timeInterval sec, 59999 millisec, $schedulerTimeInterval sec, true)
select startTime, meta_clientType, consumerKey, context, api_version, api, version,
max(requestTime) as requestTime, userId, hostName, apiPublisher, sum(total_request_count) as total_request_count, 
resourceTemplate, method, applicationName, tenantDomain, userAgent, resourcePath, request, applicationId, tier, 
throttledOut, clientIp, applicationOwner
group by consumerKey, context, api_version, userId, hostName, apiPublisher, resourceTemplate, method, userAgent, clientIp  
insert into perMinuteRequestsTemp;

from perMinuteRequestsTemp
select meta_clientType, time:extract(startTime, 'year') as year, time:extract(startTime, 'month') as month, 
time:extract(startTime, 'day') as day, time:extract(startTime, 'hour') as hour, 
time:extract(startTime, 'minute') as minute,
consumerKey, context, api_version, api, version, 
requestTime, userId, hostName, apiPublisher, total_request_count, 
resourceTemplate, method, applicationName, tenantDomain, userAgent, resourcePath, request, applicationId, 
tier, throttledOut, clientIp, applicationOwner, startTime as _timestamp
insert into perMinuteRequests;

-- api_version
from perSecondPerApiRequests
select startTime, str:concat(facetStartTime, '_', api_version) 
as uniqueId, api_version, tenantDomain, total_request_count
insert into perSecondPerApiRequestsWithUniqueId;

from perSecondPerApiRequestsWithUniqueId#window.uniqueExternalTimeBatch(uniqueId, startTime, $timeInterval sec, 59999 millisec, $schedulerTimeInterval sec, true)
select api_version, tenantDomain, sum(total_request_count) as requestsPerMinPerApi
group by api_version, tenantDomain
insert into requestPerMinPerApiStream;

-- consumer key
from perSecondPerKeyRequests
select startTime, str:concat(facetStartTime, '_', consumerKey) 
as uniqueId, consumerKey, total_request_count
insert into perSecondPerKeyRequestsWithUniqueId;

from perSecondPerKeyRequestsWithUniqueId#window.uniqueExternalTimeBatch(uniqueId, startTime, 60 sec, 59999 millisec, 10 sec)
select consumerKey, sum(total_request_count) as requestsPerMinPerKey
group by consumerKey 
insert into requestsPerMinPerKeyStream;

-- state stream
from requests
select context,method,resourceTemplate, tenantDomain, str:concat(method,' ',context,' ',resourceTemplate) as state, userId as userId, 
time:dateFormat(requestTime, 'yyyy-MM-dd HH:mm:ss') as requestTimestamp, consumerKey, applicationName, applicationOwner 
insert into stateStream;

-- responses

-- second
-- global
from responses#window.externalTimeBatch(eventTime, 1 sec, 0, $schedulerTimeInterval sec, true)
select eventTime as startTime, meta_clientType, context, api_version, api, resourceTemplate, version,
tenantDomain, hostName, apiPublisher, destination, consumerKey, resourcePath, method, response, 
avg(responseTime) as responseTime, avg(serviceTime) as serviceTime, avg(backendTime) as backendTime, 
username, max(eventTime) as eventTime, applicationName, applicationId, cacheHit, avg(responseSize) as responseSize, 
protocol, responseCode, count() as total_response_count 
group by consumerKey, context, api_version, username, resourceTemplate, hostName, apiPublisher, method, destination  
insert into perSecondResponsesTemp;

from perSecondResponsesTemp
select meta_clientType, (startTime - (startTime % 1000)) as startTime, convert((startTime - (startTime % 1000)), 'string') as facetStartTime,
time:extract(startTime, 'year') as year, 
time:extract(startTime, 'month') as month, time:extract(startTime, 'day') as day, 
time:extract(startTime, 'hour') as hour, time:extract(startTime, 'minute') as minute, 
time:extract(startTime, 'second') as second, 
context, api_version, api, resourceTemplate, version, 
tenantDomain, hostName, apiPublisher, destination, consumerKey, resourcePath, method, response, 
responseTime, serviceTime, backendTime, 
username, eventTime, applicationName, applicationId, cacheHit, responseSize, 
protocol, responseCode, total_response_count
insert into perSecondResponses;

-- api_version
from responses#window.externalTimeBatch(eventTime, 1 sec, 0, $schedulerTimeInterval sec, true)
select eventTime as startTime, api_version, tenantDomain, count() as total_response_count
group by api_version, tenantDomain
insert into perSecondPerApiResponsesTemp;

from perSecondPerApiResponsesTemp
select (startTime - (startTime % 1000)) as startTime, convert((startTime - (startTime % 1000)), 'string') as facetStartTime,
api_version, tenantDomain, total_response_count
insert into perSecondPerApiResponses;

-- minute
-- global
from perSecondResponses
select startTime, meta_clientType, str:concat(facetStartTime, '_', consumerKey, '_', context, '_', api_version, 
  '_', username, '_', hostName, '_', apiPublisher, '_', resourceTemplate, '_', method, '_', destination) as uniqueId, 
year, month, day, hour, minute, second, context, api_version, api, resourceTemplate, version, tenantDomain, 
hostName, apiPublisher, destination, consumerKey, resourcePath, method, response, responseTime, serviceTime, 
backendTime, username, eventTime, applicationName, applicationId, cacheHit, responseSize, protocol, responseCode, 
total_response_count
insert into perSecondResponsesWithUniqueId;

from perSecondResponsesWithUniqueId#window.uniqueExternalTimeBatch(uniqueId, startTime, $timeInterval sec, 59999 millisec, $schedulerTimeInterval sec, true)
select startTime, meta_clientType, context, api_version, api, resourceTemplate, version, tenantDomain,
hostName, apiPublisher, destination, consumerKey, resourcePath, method, response, avg(responseTime) as responseTime, 
avg(serviceTime) as serviceTime, avg(backendTime) as backendTime, username, max(eventTime) as eventTime, applicationName, 
applicationId, cacheHit, max(responseSize) as responseSize, protocol, responseCode, 
sum(total_response_count) as total_response_count
group by consumerKey, context, api_version, username, resourceTemplate, hostName, apiPublisher, method, destination    
insert into perMinuteResponsesTemp;

from perMinuteResponsesTemp
select meta_clientType, time:extract(startTime, 'year') as year, time:extract(startTime, 'month') as month, 
time:extract(startTime, 'day') as day, time:extract(startTime, 'hour') as hour, 
time:extract(startTime, 'minute') as minute,
context, api_version, api, resourceTemplate, version, tenantDomain, 
hostName, apiPublisher, destination, consumerKey, resourcePath, method, response, math:round(responseTime) as responseTime, 
math:round(serviceTime) as serviceTime, math:round(backendTime) as backendTime, username, eventTime, applicationName, 
applicationId, cacheHit, math:round(responseSize) as responseSize, protocol, responseCode, total_response_count, startTime as _timestamp
insert into perMinuteResponses;

-- api_version
from perSecondPerApiResponses
select startTime, str:concat(facetStartTime, '_', api_version) 
as uniqueId, api_version, tenantDomain, total_response_count
insert into perSecondPerApiResponsesWithUniqueId;

from perSecondPerApiResponsesWithUniqueId#window.uniqueExternalTimeBatch(uniqueId, startTime, $timeInterval sec, 59999 millisec, $schedulerTimeInterval sec, true)
select api_version, tenantDomain, sum(total_response_count) as responsesPerMinPerApi
group by api_version, tenantDomain
insert into responsePerMinPerApiStream;

-- execution times

-- second

from executionTimes#window.externalTimeBatch(eventTime, 1 sec, 0, $schedulerTimeInterval sec, true)
select eventTime as startTime, meta_clientType, context, api_version, api,
tenantDomain, apiPublisher, avg(apiResponseTime) as apiResponseTime, avg(securityLatency) as securityLatency,
avg(throttlingLatency) as throttlingLatency, avg(requestMediationLatency) as requestMediationLatency,
avg(responseMediationLatency) as responseMediationLatency, avg(backendLatency) as backendLatency,
avg(otherLatency) as otherLatency
group by context, api_version, tenantDomain, apiPublisher 
insert into perSecondExecutionTimesTemp;

from perSecondExecutionTimesTemp
select meta_clientType, (startTime - (startTime % 1000)) as startTime, convert((startTime - (startTime % 1000)), 'string') as facetStartTime,
time:extract(startTime, 'year') as year, 
time:extract(startTime, 'month') as month, time:extract(startTime, 'day') as day, 
time:extract(startTime, 'hour') as hour, time:extract(startTime, 'minute') as minute, 
time:extract(startTime, 'second') as second, 
context, api_version, api, tenantDomain, apiPublisher, apiResponseTime, securityLatency,
throttlingLatency, requestMediationLatency, responseMediationLatency, backendLatency, otherLatency,
startTime as eventTime, startTime as _timestamp
insert into perSecondExecutionTimes;

-- minute

from perSecondExecutionTimes
select startTime, meta_clientType, str:concat(facetStartTime, '_', context, '_', 
  api_version, '_', apiPublisher, '_', tenantDomain) 
as uniqueId, year, month, day, hour, minute, second, context, api_version, api, tenantDomain, apiPublisher, apiResponseTime, securityLatency,
throttlingLatency, requestMediationLatency, responseMediationLatency, backendLatency, otherLatency
insert into perSecondExecutionTimeWithUniqueId;

from perSecondExecutionTimeWithUniqueId#window.uniqueExternalTimeBatch(uniqueId, startTime, $timeInterval sec, 59999 millisec, $schedulerTimeInterval sec, true)
select startTime, meta_clientType, context, api_version, api,
tenantDomain, apiPublisher, avg(apiResponseTime) as apiResponseTime, avg(securityLatency) as securityLatency,
avg(throttlingLatency) as throttlingLatency, avg(requestMediationLatency) as requestMediationLatency,
avg(responseMediationLatency) as responseMediationLatency, avg(backendLatency) as backendLatency,
avg(otherLatency) as otherLatency
group by context, api_version, tenantDomain, apiPublisher  
insert into perMinuteExecutionTimeTemp;

from perMinuteExecutionTimeTemp
select meta_clientType, time:extract(startTime, 'year') as year, time:extract(startTime, 'month') as month, 
time:extract(startTime, 'day') as day, time:extract(startTime, 'hour') as hour, 
time:extract(startTime, 'minute') as minute,
context, api_version, api, tenantDomain, apiPublisher, apiResponseTime, securityLatency,
throttlingLatency, requestMediationLatency, responseMediationLatency, backendLatency, otherLatency,
startTime as eventTime, startTime as _timestamp
insert into perMinuteExecutionTimes;

                    ]]>
                </template>
            </templates>
            <parameters>
                <parameter name="timeInterval" type="long">
                    <displayName>Time Interval</displayName>
                    <description>Time period in seconds of which the summarization would happen</description>
                    <defaultValue>60</defaultValue>
                </parameter>
                <parameter name="schedulerTimeInterval" type="long">
                    <displayName>Scheduler Time Interval</displayName>
                    <description>Time period in seconds of which the summarized data would get published if no further events are received</description>
                    <defaultValue>10</defaultValue>
                </parameter>
            </parameters>
        </scenario>
        <scenario type="FrequentTierLimitHitting">
            <description>Detects frequent hitting of the tier limit</description>
            <templates>
                <template type="realtime">
                    <![CDATA[
/* Enter a unique ExecutionPlan */
@Plan:name('APIMAnalytics-FrequentTierLimitHitting-FrequentTierLimitHitting-realtime1')

@Import('org.wso2.apimgt.statistics.throttle:1.0.0')
define stream throttleStream (meta_clientType string, accessToken string, userId string, tenantDomain string, api string, api_version string, context string, apiPublisher string, throttledTime long, applicationName string, applicationId string, subscriber string, throttledOutReason string);


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

@Export('org.wso2.analytics.apim.tierLimitHittingAlert:1.0.0')
define stream tierLimitHittingAlertStream (subscriber string, apiPublisher string, api string, applicationId string, applicationName string, tenantDomain string, msg string, severity int, alertTimestamp long);

/* read data to a temporary stream*/
@info(name = 'query1')
from throttleStream[throttledOutReason == 'SUBSCRIPTION_LIMIT_EXCEEDED']#window.time( $timeInterval )
select apiPublisher, api, api_version, applicationId, applicationName, tenantDomain, subscriber, count(userId) as numHits
group by apiPublisher, api_version, applicationId
having numHits > $noTierCrossings
insert into temporarySubscriberStream;

@info(name = 'query2')
from throttleStream[throttledOutReason == 'APPLICATION_LIMIT_EXCEEDED']#window.time( $timeInterval )
select userId, apiPublisher, api, api_version, applicationId, applicationName, tenantDomain, subscriber, count(userId) as numHits
group by userId, api_version, applicationId
having numHits > $noTierCrossings
insert into temporaryUserStream;

@info(name = 'query3')
from temporarySubscriberStream#window.length(1) as a left outer join temporarySubscriberStream#window.time($alertSuppressionPeriod minute) as b
on (a.apiPublisher== b.apiPublisher and a.api_version== b.api_version and a.applicationId==b.applicationId)
select ifThenElse(a.tenantDomain == 'carbon.super', str:concat(a.apiPublisher, "@carbon.super"), a.apiPublisher) as apiPublisher, a.api, a.api_version, a.applicationId, a.applicationName, a.tenantDomain, 
ifThenElse(a.tenantDomain == 'carbon.super', str:concat(a.subscriber, "@carbon.super"), a.subscriber) as subscriber, a.numHits
having b.apiPublisher is null
insert into suppressedTemporarySubscriberStream ;

@info(name = 'query4')
from temporaryUserStream#window.length(1) as a left outer join temporaryUserStream#window.time($alertSuppressionPeriod minute) as b
on (a.userId== b.userId and a.api_version== b.api_version and a.applicationId==b.applicationId)
select a.userId, ifThenElse(a.tenantDomain == 'carbon.super', str:concat(a.apiPublisher, "@carbon.super"), a.apiPublisher) as apiPublisher, a.api, a.api_version, a.applicationId, a.applicationName, a.tenantDomain, 
ifThenElse(a.tenantDomain == 'carbon.super', str:concat(a.subscriber, "@carbon.super"), a.subscriber) as subscriber, a.numHits
having b.userId is null
insert into suppressedTemporaryUserStream;

/* send to the alert stream specific to this scenario */
@info(name = 'query5')
from suppressedTemporarySubscriberStream
select subscriber, apiPublisher, api, applicationId, applicationName, tenantDomain, "Application frequently goes beyond the allocated quota when accessing an api." as msg, $severity as severity, (time:timestampInMilliseconds()) as alertTimestamp
insert into tierLimitHittingAlertStream;

@info(name = 'query6')
from suppressedTemporaryUserStream
select subscriber, apiPublisher, api, applicationId, applicationName, tenantDomain, str:concat("User ", userId, " frequently crosses the limit set.") as msg, $severity as severity, (time:timestampInMilliseconds()) as alertTimestamp
insert into tierLimitHittingAlertStream;

/* send to general alert stream */
@info(name = 'query7')
from suppressedTemporarySubscriberStream
select "FrequentTierHittingAlert" as type, tenantDomain, str:concat("The Application ", applicationName, " owned by ", subscriber, " frequently goes beyond the allocated quota when accessing the API ", api_version,".") as msg, $severity as severity, (time:timestampInMilliseconds()) as alertTimestamp
insert into allAlertsStream;

@info(name = 'query8')
from suppressedTemporaryUserStream
select "FrequentTierHittingAlert" as type, tenantDomain, str:concat("User ", userId, " using the ", applicationName, " application owned by ", subscriber, " frequently crosses the limit set for the user.") as msg, $severity as severity, (time:timestampInMilliseconds()) as alertTimestamp
insert into allAlertsStream;
]]>
                </template>
            </templates>
            <parameters>
                <parameter name="timeInterval" type="string">
                    <displayName>Time Interval</displayName>
                    <description>Time period (day/min/sec) of which the request count would be taken.</description>
                    <defaultValue>1 day</defaultValue>
                </parameter>

                <parameter name="alertSuppressionPeriod" type="long">
                    <displayName>Alert Suppression Period in Minutes</displayName>
                    <description>Time period in minutes to wait before resending the same alert</description>
                    <defaultValue>10</defaultValue>
                </parameter>

                <parameter name="noTierCrossings" type="int">
                    <displayName>No Of Tier Crossings</displayName>
                    <description>Max Number of tier crossings for the given time window</description>
                    <defaultValue>10</defaultValue>
                </parameter>

                <parameter name="severity" type="int">
                    <displayName>Severity Level</displayName>
                    <description>Severity level of the alert:(1:severe,2:moderate,3:mild)</description>
                    <defaultValue>3</defaultValue>
                    <options>1, 2, 3</options>
                </parameter>
            </parameters>
        </scenario>

        <scenario type="AbnormalRequestCountDetection">
            <description>Detects abnormal request counts</description>
            <templates>
                <template type="realtime">
                    <![CDATA[
                      /* Enter a unique ExecutionPlan */


@Plan:name('APIMAnalytics-AbnormalRequestCountDetection-AbnormalRequestCountDetection-realtime1')

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

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


@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);
                              
@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.abnormalRequestsPerMinAlertStream:1.0.0')
define stream abnormalRequestsPerMinAlertStream (api string, userId string, tenantDomain string, applicationName string, applicationOwner string, resourceTemplate string, method string, requestsPerMin long, requestsPerMinUpperPercentile double, requestsPerMinLowerPercentile double, reason string, msg string, severity int, alertTimestamp long);

@Export('org.wso2.analytics.apim.requestPerMinStream:1.0.0')
define stream requestPerMin (api_version string, userId string, consumerKey string, resourceTemplate string, method string, requestsPerMin long);

@from(eventtable = 'analytics.table' , table.name = 'ORG_WSO2_ANALYTICS_APIM_REQUESTPERCENTILE',  primary.keys = 'api_version,userId,consumerKey,resourceTemplate,method',  wait.for.indexing = 'false')
define table requestPercentileTable (api_version string, userId string, consumerKey string, resourceTemplate string, method string, requestsPerMinUpperPercentile double, requestsPerMinLowerPercentile double);

@info(name = 'query1')
from requestPerMinStream
select api_version, userId , consumerKey, resourceTemplate, method, count(total_request_count) as requestsPerMin
group by api_version, userId , consumerKey, resourceTemplate, method
insert into requestPerMin ;

/* retrieving corresponding percentile values from table and filtering abnormal counts*/
@info(name = 'query2')
from requestPerMinStream join requestPercentileTable
on (requestPerMinStream.api_version ==requestPercentileTable.api_version and  requestPerMinStream.userId ==requestPercentileTable.userId  and
requestPerMinStream.consumerKey ==requestPercentileTable.consumerKey and requestPerMinStream.resourceTemplate ==requestPercentileTable.resourceTemplate and requestPerMinStream.method ==requestPercentileTable.method)
select requestPerMinStream.api_version as api , requestPerMinStream.userId , requestPerMinStream.tenantDomain, requestPerMinStream.applicationName, requestPerMinStream.applicationOwner, requestPerMinStream.resourceTemplate, requestPerMinStream.method, requestPerMinStream.total_request_count,  requestPercentileTable.requestsPerMinUpperPercentile,   requestPercentileTable.requestsPerMinLowerPercentile, ifThenElse(total_request_count > requestsPerMinUpperPercentile,'spike','drop') as reason,
'Abnormal request count detected during last minute.' as msg , $severity as severity, (time:timestampInMilliseconds()) as alertTimestamp
having (total_request_count > requestsPerMinUpperPercentile or total_request_count < requestsPerMinLowerPercentile)
insert into tempAbnormalRequestsPerMinAlertStreamTemp;

@info(name = 'query3')
from tempAbnormalRequestsPerMinAlertStreamTemp#window.length(1) as a left outer join tempAbnormalRequestsPerMinAlertStreamTemp#window.time($alertSuppressionPeriod minute) as b
on (a.api== b.api and a.userId== b.userId and a.applicationName==b.applicationName and a.resourceTemplate==b.resourceTemplate and a.method== b.method and a.reason== b.reason)
select a.api, a.userId, a.tenantDomain, a.applicationName, ifThenElse(a.tenantDomain == 'carbon.super', str:concat(a.applicationOwner, "@carbon.super"), a.applicationOwner) as applicationOwner, a.resourceTemplate, a.method, a.total_request_count as requestsPerMin, a.requestsPerMinUpperPercentile, a.requestsPerMinLowerPercentile, a.reason, a.msg, a.severity, a.alertTimestamp
having b.api  is null
insert into abnormalRequestsPerMinAlertStream ;

@info(name = 'query4')
from abnormalRequestsPerMinAlertStream
select 'abnormalRequestsPerMin' as type, tenantDomain, str:concat('Abnormal request count ' , ifThenElse(requestsPerMin > requestsPerMinUpperPercentile,
'spike','drop') , ' detected during last minute by userId:',userId,' using application ', applicationName,' owned by ', applicationOwner,' for http ', 
method,' method of resource template ', resourceTemplate,' in api :', api,', abnormal request count:', requestsPerMin, ".") as msg , 
severity, (time:timestampInMilliseconds()) as alertTimestamp
insert into allApimAlertStream;
                 ]]>
                </template>
            </templates>
            <parameters>
                <parameter name="alertSuppressionPeriod" type="long">
                    <displayName>Alert Suppression Period in Minutes</displayName>
                    <description>Time period to wait before resending the same alert</description>
                    <defaultValue>10</defaultValue>
                </parameter>
                <parameter name="severity" type="int">
                    <displayName>Severity Level</displayName>
                    <description>Severity level of the alert:(1:severe,2:moderate,3:mild)</description>
                    <defaultValue>2</defaultValue>
                    <options>1, 2, 3</options>
                </parameter>
            </parameters>
        </scenario>

        <scenario type="AbnormalResponseAndBackendTimeDetection">
            <description>Detects abnormal backend time and response time</description>
            <templates>
                <template type="realtime">
                    <![CDATA[
/* 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);

@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='$cacheTimeout',  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, $responseTimeSeverity 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($alertSuppressionPeriod 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, $backendTimeSeverity 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($alertSuppressionPeriod 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;
                 ]]>
                </template>
            </templates>
            <parameters>
                <parameter name="alertSuppressionPeriod" type="long">
                    <displayName>Alert Suppression Period in Minutes</displayName>
                    <description>Time period to wait before resending the same alert</description>
                    <defaultValue>10</defaultValue>
                </parameter>

                <parameter name="cacheTimeout" type="int">
                    <displayName>Cache Time-out</displayName>
                    <description>Cache time-out value in seconds</description>
                    <defaultValue>300</defaultValue>
                </parameter>
                
                <parameter name="responseTimeSeverity" type="int">
                    <displayName>Severity Level of Abnormal Response Time</displayName>
                    <description>Severity level of the abnormal response time alert:(1:severe,2:moderate,3:mild)</description>
                    <defaultValue>2</defaultValue>
                    <options>1, 2, 3</options>
                </parameter>
                <parameter name="backendTimeSeverity" type="int">
                    <displayName>Severity Level of Abnormal Backend-Response Time</displayName>
                    <description>Severity level of the abnormal backend-response time alert:(1:severe,2:moderate,3:mild)</description>
                    <defaultValue>3</defaultValue>
                    <options>1, 2, 3</options>
                </parameter>
                
            </parameters>
        </scenario>

        <scenario type="RequestPatternChangeDetection">
            <description>Change the configuration of request pattern detection</description>
            <templates>
                <template type="realtime">
                    <![CDATA[
@Plan:name('APIMAnalytics-RequestPatternChangeDetection-APIRequestPatternChangeAnalysisMetric-realtime1')


@Import('org.wso2.analytics.apim.transitionStream:1.0.0')
define stream metricStream (startState string, endState string, consumerKey string, applicationName string, applicationOwner string, 
                   userId string, tenantDomain string, requestTimestamp string);

@Export('org.wso2.analytics.apim.requestPatternChangedStream:1.0.0')
define stream requestPatternChangedStream (userId string, applicationName string, applicationOwner string, tenantDomain string, msg string, severity int, alertTimestamp long);

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


@from(eventtable = 'analytics.table' , table.name = 'org_wso2_analytics_apim_MARKOVMODELTABLE', primary.keys='startState,endState, consumerKey', indices = 'startState,endState, probability, consumerKey', caching = 'true', cache.timeout.seconds='$cacheTimeout', wait.for.indexing = 'false' )
define table MarkovModelTable (consumerKey string, startState string, endState string, probability double);

@from(eventtable = 'analytics.table' , table.name = 'org_wso2_analytics_apim_REQUESTCOUNTTABLE', primary.keys='consumerKey', indices = 'consumerKey,count', caching = 'true', cache.timeout.seconds='$cacheTimeout', wait.for.indexing = 'false')
define table RequestCountPerConsumerKeyTable(consumerKey string, count long);

@from(eventtable = 'analytics.table' , table.name = 'ORG_WSO2_ANALYTICS_APIM_REQUESTPATTERNALERTSUMMARYTABLE', primary.keys='applicationName, applicationOwner, userId, msg' , indices = 'applicationName, applicationOwner, userId, msg, lastlastAlertedTimestamp', wait.for.indexing = 'false')
define table RequestPatternAlertSummaryTable(applicationName string, applicationOwner string, userId string, msg string, lastAlertedTimestamp long);

/****************************************************** Calculate Miss Probability *******************************************************************************/
@info(name = 'query1')
from metricStream as ms left outer join MarkovModelTable
on (ms.startState == MarkovModelTable.startState and ms.endState == MarkovModelTable.endState and ms.consumerKey == MarkovModelTable.consumerKey)
select ms.userId, ms.requestTimestamp, ms.startState ,ms.endState, ifThenElse(MarkovModelTable.probability is null, 1.0, 1- MarkovModelTable.probability) as MissProbability, ms.consumerKey,  ms.applicationName, ms.applicationOwner, ms.tenantDomain
insert into metricValueStreamTemp;

/****************************************************** fetch request count for consumerkey *******************************************************************************/
@info(name = 'query2')
from metricValueStreamTemp as m join RequestCountPerConsumerKeyTable
on (consumerKey == RequestCountPerConsumerKeyTable.consumerKey)
select userId, requestTimestamp, startState, endState, MissProbability, m.consumerKey, m.applicationName, m.applicationOwner , m.tenantDomain,  count
insert into metricValueStream;

/****************************************************** Compare Normalize Miss Probability with Threshold *******************************************************/
@info(name = 'query5')
from metricValueStream [ MissProbability > $probabilityThreshold and count >= $requestCount]
select userId, requestTimestamp,  str:concat(startState,' to ',endState) as transition, MissProbability as miss_probability, consumerKey, applicationName, applicationOwner, tenantDomain
insert into IntermediateFraudStream;

@info(name = 'query6')
from IntermediateFraudStream
select userId, applicationName, applicationOwner, tenantDomain, str:concat('Abnormal request pattern detected.' ,' Suspicious API transition is: ',transition) as msg , transition, (time:timestampInMilliseconds()) as alertTimestamp
insert into requestPatternChangedStreamTemp;

@info(name = 'query7')
from requestPatternChangedStreamTemp as r left outer join RequestPatternAlertSummaryTable
on (applicationName == RequestPatternAlertSummaryTable.applicationName   and applicationOwner == RequestPatternAlertSummaryTable.applicationOwner and userId == RequestPatternAlertSummaryTable.userId and msg == RequestPatternAlertSummaryTable.msg)
select r.applicationName, r.applicationOwner, r.tenantDomain, r.userId, r.msg, r.transition, alertTimestamp , ifThenElse(lastAlertedTimestamp is null, 0l , lastAlertedTimestamp) as lastAlertedTimestamp
insert into requestPatternChangedStreamModified;

@info(name = 'query8')
from requestPatternChangedStreamModified [(alertTimestamp - lastAlertedTimestamp) > $alertSuppressionPeriod]
select userId, applicationName, applicationOwner, tenantDomain, msg, $severity as severity, alertTimestamp
insert into requestPatternChangedStream;

@info(name = 'query9')
from requestPatternChangedStream[(applicationName == RequestPatternAlertSummaryTable.applicationName   and applicationOwner == RequestPatternAlertSummaryTable.applicationOwner and userId == RequestPatternAlertSummaryTable.userId and msg == RequestPatternAlertSummaryTable.msg ) in RequestPatternAlertSummaryTable]
select  applicationName, applicationOwner,userId, msg, alertTimestamp as lastAlertedTimestamp
update RequestPatternAlertSummaryTable
on (applicationName == RequestPatternAlertSummaryTable.applicationName and  applicationOwner == RequestPatternAlertSummaryTable.applicationOwner and userId == RequestPatternAlertSummaryTable.userId and msg == RequestPatternAlertSummaryTable.msg);

@info(name = 'query10')
from requestPatternChangedStream[not ((applicationName == RequestPatternAlertSummaryTable.applicationName and applicationOwner == RequestPatternAlertSummaryTable.applicationOwner and userId == RequestPatternAlertSummaryTable.userId and msg == RequestPatternAlertSummaryTable.msg) in RequestPatternAlertSummaryTable)]
select  applicationName, applicationOwner,userId, msg,  alertTimestamp as lastAlertedTimestamp
insert into RequestPatternAlertSummaryTable;

@info(name = 'query11')
from requestPatternChangedStreamModified[(alertTimestamp - lastAlertedTimestamp) > $alertSuppressionPeriod]
select 'RequestPatternChanged' as type, tenantDomain, str:concat('Abnormal request pattern detected by user :',userId,' using application : ',applicationName, ' owned by: ',applicationOwner, ' suspicious API transition is: ',transition, '.') as msg, $severity as severity, (time:timestampInMilliseconds()) as alertTimestamp
insert into all_apim_alerts_stream;
             
                 ]]>
                </template>
            </templates>
            <parameters>
                <parameter name="regularAPITransitions" type="int">
                    <displayName>Regular API Transactions</displayName>
                    <description>Regular Number of transitions per API</description>
                    <defaultValue>1</defaultValue>
                </parameter>

                <parameter name="requestCount" type="int">
                    <displayName>Request Count</displayName>
                    <description>Request count required for training</description>
                    <defaultValue>500</defaultValue>
                </parameter>

                <parameter name="probabilityThreshold" type="float">
                    <displayName>Probability Threshold</displayName>
                    <description>Probability threshold of a normal API transition</description>
                    <defaultValue>0.95</defaultValue>
                </parameter>

                <parameter name="alertSuppressionPeriod" type="long">
                    <displayName>Alert Suppression Period</displayName>
                    <description>Time period in milliseconds to wait before resending the same alert</description>
                    <defaultValue>30*60*1000</defaultValue>
                </parameter>

                <parameter name="cacheTimeout" type="int">
                    <displayName>Cache Time-out</displayName>
                    <description>Cache time-out value in seconds</description>
                    <defaultValue>300</defaultValue>
                </parameter>

                <parameter name="severity" type="int">
                    <displayName>Severity Level</displayName>
                    <description>Severity level of the alert:(1:severe,2:moderate,3:mild)</description>
                    <defaultValue>1</defaultValue>
                    <options>1, 2, 3</options>
                </parameter>
            </parameters>
        </scenario>

        <scenario type="MarkovStateClassifier">
            <description>Clasiffier configurations for Request Patterns</description>
            <templates>
                <template type="realtime">
                    <![CDATA[
/* Enter a unique ExecutionPlan */
@Plan:name('APIMAnalytics-MarkovStateClassifier-MarkovStateClassifier-realtime1')

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

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


@Import('org.wso2.analytics.apim.stateStream:1.0.0')
define stream stateStream (context string, method string, resourceTemplate string, tenantDomain string, state string, userId string, requestTimestamp string, consumerKey string, applicationName string, applicationOwner string);

@Import('org.wso2.apimgt.statistics.requestsPerMinPerKeyStream:1.0.0')
define stream requestsPerMinPerKeyStream (consumerKey string, requestsPerMinPerKey long);
   
@Export('org.wso2.analytics.apim.transitionStream:1.0.0')
define stream stateTransitionStream (startState string, endState string, consumerKey string, applicationName string, applicationOwner string, 
                   userId string, tenantDomain string, requestTimestamp string);

@from(eventtable = 'analytics.table' , table.name = 'org_wso2_analytics_apim_REQUESTCOUNTTABLE', primary.keys='consumerKey', indices = 'consumerKey,count', wait.for.indexing = 'false')
define table RequestCountPerConsumerKeyTable(consumerKey string, count long);

@info(name = 'query1')
from requestsPerMinPerKeyStream as r left outer join RequestCountPerConsumerKeyTable
on (r.consumerKey == RequestCountPerConsumerKeyTable.consumerKey)
select r.consumerKey , ifThenElse(count is null,r.requestsPerMinPerKey,count+r.requestsPerMinPerKey) as count
insert into requestCountPerConsumerKeyTableStream;

@info(name = 'query2')
from requestCountPerConsumerKeyTableStream#window.timeBatch(1 min)
select consumerKey,count
insert into RequestCountPerConsumerKeyTable;


@info(name = 'query3')
from stateStream#amAnalytics:buildTransitions(consumerKey, userId, state, applicationName, applicationOwner, tenantDomain, requestTimestamp, $batchSize)
select startState, state as endState, consumerKey, applicationName, applicationOwner, userId, tenantDomain, requestTimestamp 
insert into stateTransitionStream;
                                     ]]>
                </template>
            </templates>
            <parameters>
                <parameter name="batchSize" type="int">
                    <displayName>Batch Size</displayName>
                    <description>Events collected before submitting to the state transition stream</description>
                    <defaultValue>10000</defaultValue>
                </parameter>
            </parameters>
        </scenario>

        <scenario type="UnusualIPAccessTemplate">
            <description>To detect accesses from new or rarely used hostnames</description>
            <templates>
                <template type="realtime">
                    <![CDATA[
/* Enter a unique ExecutionPlan */
@Plan:name('APIMAnalytics-UnusualIPAccessTemplate-UnusualIPAccessAlert-realtime1')

/* Enter a unique description for ExecutionPlan */
@Plan:description('Alerts if an access from a strange ip is detected')

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

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

@Export('org.wso2.analytics.apim.alert.unusualIPAccess:1.0.0')
define stream alertIpAccessAbnormality (type string, msg string, severity int, ip string, applicationName string, applicationOwner string, userId string, tenantDomain string, requestTime long, alertTimestamp long);

@from(eventtable = 'analytics.table', table.name = 'ORG_WSO2_ANALYTICS_APIM_IPACCESSSUMMARY', primary.keys = 'consumerKey, ip', indices = 'userId, consumerKey, ip, lastAccessedDate', wait.for.indexing = 'false')
define table IPAccessSummary (userId string, consumerKey string, ip string, lastAccessedDate long) ;

@from(eventtable = 'analytics.table', table.name = 'ORG_WSO2_ANALYTICS_APIM_IPACCESSALERTCOUNT', primary.keys = 'consumerKey, userId', indices = 'userId, consumerKey, requestCount', wait.for.indexing = 'false')
define table IPAccessAlertCount (userId string, consumerKey string, requestCount long) ;

-- Consumer key is already in the table
@info(name = 'query1')
from requestPerMinStream[consumerKey == IPAccessSummary.consumerKey in IPAccessSummary]#window.length(1) as api left outer join IPAccessSummary as ais
on api.consumerKey == ais.consumerKey and api.clientIp == ais.ip
select api.consumerKey as consumerKey, api.tenantDomain, api.applicationName,  api.applicationOwner ,  api.userId as userId, api.clientIp as requestHostName, ais.ip as ip, api.requestTime as requestTime, ais.lastAccessedDate as lastAccessedDate, total_request_count as count
insert into TempStream;

-- New consumer key
@info(name = 'query2')
from requestPerMinStream[not (consumerKey == IPAccessSummary.consumerKey in IPAccessSummary)]
select userId, consumerKey, clientIp as ip, requestTime as lastAccessedDate, total_request_count as count
insert into NewConsumerKeyStream;

@info(name = 'query3')
from NewConsumerKeyStream
select userId, consumerKey, ip, lastAccessedDate
insert into IPAccessSummary;


-- Adding the alert count
@info(name = 'query4')
from NewConsumerKeyStream
select userId, consumerKey, count as requestCount
insert into IPAccessAlertCount;

@info(name = 'query5')
from TempStream as ts join IPAccessAlertCount as iaac
on ts.consumerKey == iaac.consumerKey and ts.userId == iaac.userId
select ts.consumerKey as consumerKey, ts.tenantDomain, ts.applicationName , ts.applicationOwner , ts.userId as userId, ts.requestHostName as requestHostName, ts.ip as ip, ts.requestTime as requestTime, ts.lastAccessedDate as lastAccessedDate, (ts.count + iaac.requestCount) as requestCount
insert into ExistingUserStream;

@info(name = 'query6')
from ExistingUserStream
select userId, consumerKey, requestCount
insert into IPAccessAlertCount;

-- A new request source IP
-- generate alert and add it to the table
@info(name = 'query7')
from ExistingUserStream[ip is null and requestCount > $learningCount]
select *
insert into NewIPDetectedStream;

@info(name = 'query8')
from NewIPDetectedStream
select "UnusualIPAccessAlert" as type, "A request from a new IP detected." as msg, $severity as severity, requestHostName as ip, applicationName, applicationOwner, userId, tenantDomain, requestTime,(time:timestampInMilliseconds()) as alertTimestamp
insert into alertIpAccessAbnormality;

@info(name = 'query9')
from NewIPDetectedStream
select userId, consumerKey, requestHostName as ip, requestTime as lastAccessedDate
insert into IPAccessSummary;

-- Check if the request for the IP is received after the threshold time gap
@info(name = 'query10')
from ExistingUserStream[(not (ip is null)) and requestCount > $learningCount]
select requestHostName, consumerKey, applicationName, applicationOwner, (requestTime - lastAccessedDate) as timeBetweenAccess, requestTime, userId, tenantDomain, lastAccessedDate
insert into CheckAbnormalTimeBetweenAccessStream;

@info(name = 'query11')
from CheckAbnormalTimeBetweenAccessStream[(timeBetweenAccess/1000) > ($maxDaysBetweenAccess*24*60*60)]
select *
insert into AbnormalTimeBetweenAccessStream;

@info(name = 'query12')
from AbnormalTimeBetweenAccessStream
select "UnusualIPAccessAlert" as type, "A request from an old IP detected." as msg, $severity as severity, requestHostName as ip, applicationName, applicationOwner, userId, tenantDomain, requestTime,(time:timestampInMilliseconds()) as alertTimestamp
insert into alertIpAccessAbnormality;

-- Send all the alerts to the Global Alert Stream
@info(name = 'query13')
from alertIpAccessAbnormality
select type, tenantDomain, str:concat("A request from a ", ifThenElse(str:contains(msg, 'old'), 'old','new'), " IP (", ip, ") detected by user:" , userId,  " using application:", applicationName, " owned by ", applicationOwner, ".") as msg, severity, (time:timestampInMilliseconds()) as alertTimestamp
insert into all_apim_alerts_stream;

-- Update the request time for the lastAccessDate
@info(name = 'query14')
from CheckAbnormalTimeBetweenAccessStream[(timeBetweenAccess/1000) > 24*60*60]
select userId, consumerKey, requestHostName as ip, requestTime as lastAccessedDate
insert into IPAccessSummary;               
                 ]]>
                </template>
            </templates>
            <parameters>
                <parameter name="maxDaysBetweenAccess" type="int">
                    <displayName>Maximum Days Between Last Access</displayName>
                    <description>Maximum days between Last access from the hostname to the latest </description>
                    <defaultValue>30</defaultValue>
                </parameter>
                <parameter name="severity" type="int">
                    <displayName>Severity Level</displayName>
                    <description>Severity level of the alert:(1:severe,2:moderate,3:mild)</description>
                    <defaultValue>2</defaultValue>
                    <options>1, 2, 3</options>
                </parameter>
                <parameter name="learningCount" type="long">
                    <displayName>Learning IP Count</displayName>
                    <description>This will be used as learning count for unusual IP access alerts</description>
                    <defaultValue>500</defaultValue>
                </parameter>
            </parameters>
        </scenario>

        <scenario type="HealthAvailabilityPerMinAlert">
            <description>Monitors the API Availability</description>
            <templates>
                <template type="realtime">
                    <![CDATA[
/* 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='$cacheTimeout', 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='$cacheTimeout', 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='$cacheTimeout', 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', indices='api_version, tenantDomain, status', caching='true', cache.timeout.seconds='$cacheTimeout', 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='$cacheTimeout', 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='$cacheTimeout', 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 > $responsePerMinPattern]
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 ', $responsePerMinPattern, ' 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 > $responseTimePattern]
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 ',$responseTimePattern,' 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 > $responseCodeCount]
select api_version, ifThenElse(tenantDomain == 'carbon.super', str:concat(apiPublisher, "@carbon.super"), apiPublisher) as apiPublisher, tenantDomain, str:concat('Server error occurred continuously for ', $responseCodeCount, ' 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) > $alertSuppressionTime]
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) > $alertSuppressionTime]
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) > $statusChangeTimeInterval]
select api_version , tenantDomain, 'Available' as status
update apiAvailabilityTable
on (api_version == apiAvailabilityTable.api_version AND tenantDomain == apiAvailabilityTable.tenantDomain);


                 ]]>
                </template>
            </templates>
            <parameters>
                <parameter name="responsePerMinPattern" type="int">
                    <displayName>Number of continuous responses</displayName>
                    <description>Number of responses that should fail to pass the lower percentile</description>
                    <defaultValue>5</defaultValue>
                </parameter>


                <parameter name="responseTimePattern" type="int">
                    <displayName>Number of continuous response time failures</displayName>
                    <description>number of minutes that responses should fail to pass the lower percentile</description>
                    <defaultValue>5</defaultValue>
                </parameter>

                <parameter name="statusChangeTimeInterval" type="int">
                    <displayName>Time interval(in milliseconds) for API availability status change</displayName>
                    <description>Time duration taken to recheck and change the availability of an API</description>
                    <defaultValue>300000</defaultValue>
                </parameter>

                <parameter name="cacheTimeout" type="int">
                    <displayName>Cache Time-out</displayName>
                    <description>Cache time-out value in seconds</description>
                    <defaultValue>300</defaultValue>
                </parameter>

                <parameter name="severity" type="int">
                    <displayName>Severity Level</displayName>
                    <description>Severity level of the alert:(1:severe,2:moderate,3:mild)</description>
                    <defaultValue>2</defaultValue>
                    <options>1, 2, 3</options>
                </parameter>

                <parameter name="alertSuppressionTime" type="int">
                    <displayName>Alert Suppression period(in milliseconds)</displayName>
                    <description>Same alert will be sent only once within this time period</description>
                    <defaultValue>30*60*1000</defaultValue>
                </parameter>

                <parameter name="responseCodeCount" type="int">
                    <displayName>(500-600)Response Code Count</displayName>
                    <description>How many response with code 500-600 should come continuously to generate an alert </description>
                    <defaultValue>5</defaultValue>
                </parameter>
            </parameters>
        </scenario>


        <scenario type="ConfigureAccessToken">
            <description>Configure access token to analyse data</description>
            <templates>
                <template type="batch">
                    <executionParameters>
                    <cron>0 0 12 1/7 * ? *</cron>

                    <sparkScript>
                        <![CDATA[
                  create temporary table accessTokenRefreshTime using CarbonAnalytics options (tableName "ORG_WSO2_ANALYTICS_APIM_ACCESSTOKENREFRESHTIMEDIFFERENCE", schema "userId STRING, clientId STRING, scopes STRING, timeDifference LONG, timestamp LONG");

create temporary table accessTokenRefreshAvrgTime using CarbonAnalytics options (tableName "ORG_WSO2_ANALYTICS_APIM_ACCESSTOKENREFRESHSUMMARYTABLE", schema "userId STRING -i, clientId STRING -i, scopes STRING -i, minTimeDifference DOUBLE -i, maxTimeDifference DOUBLE -i", primaryKeys "userId, clientId, scopes");

INSERT INTO TABLE accessTokenRefreshAvrgTime
      SELECT temp.userId, temp.clientId, temp.scopes, getpercentileValue(cast(AVG(timeDifference) as double), cast(SQRT(AVG(timeDifference*timeDifference) - AVG(timeDifference)*AVG(timeDifference)) as double), cast($lowerPercentile as double)) as minTimeDifference, getpercentileValue(cast(AVG(timeDifference) as double), cast(SQRT(AVG(timeDifference*timeDifference) - AVG(timeDifference)*AVG(timeDifference)) as double), cast($upperPercentile as double)) as maxTimeDifference
      FROM
         (SELECT userId, clientId,scopes, timeDifference
          FROM accessTokenRefreshTime
            WHERE timestamp >= offsetInDays(-7)) temp
      GROUP BY userId, clientId, scopes;

                 ]]>
                    </sparkScript>

                    </executionParameters>
                </template>

            </templates>

            <parameters>
                <parameter name="lowerPercentile" type="double">
                    <displayName>Lower percentile</displayName>
                    <description>Lower percentile (value between 0 and 1) to calculate minimum time difference</description>
                    <defaultValue>0.10</defaultValue>
                </parameter>


                <parameter name="upperPercentile" type="double">
                    <displayName>Upper percentile</displayName>
                    <description>Upper percentile (value between 0 and 1) to calculate maximum time difference</description>
                    <defaultValue>0.95</defaultValue>
                </parameter>
            </parameters>


    </scenario>


        <scenario type="AbnormalTierUsageAlert">
            <description>Detects Abnormal Tier Usage</description>
            <templates>
                <template type="batch">
                    <executionParameters>
                        <cron>0 0 23 * * ?</cron>
                        <sparkScript>
                            <![CDATA[
                                      CREATE TEMPORARY TABLE ORG_WSO2_API_REQUESTS
        USING CarbonAnalytics
        OPTIONS(tableName "ORG_WSO2_APIMGT_STATISTICS_PERMINUTEREQUEST");

                CREATE TEMPORARY TABLE ORG_WSO2_API_AVG_REQ_FOR_X_DAYS
                USING CarbonAnalytics
                OPTIONS (tableName "ORG_WSO2_ANALYTICS_APIM_AVG_REQ_FOR_X_DAYS_TBL",
                   schema "applicationId STRING, api_version STRING, request_count INT"
                );

                CREATE TEMPORARY TABLE ORG_WSO2_API_ABNORMAL_ADITIONAL_DATA
                USING CarbonAnalytics
                OPTIONS (tableName "ORG_WSO2_ANALYTICS_APIM_ABNORMAL_ADITIONAL_DATA_TBL",
                   schema "applicationId STRING, api_version STRING, tenantDomain STRING, applicationName STRING, applicationOwner STRING, api STRING, tier STRING"
                );

                CREATE TEMPORARY TABLE ORG_WSO2_API_PERCENTILE
                USING CarbonAnalytics
                OPTIONS (tableName "ORG_WSO2_ANALYTICS_APIM_PERCENTILE_TBL",
                   schema "applicationId STRING, api_version STRING, requestsPerDayPercentile FLOAT"
                );

                CREATE TEMPORARY TABLE ORG_WSO2_API_ABNORMAL_USAGE_ALERT
                USING CarbonAnalytics
                OPTIONS (tableName "ORG_WSO2_ANALYTICS_APIM_ABNORMAL_USAGE_ALERT_TBL",
                   schema "applicationId STRING, api_version STRING"
                );

                CREATE TEMPORARY TABLE ORG_WSO2_API_ABNORMAL_STREAM_TABLE
                USING org.wso2.carbon.analytics.spark.event.EventStreamProvider
                OPTIONS (streamName "org.wso2.analytics.apim.alert.abnormalTierUsageAlert",
                version "1.1.0",
                payload "applicationId STRING, applicationOwner STRING, api_version STRING, tenantDomain STRING, msg STRING, severity INTEGER, alertTimestamp LONG"
                );

                CREATE TEMPORARY TABLE ORG_WSO2_API_ALL_ALERT_STREAM_TABLE
                USING org.wso2.carbon.analytics.spark.event.EventStreamProvider
                OPTIONS (streamName "org.wso2.analytics.apim.allApimAlertsStream",
                         version "1.0.0",
                         payload "type STRING, tenantDomain string, msg STRING, severity INTEGER, alertTimestamp LONG"
                );

                INSERT OVERWRITE TABLE ORG_WSO2_API_PERCENTILE
                SELECT applicationId, api_version, getpercentileValue(cast(AVG(request_count) as double), cast(SQRT(AVG(request_count*request_count) - AVG(request_count)*AVG(request_count)) as double), cast($percentile as double)) as percentile
                FROM
                   (SELECT applicationId, api_version, SUM(total_request_count) as request_count
                    FROM ORG_WSO2_API_REQUESTS
                    WHERE context is not NULL  AND now('') > convertToTimestamp('$alertStartDate') AND requestTime >= offsetInDays(-$daysConsiderForPercentilesCalculation)
                    GROUP BY convertToDate(requestTime), applicationId, api_version) ORG_WSO2_API_REQUESTS_FILTERED
                GROUP BY applicationId, api_version;

               INSERT OVERWRITE TABLE ORG_WSO2_API_AVG_REQ_FOR_X_DAYS
               SELECT applicationId, api_version, SUM(total_request_count) as request_count
               FROM ORG_WSO2_API_REQUESTS
               WHERE context is not NULL  AND requestTime >= offsetInDays(-$daysConsiderForAbnormalTierAvailability)
               GROUP BY convertToDate(requestTime), applicationId, api_version;

               INSERT OVERWRITE TABLE ORG_WSO2_API_ABNORMAL_USAGE_ALERT
               SELECT S.applicationId, S.api_version
               FROM ORG_WSO2_API_AVG_REQ_FOR_X_DAYS S, ORG_WSO2_API_PERCENTILE D
               WHERE S.applicationId = D.applicationId AND S.api_version = D.api_version  AND D.requestsPerDayPercentile  > S.request_count
               GROUP BY S.applicationId, S.api_version
               HAVING COUNT(S.api_version) >= $daysConsiderForAbnormalTierAvailability;

               INSERT OVERWRITE TABLE ORG_WSO2_API_ABNORMAL_ADITIONAL_DATA
               SELECT DISTINCT applicationId, api_version, tenantDomain, applicationName, applicationOwner, api, tier
               FROM ORG_WSO2_API_REQUESTS;

               INSERT OVERWRITE TABLE ORG_WSO2_API_ABNORMAL_STREAM_TABLE
               SELECT R.applicationId, R.applicationOwner, R.api_version, R.tenantDomain, concat(concat(concat(concat(concat(concat(concat(concat(concat('The ', R.applicationName), ' Application owned by '), R.applicationOwner), ' is consuming less than the allowed quota when accessing the '), R.api_version), ' API'), '. It currently uses a '), R.tier), ' subscription.'), $severity, now('')
               FROM ORG_WSO2_API_ABNORMAL_USAGE_ALERT A, ORG_WSO2_API_ABNORMAL_ADITIONAL_DATA R
               WHERE A.api_version = R.api_version AND A.applicationId = R.applicationId;

               INSERT OVERWRITE TABLE ORG_WSO2_API_ALL_ALERT_STREAM_TABLE
               SELECT "AbnormalTierUsage", R.tenantDomain, concat(concat(concat(concat(concat(concat(concat(concat(concat('The ', R.applicationName), ' Application owned by '), R.applicationOwner), ' is consuming less than the allowed quota when accessing the '), R.api_version), ' API'), '. It currently uses a '), R.tier), ' subscription.'),  $severity, now('')
               FROM ORG_WSO2_API_ABNORMAL_USAGE_ALERT A, ORG_WSO2_API_ABNORMAL_ADITIONAL_DATA R
               WHERE A.api_version = R.api_version AND A.applicationId = R.applicationId;
                
                            ]]>
                        </sparkScript>
                    </executionParameters>
                </template>
            </templates>


            <parameters>

                <parameter name="percentile" type="double">
                    <displayName>Percentile</displayName>
                    <description>Percentile value (between 0 and 1) uses to calculate the threshold</description>
                    <defaultValue>0.05</defaultValue>
                </parameter>
                <parameter name="alertStartDate" type="string">
                    <displayName>Alert Start Date</displayName>
                    <description>Alerting will be activated from this day onwards. format: dd/mm/yyyy</description>
                    <defaultValue>12/2/2016</defaultValue>
                </parameter>
                <parameter name="daysConsiderForPercentilesCalculation" type="int">
                    <displayName>Days Considered for Percentile Calculation</displayName>
                    <description>Number of previous days considered for percentile calculation.</description>
                    <defaultValue>30</defaultValue>
                </parameter>
                <parameter name="daysConsiderForAbnormalTierAvailability" type="int">
                    <displayName>Days Considered for Abnormal Tier Availability Calculation</displayName>
                    <description>Number of previous days considered for abnormal tier availability calculation.
                    </description>
                    <defaultValue>5</defaultValue>
                </parameter>
                <parameter name="severity" type="int">
                    <displayName>Severity Level</displayName>
                    <description>Severity level of the alert:(1:severe,2:moderate,3:mild)</description>
                    <defaultValue>2</defaultValue>
                    <options>1, 2, 3</options>
                </parameter>

            </parameters>
        </scenario>


        <scenario type="RequestPerApi">
            <description>To create RequestPerApi percentile values</description>
            <templates>
            <template type="batch">
                <executionParameters>
                    <cron>0 0 12 * * ?</cron>
                    <sparkScript>
                        <![CDATA[
                CREATE TEMPORARY TABLE REQUEST_PER_API_INFO USING CarbonAnalytics OPTIONS (tableName "org_wso2_analytics_apim_requestPerMinPerApiStream");

        CREATE TEMPORARY TABLE REQUEST_PER_API_PERCENTILE_GEN USING CarbonAnalytics OPTIONS (tableName "ORG_WSO2_ANALYTICS_APIM_REQUESTPERAPIPERCENTILE",
        schema "requestsPerMinPerApiLowerPercentile double, api_version string -i, tenantDomain string -i",
        primaryKeys "api_version, tenantDomain", mergeSchema "false"
        );

        INSERT INTO TABLE REQUEST_PER_API_PERCENTILE_GEN
        select getpercentileValue(cast(avg(requestsPerMinPerApi) as double), cast(sqrt(avg(cast(requestsPerMinPerApi
        as double)*cast(requestsPerMinPerApi as double))-avg(requestsPerMinPerApi)*avg(requestsPerMinPerApi)) as double) , cast($lowerPercentile as double) ) as requestsPerMinPerApiLowerPercentile,
        api_version as api_version, tenantDomain

        from REQUEST_PER_API_INFO group by api_version, tenantDomain; 

                 ]]>
                    </sparkScript>
                </executionParameters>
            </template>
            </templates>

            <parameters>
                <parameter name="lowerPercentile" type="double">
                    <displayName>Lower percentile</displayName>
                    <description>Lower percentile value (between 0 and 1) to calculate request per min per api lower percentile
                    </description>
                    <defaultValue>0.05</defaultValue>
                </parameter>
            </parameters>
        </scenario>


        <scenario type="RequestStatGenerator">
            <description>To generate request stats</description>
            <templates>
                <template type="batch">
                    <executionParameters>
                        <cron>0 0 23 * * ?</cron>
                        <sparkScript>
                            <![CDATA[
                              CREATE TEMPORARY TABLE REQUEST_INFO USING CarbonAnalytics OPTIONS (tableName "ORG_WSO2_APIMGT_STATISTICS_PERMINUTEREQUEST");

        CREATE TEMPORARY TABLE REQUEST_PERCENTILE_GEN USING CarbonAnalytics OPTIONS (tableName "ORG_WSO2_ANALYTICS_APIM_REQUESTPERCENTILE",
        schema "api_version string, userId string, consumerKey string, resourceTemplate string, method string, requestsPerMinUpperPercentile double, requestsPerMinLowerPercentile double ",
        primaryKeys "api_version,userId,consumerKey,resourceTemplate,method"
        );

        INSERT OVERWRITE TABLE REQUEST_PERCENTILE_GEN
        select api_version, userId, consumerKey, resourceTemplate, method,
        getpercentileValue(cast(avg(total_request_count) as double), cast(sqrt(avg(cast(total_request_count
        as double)*cast(total_request_count as double))-avg(total_request_count)*avg(total_request_count)) as double), cast($upperPercentile as double) ) as requestsPerMinUpperPercentile,
        getpercentileValue(cast(avg(total_request_count) as double), cast(sqrt(avg(cast(total_request_count
        as double)*cast(total_request_count as double))-avg(total_request_count)*avg(total_request_count)) as double) , cast($lowerPercentile as double)) as requestsPerMinLowerPercentile

        from REQUEST_INFO group by api_version, userId, consumerKey, resourceTemplate, method;

                 ]]>
                        </sparkScript>
                    </executionParameters>
                </template>
            </templates>

            <parameters>
                <parameter name="upperPercentile" type="double">
                    <displayName>Upper Percentile</displayName>
                    <description>Upper percentile value (between 0 and 1) to calculate requests per min upper percentile</description>
                    <defaultValue>0.95</defaultValue>
                </parameter>


                <parameter name="lowerPercentile" type="double">
                    <displayName>lowerPercentile</displayName>
                    <description>Lower percentile value (between 0 and 1) to calculate requests per min</description>
                    <defaultValue>0.05</defaultValue>
                </parameter>
            </parameters>
        </scenario>


        <scenario type="ResponsePerApiStatGenerator">
            <description>To create Response per api percentile values</description>
            <templates>
                <template type="batch">
                    <executionParameters>
                        <cron>0 0 23 * * ?</cron>
                        <sparkScript>
                            <![CDATA[
                CREATE TEMPORARY TABLE RESPONSE_PER_API_INFO USING CarbonAnalytics OPTIONS (tableName "org_wso2_analytics_apim_responsePerMinPerApiStream");

        CREATE TEMPORARY TABLE RESPONSE_PER_API_PERCENTILE_GEN USING CarbonAnalytics OPTIONS (tableName "ORG_WSO2_ANALYTICS_APIM_RESPONSEPERAPIPERCENTILE",
        schema "api_version string -i, tenantDomain string -i, responsesPerMinPerApiLowerPercentile double",
        primaryKeys "api_version, tenantDomain"
        );

        INSERT INTO TABLE RESPONSE_PER_API_PERCENTILE_GEN
        select api_version, tenantDomain, 
        getpercentileValue(cast(avg(responsesPerMinPerApi) as double), cast(sqrt(avg(cast(responsesPerMinPerApi
        as double)*cast(responsesPerMinPerApi as double))-avg(responsesPerMinPerApi)*avg(responsesPerMinPerApi)) as double), cast($lowerPercentile as double) ) as responsesPerMinPerApiLowerPercentile
        from RESPONSE_PER_API_INFO group by api_version, tenantDomain;

                 ]]>
                        </sparkScript>
                    </executionParameters>
                </template>
            </templates>

            <parameters>
                <parameter name="lowerPercentile" type="double">
                    <displayName>Lower percentile</displayName>
                    <description>Lower percentile value (between 0 and 1) to calculate responsesPerMinPerApiLowerPercentile</description>
                    <defaultValue>0.05</defaultValue>
                </parameter>
            </parameters>
        </scenario>


        <scenario type="ResponseTime">
            <description>To create response time percentile values</description>
            <templates>
                <template type="batch">
                    <executionParameters>
                        <cron>0 0 22 * * ?</cron>
                        <sparkScript>
                            <![CDATA[
                              CREATE TEMPORARY TABLE RESPONSE_INFO USING CarbonAnalytics OPTIONS (tableName "ORG_WSO2_APIMGT_STATISTICS_PERMINUTERESPONSE");

        CREATE TEMPORARY TABLE RESPONSE_PER_TIME_PERCENTILE_GEN USING CarbonAnalytics OPTIONS (tableName "ORG_WSO2_ANALYTICS_APIM_RESPONSETIMEPERAPIPERCENTILE",
        schema "api_version string -i, tenantDomain string -i,  responseTimePerApiUpperPercentile double ",
        primaryKeys "api_version, tenantDomain" , mergeSchema "false"
        );

        INSERT INTO TABLE RESPONSE_PER_TIME_PERCENTILE_GEN
        getpercentileValue(cast(avg(responseTime) as double), cast(sqrt(avg(cast(responseTime as double)*cast(responseTime as double))-avg(responseTime)*avg(responseTime)) as double),
        cast($upperPercentile as double) ) as responseTimePerApiUpperPercentile

        from RESPONSE_INFO group by api_version, tenantDomain;
                 ]]>
                        </sparkScript>
                    </executionParameters>
                </template>
            </templates>

            <parameters>
                <parameter name="upperPercentile" type="double">
                    <displayName>Upper percentile</displayName>
                    <description>Upper percentile value (between 0 and 1) to calculate responseTimePerApiUpperPercentile</description>
                    <defaultValue>0.95</defaultValue>
                </parameter>
            </parameters>
        </scenario>


        <scenario type="ResponseStatGenerator">
            <description>To create response percentile values</description>
            <templates>
                <template type="batch">
                    <executionParameters>
                        <cron>0 0 23 * * ?</cron>
                        <sparkScript>
                            <![CDATA[
                              CREATE TEMPORARY TABLE RESPONSE_INFO USING CarbonAnalytics OPTIONS (tableName
        "ORG_WSO2_APIMGT_STATISTICS_PERMINUTERESPONSE");


        CREATE TEMPORARY TABLE RESPONSE_PERCENTILE_GEN USING CarbonAnalytics OPTIONS (tableName "ORG_WSO2_ANALYTICS_APIM_RESPONSEPERCENTILE",
        schema "api_version string, tenantDomain string, resourceTemplate string, method string, responsePercentile double, backendPercentile double ",
        primaryKeys "api_version,tenantDomain,resourceTemplate,method"
        );


        INSERT OVERWRITE TABLE RESPONSE_PERCENTILE_GEN
        select api_version, tenantDomain, resourceTemplate, method,
        getpercentileValue(cast(avg(responseTime) as double), cast(sqrt(avg(cast(responseTime as double)*cast(responseTime as double))-avg(responseTime)*avg(responseTime)) as double), cast($upperPercentileRespondTime as double) ) as responsePercentile,
        getpercentileValue(cast(avg(backendTime) as double), cast(sqrt(avg(cast(backendTime as double)*cast(backendTime as double))-avg(backendTime)*avg(backendTime)) as double), cast($upperPercentilebackendTime as double) ) as backendPercentile
        from RESPONSE_INFO group by api_version, tenantDomain, resourceTemplate, method;

                 ]]>
                        </sparkScript>
                    </executionParameters>
                </template>
            </templates>


            <parameters>
                <parameter name="upperPercentileRespondTime" type="double">
                    <displayName>Upper percentile response time</displayName>
                    <description>Upper percentile value (between 0 and 1) to calculate responsePercentile</description>
                    <defaultValue>0.95</defaultValue>
                </parameter>
                <parameter name="upperPercentilebackendTime" type="double">
                    <displayName>Upper percentile backend time</displayName>
                    <description>Upper percentile value (between 0 and 1) to calculate backendPercentile</description>
                    <defaultValue>0.95</defaultValue>
                </parameter>
            </parameters>
        </scenario>

        <scenario type="APIM_GEO_LOCATION_STATS">
            <description>To generate Geo locations-based statistics</description>
            <templates>
                <template type="batch">
                    <executionParameters>
                        <cron>0 0 23 * * ?</cron>
                        <sparkScript>
                            <![CDATA[
                              CREATE TEMPORARY TABLE APIGeoLocationData USING CarbonJDBC OPTIONS (dataSource "WSO2AM_STATS_DB", tableName "API_REQ_GEO_LOC_SUMMARY",
                              schema "api STRING ,
                              version STRING ,
                              apiPublisher STRING ,
                              tenantDomain STRING ,
                              total_request_count INTEGER ,
                              year INTEGER ,
                              month INTEGER ,
                              day INTEGER ,
                              requestTime LONG ,
                              country STRING ,
                              city STRING ",
                              primaryKeys "api,version,apiPublisher,year,month,day,country,city,tenantDomain"
                              );
			      CREATE TEMPORARY TABLE APIRequestData USING CarbonAnalytics OPTIONS(tableName "ORG_WSO2_APIMGT_STATISTICS_PERMINUTEREQUEST", schema "
                                year INT -i, month INT -i, day INT -i, hour INT -i, minute INT -i,
                                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 BOOLEAN, 
				clientIp STRING,
                                applicationOwner STRING, 
				_timestamp LONG -i",
                                primaryKeys "year, month, day, hour, minute, consumerKey, context, api_version, userId, hostName, apiPublisher, resourceTemplate, method, userAgent, clientIp",
                                incrementalParams "GEO_APIMGT_PERMINUTE_REQUEST_DATA, DAY",
                                mergeSchema "false");
                                INSERT INTO TABLE APIGeoLocationData SELECT
                              api,version, apiPublisher,tenantDomain,
                                sum(total_request_count) as total_request_count,year,month,day,first(requestTime),getCountry(clientIp) as country,
                                getCity(clientIp) as city
                                from APIRequestData group by api,version,apiPublisher,year,month,day,clientIp,getCountry(clientIp),getCity(clientIp),tenantDomain;
                              INCREMENTAL_TABLE_COMMIT GEO_APIMGT_PERMINUTE_REQUEST_DATA;

                 ]]>
                        </sparkScript>

                    </executionParameters>
                </template>
            </templates>

            <parameters>
                <parameter name="dummyParameter" type="double">
                    <displayName>dummy parameter</displayName>
                    <description>This is a dummy parameter</description>
                    <defaultValue>0.95</defaultValue>
                </parameter>

            </parameters>

        </scenario>


    </scenarios>


    <streams>

    </streams>
</domain>
