InfluxDB state history

Allows to store item states history in InfluxDB, v1 and v2 protocols are supported.

The service provides unified database EAPI.

Setup

Use the template EVA_DIR/share/svc-tpl/svc-tpl-db-influx.yml:

# Event-to-InfluxDB database service
command: svc/eva-db-influx
workers: 2
bus:
  path: var/bus.ipc
config:
  api_version: 1 # 1 or 2
  ## v1 basic auth
  # username: user
  # password: pass
  ## v2 token auth
  # token: sometoken
  url: http://localhost:8086
  db: eva
  ## required for v2
  # org: myorg
  # event buffer time-to-live, seconds
  buf_ttl_sec: null
  # periodically submit all item states
  interval: null
  # ignore real-time events
  ignore_events: false
  # event queue size
  queue_size: 8192
  # panic on critical errors in X seconds after happened (0 - panic immediately)
  panic_in: null
  # influx client pool size (default = workers)
  #clients: 2
  # item OIDs / OID masks
  oids:
    - "#"
user: nobody

Create the service using eva-shell:

eva svc create eva.db.i1 /opt/eva4/share/svc-tpl/svc-tpl-db-influx.yml

or using the bus CLI client:

cd /opt/eva4
cat DEPLOY.yml | ./bin/yml2mp | \
    ./sbin/bus ./var/bus.ipc rpc call eva.core svc.deploy -

(see eva.core::svc.deploy for more info)

EAPI methods

See EAPI commons for the common information about the bus, types, errors and RPC calls.

state_history

Description	Gets item state history
Parameters	required
Returns	State history payload

Parameters

Name	Type	Description	Required
i	String	Item OID	yes
t_start	f64	Beginning timestamp (default: last 24 hours)	no
t_end	f64	Ending timestamp (default: now)	no
fill	String	Fill (nS/T/H/D/W e.g. 10T for 10-minute)	no
precision	u32	Round values to digits after commma	no
limit	u32	Limit records to	no
xopts	Map<String, String>	Extra: vfn=fn for value grouping (d: mean), rp=X for retention policy	no
compact	bool	Pack data in arrays according to type	no

Return payload example:

[
    {
        "status": 1,
        "t": 1652059860.0424938,
        "value": 15
    },
    {
        "status": 1,
        "t": 1652059865.045223,
        "value": 15
    },
    {
        "status": 1,
        "t": 1652059870.0452943,
        "value": 15
    },
    {
        "status": 1,
        "t": 1652059875.0443518,
        "value": 15
    }
]

state_log

Description	Gets item state log
Parameters	required
Returns	State log payload (includes OIDs, as other svcs may support get-by-mask)

Parameters

Name	Type	Description	Required
i	String	Item OID	yes
t_start	f64	Beginning timestamp (default: last 24 hours)	no
t_end	f64	Ending timestamp (default: now)	no
limit	u32	Limit records to	no
xopts	Map<String, String>	Extra: rp=X for retention policy	no

Return payload example:

[
    {
        "oid": "sensor:tests/temp",
        "status": 1,
        "t": 1652060175.0443184,
        "value": 15
    },
    {
        "oid": "sensor:tests/temp",
        "status": 1,
        "t": 1652060180.046056,
        "value": 15
    },
    {
        "oid": "sensor:tests/temp",
        "status": 1,
        "t": 1652060185.0454304,
        "value": 15
    }
]

state_push

Description	push item states into db, (payload: single item state or list)
Parameters	none
Returns	nothing

Retention policies

In InfluxDB v1, retention policies can be created as the following:

CREATE RETENTION POLICY "daily" ON "eva" DURATION 1D REPLICATION 1
CREATE CONTINUOUS QUERY "downsampled_env_temp1_30m" ON "eva" BEGIN
  SELECT mode(status) as "status",mean(value) as value
  INTO "daily"."sensor:env/temp1"
  FROM "sensor:env/temp1"
  GROUP BY time(30m)
END

To process all items with the same downsampled rate, set the continuous query to:

SELECT mode(status) as status, mean(value) as value
INTO "daily".:MEASUREMENT
FROM /.*/ WHERE time > now() - 1d
GROUP BY time(30m);