Monitor.notify() does not work

InfluxDB 2 Checks & Notifications
1

Hi:
when i use monitor.notify() to send the critic and store notification status, it throw error " runtime error @72:6-97:6: notify: failed to evaluate map function: error calling function “encode” @93:27-93:47: got table stream instead of array. Try using tableFind() or findRecord() to extract data from stream"

20240806-113208
20240806-1132081920×938 145 KB

The code as below:

import "strings"
import "regexp"
import "json"
import "influxdata/influxdb/secrets"
import "influxdata/influxdb/schema"
import "influxdata/influxdb/monitor"
import "http"
import "experimental"
import "math"
import "date"

endtime = date.truncate(t: 2024-08-06T01:06:00Z, unit: 1s) 
starttime = date.truncate(t: 2024-08-06T01:00:00Z, unit: 1s) 
offsecond = duration(v: string(v: date.second(t: starttime)) + "s") 
check = {_check_id: "0ccceff1f2f2e000", _check_name: "VelocityAlert", _type: "changeRate", tags: {}}
notification = {
    _notification_rule_id: "0cca7d2d652ad000",
    _notification_rule_name: "AnyToCrit",
    _notification_endpoint_id: "0cca7ce587077000",
    _notification_endpoint_name: "collectAlertNotifications",
}
roundTo = (n, precision) => {
    multiplier = math.pow10(n: precision)

    return math.round(x: n * multiplier) / multiplier
}
fixed = (v, p) => {
    strV = string(v: v)
    splitV = strings.split(v: strV, t: ".")
    hasFraction = length(arr: splitV) > 1
    fraction =
        if hasFraction then
            splitV[1] + strings.repeat(v: "0", i: p - strings.strlen(v: splitV[1]))
        else
            strings.repeat(v: "0", i: p)
    result = if p == 0 then "${splitV[0]}" else "${splitV[0]}.${fraction}"

    return result
}

data = from(bucket: "CS107") 
  |> range(start: date.add(d: -5m, to: starttime), stop: endtime) 
  |> filter(fn: (r) => r["_measurement"] == "CS107.RA-A705" and r["_field"] == "TI16") 
  |> aggregateWindow(every: 1m, fn: last, offset: offsecond) 
  |> fill(column: "_value", usePrevious: true) 
  |> filter(fn: (r) => r["_time"] >= starttime) 

first = data |> first() 
last = data |> last() 

result = union(tables: [first, last]) 
  |> map(fn: (r) => ({ r with _value: roundTo(n: r._value, precision: 3) })) 
  |> derivative(unit: 1h, nonNegative: false, columns: ["_value"], timeColumn: "_time") 
  |> map(fn: (r) => ({ r with _value: roundTo(n: math.abs(x: r._value), precision: 3) })) 
  |> keep(columns: ["_measurement", "_field", "_time", "_value",],) 
  |> to(bucket: "VelocityAlert") 
derivative = result |> findRecord(fn: (key) => true, idx: 0)
trigger = (r) => r["TI16"] < 1 or r["TI16"] > 2
messageFn =
    (r) =>
        "{\"AlertMessage\":\"${fixed(
                v: r["TI16"],
                p: 2,
            )}\",\"AlertTime\":\"${r._time}\",\"AlertType\":1,\"EbrCode\":\"EBR-2408050000012\",\"EquipCode\":\"CS107.RA-A705.TI16\",\"ParameterCode\":\"6a1db9ec-46b4-c70f-e77f-08dcb50d7067|1/1\",\"TagName\":\"TI16\"}"


result
  |> schema["fieldsAsCols"]()
  |> filter(fn: (r) => r["_measurement"] == "CS107.RA-A705")
  |> monitor["check"](data: check, messageFn: messageFn, crit: trigger)
  |> filter(fn: trigger)
  |> monitor["notify"](
    data: notification,
    endpoint:
        http.endpoint(
            url: "https://esb-test.porton.cn/restcloud/sebriot/api/collectAlertNotifications",
        )(
             mapFn: (r) => {
                body = {
                    EbrCode: "EBR-2408050000012",
                    ParameterCode: "6a1db9ec-46b4-c70f-e77f-08dcb50d7067|1/1",
                    EquipCode: "CS107.RA-A705.TI16",
                    AlertTime: r._time,
                    AlertType: 1,
                    AlertMessage: fixed(v: r["TI16"], p: 2),
                    TagName: "TI16",
                    First: first,
                    Last: last,
                }

                return {
                    headers: {"Content-Type": "application/json"},
                    data: json.encode(v: body),
                }
            },
        ),
    )
2

@Leo I believe the error stems from calling the first and last variables in the context of the mapFn parameter. This parameter builds a Flux record that then gets JSON-encoded and sent as the notification request body. The first and last variables return streams of tables, which cannot be JSON-encoded.

first = data |> first() 
last = data |> last()

You need to extract a row from each of this tables and potentially a scalar value in that row. If you want to send the entire row as part of the notification body, update the variables to this:

first = data |> first() |> findRecord(fn: (key) => true, idx: 0)
last = data |> last() |> findRecord(fn: (key) => true, idx: 0)

If you just want to send the _value from these rows, do this:

first = (data |> first() |> findRecord(fn: (key) => true, idx: 0))._value
last = (data |> last() |> findRecord(fn: (key) => true, idx: 0))._value
3

yes, that does work. thanks for your solution.