From 23fe7d2ed23a303024e95ce349acd6e4d6d8c34a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Reinhard=20X=2E=20F=C3=BCrst?= Date: Thu, 20 Aug 2026 21:31:06 +0200 Subject: [PATCH] =?UTF-8?q?VictoriaMetrics=20als=20zus=C3=A4tzliche=20Date?= =?UTF-8?q?nbank-Option=20f=C3=BCr=20Messwerte=20erg=C3=A4nzt=20(STORE/DBA?= =?UTF-8?q?SE=3Dvictoria)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Dritte, zu mongo/influx exklusive Auswahl für die laufenden Messwerte. Schreibpfad nutzt das bestehende Influx-Line-Protocol unverändert (common/victoria_post.js); Lesepfad (sensorapi/databases/victoria.js + victoria2json.js) holt Rohdaten per VictoriaMetrics' /api/v1/export und bucketet/aggregiert stundenweise clientseitig, nach Mongo-Konvention (Stunden-Start als Label, kein Zeit-Shift nötig wie bei Influx). Scope bewusst auf die schon heute per DBASE umschaltbaren Funktionen begrenzt (getActData/getNoiseAVGData) - getAvgData/getLongAvg/getGeigerData bleiben wie bisher. Docker-Compose um victoriametrics-Service ergänzt (Retention explizit auf 100y gesetzt, da VictoriaMetrics sonst nach 1 Monat Daten löscht). Co-Authored-By: Claude Sonnet 5 --- .gitignore | 6 ++ README.md | 18 +++++- common/victoria_post.js | 55 ++++++++++++++++ noisesensors/.env.example | 5 +- noisesensors/docker-compose.yml | 24 +++++-- noisesensors/readarchive-stack.yml | 4 +- readarchive/readFromcsv.js | 20 +++++- readin/fetchnewdata.js | 11 +++- sensorapi/actions/getsensorData.js | 5 +- sensorapi/databases/victoria.js | 67 +++++++++++++++++++ sensorapi/sensorspecials/noise.js | 4 ++ sensorapi/utilities/victoria2json.js | 97 ++++++++++++++++++++++++++++ 12 files changed, 302 insertions(+), 14 deletions(-) create mode 100644 common/victoria_post.js create mode 100644 sensorapi/databases/victoria.js create mode 100644 sensorapi/utilities/victoria2json.js diff --git a/.gitignore b/.gitignore index 95300c5..e5e80a9 100644 --- a/.gitignore +++ b/.gitignore @@ -9,3 +9,9 @@ log/ # Produktionsstand vom Server, nur Referenz noisesensors/vonRemote/ + +# Laufzeitdaten der lokal getesteten Container (DB-Dateien, aktuelle Messwerte) +noisesensors/data/mongo/data/ +noisesensors/data/influx/ +noisesensors/data/victoria/ +noisesensors/data/aktdata.json diff --git a/README.md b/README.md index 07d30b9..93e19f7 100644 --- a/README.md +++ b/README.md @@ -20,6 +20,8 @@ Die Datei **noise.tgz** enthält die folgenden Dateien und Verzeichnisse: | | | +- create.js | + influx | | + data + | + victoria + | | + data : file, : directory ~~~ @@ -42,6 +44,10 @@ hier liegt die Datei **create.js**, mit deren Hilfe bei ersten Start der Datenba Verzeichnis für die Influx-Datenbank * **data/influx/data** Hier dann die eingelesenen Daten +* **data/victoria** +Verzeichnis für die VictoriaMetrics-Datenbank (optionale Alternative zu Influx, siehe unten) +* **data/victoria/data** +Hier dann die eingelesenen Daten ### Aufrufe @@ -60,6 +66,8 @@ STORE=influx ~~~ \, \, \, \ und \ sind anzupassen. +**STORE** legt fest, in welche Datenbank(en) *readin* die laufenden Messwerte schreibt: `influx`, `mongo`, `victoria` oder `both` (= mongo **und** influx gleichzeitig; Default). `victoria` ist dabei ein zu `influx` alternativer, exklusiver Wert (nicht über `both` kombinierbar). *sensorapi* liest die Messwerte unabhängig davon über **DBASE** (dieselben Werte `influx`/`mongo`/`victoria`, siehe Container-Beschreibung unten). + Danach einloggen in das Docker-Registry auf *citysensor.de*: ~~~ @@ -91,7 +99,7 @@ Im **Portainer** als eigenes Stack anlegen (Inhalt von `readarchive-stack.yml`) `-s` | `START` | gestern `-e` | `END` | Startdatum + 1 Tag (**exklusiv**) `-t` | `TYP` | `noise` - `-d` | `DBASE` | `both`, in der Stack-Datei auf `influx` vorbelegt + `-d` | `DBASE` | `mongo`, `influx`, `victoria` oder `both` (= mongo+influx); in der Stack-Datei auf `influx` vorbelegt *Deploy the stack* startet den Lauf, der Container endet danach und bleibt als „Exited" stehen. Für den nächsten Zeitraum nur die Variablen ändern und *Update the stack*. Der Haupt-Stack wird dabei nicht angefasst. @@ -143,13 +151,19 @@ Die Influx-Datenbank. Hier werden in dem Bucket *sensor_data* die reinen Messwe http://\:8086 Logindaten: Entsprechen der im Portainer hinterlegten. +* **victoriametrics** +Alternative zur Influx-Datenbank für die reinen Messwerte, wählbar über `STORE`/`DBASE=victoria`. Läuft als Single-Node-Instanz auf **Port 8428**, ohne Login/Auth. +**Wichtig:** VictoriaMetrics löscht standardmäßig Daten, die älter als 1 Monat sind (`-retentionPeriod=1`). Im `docker-compose.yml` ist das Flag deshalb explizit auf `--retentionPeriod=100y` gesetzt — das **nicht** entfernen, sonst verschwinden alte Messwerte nach 30 Tagen kommentarlos. +**Zugriff:** +http://\:8428 (z.B. `/api/v1/export?match[]={sid="..."}` zum Rohdaten-Export) + ## Sourcen für die Container Das gesamte Projekt ist im GitHub unter **laermsensor-stack** abgelegt. Für jeden der 4 Container existiert darunter ein Verzeichnis: **readin**, **readarchive**, **sensorapi** und **noise**. In diesen Verzeichnissen sind alle benötigten Sourcen enthalten. -Daneben gibt es **common** mit den Modulen, die sich *readin* und *readarchive* teilen: `mongo.js`, `influx_post.js`, `logit.js` und `statistics.js`. Sie lagen früher in beiden Komponenten doppelt und sind auseinandergelaufen. Wer dort etwas ändert, ändert es für beide — nach einer Änderung also **beide** Container neu bauen. +Daneben gibt es **common** mit den Modulen, die sich *readin* und *readarchive* teilen: `mongo.js`, `influx_post.js`, `victoria_post.js`, `logit.js` und `statistics.js`. Sie lagen früher in beiden Komponenten doppelt und sind auseinandergelaufen. Wer dort etwas ändert, ändert es für beide — nach einer Änderung also **beide** Container neu bauen. Sollte was geändert werden, so muss mit **deploy.sh** der Container neu erzeugt und auf die Registry (siehe oben) gepushed werden. Der Build-Kontext von *readin* und *readarchive* ist wegen `common/` das Wurzelverzeichnis des Repositories; `deploy.sh` wechselt selbst dorthin und kann aus dem Komponentenverzeichnis aufgerufen werden. diff --git a/common/victoria_post.js b/common/victoria_post.js new file mode 100644 index 0000000..845c843 --- /dev/null +++ b/common/victoria_post.js @@ -0,0 +1,55 @@ +/* Zugriff auf VictoriaMetrics per HTTP (InfluxDB-Line-Protocol-kompatibler Write-Endpunkt) + +Gemeinsam genutzt von readin (Live-API) und readarchive (CSV-Archiv). +*/ + +import axios from 'axios' +import { logit, logerror } from './logit.js' +import { statistics } from './statistics.js' +import { DateTime } from 'luxon' + +let VICTORIAHOST = process.env.VICTORIAHOST || "localhost" +let VICTORIAPORT = process.env.VICTORIAPORT || 8428 + +const VICTORIAURL_WRITE = `http://${VICTORIAHOST}:${VICTORIAPORT}/write?precision=ms` + +// `${e}` liefert bei einem AggregateError nur "AggregateError" - die einzelnen +// Verbindungsfehler (je einer pro aufgeloester IP) stecken in e.errors bzw. e.cause.errors +function describeError(e) { + const sub = e.errors || e.cause?.errors + if (sub?.length) { + return `${e.message || e.name}: ` + sub.map(s => `${s.code} ${s.address}:${s.port}`).join(', ') + } + if (e.response) { + return `HTTP ${e.response.status} ${JSON.stringify(e.response.data)}` + } + return `${e.code ? e.code + ' ' : ''}${e.message || e}` +} + +// liefert true, wenn VictoriaMetrics die Daten uebernommen hat, sonst false +export const victoriaWrite = async (data) => { + let start = DateTime.now() + let ok = false + try { + const ret = await axios({ + method: 'post', + url: VICTORIAURL_WRITE, + data: data, + headers: { + Accept: 'application/json', + 'Content-Type': 'text/plain; charset=utf-8' + }, + timeout: 10000, + }) + if (ret.status != 204) { + logerror(`doWrite2API Status: ${ret.status}`) + } else { + ok = true + } + } catch (e) { + logerror(`doWrite2API ${VICTORIAURL_WRITE} ${describeError(e)}`) + } + statistics['writeVictoriaData[sensor_data]Time'] = DateTime.now().diff(start, ['seconds']).toObject().seconds + logit(`Victoria-Write-Time: ${start.diffNow('seconds').toObject().seconds * -1} sec`) + return ok +} diff --git a/noisesensors/.env.example b/noisesensors/.env.example index d124690..c99e312 100644 --- a/noisesensors/.env.example +++ b/noisesensors/.env.example @@ -9,9 +9,12 @@ MONGO_ROOT_PASSWORD= LOCALDIR= -# Datenbank, in welche die laufenden Messwerte gespeichert werden (kannn 'influx' oder 'mongo' sein) +# Datenbank, in welche die laufenden Messwerte gespeichert werden (kann 'influx', 'mongo' oder 'victoria' sein) STORE=influx +# Datenbank, aus welcher sensorapi die Messwerte liest (kann 'influx', 'mongo' oder 'victoria' sein) +DBASE=influx + # Secret zum Signieren der Session-Cookies von esp2sensor (langer Zufallsstring) ESP2SENSOR_SESSION_SECRET= diff --git a/noisesensors/docker-compose.yml b/noisesensors/docker-compose.yml index 44ad5ed..e825c83 100644 --- a/noisesensors/docker-compose.yml +++ b/noisesensors/docker-compose.yml @@ -26,20 +26,33 @@ services: DOCKER_INFLUXDB_INIT_ORG: citysensor DOCKER_INFLUXDB_INIT_BUCKET: sensor_data DOCKER_INFLUXDB_INIT_ADMIN_TOKEN: ${DOCKER_INFLUXTOKEN} - restart: - unless-stopped - container_name: influxdb + restart: + unless-stopped + container_name: influxdb + + victoriametrics: + image: victoriametrics/victoria-metrics:v1.102.0 + ports: + - "8428:8428" + volumes: + - ${LOCALDIR}/noisesensors/data/victoria/data:/victoria-metrics-data + command: + - "--storageDataPath=/victoria-metrics-data" + - "--retentionPeriod=100y" + restart: unless-stopped + container_name: victoriametrics readin: image: docker.citysensor.de/readin environment: MONGOHOST: mongodb INFLUXHOST: influxdb + VICTORIAHOST: victoriametrics TYP: "[\"noise\"]" STORE: ${STORE:-mongo} MONGOAUTH: true MONGOUSRP: ${MONGO_ROOT_USERNAME}:${MONGO_ROOT_PASSWORD} - INFLUXTOKEN: ${DOCKER_INFLUXTOKEN} + INFLUXTOKEN: ${DOCKER_INFLUXTOKEN} volumes: - ${LOCALDIR}/noisesensors/log:/var/log # Pfad im Image ist jetzt /opt/app/readin, weil readin und readarchive @@ -67,7 +80,8 @@ services: environment: INFLUXHOST: influxdb INFLUXTOKEN: ${DOCKER_INFLUXTOKEN} - DBASE: influx + VICTORIAHOST: victoriametrics + DBASE: ${DBASE:-influx} MONGOHOST: mongodb MONGOAUTH: true MONGOUSRP: ${MONGO_ROOT_USERNAME}:${MONGO_ROOT_PASSWORD} diff --git a/noisesensors/readarchive-stack.yml b/noisesensors/readarchive-stack.yml index 1ef7104..c8e18ac 100644 --- a/noisesensors/readarchive-stack.yml +++ b/noisesensors/readarchive-stack.yml @@ -16,7 +16,8 @@ # # readFromcsv.js liest jeden Parameter auch aus einer Env-Variablen: # -s = START (Default: gestern), -e = END (Default: Startdatum + 1 Tag, -# exklusiv), -t = TYP (Default: noise), -d = DBASE (Default: both), +# exklusiv), -t = TYP (Default: noise), -d = DBASE ('mongo', 'influx', +# 'victoria' oder 'both' = mongo+influx; Default: both), # -p = PARALLEL (Default: 4) - so viele Sensoren werden gleichzeitig geholt. # # Voraussetzungen: @@ -39,6 +40,7 @@ services: environment: MONGOHOST: mongodb INFLUXHOST: influxdb + VICTORIAHOST: victoriametrics MONGOAUTH: "true" MONGOUSRP: ${MONGO_ROOT_USERNAME}:${MONGO_ROOT_PASSWORD} INFLUXTOKEN: ${DOCKER_INFLUXTOKEN} diff --git a/readarchive/readFromcsv.js b/readarchive/readFromcsv.js index b882521..b08974b 100644 --- a/readarchive/readFromcsv.js +++ b/readarchive/readFromcsv.js @@ -33,7 +33,7 @@ const DBASE = process.env.DBASE || 'both' // Mehr als 4 bringt nichts: ab da liefert das Archiv nicht mehr schneller. const PARALLEL = parseInt(process.env.PARALLEL) || 4 -const DATABASE = ['mongo', 'influx', 'both'] +const DATABASE = ['mongo', 'influx', 'victoria', 'both'] const DEVELOP = process.env.DEVELOP || false; const LIVE = (process.env.LIVE == "true") || true @@ -47,6 +47,7 @@ import axios from 'axios' import https from 'https' import * as mongo from '../common/mongo.js' import * as influx from '../common/influx_post.js' +import * as victoria from '../common/victoria_post.js' import fs from 'fs' import csv from 'csvtojson' import pkg from './package.json' with { type: "json" } @@ -67,6 +68,8 @@ const newProps = [] // Sensor/Tag-Paare, deren Influx-Write fehlgeschlagen ist - Auswertung am Ende von main() const failedInfluxWrites = [] +// Sensor/Tag-Paare, deren VictoriaMetrics-Write fehlgeschlagen ist - Auswertung am Ende von main() +const failedVictoriaWrites = [] const saveList = (list) => { fs.writeFileSync('data/list.json', JSON.stringify(list)) @@ -397,6 +400,12 @@ async function enterOneSensorinDB(client, sid, erg, dataline, day, styp, args) { failedInfluxWrites.push({sensorid: sid, day: day}) } } + if ((args.database === 'victoria') && (dataline !== '')) { + const ok = await victoria.victoriaWrite(dataline) + if (!ok) { + failedVictoriaWrites.push({sensorid: sid, day: day}) + } + } if((args.database === 'both') || (args.database === 'mongo')) { if(erg.length !== 0) { await mongo.writeDataArray(client, mongo.dataCollName(styp), erg) @@ -507,7 +516,7 @@ function parse_cmdline(argv) { console.log(" will be used and endDate = startDate + 1 day") console.log(" -t sensorType: if given, only those sensors will be used; default: 'noise'"); console.log(" allowed types: 'noise', 'radiactivity', 'pm' and 'thp'") - console.log(" -d database: 'mongo', 'influx' or 'both'; default: 'both'") + console.log(" -d database: 'mongo', 'influx', 'victoria' or 'both' ('both' = mongo + influx); default: 'both'") console.log(` -p parallel: number of sensors fetched at the same time; default: ${PARALLEL}`) console.log(" -v version: show version"); console.log(" -h this help text"); @@ -555,6 +564,13 @@ async function main() { } process.exitCode = 1 // damit cron/docker den Fehlschlag sieht } + if (failedVictoriaWrites.length > 0) { + logerror(`${failedVictoriaWrites.length} Victoria-Writes fehlgeschlagen - diese Daten fehlen in VictoriaMetrics:`) + for (const f of failedVictoriaWrites) { + logerror(` Sensor ${f.sensorid}, Tag ${f.day}`) + } + process.exitCode = 1 // damit cron/docker den Fehlschlag sieht + } let dauer = starttime.diffNow('seconds').toObject().seconds * -1 const duration1 = Duration.fromObject({ seconds: dauer }); const output1 = duration1.toFormat('hh:mm:ss'); diff --git a/readin/fetchnewdata.js b/readin/fetchnewdata.js index 256070d..4e6910b 100644 --- a/readin/fetchnewdata.js +++ b/readin/fetchnewdata.js @@ -14,13 +14,14 @@ // TYP: Ist ein Array von Strings, wenn nur einzelne Typen gespeichert werden sollen, also z.B.: // ['pm', 'noise'] const TYP = process.env.TYP || '' -const STORE = (process.env.STORE || 'both').toLowerCase() // 'mongo' | 'influx' | 'both' +const STORE = (process.env.STORE || 'both').toLowerCase() // 'mongo' | 'influx' | 'victoria' | 'both' ('both' = mongo + influx) import { doReadfromAPI as readin } from './readdata.js' import { constructDBaseEntries as parse} from './parse.js' import * as mongo from '../common/mongo.js' import * as influx from '../common/influx_post.js' +import * as victoria from '../common/victoria_post.js' import { logit, logerror } from '../common/logit.js' import { statistics } from '../common/statistics.js' import { DateTime } from 'luxon' @@ -62,6 +63,11 @@ const fetchNewData = async (args) => { await influx.influxWrite(idata) } + // write sensor data to VictoriaMetrics + if(args.victoria) { + await victoria.victoriaWrite(idata) + } + // write properties to mongoDB await mongo.bulkWrite(client, mongo.property_coll, props) } @@ -78,7 +84,7 @@ const fetchNewData = async (args) => { function parse_cmdline(argv) { let parser = new mod_getopt.BasicParser('i(influx)m(mongo)t:(typ)h(help)v(version)',argv); let option; - let ret = {influx: STORE === 'both' || STORE === 'influx', mongo: STORE === 'both' || STORE === 'mongo', typ: TYP} + let ret = {influx: STORE === 'both' || STORE === 'influx', mongo: STORE === 'both' || STORE === 'mongo', victoria: STORE === 'victoria', typ: TYP} while((option = parser.getopt()) !== undefined) { switch(option.option) { case 'i': @@ -111,6 +117,7 @@ function parse_cmdline(argv) { console.log("Params:"); console.log(" -i use only InfluxDB to store the data (default use both, InfluxDB and MongoDB))"); console.log(" -m use only MongoDB to store the data (default use both, InfluxDB and MongoDB)"); + console.log(" (VictoriaMetrics storage is selected exclusively via STORE=victoria, no CLI flag)"); console.log(" -t [ sensorType, ..]: if given, only those sensors will be used (ex: laerm) default: all"); console.log(" MUST BE AN ARRAY!; allowed types: 'pm', 'noise', 'radiactivity', 'thp', 'gps'.") console.log(" -v version: show version"); diff --git a/sensorapi/actions/getsensorData.js b/sensorapi/actions/getsensorData.js index c8e5ece..96d6115 100644 --- a/sensorapi/actions/getsensorData.js +++ b/sensorapi/actions/getsensorData.js @@ -4,6 +4,7 @@ const DBASE = process.env.DBASE || 'mongo' import {DateTime} from "luxon" import * as influx from "../databases/influx.js" import * as mongo from "../databases/mongo.js" +import * as victoria from "../databases/victoria.js" import {returnOnError} from "../utilities/reporterror.js" import {csv2Json} from "../utilities/csv2json.js" import checkParams from "../utilities/checkparams.js" @@ -155,8 +156,10 @@ export async function getActData(opts) { return await mongo.fetchActData(opts) } else if (DBASE === 'influx') { return await influx.fetchActData(opts) + } else if (DBASE === 'victoria') { + return await victoria.fetchActData(opts) } - return {err: 'DBASEUNKNOWN', values: []} + return {err: 'DBASEUNKNOWN', values: []} } diff --git a/sensorapi/databases/victoria.js b/sensorapi/databases/victoria.js new file mode 100644 index 0000000..c319185 --- /dev/null +++ b/sensorapi/databases/victoria.js @@ -0,0 +1,67 @@ +// Access to VictoriaMetrics via HTTP (Prometheus-compatible /api/v1/export, +// no Flux support - see common/victoria_post.js for the write side, which +// reuses the InfluxDB line-protocol format on VictoriaMetrics' /write endpoint). + +import axios from 'axios' +import { logit, logerror } from '../utilities/logit.js' +import { returnOnError } from "../utilities/reporterror.js" +import { exportToRows, bucketNoiseAVG } from "../utilities/victoria2json.js" + +let VICTORIAHOST = process.env.VICTORIAHOST || "localhost" +let VICTORIAPORT = process.env.VICTORIAPORT || 8428 + +const VICTORIAURL_EXPORT = `http://${VICTORIAHOST}:${VICTORIAPORT}/api/v1/export` + +// opts.start/opts.stop arrive as Flux range() fragments ("start: " / +// "stop: ") built by calcRange() in getsensorData.js - strip the Flux +// keyword the same way sensorapi/databases/mongo.js already does. +const isoStart = (opts) => opts.start.slice(7) +const isoStop = (opts) => opts.stop.slice(6) + +const victoriaExport = async (matchSelector, start, end) => { + let erg = { values: '', err: null } + try { + let ret = await axios({ + method: 'get', + url: VICTORIAURL_EXPORT, + params: { 'match[]': matchSelector, start, end }, + timeout: 10000, + transformResponse: [(data) => data], // response body is ndjson, not a single JSON document - keep it raw + }) + if (ret.status !== 200) { + return returnOnError(erg, 'RESPSTATUS', victoriaExport.name, ret.status) + } + erg.values = ret.data + } catch (e) { + return returnOnError(erg, e, victoriaExport.name) + } + return erg +} + +export const fetchActData = async (opts) => { + let ret = { err: null, values: [] } + const match = `{__name__=~"noise_(LAeq|LA_min|LA_max|E10tel_eq)", sid="${opts.sensorid}"}` + let { values, err } = await victoriaExport(match, isoStart(opts), isoStop(opts)) + if (err) { + return returnOnError(ret, err, fetchActData.name) + } + ret.values = exportToRows(values, opts.sort) + if (ret.values.length === 0) { + return returnOnError(ret, 'NODATA', fetchActData.name) + } + return ret +} + +export const fetchNoiseAVGData = async (opts) => { + let ret = { err: null, values: [] } + const match = `{__name__=~"noise_(E10tel_eq|LA_max)", sid="${opts.sensorid}"}` + let { values, err } = await victoriaExport(match, isoStart(opts), isoStop(opts)) + if (err) { + return returnOnError(ret, err, fetchNoiseAVGData.name) + } + ret.values = bucketNoiseAVG(values, opts.peak, opts.long) + if (ret.values.length === 0) { + return returnOnError(ret, 'NODATA', fetchNoiseAVGData.name) + } + return ret +} diff --git a/sensorapi/sensorspecials/noise.js b/sensorapi/sensorspecials/noise.js index 6d37807..f11c81c 100644 --- a/sensorapi/sensorspecials/noise.js +++ b/sensorapi/sensorspecials/noise.js @@ -9,6 +9,7 @@ import {DateTime} from 'luxon' import { translate as trans } from '../routes/api.js' import * as influx from "../databases/influx.js" import * as mongo from "../databases/mongo.js" +import * as victoria from "../databases/victoria.js" import { setoptionfromtable } from "../utilities/chartoptions.js" export const getNoiseData = async (params, possibles, props) => { @@ -451,6 +452,9 @@ const getNoiseAVGData = async (opts) => { for (let x=0; x < ret.values.length; x++) { ret.values[x].datetime = DateTime.fromISO(ret.values[x].datetime).toUTC().minus({hours:1}).toFormat("yyyy-LL-dd'T'HH:mm:ss'Z'") } + } else if (DBASE === 'victoria') { + // victoria.js buckets by hour-start itself (like mongo), so no shift is needed here + ret = await victoria.fetchNoiseAVGData(opts) } else { ret.err = 'DBASEUNKNOWN' } diff --git a/sensorapi/utilities/victoria2json.js b/sensorapi/utilities/victoria2json.js new file mode 100644 index 0000000..aba0325 --- /dev/null +++ b/sensorapi/utilities/victoria2json.js @@ -0,0 +1,97 @@ +// Parse VictoriaMetrics' /api/v1/export ndjson responses into the same row +// shapes influx.js builds via csv2Json() / pivot(), for the noise-specific +// fields (measurement "noise", separator "_", i.e. metric names noise_LAeq, +// noise_LA_min, noise_LA_max, noise_E10tel_eq). + +import { DateTime } from 'luxon' + +const parseExportLines = (ndjsonBody) => { + return String(ndjsonBody) + .split('\n') + .filter((line) => line.trim() !== '') + .map((line) => { + try { + return JSON.parse(line) + } catch (e) { + return null + } + }) + .filter((series) => series !== null) +} + +const fieldName = (metricName) => metricName.replace(/^noise_/, '') + +// Merges the per-field series (one ndjson line per field) back into rows of +// {datetime, LAeq, LA_min, LA_max, E10tel_eq}, matched by raw timestamp - +// mirrors influx's pivot(rowKey:["_time"], columnKey:["_field"]). +export const exportToRows = (ndjsonBody, sort) => { + const rowsByTime = new Map() + for (const series of parseExportLines(ndjsonBody)) { + const field = fieldName(series.metric.__name__) + const { values, timestamps } = series + for (let i = 0; i < timestamps.length; i++) { + const ts = timestamps[i] + let row = rowsByTime.get(ts) + if (!row) { + row = { datetime: DateTime.fromMillis(ts).toUTC().toISO() } + rowsByTime.set(ts, row) + } + row[field] = values[i] + } + } + const rows = Array.from(rowsByTime.values()) + rows.sort((a, b) => (a.datetime < b.datetime ? -1 : a.datetime > b.datetime ? 1 : 0)) + if (sort === -1) { + rows.reverse() + } + return rows +} + +// Buckets raw E10tel_eq/LA_max samples by hour (truncated to the start of the +// hour of each sample, same convention as sensorapi/databases/mongo.js's +// $dateToString hour-grouping) and computes n_AVG/n_sum/count/peakcount - +// equivalent to influx.js's aggregateWindow()/reduce() Flux pipeline. +// Bucketing by hour-start (not hour-end, as Influx does) means no extra +// 1-hour shift is needed downstream, unlike the DBASE === 'influx' branch. +export const bucketNoiseAVG = (ndjsonBody, peak, long) => { + const buckets = new Map() + for (const series of parseExportLines(ndjsonBody)) { + const field = fieldName(series.metric.__name__) + if (field !== 'E10tel_eq' && field !== 'LA_max') { + continue + } + const { values, timestamps } = series + for (let i = 0; i < timestamps.length; i++) { + const key = DateTime.fromMillis(timestamps[i]).toUTC().startOf('hour').toISO() + let b = buckets.get(key) + if (!b) { + b = { sum: 0, count: 0, peakcount: 0 } + buckets.set(key, b) + } + if (field === 'E10tel_eq') { + b.sum += values[i] + b.count += 1 + } else { + if (values[i] >= peak) { + b.peakcount += 1 + } + } + } + } + const rows = Array.from(buckets.entries()) + .filter(([, b]) => b.count > 0) // matches Influx's inner join: hours without E10tel_eq samples are dropped + .map(([datetime, b]) => { + let row = { + datetime, + n_AVG: 10 * Math.log10(b.sum / b.count), + peakcount: b.peakcount, + } + if (long) { + row.count = b.count + row.n_sum = b.sum + } + return row + }) + rows.sort((a, b) => (a.datetime < b.datetime ? -1 : a.datetime > b.datetime ? 1 : 0)) + return rows +}