How can I compare current value in my stream with previous value in tickscript?

telegraf
influxql
kapacitor
influxdb
chronograf

#1

I am working on kapacitor alerts to generate alerts on data in influxDB. I am receiving a new value from the stream when a request has been made. I want to compare the current value in my stream I received with the previous value I received just before it to see if current value is greater than previous value or not. If it is greater than previous value, then I want to generate an alert on chronograf using kapacitor and tickscript.

How do I achieve this in tickscript? Any help is highly appreciated. I am new to tickscript.


#2

Hi ,
maybe the derivativenode can be used ,
If the current value is greater than the previous value the derivative will be > 0 ,

DerivativeNode

Computes the derivative via: (current - previous ) / ( time_difference / unit)

Update : maybe this is also an option … https://docs.influxdata.com/kapacitor/v1.5/nodes/influx_q_l_node/#difference


#3

@MarcV Can you give the tickscript? I am new to it. I need the tickscript for fetching the current and previous values and computing this derivative.


#4

Hi the link I posted earlier contains a part of the script you need.
Does it help you to create the script ?


#5

Hi here is an example , best regards ,

create a script task1.tick with the following content

dbrp "telegraf"."autogen"

stream

 |from()
   .measurement('numbers')

 |derivative('value')

 |alert()
  .info(lambda:"value" > 0 )

to create the task in kapacitor

kapacitor define task1 -tick task1.tick

to enable the task in kapacitor

kapacitor enable task1

And to test the alert , you can start the influx command line interface and insert some points …

influx -database telegraf

insert numbers value=1
insert numbers value=1
insert numbers value=2

and to monitor the task

you can use

kapacitor watch task1   

good luck :slight_smile:

2019-03-14T11:01:00Z


#6

Hey @MarcV I’ve been using this code and it’s not sending an alert. It’s the same as what you did. I am not sure why. Could you please give your feedback?

var data = stream
    |from()
        .database(db)
        .retentionPolicy(rp)
        .measurement(measurement)
        .groupBy(groupBy)
        .where(whereFilter)
    |derivative('QCOUNT')
        .as('value')

var trigger = data
    |alert()
        .crit(lambda: "value" > crit)
        .message(message)
        .id(idVar)
        .idTag(idTag)
        .levelTag(levelTag)
        .messageField(messageField)
        .durationField(durationField)
        .details('''
        <b>QCOUNT</b> [{{ index .Fields "QCOUNT" }}]</br>
        ''')
        .email()
        .to('karthik@gmail.com')

#7

So I am sending data every 5 minutes. And I wanted the derivative to work on this for every 5 minutes. That is check the value at the end of 5th minute with the one before 5 minutes and if it’s non-negative, which means the value at the end of 5th minute is greater than the before value, send an email alert.

But this seems to be not working. Any help appreciated. I am going through documentation. Thanks for sharing. I am totally new to this. So I am looking for help.


#8

Hi ,

do you see alerts triggerd in the Kapacitor logfile ?

Can you try with “value” > 0 in your lambda expression ?


#9

@MarcV I changed it to “value” > 0 in lambda expression and it’s still not working.

I don’t see alerts triggered in Kapacitor logfile either. Any idea why?


#10

This is my whole script right now.

var data = stream
    |from()
        .database(db)
        .retentionPolicy(rp)
        .measurement(measurement)
        .groupBy(groupBy)
        .where(whereFilter)
    |derivative('QCOUNT')
        .unit(10s)
        .nonNegative()
        .as('value')

var trigger = data
    |alert()
        .crit(lambda: "value" > 0)
        .message(message)
        .id(idVar)
        .idTag(idTag)
        .levelTag(levelTag)
        .messageField(messageField)
        .durationField(durationField)
        .details(details)
        .email()
        .to('karthik@gmail.com')

trigger
    |eval(lambda: float("value"))
        .as('value')
        .keep()
    |influxDBOut()
        .create()
        .database(outputDB)
        .retentionPolicy(outputRP)
        .measurement(outputMeasurement)
        .tag('alertName', name)
        .tag('triggerType', triggerType)

trigger
    |httpOut('output')

#11

I am sending the QCOUNT variable through REST call into my influxDb and this REST call runs every 5 minutes. So I want to compare the values at these 5 minutes interval and above is the script I wrote for it. Any help is greatly appreciated.


#12

Hi it is hard to tell but
Can you check with

   kapacitor show yourtask

To see if the task processes data
And

  kapacitor stats ingress

To see if Kapacitor is getting data from the measurement ?
Is the value of QCOUNT higher than the
Previous value ?


#13

Hi ,
@adarsh_hatwar
talks about these relative alerts in chronograf in this post ,
https://community.influxdata.com/t/alerting-if-mean-of-a-value-is-above-a-threshold-value-for-a-certain-period-of-time/8969

I have tried it out for your script and chronograf came up with this …

var db = 'dbcpu'
var rp = 'autogen'
var measurement = 'cpu'
var groupBy = []
var whereFilter = lambda: ("cpu" == 'cpu-total')
var name = 'compare'
var idVar = name
var message = ''
var idTag = 'alertID'
var levelTag = 'level'
var messageField = 'message'
var durationField = 'duration'
var outputDB = 'chronograf'
var outputRP = 'autogen'
var outputMeasurement = 'alerts'
var triggerType = 'relative'
var shift = 5m
var crit = 0
var data = stream
    |from()
        .database(db)
        .retentionPolicy(rp)
        .measurement(measurement)
        .groupBy(groupBy)
        .where(whereFilter)
    |eval(lambda: "usage_idle")
        .as('value')

var past = data
    |shift(shift)

var current = data

var trigger = past
    |join(current)
        .as('past', 'current')
    |eval(lambda: float("current.value" - "past.value"))
        .keep()
        .as('value')
    |alert()
        .crit(lambda: "value" > crit)
        .message(message)
        .id(idVar)
        .idTag(idTag)
        .levelTag(levelTag)
        .messageField(messageField)
        .durationField(durationField)
        .stateChangesOnly()

trigger
    |eval(lambda: float("value"))
        .as('value')
        .keep()
    |influxDBOut()
        .create()
        .database(outputDB)
        .retentionPolicy(outputRP)
        .measurement(outputMeasurement)
        .tag('alertName', name)
        .tag('triggerType', triggerType)

trigger
    |httpOut('output')

#14

@MarcV This won’t work if the data comes continuously right? Say, I have people dumping QCOUNT values at random intervals, unlike every 5 minutes, it won’t work I guess. What I am trying to achieve is that, at any random intervals of data being dumped into influxDB, look at the current value that is being inserted, look at the past value, and if the current value is greater than the past value, then generate an alert. How do I go about this?

And thanks for all your effort. I really appreciated your help.


#15

Hi ,
have you found a solution ?
what is your task name ?
Can you share the following information assuming the name of your task is mytask
and assuming the task is enabled ?

 kapacitor show mytask

you can also check for errors while the task is executing with :

 kapacitor watch mytask

#16

@MarcV The task is enabled. It’s still not working. Like I said, the problem I am trying to solve is, to take the previous value and compare with current value in a stream. This QCOUNT variable can be posted in any random interval. Like someone can post that into influxDB whenever they want. When a new value comes in, I want to compare it with previous value. How do I achieve this?


#17

@gvkarthik93
Hi, have you looked at this?
Elaborating your use case on your alerts being real time would help a bit.


#18

@adarsh_hatwar I have looked at that example. How do I get the previous value from the same stream of data using that example. Like I mentioned about, I have stream of data coming in. That is I have QCOUNT variable coming in randomly. When a new QCOUNT comes in, how do I get the previous QCOUNT, compare both and generate alert?


#19

Hi can you share your complete script ?

I will test it with exactly the same script.


#20

@MarcV This is the complete script I am trying.

var db = 'telegraf'

var rp = 'autogen'

var measurement = 'E Plus'

var groupBy = []

var whereFilter = lambda: ("domain" == 'gxgame') AND ("statsName" == 'Q Count')

var name = 'qcounttest0318'

var idVar = name

var message = 'subject {{ index .Tags "value" }} {{ index .Fields "value" }}'

var idTag = 'alertID'

var levelTag = 'level'

var messageField = 'message'

var durationField = 'duration'

var outputDB = 'chronograf'

var outputRP = 'autogen'

var outputMeasurement = 'alerts'

var triggerType = 'threshold'

var details = 'message'

var crit = 0

var data = stream
|from()
    .database(db)
    .retentionPolicy(rp)
    .measurement(measurement)
    .groupBy(groupBy)
    .where(whereFilter)
|window()
    .period(10m)
    .every(10m)
    .align()
|derivative('QCOUNT')
    .as('value')

var trigger = data
|alert()
    .crit(lambda: "value" > crit)
    .message(message)
    .id(idVar)
    .idTag(idTag)
    .levelTag(levelTag)
    .messageField(messageField)
    .durationField(durationField)
    .details(details)
    .email()
    .to('karthik@gmail.com')

trigger
|eval(lambda: float("value"))
    .as('value')
    .keep()
|influxDBOut()
    .create()
    .database(outputDB)
    .retentionPolicy(outputRP)
    .measurement(outputMeasurement)
    .tag('alertName', name)
    .tag('triggerType', triggerType)

trigger
|httpOut('output')

Any help with the use case I mentioned is appreciated. What I want to achieve is whenever there is a stream of data, look at previous value and compare it with current value and generate alert. This is irregular stream. So people can send data into influxDB anytime.