stroblmeandClaude Opus 5
f00045d6b6
The Influx client was built with no timeout, so every query and write fell through to influxdb-client's own 10 s default — invisible to a flow and unchangeable. The param is in seconds like its peers; the client counts in milliseconds, so the call sites convert. The publisher backlog was a module constant, read once at import. It is the depth at which the oldest payload is dropped and the node goes degraded, and a node that bursts wants more than one that trickles, so it moves to Params and is read where the queue is built. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01K1moruzue2kTJd3uVisgNk
610 lines
23 KiB
Python
610 lines
23 KiB
Python
"""InfluxDB nodes: write points from messages, or read a query into them."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from collections.abc import Iterable
|
|
from typing import Any
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
|
|
from fluksio.flow.messages import MessageSpec
|
|
from fluksio.flow.nodes.base import Node, NodeResult
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class InfluxDbNode(Node):
|
|
"""
|
|
InfluxDB node for writing to and reading from InfluxDB.
|
|
|
|
This node can perform both write and read operations independently:
|
|
|
|
- **Write operation**: Triggered when upstream dependencies are satisfied
|
|
(data flows in via ``requires``). Writes data points to InfluxDB based
|
|
on the ``writes`` configuration in params.
|
|
- **Read operation**: Performed when the node provides data to downstream
|
|
nodes via ``provides``, based on the ``queries`` configuration in params.
|
|
|
|
Both operations use a similar configuration pattern in params, making the
|
|
API consistent and the input/output data simple (just values).
|
|
|
|
- **Query passthrough**: an incoming message holding a ``flux`` key is run
|
|
as written, and every row comes back on the first output port as
|
|
``{"rows": [{ts, value, field, measurement, tags}], **echo}``.
|
|
|
|
The passthrough is what keeps a database node a database node. It holds the
|
|
credentials and the connection and nothing else: building a query and
|
|
shaping its rows are ordinary Python nodes on either side, so a dashboard
|
|
widget never learns which database answered it. A chart drawing an
|
|
InfluxDB series is::
|
|
|
|
[chart] -record-> [py: build] -record{flux, ...}-> [influx]
|
|
[chart] <-series- [py: parse] <-json{rows, ...}---'
|
|
|
|
Nothing rate-limits the requests here; an ``interval`` on the request port
|
|
is what holds back a caller asking too often.
|
|
|
|
:param requires: Messages to write to InfluxDB. The actual value from each
|
|
message is written according to the corresponding config in ``writes``.
|
|
:type requires: MessageSpec | list[MessageSpec] | None
|
|
:param provides: Messages to read from InfluxDB. Each message gets its value
|
|
from a query defined in ``queries``.
|
|
:type provides: MessageSpec | list[MessageSpec] | None
|
|
:param params: Parameters dict containing:
|
|
- ``url`` (str): InfluxDB server URL (required)
|
|
- ``token`` (str): Authentication token (required)
|
|
- ``org`` (str): Organization name (required)
|
|
- ``bucket`` (str): Bucket name (required)
|
|
- ``write_precision`` (str): Write precision ("ns", "us", "ms", "s"), default "ms"
|
|
- ``query_range`` (str): Default time range for queries, e.g., "-1h", "-24h"
|
|
- ``timeout`` (float): Deadline for a request in seconds (default: 10.0)
|
|
- ``writes`` (dict): Write configurations keyed by message name, each with:
|
|
- ``measurement`` (str): Measurement name to write to
|
|
- ``field`` (str): Field name to write (default: "value")
|
|
- ``tags`` (dict): Static tags to add to each point
|
|
- ``queries`` (dict): Query configurations keyed by message name, each with:
|
|
- ``measurement`` (str): Measurement name to query
|
|
- ``field`` (str): Field name to retrieve (default: "value")
|
|
- ``tags`` (dict): Optional tag filters
|
|
- ``range`` (str): Optional time range override
|
|
- ``aggregation`` (str): Aggregation function ("mean", "last", "first", "max", "min")
|
|
:type params: dict
|
|
:param name: Optional name for the node.
|
|
:type name: str | None
|
|
|
|
:raises ValueError: If required params are missing or both requires and provides are empty.
|
|
|
|
:example:
|
|
Write-only node (writes temperature values):
|
|
|
|
>>> writer = InfluxDbNode(
|
|
... requires=[MessageSpec(name="temperature", dtype=DType.FLOAT)],
|
|
... params={
|
|
... "url": "http://localhost:8086",
|
|
... "token": "my-token",
|
|
... "org": "my-org",
|
|
... "bucket": "sensors",
|
|
... "writes": {
|
|
... "temperature": {
|
|
... "measurement": "environment",
|
|
... "field": "temp_celsius",
|
|
... "tags": {"location": "room1", "sensor": "dht22"},
|
|
... }
|
|
... },
|
|
... },
|
|
... )
|
|
|
|
Read-only node (queries average temperature):
|
|
|
|
>>> reader = InfluxDbNode(
|
|
... provides=[MessageSpec(name="avg_temperature", dtype=DType.FLOAT)],
|
|
... params={
|
|
... "url": "http://localhost:8086",
|
|
... "token": "my-token",
|
|
... "org": "my-org",
|
|
... "bucket": "sensors",
|
|
... "queries": {
|
|
... "avg_temperature": {
|
|
... "measurement": "environment",
|
|
... "field": "temp_celsius",
|
|
... "tags": {"location": "room1"},
|
|
... "range": "-1h",
|
|
... "aggregation": "mean",
|
|
... }
|
|
... },
|
|
... },
|
|
... )
|
|
|
|
Combined read/write node:
|
|
|
|
>>> node = InfluxDbNode(
|
|
... requires=[MessageSpec(name="raw_temp", dtype=DType.FLOAT)],
|
|
... provides=[MessageSpec(name="avg_temp", dtype=DType.FLOAT)],
|
|
... params={
|
|
... "url": "http://localhost:8086",
|
|
... "token": "my-token",
|
|
... "org": "my-org",
|
|
... "bucket": "sensors",
|
|
... "writes": {
|
|
... "raw_temp": {
|
|
... "measurement": "temperature",
|
|
... "field": "value",
|
|
... "tags": {"source": "sensor"},
|
|
... }
|
|
... },
|
|
... "queries": {
|
|
... "avg_temp": {
|
|
... "measurement": "temperature",
|
|
... "field": "value",
|
|
... "aggregation": "mean",
|
|
... "range": "-5m",
|
|
... }
|
|
... },
|
|
... },
|
|
... )
|
|
"""
|
|
|
|
# Writing the same point twice doubles it in the series.
|
|
idempotent = False
|
|
|
|
__slots__ = (
|
|
"url",
|
|
"token",
|
|
"org",
|
|
"bucket",
|
|
"write_precision",
|
|
"query_range",
|
|
"timeout",
|
|
"writes",
|
|
"queries",
|
|
"_write_client",
|
|
"_query_client",
|
|
)
|
|
|
|
class Params(BaseModel):
|
|
model_config = ConfigDict(extra="allow")
|
|
|
|
url: str
|
|
token: str = Field(json_schema_extra={"x-secret": True})
|
|
org: str
|
|
bucket: str
|
|
write_precision: str = "ms"
|
|
query_range: str = "-1h"
|
|
# Bounds every request to the server. Without one the client falls
|
|
# back to its own default, which no flow can see or change. Held in
|
|
# seconds like every other node; the client counts in milliseconds.
|
|
timeout: float = Field(
|
|
default=10.0,
|
|
gt=0,
|
|
description="Give up on a query or write after this many seconds.",
|
|
)
|
|
# Per-port write and query configuration.
|
|
writes: dict[str, dict[str, Any]] = {}
|
|
queries: dict[str, dict[str, Any]] = {}
|
|
|
|
@classmethod
|
|
def instance_key(cls, params: dict[str, Any]) -> str | None:
|
|
"""The bucket, which is the thing several flows share."""
|
|
url, bucket = params.get("url"), params.get("bucket")
|
|
return f"{url}/{bucket}" if url and bucket else None
|
|
|
|
def __init__(
|
|
self,
|
|
requires: MessageSpec | Iterable[MessageSpec] = (),
|
|
provides: MessageSpec | Iterable[MessageSpec] = (),
|
|
params: dict[str, Any] | None = None,
|
|
name: str | None = None,
|
|
):
|
|
cfg = self.Params.model_validate(params or {})
|
|
|
|
requires = Node._normalize_ports(requires)
|
|
provides = Node._normalize_ports(provides)
|
|
if not requires and not provides:
|
|
raise ValueError(
|
|
"An InfluxDB node needs either inputs (to write) or outputs (to read)"
|
|
)
|
|
|
|
self.url = cfg.url
|
|
self.token = cfg.token
|
|
self.org = cfg.org
|
|
self.bucket = cfg.bucket
|
|
self.write_precision = cfg.write_precision
|
|
self.query_range = cfg.query_range
|
|
self.timeout = cfg.timeout
|
|
self.writes = cfg.writes
|
|
self.queries = cfg.queries
|
|
|
|
# Lazy-initialized clients
|
|
self._write_client = None
|
|
self._query_client = None
|
|
|
|
# Set default name
|
|
if name is None:
|
|
name = f"influxdb_{self.bucket}"
|
|
|
|
# Initialize parent
|
|
# The handler function depends on what operations are configured
|
|
super().__init__(
|
|
f=self._handler,
|
|
requires=requires,
|
|
provides=provides,
|
|
params=params,
|
|
name=name,
|
|
)
|
|
|
|
def _handler(self, params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None:
|
|
"""
|
|
Handle incoming data - write to InfluxDB, run a query, or both.
|
|
|
|
This method is called when upstream dependencies (requires) are satisfied.
|
|
|
|
An incoming dict carrying a ``flux`` key is a query request rather than
|
|
something to store: it is executed as written and answered on the first
|
|
output port. Everything else on the inputs is written as points.
|
|
|
|
:param params: Node parameters.
|
|
:type params: dict
|
|
:param kwargs: Incoming data from upstream nodes.
|
|
:type kwargs: Any
|
|
:returns: Query results if provides is configured, None otherwise.
|
|
:rtype: dict | None
|
|
"""
|
|
# ponytail: one request/answer pair per node — the first input holding
|
|
# a "flux" key is the request and the answer goes out on the first
|
|
# output port. A second query stream means a second node.
|
|
request = next(
|
|
(v for v in kwargs.values() if isinstance(v, dict) and "flux" in v), None
|
|
)
|
|
writes = {k: v for k, v in kwargs.items() if v is not request}
|
|
|
|
if writes:
|
|
self._write_points(writes)
|
|
|
|
if request is not None:
|
|
if not self.output_ports:
|
|
raise ValueError(
|
|
f"Node '{self.name}' was sent a query but has no output to "
|
|
"answer on"
|
|
)
|
|
return {self.output_ports[0].port: self._run_flux(request)}
|
|
|
|
# If we have provides, perform queries
|
|
if self.provides:
|
|
return self._query_data()
|
|
|
|
return None
|
|
|
|
def _run_flux(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
"""
|
|
Execute a Flux query exactly as it was handed over, and answer its rows.
|
|
|
|
The node stays out of the query's business: building it is the job of
|
|
whatever produced the request, and shaping the rows into something a
|
|
widget draws is the job of whatever reads the answer. Every field of
|
|
the request except ``flux`` is echoed back untouched, which is how a
|
|
caller tells its own answer from someone else's.
|
|
|
|
:param request: The query and whatever the caller wants echoed.
|
|
:type request: dict
|
|
:returns: ``{"rows": [...], **echo}``.
|
|
:rtype: dict
|
|
"""
|
|
from influxdb_client import InfluxDBClient
|
|
|
|
flux = str(request["flux"])
|
|
echo = {key: value for key, value in request.items() if key != "flux"}
|
|
|
|
logger.info("Running Flux for node '%s': %s", self.name, flux)
|
|
with InfluxDBClient(
|
|
url=self.url,
|
|
token=self.token,
|
|
org=self.org,
|
|
timeout=int(self.timeout * 1000),
|
|
) as client:
|
|
tables = client.query_api().query(flux, org=self.org)
|
|
|
|
rows = [
|
|
{
|
|
"ts": record.get_time().timestamp() if record.get_time() else None,
|
|
"value": record.get_value(),
|
|
# Read off `values` rather than the getters: a query that keeps
|
|
# only what it needs drops these columns, and the getters raise.
|
|
"field": record.values.get("_field"),
|
|
"measurement": record.values.get("_measurement"),
|
|
"tags": {
|
|
key: value
|
|
for key, value in record.values.items()
|
|
if not key.startswith("_") and key not in ("result", "table")
|
|
},
|
|
}
|
|
for table in tables
|
|
for record in table.records
|
|
]
|
|
return {"rows": rows, **echo}
|
|
|
|
def _write_points(self, data: dict[str, Any]) -> None:
|
|
"""
|
|
Write data points to InfluxDB using configuration from ``writes``.
|
|
|
|
The write configuration is looked up in ``self.writes`` by message name.
|
|
Each config specifies the measurement, field, and tags. The actual value
|
|
comes from the incoming data.
|
|
|
|
:param data: Data to write, keyed by port name. Values can be:
|
|
- Simple values (float, int, str, bool): Written using config from ``writes``
|
|
- List of values: Each value written as a separate point
|
|
- Dict with "value" key: Value extracted and written using config
|
|
- Dict with "value" and "tags" keys: Value written with merged tags
|
|
:type data: dict
|
|
"""
|
|
from influxdb_client import InfluxDBClient, Point, WritePrecision
|
|
from influxdb_client.client.write_api import SYNCHRONOUS
|
|
|
|
try:
|
|
with InfluxDBClient(
|
|
url=self.url,
|
|
token=self.token,
|
|
org=self.org,
|
|
timeout=int(self.timeout * 1000),
|
|
) as client:
|
|
write_api = client.write_api(write_options=SYNCHRONOUS)
|
|
|
|
precision_map = {
|
|
"ns": WritePrecision.NS,
|
|
"us": WritePrecision.US,
|
|
"ms": WritePrecision.MS,
|
|
"s": WritePrecision.S,
|
|
}
|
|
precision = precision_map.get(self.write_precision, WritePrecision.MS)
|
|
|
|
for msg_name, msg_value in data.items():
|
|
# Get write configuration for this message
|
|
write_config = self.writes.get(msg_name, {})
|
|
|
|
# Get measurement, field, and base tags from config
|
|
measurement = write_config.get("measurement", msg_name)
|
|
field = write_config.get("field", "value")
|
|
base_tags = write_config.get("tags", {})
|
|
|
|
# Handle list of values (batch write)
|
|
values_to_write = (
|
|
msg_value if isinstance(msg_value, list) else [msg_value]
|
|
)
|
|
|
|
for item in values_to_write:
|
|
# Extract value and optional runtime tags
|
|
if isinstance(item, dict):
|
|
value = item.get("value", item)
|
|
runtime_tags = item.get("tags", {})
|
|
# If no "value" key, treat the whole dict as invalid
|
|
if "value" not in item and not isinstance(
|
|
value, (int, float, str, bool)
|
|
):
|
|
logger.warning(
|
|
"Skipping invalid item in node '%s': %s",
|
|
self.name,
|
|
item,
|
|
)
|
|
continue
|
|
else:
|
|
value = item
|
|
runtime_tags = {}
|
|
|
|
if value is None:
|
|
logger.info(
|
|
"Skipping None value for '%s' in node '%s'",
|
|
msg_name,
|
|
self.name,
|
|
)
|
|
continue
|
|
|
|
# Merge base tags with runtime tags (runtime takes precedence)
|
|
tags = {**base_tags, **runtime_tags}
|
|
|
|
# Build the point
|
|
point = Point(measurement)
|
|
|
|
for tag_key, tag_value in tags.items():
|
|
point = point.tag(tag_key, str(tag_value))
|
|
|
|
point = point.field(field, value)
|
|
|
|
# Write the point
|
|
write_api.write(
|
|
bucket=self.bucket,
|
|
org=self.org,
|
|
record=point,
|
|
write_precision=precision,
|
|
)
|
|
|
|
logger.info(
|
|
"Wrote to InfluxDB from node '%s': %s.%s=%s, tags=%s",
|
|
self.name,
|
|
measurement,
|
|
field,
|
|
value,
|
|
tags,
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error("InfluxDB write error in node '%s': %s", self.name, e)
|
|
raise
|
|
|
|
def _query_data(self) -> dict[str, Any]:
|
|
"""
|
|
Query data from InfluxDB based on provides configuration.
|
|
|
|
:returns: Dict of port name to queried value.
|
|
:rtype: dict
|
|
"""
|
|
from influxdb_client import InfluxDBClient
|
|
|
|
results = {}
|
|
|
|
try:
|
|
with InfluxDBClient(
|
|
url=self.url,
|
|
token=self.token,
|
|
org=self.org,
|
|
timeout=int(self.timeout * 1000),
|
|
) as client:
|
|
query_api = client.query_api()
|
|
|
|
for spec in self.output_ports:
|
|
msg_name = spec.port
|
|
query_config = self.queries.get(msg_name, {})
|
|
|
|
measurement = query_config.get("measurement", msg_name)
|
|
field = query_config.get("field", "value")
|
|
tags = query_config.get("tags", {})
|
|
time_range = query_config.get("range", self.query_range)
|
|
aggregation = query_config.get("aggregation", "last")
|
|
|
|
# Build Flux query
|
|
flux_query = self._build_flux_query(
|
|
measurement=measurement,
|
|
field=field,
|
|
tags=tags,
|
|
time_range=time_range,
|
|
aggregation=aggregation,
|
|
)
|
|
|
|
logger.info(
|
|
"Executing InfluxDB query for '%s' in node '%s': %s",
|
|
msg_name,
|
|
self.name,
|
|
flux_query,
|
|
)
|
|
|
|
# Execute query
|
|
tables = query_api.query(flux_query, org=self.org)
|
|
|
|
# Extract result
|
|
value = self._extract_query_result(tables, spec)
|
|
|
|
if value is not None:
|
|
results[msg_name] = value
|
|
logger.info(
|
|
"Query result for '%s' in node '%s': %s",
|
|
msg_name,
|
|
self.name,
|
|
value,
|
|
)
|
|
else:
|
|
logger.info(
|
|
"No data found for '%s' in node '%s'",
|
|
msg_name,
|
|
self.name,
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error("InfluxDB query error in node '%s': %s", self.name, e)
|
|
raise
|
|
|
|
return results
|
|
|
|
def _build_flux_query(
|
|
self,
|
|
measurement: str,
|
|
field: str,
|
|
tags: dict[str, Any],
|
|
time_range: str,
|
|
aggregation: str,
|
|
) -> str:
|
|
"""
|
|
Build a Flux query string.
|
|
|
|
:param measurement: Measurement name.
|
|
:type measurement: str
|
|
:param field: Field name.
|
|
:type field: str
|
|
:param tags: Tag filters.
|
|
:type tags: dict
|
|
:param time_range: Time range (e.g., "-1h").
|
|
:type time_range: str
|
|
:param aggregation: Aggregation function.
|
|
:type aggregation: str
|
|
:returns: Flux query string.
|
|
:rtype: str
|
|
"""
|
|
# Base query
|
|
query_parts = [
|
|
f'from(bucket: "{self.bucket}")',
|
|
f" |> range(start: {time_range})",
|
|
f' |> filter(fn: (r) => r["_measurement"] == "{measurement}")',
|
|
f' |> filter(fn: (r) => r["_field"] == "{field}")',
|
|
]
|
|
|
|
# Add tag filters
|
|
for tag_key, tag_value in tags.items():
|
|
query_parts.append(
|
|
f' |> filter(fn: (r) => r["{tag_key}"] == "{tag_value}")'
|
|
)
|
|
|
|
# Add aggregation
|
|
aggregation_map = {
|
|
"mean": "mean()",
|
|
"last": "last()",
|
|
"first": "first()",
|
|
"max": "max()",
|
|
"min": "min()",
|
|
"sum": "sum()",
|
|
"count": "count()",
|
|
}
|
|
|
|
if aggregation in aggregation_map:
|
|
query_parts.append(f" |> {aggregation_map[aggregation]}")
|
|
else:
|
|
# Default to last value
|
|
query_parts.append(" |> last()")
|
|
|
|
return "\n".join(query_parts)
|
|
|
|
def _extract_query_result(self, tables: Any, spec: MessageSpec) -> Any:
|
|
"""
|
|
Extract a single value from query result tables.
|
|
|
|
:param tables: InfluxDB query result tables.
|
|
:param spec: The port the value is destined for.
|
|
:type spec: MessageSpec
|
|
:returns: Extracted and typed value, or None if no data.
|
|
:rtype: Any
|
|
"""
|
|
for table in tables:
|
|
for record in table.records:
|
|
value = record.get_value()
|
|
if value is not None:
|
|
try:
|
|
return spec.coerce(value)
|
|
except (ValueError, TypeError):
|
|
return value
|
|
|
|
return None
|
|
|
|
def inject(
|
|
self, outputs: dict[str, Any] | None = None, durable: bool | None = None
|
|
) -> NodeResult:
|
|
"""
|
|
Inject queried data into the pipeline.
|
|
|
|
For InfluxDbNode, inject performs a query operation and injects
|
|
the results into the pipeline. This is useful for trigger-style
|
|
usage where you want to periodically query InfluxDB.
|
|
|
|
:param outputs: Optional pre-set outputs (usually None for queries).
|
|
:type outputs: dict | None
|
|
:returns: Query results injected into the pipeline.
|
|
:rtype: dict | None
|
|
"""
|
|
if self._pipeline is None:
|
|
raise RuntimeError("Node must be bound to a pipeline to inject")
|
|
|
|
# Without given values, injecting means running the configured queries.
|
|
if not outputs:
|
|
if not self.provides:
|
|
return None
|
|
outputs = self._query_data()
|
|
|
|
return self._pipeline.trigger(self, self._to_messages(outputs), durable=durable)
|