InfluxDB 2: how to check value from the stream, and conditionally write new value to another measurement/tags

[RavilN]

Hi, I have InfluxDB 2 setup, connecting to its Chronograph GUI over port 9999. From the Tasks menu, trying to create a task, which would check if a value for some measurement/tags combination is exceeding some threshold value, and then if it exceeds, write some value on some another measurement/tag.
For example, if a field “value” with measurement equal to “pressure”, tag “location” equal to “pump1” has a field “value” greater than 100, write text value “Pressure is higher than 100!” into measurement “status”, tag “location”.
And this needs to be done only once while value exceeds 100. When it goes back to below 100, then should change the value to “OK” once.

Tried to use Alerts feature, but with it, I don’t see notification endpoint to write to InfluxDB. There are only notification endpoints to select from HTTP, slack, and pagerduty.
Thank you, Ravil

[RavilN]

In the end, I came to a somewhat working query, using monitor.check function. Probably could be better, enough for the purpose for now.

import “influxdata/influxdb/monitor”
option task = {name: “Check Pump Pressure”, every: 5s, offset: 1s}
currentTime = now()
join(tables: {ip: from(bucket: “ogamma”)
|> range(start: -task.every)
|> filter(fn: (r) =>
(r._measurement == “InputPressure” and r.unit == “Pump” and r._field == “v”))
|> drop(columns: [“_source_timestamp”])
|> mean(column: “_value”)
|> map(fn: (r) =>
({
_time: currentTime,
_value: r._value,
unit: r.unit,
_measurement: “OutputPressure”,
})), op: from(bucket: “ogamma”)
|> range(start: -task.every)
|> filter(fn: (r) =>
(r._measurement == “OutputPressure” and r.unit == “Pump” and r._field == “v”))
|> drop(columns: [“_source_timestamp”])
|> mean(column: “_value”)
|> map(fn: (r) =>
({
_time: currentTime,
_value: r._value,
unit: r.unit,
_measurement: “OutputPressure”,
}))}, on: [“_time”, “_measurement”])
|> map(fn: (r) =>
({
_time: r._time,
_pd: r._value_op - r._value_ip,
unit: r.unit_op,
_measurement: “OutputPressure”,
}))
|> monitor.check(
crit: (r) =>
(r._pd < 10.0),
warn: (r) =>
(r._pd < 15.0),
info: (r) =>
(r._pd < 25.0),
ok: (r) =>
(r._pd >= 25.0),
messageFn: (r) =>
(if r._pd < 10.0 then “Critical alert!! error: Pressure difference is too low: ${string(v: r._pd)}!” else if r._pd < 15.0 then “Warning! Pressure difference is low: ${string(v: r._pd)}!” else if r._pd < 25.0 then “Pressure difference is: ${string(v: r._pd)}!” else “ok”),
data: {
_check_name: “Pressure difference check”,
_check_id: “pres_diff”,
_type: “threshold”,
tags: {unit: “Pump”},
_measurement: “statuses”,
},
)