wso2 / apim-analytics-publisher Goto Github PK

Description
Design and develop client code to connect to event hub and publish the given analytics events in a resilient manner. Following functional and non-functional requirements should be met

• Should publish events to EventHub in a resilient and fail safe manner. At least once QoS guarantee should be achieved
• Should accepts events though a non-blocking bounded queue to have minimal impact on the GW
• Separate parallel workers should consume the queue and publish to Event Hub using Event Hub Java SDK and Azure Schema Registry
• In an event of connection loss, once the queue is full should drop any newly arriving events till connection is re-established and queue is flushed.

Sub Tasks

• Improve metric reporter interface
• Implement queuing async publisher
• Add error handling
• Add schema validation

This will be tracked as a sub task in this issue.

Description:
In the current architecture, only a single EventHubProducerClient is sending events to eventhub. That introduce a publishing ceiling below the maximum TPS a gateway. Hence we need to change the architecture to allow multiple parallel producer clients while maintaining the same error handling requirements. This issue will be used to track that improvement

Description:
Current implementation publishes messages using the transport thread in sync manner. This is going to affect GW performance. Also we are publishing one event at a time which is not efficient. Hence we need to improve the publisher. This task is to track that progress.

Related Issues:
#8

Description:

In microgateway based scenario, the gateway should be able to publish the organization id to the analytics cloud. For that, it is required to have a new schema.

Suggested Labels:

Suggested Assignees:

Affected Product Version:
v1.0.0-ALPHA-2

OS, DB, other environment details and versions:

Steps to reproduce:

Related Issues:

Description:
Implement a HTTP client to connect to Auth API and retrieve SAS token. Reconnection service should make sure SAS tokens are refreshed before expiry. Below are the functional and non-functional requirements

• Should connect to auth API and retrieve token
• In an event of failure should retry in backoff fashion. This should not affect gateway startup
• Reconnection service should keep track of the expiry and refresh tokens so that publisher client will not encounter authentication issues
• All authentication issues should be handed over to reconnection service

Description:
Current handler impl initialize the MetricReporter only when a request is made. But we need it to be intialialized at the server start up to make sure that any problems with the configurations are noticed and handled before traffic reaches the instance.

Description:
Since SAS tokens are renewing automatically need to provide token expiry time correct to generate a new token.

Description:
Used an Invalid Token (token used to publish Analytics Events) while starting the Gateway. Invoked an API, which was working earlier. Got the following error while invoking the API.

com.azure.core.amqp.exception.AmqpException: status-code: 401, status-description: ExpiredToken: The token is expired. Expiration time: '2021-02-01 06:19:27Z', errorContext[NAMESPACE: analytics-dev-eventhub.servicebus.windows.net, PATH: $cbs, REFERENCE_ID: cbs:receiver, LINK_CREDIT: 0]

Received the following error as the response:

{"Fault":{"faultcode":"soapenv:Server","faultstring":"status-code: 401, status-description: ExpiredToken: The token is expired. Expiration time: '2021-02-01 06:19:27Z', errorContext[NAMESPACE: analytics-dev-eventhub.servicebus.windows.net, PATH: $cbs, REFERENCE_ID: cbs:receiver, LINK_CREDIT: 0]","detail":null}

Description
This issue is to track the progress of analytics publishing effort of APIM 4.0.0. Below is the high-level design diagram which will be updated in due course.

Below are the subtasks related to this high-level objective.

• Design and develop common metric reporter interface to accept metrics
• Design and develop client to Auth API
• Design and develop resilient event hub client
• Design and develop synapse instrumentation
• Design and develop Auth API

Publishing even schema can be found here

Description
APIM analytics metrics publisher needs to facilitate multiple analytics technologies. To achieve that implementation publisher implementation should be extensible. Common Metric Reporter interface will be introduced which can be extended by different technologies. Common interface should handle relevant configs and extracted metric information.
This will be tracked as a subtask of this issue.

Description:
This issue is to track unit test implementation for publisher component. As a part of this we need to find a suitable way to unit test publisher client. At the end of the exercise unit test coverage should be beyond 80%

Description:
In certain situations additional attributes need to be published from Gateways. Ability should be provided to publish custom attributes from client side.

• Currently createMetricReporter get class name as argument and use default if null pass
• Better to have another createMetricReporter() without class name as argument

ex:

public MetricReporter createMetricReporter(String fullyQualifiedClassName, Map<String, String> properties) throws MetricCreationException {
   this.createMetricReporter("org.wso2.am.analytics.publisher.reporter.ChoreoAnalyticsMetricReporter", properties);
}

Description:
Currently WSO2 Token Credential token expiry time is hard coded. Need to change that to extract the value from retrieved token and set

Description:
Error handling of the eventhub client should be implemented in a way that event loss should be minimal and impact on the Gateway is minimal. Following error scenarios will be handled and detailed description on how each scenario is handled mentioned in this doc.

• Scenario 01 - Network loss when starting up the instance
• Scenario 02 - Authentication failure when starting up the instance
• Scenario 03 - Intermittent connection issue
• Scenario 04 - Intermittent authentication issue due to authentication service down time
• Scenario 05 - Unexpected authentication issue due to user revocation(user token and publisher)
• Scenario 06 - Unexpected authentication issue due to publisher suspension(user token and publisher)

Description:
Currently event attributes are only validated by key name. But there has to be more sophisticated validation to make sure stale data doesn't make it up to processing system.

Description
The authentication server is responsible for authenticating agents and issue a new SAS token that can be used to publish events. Token generation information is kept inside a database for each deployment. Auth server authenticates the user with incoming JWT with a token endpoint. consumer key is taken from the JWT and extracts the relevant publisher name and the tokens need to generate the SAS token. Generated SAS token reply back to the client.

Description:
Current metric reporter accepts configs and events as a Map. This was done to ease the integration for different technologies that we will have in future. However this makes the usage of the API ambiguous as there is no control on what u can add to the map. Hence builders will be introduced to create more tighter control

Related Issues:
#7

Description:
Following error is thrown when the first request is made

[2021-02-17 17:56:58,533] ERROR - APIMgtResponseHandler Cannot publish response event. Index 1 out of bounds for length 1
java.lang.ArrayIndexOutOfBoundsException: Index 1 out of bounds for length 1
        at org.wso2.carbon.apimgt.gateway.handlers.analytics.APIMgtResponseHandler.mediate_aroundBody0(APIMgtResponseHandler.java:161) [org.wso2.carbon.apimgt.gateway_6.8.210.jar:?]
        at org.wso2.carbon.apimgt.gateway.handlers.analytics.APIMgtResponseHandler.mediate(APIMgtResponseHandler.java:63) [org.wso2.carbon.apimgt.gateway_6.8.210.jar:?]
        at org.apache.synapse.mediators.ext.ClassMediator.mediate(ClassMediator.java:94) [synapse-core_2.1.7.wso2v203.jar:2.1.7-wso2v203]
        at org.apache.synapse.mediators.AbstractListMediator.mediate(AbstractListMediator.java:109) [synapse-core_2.1.7.wso2v203.jar:2.1.7-wso2v203]
        at org.apache.synapse.mediators.AbstractListMediator.mediate(AbstractListMediator.java:71) [synapse-core_2.1.7.wso2v203.jar:2.1.7-wso2v203]
        at org.apache.synapse.mediators.base.SequenceMediator.mediate(SequenceMediator.java:158) [synapse-core_2.1.7.wso2v203.jar:2.1.7-wso2v203]
        at org.apache.synapse.api.Resource.process(Resource.java:342) [synapse-core_2.1.7.wso2v203.jar:2.1.7-wso2v203]
        at org.apache.synapse.api.API.process(API.java:404) [synapse-core_2.1.7.wso2v203.jar:2.1.7-wso2v203]
        at org.apache.synapse.api.AbstractApiHandler.apiProcessNonDefaultStrategy(AbstractApiHandler.java:107) [synapse-core_2.1.7.wso2v203.jar:2.1.7-wso2v203]
        at org.apache.synapse.api.AbstractApiHandler.identifyAPI(AbstractApiHandler.java:127) [synapse-core_2.1.7.wso2v203.jar:2.1.7-wso2v203]
        at org.apache.synapse.api.AbstractApiHandler.dispatchToAPI(AbstractApiHandler.java:54) [synapse-core_2.1.7.wso2v203.jar:2.1.7-wso2v203]
        at org.apache.synapse.api.rest.RestRequestHandler.dispatchToAPI(RestRequestHandler.java:75) [synapse-core_2.1.7.wso2v203.jar:2.1.7-wso2v203]
        at org.apache.synapse.api.rest.RestRequestHandler.process(RestRequestHandler.java:54) [synapse-core_2.1.7.wso2v203.jar:2.1.7-wso2v203]
        at org.apache.synapse.rest.RESTRequestHandler.process(RESTRequestHandler.java:54) [synapse-core_2.1.7.wso2v203.jar:2.1.7-wso2v203]
        at org.apache.synapse.core.axis2.Axis2SynapseEnvironment.injectMessage(Axis2SynapseEnvironment.java:344) [synapse-core_2.1.7.wso2v203.jar:2.1.7-wso2v203]
        at org.apache.synapse.core.axis2.SynapseCallbackReceiver.handleMessage(SynapseCallbackReceiver.java:607) [synapse-core_2.1.7.wso2v203.jar:2.1.7-wso2v203]
        at org.apache.synapse.core.axis2.SynapseCallbackReceiver.receive(SynapseCallbackReceiver.java:200) [synapse-core_2.1.7.wso2v203.jar:2.1.7-wso2v203]
        at org.apache.axis2.engine.AxisEngine.receive(AxisEngine.java:180) [axis2_1.6.1.wso2v41.jar:?]
        at org.apache.synapse.transport.passthru.ClientWorker.run(ClientWorker.java:290) [synapse-nhttp-transport_2.1.7.wso2v203.jar:?]
        at org.apache.axis2.transport.base.threads.NativeWorkerPool$1.run(NativeWorkerPool.java:172) [axis2_1.6.1.wso2v41.jar:?]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
        at java.lang.Thread.run(Thread.java:834) [?:?]

Description:
In the current impl we initialize ua parser for each event which is costly. Also the parse method of the ua parser takes up significant amount of memory. Below solutions are suggested to improve this.

• Use static variable or a singleton class to acquire ua parser
• Use a cache map to make minimise ua parsing

Description:
EventHubClient.flushEvent() does not contain error handling and retrying logic. Hence if an issue occurs when events are being flushed we might loose the events.