Skip to main content

Tap Stream Processing Guide

The Tap service unlocks the power of stream processing inside Morio without the complexity. It provides a low-code way to write your own stream processing logic and have Morio run it for you.

Some JavaScript required

With Morio, you can and write your own stream processor in JavaScript.

If you are not familiar with JavaScript, it is a beginner-friendly language and you do not need to master it to start creating valuable stream processing logic. Instead, you can start from our examples and build up from there, or look at the various stream processors on MorioHub for inspiration.

If you prefer another language, you can always handle these aspects yourself and talk directly to Kafka.

Creating your first stream processor​

To create a stream processor within the Tap service means to create a so-called stream processor object or .

It is a Javascript object that holds our stream processing logic, as well as a little bit of configuration. Below is a fully functional example:

A fully function stream processor object (SPO) for the Tap service
export default {
  id: 'my-first-stream-processor',
  info: "This example generates a note when 'error' is found in journald/cron logs",
  settings: {
    topics: ['logs'],
    modules: ['linux-system'],
    datasets: ['cron'],
  },
  handler: (params) => {
    const { data, tools } = params
    if (tools.get(data, 'message' '').toString().toLowerCase().includes('error')) {
      tools.cache.note('Error detected in cron logs', data)
    }
  }
}

It only took 15 lines of code to create our own . Of those, the stream processor logic itself take up 6 lines. That’s because the Tap service will take care of everything for us, we just have to tell it exactly what data we want.

Let’s go through these lines to clarify what we just did:

The default export​

The first thing to notice is that we are using a default export:

export default {
  // rest of the code
}

Your must be the default export.

See also

See File structure in the Tap Stream Processors reference documentation for more examples.

The processor ID​

The id property of an holds a string that is the ID of the stream processor.

  id: 'my-first-stream-processor',

Each should have an unique ID.

See also

Refer to the reference documentation to learn more about the importance of this ID, and in what scenario you may want to deliberately reuse the same ID.

Info about the processor​

The info property of the holds a string that is a description of the stream processor. It is intended for humans to clarify the role and purpose of the stream processor.

  info: "This example generates a note when 'error' is found in journald/cron logs",

The processor settings​

The settings property of the is where we configure the kind of data we want to receive:

  settings: {
    topics: ['logs'],
    modules: ['linux-system'],
    datasets: ['cron'],
  },

Think of this as a subscription with three levels of increasing granularity:

  • settings.topics holds a list of kafka topics to subscribe to. At least 1 topic is mandatory.
  • settings.modules holds a list of Morio modules to subscribe to. If not specified, we receive data from all modules.
  • settings.datasets holds a list of Morio datasets to subscribe to. If not specified, we receive data from all datasets.

In our example, we we receive all data that: Comes from the logs topic AND is generated by the linux-system module AND has the dataset cron.

In other words, our processor we will receive all cron logs.

See also

Your stream processor can also expose custom settings that can be controlled by the user. Refer to the reference documentation for all details.

The stream processor handler function​

The handler property of the holds the function that will receive the data to be processed.

  handler: (params) => {
    const { data, tools } = params
    if (tools.get(data, 'message' '').toString().toLowerCase().includes('error')) {
      tools.cache.note('Error detected in cron logs', data)
    }
  }

Specifically, each time a message matches what we configured in settings — in our case, when the topic is logs, the module is linux-system, and the dataset is cron — this function will be called and passed the data.

Destructuring params​

But our function gets passed more than just the data. You can see that we are using desctructuring to extract not only data but also tools from params:

    const { data, tools } = params
  • params.data: Holds the data read from Kafka, parsed as JSON
  • params.tools: Holds a bunch of helpers to facilitate stream processing
See also

There is more to params, and much more to params.tools, refer to the reference documentation on Tap params for all details.

Also, now that you know params exists and can be destructured, we can write this in a more elegant way by moving the destructuring to the function parameters:

  handler: ({ data, tools }) => {
    if (tools.get(data, 'message' '').toLowerCase().includes('error')) {
      tools.cache.note('Error detected in cron logs', data)
    }
  }

Writing robust detections​

Inside our function, our detection logic is really just this one line:

    if (tools.get(data, 'message' '').toString().toLowerCase().includes('error')) {

A more naïve approach would have been to write something like this:

    if (data.message.includes('error')) {

However, this assumes that data.message exists and is of type String. If it does not exist (is undefined) or perhaps holds and Object, then calling .toLowerCase() on it will cause an error.

When an error occurs in our handler function, Morio’s Tap service will catch it for us, and discard it. In other words, the Tap service will keep on going, but this particular message — as well as any other messages that cause an error — will not be processed.

Since we do not always know what kind of data, we are going to process, we need to write robust detection logic that does not make assumptions. Which is why we end up with:

    if (tools.get(data, 'message' '').toString().toLowerCase().includes('error')) {

This guards against:

  • data.message being undefined, by using tools.get which provides lodash.get.
  • Ensures we work with a string by calling .toString()
  • Ensures we match Error, error, and ERROR by calling .toLowerCase()
  • Before finally checking whether it includes the string error
tip

When writing detection logic, keep in mind to not make assumptions about the data you are processing.

Take action​

Once our detection logic is triggered, we are going to take action. In this case, we are creating :

      tools.cache.note('Error detected in cron logs', data)

What action to take depends on your use-case. Here we have created a note, which exist to avoid logs snowballing.

Embrace the event-driven way

Our recommendation is to resist the urge to do something complex, and instead create an event. Then, let a different stream processor read from the events topic, and take further action.

Doing so avoids tight-coupling of your detection logic with remediation. This sort of event-driven automation is both flexible and powerful, but requires some getting used to.

Beware of the logging snowball effect​

One thing to avoid inside a stream processor function is to log data. Especially when you are processing log data with that stream processor:

  • Imagine you have a stream processor that is processing log data.
  • When that stream processor logs themselves, those logs will be picked up by the local Morio client running on the Morio collected, and will end up in the logs topic
  • From there they will be picked up by our stream processor, which will generate more logs… and round and round we go!

This creates a feedback loop, or snowball effect, where the logs generated by the stream processor get picked up again by the same stream processor, ever-increasing the amount of log data flowing through the system.

For this reason, use to log data inside the stream processor. This is a parallel logging channel that adds messages to the Morio cache which are made available through the UI and API, thus avoiding this feedback loop.