@App:name("telemetry-app")
@App:description("Description of the plan")

-- Please refer to https://docs.wso2.com/display/SP400/Quick+Start+Guide on getting started with SP editor.

@source(type='telemetry-receiver', @map(type='keyvalue', fail.on.missing.attribute='false', @attributes(source_uid = 'source.uid' , source_ip = 'source.ip' , source_labels = 'source.labels' , source_name = 'source.name' , source_namespace = 'source.namespace' , source_principal = 'source.principal' , source_owner = 'source.owner' , source_workload_uid = 'source.workload.uid' , source_workload_name = 'source.workload.name' , source_workload_namespace = 'source.workload.namespace' , destination_uid = 'destination.uid' , destination_ip = 'destination.ip' , destination_port = 'destination.port' , destination_labels = 'destination.labels' , destination_name = 'destination.name' , destination_namespace = 'destination.namespace' , destination_principal = 'destination.principal' , destination_owner = 'destination.owner' , destination_workload_uid = 'destination.workload.uid' , destination_workload_name = 'destination.workload.name' , destination_workload_namespace = 'destination.workload.namespace' , destination_container_name = 'destination.container.name' , destination_container_image = 'destination.container.image' , destination_service_host = 'destination.service.host' , destination_service_uid = 'destination.service.uid' , destination_service_name = 'destination.service.name' , destination_service_namespace = 'destination.service.namespace' , request_headers = 'request.headers' , request_id = 'request.index' , request_path = 'request.path' , request_host = 'request.host' , request_method = 'request.method' , request_reason = 'request.reason' , request_referer = 'request.referer' , request_scheme = 'request.scheme' , request_size = 'request.size' , request_total_size = 'request.total_size' , request_time_sec = 'request.time_sec' , request_time_nanosec = 'request.time_nanosec' , request_useragent = 'request.useragent' , response_headers = 'response.headers' , response_size = 'response.size' , response_total_size = 'response.total_size' , response_time_sec = 'response.time_sec' , response_time_nanosec = 'response.time_nanosec' , response_duration_sec = 'response.duration_sec' , response_duration_nanosec = 'response.duration_nanosec' , response_code = 'response.code' , response_grpc_status = 'response.grpc_status' , response_grpc_message = 'response.grpc_message' , connection_id = 'connection.index' , connection_event = 'connection.event' , connection_received_bytes = 'connection.received.bytes' , connection_received_bytes_total = 'connection.received.bytes_total' , connection_sent_bytes = 'connection.sent.bytes' , connection_sent_bytes_total = 'connection.sent.bytes_total' , connection_duration_sec = 'connection.duration_sec' , connection_duration_nanosec = 'connection.duration_nanosec' , connection_mtls = 'connection.mtls' , connection_requested_server_name = 'connection.requested_server_name' , context_protocol = 'context.protocol' , context_time_sec = 'context.time_sec' , context_time_nanosec = 'context.time_nanosec' , context_reporter_kind = 'context.reporter.kind' , context_reporter_uid = 'context.reporter.uid' , api_service = 'api.service' , api_version = 'api.version' , api_operation = 'api.operation' , api_protocol = 'api.protocol' , request_auth_principal = 'request.auth.principal' , request_auth_audiences = 'request.auth.audiences' , request_auth_presenter = 'request.auth.presenter' , request_auth_claims = 'request.auth.claims' , request_api_key = 'request.api_key' , check_error_code = 'check.error_code' , check_error_message = 'check.error_message' , check_cache_hit = 'check.cache_hit' , quota_cache_hit = 'quota.cache_hit')))
-- @sink(type='log', prefix='LOGGER')
define stream TelemetryStream (source_uid string, source_ip string, source_labels string, source_name string, source_namespace string, source_principal string, source_owner string, source_workload_uid string, source_workload_name string, source_workload_namespace string, destination_uid string, destination_ip string, destination_port long, destination_labels string, destination_name string, destination_namespace string, destination_principal string, destination_owner string, destination_workload_uid string, destination_workload_name string, destination_workload_namespace string, destination_container_name string, destination_container_image string, destination_service_host string, destination_service_uid string, destination_service_name string, destination_service_namespace string, request_headers string, request_id string, request_path string, request_host string, request_method string, request_reason string, request_referer string, request_scheme string, request_size long, request_total_size long, request_time_sec long, request_time_nanosec int, request_useragent string, response_headers string, response_size long, response_total_size long, response_time_sec long, response_time_nanosec int, response_duration_sec long, response_duration_nanosec int, response_code long, response_grpc_status string, response_grpc_message string, connection_id string, connection_event string, connection_received_bytes long, connection_received_bytes_total long, connection_sent_bytes long, connection_sent_bytes_total long, connection_duration_sec long, connection_duration_nanosec int, connection_mtls bool, connection_requested_server_name string, context_protocol string, context_time_sec long, context_time_nanosec int, context_reporter_kind string, context_reporter_uid string, api_service string, api_version string, api_operation string, api_protocol string, request_auth_principal string, request_auth_audiences string, request_auth_presenter string, request_auth_claims string, request_api_key string, check_error_code long, check_error_message string, check_cache_hit bool, quota_cache_hit bool);


-- @sink(type='log', prefix='SUMMARY')
-- define stream SummaryStream (request_host string, request_method string, request_path string, secCount long);

-- from TelemetryStream#window.timeBatch(10 sec) select request_host, request_method, request_path, count() as secCount
-- group by request_host, request_method, request_path
-- insert into SummaryStream;

@sink(type='log', prefix='SUMMARY')
define stream RequestsStream(timestamp long, serverName string, serviceName string, serviceMethod string, responseTime double, httpResponseCode int, userAgent string, requestIP string);

define stream ProcessedRequestsStream(timestamp long, serverName string, serviceName string, serviceMethod string, responseTime double, httpRespGroup string, userAgent string,
requestIP string);

@store(type="rdbms",datasource="HTTP_ANALYTICS_DB")
define table ServiceTable( serverName string, serviceName string);

@store(type="rdbms",datasource="HTTP_ANALYTICS_DB")
define aggregation RequestAggregation from ProcessedRequestsStream
select serverName, serviceName, serviceMethod, httpRespGroup, count() as numRequests, avg(responseTime) as avgRespTime
group by serverName, serviceName, serviceMethod, httpRespGroup
aggregate   every sec...year;

@info(name='ConvertToRequestStream')
from TelemetryStream [context_reporter_kind  == "inbound"] select request_time_sec as timestamp, destination_uid as serverName, destination_service_name as serviceName, request_method as serviceMethod, convert((response_duration_nanosec/1000000), 'double') as responseTime, cast(response_code, 'int') as httpResponseCode,
request_useragent as userAgent, source_uid as requestIP
insert into RequestsStream;


@info(name='FilterUniqueServerServiceQuery')
from RequestsStream#window.unique:first(serverName,serviceName)
select serverName, serviceName
insert into ServiceTable;

@info(name='HTTPResponseGroupProcessQuery')
from RequestsStream
select timestamp, serverName, serviceName, serviceMethod, cast(responseTime, 'double') as responseTime, str:concat(convert(httpResponseCode/100, "string"), 'xx') as httpRespGroup, userAgent, requestIP
insert into ProcessedRequestsStream;
