Overview

Monitoring data integration flows in real-time is crucial for ensuring optimal performance, troubleshooting issues, and gaining insights into the operations. Etlworks allows you to collect statistics and metrics from running flows and send this data to third-party tools like Grafana for visualization and analysis. This guide explains how to set up real-time data collection and send metrics to third party tool such as Grafana.

Prerequisites

1. Etlworks instance: Ensure you have the Etlworks Instance set up and running.
2. Data collection tool which supports ingesting data using API : Ensure you have a data collection tool, for example Grafana installed and accessible.
3. Webhook+Listener: Use a webhook and listener to capture real-time data.

How it works

The listener is configured in the Etlworks instance to receive a payload sent by the webhook. This webhook is triggered whenever an Etlworks flow is executed, either by the scheduler, manually, by API call or by Integration Agent.

When the listener receives the payload, it triggers a flow that transforms the payload and sends it to a third-party tool, such as Grafana, via an API call.

Step-by-Step Guide

Create listener, webhook,  connections and formats

Step 1. Create a new listener.

Set parameters:

• URL patten: /flow-exec-webhook 
• METHOD: POST
• Auth Type: Basic

Keep all other values unchanged. 

Step 2. Create new webhook.

Set parameters:

• URL: the same URL is auto-generated by the Listener.

You can copy the full listener URL and paste it in the Webhook URL.

• Events: Executed

Configure Basic authentication. You can use credentials for any active user with super admin, admin, executor or API user roles. 

Keep all other values unchanged. We will return back to this screen later.

Step 3. Create new Memory connection. Don't change any parameters

Step 4. Create new CLOB format. Don't change any settings.

Step 5. Create new JSON format. For now keep all the default settings. We will get back to this format later.

Step 6. Create a new HTTP connection which will be sending data to third-party data collection tool, such as Grafana.

For testing purposes we will create a dummy connection to send data to https://webhook.site/. later, we will modify this connection to send data to a specific third party data collection tool.

Enter parameters:

URL: the unique webhook.site URL

Method: POST

Content Type Header: text/plain (we will modify it later)

Payload: {grafana_payload}. You can use any name enclosed in {}.

Create flow which transforms the payload sent by webhook

Step 7. Create new web service to file flow.

Step 8. Add new source to destination transformation where:  

• SOURCE CONNECTION - Listener created in step 1.
• SOURCE FORMAT - JSON format created in step 5.
• DESTINATION CONNECTION - Memory connection created in step 3.
• DESTINATION FORMAT: CLOB format created in step 4.
• TO - just enter word temp.

Save the flow.

Create flow which sends data to data collection tool, such a Grafana

Step 9. Create new Call HTTP endpoint flow.

Step 10. Select HTTP connection created in step 6.

Save the flow.

Create flow which transforms the payload and sends it to data collection tool

Step 11. Create new nested flow.

Step 12. Add flows web service to file and Call HTTP endpoint added in steps 7 and 8 (in that order) to the nested flow.

Save the nested flow.

We will now need to create a transformation which modifies the payload sent by a webhook to a payload required by a third party data collection tool. We will use Grafana as an example.

Create a code transformation which transforms the payload to Grafana compatible

Step 13. Open JSON format created in step 5 and navigate to Preprocessor field.

Enter the following code:

var props = com.toolsverse.config.SystemConfig.instance().getProperties();

// payload received from webhook
var audit = JSON.parse(message).entity;

var errorDetail = '';

if (audit.exception) {
    // transform the exception
    errorDetail = audit.exception.replace(/\r?\n|\r/g, " ").replaceAll("=", "="); 
}

var task = audit.eventName ? audit.eventName : (audit.agentName ? audit.agentName : audit.flowName);

var grafanaString = 'Task=' + task 
+ ' Pipeline=' + audit.flowName 
+ ' RunId=' + audit.auditId 
+ ' StartTimeMs=' + audit.started 
+ ' StopTimeMS=' + audit.ended 
+ ' Duration=' + audit.duration 
+ ' State=' + audit.status 
+ ' ErrorDetail=' + errorDetail 
+ ' ErrorComplete=' + errorDetail;

var payload = {
 streams: [
 {
     stream: {
        label: "value",
        service: "ETLworks",
        team: "integrations",
        suite: "suite",
        department: "department",
        environment: "production",
        customer: "customer",
        task: task},
 values: [
     [(audit.started*1000000).toString(), grafanaString]
 ]}
]};


// sends the value of the global variable grafana_payload. 
// This will be sent to the Grafana
props.put('grafana_payload', JSON.stringify(payload));

// return the stringifyed payload
value = JSON.stringify(payload);

This is just an example of creating a payload which is compatible with Grafana. Modify if needed.

Modify the webhook to exclude the flow which sends data to third-party tool

By default the webhook created in step 2 will be triggered for all flows, including the flow created in step 11 which transforms and sends data to the third party collection tool which will cause a recursion: running flow created in step 11 triggers the webhook which triggers the flow created in step 11.

Step 13. To avoid the recursion navigate to the webhook created in step 2 and add flow created in step 11 to the Exclude Flows list. 

If you want to trigger data collection for specific flows only modify the Include Flows list.

Schedule the flow which transforms the payload and sends to data collection tool

The real-time event-driven flow triggered by the listener must be scheduled.

Step 14. Click the Schedule button next to the nested flow created in step 11.

Click Add schedule.

Keep all the default settings.

Click Save.

In about one minute the steady green indicator will appear next to the flow name. It means that the listener is online and ready to receive calls. 

The integration is now up and running.

Test integration

To test the integration simply execute any flow. Check the payload received by https://webhook.site/. Here is an example:

Configure an integration with an actual data collection tool

Once you are happy with a payload modify the HTTP connection created in step 6 to send the payload to the actual data collection tool. Here is an example for Grafana: