How to write back Kapacitor alerts into influx

There is a problem: you want to save alerts from Kapacitor back into InfluxDB, but when you use |InfluxDBOut, all you get is original data which had trigger alert with zero information on the alert itself (its level, message, id, etc).

Most people at that moment want just to have a table representation of the “kapacitor show-topic” output, and a simple|InfluxDBOut does not help.

It took me about two hours to find a way. Good news, there is a way to write alerts into any measurement into any database in Influx.

The wrong approach is:

alert()
  .warn(lambda: "x" > 0)
  .id('prophecy')
  .warn('Its a warning')
  .topic('bad_things_will_happen')
 | InfluxDBOut()
      .database(' apocrypha')
      .measurement('warnings')

It will put only original fields into influx. No ‘bad_things_will_happen’, no ‘prophecy’, no WARN level.

The solution

To fix this one need to add tags and fields into alert. It’s strange, but when you write alert().id(something) it does not add anything into data in the pipeline. It only add data to the actions of the alert, not to the data in the pipeline.

To add some data into pipeline you need to do this explicitly:

alert()
  .warn(lambda: "x" > 0)
  .id('prophecy')
  .warn('Its a warning')
  .topic('bad_things_will_happen')
  .idTag('alert_id')
  .levelTag('level')
  .messageField('message')
  .topicTag('topic_id')
 | InfluxDBOut()
      .database(' apocrypha')
      .measurement('warnings')

Pay attention to those .levelTag and .messageField — they add required information into pipeline with a specified name.

It’s all in the docs but it really hard to understand that they are needed in this situation.

Reducing clutter

If you want to cut some fields from data stream it’s easy to do. Use Delete node or evalnode with .keep() function:

alert()
  .warn(lambda: "x" > 0)
  .id('prophecy')
  .warn('Its a warning')
  .topic('bad_things_will_happen')
  .idTag('alert_id')
  .levelTag('level')
  .messageField('message')
|eval()
  .keep('message')
| InfluxDBOut()
      .database(' apocrypha')
      .measurement('warnings')

Please, pay attention to difference between tags and fields. You can store f.e. level as a tag (with .levelTag) or as a field with .levelField). But you need at least one field to be present for each entry to have it saved into influx.