InfluxDB: structured Flux pipelines

import { InfluxDBStore } from 'ts-prorm-orm/nosql/influxdb';

const store = new InfluxDBStore({ url: 'http://localhost:8086', token: process.env.INFLUX_TOKEN!, org: 'acme', bucket: 'telemetry' });
await store.connect();

const rows = await store.runFlux({
  range: { start: '-24h', stop: { now: true } },
  pipeline: [
    { op: 'filter', measurement: 'cpu', field: 'usage_user', tags: { host: 'web-1' } },
    { op: 'aggregateWindow', every: '5m', fn: 'mean', createEmpty: false },
    { op: 'sort', columns: ['_time'], desc: true },
    { op: 'limit', n: 100 },
  ],
  yield: 'cpu_5m',
});

from(bucket: "telemetry")
  |> range(start: -24h, stop: now())
  |> filter(fn: (r) => r._measurement == "cpu" and r._field == "usage_user" and r.host == "web-1")
  |> aggregateWindow(every: 5m, fn: mean, createEmpty: false)
  |> sort(columns: ["_time"], desc: true)
  |> limit(n: 100)
  |> yield(name: "cpu_5m")

import { Op } from 'ts-prorm-orm';
import { InfluxDBStore } from 'ts-prorm-orm/nosql/influxdb';

const store = new InfluxDBStore({ url: 'http://localhost:8086', token: process.env.INFLUX_TOKEN!, org: 'acme', bucket: 'telemetry' });
await store.connect();

try {
  // write two points; buffered until flush()
  store.writeMeasurement({ measurement: 'http', tags: { route: '/api/orders' }, fields: { duration_ms: 42, status: 200 }, intFields: ['status'] });
  store.writeMeasurement({ measurement: 'http', tags: { route: '/health' }, fields: { duration_ms: 1, status: 200 }, intFields: ['status'] });
  await store.flush();

  // 1. p95 latency per route in 15-minute windows
  const p95 = await store.runFlux({
    range: { start: '-6h' },
    pipeline: [
      { op: 'filter', measurement: 'http', field: 'duration_ms', where: { status: { [Op.gte]: 200 }, route: { [Op.ne]: '/health' } } },
      { op: 'group', columns: ['route'] },
      { op: 'aggregateWindow', every: '15m', fn: 'quantile', q: 0.95, createEmpty: false },
      { op: 'yield', name: 'p95' },
    ],
  });

  // 2. fields to columns, then a computed column
  const busy = await store.runFlux({
    range: { start: '-1h' },
    pipeline: [
      { op: 'filter', measurement: 'cpu', tags: { host: 'web-1' } },
      { op: 'pivot', rowKey: ['_time'], columnKey: ['_field'], valueColumn: '_value' },
      { op: 'map', columns: { busy: { arith: '-', left: { value: 100 }, right: { col: 'usage_idle' } } } },
      { op: 'keep', columns: ['_time', 'host', 'busy'] },
    ],
  });

  // 3. per-second rate of a counter, top five
  const top = await store.runFlux({
    range: { start: '-1h' },
    pipeline: [
      { op: 'filter', measurement: 'net', field: 'bytes_recv' },
      { op: 'derivative', unit: '1s', nonNegative: true },
      { op: 'top', n: 5, columns: ['_value'] },
    ],
  });

  console.log(p95.length, busy.length, top.length);
} finally {
  await store.disconnect(); // flushes and closes the write APIs
}

// 1.
from(bucket: "telemetry")
  |> range(start: -6h)
  |> filter(fn: (r) => r._measurement == "http" and r._field == "duration_ms" and (r.status >= 200.0 and r.route != "/health"))
  |> group(columns: ["route"])
  |> aggregateWindow(every: 15m, fn: (column, tables=<-) => tables |> quantile(q: 0.95, column: column), createEmpty: false)
  |> yield(name: "p95")

// 2.
from(bucket: "telemetry")
  |> range(start: -1h)
  |> filter(fn: (r) => r._measurement == "cpu" and r.host == "web-1")
  |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
  |> map(fn: (r) => ({r with busy: (100.0 - r.usage_idle)}))
  |> keep(columns: ["_time", "host", "busy"])

// 3.
from(bucket: "telemetry")
  |> range(start: -1h)
  |> filter(fn: (r) => r._measurement == "net" and r._field == "bytes_recv")
  |> derivative(unit: 1s, nonNegative: true)
  |> top(n: 5, columns: ["_value"])

interface FluxQuerySpec {
  bucket?: string;         // from(bucket:)   — falls back to the store's default
  bucketID?: string;       // from(bucketID:)
  union?: FluxQuerySpec[]; // union(tables: [...])
  join?: FluxJoinSpec;     // join(tables: {left, right}, on:)
  range?: FluxRange;       // range(start:, stop:) — required for a bucket source
  pipeline?: FluxStage[];  // the stages, in order
  yield?: string;          // shorthand for a trailing yield(name:)
}

compileFlux({
  union: [
    { bucket: 'a', range: { start: '-1h' }, pipeline: [{ op: 'filter', measurement: 'cpu' }] },
    { bucket: 'a', range: { start: '-1h' }, pipeline: [{ op: 'filter', measurement: 'mem' }] },
  ],
  pipeline: [{ op: 'sort', desc: true }],
});

_t0 = from(bucket: "a")
  |> range(start: -1h)
  |> filter(fn: (r) => r._measurement == "cpu")

_t1 = from(bucket: "a")
  |> range(start: -1h)
  |> filter(fn: (r) => r._measurement == "mem")

union(tables: [_t0, _t1])
  |> sort(columns: ["_time"], desc: true)

|> aggregateWindow(every: 1h, fn: (column, tables=<-) => tables |> quantile(q: 0.99, column: column))

{ op: 'filter', measurement: 'cpu', field: ['usage_user', 'usage_system'], tags: { host: 'web-1' } }

{ op: 'filter', where: { retries: { [Op.gte]: { int: 3 } } } }