influxdb

[mischa]

I want to store LM object (values) data in an influxdb database, does anybody have any experience with this?

Their documentation (https://docs.influxdata.com/influxdb/v1....ting_data/) says that you can use the http api for storing data like this;

Code:

curl -i -XPOST 'http://localhost:8086/write?db=mydb' --data-binary 'cpu_load_short,host=server01,region=us-west value=0.64 1434055562000000000'

Is this possible from lua in the LM? 

Thanks Mischa

[admin]

Code:

http = require('socket.http')
http.TIMEOUT = 5
res, err = http.request('http://192.168.1.2:8086/write?db=mydb', 'cpu_load_short,host=server01,region=us-west value=0.64')
log(res, err)

[mischa]

Code:

http = require('socket.http')
http.TIMEOUT = 5
res, err = http.request('http://192.168.1.2:8086/write?db=mydb', 'cpu_load_short,host=server01,region=us-west value=0.64')
log(res, err)

Thanks, works perfectly.

[eirik]

Hi 
I have tried this influxdb-connection and got it to work.
I want to write several tags to influxdb and can't really figure out the best way to do this in scripts.....

Data like temperatures, Co2 and also presence objects are data I need to log i influx. 
Could anyone give me a push in the right direction here. 
Planning to use this with grafana. 

Everything is up and running but can't figure out the most vital part:-)

Eirik

[myg]

This code will send everything it captures to influx, then you can filter whatever you want in grafana. Create resident script with zero delay

Code:

local socket = require('socket')
local http = require('socket.http')
http.TIMEOUT = 5


local dt_cache = {}
function get_value(addr, datahex)
  local dt = dt_cache[addr]

  if(not dt) then
    dt = grp.find(addr)
    dt_cache[addr] = dt
  end
 
  return knxdatatype.decode(datahex, dt.datatype), dt
end

function knx_callback(event)
  local addr = event.dst
  local value, dt = get_value(addr, event.datahex)

  send_metric('rawdata', dt.name, addr, value)
end


function send_metric(table, name, addr, value)
  if(name == nil or name == '') then
    return
  end
 
  name = string.gsub(name, ' ', '\\ ')
 
  local url = 'http://IP:8086/write?db=knxdb'
  local args
  if(type(value) == 'boolean') then
    args = string.format('%s,name=%s,addr=%s state=%s', table, name, addr, value)
  else
    -- most likely number
    args = string.format('%s,name=%s,addr=%s value=%s', table, name, addr, value)
  end
  res, err = http.request(url, args)
  if(err ~= 204) then
    log('error sending to influx', res, err, args)
  end
end

function run_loop()
  local bus = require('localbus').new(1)
  bus:sethandler('groupwrite', knx_callback)
  local busfd = socket.fdmaskset(bus:getfd(), 'r')

 
  while(true) do
    res, fbus = socket.selectfds(10, busfd)
    if(fbus) then
      bus:step()
    end
  end
end

run_loop()

[thomasoppida]

Hi, I am wondering if there is a working script to do the same for Influx V2, where you need a token, define organization and push to correct bucket.

kind regards 
Thomas

[admin]

Modified send_metric function for the previous example. Replace IP/BUCKET/ORG/TOKEN placeholders with your parameters.

Code:

function send_metric(table, name, addr, value)
  if name == nil or name == '' then
    return
  end

  name = string.gsub(name, ' ', '\\ ')

  local url = 'http://IP:8086/api/v2/write?bucket=BUCKET&org=ORG'
  local body

  if type(value) == 'boolean' then
    body = string.format('%s,name=%s,addr=%s state=%s', table, name, addr, value)
  else
    -- most likely number
    body = string.format('%s,name=%s,addr=%s value=%s', table, name, addr, value)
  end

  local res, code = http.request({
    url = url,
    method = 'POST',
    body = body,
    headers = {
      Authorization = 'Token TOKEN'
    }
  })

  if code ~= 204 then
    log('error sending to influx', res, code, body)
  end
end

[thomasoppida]

admin pid='24163' dateline='1639476013']Modified send_metric function for the previous example. Replace IP/BUCKET/ORG/TOKEN placeholders with your parameters.

Code:

function send_metric(table, name, addr, value)
  if name == nil or name == '' then
    return
  end

  name = string.gsub(name, ' ', '\\ ')

  local url = 'http://IP:8086/api/v2/write?bucket=BUCKET&org=ORG'
  local body

  if type(value) == 'boolean' then
    body = string.format('%s,name=%s,addr=%s state=%s', table, name, addr, value)
  else
    -- most likely number
    body = string.format('%s,name=%s,addr=%s value=%s', table, name, addr, value)
  end

  local res, code = http.request({
    url = url,
    method = 'POST',
    body = body,
    headers = {
      Authorization = 'Token TOKEN'
    }
  })

  if code ~= 204 then
    log('error sending to influx', res, code, body)
  end
end

Thank you for responding Admin. I tried to put in your code in Postman just to verify that access is ok.
I can see to get response in postman I have to set Authorization to Oauth 2.0 and I get the header prefix Bearer. 

Do I replace Token with Bearer instead in the script? I didn't get this to work either... Suggestions?

[admin]

It should be Token not Bearer. That's what the documentation states and what worked for me. What kind of error are you getting? Have you checked that the user with this token has correct write access rights to the bucket?

[jensmagnar]

Hi, I have tried the influxdb V2 script above. Get's the following response in the log for all tags (here an example writing brightness value from weather station). Any ideas?

------------------
InfluxWrite 23.12.2021 15:50:52
* arg: 1
  * string: error sending to influx
* arg: 2
  * string: <html>
<head><title>405 Not Allowed</title></head>
<body>
<center><h1>405 Not Allowed</h1></center>
<hr><center>nginx/1.20.2</center>
</body>
</html>

* arg: 3
  * number: 405
* arg: 4
  * string: rawdata,name=Brightness,addr=0/0/3 value=328.96

[admin]

Check correctness of the url that you are sending to.

[jensmagnar]

Check correctness of the url that you are sending to.

Hi, I finally managed to write to my Influxdb. The url needs to be at this form: 'https://IP/api/v2/write?bucket=BUCKET&org=ORG'

[Re-G]

Check correctness of the url that you are sending to.

Hi.

How to set filter for addresses which are sending to influx. I would like to send just part of it (yes i know i can filter them in grafana) for example:

send address form range:

from 1/0/0 to 2/3/15 and from 4/6/7 to 5/1/15.

Can You help me.

Thanks.

[admin]

Replace this script part:

Code:

function knx_callback(event)
  local addr = event.dst
  local value, dt = get_value(addr, event.datahex)

  send_metric('rawdata', dt.name, addr, value)
end

With this:

Code:

local range1s = buslib.encodega('1/0/0')
local range1e = buslib.encodega('2/3/15')

local range2s = buslib.encodega('4/6/7')
local range2e = buslib.encodega('5/1/15')

function knx_callback(event)
  local id = event.dstraw

  if (range1s <= id and id <= range1e) or (range2s <= id and id <= range2e) then
    local addr = event.dst
    local value, dt = get_value(addr, event.datahex)
    send_metric('rawdata', dt.name, addr, value)
  end
end

[Re-G]

Replace this script part:

Code:

function knx_callback(event)
  local addr = event.dst
  local value, dt = get_value(addr, event.datahex)

  send_metric('rawdata', dt.name, addr, value)
end

With this:

Code:

local range1s = buslib.encodega('1/0/0')
local range1e = buslib.encodega('2/3/15')

local range2s = buslib.encodega('4/6/7')
local range2e = buslib.encodega('5/1/15')

function knx_callback(event)
  local id = event.dstraw

  if (range1s <= id and id <= range1e) or (range2s <= id and id <= range2e) then
    local addr = event.dst
    local value, dt = get_value(addr, event.datahex)
    send_metric('rawdata', dt.name, addr, value)
  end
end

thanks.

[sx3]

Hello,

I'm trying out this influx integration. I get the following error in the error-log.
It managed to write about 20 adresses to influx, before failing.
It's a resident script with 0 deley.

Resident script:15: attempt to index local 'dt' (a nil value)
stack traceback:
Resident script:15: in function 'get_value'
Resident script:20: in function <Resident script:18>
Library localbus: in function ''
Library localbus: in function ''
Library localbus: in function 'step'
Resident script:66: in function 'run_loop'

And the code (a little bit copy paste here and there from this thread)

Code:

local socket = require('socket')
local http = require('socket.http')
http.TIMEOUT = 5


local dt_cache = {}
function get_value(addr, datahex)
  local dt = dt_cache[addr]

  if(not dt) then
    dt = grp.find(addr)
    dt_cache[addr] = dt
  end
 
  return knxdatatype.decode(datahex, dt.datatype), dt
end

function knx_callback(event)
  local addr = event.dst
  local value, dt = get_value(addr, event.datahex)

  send_metric('rawdata', dt.name, addr, value)
end


function send_metric(table, name, addr, value)
  if name == nil or name == '' then
    return
  end

  name = string.gsub(name, ' ', '\\ ')

  local url = 'http://192.168.1.200:8086/api/v2/write?bucket=XXX&org=XXX'
  local body

  if type(value) == 'boolean' then
    body = string.format('%s,name=%s,addr=%s state=%s', table, name, addr, value)
  else
    -- most likely number
    body = string.format('%s,name=%s,addr=%s value=%s', table, name, addr, value)
  end

  local res, code = http.request({
    url = url,
    method = 'POST',
    body = body,
    headers = {
      Authorization = 'Token XXXXX'
    }
  })

  if code ~= 204 then
    log('error sending to influx', res, code, body)
  end
end

function run_loop()
  local bus = require('localbus').new(1)
  bus:sethandler('groupwrite', knx_callback)
  local busfd = socket.fdmaskset(bus:getfd(), 'r')

 
  while(true) do
    res, fbus = socket.selectfds(10, busfd)
    if(fbus) then
      bus:step()
    end
  end
end

run_loop()

[admin]

Most likely you have some group writes coming from addresses that are not present on LM.
Replace get_value/knx_callback functions with this and see if it works:

Code:

function get_value(addr, datahex)
  local obj = dt_cache[addr]

  if obj == nil then
    obj = grp.find(addr)
    dt_cache[ addr ] = obj or false
  end
  
  if obj then
    return knxdatatype.decode(datahex, obj.datatype), obj
  end
end

function knx_callback(event)
  local addr = event.dst
  local value, obj = get_value(addr, event.datahex)

  if value ~= nil then
    send_metric('rawdata', obj.name, addr, value)
  end
end

[sx3]

Yes, I have alot of GA that is beeing used outside om LM.

Tried your code but it gives me following error

Also wonder how the string would like like if I want to transmit a modified Timestamp, and where should it be placed?
For example (strTime = os.time() + 3600)

Code:

* arg: 1
  * string: error sending to influx
* arg: 2
  * string: {"code":"invalid","message":"unable to parse 'rawdata,name=Heru\\ -\\ Max\\ exhaust\\ fan\\ speed,\\ EC,addr=2/3/5 value=72': missing tag value"}
* arg: 3
  * number: 400
* arg: 4
  *

Code:

local socket = require('socket')
local http = require('socket.http')
http.TIMEOUT = 5


local dt_cache = {}
function get_value(addr, datahex)
  local obj = dt_cache[addr]

  if obj == nil then
    obj = grp.find(addr)
    dt_cache[ addr ] = obj or false
  end
 
  if obj then
    return knxdatatype.decode(datahex, obj.datatype), obj
  end
end

function knx_callback(event)
  local addr = event.dst
  local value, obj = get_value(addr, event.datahex)

  if value ~= nil then
    send_metric('rawdata', obj.name, addr, value)
  end
end


function send_metric(table, name, addr, value)
  if name == nil or name == '' then
    return
  end

  name = string.gsub(name, ' ', '\\ ')

  local url = 'http://192.168.1.200:8086/api/v2/write?bucket=xxx&org=xxx'
  local body

  if type(value) == 'boolean' then
    body = string.format('%s,name=%s,addr=%s state=%s', table, name, addr, value)
  else
    -- most likely number
    body = string.format('%s,name=%s,addr=%s value=%s', table, name, addr, value)
  end

  local res, code = http.request({
    url = url,
    method = 'POST',
    body = body,
    headers = {
      Authorization = 'Token XXXX'
    }
  })

  if code ~= 204 then
    log('error sending to influx', res, code, body)
  end
end

function run_loop()
  local bus = require('localbus').new(1)
  bus:sethandler('groupwrite', knx_callback)
  local busfd = socket.fdmaskset(bus:getfd(), 'r')

 
  while(true) do
    res, fbus = socket.selectfds(10, busfd)
    if(fbus) then
      bus:step()
    end
  end
end

run_loop()

[sx3]

Anyone have suggestions? Do I have too much "spaces" in my GA names?
And also the Timestamp would be interesting to learn more about.

[admin]

The problem is not with spaces but with comma character in the object name.