Tap params

A tap includes a handler function that will do the actual stream processing. Its signature looks like this:

Example stream processing handler function
function(params) {
  // do something here
}

This page lists everything included in params, in other words, everything that is passed into your handler function that you can work with.

`params.data`

Holds the data of the current message (the value as received by the consumer), parsed as JSON.

params.data
function(params) {
  const { data } = params
}

`params.dataset`

Holds the Morio dataset name, if Tap was able to extract it.

params.dataset
function(params) {
  const { dataset } = params
}

`params.iid`

Holds the Morio Inventory ID () of the host who generated the data, if Tap was able to extract it.

params.iid
function(params) {
  const { iid } = params
}

`params.module`

Holds the Morio module name, if Tap was able to extract it.

params.module
function(params) {
  const { module } = params
}

`params.node`

Holds information on the node the Tap service is running on:

params.node.fqdn: The node’s fully qualified domain name
params.node.hostname: The node’s host name
params.node.ip: The node’s IP address
params.node.serial: The Morio node serial
params.node.settings: The Morio settings serial
params.node.cluster: The Morio cluster ID the node is part of
params.node.reaper: Is true if this node should run the tap reaper

`params.offset`

Rarely used

Holds the partition offset of the current message (the offset as received by the consumer).

`params.partition`

Rarely used

Holds the partition of the current message (the partition as received by the consumer).

`params.processor`

Rarely used

Holds the itself. In other words, the stream processor object of which the invoked handler function is part.

`params.settings`

Holds the resolved stream processor settings. In other words, this holds the settings to take into account, which may differ from the initial settings as available from params.processor.settings.

params.settings
function(params) {
  const { settings } = params
}

`params.timestamp`

Rarely used

This is the timestamp as reported by Kafka. You typically want to use the timestamp embedded in params.data instead.

`params.tools`

Holds the tools object which is how the Tap service does dependency injection. This tools object provides everything you need for a typically stream processing scenario.

params.settings
function(params) {
  const { settings } = params
}

Below is a full list of what params.tools provides:

`params.tools.cache`

tools.cache holds specific cache-related helpers:

`params.tools.cache.audit`

Holds a helper method to cache audit data:

/**
 * Caches audit data, with optional custom settings
 *
 * @param {Object} data - The data to cache
 * @param {Object} params - The params as passed to the handler
 * @param {Object} [custom] - Custom settings to apply
 * @param {Number} [custom.cap] - Number of audit data entries to keep
 * @param {Number} [custom.hostCap] - Number of per-host audit data entries to keep
 * @param {Number} [custom.userCap] - Number of per-user audit data entries to keep
 * @returns {Promise} - Function does not return, but since it's async, a Promise is implicitly returned
 */
async function audit (data, params, custom = {}) { /* ... */ }

Use it as:

function(params) {
  const { tools, data } = params
  // You should make sure data is properly structured
  tools.cache.audit(data, params)
}

tip

Settings passed in by the handler will take precedence over processor-wide settings:

const { 
  cap = 150,
  hostCap = 25,
  userCap = 25
} = { ...params.settings, ...custom }

The hostCap relies on data.morio.iid being set
The userCap relies on data.user.name being set

`params.tools.cache.event`

Holds a helper method to cache event data:

/**
 * Caches event data, with optional settings for the cache trim behaviour
 *
 * @param {Object} data - The data to cache
 * @param {Object} params - The params as passed to the handler
 * @param {Object} [custom] - Custom settings to apply
 * @param {Number} [custom.cap] - Number of event data entries to keep
 * @returns {Promise} - Function does not return, but since it's async, a Promise is implicitly returned
 */
async function event(data, params, custom = {}) { /* ... */ }

Use it as:

function(params) {
  const { tools, data } = params
  // You should make sure data is properly structured
  tools.cache.event(data, params)
}

tip

Settings passed in by the handler will take precedence over processor-wide settings:

const { cap = 150 } = { ...params.settings, ...custom }

`params.tools.cache.healthcheck`

Holds a helper method to cache healthcheck data:

/**
 * Caches healthcheck data, with optional settings for the cache trim behaviour
 *
 * @param {Object} data - The data to cache
 * @param {Object} [settings] - The settings that control cache trimming
 * @param {Number} [settings.cap] - Number of healthcheck data entries to keep
 * @param {Number} [settings.hostCap] - Number of per-host healthcheck data entries to keep
 * @returns {Promise} - Function does not return, but since it's async, a Promise is implicitly returned
 */
async function healthcheck(data, params, custom = {}) { /* ... */ }

Use it as:

function(params) {
  const { tools, data } = params
  // You should make sure data is properly structured
  tools.cache.healthcheck(data, params)
}

tip

Settings passed in by the handler will take precedence over processor-wide settings:

const { 
  cap = 300,
  hostCap = 100 
} = { ...params.settings, ...custom }

The hostCap relies on data.morio.iid being set

`params.tools.cache.logline`

Holds a helper method to cache log data:

/**
 * Caches log data, with optional settings for the cache trim behaviour
 *
 * @param {Object} data - The data to cache
 * @param {Object} [settings] - The settings that control cache trimming
 * @param {Number} [settings.cap] - Number of log data entries to keep
 * @param {Number} [settings.dataset] - Dataset name to use
 * @returns {Promise} - Function does not return, but since it's async, a Promise is implicitly returned
 */
async function logline(data, params, custom = {}) { /* ... */ }

Use it as:

function(params) {
  const { tools, data } = params
  // You should make sure data is properly structured
  tools.cache.logline(data, params)
}

tip

Settings passed in by the handler will take precedence over processor-wide settings:

const { cap = 50 } = { ...params.settings, ...custom }
const dataset = custom.dataset ? custom.dataset : params.dataset || '*'

`params.tools.cache.metricset`

Holds a helper method to cache metrics data:

/**
 * Caches metrics data, with optional settings for the cache trim behaviour
 *
 * @param {Object} data - The metric(set) to cache
 * @param {Object} [settings] - The settings that control cache trimming
 * @param {Number} [settings.cap] - Number of metrics data entries to keep
 * @param {Number} [settings.dataset] - Dataset name to use
 * @returns {Promise} - Function does not return, but since it's async, a Promise is implicitly returned
 */
async function metricset(data, params, custom = {}) { /* ... */ }

Use it as:

function(params) {
  const { tools, data } = params
  // You should make sure data is properly structured
  tools.cache.metricset(data, params)
}

tip

Settings passed in by the handler will take precedence over processor-wide settings:

const { cap = 150 } = { ...params.settings, ...custom }
const dataset = custom.dataset ? custom.dataset : params.dataset || '*'

`params.tools.cache.note`

Holds a helper method to cache .

/**
 * Caches a note
 *
 * @param {String} title - The note title
 * @param {Object} [data] - Optional note data
 * @returns {Promise} - Function does not return, but since it's async, a Promise is implicitly returned
 */
async function note(title, data = {}) { /* ... */ }

Use it as:

function(params) {
  const { tools } = params
  tools.cache.note("I am a note", { and: "I am note data })
}

`params.tools.cache.top`

Holds a helper method to cache top-x data, typically metrics, although it can be any numeric data.

/**
 * Caches top-x data
 *
 * @param {String} key - The key to cache under
 * @param {Array} data - The data to cache
 * @param {Number} [limit] - The number of top entries to keep
 * @returns {Promise} - Function does not return, but since it's async, a Promise is implicitly returned
 */
async function top(key, data, limit = 10) { /* ... */ }

Use it as:

function(params) {
  const { tools } = params
  // You can use a single data entry
  tools.cache.top("top-load", [ data.morio.iid, data.cpu.load-15 ])
  // Or provide multiple data entries
  tools.cache.top(
    "top-disk", 
    [
      [ `${data.morio.iid}|${data.disk1,name}`, data.disk1.used.pct ],
      [ `${data.morio.iid}|${data.disk2,name}`, data.disk2.used.pct ],
      [ `${data.morio.iid}|${data.disk3,name}`, data.disk3.used.pct ],
    ]
  )
}

tip

Under the hood, this will maintain a sorted set to which it will push data with ZADD and then trim it with ZREMRANGEBYRANK.

This allows you to maintain a top-x ranking for any kind of (numeric) data.

By default, it will keep a top-10, but you can override that with the limit parameter.

`params.tools.cache.ioredis`

Expert use

This is an alias for params.tools.valkey.

`params.tools.create`

Holds various helpers to create things:

`params.tools.create.context`

Creates a Morio context from the parameters you pass it. This method is variadic.

/**
 * Generates a context string from any parameters you pass it
 *
 * @param {...String} data - The (variadic) input data
 * @returns {String} - The context string
 */
function context(...data) { /* ... */ }

Use it as:

function(params) {
  const { tools } = params
  const context = tools.create.context('from', 'these', 'various', 'strings')
}

tip

The purpose of this helper is to ensure consistent context keys across Morio data. As such, you should use this rather than contruct your own context string.

`params.tools.create.debugHelper`

Creates an object to help you debug your stream processor code. This takes an id as paramter and returns an object with 3 properties:

start: Starts the debug
msg: Adds a debug message
end: Ends the debug

The debug generates , no other side effects.

/**
 * Generates a debug helper object
 *
 * @param {String} id - The debug ID helps to identify its debug output (the notes)
 * @returns {Object} - The debug helper object
 */
function debugHelper(id) { /* ... */ }

Use it as:

function(params) {
  const { tools } = params
  const debug = tools.create.debugHelper('my-processor')
  debug.start()
  try {
    debug.msg('Before risky')
    // do something that might fail
    const result = risky()
    debug.msg('After risky', result)
  }
  catch (err) {
    debug('It failed', err)
  }
  debug.end()
}

tip

Whereas the start and end methods of the debug helper take no arguments, the msg method takes a String argument as message, and an optional second argument as data to include with the message.

`params.tools.create.elasticId`

Expert use

A NodeJS implementation of the algorithm used by the Beats add_id processor:

/**
 * This is a NodeJS implementation of the algorithm used by Beats add_id processor:
 * - Generate 20 random bytes
 * - Convert to base64
 * - Replace + with - and / with _ (to make it URL-safe)
 *
 * @return {String} id - The unique id
 */
function elasticId() { /* ... */ }

Use it as:

function(params) {
  const { tools } = params
  const id = tools.create.elasticId()
}

tip

Since this is specific to the Elasticsearch storage backend, how to use this is outside the scope of Morio.

However, the idea here is that you can generate an ID prior to ingestion by Elasticsearch, so that you have that ID and it is the same as teh document ID after ingestion.

`params.tools.create.hash`

Creates a (sha256) hash from the data you pass it.

/**
 * Generates a SHA-256 hash
 *
 * @param {any} data - The input data
 * @returns {String} - The hash
 */
function hash(data) { /* ... */ }

Use it as:

function(params) {
  const { tools, data } = params
  const hash = tools.create.hash(data)
}

`params.tools.create.key`

Creates a Morio cache key from the parameters you pass it. This method is variadic.

/**
 * Generates a cache key from any parameters you pass it
 *
 * @param {...String} data - The (variadic) input data
 * @returns {String} - The cache key
 */
function key(...data) { /* ... */ }

Use it as:

function(params) {
  const { tools } = params
  const key = tools.create.key('from', 'these', 'various', 'strings')
}

tip

The purpose of this helper is to ensure consistent cache keys across Morio data. As such, you should use this rather than contruct your own cache key.

`params.tools.create.uuid`

Creates a (v4) UUID.

/**
 * Generates a UUIDv4
 *
 * @returns {String} - The UUID
 */
function uuid() { /* ... */ }

Use it as:

function(params) {
  const { tools } = params
  const uuid = tools.create.uuid()
}

`params.tools.extract`

Holds various helpers to extract specific info from the data:

params.tools.extract.agent: Extracts the name of the agent
params.tools.extract.check: Extracts the healthcheck URL
params.tools.extract.hostname: Extracts the host name
params.tools.extract.id: Extracts the message ID
params.tools.extract.iid: Extracts the
params.tools.extract.metricset: Extracts the metricset name
params.tools.extract.module: Extracts the module name
params.tools.extract.timestamp: Extracts the timestamp

tip

All of these methods take a single argument: data which should be the data received from Kafka.

/**
 * All extract functions have the same signature
 *
 * @param {Object} data - The data received from Kafka
 * @returns {String} - The extracted data
 */
function(data) { }

Note that:

These methods will attempt to extract the relevant data, but return a fallback value when it cannot be found. So they always return a value.
Not all methods can be used with any data. For example extracting the healthcheck URL should only be used on healthcheck data.

`params.tools.format`

Holds various helpers to format data:

params.tools.format.clean: Will force the input to a input, convert to lowercase, and trim leading and trailing whitespace
params.tools.format.escape: An alias for NodeJS’s querystring.escape
params.tools.format.stringify: Will convert input to a string, and stringify objects to JSON
params.tools.format.shortUuid: Will return the input string trimmed to maximum 5 characters

tip

All of these methods take a single argument: input which should be a string.

/**
 * All format functions have the same signature
 *
 * @param {String} input - The data received from Kafka
 * @returns {String} - The formatted data
 */
function(data) { }

`params.tools.get`

Provides lodash.get, a utility function that retrieves the value of a nested property from an object safely.

`params.tools.ip`

Holds helpers to work with IP addresses:

`params.tools.ip.matchCidr`

A helper method to check whether an IP address is part of a CIDR block.

/*
 * Helper method to match a and IP to a CIDR block
 *
 * @param {string} ip - The IP address, eg: 10.6.66.123
 * @param {string} cidr - The network in CIDR notation, eg: 10.6.66.0/24
 * @return {boolean} match - True if the IP address is in the CIDR block, false if not
 */
function matchCidr(ip, cidr) { /* ... */ }

`params.tools.kafka`

Holds the Kafka consumer and producer:

`params.tools.kafka.consumer`

Expert use

Holds the Kafka consumer, a KafkaJS consumer instance.

`params.tools.kafka.producer`

Expert use

Holds the Kafka producer, a KafkaJS producer instance.

`params.tools.link`

Helper methods to create HTTP links or href data:

`params.tools.link.md`

Helper methods to generate markdown syntax for links:

`params.tools.link.md.to`

Takes a URL and optional title and returns a markdown-formatted link:

/*
 * Creates a markdown link
 *
 * @param {String} href - The HREF to link to
 * @param {String} [title] - The title/text to use for the link, will use href if not set
 * @return {String} md - The markdown code
 */
function to(href, title=false) { /* ... */ }

`params.tools.link.md.audit`

Creates markdown links to pages where one can audit the cached data:

`params.tools.link.md.audit.host`

Creates a Markdown link to a page with a host’s audit data:

/*
 * Creates a Markdown link to a host's audit page
 *
 * @param {String} uuid - The host's UUID
 * @param {String} title - The title to use for the link
 * @return {String} md - The markdown code
 */
function host(uuid, title=false) { /* ... */ }

`params.tools.link.md.audit.user`

Creates a Markdown link to a page with a user’s audit data:

/*
 * Creates a Markdown link to a user's audit page
 *
 * @param {String} username - The username
 * @param {String} title - The title to use for the link
 * @return {String} md - The markdown code
 */
function user(uuid, title=false) { /* ... */ }

`params.tools.link.md.healthcheck`

Creates a Markdown link to a page with cached healthcheck data:

/*
 * Creates a Markdown link to a page with cached healthcheck data
 *
 * @param {String} id - The healthcheck ID
 * @param {String} title - The title to use for the link
 * @return {String} md - The markdown code
 */
function healthcheck(id, title=false) { /* ... */ }

`params.tools.link.md.inventory`

Holds helpers to link to data in the Morio inventory:

`params.tools.link.md.inventory.host`

Creates a Markdown link to a host in the Morio inventory:

/*
 * Creates a Markdown link to a host in the Morio inventory
 *
 * @param {String} uuid - The iid of the host
 * @param {String} title - The title to use for the link
 * @return {String} md - The markdown code
 */
function host(id, title=false) { /* ... */ }

`params.tools.link.md.logset`

Creates a Markdown link to a page with cached log data:

/*
 * Creates a Markdown link to a page with cached log data
 *
 * @param {String} iid - The host's iid (UUID)
 * @param {String} module - The Morio module name
 * @param {String} dataset - The Morio dataset name
 * @param {String} title - The title to use for the link
 * @return {String} md - The markdown code
 */
function logset(iid, module, dataset, title=false) { /* ... */ }

`params.tools.link.md.metricset`

Creates a Markdown link to a page with cached metrics data:

/*
 * Creates a Markdown link to a page with cached metrics data
 *
 * @param {String} iid - The host's iid (UUID)
 * @param {String} module - The Morio module name
 * @param {String} dataset - The Morio dataset name
 * @param {String} title - The title to use for the link
 * @return {String} md - The markdown code
 */
function metricset(iid, module, dataset, title=false) { /* ... */ }

`params.tools.link.href`

Helper methods to generate a URL to be used in a link’s href attribute:

tip

This holds the same helper methods as params.tools.link.md, but these return a URL instead of Markdown code.

As such, they do not take the optional title paramter.

`params.tools.link.href.to`

Similar as params.tools.link.md.to, but returns a URL instead of Markdown code, and as such does not take the title argument.

`params.tools.link.href.audit`

`params.tools.link.href.audit.host`

Similar as params.tools.link.md.audit.host, but returns a URL instead of Markdown code, and as such does not take the title argument.

`params.tools.link.href.audit.user`

Similar as params.tools.link.md.audit.user, but returns a URL instead of Markdown code, and as such does not take the title argument.

`params.tools.link.href.healthcheck`

Similar as params.tools.link.md.healthcheck, but returns a URL instead of Markdown code, and as such does not take the title argument.

`params.tools.link.href.inventory`

`params.tools.link.md.inventory.host`

Similar as params.tools.link.md.inventory.host, but returns a URL instead of Markdown code, and as such does not take the title argument.

`params.tools.link.md.logset`

Similar as params.tools.link.md.logset, but returns a URL instead of Markdown code, and as such does not take the title argument.

`params.tools.link.md.metricset`

Similar as params.tools.link.md.metricset, but returns a URL instead of Markdown code, and as such does not take the title argument.

`params.tools.logCacheErrors`

A helper method that you can pass as a callback to ioredis operations.

Keep in mind is that this will not actually log. Instead it will create .

/**
 * A helper function to log ValKey pipeline errors
 *
 * @param {object} result - The result from the pipeline
 * @param {object} [info] - Optional additional data
 */
function logCacheErrors (result, info) { /* ... */ }

Use it as:

function processor(params) {
  const { tools, data } = params
  const pipeline = tools.valkey
    .pipeline()
    .lpush('example-key', data.some.value)
    .exec(result => tools.logCacheErrors(result, { optional: 'data goes here' }))
}

`params.tools.set`

Provides lodash.set, a utility function that safely mutates an object by setting a (nested) property on it.

`params.tools.produce`

Holds helper methods to produce data to a Kafka topic:

params.tools.produce.alarm: Writes to the alarms topic
params.tools.produce.alert: Writes to the alerts topic
params.tools.produce.event: Writes to the events topic
params.tools.produce.notification: Writes to the notifications topic

tip

All of these take a single data argument, which should hold the data to write to Kafka.

Please ensure your data is properly structured.

`params.tools.time`

Holds helper methods to handle time:

`params.tools.time.ms2s`

Takes a number of milliseconds as input, returns an integer amount of seconds.

`params.tools.time.now`

Returns the current timestamp (Date.now()), takes no arguments.

`params.tools.time.when`

An alias of params.tools.extract.timestamp

`params.tools.valkey`

Expert use

Holds the instantiated Valkey/Redis client, a pre-configured instance of ioredis.
Use this if you want to do advanced Valkey/Redis operations that go beyond simple caching.

`params.tools.when`

An alias of params.tools.extract.timestamp

`params.topic`

Rarely used

The topic the message originated from. This is rarely relevant as stream processors typically subscribe to one topic only.

params.data
params.dataset
params.iid
params.module
params.node
params.offset
params.partition
params.processor
params.settings
params.timestamp
params.tools
params.topic

params.data​

params.dataset​

params.iid​

params.module​

params.node​

params.offset​

params.partition​

params.processor​

params.settings​

params.timestamp​

params.tools​

params.tools.cache​

params.tools.cache.audit​

params.tools.cache.event​

params.tools.cache.healthcheck​

params.tools.cache.logline​

params.tools.cache.metricset​

params.tools.cache.note​

params.tools.cache.top​

params.tools.cache.ioredis​

params.tools.create​

params.tools.create.context​

params.tools.create.debugHelper​

params.tools.create.elasticId​

params.tools.create.hash​

params.tools.create.key​

params.tools.create.uuid​

params.tools.extract​

params.tools.format​

params.tools.get​

params.tools.ip​

params.tools.ip.matchCidr​

params.tools.kafka​

params.tools.kafka.consumer​

params.tools.kafka.producer​

params.tools.link​

params.tools.link.md​

params.tools.link.md.to​

params.tools.link.md.audit​

params.tools.link.md.audit.host​

params.tools.link.md.audit.user​

params.tools.link.md.healthcheck​

params.tools.link.md.inventory​

params.tools.link.md.inventory.host​

params.tools.link.md.logset​

params.tools.link.md.metricset​

params.tools.link.href​

params.tools.link.href.to​

params.tools.link.href.audit​

params.tools.link.href.audit.host​

params.tools.link.href.audit.user​

params.tools.link.href.healthcheck​

params.tools.link.href.inventory​

params.tools.link.md.inventory.host​

params.tools.link.md.logset​

params.tools.link.md.metricset​

params.tools.logCacheErrors​

params.tools.set​

params.tools.produce​

params.tools.time​

params.tools.time.ms2s​

params.tools.time.now​

params.tools.time.when​

params.tools.valkey​

params.tools.when​

params.topic​

`params.data`

`params.dataset`

`params.iid`

`params.module`

`params.node`

`params.offset`

`params.partition`

`params.processor`

`params.settings`

`params.timestamp`

`params.tools`

`params.tools.cache`

`params.tools.cache.audit`

`params.tools.cache.event`

`params.tools.cache.healthcheck`

`params.tools.cache.logline`

`params.tools.cache.metricset`

`params.tools.cache.note`

`params.tools.cache.top`

`params.tools.cache.ioredis`

`params.tools.create`

`params.tools.create.context`

`params.tools.create.debugHelper`

`params.tools.create.elasticId`

`params.tools.create.hash`

`params.tools.create.key`

`params.tools.create.uuid`

`params.tools.extract`

`params.tools.format`

`params.tools.get`

`params.tools.ip`

`params.tools.ip.matchCidr`

`params.tools.kafka`

`params.tools.kafka.consumer`

`params.tools.kafka.producer`

`params.tools.link`

`params.tools.link.md`

`params.tools.link.md.to`

`params.tools.link.md.audit`

`params.tools.link.md.audit.host`

`params.tools.link.md.audit.user`

`params.tools.link.md.healthcheck`

`params.tools.link.md.inventory`

`params.tools.link.md.inventory.host`

`params.tools.link.md.logset`

`params.tools.link.md.metricset`

`params.tools.link.href`

`params.tools.link.href.to`

`params.tools.link.href.audit`

`params.tools.link.href.audit.host`

`params.tools.link.href.audit.user`

`params.tools.link.href.healthcheck`

`params.tools.link.href.inventory`

`params.tools.link.md.inventory.host`

`params.tools.link.md.logset`

`params.tools.link.md.metricset`

`params.tools.logCacheErrors`

`params.tools.set`

`params.tools.produce`

`params.tools.time`

`params.tools.time.ms2s`

`params.tools.time.now`

`params.tools.time.when`

`params.tools.valkey`

`params.tools.when`

`params.topic`