Flow / Flow use cases / Customized alert reports

Customized alert reports

[ Description ] [ Flow configuration ] [ Result ] [ Import this Flow ]

Description

A Flow that pulls an alert count from 2 different time periods, combines and transforms it into a specific format, and sends a report with the results via email to the required recipients.

This Flow checks the number of alerts triggered in the last hour and compares them with the alerts received over the same day and hour of the previous week. The info comparing both alert counts is sent via email every x minutes.

We've defined 2 different Flows for this use case. The final result is the same, but each works with different sequences and units. You can check the other version of this Flow by clicking the Version #2 tab at the top of this article.

Flow configuration

We've divided the configuration of this Flow into 2 different parts:

Units - See the required units for this Flow and how to configure them.
Links - Check how to link the units of this Flow to define the required sequence.

Units

The following table describes the units needed to create this Flow, and how to configure them.

Unit type Description Configuration

Generator

This unit will fire a signal event every x minutes (in this example, every minute) to activate the Flow.

We will enrich these events using the following units and will use them to activate the required queries.

Drag a Generator unit to the canvas and open its configuration options by double-clicking it.
On the General tab, add a Name to the unit (in this example, we called it Clock), and specify the Time field name that will contain the output event times (in this example, eventdate).
Then, on the Period tab, enter 60000 in the Millis field. Also, switch on the Exact, Aligned, and Drop past periods toggles.
Click Apply to save the configuration.

Map

We will use a couple of Map units to enrich the events with time information.

As said above, we will be comparing alerts triggered the last hour with alerts triggered over the same day and hour of the previous week. To do it, we will use these Map units to specify the start and end date of each period.

Drag 2 Map units to the canvas and follow the steps below to configure them:

Open the configuration options of the first Map unit by double-clicking it.
On the General tab, add a Name to the unit (in this example, we called it prepareQueryTimeInterval), and leave the Language field as default.

Then, on the Fields to add tab, you must add 3 different fields by clicking the + icon:

This column will contain the end date of the period to be analyzed. In this case, we need the current time as end date.

Field name - endDate
Type - Long
Expression - eventdate.getTime()

This column will contain the start date of the period to be analyzed. In this case, we need to get 1 hour before the current time.

Field name - startDate
Type - Long
Expression - endDate - java.time.Duration.ofHours(1).toMillis()

We will use this column to identify the time period of the events.

Field name - header
Type - String
Expression - "Today"

Click Apply to save the configuration.

Open the configuration options of the second Map unit by double-clicking it.
On the General tab, add a Name to the unit (in this example, we called it preparePastQueryTimeInterval), and leave the Language field as default.

Then, on the Fields to add tab, you must add 3 different fields by clicking the + icon:

This column will contain the end date of the period to be analyzed. In this case, we need the current time of the previous week as end date.

Field name - endDate
Type - Long
Expression - eventdate.getTime() - java.time.Duration.ofDays(7).toMillis()

This column will contain the start date of the period to be analyzed. In this case, we need to get 1 hour before the current time of the previous week.

Field name - startDate
Type - Long
Expression - endDate - java.time.Duration.ofHours(1).toMillis()

We will use this column to identify the time period of the events.

Field name - header
Type - String
Expression - "Last week"

Click Apply to save the configuration.

Devo Full Query

We will use a Devo Full Query unit to specify the query that defines the alerts to be compared.

We will be querying the siem.logtrust.alert.info table, which receives all the alerts triggered in your domain.

Drag a Devo Full Query unit to the canvas and configure it as follows:

Open the configuration options of the Devo Full Query unit by double-clicking it.
On the General tab, give the unit a Name (in this example, we called it alertTriggeredQuery) and enter the following in the Query field:
```
from siem.logtrust.alert.info
select eventdate, context
group by context
select first(eventdate) as eventdate, count() as count
```
Then, on the From event tab, choose the startDate and endDate columns in the Start time and End time fields.
Click Apply to save the configuration.

Reducer

We will use a Reducer unit to format all the events received during each of the query periods set.

Once this unit has stored all the alert info of a period, it will be sent to an additional Reducer unit, which will combine it with the info of the other period and generate a report.

Drag a Reducer unit to the canvas and configure it as follows:

Open the configuration options of the Reducer unit by double-clicking it.
Give the unit a Name (in this example, we called it prepareMessage).
Leave the Language as default (Groovy) and enter two double quotation marks "" in the Init Value field.
Set the Field name as message and choose String as Accumulator Type.

Then, enter the following in the Expression field. This expression will format all the events received.

__acc__ + 
"<tr> <td>"+new Date(startDate)+"</td>"+
"<td>"+context+"</td>"+
"<td>"+count+"</td>"+
"</tr>"

Click Apply to save the configuration.

Reducer

We will use this Reducer to combine the information formatted by the previously added Reducer unit and send the final report with the alert info from both time periods.

Drag a Reducer unit to the canvas and configure it as follows:

Open the configuration options of this Reducer unit by double-clicking it.
Give the unit a Name (in this example, we called it combinedMessage).
Leave the Language as default (Groovy) and enter two double quotation marks "" in the Init Value field.
Set the Field name as combinedMessage and choose String as Accumulator Type.

Then, enter the following in the Expression field:

__acc__ + 
"<tr> <td></td>"+
"<td>"+header+"</td>"+
"<td></td>"+
"</tr>" + message

Email Sink

We will use this unit to send the reports generated to the required user emails.

Drag an Email Sink unit to the canvas and configure it as follows:

Open the configuration options of the unit by double-clicking it. In the General tab, add a Name to the unit (in this example, we called it notify), add a Subject for the emails to be sent (in this example, Alert Report).
Then, in the Message field, you must enter the text to be sent in the mails. For this example, we added the following (where combinedMessage is a variable field that will be automatically replaced by the corresponding values)
```
<table>
  <tr>
    <th>Date</th>
    <th>Alert</th>
    <th>Count</th>
    %%{combinedMessage}
  </tr>
</table>
```
Then, click the To recipients tab and add as many email addresses as required by clicking the + icon.
Click Apply to save the configuration.

Links

Once you have added and configured all the units of the Flow, you must link them as follows:

Step 1
First, you must link the out port of the Generator unit to the reset port of the second Reducer unit (combineMessages). This first step is required to reset the final Reducer every time the Flow is activated. This unit will store and combine the data to be sent in the report, so we need to reset it before the next report is generated.
Step 2
Then, we must define the links to feed the final Reducer unit with the alert information for the first period of time (last 24 hours). To do it, create the following links: Link the out port of the Generator unit to the in port of the first Map unit (prepareQueryTimeInterval). Events will be enriched with the fields defined in this Map unit to indicate the time period. Then, link the out port of the prepareQueryTimeInterval unit to the in port of the Devo Full Query unit. This will trigger the query defined in the unit. Now, we must define the links that will feed the first Reducer unit with the alert information of the first period: Link the init port of the Devo Full Query unit to the reset port of the Reducer unit. This will reset the Reducer unit before storing the information. Link the data port of the Devo Full Query unit to the in port of the Reducer unit. Query events will be sent to the Reducer through this link. Link the end port of the Devo Full Query unit to the get port of the Reducer unit. The Reducer will evaluate the expression defined in the unit and will emit the formatted info through its current port once the query ends. Finally, link the current port of the first Reducer unit (prepareMessage) to the in port of the second Reducer unit (combineMessages). The alert information prepared by the first Reducer will be stored here until the data of the second time period arrives.
Step 3
Now, we must define the links to feed the final Reducer unit with the alert information for the second period of time (same day and hour but a week before). To do it, create the following links: Link the out port of the Generator unit to the in port of the second Map unit (preparePastQueryTimeInterval). Events will be enriched with the fields defined in this Map unit to indicate the time period. Then, link the out port of the preparePastQueryTimeInterval unit to the in port of the Devo Full Query unit. This will trigger the query defined in the unit. The alert information will be formatted by the first Reducer unit following the flow defined in step 3 of the previous section (the Reducer is reset, then fed with the alert information of this period and finally sent via the current port when the query ends). Finally, the information will be sent to the second Reducer unit (combineMessages) through the link defined on the previous section. The alert information of both time periods is now stored in the final Reducer.
Step 4
Finally, when the second Reducer has the information from both time periods, it will be sent via email to the required recipients. To do it: Link the out port of the Generator unit to the get port of the second Reducer unit (combineMessages). This will make the Reducer unit emmit the final report through its current port. Then, link the current port of the Reducer to the in port of the Devo Sink Mail unit. The report will be sent to the recipients indicated in this unit.

Result

Once you're done, remember to save your Flow so you can start running it. To do it, click the Save button on the Flow toolbar.

Once you have defined the whole Flow and saved it, click the Start button to activate it. If everything is correctly configured, the Flow will send an email to the given addresses with a report comparing the alert count over the last hour with the alert count over the same hour and day of the previous week.

Import this Flow

Download this Flow in JSON format by clicking here and import it to your domain clicking the Import option at the top of the Flow canvas. If the JSON object opens on a new browser window automatically, copy all the content, paste it into your notepad and save it in .txt or .json format.

Remember to add the required target email addresses as explained above before running the Flow.

[ Description ] [ Flow configuration ] [ Result ] [ Import this Flow ]

Description

A Flow that pulls an alert count from 2 different time periods, combines and transforms it into a specific format, and sends a report with the results via email to the required recipients.

The top branch counts the alerts triggered over the last hour.
The bottom branch counts the alerts triggered over the same hour, seven days previously.

Execution order in Flow

In Flows with different branches like this one, note that branches are always executed in a specific sequential order. You can check the order by hovering over the links that go out from a specific port and checking the number that appears. In this specific Flow, we need the top branch to be executed before the other one.

To change the execution order, click a link, select the arrows icon that appears and click the top or bottom arrow button to set the order of that link higher or lower. Learn more about this in Working with links.

Flow configuration

The following table describes the units needed to create this Flow, as well as how to configure and link them.

Unit type Description Configuration

Generator

This unit will fire an event every x minutes (in this example, every minute).

We will enrich these events using the following units and will use them to activate the required queries.

Drag a Generator unit to the canvas and open its configuration options by double-clicking it.
On the General tab, add a Name to the unit (in this example, we called it Clock), and specify the Time field name that will contain the output event times (in this example, eventdate).
Then, on the Period tab, enter 60000 in the Millis field. Also, switch on the Exact, Aligned, and Drop past periods toggles.
Click Apply to save the configuration.

Map

We will use a couple of Map units to enrich the events with time information.

Drag 2 Map units to the canvas and link their in ports to the out port of the Generator unit.

As mentioned in the introduction of this article, Flows follow a sequential order. In this case, we want the top branch of the Flow to activate before the bottom want. To do this, you must first link the Generator unit to the top Map (prepareQueryTimeInterval), and then to the bottom one (preparePastQuertTimeInterval).

As said above, you can always change the execution order by clicking a link, selecting the arrows icon that appears, and clicking the top or bottom arrow button to set the order of that link higher or lower. Learn more about this in Working with links.

Now follow the steps below to configure both units:

Open the configuration options of the first Map unit by double-clicking it.
On the General tab, add a Name to the unit (in this example, we called it prepareQueryTimeInterval), and leave the Language field as default.

Then, on the Fields to add tab, you must add 3 different fields by clicking the + icon:

This column will contain the end date of the period to be analyzed. In this case, we need the current time as end date.

Field name - endDate
Type - Long
Expression - eventdate.getTime()

This column will contain the start date of the period to be analyzed. In this case, we need to get 1 hour before the current time.

Field name - startDate
Type - Long
Expression - endDate - java.time.Duration.ofHours(1).toMillis()

We will use this column to identify the time period of the events.

Field name - header
Type - String
Expression - "Today"

Click Apply to save the configuration.

Open the configuration options of the second Map unit by double-clicking it.
On the General tab, add a Name to the unit (in this example, we called it preparePastQueryTimeInterval), and leave the Language field as default.

Then, on the Fields to add tab, you must add 3 different fields by clicking the + icon:

This column will contain the end date of the period to be analyzed. In this case, we need the current time of the previous week as end date.

Field name - endDate
Type - Long
Expression - eventdate.getTime() - java.time.Duration.ofDays(7).toMillis()

This column will contain the start date of the period to be analyzed. In this case, we need to get 1 hour before the current time of the previous week.

Field name - startDate
Type - Long
Expression - endDate - java.time.Duration.ofHours(1).toMillis()

We will use this column to identify the time period of the events.

Field name - header
Type - String
Expression - "Last week"

Click Apply to save the configuration.

Devo Full Query

We will use a couple of Devo Full Query units to specify the query that defines the alerts to be compared.

We will be querying the siem.logtrust.alert.info table, which receives all the alerts triggered in your domain.

Drag 2 Devo Full Query units to the canvas and link their in ports to the out ports of the Map units, as follows:

Now follow the steps below to configure both units:

Open the configuration options of the first Devo Full Query unit by double-clicking it.
On the General tab, give the unit a Name (in this example, we called it alertTriggeredQuery) and enter the following in the Query field:
```
from siem.logtrust.alert.info
select eventdate, context
group by context
select first(eventdate) as eventdate, count() as count
```
Then, on the From event tab, choose the startDate and endDate columns in the Start time and End time fields.
Click Apply to save the configuration.

Open the configuration options of the second Devo Full Query unit by double-clicking it.
On the General tab, give the unit a Name (in this example, we called it alertTriggeredPastQuery) and enter the following in the Query field:
```
from siem.logtrust.alert.info
select eventdate, context
group by context
select first(eventdate) as eventdate, count() as count
```
Then, on the From event tab, choose the startDate and endDate columns in the Start time and End time fields.
Click Apply to save the configuration.

Reducer

We will use a couple of Reducer units to store all the events received during the query periods set, combine and format them and finally emit an only event with all the alerts triggered during those periods.

Drag 2 Reducer units to the canvas and link them to the Devo Full Query units as follows:

Link the init port of the Devo Full Query units to the reset port of the Reducer units. This way, each time a query is started, the Reducer units will be reset with the value specified in the Init Value of the unit. We will set this initial value as empty.
Link the data port of the Devo Full Query units to the in port of the Reducer units. The expression set in the Reducer units will be evaluated and stored with the corresponding query values. We will use an HTML expression to format the resulting reports.
Link the end port of the Devo Full Query units to the get port of the Reducer units. When the query ends, a signal event will be sent to the get port of the Reducer units. This will make the Reducer units emit an only event with all the stored information through the output current ports.

Now follow the steps below to configure both units:

Open the configuration options of the first Reducer unit by double-clicking it.
Give the unit a Name (in this example, we called it prepareMessage).
Leave the Language as default (Groovy) and enter "" in the Init Value field.
Set the Field name as message and choose String as Accumulator Type.

Then, enter the following in the Expression field:

__acc__ + 
"<tr> <td>"+new Date(startDate)+"</td>"+
"<td>"+context+"</td>"+
"<td>"+count+"</td>"+
"</tr>"

Click Apply to save the configuration.

Open the configuration options of the second Reducer unit by double-clicking it.
Give the unit a Name (in this example, we called it preparePastMessage).
Leave the Language as default (Groovy) and enter "" in the Init Value field.
Set the Field name as message and choose String as Accumulator Type.

Then, enter the following in the Expression field:

__acc__ + 
"<tr> <td>"+new Date(startDate)+"</td>"+
"<td>"+context+"</td>"+
"<td>"+count+"</td>"+
"</tr>"

Click Apply to save the configuration.

Reducer

We will use an additional Reducer to combine the information stored by the previously added Reducer units and send the final report with the alert info from both time periods.

Drag a Reducer unit to the canvas and link it to the previously added Reducers in the following order:

First, link the current port of the first Reducer unit (prepareMessage) to the reset port of the new Reducer unit. This will reset this Reducer unit each time information goes out from the first Reducer, which is the first one in our sequence.
Then, link the current port of the first Reducer unit (prepareMessage) to the in port of the new Reducer unit. The Reducer will store the information that gets through the in port.
Now, link the current port of the second Reducer unit (preparePastMessage) to the in port of the new Reducer unit. This will add the information of the unit to the new Reducer

Finally, link the current port of the second Reducer unit (preparePastMessage) to the get port of the new Reducer. This will make the new Reducer emmit the report with all the information stored. The final report will go out through the current port of this Reducer.

Execution order

As said in the introduction of this article, remember that the link order matters. You must link the ports in the order described above.

This is the sequence order of this part of the Flow:

(1) The combineMessages Reducer is reset.

(2) The combineMessages Reducer gets the information stored by the prepareMessage Reducer.

(3) The combineMessages Reducer gets the information stored by the preparePasrMessage Reducer.

(4) The combineMessages Reducer is notified to emit the report with all the information stored.

Then, open the configuration options of the unit by double-clicking it. Add a Name to the unit (in this example, we called it combineMessages).
Leave the Language as default (Groovy) and enter "" in the Init Value field.
Set the Field name as combinedMessage and choose String as Accumulator Type.

Then, enter the following in the Expression field:

__acc__ + 

"<tr> <td></td>"+
"<td>"+header+"</td>"+
"<td></td>"+
"</tr>" + message

Click Apply to save the configuration.

Email Sink

We will use this unit to send the reports generated to the required user emails.

Drag an Email Sink unit to the canvas and link the current port of the Reducer unit to the in port of this unit, as follows:
Then, open the configuration options of the unit by double-clicking it. In the General tab, add a Name to the unit (in this example, we called it notify), add a Subject for the emails to be sent (in this example, Alert Report).
Then, in the Message field, you must enter the text to be sent in the mails. For this example, we added the following (where combinedMessage is a variable field that will be automatically replaced by the corresponding values)
```
<table>
  <tr>
    <th>Date</th>
    <th>Alert</th>
    <th>Count</th>
    %%{combinedMessage}
  </tr>
</table>
```
Then, click the To recipients tab and add as many email addresses as required by clicking the + icon.
Click Apply to save the configuration.

Once you're done, remember to save your Flow so you can start running it. To do it, click the Save button on the Flow toolbar.

Result

Import this Flow

Remember to add the required target email addresses as explained above before running the Flow.

See the following video to learn more about version #1 of this Flow:

Download as PDF