From 67a29431955014d6100c3e6f465c54a6c1590432 Mon Sep 17 00:00:00 2001 From: Aliaksei Kharlap Date: Mon, 3 Aug 2026 20:49:29 +0300 Subject: [PATCH 1/4] refactor valuecounter to use utils-package --- influxdata/library/plugin_library.json | 4 +- influxdata/valuecounter/README.md | 29 +- influxdata/valuecounter/manifest.toml | 4 +- influxdata/valuecounter/requirements-dev.txt | 1 + influxdata/valuecounter/requirements.txt | 1 + influxdata/valuecounter/test_valuecounter.py | 613 ++++++++++-------- influxdata/valuecounter/valuecounter.py | 328 ++++------ .../valuecounter/valuecounter_config.toml | 2 +- 8 files changed, 501 insertions(+), 481 deletions(-) diff --git a/influxdata/library/plugin_library.json b/influxdata/library/plugin_library.json index bb8986b..a076235 100644 --- a/influxdata/library/plugin_library.json +++ b/influxdata/library/plugin_library.json @@ -358,8 +358,8 @@ "author": "InfluxData", "docs_file_link": "https://github.com/influxdata/influxdb3_plugins/blob/main/influxdata/valuecounter/README.md", "required_plugins": [], - "required_libraries": [], - "last_update": "2026-06-10", + "required_libraries": ["influxdata-plugin-utils>=0.3.0"], + "last_update": "2026-08-03", "trigger_types_supported": ["scheduler", "data_writes"] }, { diff --git a/influxdata/valuecounter/README.md b/influxdata/valuecounter/README.md index bdfbbfc..ae36508 100644 --- a/influxdata/valuecounter/README.md +++ b/influxdata/valuecounter/README.md @@ -25,11 +25,12 @@ This plugin includes a JSON metadata schema in its docstring that defines suppor ### Data write trigger parameters -| Parameter | Type | Default | Description | -|------------------|---------|------------------|----------------------------------------------------------------------------------------------------------| -| `period_seconds` | integer | `60` | Emission period in seconds. Cache TTL is set to `2 * period_seconds`. Accepts `period=60s` form via TOML | -| `output_suffix` | string | `_valuecounts` | Suffix appended to the source measurement name for rollup output. Must be non-empty | -| `dest_database` | string | trigger's own DB | Optional database to write rollups to via `write_sync_to_db` | +| Parameter | Type | Default | Description | +|------------------|---------|------------------|----------------------------------------------------------------------------------------------------------------------------------| +| `period_seconds` | integer | `60` | Emission period in seconds, at least `1`. Cache TTL is set to `2 * period_seconds` | +| `period` | string | `60s` | Emission period as a duration, at least `1s`. Units: `s`, `min`, `h`, `d`, `w`. Overridden by `period_seconds` when both are set | +| `output_suffix` | string | `_valuecounts` | Suffix appended to the source measurement name for rollup output. Must be non-empty | +| `dest_database` | string | trigger's own DB | Optional database to write rollups to via `write_sync_to_db` | ### Scheduled trigger parameters @@ -46,9 +47,9 @@ Mode B is drift-based: the trigger's `every:` spec is the only cadence |--------------------|--------|---------|----------------------------------------------------------------------------------| | `config_file_path` | string | none | TOML config file path relative to `PLUGIN_DIR` (required for TOML configuration) | -*To use a TOML configuration file, set the `PLUGIN_DIR` environment variable and specify the `config_file_path` in the trigger arguments.* This is in addition to the `--plugin-dir` flag when starting InfluxDB 3. +*To use a TOML configuration file, set the `PLUGIN_DIR` environment variable and specify the `config_file_path` in the trigger arguments.* This is in addition to the `--plugin-dir` flag when starting InfluxDB 3. Relative paths are resolved against `PLUGIN_DIR`, then `INFLUXDB3_PLUGIN_DIR`, then the parent of `VIRTUAL_ENV`. -If `config_file_path` is set, no other inline arguments may be set on the trigger. The TOML accepts the same keys as inline arguments; Mode A may use `period = "60s"` in place of `period_seconds = 60`. +If `config_file_path` is set, no other inline arguments may be set on the trigger. The TOML accepts the same keys as inline arguments. #### Example TOML configuration @@ -65,7 +66,7 @@ The rollup measurement name is always ``. The defau ## Software Requirements - **InfluxDB 3 Core/Enterprise**: with the Processing Engine enabled -- **Python packages**: none (standard library only) +- **Python packages**: `influxdata-plugin-utils>=0.3.0` ## Installation steps @@ -79,7 +80,11 @@ The rollup measurement name is always ``. The defau --plugin-dir ~/.plugins ``` -2. No additional Python packages required for this plugin. +2. Install required Python packages: + + ```bash + influxdb3 install package "influxdata-plugin-utils>=0.3.0" + ``` ## Trigger setup @@ -207,8 +212,8 @@ The rollup measurement `payment_attempts_valuecounts` is written to the `analyti - `valuecounter.py`: The main plugin code containing `process_writes` (Mode A) and `process_scheduled_call` (Mode B) - `valuecounter_config.toml`: Example TOML configuration with both Mode A and Mode B shapes -- `test_valuecounter.py`: Pytest suite (82 tests, runs without a live InfluxDB 3 server) -- `requirements.txt`: Runtime dependencies (empty for this plugin) +- `test_valuecounter.py`: Pytest suite (79 tests, runs without a live InfluxDB 3 server) +- `requirements.txt`: Runtime dependencies (`influxdata-plugin-utils>=0.3.0`) - `requirements-dev.txt`: Development dependencies (`pytest`) ### Logging @@ -251,7 +256,7 @@ Handles Mode B scheduled invocations. On the first fire, only the cadence anchor #### Issue: Source table tag list changed and rollups are missing tags -**Solution**: The plugin caches tag-column names from `information_schema.columns` for one hour per source table. Schema changes are picked up within an hour. To force a refresh, delete the cache key `vc:tags:`. +**Solution**: The plugin caches tag-column names from `information_schema.columns` for one hour per source table. Schema changes are picked up within an hour. To force a refresh, delete the cache key `shared:tags:
`. #### Issue: Both modes installed against the same source table diff --git a/influxdata/valuecounter/manifest.toml b/influxdata/valuecounter/manifest.toml index b2b5e25..d4ee07f 100644 --- a/influxdata/valuecounter/manifest.toml +++ b/influxdata/valuecounter/manifest.toml @@ -2,7 +2,7 @@ manifest_schema_version = "1.2" [plugin] name = "valuecounter" -version = "0.2.0" +version = "0.3.0" description = "Counts unique field values from data-write or scheduled inputs and writes rollup measurements with per-value counts." triggers = ["process_writes", "process_scheduled_call"] homepage = "https://www.influxdata.com/" @@ -16,4 +16,4 @@ exclude = [ [dependencies] database_version = ">=3.8.2" -python = [] +python = ["influxdata-plugin-utils>=0.3.0"] diff --git a/influxdata/valuecounter/requirements-dev.txt b/influxdata/valuecounter/requirements-dev.txt index e079f8a..2a50709 100644 --- a/influxdata/valuecounter/requirements-dev.txt +++ b/influxdata/valuecounter/requirements-dev.txt @@ -1 +1,2 @@ pytest +influxdata-plugin-utils>=0.3.0 diff --git a/influxdata/valuecounter/requirements.txt b/influxdata/valuecounter/requirements.txt index e69de29..46a2877 100644 --- a/influxdata/valuecounter/requirements.txt +++ b/influxdata/valuecounter/requirements.txt @@ -0,0 +1 @@ +influxdata-plugin-utils>=0.3.0 diff --git a/influxdata/valuecounter/test_valuecounter.py b/influxdata/valuecounter/test_valuecounter.py index 7354d4d..5b50851 100644 --- a/influxdata/valuecounter/test_valuecounter.py +++ b/influxdata/valuecounter/test_valuecounter.py @@ -6,21 +6,26 @@ from collections import OrderedDict from typing import Optional + class InfluxDBError(Exception): pass + class InvalidMeasurementError(InfluxDBError): pass + class InvalidKeyError(InfluxDBError): pass + class InvalidLineError(InfluxDBError): pass + class LineBuilder: def __init__(self, measurement: str): - if ' ' in measurement: + if " " in measurement: raise InvalidMeasurementError("Measurement name cannot contain spaces") self.measurement = measurement self.tags: OrderedDict[str, str] = OrderedDict() @@ -31,32 +36,42 @@ def _validate_key(self, key: str, key_type: str) -> None: """Validate that a key does not contain spaces, commas, or equals signs.""" if not key: raise InvalidKeyError(f"{key_type} key cannot be empty") - if ' ' in key: + if " " in key: raise InvalidKeyError(f"{key_type} key '{key}' cannot contain spaces") - if ',' in key: + if "," in key: raise InvalidKeyError(f"{key_type} key '{key}' cannot contain commas") - if '=' in key: + if "=" in key: raise InvalidKeyError(f"{key_type} key '{key}' cannot contain equals signs") def _escape_measurement(self, value: str) -> str: """Escape characters in measurement names according to line protocol.""" - return value.replace(',', '\\,').replace(' ', '\\ ') + return value.replace(",", "\\,").replace(" ", "\\ ") def _escape_tag_value(self, value: str) -> str: """Escape characters in tag values according to line protocol.""" - return value.replace('\\', '\\\\').replace(',', '\\,').replace('=', '\\=').replace(' ', '\\ ') + return ( + value.replace("\\", "\\\\") + .replace(",", "\\,") + .replace("=", "\\=") + .replace(" ", "\\ ") + ) def _escape_field_key(self, value: str) -> str: """Escape characters in field keys according to line protocol.""" - return value.replace('\\', '\\\\').replace(',', '\\,').replace('=', '\\=').replace(' ', '\\ ') + return ( + value.replace("\\", "\\\\") + .replace(",", "\\,") + .replace("=", "\\=") + .replace(" ", "\\ ") + ) - def tag(self, key: str, value: str) -> 'LineBuilder': + def tag(self, key: str, value: str) -> "LineBuilder": """Add a tag to the line protocol.""" self._validate_key(key, "tag") self.tags[key] = str(value) return self - def uint64_field(self, key: str, value: int) -> 'LineBuilder': + def uint64_field(self, key: str, value: int) -> "LineBuilder": """Add an unsigned integer field to the line protocol.""" self._validate_key(key, "field") if value < 0: @@ -64,34 +79,34 @@ def uint64_field(self, key: str, value: int) -> 'LineBuilder': self.fields[key] = f"{value}u" return self - def int64_field(self, key: str, value: int) -> 'LineBuilder': + def int64_field(self, key: str, value: int) -> "LineBuilder": """Add an integer field to the line protocol.""" self._validate_key(key, "field") self.fields[key] = f"{value}i" return self - def float64_field(self, key: str, value: float) -> 'LineBuilder': + def float64_field(self, key: str, value: float) -> "LineBuilder": """Add a float field to the line protocol.""" self._validate_key(key, "field") # Check if value has no decimal component self.fields[key] = f"{int(value)}.0" if value % 1 == 0 else str(value) return self - def string_field(self, key: str, value: str) -> 'LineBuilder': + def string_field(self, key: str, value: str) -> "LineBuilder": """Add a string field to the line protocol.""" self._validate_key(key, "field") # Escape quotes and backslashes in string values - escaped_value = value.replace('\\', '\\\\').replace('"', '\\"') + escaped_value = value.replace("\\", "\\\\").replace('"', '\\"') self.fields[key] = f'"{escaped_value}"' return self - def bool_field(self, key: str, value: bool) -> 'LineBuilder': + def bool_field(self, key: str, value: bool) -> "LineBuilder": """Add a boolean field to the line protocol.""" self._validate_key(key, "field") - self.fields[key] = 't' if value else 'f' + self.fields[key] = "t" if value else "f" return self - def time_ns(self, timestamp_ns: int) -> 'LineBuilder': + def time_ns(self, timestamp_ns: int) -> "LineBuilder": """Set the timestamp in nanoseconds.""" self._timestamp_ns = timestamp_ns return self @@ -103,7 +118,7 @@ def build(self) -> str: # Add tags if present if self.tags: - tags_str = ','.join( + tags_str = ",".join( f"{key}={self._escape_tag_value(value)}" for key, value in self.tags.items() ) @@ -113,7 +128,7 @@ def build(self) -> str: if not self.fields: raise InvalidLineError(f"At least one field is required: {line}") - fields_str = ','.join( + fields_str = ",".join( f"{self._escape_field_key(key)}={value}" for key, value in self.fields.items() ) @@ -130,11 +145,14 @@ class FakeCache: def __init__(self): self._d = {} self._ttls = {} + def get(self, key, default=None, use_global=None): return self._d.get(key, default) + def put(self, key, value, ttl=None, use_global=None): self._d[key] = value self._ttls[key] = ttl + def delete(self, key, use_global=None): return self._d.pop(key, None) is not None @@ -148,17 +166,26 @@ def __init__(self): self.fail_write = False self.query_responses = {} self.queries_run = [] - def info(self, m): self.logs.append(("info", m)) - def warn(self, m): self.logs.append(("warn", m)) - def error(self, m): self.logs.append(("error", m)) + + def info(self, m): + self.logs.append(("info", m)) + + def warn(self, m): + self.logs.append(("warn", m)) + + def error(self, m): + self.logs.append(("error", m)) + def write_sync(self, line, no_sync=False): if self.fail_write: raise RuntimeError("simulated write failure") self.writes.append(line.build()) + def write_sync_to_db(self, db_name, line, no_sync=False): if self.fail_write: raise RuntimeError("simulated write failure") self.cross_db_writes.append((db_name, line.build())) + def query(self, sql, params=None): self.queries_run.append((sql, params)) for substr, rows in self.query_responses.items(): @@ -179,23 +206,30 @@ def test_vendored_line_builder_basic(): def test_stringify_int(): assert _stringify_value(200) == "200" + def test_stringify_float(): assert _stringify_value(1.5) == "1.5" + def test_stringify_str(): assert _stringify_value("OK") == "OK" + def test_stringify_bool_true(): - assert _stringify_value(True) == "true" # telegraf-style lowercase + assert _stringify_value(True) == "true" # telegraf-style lowercase + def test_stringify_bool_false(): assert _stringify_value(False) == "false" + def test_stringify_none(): - assert _stringify_value(None) is None # signals "skip" + assert _stringify_value(None) is None # signals "skip" + def test_stringify_unknown_falls_back_to_repr(): class Weird: ... + out = _stringify_value(Weird()) assert isinstance(out, str) assert "Weird" in out @@ -207,14 +241,20 @@ class Weird: ... def test_sanitize_no_change_for_plain(): assert _sanitize_field_name("status_200") == "status_200" + def test_sanitize_spaces(): assert _sanitize_field_name("not found") == "not_found" + def test_sanitize_hyphen(): assert _sanitize_field_name("User-Agent") == "User_Agent" + def test_sanitize_punctuation(): - assert _sanitize_field_name("500 Internal Server Error") == "500_Internal_Server_Error" + assert ( + _sanitize_field_name("500 Internal Server Error") == "500_Internal_Server_Error" + ) + def test_sanitize_unicode_replaced(): assert _sanitize_field_name("café") == "caf_" @@ -227,211 +267,245 @@ def test_validate_accepts_plain(): _validate_identifier("http_requests", "table") _validate_identifier("status", "field") + def test_validate_accepts_leading_underscore(): _validate_identifier("_internal", "table") + def test_validate_rejects_leading_digit(): with pytest.raises(ValueError): _validate_identifier("1status", "field") + def test_validate_rejects_space(): with pytest.raises(ValueError): _validate_identifier("status code", "field") + def test_validate_rejects_dash(): with pytest.raises(ValueError): _validate_identifier("user-agent", "field") + def test_validate_rejects_semicolon(): with pytest.raises(ValueError): _validate_identifier("status;DROP TABLE users", "table") + def test_validate_rejects_quote(): with pytest.raises(ValueError): _validate_identifier('status"', "field") + def test_validate_rejects_too_long(): with pytest.raises(ValueError): _validate_identifier("a" * 129, "table") + def test_validate_message_includes_what_and_value(): with pytest.raises(ValueError, match="invalid table:.*'bad name'"): _validate_identifier("bad name", "table") -from valuecounter import _parse_duration +from valuecounter import _parse_period_seconds def test_parse_seconds(): - assert _parse_duration("60s") == 60 + assert _parse_period_seconds("60s") == 60 + + +def test_parse_legacy_bare_minutes(): + assert _parse_period_seconds("5m") == 300 + def test_parse_minutes(): - assert _parse_duration("5m") == 300 + assert _parse_period_seconds("5min") == 300 + def test_parse_hours(): - assert _parse_duration("1h") == 3600 + assert _parse_period_seconds("1h") == 3600 + def test_parse_days(): - assert _parse_duration("1d") == 86400 + assert _parse_period_seconds("1d") == 86400 + def test_parse_with_whitespace(): - assert _parse_duration(" 30 s ") == 30 + assert _parse_period_seconds(" 30 s ") == 30 + def test_parse_rejects_missing_unit(): - with pytest.raises(ValueError, match="invalid duration"): - _parse_duration("60") + with pytest.raises(ValueError, match="Invalid duration"): + _parse_period_seconds("60") + + +def test_parse_rejects_sub_second(): + with pytest.raises(ValueError, match="at least 1 second"): + _parse_period_seconds("60ms") -def test_parse_rejects_unknown_unit(): - with pytest.raises(ValueError, match="invalid duration"): - _parse_duration("60ms") def test_parse_rejects_empty(): with pytest.raises(ValueError): - _parse_duration("") + _parse_period_seconds("") + + +from valuecounter import _resolve_config -from valuecounter import Config, _parse_inline_args +def _write_tmp_toml(tmp_path, content): + path = tmp_path / "vc.toml" + path.write_text(content) + return path -def test_inline_args_mode_a_minimal(): - cfg = _parse_inline_args({"fields": "status method"}, mode="wal") +def test_resolve_inline_mode_a_minimal(): + cfg = _resolve_config({"fields": "status method"}, mode="wal") assert cfg.fields == ["status", "method"] assert cfg.output_suffix == "_valuecounts" assert cfg.dest_database == "" assert cfg.period_seconds == 60 assert cfg.table == "" -def test_inline_args_mode_b_minimal(): - cfg = _parse_inline_args({"table": "http_requests", "fields": "status"}, mode="scheduled") + +def test_resolve_inline_mode_b_minimal(): + cfg = _resolve_config( + {"table": "http_requests", "fields": "status"}, mode="scheduled" + ) assert cfg.table == "http_requests" assert cfg.fields == ["status"] -def test_inline_args_mode_a_full(): - cfg = _parse_inline_args( - {"fields": "status method", "output_suffix": "_vc", - "period_seconds": "30", "dest_database": "rollups"}, + +def test_resolve_inline_mode_a_full(): + cfg = _resolve_config( + { + "fields": "status method", + "output_suffix": "_vc", + "period_seconds": "30", + "dest_database": "rollups", + }, mode="wal", ) assert cfg.period_seconds == 30 assert cfg.output_suffix == "_vc" assert cfg.dest_database == "rollups" -def test_inline_args_rejects_unknown_key(): + +def test_resolve_period_seconds_overrides_period(): + cfg = _resolve_config( + {"fields": "a", "period": "5m", "period_seconds": "30"}, mode="wal" + ) + assert cfg.period_seconds == 30 + + +def test_resolve_rejects_unknown_arg(): with pytest.raises(ValueError, match="unknown arg"): - _parse_inline_args({"fields": "a", "bogus": "x"}, mode="wal") + _resolve_config({"fields": "a", "bogus": "x"}, mode="wal") -def test_inline_args_mode_a_rejects_table(): - with pytest.raises(ValueError, match="table"): - _parse_inline_args({"fields": "a", "table": "http_requests"}, mode="wal") -def test_inline_args_mode_b_rejects_period_seconds(): - with pytest.raises(ValueError, match="period"): - _parse_inline_args( +def test_resolve_mode_a_rejects_table(): + with pytest.raises(ValueError, match="trigger-spec.*inline args"): + _resolve_config({"fields": "a", "table": "http_requests"}, mode="wal") + + +def test_resolve_mode_b_rejects_period_seconds(): + with pytest.raises(ValueError, match="drift-based"): + _resolve_config( {"table": "t", "fields": "a", "period_seconds": "60"}, mode="scheduled" ) -def test_inline_args_mode_b_rejects_period(): - with pytest.raises(ValueError, match="period"): - _parse_inline_args( + +def test_resolve_mode_b_rejects_period(): + with pytest.raises(ValueError, match="drift-based"): + _resolve_config( {"table": "t", "fields": "a", "period": "60s"}, mode="scheduled" ) -import os -import tempfile -from valuecounter import _load_toml_config - - -def _write_tmp_toml(content): - f = tempfile.NamedTemporaryFile(suffix=".toml", delete=False, mode="w") - f.write(content) - f.close() - return f.name +def test_resolve_rejects_unknown_mode(): + with pytest.raises(ValueError, match="unknown mode"): + _resolve_config({"fields": "a"}, mode="bogus") -def test_toml_mode_a_minimal(): - p = _write_tmp_toml('fields = ["status", "method"]\nperiod = "30s"\n') - cfg = _load_toml_config(p, mode="wal") +def test_resolve_toml_mode_a(monkeypatch, tmp_path): + monkeypatch.setenv("PLUGIN_DIR", str(tmp_path)) + _write_tmp_toml(tmp_path, 'fields = ["status", "method"]\nperiod = "30s"\n') + cfg = _resolve_config({"config_file_path": "vc.toml"}, mode="wal") assert cfg.fields == ["status", "method"] assert cfg.period_seconds == 30 - os.unlink(p) -def test_toml_mode_b_minimal(): - p = _write_tmp_toml('table = "http_requests"\nfields = ["status"]\n') - cfg = _load_toml_config(p, mode="scheduled") +def test_resolve_toml_mode_b(monkeypatch, tmp_path): + monkeypatch.setenv("PLUGIN_DIR", str(tmp_path)) + _write_tmp_toml(tmp_path, 'table = "http_requests"\nfields = ["status"]\n') + cfg = _resolve_config({"config_file_path": "vc.toml"}, mode="scheduled") assert cfg.table == "http_requests" assert cfg.fields == ["status"] - os.unlink(p) -def test_toml_rejects_unknown_key(): - p = _write_tmp_toml('fields = ["a"]\nbogus = 1\n') - with pytest.raises(ValueError, match="unknown"): - _load_toml_config(p, mode="wal") - os.unlink(p) +def test_resolve_toml_rejects_unknown_key(monkeypatch, tmp_path): + monkeypatch.setenv("PLUGIN_DIR", str(tmp_path)) + _write_tmp_toml(tmp_path, 'fields = ["a"]\nbogus = 1\n') + with pytest.raises(ValueError, match="unknown TOML key"): + _resolve_config({"config_file_path": "vc.toml"}, mode="wal") -def test_toml_parse_error(): - p = _write_tmp_toml('this is not valid toml = =') +def test_resolve_toml_parse_error(monkeypatch, tmp_path): + monkeypatch.setenv("PLUGIN_DIR", str(tmp_path)) + _write_tmp_toml(tmp_path, "this is not valid toml = =") with pytest.raises(ValueError): - _load_toml_config(p, mode="wal") - os.unlink(p) + _resolve_config({"config_file_path": "vc.toml"}, mode="wal") -def test_toml_missing_file(): +def test_resolve_toml_missing_file(monkeypatch, tmp_path): + monkeypatch.setenv("PLUGIN_DIR", str(tmp_path)) with pytest.raises(FileNotFoundError): - _load_toml_config("/tmp/definitely_not_here.toml", mode="wal") + _resolve_config({"config_file_path": "definitely_not_here.toml"}, mode="wal") -from valuecounter import _resolve_config +def test_resolve_toml_path_needs_plugin_dir(monkeypatch): + for name in ("PLUGIN_DIR", "INFLUXDB3_PLUGIN_DIR", "VIRTUAL_ENV"): + monkeypatch.delenv(name, raising=False) + with pytest.raises(ValueError, match="Cannot resolve plugin directory"): + _resolve_config({"config_file_path": "vc.toml"}, mode="wal") -def test_resolve_inline_only(): - cfg = _resolve_config({"fields": "status"}, plugin_dir="/tmp", mode="wal") - assert cfg.fields == ["status"] - def test_resolve_neither_fields_nor_toml_raises(): with pytest.raises(ValueError, match="fields"): - _resolve_config({}, plugin_dir="/tmp", mode="wal") + _resolve_config({}, mode="wal") + def test_resolve_empty_fields_raises(): with pytest.raises(ValueError, match="empty"): - _resolve_config({"fields": ""}, plugin_dir="/tmp", mode="wal") + _resolve_config({"fields": ""}, mode="wal") + def test_resolve_mode_b_without_table_raises(): with pytest.raises(ValueError, match="table"): - _resolve_config({"fields": "status"}, plugin_dir="/tmp", mode="scheduled") + _resolve_config({"fields": "status"}, mode="scheduled") + -def test_resolve_both_inline_and_toml_raises(): - p = _write_tmp_toml('fields = ["status"]\n') +def test_resolve_both_inline_and_toml_raises(monkeypatch, tmp_path): + monkeypatch.setenv("PLUGIN_DIR", str(tmp_path)) + _write_tmp_toml(tmp_path, 'fields = ["status"]\n') with pytest.raises(ValueError, match="either.*both"): - _resolve_config( - {"config_file_path": os.path.basename(p), "fields": "status"}, - plugin_dir=os.path.dirname(p), - mode="wal", - ) - os.unlink(p) + _resolve_config({"config_file_path": "vc.toml", "fields": "status"}, mode="wal") + def test_resolve_validates_identifier_table(): with pytest.raises(ValueError, match="invalid table"): _resolve_config( - {"table": "http requests", "fields": "status"}, - plugin_dir="/tmp", mode="scheduled", + {"table": "http requests", "fields": "status"}, mode="scheduled" ) + def test_resolve_validates_identifier_field(): with pytest.raises(ValueError, match="invalid field"): - _resolve_config( - {"fields": "status method-name"}, plugin_dir="/tmp", mode="wal", - ) + _resolve_config({"fields": "status method-name"}, mode="wal") def test_resolve_rejects_empty_output_suffix(): with pytest.raises(ValueError, match="output_suffix"): - _resolve_config( - {"fields": "status", "output_suffix": ""}, plugin_dir="/tmp", mode="wal", - ) + _resolve_config({"fields": "status", "output_suffix": ""}, mode="wal") from valuecounter import _series_key @@ -442,11 +516,13 @@ def test_series_key_stable_across_tag_order(): k2 = _series_key("http_requests", {"endpoint": "/api", "host": "web-1"}) assert k1 == k2 + def test_series_key_distinct_for_distinct_tags(): k1 = _series_key("http_requests", {"host": "web-1"}) k2 = _series_key("http_requests", {"host": "web-2"}) assert k1 != k2 + def test_series_key_includes_table(): k1 = _series_key("table_a", {"host": "x"}) k2 = _series_key("table_b", {"host": "x"}) @@ -454,6 +530,7 @@ def test_series_key_includes_table(): assert "table_a" in k1 assert "table_b" in k2 + def test_series_key_empty_tags(): k = _series_key("http_requests", {}) assert k.startswith("http_requests:") @@ -467,31 +544,40 @@ def test_extract_tags_subset(): out = _extract_tags(row, ["host", "endpoint"]) assert out == {"host": "web-1", "endpoint": "/api"} + def test_extract_tags_missing_tag_returns_partial(): row = {"host": "web-1", "status": 200} out = _extract_tags(row, ["host", "endpoint"]) assert out == {"host": "web-1"} + def test_extract_tags_empty_list(): row = {"status": 200} assert _extract_tags(row, []) == {} + def test_extract_tags_null_tag_returns_none(): row = {"host": "web-1", "endpoint": None} assert _extract_tags(row, ["host", "endpoint"]) is None -from valuecounter import _build_line +import valuecounter +from valuecounter import _build_rollup_line + + +@pytest.fixture(autouse=True) +def _inject_line_builder(monkeypatch): + """The engine injects LineBuilder as a global; tests use the vendored copy.""" + monkeypatch.setattr(valuecounter, "LineBuilder", LineBuilder, raising=False) def test_build_line_basic(): - lb = _build_line( + lb = _build_rollup_line( "http_requests", {"host": "web-1"}, {"status_200": 3, "status_500": 1}, "_valuecounts", 1731000060000000000, - LineBuilder, ) out = lb.build() assert out.startswith("http_requests_valuecounts,host=web-1 ") @@ -499,34 +585,15 @@ def test_build_line_basic(): assert "status_500=1i" in out assert out.endswith(" 1731000060000000000") + def test_build_line_tagless(): - lb = _build_line("t", {}, {"a_1": 2}, "_vc", 1, LineBuilder) + lb = _build_rollup_line("t", {}, {"a_1": 2}, "_vc", 1) assert lb.build() == "t_vc a_1=2i 1" -def test_build_line_empty_counts_returns_none(): - assert _build_line("t", {}, {}, "_vc", 1, LineBuilder) is None - -from valuecounter import _BatchLines - - -def test_batchlines_joins(): - lb1 = LineBuilder("m1"); lb1.int64_field("x", 1).time_ns(1) - lb2 = LineBuilder("m2"); lb2.int64_field("y", 2).time_ns(2) - b = _BatchLines([lb1, lb2]) - assert b.build() == "m1 x=1i 1\nm2 y=2i 2" - -def test_batchlines_empty_raises(): - b = _BatchLines([]) - with pytest.raises(ValueError, match="no lines"): - b.build() - -def test_batchlines_caches_build(): - lb = LineBuilder("m"); lb.int64_field("x", 1).time_ns(1) - b = _BatchLines([lb]) - out1 = b.build() - out2 = b.build() - assert out1 == out2 +def test_build_line_empty_counts_raises(): + with pytest.raises(ValueError, match="no fields"): + _build_rollup_line("t", {}, {}, "_vc", 1) from valuecounter import _build_scheduled_query @@ -542,6 +609,7 @@ def test_build_query_with_tags(): assert "GROUP BY" in sql assert "COUNT(*)" in sql.upper() or "count(*)" in sql + def test_build_query_tagless(): sql = _build_scheduled_query("t", [], "field_a") # No tag projection prefix or GROUP BY tag prefix; only the watched field @@ -553,37 +621,6 @@ def test_build_query_tagless(): assert "SELECT ," not in sql.replace(" ", "")[:8] -from valuecounter import _get_tag_names - - -def test_get_tag_names_cache_miss_runs_query_and_caches(): - fake = FakeInfluxdb3Local() - fake.query_responses["information_schema.columns"] = [ - {"column_name": "host"}, - {"column_name": "endpoint"}, - ] - out = _get_tag_names(fake, "http_requests") - assert out == ["host", "endpoint"] - assert len(fake.queries_run) == 1 - # Cached for 1h - assert fake.cache._d.get("vc:tags:http_requests") == ["host", "endpoint"] - assert fake.cache._ttls.get("vc:tags:http_requests") == 3600 - -def test_get_tag_names_cache_hit_skips_query(): - fake = FakeInfluxdb3Local() - fake.cache.put("vc:tags:http_requests", ["host"], ttl=3600) - out = _get_tag_names(fake, "http_requests") - assert out == ["host"] - assert fake.queries_run == [] - -def test_get_tag_names_empty_result_caches_empty_list(): - fake = FakeInfluxdb3Local() - fake.query_responses["information_schema.columns"] = [] - out = _get_tag_names(fake, "tagless") - assert out == [] - assert fake.cache._d.get("vc:tags:tagless") == [] - - from valuecounter import _query_field_distribution @@ -593,8 +630,12 @@ def test_query_field_distribution_calls_query_with_string_params(): {"host": "a", "status": "200", "cnt": 3}, ] rows = _query_field_distribution( - fake, "http_requests", ["host"], "status", - start_ns=1_000_000_000, end_ns=2_000_000_000, + fake, + "http_requests", + ["host"], + "status", + start_ns=1_000_000_000, + end_ns=2_000_000_000, ) assert rows == [{"host": "a", "status": "200", "cnt": 3}] # Verify params dict was decimal strings @@ -613,32 +654,34 @@ def _dt_ns(ns): return datetime.fromtimestamp(ns / 1_000_000_000, tz=timezone.utc) -def test_scheduled_first_fire_sets_anchor_no_emit(monkeypatch, tmp_path): +def test_scheduled_first_fire_sets_anchor_no_emit(): fake = FakeInfluxdb3Local() - # Make process_scheduled_call see LineBuilder as a module-level injected global - monkeypatch.setattr("valuecounter.LineBuilder", LineBuilder, raising=False) - monkeypatch.setenv("PLUGIN_DIR", str(tmp_path)) call_time = _dt_ns(1_000_000_000_000_000_000) process_scheduled_call( - fake, call_time, args={"table": "http_requests", "fields": "status"}, + fake, + call_time, + args={"table": "http_requests", "fields": "status"}, ) # Anchor written, no rollup emitted - assert fake.cache._d.get("vc:scheduled:last_call_ns:http_requests") == 1_000_000_000_000_000_000 + assert ( + fake.cache._d.get("vc:scheduled:last_call_ns:http_requests") + == 1_000_000_000_000_000_000 + ) assert fake.writes == [] # Info log emitted assert any("first fire" in m for _, m in fake.logs) -def test_scheduled_second_fire_emits_and_advances_anchor(monkeypatch, tmp_path): +def test_scheduled_second_fire_emits_and_advances_anchor(): fake = FakeInfluxdb3Local() - monkeypatch.setattr("valuecounter.LineBuilder", LineBuilder, raising=False) - monkeypatch.setenv("PLUGIN_DIR", str(tmp_path)) # Pre-seed anchor and tag-name cache - fake.cache.put("vc:scheduled:last_call_ns:http_requests", 1_000_000_000_000_000_000, ttl=None) - fake.cache.put("vc:tags:http_requests", ["host"], ttl=3600) + fake.cache.put( + "vc:scheduled:last_call_ns:http_requests", 1_000_000_000_000_000_000, ttl=None + ) + fake.cache.put("shared:tags:http_requests", ["host"], ttl=3600) fake.query_responses['"status"'] = [ {"host": "a", "status": "200", "cnt": 3}, {"host": "a", "status": "500", "cnt": 1}, @@ -646,7 +689,9 @@ def test_scheduled_second_fire_emits_and_advances_anchor(monkeypatch, tmp_path): call_time = _dt_ns(2_000_000_000_000_000_000) process_scheduled_call( - fake, call_time, args={"table": "http_requests", "fields": "status"}, + fake, + call_time, + args={"table": "http_requests", "fields": "status"}, ) assert len(fake.writes) == 1 @@ -654,21 +699,23 @@ def test_scheduled_second_fire_emits_and_advances_anchor(monkeypatch, tmp_path): assert "http_requests_valuecounts,host=a status_200=3i,status_500=1i" in out assert out.endswith(" 2000000000000000000") # Anchor advanced - assert fake.cache._d["vc:scheduled:last_call_ns:http_requests"] == 2_000_000_000_000_000_000 + assert ( + fake.cache._d["vc:scheduled:last_call_ns:http_requests"] + == 2_000_000_000_000_000_000 + ) -def test_scheduled_write_failure_does_not_advance_anchor(monkeypatch, tmp_path): +def test_scheduled_write_failure_does_not_advance_anchor(): fake = FakeInfluxdb3Local() - monkeypatch.setattr("valuecounter.LineBuilder", LineBuilder, raising=False) - monkeypatch.setenv("PLUGIN_DIR", str(tmp_path)) fake.cache.put("vc:scheduled:last_call_ns:t", 1_000_000_000_000_000_000, ttl=None) - fake.cache.put("vc:tags:t", ["host"], ttl=3600) + fake.cache.put("shared:tags:t", ["host"], ttl=3600) fake.query_responses['"f"'] = [{"host": "a", "f": "x", "cnt": 1}] fake.fail_write = True process_scheduled_call( - fake, _dt_ns(2_000_000_000_000_000_000), + fake, + _dt_ns(2_000_000_000_000_000_000), args={"table": "t", "fields": "f"}, ) @@ -676,17 +723,16 @@ def test_scheduled_write_failure_does_not_advance_anchor(monkeypatch, tmp_path): assert any("write failed" in m for _, m in fake.logs) -def test_scheduled_empty_window_advances_anchor(monkeypatch, tmp_path): +def test_scheduled_empty_window_advances_anchor(): fake = FakeInfluxdb3Local() - monkeypatch.setattr("valuecounter.LineBuilder", LineBuilder, raising=False) - monkeypatch.setenv("PLUGIN_DIR", str(tmp_path)) fake.cache.put("vc:scheduled:last_call_ns:t", 1_000_000_000_000_000_000, ttl=None) - fake.cache.put("vc:tags:t", ["host"], ttl=3600) + fake.cache.put("shared:tags:t", ["host"], ttl=3600) # query_responses empty → no rows in window process_scheduled_call( - fake, _dt_ns(2_000_000_000_000_000_000), + fake, + _dt_ns(2_000_000_000_000_000_000), args={"table": "t", "fields": "f"}, ) @@ -695,17 +741,16 @@ def test_scheduled_empty_window_advances_anchor(monkeypatch, tmp_path): assert any("no rows" in m for _, m in fake.logs) -def test_scheduled_cross_db_write(monkeypatch, tmp_path): +def test_scheduled_cross_db_write(): fake = FakeInfluxdb3Local() - monkeypatch.setattr("valuecounter.LineBuilder", LineBuilder, raising=False) - monkeypatch.setenv("PLUGIN_DIR", str(tmp_path)) fake.cache.put("vc:scheduled:last_call_ns:t", 1_000_000_000_000_000_000, ttl=None) - fake.cache.put("vc:tags:t", [], ttl=3600) # tagless + fake.cache.put("shared:tags:t", [], ttl=3600) # tagless fake.query_responses['"f"'] = [{"f": "x", "cnt": 5}] process_scheduled_call( - fake, _dt_ns(2_000_000_000_000_000_000), + fake, + _dt_ns(2_000_000_000_000_000_000), args={"table": "t", "fields": "f", "dest_database": "rollups"}, ) @@ -716,20 +761,19 @@ def test_scheduled_cross_db_write(monkeypatch, tmp_path): assert lp.startswith("t_valuecounts f_x=5i") -def test_scheduled_sanitization_collision_sums_and_warns(monkeypatch, tmp_path): +def test_scheduled_sanitization_collision_sums_and_warns(): fake = FakeInfluxdb3Local() - monkeypatch.setattr("valuecounter.LineBuilder", LineBuilder, raising=False) - monkeypatch.setenv("PLUGIN_DIR", str(tmp_path)) fake.cache.put("vc:scheduled:last_call_ns:t", 1_000_000_000_000_000_000, ttl=None) - fake.cache.put("vc:tags:t", ["host"], ttl=3600) + fake.cache.put("shared:tags:t", ["host"], ttl=3600) fake.query_responses['"f"'] = [ - {"host": "a", "f": "not found", "cnt": 2}, # sanitized: f_not_found - {"host": "a", "f": "not_found", "cnt": 3}, # already plain: f_not_found + {"host": "a", "f": "not found", "cnt": 2}, # sanitized: f_not_found + {"host": "a", "f": "not_found", "cnt": 3}, # already plain: f_not_found ] process_scheduled_call( - fake, _dt_ns(2_000_000_000_000_000_000), + fake, + _dt_ns(2_000_000_000_000_000_000), args={"table": "t", "fields": "f"}, ) @@ -738,17 +782,16 @@ def test_scheduled_sanitization_collision_sums_and_warns(monkeypatch, tmp_path): assert any("collision" in m for lvl, m in fake.logs if lvl == "warn") -def test_scheduled_call_time_ns_preserves_microseconds(monkeypatch, tmp_path): +def test_scheduled_call_time_ns_preserves_microseconds(): fake = FakeInfluxdb3Local() - monkeypatch.setattr("valuecounter.LineBuilder", LineBuilder, raising=False) - monkeypatch.setenv("PLUGIN_DIR", str(tmp_path)) fake.cache.put("vc:scheduled:last_call_ns:t", 1_700_000_000_000_000_000, ttl=None) - fake.cache.put("vc:tags:t", ["host"], ttl=3600) + fake.cache.put("shared:tags:t", ["host"], ttl=3600) fake.query_responses['"f"'] = [{"host": "a", "f": "x", "cnt": 1}] # datetime with microseconds = 123456 -> expected ns suffix carries it exactly from datetime import datetime, timezone + call_time = datetime(2026, 1, 1, 0, 0, 0, 123456, tzinfo=timezone.utc) expected_ns = int(call_time.timestamp()) * 1_000_000_000 + 123456 * 1000 @@ -758,30 +801,31 @@ def test_scheduled_call_time_ns_preserves_microseconds(monkeypatch, tmp_path): assert fake.writes[0].endswith(f" {expected_ns}") -import time from valuecounter import process_writes -def test_wal_first_batch_accumulates_no_emit(monkeypatch, tmp_path): +def test_wal_first_batch_accumulates_no_emit(monkeypatch): fake = FakeInfluxdb3Local() - monkeypatch.setattr("valuecounter.LineBuilder", LineBuilder, raising=False) - monkeypatch.setenv("PLUGIN_DIR", str(tmp_path)) - fake.cache.put("vc:tags:http_requests", ["host"], ttl=3600) + fake.cache.put("shared:tags:http_requests", ["host"], ttl=3600) # Freeze time so period gate cannot fire fixed_now = 1_000_000_000_000_000_000 monkeypatch.setattr("valuecounter.time.time_ns", lambda: fixed_now) - table_batches = [{ - "table_name": "http_requests", - "rows": [ - {"host": "web-1", "status": 200, "time": fixed_now - 1000}, - {"host": "web-1", "status": 500, "time": fixed_now - 500}, - {"host": "web-1", "status": 200, "time": fixed_now - 100}, - ], - }] + table_batches = [ + { + "table_name": "http_requests", + "rows": [ + {"host": "web-1", "status": 200, "time": fixed_now - 1000}, + {"host": "web-1", "status": 500, "time": fixed_now - 500}, + {"host": "web-1", "status": 200, "time": fixed_now - 100}, + ], + } + ] - process_writes(fake, table_batches, args={"fields": "status", "period_seconds": "60"}) + process_writes( + fake, table_batches, args={"fields": "status", "period_seconds": "60"} + ) # No emit yet — last_emit_ns == now_ns means period not elapsed assert fake.writes == [] @@ -795,34 +839,41 @@ def test_wal_first_batch_accumulates_no_emit(monkeypatch, tmp_path): assert series_entry["last_emit_ns"] == fixed_now -def test_wal_period_elapsed_emits_and_subtracts(monkeypatch, tmp_path): +def test_wal_period_elapsed_emits_and_subtracts(monkeypatch): fake = FakeInfluxdb3Local() - monkeypatch.setattr("valuecounter.LineBuilder", LineBuilder, raising=False) - monkeypatch.setenv("PLUGIN_DIR", str(tmp_path)) - fake.cache.put("vc:tags:t", ["host"], ttl=3600) + fake.cache.put("shared:tags:t", ["host"], ttl=3600) # Pre-seed series state with last_emit_ns far in the past sh = _series_key("t", {"host": "a"}) - fake.cache.put(f"vc:wal:{sh}", { - "table": "t", "tags": {"host": "a"}, - "counts": {"status_200": 5}, - "last_emit_ns": 0, - }, ttl=120) + fake.cache.put( + f"vc:wal:{sh}", + { + "table": "t", + "tags": {"host": "a"}, + "counts": {"status_200": 5}, + "last_emit_ns": 0, + }, + ttl=120, + ) fake.cache.put("vc:wal:_index:t", [sh], ttl=120) fixed_now = 1_000_000_000_000_000_000 monkeypatch.setattr("valuecounter.time.time_ns", lambda: fixed_now) # New batch adds two more 200s (so during snapshot counts=5, after add counts=7) - table_batches = [{ - "table_name": "t", - "rows": [ - {"host": "a", "status": 200, "time": fixed_now - 100}, - {"host": "a", "status": 200, "time": fixed_now - 50}, - ], - }] + table_batches = [ + { + "table_name": "t", + "rows": [ + {"host": "a", "status": 200, "time": fixed_now - 100}, + {"host": "a", "status": 200, "time": fixed_now - 50}, + ], + } + ] - process_writes(fake, table_batches, args={"fields": "status", "period_seconds": "60"}) + process_writes( + fake, table_batches, args={"fields": "status", "period_seconds": "60"} + ) # One emit with counts=7 (5 pre + 2 new) assert len(fake.writes) == 1 @@ -833,26 +884,32 @@ def test_wal_period_elapsed_emits_and_subtracts(monkeypatch, tmp_path): assert new_state["last_emit_ns"] == fixed_now -def test_wal_write_failure_retains_counts(monkeypatch, tmp_path): +def test_wal_write_failure_retains_counts(monkeypatch): fake = FakeInfluxdb3Local() - monkeypatch.setattr("valuecounter.LineBuilder", LineBuilder, raising=False) - monkeypatch.setenv("PLUGIN_DIR", str(tmp_path)) - fake.cache.put("vc:tags:t", ["host"], ttl=3600) + fake.cache.put("shared:tags:t", ["host"], ttl=3600) sh = _series_key("t", {"host": "a"}) - fake.cache.put(f"vc:wal:{sh}", { - "table": "t", "tags": {"host": "a"}, - "counts": {"status_200": 3}, - "last_emit_ns": 0, - }, ttl=120) + fake.cache.put( + f"vc:wal:{sh}", + { + "table": "t", + "tags": {"host": "a"}, + "counts": {"status_200": 3}, + "last_emit_ns": 0, + }, + ttl=120, + ) fake.cache.put("vc:wal:_index:t", [sh], ttl=120) fixed_now = 1_000_000_000_000_000_000 monkeypatch.setattr("valuecounter.time.time_ns", lambda: fixed_now) fake.fail_write = True - process_writes(fake, [{"table_name": "t", "rows": []}], - args={"fields": "status", "period_seconds": "60"}) + process_writes( + fake, + [{"table_name": "t", "rows": []}], + args={"fields": "status", "period_seconds": "60"}, + ) # Counts retained for retry; last_emit_ns NOT advanced state = fake.cache._d[f"vc:wal:{sh}"] @@ -861,51 +918,63 @@ def test_wal_write_failure_retains_counts(monkeypatch, tmp_path): assert any("emit failed" in m for _, m in fake.logs) -def test_wal_index_pruning_removes_expired_hashes(monkeypatch, tmp_path): +def test_wal_index_pruning_removes_expired_hashes(monkeypatch): fake = FakeInfluxdb3Local() - monkeypatch.setattr("valuecounter.LineBuilder", LineBuilder, raising=False) - monkeypatch.setenv("PLUGIN_DIR", str(tmp_path)) - fake.cache.put("vc:tags:t", ["host"], ttl=3600) + fake.cache.put("shared:tags:t", ["host"], ttl=3600) sh_live = _series_key("t", {"host": "a"}) sh_dead = _series_key("t", {"host": "b"}) - fake.cache.put(f"vc:wal:{sh_live}", { - "table": "t", "tags": {"host": "a"}, - "counts": {}, "last_emit_ns": 0, - }, ttl=120) + fake.cache.put( + f"vc:wal:{sh_live}", + { + "table": "t", + "tags": {"host": "a"}, + "counts": {}, + "last_emit_ns": 0, + }, + ttl=120, + ) # sh_dead deliberately NOT in the cache (simulating TTL expiry) fake.cache.put("vc:wal:_index:t", [sh_live, sh_dead], ttl=120) fixed_now = 1_000_000_000_000_000_000 monkeypatch.setattr("valuecounter.time.time_ns", lambda: fixed_now) - process_writes(fake, [{"table_name": "t", "rows": []}], - args={"fields": "status", "period_seconds": "60"}) + process_writes( + fake, + [{"table_name": "t", "rows": []}], + args={"fields": "status", "period_seconds": "60"}, + ) pruned = fake.cache._d["vc:wal:_index:t"] assert pruned == [sh_live] -def test_wal_cross_db_write(monkeypatch, tmp_path): +def test_wal_cross_db_write(monkeypatch): fake = FakeInfluxdb3Local() - monkeypatch.setattr("valuecounter.LineBuilder", LineBuilder, raising=False) - monkeypatch.setenv("PLUGIN_DIR", str(tmp_path)) - fake.cache.put("vc:tags:t", ["host"], ttl=3600) + fake.cache.put("shared:tags:t", ["host"], ttl=3600) sh = _series_key("t", {"host": "a"}) - fake.cache.put(f"vc:wal:{sh}", { - "table": "t", "tags": {"host": "a"}, - "counts": {"status_200": 1}, - "last_emit_ns": 0, - }, ttl=120) + fake.cache.put( + f"vc:wal:{sh}", + { + "table": "t", + "tags": {"host": "a"}, + "counts": {"status_200": 1}, + "last_emit_ns": 0, + }, + ttl=120, + ) fake.cache.put("vc:wal:_index:t", [sh], ttl=120) fixed_now = 1_000_000_000_000_000_000 monkeypatch.setattr("valuecounter.time.time_ns", lambda: fixed_now) - process_writes(fake, [{"table_name": "t", "rows": []}], - args={"fields": "status", "period_seconds": "60", - "dest_database": "rollups"}) + process_writes( + fake, + [{"table_name": "t", "rows": []}], + args={"fields": "status", "period_seconds": "60", "dest_database": "rollups"}, + ) assert fake.writes == [] assert len(fake.cross_db_writes) == 1 diff --git a/influxdata/valuecounter/valuecounter.py b/influxdata/valuecounter/valuecounter.py index 4827c9f..f035f6f 100644 --- a/influxdata/valuecounter/valuecounter.py +++ b/influxdata/valuecounter/valuecounter.py @@ -17,7 +17,13 @@ { "name": "period_seconds", "example": "60", - "description": "Emission period in seconds. Cache TTL = 2x this value. Defaults to 60.", + "description": "Emission period in seconds, at least 1. Cache TTL = 2x this value. Defaults to 60.", + "required": false + }, + { + "name": "period", + "example": "5min", + "description": "Emission period as a duration (e.g., '30s', '5min', '1h'), at least 1s. Units: 's', 'min', 'h', 'd', 'w'. Overridden by period_seconds when both are set.", "required": false }, { @@ -69,13 +75,19 @@ """ import hashlib -import os import re import time -import tomllib import uuid from dataclasses import dataclass -from pathlib import Path + +from influxdata_plugin_utils.config import Validator, load_plugin_config +from influxdata_plugin_utils.introspection import get_tag_names +from influxdata_plugin_utils.parsing import ( + parse_delimited_list, + parse_int, + parse_timedelta, +) +from influxdata_plugin_utils.write import build_line, write_data # At server runtime LineBuilder is injected as a builtin. In test environments # pytest patches this module-level name to a vendored copy. The reference in @@ -110,15 +122,16 @@ def _validate_identifier(name, what): raise ValueError(f"invalid {what}: {name!r}") -_DURATION_RE = re.compile(r"^\s*(\d+)\s*([smhd])\s*$") -_DURATION_UNITS = {"s": 1, "m": 60, "h": 3600, "d": 86400} +_BARE_MINUTES_RE = re.compile(r"^\s*(\d+)\s*m\s*$") -def _parse_duration(s): - m = _DURATION_RE.match(s) - if not m: - raise ValueError(f"invalid duration: {s!r} (expected e.g. '60s', '5m', '1h', '1d')") - return int(m.group(1)) * _DURATION_UNITS[m.group(2)] +def _parse_period_seconds(raw): + match = _BARE_MINUTES_RE.match(str(raw)) + text = f"{match.group(1)}min" if match else str(raw) + seconds = int(parse_timedelta(text).total_seconds()) + if seconds < 1: + raise ValueError(f"invalid period: {raw!r} (must be at least 1 second)") + return seconds @dataclass @@ -130,126 +143,82 @@ class Config: table: str = "" -_MODE_A_ALLOWED = {"fields", "output_suffix", "period_seconds", "period", "dest_database", "config_file_path"} -_MODE_B_ALLOWED = {"table", "fields", "output_suffix", "dest_database", "config_file_path"} +_MODE_ALLOWED = { + "wal": {"fields", "output_suffix", "period", "period_seconds", "dest_database"}, + "scheduled": {"table", "fields", "output_suffix", "dest_database"}, +} -def _parse_inline_args(args, mode): - if mode == "wal": - allowed = _MODE_A_ALLOWED - elif mode == "scheduled": - allowed = _MODE_B_ALLOWED - else: - raise ValueError(f"unknown mode: {mode!r}") +def _parse_fields(raw): + """Field names split on any whitespace; TOML may deliver a list already.""" + if isinstance(raw, (list, tuple)): + return parse_delimited_list(raw) + return str(raw).split() - for key in args: - if key not in allowed: - # special-case better error messages for mode-incompatible knobs - if key == "table" and mode == "wal": - raise ValueError( - "vc-wal: 'table' is determined by the trigger-spec, not inline args" - ) - if key in ("period_seconds", "period") and mode == "scheduled": - raise ValueError( - "vc-scheduled: 'period_seconds'/'period' is not used; Mode B is drift-based" - ) - raise ValueError(f"unknown arg: {key!r}") - - cfg = Config(fields=[]) - if "fields" in args: - cfg.fields = args["fields"].split() - if "output_suffix" in args: - cfg.output_suffix = args["output_suffix"] - if "dest_database" in args: - cfg.dest_database = args["dest_database"] - if "table" in args: - cfg.table = args["table"] - if "period" in args: - cfg.period_seconds = _parse_duration(args["period"]) - if "period_seconds" in args: - cfg.period_seconds = int(args["period_seconds"]) - return cfg +_COMMON_VALIDATORS = [ + Validator("fields", default="", cast=_parse_fields), + Validator("output_suffix", default="_valuecounts", cast=str), + Validator("dest_database", default="", cast=str), +] -def _load_toml_config(path, mode): - if mode == "wal": - allowed = _MODE_A_ALLOWED - {"config_file_path"} - elif mode == "scheduled": - allowed = _MODE_B_ALLOWED - {"config_file_path"} - else: - raise ValueError(f"unknown mode: {mode!r}") +# Only mode-valid keys get a validator, so registered defaults never trip the +# unknown-key check below. +_MODE_VALIDATORS = { + "wal": _COMMON_VALIDATORS, + "scheduled": _COMMON_VALIDATORS + [Validator("table", default="", cast=str)], +} - try: - with open(path, "rb") as f: - data = tomllib.load(f) - except FileNotFoundError: - raise - except tomllib.TOMLDecodeError as e: - raise ValueError(f"TOML parse error in {path}: {e}") - - for key in data: - if key not in allowed: - if key == "table" and mode == "wal": - raise ValueError( - "vc-wal: 'table' is determined by the trigger-spec, not the TOML" - ) - if key in ("period_seconds", "period") and mode == "scheduled": - raise ValueError( - "vc-scheduled: 'period'/'period_seconds' is not used; Mode B is drift-based" - ) - raise ValueError(f"unknown TOML key: {key!r}") - - cfg = Config(fields=[]) - if "fields" in data: - f = data["fields"] - if isinstance(f, str): - f = f.split() - cfg.fields = list(f) - if "output_suffix" in data: - cfg.output_suffix = data["output_suffix"] - if "dest_database" in data: - cfg.dest_database = data["dest_database"] - if "table" in data: - cfg.table = data["table"] - if "period" in data: - cfg.period_seconds = _parse_duration(data["period"]) - if "period_seconds" in data: - cfg.period_seconds = int(data["period_seconds"]) - return cfg + +def _reject_unknown_keys(loaded, mode, from_toml): + key_label = "TOML key" if from_toml else "arg" + source_label = "the TOML" if from_toml else "inline args" + allowed = _MODE_ALLOWED[mode] + + for key in (name.lower() for name in loaded.as_dict()): + if key in allowed: + continue + # special-case better error messages for mode-incompatible knobs + if key == "table" and mode == "wal": + raise ValueError( + f"vc-wal: 'table' is determined by the trigger-spec, not {source_label}" + ) + if key in ("period", "period_seconds") and mode == "scheduled": + raise ValueError( + "vc-scheduled: 'period'/'period_seconds' is not used; Mode B is drift-based" + ) + raise ValueError(f"unknown {key_label}: {key!r}") -def _resolve_config(args, plugin_dir, mode): - args = dict(args) # defensive copy - cfp = args.pop("config_file_path", None) +def _resolve_config(args, mode): + if mode not in _MODE_ALLOWED: + raise ValueError(f"unknown mode: {mode!r}") - if cfp is not None and args: + args = dict(args or {}) # defensive copy + config_file_path = args.get("config_file_path") + if config_file_path is not None and len(args) > 1: raise ValueError("set either config_file_path or inline args, not both") - if cfp is not None: - if plugin_dir is not None: - path = os.path.join(plugin_dir, cfp) - else: - # Fallbacks for servers where the operator has not exported PLUGIN_DIR: - # - INFLUXDB3_PLUGIN_DIR: set when the server is configured via env var - # - VIRTUAL_ENV: exported by the processing engine; default venv is /.venv - candidates = [] - if influxdb3_plugin_dir := os.environ.get("INFLUXDB3_PLUGIN_DIR"): - candidates.append(influxdb3_plugin_dir) - if virtual_env := os.environ.get("VIRTUAL_ENV"): - candidates.append(str(Path(virtual_env).parent)) - - path = None - for base in candidates: - candidate = os.path.join(base, cfp) - if os.path.exists(candidate): - path = candidate - break - if path is None: - # Original default: resolve against the current directory. - path = os.path.join(".", cfp) - cfg = _load_toml_config(path, mode=mode) - else: - cfg = _parse_inline_args(args, mode=mode) + loaded = load_plugin_config( + args, + validators=_MODE_VALIDATORS[mode], + source="toml" if config_file_path else "args", + ) + _reject_unknown_keys(loaded, mode, from_toml=bool(config_file_path)) + + cfg = Config( + fields=[str(f) for f in loaded.fields], + output_suffix=loaded.output_suffix, + dest_database=loaded.dest_database, + table=str(loaded.get("table") or ""), + ) + + if mode == "wal": + # 'period_seconds' wins when both spellings are present + if (period := loaded.get("period")) is not None: + cfg.period_seconds = _parse_period_seconds(period) + if (period_seconds := loaded.get("period_seconds")) is not None: + cfg.period_seconds = parse_int(period_seconds, minimum=1) if not cfg.fields: raise ValueError("config error: 'fields' is empty or missing") @@ -264,7 +233,9 @@ def _resolve_config(args, plugin_dir, mode): _validate_identifier(f, "field") if cfg.output_suffix == "": - raise ValueError("config error: 'output_suffix' cannot be empty (would risk feedback loop in Mode A and ambiguity in Mode B)") + raise ValueError( + "config error: 'output_suffix' cannot be empty (would risk feedback loop in Mode A and ambiguity in Mode B)" + ) return cfg @@ -287,35 +258,14 @@ def _extract_tags(row, tag_names): return out -def _build_line(table, tags, counts, output_suffix, ts_ns, line_builder_cls): - if not counts: - return None - lb = line_builder_cls(f"{table}{output_suffix}") - for k, v in tags.items(): - lb.tag(k, str(v)) - for k, v in counts.items(): - lb.int64_field(k, int(v)) - lb.time_ns(ts_ns) - return lb - - -def _get_tag_names(influxdb3_local, table_name): - key = f"vc:tags:{table_name}" - cached = influxdb3_local.cache.get(key) - if cached is not None: - return cached - res = influxdb3_local.query( - """ - SELECT column_name - FROM information_schema.columns - WHERE table_name = $tbl - AND data_type = 'Dictionary(Int32, Utf8)' - """, - {"tbl": table_name}, +def _build_rollup_line(table, tags, counts, output_suffix, ts_ns): + return build_line( + LineBuilder, + f"{table}{output_suffix}", + tags=tags, + fields=counts, + time_ns=ts_ns, ) - tag_names = [r["column_name"] for r in res] - influxdb3_local.cache.put(key, tag_names, ttl=3600) - return tag_names def _build_scheduled_query(table, tag_names, field_name): @@ -339,33 +289,23 @@ def _build_scheduled_query(table, tag_names, field_name): ) -def _query_field_distribution(influxdb3_local, table, tag_names, field_name, start_ns, end_ns): +def _query_field_distribution( + influxdb3_local, table, tag_names, field_name, start_ns, end_ns +): sql = _build_scheduled_query(table, tag_names, field_name) params = {"start_ns": str(start_ns), "end_ns": str(end_ns)} return influxdb3_local.query(sql, params) -class _BatchLines: - def __init__(self, line_builders): - self._line_builders = list(line_builders) - self._built = None - - def build(self): - if self._built is None: - lines = [b.build() for b in self._line_builders] - if not lines: - raise ValueError("batch_write received no lines to build") - self._built = "\n".join(lines) - return self._built - - def process_scheduled_call(influxdb3_local, call_time, args=None): task_id = uuid.uuid4().hex[:8] # call_time arrives as a PyDateTime per system_py.rs:847,867 - call_time_ns = int(call_time.timestamp()) * 1_000_000_000 + call_time.microsecond * 1000 + call_time_ns = ( + int(call_time.timestamp()) * 1_000_000_000 + call_time.microsecond * 1000 + ) - cfg = _resolve_config(args or {}, os.environ.get("PLUGIN_DIR"), mode="scheduled") + cfg = _resolve_config(args, mode="scheduled") anchor_key = f"vc:scheduled:last_call_ns:{cfg.table}" last_call_ns = influxdb3_local.cache.get(anchor_key) @@ -377,7 +317,7 @@ def process_scheduled_call(influxdb3_local, call_time, args=None): ) return - tag_names = _get_tag_names(influxdb3_local, cfg.table) + tag_names = get_tag_names(influxdb3_local, cfg.table) for t in tag_names: _validate_identifier(t, "tag column") @@ -390,8 +330,12 @@ def process_scheduled_call(influxdb3_local, call_time, args=None): for field_name in cfg.fields: try: rows = _query_field_distribution( - influxdb3_local, cfg.table, tag_names, field_name, - window_start_ns, window_end_ns, + influxdb3_local, + cfg.table, + tag_names, + field_name, + window_start_ns, + window_end_ns, ) except Exception as e: influxdb3_local.error( @@ -431,21 +375,20 @@ def process_scheduled_call(influxdb3_local, call_time, args=None): return builders = [ - _build_line(ss["table"], ss["tags"], ss["counts"], cfg.output_suffix, - call_time_ns, LineBuilder) + _build_rollup_line( + ss["table"], ss["tags"], ss["counts"], cfg.output_suffix, call_time_ns + ) for ss in series.values() ] - builders = [b for b in builders if b is not None] - if not builders: - influxdb3_local.cache.put(anchor_key, call_time_ns, ttl=None) - return try: - batch = _BatchLines(builders) - if cfg.dest_database: - influxdb3_local.write_sync_to_db(cfg.dest_database, batch, no_sync=True) - else: - influxdb3_local.write_sync(batch, no_sync=True) + write_data( + influxdb3_local, + builders, + retries=0, + no_sync=True, + database=cfg.dest_database or None, + ) except Exception as e: influxdb3_local.error(f"[{task_id}] vc-scheduled: write failed: {e}") return # anchor unchanged → next fire's window covers two periods @@ -457,7 +400,7 @@ def process_writes(influxdb3_local, table_batches, args=None): task_id = uuid.uuid4().hex[:8] now_ns = time.time_ns() - cfg = _resolve_config(args or {}, os.environ.get("PLUGIN_DIR"), mode="wal") + cfg = _resolve_config(args, mode="wal") period_ns = cfg.period_seconds * 1_000_000_000 ttl = 2 * cfg.period_seconds @@ -470,7 +413,7 @@ def process_writes(influxdb3_local, table_batches, args=None): by_table.setdefault(name, []).extend(batch["rows"]) for table_name, rows in by_table.items(): - tag_names = _get_tag_names(influxdb3_local, table_name) + tag_names = get_tag_names(influxdb3_local, table_name) for t in tag_names: _validate_identifier(t, "tag column") @@ -527,7 +470,9 @@ def process_writes(influxdb3_local, table_batches, args=None): continue if now_ns - state["last_emit_ns"] < period_ns: continue - to_emit.append((sh, state["table"], dict(state["tags"]), dict(state["counts"]))) + to_emit.append( + (sh, state["table"], dict(state["tags"]), dict(state["counts"])) + ) # Prune index — only live hashes survive if live != active_hashes: @@ -540,19 +485,18 @@ def process_writes(influxdb3_local, table_batches, args=None): continue builders = [ - _build_line(t_, tags, counts, cfg.output_suffix, now_ns, LineBuilder) + _build_rollup_line(t_, tags, counts, cfg.output_suffix, now_ns) for _sh, t_, tags, counts in to_emit ] - builders = [b for b in builders if b is not None] - if not builders: - continue try: - batch = _BatchLines(builders) - if cfg.dest_database: - influxdb3_local.write_sync_to_db(cfg.dest_database, batch, no_sync=True) - else: - influxdb3_local.write_sync(batch, no_sync=True) + write_data( + influxdb3_local, + builders, + retries=0, + no_sync=True, + database=cfg.dest_database or None, + ) except Exception as e: influxdb3_local.error( f"[{task_id}] vc-wal: emit failed, counts retained for retry: {e}" diff --git a/influxdata/valuecounter/valuecounter_config.toml b/influxdata/valuecounter/valuecounter_config.toml index 80ad4a6..917ed75 100644 --- a/influxdata/valuecounter/valuecounter_config.toml +++ b/influxdata/valuecounter/valuecounter_config.toml @@ -7,7 +7,7 @@ # Mode A — WAL trigger (process_writes) # ------------------------------------------------------------ # 'table' is determined by the WAL trigger spec (`table:`); do NOT set it here. -# period = "60s" +# period = "60s" # or period_seconds = 60; units: s, min, h, d, w # fields = ["status", "method"] # output_suffix = "_valuecounts" # dest_database = "rollups" From a6925ac7cac6a6fb3e8b8a6f0dc35235fad4fff1 Mon Sep 17 00:00:00 2001 From: Aliaksei Kharlap Date: Wed, 5 Aug 2026 20:37:49 +0300 Subject: [PATCH 2/4] refactor threshold_deadman_checks to use utils-package --- influxdata/library/plugin_library.json | 4 +- influxdata/threshold_deadman_checks/README.md | 48 +- .../threshold_deadman_checks/manifest.toml | 4 +- .../threshold_deadman_checks/requirements.txt | 1 + .../threshold_deadman_checks_plugin.py | 1156 ++++++++--------- .../threshold_deadman_config_data_writes.toml | 17 +- .../threshold_deadman_config_scheduler.toml | 19 +- 7 files changed, 605 insertions(+), 644 deletions(-) diff --git a/influxdata/library/plugin_library.json b/influxdata/library/plugin_library.json index a076235..03e49c5 100644 --- a/influxdata/library/plugin_library.json +++ b/influxdata/library/plugin_library.json @@ -138,8 +138,8 @@ "path": "influxdata/notifier/notifier_plugin.py" } ], - "required_libraries": ["requests"], - "last_update": "2025-06-16", + "required_libraries": ["influxdata-plugin-utils>=0.3.0", "requests"], + "last_update": "2026-08-03", "trigger_types_supported": ["scheduler", "data_writes"] }, { diff --git a/influxdata/threshold_deadman_checks/README.md b/influxdata/threshold_deadman_checks/README.md index 8be5f57..e2e3be2 100644 --- a/influxdata/threshold_deadman_checks/README.md +++ b/influxdata/threshold_deadman_checks/README.md @@ -17,11 +17,11 @@ This plugin includes a JSON metadata schema in its docstring that defines suppor ### Required parameters -| Parameter | Type | Default | Description | -|---------------|--------|----------|---------------------------------------------------------------------------------------------------| -| `measurement` | string | required | Measurement to monitor for deadman alerts and aggregation-based conditions | -| `senders` | string | required | Dot-separated notification channels with multi-channel notification integration | -| `window` | string | required | Time window for periodic data presence checking | +| Parameter | Type | Default | Description | +|---------------|--------|----------|------------------------------------------------------------------------------------------------------------------------------------------| +| `measurement` | string | required | Measurement to monitor for deadman alerts and aggregation-based conditions | +| `senders` | string | required | Dot-separated notification channels with multi-channel notification integration | +| `window` | string | required | Time window for periodic data presence checking. Format: ``, units: `s`, `min`, `h`, `d`, `w`. Must be a positive duration | ### Data write trigger parameters @@ -33,12 +33,12 @@ This plugin includes a JSON metadata schema in its docstring that defines suppor ### Threshold check parameters -| Parameter | Type | Default | Description | -|----------------------------|---------|---------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------| -| `field_aggregation_values` | string | none | Multi-level aggregation conditions with aggregation support for avg, min, max, count, sum, median, stddev, first_value, last_value, var, and approx_median values | -| `deadman_check` | boolean | false | Enable deadman detection to monitor for data absence and missing data streams | -| `interval` | string | "5min" | Configurable aggregation time interval for batch processing with performance optimization | -| `trigger_count` | number | 1 | Configurable triggers requiring multiple consecutive failures before alerting | +| Parameter | Type | Default | Description | +|----------------------------|---------|---------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| `field_aggregation_values` | string | none | Multi-level aggregation conditions with aggregation support for avg, min, max, count, sum, median, stddev, first_value, last_value, var, and approx_median values | +| `deadman_check` | boolean | false | Enable deadman detection to monitor for data absence and missing data streams | +| `interval` | string | "5min" | Aggregation time interval used in `DATE_BIN`. Format: ``, units: `s`, `min`, `h`, `d`, `w` | +| `trigger_count` | number | 1 | Breaches required before alerting. Threshold checks count consecutive breaches per row identifier, including across the time bins of a single run; deadman checks count consecutive runs without data | ### Notification parameters @@ -57,7 +57,11 @@ This plugin includes a JSON metadata schema in its docstring that defines suppor |--------------------|--------|---------|----------------------------------------------------------------------------------| | `config_file_path` | string | none | TOML config file path relative to `PLUGIN_DIR` (required for TOML configuration) | -*To use a TOML configuration file, set the `PLUGIN_DIR` environment variable and specify the `config_file_path` in the trigger arguments.* This is in addition to the `--plugin-dir` flag when starting InfluxDB 3. +*To use a TOML configuration file, set the `PLUGIN_DIR` environment variable and specify the `config_file_path` in the trigger arguments.* This is in addition to the `--plugin-dir` flag when starting InfluxDB 3. Relative paths are resolved against the first directory that is set: `PLUGIN_DIR`, then `INFLUXDB3_PLUGIN_DIR`, then the parent of `VIRTUAL_ENV`. Only that directory is used — the file is not looked up in the remaining ones. + +When `config_file_path` is set, the TOML file provides the whole configuration and inline trigger arguments are ignored. `INFLUXDB3_AUTH_TOKEN` from the environment still applies when `influxdb3_auth_token` is not set in the file. In TOML, `senders`, `field_conditions`, and `field_aggregation_values` use native structures instead of the inline string formats. + +Data write triggers cache the loaded configuration for 10 minutes to keep the write path fast, so configuration changes take effect within that window. Example TOML configuration files provided: @@ -77,6 +81,7 @@ The plugin assumes that the table schema is already defined in the database, as ## Software requirements - **InfluxDB v3 Core/Enterprise**: with the Processing Engine enabled. +- **Python packages**: `influxdata-plugin-utils>=0.3.0`, `requests` - **Notification Sender Plugin for InfluxDB 3**: This plugin is required for sending notifications. See the [influxdata/notifier plugin](../notifier/README.md). ## Installation steps @@ -94,6 +99,7 @@ The plugin assumes that the table schema is already defined in the database, as 2. **Install required Python packages**: ```bash + influxdb3 install package "influxdata-plugin-utils>=0.3.0" influxdb3 install package requests ``` @@ -110,7 +116,7 @@ influxdb3 create trigger \ --database mydb \ --path "gh:influxdata/threshold_deadman_checks/threshold_deadman_checks_plugin.py" \ --trigger-spec "every:10m" \ - --trigger-arguments "measurement=cpu,senders=slack,field_aggregation_values=temp:avg@>=30-ERROR,window=10m,trigger_count=3,deadman_check=true,slack_webhook_url=$SLACK_WEBHOOK_URL" \ + --trigger-arguments "measurement=cpu,senders=slack,field_aggregation_values=temp:avg@>=30-ERROR,window=10min,trigger_count=3,deadman_check=true,slack_webhook_url=$SLACK_WEBHOOK_URL" \ threshold_scheduler ``` @@ -159,7 +165,7 @@ influxdb3 create trigger \ --database sensors \ --path "gh:influxdata/threshold_deadman_checks/threshold_deadman_checks_plugin.py" \ --trigger-spec "every:5m" \ - --trigger-arguments "measurement=heartbeat,senders=slack,window=5m,deadman_check=true,slack_webhook_url=$SLACK_WEBHOOK_URL" \ + --trigger-arguments "measurement=heartbeat,senders=slack,window=5min,deadman_check=true,slack_webhook_url=$SLACK_WEBHOOK_URL" \ heartbeat_monitor influxdb3 enable trigger --database sensors heartbeat_monitor @@ -185,7 +191,7 @@ influxdb3 create trigger \ --database sensors \ --path "gh:influxdata/threshold_deadman_checks/threshold_deadman_checks_plugin.py" \ --trigger-spec "every:15m" \ - --trigger-arguments "measurement=heartbeat,senders=sms,window=10m,deadman_check=true,trigger_count=2,twilio_from_number=+1234567890,twilio_to_number=+0987654321,notification_deadman_text=CRITICAL: No heartbeat data from \$table between \$time_from and \$time_to" \ + --trigger-arguments "measurement=heartbeat,senders=sms,window=10min,deadman_check=true,trigger_count=2,twilio_from_number=+1234567890,twilio_to_number=+0987654321,notification_deadman_text=CRITICAL: No heartbeat data from \$table between \$time_from and \$time_to" \ heartbeat_monitor ``` @@ -198,7 +204,7 @@ influxdb3 create trigger \ --database monitoring \ --path "gh:influxdata/threshold_deadman_checks/threshold_deadman_checks_plugin.py" \ --trigger-spec "every:5m" \ - --trigger-arguments "measurement=system_metrics,senders=slack.discord,field_aggregation_values='cpu_usage:avg@>=80-WARN cpu_usage:avg@>=95-ERROR memory_usage:max@>=90-WARN',window=5m,interval=1min,trigger_count=3,slack_webhook_url=$SLACK_WEBHOOK_URL,discord_webhook_url=$DISCORD_WEBHOOK_URL" \ + --trigger-arguments "measurement=system_metrics,senders=slack.discord,field_aggregation_values='cpu_usage:avg@>=80-WARN cpu_usage:avg@>=95-ERROR memory_usage:max@>=90-WARN',window=5min,interval=1min,trigger_count=3,slack_webhook_url=$SLACK_WEBHOOK_URL,discord_webhook_url=$DISCORD_WEBHOOK_URL" \ system_threshold_monitor ``` @@ -228,7 +234,7 @@ influxdb3 create trigger \ --database comprehensive \ --path "gh:influxdata/threshold_deadman_checks/threshold_deadman_checks_plugin.py" \ --trigger-spec "every:10m" \ - --trigger-arguments "measurement=temperature_sensors,senders=whatsapp,field_aggregation_values='temperature:avg@>=35-WARN temperature:max@>=40-ERROR',window=15m,deadman_check=true,trigger_count=2,twilio_from_number=+1234567890,twilio_to_number=+0987654321" \ + --trigger-arguments "measurement=temperature_sensors,senders=whatsapp,field_aggregation_values='temperature:avg@>=35-WARN temperature:max@>=40-ERROR',window=15min,deadman_check=true,trigger_count=2,twilio_from_number=+1234567890,twilio_to_number=+0987654321" \ comprehensive_sensor_monitor ``` @@ -269,7 +275,7 @@ Handles real-time threshold monitoring on data writes. Evaluates incoming data a #### Issue: False positive alerts -**Solution**: Increase `trigger_count` to require more consecutive failures. Adjust threshold values to be less sensitive. Consider longer aggregation intervals for noisy data. +**Solution**: Increase `trigger_count` to require more consecutive breaches. In scheduled mode every `DATE_BIN` bin of the window counts as a breach, so keep `window`, `interval`, and `trigger_count` aligned. Adjust threshold values to be less sensitive. Consider longer aggregation intervals for noisy data. #### Issue: Missing deadman alerts @@ -333,12 +339,14 @@ Handles real-time threshold monitoring on data writes. Evaluates incoming data a - `$op_sym`: Operator symbol - `$compare_val`: Threshold value - `$actual`: Actual field value +- `$trigger_count`: Consecutive matches required before alerting +- `$row`: Unique identifier ### Row identification -The `row` variable uniquely identifies alert contexts using format: `measurement:level:tag1=value1:tag2=value2` +The `row` variable uniquely identifies alert contexts using format: `measurement:field[:aggregation]:level:tag1=value1:tag2=value2` (`aggregation` is present for scheduled threshold checks only). Tags without a value are omitted. -This ensures trigger counts are maintained independently for each unique combination of measurement, severity level, and tag values. +Trigger counts are maintained independently for each unique combination of measurement, field, aggregation, severity level, tag values, **and the condition's operator and threshold** — two conditions that differ only by threshold never share a count. ## Questions/Comments diff --git a/influxdata/threshold_deadman_checks/manifest.toml b/influxdata/threshold_deadman_checks/manifest.toml index a57b301..c49be1d 100644 --- a/influxdata/threshold_deadman_checks/manifest.toml +++ b/influxdata/threshold_deadman_checks/manifest.toml @@ -2,7 +2,7 @@ manifest_schema_version = "1.3" [plugin] name = "threshold_deadman_checks" -version = "1.2.0" +version = "2.0.0" description = "Provides comprehensive monitoring capabilities including deadman alerts and aggregation-based threshold checks. Supports both scheduler and data write triggers with multi-channel notifications." triggers = ["process_writes", "process_scheduled_call"] homepage = "https://www.influxdata.com/" @@ -16,7 +16,7 @@ exclude = [ [dependencies] database_version = ">=3.0.0" -python = ["requests"] +python = ["influxdata-plugin-utils>=0.3.0", "requests"] [[dependencies.plugins]] index_url = "https://github.com/influxdata/influxdb3_plugins/releases/download/registry/index.json" diff --git a/influxdata/threshold_deadman_checks/requirements.txt b/influxdata/threshold_deadman_checks/requirements.txt index 663bd1f..3349a3f 100644 --- a/influxdata/threshold_deadman_checks/requirements.txt +++ b/influxdata/threshold_deadman_checks/requirements.txt @@ -1 +1,2 @@ +influxdata-plugin-utils>=0.3.0 requests \ No newline at end of file diff --git a/influxdata/threshold_deadman_checks/threshold_deadman_checks_plugin.py b/influxdata/threshold_deadman_checks/threshold_deadman_checks_plugin.py index 41fe3aa..edb109b 100644 --- a/influxdata/threshold_deadman_checks/threshold_deadman_checks_plugin.py +++ b/influxdata/threshold_deadman_checks/threshold_deadman_checks_plugin.py @@ -22,8 +22,8 @@ }, { "name": "window", - "example": "5m", - "description": "Time window to check for data (e.g., '5m' for 5 minutes).", + "example": "5min", + "description": "Time window to check for data (e.g., '5min' for 5 minutes). Valid units: s, min, h, d, w. Must be a positive duration.", "required": true }, { @@ -35,7 +35,7 @@ { "name": "trigger_count", "example": "3", - "description": "Number of consecutive failed checks before sending an alert. Default: 1.", + "description": "Number of condition breaches before sending an alert. Threshold checks count consecutive breaches per row identifier, including across the time bins of a single run; deadman checks count consecutive runs without data. Default: 1.", "required": false }, { @@ -266,15 +266,21 @@ import random import re import time -import tomllib import uuid from collections import defaultdict from datetime import datetime, timedelta, timezone -from pathlib import Path from string import Template from urllib.parse import urlparse import requests +from influxdata_plugin_utils.config import Validator, load_plugin_config +from influxdata_plugin_utils.introspection import get_table_names, get_tag_names +from influxdata_plugin_utils.parsing import ( + parse_bool, + parse_delimited_list, + parse_int, + parse_timedelta, +) # Supported comparison operators _OP_FUNCS = { @@ -303,41 +309,134 @@ # List of keywords to exclude from argument validation in AVAILABLE_SENDERS EXCLUDED_KEYWORDS = ["headers", "token", "sid"] - -def get_all_measurements(influxdb3_local) -> list[str]: +# Alert severity levels accepted in conditions +ALLOWED_MESSAGE_LEVELS = ("INFO", "WARN", "ERROR", "CRITICAL") + +# Aggregations supported in field_aggregation_values +AVAILABLE_AGGREGATIONS = ( + "avg", + "count", + "sum", + "min", + "max", + "median", + "stddev", + "first_value", + "last_value", + "var", + "approx_median", +) + +_DEFAULT_NOTIFICATION_TEXT = ( + "[$level] InfluxDB 3 alert triggered. Condition $field $op_sym $compare_val " + "matched $trigger_count times($actual) — matched in row $row." +) +_DEFAULT_DEADMAN_TEXT = ( + "Deadman Alert: No data received from $table from $time_from to $time_to." +) +_DEFAULT_THRESHOLD_TEXT = ( + "[$level] Threshold Alert on table $table: $aggregation of $field $op_sym " + "$compare_val (actual: $actual) — matched in row $row." +) + + +def parse_window(raw) -> timedelta: + """Parse a check window, rejecting non-positive durations.""" + window: timedelta = parse_timedelta(raw) + if window <= timedelta(0): + raise ValueError(f"Invalid window: {raw!r} (must be a positive duration)") + return window + + +_COMMON_VALIDATORS = [ + Validator("measurement", required=True, cast=str), + Validator("senders", required=True), + Validator("trigger_count", default=1, cast=lambda raw: parse_int(raw, minimum=1)), + Validator( + "port_override", + default=8181, + cast=lambda raw: parse_int(raw, minimum=1, maximum=65535), + ), + Validator("notification_path", default="notify", cast=str), +] + +_WRITES_VALIDATORS = _COMMON_VALIDATORS + [ + Validator("field_conditions", required=True), + Validator("notification_text", default=_DEFAULT_NOTIFICATION_TEXT, cast=str), +] + +_SCHEDULED_VALIDATORS = _COMMON_VALIDATORS + [ + Validator("deadman_check", default=False, cast=parse_bool), + Validator("window", required=True, cast=parse_window), + Validator("interval", default="5min", cast=parse_timedelta), + Validator("notification_deadman_text", default=_DEFAULT_DEADMAN_TEXT, cast=str), + Validator("notification_threshold_text", default=_DEFAULT_THRESHOLD_TEXT, cast=str), +] + +_WRITES_CONFIG_CACHE_KEY = "thresholds:writes_config" +_WRITES_CONFIG_TTL_SECONDS = 10 * 60 + + +def _load_config( + influxdb3_local, args: dict, validators: list, task_id: str +) -> dict | None: """ - Retrieves a list of all tables of type 'BASE TABLE' from cache or the current InfluxDB database. + Load the plugin configuration, applying defaults and type casts. + + A TOML file referenced by 'config_file_path' replaces the inline arguments; + INFLUXDB3_AUTH_TOKEN from the environment is used when the token is not + configured explicitly. Args: influxdb3_local: InfluxDB client instance. + args (dict): Runtime arguments of the trigger. + validators (list): Validators providing defaults and casts for the mode. + task_id (str): Unique task identifier. Returns: - list[str]: List of table names (e.g., ["cpu", "memory", "disk"]). + dict | None: Config values keyed by lower-case name, or None if loading failed. """ - # check cache first - measurements: list = influxdb3_local.cache.get("measurements") - if measurements: - return measurements - - # if not in cache, query the database - result: list = influxdb3_local.query("SHOW TABLES") - measurements = [ - row["table_name"] for row in result if row.get("table_type") == "BASE TABLE" - ] + config_file_path = (args or {}).get("config_file_path") + if config_file_path and not str(config_file_path).endswith(".toml"): + influxdb3_local.error( + f"[{task_id}] Invalid config file format: expected a .toml file" + ) + return None + + try: + loaded = load_plugin_config( + args, + validators=validators, + env_keys=["INFLUXDB3_AUTH_TOKEN"], + source="toml" if config_file_path else "args", + ) + except Exception as e: + influxdb3_local.error(f"[{task_id}] Failed to load configuration: {e}") + return None + + return {key.lower(): value for key, value in loaded.as_dict().items()} - # cache the result for 1 hour - influxdb3_local.cache.put(f"measurements", measurements, 60 * 60) - return measurements +def get_measurement_tags(influxdb3_local, measurement: str, task_id: str) -> list[str]: + """Return the cached tag names of a measurement, logging when it has none.""" + tags: list[str] = get_tag_names(influxdb3_local, measurement) + if not tags: + # an empty list stays cached for an hour and would hide tags added later + tags = get_tag_names(influxdb3_local, measurement, use_cache=False) + if not tags: + influxdb3_local.info( + f"[{task_id}] No tags found for measurement '{measurement}'." + ) + return tags -def parse_senders(influxdb3_local, args: dict, task_id: str) -> dict: +def parse_senders(influxdb3_local, config: dict, task_id: str) -> dict: """ - Parse and validate sender configurations from input arguments. + Parse and validate sender configurations from the loaded config. Args: influxdb3_local: InfluxDB client instance. - args (dict): Input arguments containing "senders" and related configs. + config (dict): Loaded config containing "senders" and related settings. task_id (str): Unique task identifier. Returns: @@ -347,39 +446,31 @@ def parse_senders(influxdb3_local, args: dict, task_id: str) -> dict: Exception: If no valid senders are found. """ senders_config: defaultdict = defaultdict(dict) - senders: str | list = args.get("senders") - - if args["use_config_file"]: - if not isinstance(senders, list): - raise Exception( - f"[{task_id}] 'senders' must be a list when using config file" - ) - else: - senders = senders.split(".") + senders: list = parse_delimited_list(config["senders"], sep=".") for sender in senders: if sender not in AVAILABLE_SENDERS: influxdb3_local.warn(f"[{task_id}] Invalid sender type: {sender}") continue for key in AVAILABLE_SENDERS[sender]: - if key not in args and not any(ex in key for ex in EXCLUDED_KEYWORDS): + if key not in config and not any(ex in key for ex in EXCLUDED_KEYWORDS): influxdb3_local.warn( f"[{task_id}] Missing required argument for {sender}: {key}" ) senders_config.pop(sender, None) break if "url" in key and not validate_webhook_url( - influxdb3_local, sender, args[key], task_id + influxdb3_local, sender, config[key], task_id ): senders_config.pop(sender, None) break - if key not in args: + if key not in config: continue - senders_config[sender][key] = args[key] + senders_config[sender][key] = config[key] if not senders_config: - raise Exception(f"[{task_id}] No valid senders configured") + raise Exception("No valid senders configured") return senders_config @@ -435,61 +526,40 @@ def _coerce_value(raw: str) -> str | int | float | bool: return raw -def parse_field_conditions(influxdb3_local, args: dict, task_id: str) -> list: - """ - Parse a semicolon-separated list of field conditions or use values from config file. +def _conditions_from_entries(influxdb3_local, entries: list, task_id: str) -> list: + """Parse field conditions given as [field, operator, value, level] entries.""" + conditions: list = [] - Each condition has the form: - - where is one of: >, <, >=, <=, ==, != - Multiple conditions are separated by semicolons ':'. + for part in entries: + if not isinstance(part, (list, tuple)) or len(part) != 4: + influxdb3_local.warn( + f"[{task_id}] Invalid condition '{part}', expected [field, operator, value, level]" + ) + continue + field: str = str(part[0]) + op: str = str(part[1]) + if op not in _OP_FUNCS: + influxdb3_local.warn( + f"[{task_id}] Unsupported operator '{op}' in condition '{part}'" + ) + continue + value = part[2] + level: str = str(part[3]).strip().upper() + if level not in ALLOWED_MESSAGE_LEVELS: + influxdb3_local.warn( + f"[{task_id}] Invalid message level '{part[3]}' in condition '{part}'" + ) + continue + conditions.append((field, op, _OP_FUNCS[op], value, level)) - Args: - influxdb3_local: InfluxDB client instance. - args (dict): Input arguments containing "field_conditions". - task_id (str): Unique task identifier. + return conditions - Returns: - List of lists: [field_name (str), operator_fn (callable), compare_value, level] - Example: - parse_field_conditions("temp>30-ERROR:status=='ok'-INFO:count<=100-WARN") - [ - ["temp", operator.gt, 30, ERROR], - ["status", operator.eq, "ok", INFO], - ["count", operator.le, 100, WARN] - ] - """ - allowed_message_levels: tuple = ("INFO", "WARN", "ERROR", "CRITICAL") - cond_input: str | list = args.get("field_conditions") +def _conditions_from_string(influxdb3_local, raw: str, task_id: str) -> list: + """Parse field conditions given as '-' joined by ':'.""" conditions: list = [] - if args["use_config_file"]: - if not isinstance(cond_input, list): - raise Exception( - f"[{task_id}] 'field_conditions' must be a list when using config file" - ) - for part in cond_input: - field: str = str(part[0]) - op: str = str(part[1]) - if op not in _OP_FUNCS: - influxdb3_local.warn( - f"[{task_id}] Unsupported operator '{op}' in condition '{part}'" - ) - continue - value = part[2] - level = part[3] - if level not in allowed_message_levels: - influxdb3_local.warn( - f"[{task_id}] Invalid message level '{level}' in condition '{part}'" - ) - continue - conditions.append((field, _OP_FUNCS[op], value, level)) - if not conditions: - raise Exception(f"[{task_id}] No valid field conditions provided.") - return conditions - - for part in cond_input.split(":"): + for part in raw.split(":"): part = part.strip() if not part: continue @@ -503,7 +573,7 @@ def parse_field_conditions(influxdb3_local, args: dict, task_id: str) -> list: cond_expr, level = part.rsplit("-", 1) level = level.strip().upper() - if level not in allowed_message_levels: + if level not in ALLOWED_MESSAGE_LEVELS: influxdb3_local.warn( f"[{task_id}] Invalid message level '{level}' in condition '{part}'" ) @@ -523,41 +593,52 @@ def parse_field_conditions(influxdb3_local, args: dict, task_id: str) -> list: continue value = _coerce_value(raw_val) - conditions.append([field, _OP_FUNCS[op], value, level]) + conditions.append((field, op, _OP_FUNCS[op], value, level)) - if not conditions: - raise Exception(f"[{task_id}] No valid field conditions provided.") return conditions -def parse_port_override(args: dict, task_id: str) -> int: +def parse_field_conditions(influxdb3_local, config: dict, task_id: str) -> list: """ - Parse and validate the 'port_override' argument, converting it from string to int. + Parse the field conditions used by the data write trigger. + + Conditions come either as entries of [field, operator, value, level] (TOML) or + as a string of '-' expressions separated by ':'. Args: - args (dict): Runtime arguments containing 'port_override'. - task_id (str): Unique task identifier for logging context. + influxdb3_local: InfluxDB client instance. + config (dict): Loaded config containing "field_conditions". + task_id (str): Unique task identifier. Returns: - int: Parsed port number (1–65535), or 8181 if not provided. + list[tuple]: Tuples of (field_name, operator, operator_fn, compare_value, level). Raises: - Exception: If 'port_override' is provided but is not a valid integer in the range 1–65535. - """ - raw: str | int = args.get("port_override", 8181) + Exception: If the value has an unsupported type or no valid conditions are found. - try: - port = int(raw) - except (TypeError, ValueError): - raise Exception(f"[{task_id}] Invalid port_override, not an integer: {raw!r}") + Example: + "temp>30-ERROR:status=='ok'-INFO:count<=100-WARN" + [ + ("temp", ">", operator.gt, 30, "ERROR"), + ("status", "==", operator.eq, "ok", "INFO"), + ("count", "<=", operator.le, 100, "WARN"), + ] + """ + raw: str | list = config["field_conditions"] - # Validate port range - if not (1 <= port <= 65535): + if isinstance(raw, (list, tuple)): + conditions = _conditions_from_entries(influxdb3_local, raw, task_id) + elif isinstance(raw, str): + conditions = _conditions_from_string(influxdb3_local, raw, task_id) + else: raise Exception( - f"[{task_id}] Invalid port_override, must be between 1 and 65535: {port}" + "'field_conditions' must be a list of entries or a string, " + f"got {type(raw).__name__}" ) - return port + if not conditions: + raise Exception("No valid field conditions provided.") + return conditions def interpolate_notification_text(text: str, row_data: dict) -> str: @@ -589,8 +670,8 @@ def send_notification( payload (dict): Dict to serialize as JSON in the POST body. task_id (str): Unique task identifier. - Raises: - requests.RequestException: If all retries fail or a non-2xx response is received. + Request failures and non-2xx responses are retried; after the final attempt + the error is logged and the alert is dropped. """ url: str = f"http://localhost:{port}/api/v3/engine/{path}" headers: dict = { @@ -626,110 +707,132 @@ def send_notification( ) +def generate_cache_key( + measurement: str, + field: str, + level: str, + row: dict, + tags: list, + aggregation: str | None = None, +) -> str: + """Generate the row identifier used in alerts ($row). Aggregation is optional.""" + base_parts: list = [measurement, field] + if aggregation: + base_parts.append(aggregation) + base_parts.append(level) + + cache_key: str = ":".join(base_parts) + + for tag in sorted(tags): + tag_value = row.get(tag) + # tags without a value are skipped: line protocol has no empty tag values + if tag_value is not None: + cache_key += f":{tag}={tag_value}" + + return cache_key + + +def generate_counter_key(row_identifier: str, op_sym: str, compare_value) -> str: + """ + Generate the cache key of the breach counter for one condition. + + Conditions that differ only by operator or threshold share a row identifier, so both + are part of the counter key to keep their counts independent. + """ + return f"{row_identifier}|{op_sym}|{compare_value!r}" + + +def record_breach( + influxdb3_local, cache_key: str, trigger_count: int +) -> tuple[bool, int]: + """ + Count one consecutive condition breach for the given cache key. + + The counter is reset as soon as the alert is due, so the next alert requires + another 'trigger_count' consecutive breaches. + + Args: + influxdb3_local: InfluxDB client instance. + cache_key (str): Key identifying the condition and row. + trigger_count (int): Number of consecutive breaches required to alert. + + Returns: + tuple[bool, int]: Whether an alert is due, and the current breach number. + """ + cached_value = influxdb3_local.cache.get(cache_key) + breach_number: int = (int(cached_value) if cached_value is not None else 0) + 1 + + if breach_number >= trigger_count: + influxdb3_local.cache.put(cache_key, "0") + return True, breach_number + + influxdb3_local.cache.put(cache_key, str(breach_number)) + return False, breach_number + + def process_writes(influxdb3_local, table_batches: list, args: dict): """ Process incoming data writes and trigger notifications if field conditions are met for a specified number of times. """ - task_id: str = str(uuid.uuid4()) - influxdb3_local.info(f"[{task_id}] Starting writes process with args: {args}") - - # Override args with config file if specified - if args: - if path := args.get("config_file_path", None): - if not path.endswith(".toml"): - influxdb3_local.error( - f"[{task_id}] Invalid config file format: expected a .toml file" - ) - return - try: - plugin_dir_var: str | None = os.getenv("PLUGIN_DIR", None) - if plugin_dir_var: - file_path = Path(plugin_dir_var) / path - else: - # Fallbacks for servers where the operator has not exported PLUGIN_DIR: - # - INFLUXDB3_PLUGIN_DIR: set when the server is configured via env var - # - VIRTUAL_ENV: exported by the processing engine; default venv is /.venv - candidates: list[str] = [] - if influxdb3_plugin_dir := os.environ.get("INFLUXDB3_PLUGIN_DIR"): - candidates.append(influxdb3_plugin_dir) - if virtual_env := os.environ.get("VIRTUAL_ENV"): - candidates.append(str(Path(virtual_env).parent)) - - resolved = None - for base in candidates: - candidate = Path(base) / path - if candidate.exists(): - resolved = candidate - break - - if resolved is None: - candidates_str = ", ".join(candidates) if candidates else "none available" - influxdb3_local.error( - f"[{task_id}] PLUGIN_DIR env var not set and config file path " - f"'{path}' was not found via fallbacks (tried: {candidates_str})" - ) - return - file_path = resolved - influxdb3_local.info(f"[{task_id}] Reading config file {file_path}") - with open(file_path, "rb") as f: - args = tomllib.load(f) - args["use_config_file"] = True - influxdb3_local.info(f"[{task_id}] New args content: {args}") - except Exception: - influxdb3_local.error(f"[{task_id}] Failed to read config file") - return - else: - args["use_config_file"] = False + if not table_batches: + return - if ( - not args - or "measurement" not in args - or "field_conditions" not in args - or "senders" not in args - ): - influxdb3_local.error( - f"[{task_id}] Missing required arguments: measurement, field_conditions, or senders" + task_id: str = str(uuid.uuid4()) + config: dict | None = influxdb3_local.cache.get(_WRITES_CONFIG_CACHE_KEY) + if config is None: + config = _load_config(influxdb3_local, args, _WRITES_VALIDATORS, task_id) + if config is None: + return + influxdb3_local.cache.put( + _WRITES_CONFIG_CACHE_KEY, config, _WRITES_CONFIG_TTL_SECONDS ) - return - measurement: str = args["measurement"] - all_measurements: list = get_all_measurements(influxdb3_local) + measurement: str = config["measurement"] + all_measurements: list = get_table_names(influxdb3_local) if measurement not in all_measurements: influxdb3_local.error( f"[{task_id}] Measurement '{measurement}' not found in database" ) return + # an 'all_tables' trigger also receives batches of other tables + monitored_batches: list = [ + table_batch + for table_batch in table_batches + if table_batch["table_name"] == measurement + ] + if not monitored_batches: + return + + influxdb3_local.info(f"[{task_id}] Starting writes process") + try: - trigger_count: int = int(args.get("trigger_count", 1)) - senders_config: dict = parse_senders(influxdb3_local, args, task_id) - field_conditions: list = parse_field_conditions(influxdb3_local, args, task_id) + trigger_count: int = config["trigger_count"] + senders_config: dict = parse_senders(influxdb3_local, config, task_id) + field_conditions: list = parse_field_conditions( + influxdb3_local, config, task_id + ) influxdb3_local.info(f"[{task_id}] Field conditions: {field_conditions}") - port_override: int = parse_port_override(args, task_id) - notification_path: str = args.get("notification_path", "notify") - influxdb3_auth_token: str = args.get("influxdb3_auth_token") or os.getenv( - "INFLUXDB3_AUTH_TOKEN" + port_override: int = config["port_override"] + notification_path: str = config["notification_path"] + influxdb3_auth_token: str = ( + config.get("influxdb3_auth_token") + or os.getenv("INFLUXDB3_AUTH_TOKEN") + or "" ) - if influxdb3_auth_token is None: + if not influxdb3_auth_token: influxdb3_local.error( f"[{task_id}] Missing required argument: influxdb3_auth_token" ) return - notification_tpl: str = args.get( - "notification_text", - "[$level] InfluxDB 3 alert triggered. Condition $field $op_sym $compare_val matched $trigger_count times($actual) — matched in row $row.", - ) + notification_tpl: str = config["notification_text"] - for table_batch in table_batches: - table_name: str = table_batch["table_name"] - if table_name != measurement: - continue - - tags: list = get_tag_names(influxdb3_local, table_name, task_id) + tags: list = get_measurement_tags(influxdb3_local, measurement, task_id) + for table_batch in monitored_batches: for row in table_batch["rows"]: - for field, compare_fn, compare_val, level in field_conditions: + for field, op_sym, compare_fn, compare_val, level in field_conditions: if field not in row: influxdb3_local.warn( f"[{task_id}] Field '{field}' not found in row: {row}" @@ -738,105 +841,88 @@ def process_writes(influxdb3_local, table_batches: list, args: dict): actual = row[field] cache_key: str = generate_cache_key( - table_name, field, level, row, tags + measurement, field, level, row, tags + ) + counter_key: str = generate_counter_key( + cache_key, op_sym, compare_val + ) + if not compare_fn(actual, compare_val): + influxdb3_local.cache.put(counter_key, "0") + continue + + alert_due, breach_number = record_breach( + influxdb3_local, counter_key, trigger_count ) - if compare_fn(actual, compare_val): - cache_value = influxdb3_local.cache.get(cache_key) - current_count = ( - int(cache_value) if cache_value is not None else 0 - ) - # reconstruct operator symbol from function - op_sym = next( - sym for sym, fn in _OP_FUNCS.items() if fn is compare_fn + if not alert_due: + influxdb3_local.warn( + f"[{task_id}] [{level}] Condition {field} {op_sym} {compare_val!r} matched in row {cache_key} ({actual!r}) for the {breach_number}/{trigger_count} time. Skipping alert." ) + continue - if current_count >= (trigger_count - 1): - notification_text = interpolate_notification_text( - notification_tpl, - { - "level": level, - "row": cache_key, - "field": field, - "op_sym": op_sym, - "compare_val": compare_val, - "trigger_count": trigger_count, - "actual": actual, - }, - ) - - payload: dict = { - "notification_text": notification_text, - "senders_config": senders_config, - } - - influxdb3_local.error( - f"[{task_id}] [{level}] Condition {field} {op_sym} {compare_val!r} matched in row {cache_key} {trigger_count} times ({actual!r}), sending alert" - ) - send_notification( - influxdb3_local, - port_override, - notification_path, - influxdb3_auth_token, - payload, - task_id, - ) - influxdb3_local.cache.put(cache_key, "0") - else: - influxdb3_local.warn( - f"[{task_id}] [{level}] Condition {field} {op_sym} {compare_val!r} matched in row {cache_key} ({actual!r}) for the {current_count + 1}/{trigger_count} time. Skipping alert." - ) - influxdb3_local.cache.put(cache_key, str(current_count + 1)) - - else: - influxdb3_local.cache.put(cache_key, "0") + notification_text = interpolate_notification_text( + notification_tpl, + { + "level": level, + "row": cache_key, + "field": field, + "op_sym": op_sym, + "compare_val": compare_val, + "trigger_count": trigger_count, + "actual": actual, + }, + ) + + payload: dict = { + "notification_text": notification_text, + "senders_config": senders_config, + } + + influxdb3_local.error( + f"[{task_id}] [{level}] Condition {field} {op_sym} {compare_val!r} matched in row {cache_key} {trigger_count} times ({actual!r}), sending alert" + ) + send_notification( + influxdb3_local, + port_override, + notification_path, + influxdb3_auth_token, + payload, + task_id, + ) except Exception as e: influxdb3_local.error(f"[{task_id}] Error: {str(e)}") -def parse_window(args: dict, task_id: str) -> timedelta: +def interval_literal(interval: timedelta) -> str: """ - Parses the 'window' argument from args and converts it into a timedelta object. - Represents the size of the query window. + Render a DATE_BIN interval literal for the aggregation interval. Args: - args (dict): Dictionary with the 'window' key (e.g., {"window": "2h"}). - task_id (str): Unique identifier for the current task, used for logging. + interval (timedelta): Aggregation interval. Returns: - timedelta: Parsed time delta for the window. + str: Interval literal, e.g. "600 seconds". Raises: - Exception: If the window is missing or has an invalid format or unit. - - Example input: - args = {"window": "3d"} # valid units: 's', 'min', 'h', 'd', 'w' + ValueError: If the interval is shorter than one second. """ - valid_units: dict = { - "s": "seconds", - "min": "minutes", - "h": "hours", - "d": "days", - "w": "weeks", - } - - window: str = args.get("window") - - match = re.fullmatch(r"(\d+)([a-zA-Z]+)", window) - if match: - number, unit = match.groups() - number = int(number) + seconds: int = int(interval.total_seconds()) + if seconds < 1: + raise ValueError( + f"Invalid interval: {seconds} seconds (must be at least 1 second)" + ) + return f"{seconds} seconds" - if number >= 1 and unit in valid_units: - return timedelta(**{valid_units[unit]: number}) - raise Exception(f"[{task_id}] Invalid interval format: {window}.") +def quote_identifier(identifier: str) -> str: + """Quote a SQL identifier, escaping embedded double quotes.""" + return '"' + str(identifier).replace('"', '""') + '"' def generate_fields_string( field_aggregation_values: dict, - interval: tuple, + interval: str, tags_list: list, ): """ @@ -844,28 +930,34 @@ def generate_fields_string( Args: field_aggregation_values: dict - interval (tuple[int, str]): Tuple of interval magnitude and unit (e.g., (10, 'minutes')). + interval (str): DATE_BIN interval literal (e.g., "600 seconds"). tags_list (list): List of tag names to include in the query. Returns: str: SQL SELECT clause string including DATE_BIN, aggregations and tags. """ - query: str = f"DATE_BIN(INTERVAL '{interval[0]} {interval[1]}', time, '1970-01-01T00:00:00Z') AS _time" + query: str = ( + f"DATE_BIN(INTERVAL '{interval}', time, '1970-01-01T00:00:00Z') AS _time" + ) for field_name, aggregation_value_list in field_aggregation_values.items(): - for aggregation, op_fn, value, level in aggregation_value_list: - if f'{aggregation}("{field_name}")' in query: + quoted_field: str = quote_identifier(field_name) + for aggregation, *_ in aggregation_value_list: + # Dedupe by alias: several conditions may share one aggregation, and + # duplicate projection names are rejected by the query planner. + alias: str = quote_identifier(f"{field_name}_{aggregation}") + if f"as {alias}" in query: continue query += ",\n" # Add ORDER BY time for first_value and last_value to ensure correct temporal ordering - if aggregation in ('first_value', 'last_value'): - query += f'\t{aggregation}("{field_name}" ORDER BY time) as "{field_name}_{aggregation}"' + if aggregation in ("first_value", "last_value"): + query += f"\t{aggregation}({quoted_field} ORDER BY time) as {alias}" else: - query += f'\t{aggregation}("{field_name}") as "{field_name}_{aggregation}"' + query += f"\t{aggregation}({quoted_field}) as {alias}" for tag in tags_list: - query += f',\n\t"{tag}"' + query += f",\n\t{quote_identifier(tag)}" return query @@ -882,7 +974,7 @@ def generate_group_by_string(tags_list: list): """ group_by_clause: str = "_time" for tag in tags_list: - group_by_clause += f', "{tag}"' + group_by_clause += f", {quote_identifier(tag)}" return group_by_clause @@ -890,7 +982,7 @@ def build_query( field_aggregation_values: dict, measurement: str, tags_list: list[str], - interval: tuple, + interval: str, start_time: datetime, end_time: datetime, ) -> str: @@ -901,7 +993,7 @@ def build_query( field_aggregation_values: dict for aggregation building measurement: source measurement name tags_list: list of tag keys to GROUP BY - interval: (magnitude, unit) for DATE_BIN + interval: DATE_BIN interval literal (e.g., "600 seconds") start_time: UTC datetime for WHERE time > ... end_time: UTC datetime for WHERE time < ... @@ -915,179 +1007,76 @@ def build_query( # GROUP BY clause group_by: str = generate_group_by_string(tags_list) - # ISO timestamps - start_iso: str = start_time.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") - end_iso: str = end_time.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + # ISO timestamps, microsecond precision so consecutive windows tile exactly + start_iso: str = start_time.astimezone(timezone.utc).strftime( + "%Y-%m-%dT%H:%M:%S.%fZ" + ) + end_iso: str = end_time.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ") query: str = f""" SELECT {fields_clause} FROM - '{measurement}' + {quote_identifier(measurement)} WHERE time >= '{start_iso}' - AND + AND time < '{end_iso}' GROUP BY {group_by} + ORDER BY + _time """ return query -def get_tag_names(influxdb3_local, measurement: str, task_id: str) -> list[str]: - """ - Retrieves the list of tag names for a measurement from cache or the database. - - Args: - influxdb3_local: InfluxDB client instance. - measurement (str): Name of the measurement to query. - task_id (str): The task ID. - - Returns: - list[str]: List of tag names with 'Dictionary(Int32, Utf8)' data type. - """ - # check cache first - tags: list = influxdb3_local.cache.get(f"{measurement}_tags") - if tags: - return tags - - # if not in cache, query the database - query: str = """ - SELECT column_name - FROM information_schema.columns - WHERE table_name = $measurement - AND data_type = 'Dictionary(Int32, Utf8)' - """ - res: list[dict] = influxdb3_local.query(query, {"measurement": measurement}) - - if not res: - influxdb3_local.info( - f"[{task_id}] No tags found for measurement '{measurement}'." - ) - return [] - - tag_names: list[str] = [tag["column_name"] for tag in res] - - # cache the result for 1 hour - influxdb3_local.cache.put(f"{measurement}_tags", tag_names, 60 * 60) - - return tag_names - - -def parse_time_interval(args: dict, task_id: str) -> tuple[int, str]: - """ - Parses the interval string into a tuple of magnitude and unit. - - Supports time units: seconds (s), minutes (min), hours (h), days (d). - - Args: - args (dict): Dictionary containing configuration parameters, including the 'interval' key - with a string in the format '' (e.g., '10min', '2s', '1h'). - task_id (str): The task ID. - - Returns: - tuple[int, str]: A tuple containing the magnitude (integer) and the unit (e.g., 'minutes' or 'days'). - For months, quarters, and years, the magnitude is the equivalent number of days, and the unit is 'days'. - - Raises: - Exception: If the interval format is invalid, the unit is not supported, or the magnitude is less than 1. - """ - unit_mapping: dict = {"s": "seconds", "min": "minutes", "h": "hours", "d": "days"} - valid_units = unit_mapping.keys() - - interval: str = args.get("interval", "5min") - - match = re.fullmatch(r"(\d+)([a-zA-Z]+)", interval) - if match: - number_part, unit = match.groups() - magnitude = int(number_part) - if unit in valid_units and magnitude >= 1: - return magnitude, unit_mapping[unit] - - raise Exception(f"[{task_id}] Invalid interval format: {interval}.") - - -def parse_field_aggregation_values( - influxdb3_local, args: dict, task_id: str -) -> dict[str, list] | None: - """ - Parses field aggregation values with comparison operators and message levels or use values from config file. +def _aggregations_from_mapping( + influxdb3_local, raw: dict, task_id: str +) -> dict[str, list]: + """Parse aggregation conditions given as {field: [[aggregation, op, value, level], ...]}.""" + result: dict[str, list] = {} - Args: - influxdb3_local: InfluxDB client instance. - args (dict): Contains the 'field_aggregation_values' key with space-separated strings, - e.g., 'field:avg@>=10-INFO field2:min@<5.0-WARN'. - task_id (str): Task identifier (used for logging/warnings). + for field, conditions in raw.items(): + try: + for aggregation, op, value, level in conditions: + if aggregation not in AVAILABLE_AGGREGATIONS: + influxdb3_local.warn( + f"[{task_id}] Unsupported aggregation '{aggregation}', skipping..." + ) + continue + message_level: str = str(level).strip().upper() + if message_level not in ALLOWED_MESSAGE_LEVELS: + influxdb3_local.warn( + f"[{task_id}] Invalid message level '{level}', skipping..." + ) + continue + if op not in _OP_FUNCS: + influxdb3_local.warn( + f"[{task_id}] Invalid operator '{op}', skipping..." + ) + continue + entry: list = [aggregation, op, _OP_FUNCS[op], value, message_level] + result.setdefault(field, []).append(entry) + except Exception as e: + influxdb3_local.warn( + f"[{task_id}] Error parsing field aggregation values for field '{field}': {e}" + ) + continue - Returns: - dict[str, list[list[str, callable, float, str]]]: Dictionary mapping field names to a list of lists: - [aggregation, comparison_operator_fn, threshold_value, message_level]. + return result - Raises: - Exception: If no valid entries are found. - """ - available_aggregations: tuple = ( - "avg", - "count", - "sum", - "min", - "max", - "median", - "stddev", - "first_value", - "last_value", - "var", - "approx_median", - ) - allowed_operators: tuple = (">", "<", ">=", "<=", "==", "!=") - allowed_message_levels: tuple = ("INFO", "WARN", "ERROR", "CRITICAL") - raw_input: str | None = args.get("field_aggregation_values") - if raw_input is None: - return {} +def _aggregations_from_string( + influxdb3_local, raw: str, task_id: str +) -> dict[str, list]: + """Parse aggregation conditions given as 'field:aggregation@-' pairs.""" result: dict[str, list] = {} - if args["use_config_file"]: - if not isinstance(raw_input, dict): - raise Exception( - f"[{task_id}] field_aggregation_values must be a dictionary when using config file" - ) - for field, conditions in raw_input.items(): - try: - for aggregation, op, value, level in conditions: - if aggregation not in available_aggregations: - influxdb3_local.warn( - f"[{task_id}] Unsupported aggregation '{aggregation}', skipping..." - ) - continue - if level not in allowed_message_levels: - influxdb3_local.warn( - f"[{task_id}] Invalid message level '{level}', skipping..." - ) - continue - if op not in allowed_operators: - influxdb3_local.warn( - f"[{task_id}] Invalid operator '{op}', skipping..." - ) - continue - entry: list = [aggregation, _OP_FUNCS[op], value, level] - result.setdefault(field, []).append(entry) - except Exception as e: - influxdb3_local.warn( - f"[{task_id}] Error parsing field aggregation values for field '{field}': {e}" - ) - continue - - if not result: - raise Exception(f"[{task_id}] No valid field aggregation values provided.") - return result - # Strip quotes around the string if present - if raw_input[0] == raw_input[-1] and raw_input[0] in ('"', "'"): - raw_input = raw_input[1:-1] + if len(raw) > 1 and raw[0] == raw[-1] and raw[0] in ('"', "'"): + raw = raw[1:-1] - pairs = raw_input.split(" ") - for pair in pairs: + for pair in raw.split(" "): if not pair or ":" not in pair: influxdb3_local.warn( f"[{task_id}] Invalid format in pair '{pair}', skipping..." @@ -1103,21 +1092,25 @@ def parse_field_aggregation_values( aggregation, value_expr = agg_expr.split("@", 1) aggregation = aggregation.strip() - if aggregation not in available_aggregations: + if aggregation not in AVAILABLE_AGGREGATIONS: influxdb3_local.warn( f"[{task_id}] Unsupported aggregation '{aggregation}', skipping..." ) continue # Strip quotes around the value expression if present - if value_expr[0] == value_expr[-1] and value_expr[0] in ('"', "'"): + if ( + len(value_expr) > 1 + and value_expr[0] == value_expr[-1] + and value_expr[0] in ('"', "'") + ): value_expr = value_expr[1:-1] # Extract comparison operator matched_op = next( ( op - for op in sorted(allowed_operators, key=len, reverse=True) + for op in sorted(_OP_FUNCS, key=len, reverse=True) if value_expr.startswith(op) ), None, @@ -1139,7 +1132,7 @@ def parse_field_aggregation_values( ) continue - if level not in allowed_message_levels: + if level not in ALLOWED_MESSAGE_LEVELS: influxdb3_local.warn( f"[{task_id}] Invalid message level '{level}', skipping..." ) @@ -1153,36 +1146,54 @@ def parse_field_aggregation_values( ) continue - entry: list = [aggregation, _OP_FUNCS[matched_op], value, level] + entry: list = [aggregation, matched_op, _OP_FUNCS[matched_op], value, level] result.setdefault(field_name.strip(), []).append(entry) - if not result: - raise Exception(f"[{task_id}] No valid field aggregation values provided.") - return result -def generate_cache_key( - measurement: str, - field: str, - level: str, - row: dict, - tags: list, - aggregation: str | None = None, -) -> str: - """Generate cache key based on input parameters. Aggregation is optional.""" - base_parts: list = [measurement, field] - if aggregation: - base_parts.append(aggregation) - base_parts.append(level) +def parse_field_aggregation_values( + influxdb3_local, config: dict, task_id: str +) -> dict[str, list]: + """ + Parse the aggregation conditions used by the scheduled trigger. - cache_key: str = ":".join(base_parts) + Conditions come either as {field: [[aggregation, operator, value, level], ...]} + (TOML) or as a string of 'field:aggregation@-' pairs separated + by spaces, e.g. 'field:avg@>=10-INFO field2:min@<5.0-WARN'. - for tag in sorted(tags): - tag_value = row.get(tag, "None") - cache_key += f":{tag}={tag_value}" + Args: + influxdb3_local: InfluxDB client instance. + config (dict): Loaded config, optionally containing "field_aggregation_values". + task_id (str): Unique task identifier. - return cache_key + Returns: + dict[str, list]: Field name mapped to a list of + [aggregation, operator, operator_fn, threshold_value, message_level]. + + Raises: + Exception: If the value has an unsupported type, or if it is provided but no + valid entries are found. + """ + raw: str | dict | None = config.get("field_aggregation_values") + if raw is None: + return {} + + if isinstance(raw, dict): + result = _aggregations_from_mapping(influxdb3_local, raw, task_id) + elif isinstance(raw, str): + if not raw.strip(): + return {} + result = _aggregations_from_string(influxdb3_local, raw, task_id) + else: + raise Exception( + "'field_aggregation_values' must be a mapping or a string, " + f"got {type(raw).__name__}" + ) + + if not result: + raise Exception("No valid field aggregation values provided.") + return result def process_scheduled_call(influxdb3_local, call_time: datetime, args: dict): @@ -1197,70 +1208,18 @@ def process_scheduled_call(influxdb3_local, call_time: datetime, args: dict): "measurement", "senders", "influxdb3_auth_token", "window", and other alert settings. """ task_id: str = str(uuid.uuid4()) - influxdb3_local.info(f"[{task_id}] Starting scheduled call with args: {args} and call_time: {call_time}") + influxdb3_local.info( + f"[{task_id}] Starting scheduled call with call_time: {call_time}" + ) - # Override args with config file if specified - if args: - if path := args.get("config_file_path", None): - if not path.endswith(".toml"): - influxdb3_local.error( - f"[{task_id}] Invalid config file format: expected a .toml file" - ) - return - try: - plugin_dir_var: str | None = os.getenv("PLUGIN_DIR", None) - if plugin_dir_var: - file_path = Path(plugin_dir_var) / path - else: - # Fallbacks for servers where the operator has not exported PLUGIN_DIR: - # - INFLUXDB3_PLUGIN_DIR: set when the server is configured via env var - # - VIRTUAL_ENV: exported by the processing engine; default venv is /.venv - candidates: list[str] = [] - if influxdb3_plugin_dir := os.environ.get("INFLUXDB3_PLUGIN_DIR"): - candidates.append(influxdb3_plugin_dir) - if virtual_env := os.environ.get("VIRTUAL_ENV"): - candidates.append(str(Path(virtual_env).parent)) - - resolved = None - for base in candidates: - candidate = Path(base) / path - if candidate.exists(): - resolved = candidate - break - - if resolved is None: - candidates_str = ", ".join(candidates) if candidates else "none available" - influxdb3_local.error( - f"[{task_id}] PLUGIN_DIR env var not set and config file path " - f"'{path}' was not found via fallbacks (tried: {candidates_str})" - ) - return - file_path = resolved - influxdb3_local.info(f"[{task_id}] Reading config file {file_path}") - with open(file_path, "rb") as f: - args = tomllib.load(f) - args["use_config_file"] = True - influxdb3_local.info(f"[{task_id}] New args content: {args}") - except Exception: - influxdb3_local.error(f"[{task_id}] Failed to read config file") - return - else: - args["use_config_file"] = False - - # Configuration - if ( - not args - or "measurement" not in args - or "senders" not in args - or "window" not in args - ): - influxdb3_local.error( - f"[{task_id}] Missing required arguments: measurement, senders, or window" - ) + config: dict | None = _load_config( + influxdb3_local, args, _SCHEDULED_VALIDATORS, task_id + ) + if config is None: return - measurement: str = args["measurement"] - all_measurements: list = get_all_measurements(influxdb3_local) + measurement: str = config["measurement"] + all_measurements: list = get_table_names(influxdb3_local) if measurement not in all_measurements: influxdb3_local.error( f"[{task_id}] Measurement '{measurement}' not found in database" @@ -1268,55 +1227,56 @@ def process_scheduled_call(influxdb3_local, call_time: datetime, args: dict): return try: - trigger_count: int = int(args.get("trigger_count", 1)) - senders_config: dict = parse_senders(influxdb3_local, args, task_id) + trigger_count: int = config["trigger_count"] + senders_config: dict = parse_senders(influxdb3_local, config, task_id) field_aggregation_values: dict = parse_field_aggregation_values( - influxdb3_local, args, task_id + influxdb3_local, config, task_id + ) + influxdb3_local.info( + f"[{task_id}] Field aggregation conditions: {field_aggregation_values}" ) - influxdb3_local.info(f"[{task_id}] Field aggregation conditions: {field_aggregation_values}") - deadman_check: bool = True if args.get("deadman_check") else False + deadman_check: bool = config["deadman_check"] if not field_aggregation_values and not deadman_check: influxdb3_local.error( "For the plugin to work, you must provide a valid field_aggregation_values parameter or set deadman_check to True" ) return - port_override: int = parse_port_override(args, task_id) - notification_path: str = args.get("notification_path", "notify") - influxdb3_auth_token: str = args.get("influxdb3_auth_token") or os.getenv( - "INFLUXDB3_AUTH_TOKEN" + port_override: int = config["port_override"] + notification_path: str = config["notification_path"] + influxdb3_auth_token: str = ( + config.get("influxdb3_auth_token") + or os.getenv("INFLUXDB3_AUTH_TOKEN") + or "" ) - if influxdb3_auth_token is None: + if not influxdb3_auth_token: influxdb3_local.error( f"[{task_id}] Missing required environment variable: INFLUXDB3_AUTH_TOKEN" ) return - notification_tpl_deadman: str = args.get( - "notification_deadman_text", - "Deadman Alert: No data received from $table from $time_from to $time_to.", - ) - notification_tpl_threshold: str = args.get( - "notification_threshold_text", - "[$level] Threshold Alert on table $table: $aggregation of $field $op_sym $compare_val (actual: $actual) — matched in row $row.", - ) + notification_tpl_deadman: str = config["notification_deadman_text"] + notification_tpl_threshold: str = config["notification_threshold_text"] - tags: list = get_tag_names(influxdb3_local, measurement, task_id) - window: timedelta = parse_window(args, task_id) - interval: tuple = parse_time_interval(args, task_id) - time_to = call_time.astimezone(timezone.utc) + tags: list = get_measurement_tags(influxdb3_local, measurement, task_id) + window: timedelta = config["window"] + interval: str = interval_literal(config["interval"]) + time_to: datetime = call_time.replace(tzinfo=timezone.utc) time_from: datetime = time_to - window - influxdb3_local.info(f"[{task_id}] Querying data in '{measurement}' from {time_from} to {time_to}") + influxdb3_local.info( + f"[{task_id}] Querying data in '{measurement}' from {time_from} to {time_to}" + ) query: str = build_query( field_aggregation_values, measurement, tags, interval, time_from, time_to ) results: list = influxdb3_local.query(query) if not results and deadman_check: - cache_value: str | None = influxdb3_local.cache.get(measurement) - current_count = int(cache_value) if cache_value is not None else 0 + alert_due, breach_number = record_breach( + influxdb3_local, measurement, trigger_count + ) - if current_count >= (trigger_count - 1): + if alert_due: influxdb3_local.error( f"[{task_id}] No data found in '{measurement}' from {time_from} to {time_to} for {trigger_count} times. Sending alert." ) @@ -1339,24 +1299,26 @@ def process_scheduled_call(influxdb3_local, call_time: datetime, args: dict): payload, task_id, ) - influxdb3_local.cache.put(measurement, "0") else: influxdb3_local.warn( - f"[{task_id}] No data found in '{measurement}' from {time_from} to {time_to} for {current_count + 1}/{trigger_count} times. Skipping alert." + f"[{task_id}] No data found in '{measurement}' from {time_from} to {time_to} for {breach_number}/{trigger_count} times. Skipping alert." ) - influxdb3_local.cache.put(measurement, str(current_count + 1)) else: influxdb3_local.cache.put(measurement, "0") - influxdb3_local.info(f"[{task_id}] Query executed, {len(results)} records returned") + influxdb3_local.info( + f"[{task_id}] Query executed, {len(results)} records returned" + ) for row in results: for field, aggregation_values in field_aggregation_values.items(): - for aggregation, compare_fn, compare_value, level in aggregation_values: - cache_key: str = generate_cache_key( - measurement, field, level, row, tags, aggregation - ) - + for ( + aggregation, + op_sym, + compare_fn, + compare_value, + level, + ) in aggregation_values: if f"{field}_{aggregation}" not in row: influxdb3_local.warn( f"[{task_id}] Field '{field}_{aggregation}' not found in results received" @@ -1364,56 +1326,56 @@ def process_scheduled_call(influxdb3_local, call_time: datetime, args: dict): continue actual = row[f"{field}_{aggregation}"] - if compare_fn(actual, compare_value): - cache_value: str | None = influxdb3_local.cache.get(cache_key) - current_count = ( - int(cache_value) if cache_value is not None else 0 - ) + cache_key: str = generate_cache_key( + measurement, field, level, row, tags, aggregation + ) + counter_key: str = generate_counter_key( + cache_key, op_sym, compare_value + ) + if not compare_fn(actual, compare_value): + influxdb3_local.cache.put(counter_key, "0") + continue - # reconstruct operator symbol from function - op_sym = next( - sym for sym, fn in _OP_FUNCS.items() if fn is compare_fn + alert_due, breach_number = record_breach( + influxdb3_local, counter_key, trigger_count + ) + + if not alert_due: + influxdb3_local.warn( + f"[{task_id}] Condition for row {cache_key} ({aggregation}({field}) {op_sym} {compare_value!r}) matched ({actual!r}) for the {breach_number}/{trigger_count} time. Skipping alert." ) + continue + + notification_text = interpolate_notification_text( + notification_tpl_threshold, + { + "level": level, + "field": field, + "table": measurement, + "row": cache_key, + "op_sym": op_sym, + "aggregation": aggregation, + "compare_val": compare_value, + "actual": actual, + }, + ) + + payload: dict = { + "notification_text": notification_text, + "senders_config": senders_config, + } - if current_count >= (trigger_count - 1): - notification_text = interpolate_notification_text( - notification_tpl_threshold, - { - "level": level, - "field": field, - "table": measurement, - "row": cache_key, - "op_sym": op_sym, - "aggregation": aggregation, - "compare_val": compare_value, - "actual": actual, - }, - ) - - payload: dict = { - "notification_text": notification_text, - "senders_config": senders_config, - } - - influxdb3_local.error( - f"[{task_id}] Condition on {measurement}: {aggregation}({field}) {op_sym} {compare_value!r} matched {trigger_count} times in row {cache_key} ({actual!r}), sending alert" - ) - send_notification( - influxdb3_local, - port_override, - notification_path, - influxdb3_auth_token, - payload, - task_id, - ) - influxdb3_local.cache.put(cache_key, "0") - else: - influxdb3_local.warn( - f"[{task_id}] Condition for row {cache_key} ({aggregation}({field}) {op_sym} {compare_value!r}) matched ({actual!r}) for the {current_count + 1}/{trigger_count} time. Skipping alert." - ) - influxdb3_local.cache.put(cache_key, str(current_count + 1)) - else: - influxdb3_local.cache.put(cache_key, "0") + influxdb3_local.error( + f"[{task_id}] Condition on {measurement}: {aggregation}({field}) {op_sym} {compare_value!r} matched {trigger_count} times in row {cache_key} ({actual!r}), sending alert" + ) + send_notification( + influxdb3_local, + port_override, + notification_path, + influxdb3_auth_token, + payload, + task_id, + ) except Exception as e: influxdb3_local.error(f"[{task_id}] Error: {str(e)}") diff --git a/influxdata/threshold_deadman_checks/threshold_deadman_config_data_writes.toml b/influxdata/threshold_deadman_checks/threshold_deadman_config_data_writes.toml index 37f417f..c2f5e40 100644 --- a/influxdata/threshold_deadman_checks/threshold_deadman_config_data_writes.toml +++ b/influxdata/threshold_deadman_checks/threshold_deadman_config_data_writes.toml @@ -64,22 +64,17 @@ field_conditions = [["field1", ">", 0.0, "your_level"]] # e.g., [["temp", ">", # Optional: Custom HTTP headers (Base64-encoded JSON string) #http_headers = "your_http_headers" # e.g., "eyJhdXRoIjogIkJlYXJlciBZT1VSX1RPS0VOIn0=" -# --- SMS (Twilio) --- -# Twilio Account SID (required for SMS, or via TWILIO_SID env var) +# --- SMS and WhatsApp (Twilio) --- +# Both channels use the same four parameters; set them once. +# Twilio Account SID (required, or via TWILIO_SID env var) #twilio_sid = "your_twilio_sid" # e.g., "ACXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX" -# Twilio Auth Token (required for SMS, or via TWILIO_TOKEN env var) +# Twilio Auth Token (required, or via TWILIO_TOKEN env var) #twilio_token = "your_twilio_token" # e.g., "your_auth_token" -# Twilio sender number (required for SMS, format: +1234567890) +# Sender number (required, format: +1234567890). For WhatsApp use the approved WhatsApp sender #twilio_from_number = "your_twilio_from_number" # e.g., "+1234567890" -# Recipient phone number (required for SMS, format: +0987654321) +# Recipient number (required, format: +0987654321) #twilio_to_number = "your_twilio_to_number" # e.g., "+0987654321" -# --- WhatsApp (via Twilio) --- -# WhatsApp sender number (required for WhatsApp, format: +1234567890) -#whatsapp_from_number = "your_whatsapp_from_number" # e.g., "+1234567890" -# WhatsApp recipient number (required for WhatsApp, format: +0987654321) -#whatsapp_to_number = "your_whatsapp_to_number" # e.g., "+0987654321" - ###### Example: Create Trigger Using This Config ###### # influxdb3 create trigger \ # --database your_database_name \ diff --git a/influxdata/threshold_deadman_checks/threshold_deadman_config_scheduler.toml b/influxdata/threshold_deadman_checks/threshold_deadman_config_scheduler.toml index 5fa6512..d83967c 100644 --- a/influxdata/threshold_deadman_checks/threshold_deadman_config_scheduler.toml +++ b/influxdata/threshold_deadman_checks/threshold_deadman_config_scheduler.toml @@ -58,22 +58,17 @@ # Optional: Custom HTTP headers (Base64-encoded JSON string) #http_headers = "your_http_headers" # e.g., "eyJhdXRoIjogIkJlYXJlciBZT1VSX1RPS0VOIn0=" -# --- SMS (Twilio) --- -# Twilio Account SID (required for SMS, or via TWILIO_SID env var) +# --- SMS and WhatsApp (Twilio) --- +# Both channels use the same four parameters; set them once. +# Twilio Account SID (required, or via TWILIO_SID env var) #twilio_sid = "your_twilio_sid" # e.g., "ACXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX" -# Twilio Auth Token (required for SMS, or via TWILIO_TOKEN env var) +# Twilio Auth Token (required, or via TWILIO_TOKEN env var) #twilio_token = "your_twilio_token" # e.g., "your_auth_token" -# Twilio sender number (required for SMS, format: +1234567890) +# Sender number (required, format: +1234567890). For WhatsApp use the approved WhatsApp sender #twilio_from_number = "your_twilio_from_number" # e.g., "+1234567890" -# Recipient phone number (required for SMS, format: +0987654321) +# Recipient number (required, format: +0987654321) #twilio_to_number = "your_twilio_to_number" # e.g., "+0987654321" -# --- WhatsApp (via Twilio) --- -# WhatsApp sender number (required for WhatsApp, format: +1234567890) -#whatsapp_from_number = "your_whatsapp_from_number" # e.g., "+1234567890" -# WhatsApp recipient number (required for WhatsApp, format: +0987654321) -#whatsapp_to_number = "your_whatsapp_to_number" # e.g., "+0987654321" - ########## Required Parameters ########## # InfluxDB measurement (table) to monitor for deadman and threshold checks @@ -86,7 +81,7 @@ senders = ["your_channel"] # e.g., ["slack"], ["slack", "http"] # Time window to check for data and threshold conditions # Format: , where unit is s (seconds), min (minutes), h (hours), d (days), w (weeks) -window = "your_window" # e.g., "10m", "1h" +window = "your_window" # e.g., "10min", "1h" # Aggregation-based threshold conditions (Required if deadman check is disabled) # Format: {field = [[aggregation, operator, value, level], ...], ...} From eacf16fe4404375445d11757a3fbf87f65b5f493 Mon Sep 17 00:00:00 2001 From: Aliaksei Kharlap Date: Wed, 5 Aug 2026 22:13:27 +0300 Subject: [PATCH 3/4] add tests --- influxdata/threshold_deadman_checks/README.md | 3 + .../requirements-dev.txt | 3 + .../test_threshold_deadman_checks.py | 705 ++++++++++++++++++ 3 files changed, 711 insertions(+) create mode 100644 influxdata/threshold_deadman_checks/requirements-dev.txt create mode 100644 influxdata/threshold_deadman_checks/test_threshold_deadman_checks.py diff --git a/influxdata/threshold_deadman_checks/README.md b/influxdata/threshold_deadman_checks/README.md index e2e3be2..d3b9935 100644 --- a/influxdata/threshold_deadman_checks/README.md +++ b/influxdata/threshold_deadman_checks/README.md @@ -246,6 +246,9 @@ influxdb3 create trigger \ - `threshold_deadman_checks_plugin.py`: The main plugin code containing handlers for scheduled and data write triggers - `threshold_deadman_config_scheduler.toml`: Example TOML configuration for scheduled triggers - `threshold_deadman_config_data_writes.toml`: Example TOML configuration for data write triggers +- `test_threshold_deadman_checks.py`: Pytest suite (59 tests, runs without a live InfluxDB 3 server) +- `requirements.txt`: Runtime dependencies (`influxdata-plugin-utils>=0.3.0`, `requests`) +- `requirements-dev.txt`: Development dependencies (`pytest`) ### Logging diff --git a/influxdata/threshold_deadman_checks/requirements-dev.txt b/influxdata/threshold_deadman_checks/requirements-dev.txt new file mode 100644 index 0000000..cbc3288 --- /dev/null +++ b/influxdata/threshold_deadman_checks/requirements-dev.txt @@ -0,0 +1,3 @@ +pytest +influxdata-plugin-utils>=0.3.0 +requests diff --git a/influxdata/threshold_deadman_checks/test_threshold_deadman_checks.py b/influxdata/threshold_deadman_checks/test_threshold_deadman_checks.py new file mode 100644 index 0000000..d85c533 --- /dev/null +++ b/influxdata/threshold_deadman_checks/test_threshold_deadman_checks.py @@ -0,0 +1,705 @@ +import json +import os +from datetime import datetime, timedelta, timezone + +import pytest + +import threshold_deadman_checks_plugin as plugin + + +class FakeCache: + def __init__(self): + self.store = {} + self.ttls = {} + + def get(self, key, default=None, use_global=None): + return self.store.get(key, default) + + def put(self, key, value, ttl=None, use_global=None): + self.store[key] = value + self.ttls[key] = ttl + + def delete(self, key, use_global=None): + return self.store.pop(key, None) is not None + + +class FakeInfluxdb3Local: + """Stub of the runtime client: logging, trigger-local cache and queries.""" + + def __init__(self, tables=("cpu",), tags=("host",), rows=None): + self.cache = FakeCache() + self.logs = [] + self.tables = list(tables) + self.tags = list(tags) + self.rows = rows or [] + self.queries = [] + + def info(self, message): + self.logs.append(("info", message)) + + def warn(self, message): + self.logs.append(("warn", message)) + + def error(self, message): + self.logs.append(("error", message)) + + def query(self, query, params=None): + self.queries.append(query) + if "SHOW TABLES" in query: + return [{"table_name": t, "table_type": "BASE TABLE"} for t in self.tables] + if "information_schema" in query: + return [{"column_name": tag} for tag in self.tags] + return self.rows + + def messages(self, level=None): + return [m for lvl, m in self.logs if level is None or lvl == level] + + +class FakeResponse: + def __init__(self, status_code=200): + self.status_code = status_code + + def raise_for_status(self): + if self.status_code >= 400: + raise plugin.requests.HTTPError(f"{self.status_code} Server Error") + + def json(self): + return {"results": "recorded"} + + +@pytest.fixture +def sent(monkeypatch): + """Collect notification payloads instead of posting them.""" + posts = [] + + def fake_post(url, headers=None, data=None, timeout=None): + posts.append({"url": url, "headers": headers, "payload": json.loads(data)}) + return FakeResponse() + + monkeypatch.setattr(plugin.requests, "post", fake_post) + monkeypatch.setattr(plugin.time, "sleep", lambda seconds: None) + return posts + + +@pytest.fixture +def host_timezone(): + """Switch the process timezone and restore it afterwards.""" + original = os.environ.get("TZ") + + def use(name): + os.environ["TZ"] = name + plugin.time.tzset() + + yield use + if original is None: + os.environ.pop("TZ", None) + else: + os.environ["TZ"] = original + plugin.time.tzset() + + +@pytest.fixture +def plugin_dir(monkeypatch, tmp_path): + monkeypatch.setenv("PLUGIN_DIR", str(tmp_path)) + monkeypatch.delenv("INFLUXDB3_AUTH_TOKEN", raising=False) + return tmp_path + + +WRITES_ARGS = { + "measurement": "cpu", + "field_conditions": "temp>30-WARN", + "senders": "http", + "http_webhook_url": "https://example.com/hook", + "influxdb3_auth_token": "tok", +} +SCHEDULED_ARGS = { + "measurement": "cpu", + "senders": "http", + "http_webhook_url": "https://example.com/hook", + "influxdb3_auth_token": "tok", + "window": "10min", + "interval": "1min", +} + + +def batch(rows, table="cpu"): + return [{"table_name": table, "rows": rows}] + + +# --- parsing ---------------------------------------------------------------- + + +@pytest.mark.parametrize( + "condition, actual, matches", + [ + ("temp>30-WARN", 40.0, True), + ("temp<30-WARN", 40.0, False), + ("temp>=40-WARN", 40.0, True), + ("temp<=40-WARN", 40.0, True), + ("status=='ok'-INFO", "ok", True), + ("status!='ok'-INFO", "ok", False), + ], +) +def test_conditions_from_string_operators(condition, actual, matches): + field, op_sym, compare_fn, value, level = plugin.parse_field_conditions( + FakeInfluxdb3Local(), {"field_conditions": condition}, "tid" + )[0] + + assert compare_fn(actual, value) is matches + assert condition.startswith(field) and op_sym in condition + + +def test_conditions_from_entries_normalizes_level_and_keeps_symbol(): + conditions = plugin.parse_field_conditions( + FakeInfluxdb3Local(), + {"field_conditions": [["temp", ">", 30.0, "warn"]]}, + "tid", + ) + + assert conditions == [("temp", ">", plugin.operator.gt, 30.0, "WARN")] + + +def test_conditions_from_entries_skips_malformed_and_keeps_valid(): + client = FakeInfluxdb3Local() + + conditions = plugin.parse_field_conditions( + client, + {"field_conditions": [["temp", ">", 30.0, "WARN"], ["cpu", ">"], "junk"]}, + "tid", + ) + + assert [c[0] for c in conditions] == ["temp"] + assert len([m for m in client.messages("warn") if "Invalid condition" in m]) == 2 + + +def test_conditions_reject_unsupported_type(): + with pytest.raises(Exception, match="must be a list of entries or a string"): + plugin.parse_field_conditions( + FakeInfluxdb3Local(), {"field_conditions": 42}, "tid" + ) + + +def test_conditions_reject_when_nothing_valid(): + with pytest.raises(Exception, match="No valid field conditions"): + plugin.parse_field_conditions( + FakeInfluxdb3Local(), {"field_conditions": "temp>30-NOSUCH"}, "tid" + ) + + +def test_aggregations_from_string(): + parsed = plugin.parse_field_aggregation_values( + FakeInfluxdb3Local(), + {"field_aggregation_values": "temp:avg@>30-ERROR temp:max@<5.0-info"}, + "tid", + ) + + assert parsed == { + "temp": [ + ["avg", ">", plugin.operator.gt, 30.0, "ERROR"], + ["max", "<", plugin.operator.lt, 5.0, "INFO"], + ] + } + + +def test_aggregations_from_mapping_normalizes_level(): + parsed = plugin.parse_field_aggregation_values( + FakeInfluxdb3Local(), + {"field_aggregation_values": {"temp": [["max", ">", 30.0, "error"]]}}, + "tid", + ) + + assert parsed == {"temp": [["max", ">", plugin.operator.gt, 30.0, "ERROR"]]} + + +@pytest.mark.parametrize("raw", [None, "", " "]) +def test_aggregations_absent_or_blank_is_empty(raw): + config = {} if raw is None else {"field_aggregation_values": raw} + + assert ( + plugin.parse_field_aggregation_values(FakeInfluxdb3Local(), config, "tid") == {} + ) + + +def test_aggregations_reject_unsupported_type(): + with pytest.raises(Exception, match="must be a mapping or a string"): + plugin.parse_field_aggregation_values( + FakeInfluxdb3Local(), + {"field_aggregation_values": [["temp", "avg", ">", 30, "ERROR"]]}, + "tid", + ) + + +@pytest.mark.parametrize( + "condition, expected", + [ + ("temp>30-WARN", 30), + ("temp>30.5-WARN", 30.5), + ("flag==true-WARN", True), + ("status=='ok'-WARN", "ok"), + ('status=="ok"-WARN', "ok"), + ], +) +def test_conditions_coerce_value_types(condition, expected): + value = plugin.parse_field_conditions( + FakeInfluxdb3Local(), {"field_conditions": condition}, "tid" + )[0][3] + + assert value == expected and isinstance(value, type(expected)) + + +def test_senders_collects_channel_arguments(): + senders = plugin.parse_senders( + FakeInfluxdb3Local(), + { + "senders": "http.whatsapp", + "http_webhook_url": "https://example.com/hook", + "twilio_sid": "ACdummy", + "twilio_token": "dummy", + "twilio_from_number": "+1234567890", + "twilio_to_number": "+0987654321", + }, + "tid", + ) + + assert sorted(senders) == ["http", "whatsapp"] + assert sorted(senders["whatsapp"]) == [ + "twilio_from_number", + "twilio_sid", + "twilio_to_number", + "twilio_token", + ] + + +def test_senders_drops_channel_without_required_argument(): + client = FakeInfluxdb3Local() + + senders = plugin.parse_senders( + client, + { + "senders": "slack.discord", + "slack_webhook_url": "https://hooks.slack.com/services/TEST", + }, + "tid", + ) + + assert list(senders) == ["slack"] + assert any("discord_webhook_url" in m for m in client.messages("warn")) + + +def test_senders_reject_when_nothing_valid(): + with pytest.raises(Exception, match="No valid senders configured"): + plugin.parse_senders(FakeInfluxdb3Local(), {"senders": "discord"}, "tid") + + +@pytest.mark.parametrize("raw, expected", [("10min", 600), ("2h", 7200), ("30s", 30)]) +def test_parse_window_accepts_positive(raw, expected): + assert plugin.parse_window(raw) == timedelta(seconds=expected) + + +@pytest.mark.parametrize("raw", ["0min", "0s"]) +def test_parse_window_rejects_non_positive(raw): + with pytest.raises(ValueError, match="must be a positive duration"): + plugin.parse_window(raw) + + +# --- keys and counters ------------------------------------------------------ + + +def test_row_identifier_includes_aggregation_and_sorted_tags(): + row = {"host": "a", "region": "eu"} + + assert ( + plugin.generate_cache_key("cpu", "temp", "WARN", row, ["region", "host"], "avg") + == "cpu:temp:avg:WARN:host=a:region=eu" + ) + + +def test_row_identifier_skips_tag_without_value(): + row = {"host": None} + + assert ( + plugin.generate_cache_key("cpu", "temp", "WARN", row, ["host"]) + == "cpu:temp:WARN" + ) + + +def test_counter_key_separates_operator_and_threshold(): + row_id = "cpu:temp:WARN:host=a" + + keys = { + plugin.generate_counter_key(row_id, ">", 30.0), + plugin.generate_counter_key(row_id, ">", 20.0), + plugin.generate_counter_key(row_id, ">=", 30.0), + } + + assert len(keys) == 3 + assert all(key.startswith(row_id) for key in keys) + + +def test_record_breach_accumulates_then_alerts_and_resets(): + client = FakeInfluxdb3Local() + + assert plugin.record_breach(client, "k", 3) == (False, 1) + assert plugin.record_breach(client, "k", 3) == (False, 2) + assert plugin.record_breach(client, "k", 3) == (True, 3) + assert client.cache.get("k") == "0" + + +def test_interpolate_notification_text_fills_all_variables(): + text = plugin.interpolate_notification_text( + "[$level] $table $aggregation($field) $op_sym $compare_val actual=$actual " + "count=$trigger_count row=$row", + { + "level": "WARN", + "table": "cpu", + "aggregation": "avg", + "field": "temp", + "op_sym": ">", + "compare_val": 30.0, + "actual": 40.0, + "trigger_count": 2, + "row": "cpu:temp:avg:WARN:host=a", + }, + ) + + assert text == ( + "[WARN] cpu avg(temp) > 30.0 actual=40.0 count=2 row=cpu:temp:avg:WARN:host=a" + ) + + +def test_interpolate_notification_text_keeps_unknown_variables(): + assert plugin.interpolate_notification_text( + "$field $missing", {"field": "temp"} + ) == ("temp $missing") + + +# --- SQL generation --------------------------------------------------------- + + +@pytest.mark.parametrize( + "identifier, expected", + [("temp", '"temp"'), ('te"mp', '"te""mp"'), ("ho st", '"ho st"')], +) +def test_quote_identifier(identifier, expected): + assert plugin.quote_identifier(identifier) == expected + + +def test_interval_literal_rejects_sub_second(): + with pytest.raises(ValueError, match="at least 1 second"): + plugin.interval_literal(timedelta(milliseconds=500)) + + +def test_build_query_quotes_identifiers_dedupes_aliases_and_orders_bins(): + aggregations = { + 'te"mp': [ + ["first_value", ">", plugin.operator.gt, 30.0, "ERROR"], + ["first_value", ">", plugin.operator.gt, 10.0, "WARN"], + ] + } + + query = plugin.build_query( + aggregations, + "cpu", + ["ho st"], + plugin.interval_literal(timedelta(minutes=1)), + datetime(2026, 8, 5, 11, 50, tzinfo=timezone.utc), + datetime(2026, 8, 5, 12, 0, tzinfo=timezone.utc), + ) + + assert query.count('as "te""mp_first_value"') == 1 + assert 'first_value("te""mp" ORDER BY time)' in query + assert 'FROM\n "cpu"' in query + assert 'GROUP BY\n _time, "ho st"' in query + assert query.rstrip().endswith("ORDER BY\n _time") + assert "INTERVAL '60 seconds'" in query + assert "time >= '2026-08-05T11:50:00.000000Z'" in query + + +# --- configuration ---------------------------------------------------------- + + +def test_load_config_reports_missing_required_argument(plugin_dir): + client = FakeInfluxdb3Local() + + config = plugin._load_config( + client, {"senders": "http"}, plugin._WRITES_VALIDATORS, "tid" + ) + + assert config is None + assert any("measurement is required" in m for m in client.messages("error")) + + +def test_load_config_rejects_non_toml_path(plugin_dir): + client = FakeInfluxdb3Local() + + config = plugin._load_config( + client, {"config_file_path": "conf.txt"}, plugin._WRITES_VALIDATORS, "tid" + ) + + assert config is None + assert any("expected a .toml file" in m for m in client.messages("error")) + + +def test_load_config_from_toml_uses_native_structures(plugin_dir): + (plugin_dir / "conf.toml").write_text( + 'measurement = "cpu"\n' + 'senders = ["http"]\n' + 'http_webhook_url = "https://example.com/hook"\n' + 'field_conditions = [["temp", ">", 30.0, "WARN"]]\n' + ) + + config = plugin._load_config( + FakeInfluxdb3Local(), + {"config_file_path": "conf.toml"}, + plugin._WRITES_VALIDATORS, + "tid", + ) + + assert config["field_conditions"] == [["temp", ">", 30.0, "WARN"]] + assert config["trigger_count"] == 1 + assert config["notification_path"] == "notify" + + +def test_blank_token_falls_back_to_environment(monkeypatch, plugin_dir, sent): + monkeypatch.setenv("INFLUXDB3_AUTH_TOKEN", "env-tok") + client = FakeInfluxdb3Local() + + plugin.process_writes( + client, + batch([{"host": "a", "temp": 40.0}]), + {**WRITES_ARGS, "influxdb3_auth_token": ""}, + ) + + assert sent[0]["headers"]["Authorization"] == "Bearer env-tok" + + +# --- data write flow -------------------------------------------------------- + + +def test_writes_alerts_on_trigger_count_and_resets_on_non_breach(plugin_dir, sent): + client = FakeInfluxdb3Local() + args = {**WRITES_ARGS, "trigger_count": "2"} + + plugin.process_writes(client, batch([{"host": "a", "temp": 40.0}]), args) + assert sent == [] + + plugin.process_writes(client, batch([{"host": "a", "temp": 10.0}]), args) + plugin.process_writes(client, batch([{"host": "a", "temp": 41.0}]), args) + assert sent == [] + + plugin.process_writes(client, batch([{"host": "a", "temp": 42.0}]), args) + assert len(sent) == 1 + assert sent[0]["url"] == "http://localhost:8181/api/v3/engine/notify" + assert ( + sent[0]["payload"]["notification_text"] + == "[WARN] InfluxDB 3 alert triggered. Condition temp > 30 matched 2 times(42.0) " + "— matched in row cpu:temp:WARN:host=a." + ) + + +def test_writes_evaluates_every_condition_and_row(plugin_dir, sent): + client = FakeInfluxdb3Local() + + plugin.process_writes( + client, + batch([{"host": "a", "temp": 60.0}, {"host": "b", "temp": 40.0}]), + { + **WRITES_ARGS, + "field_conditions": "temp>30-WARN:temp>50-ERROR", + "notification_text": "$level $compare_val $row", + }, + ) + + assert [p["payload"]["notification_text"] for p in sent] == [ + "WARN 30 cpu:temp:WARN:host=a", + "ERROR 50 cpu:temp:ERROR:host=a", + "WARN 30 cpu:temp:WARN:host=b", + ] + + +def test_writes_warn_when_field_missing_in_row(plugin_dir, sent): + client = FakeInfluxdb3Local() + + plugin.process_writes(client, batch([{"host": "a", "other": 1.0}]), WRITES_ARGS) + + assert sent == [] + assert any("Field 'temp' not found" in m for m in client.messages("warn")) + + +def test_writes_respect_port_override_and_notification_path(plugin_dir, sent): + client = FakeInfluxdb3Local() + + plugin.process_writes( + client, + batch([{"host": "a", "temp": 40.0}]), + {**WRITES_ARGS, "port_override": "8182", "notification_path": "custom/path"}, + ) + + assert sent[0]["url"] == "http://localhost:8182/api/v3/engine/custom/path" + assert sent[0]["payload"]["senders_config"] == { + "http": {"http_webhook_url": "https://example.com/hook"} + } + + +def test_writes_exits_before_loading_config_for_other_tables(plugin_dir, sent): + client = FakeInfluxdb3Local() + + plugin.process_writes( + client, batch([{"host": "a", "temp": 40.0}], table="mem"), WRITES_ARGS + ) + + assert sent == [] + assert client.messages() == [] + + +def test_writes_caches_config_between_invocations(plugin_dir, sent): + client = FakeInfluxdb3Local() + + plugin.process_writes(client, batch([{"host": "a", "temp": 40.0}]), WRITES_ARGS) + cached = client.cache.get(plugin._WRITES_CONFIG_CACHE_KEY) + plugin.process_writes( + client, batch([{"host": "a", "temp": 41.0}]), {"measurement": "cpu"} + ) + + assert cached["measurement"] == "cpu" + assert ( + client.cache.ttls[plugin._WRITES_CONFIG_CACHE_KEY] + == plugin._WRITES_CONFIG_TTL_SECONDS + ) + assert len(sent) == 2 + + +def test_writes_pick_up_tag_added_after_a_tagless_run(plugin_dir, sent): + client = FakeInfluxdb3Local(tags=()) + + plugin.process_writes(client, batch([{"temp": 40.0}]), WRITES_ARGS) + client.tags = ["host"] + plugin.process_writes(client, batch([{"host": "a", "temp": 41.0}]), WRITES_ARGS) + + rows = [p["payload"]["notification_text"].rsplit("row ", 1)[1] for p in sent] + assert rows == ["cpu:temp:WARN.", "cpu:temp:WARN:host=a."] + + +def test_writes_report_unknown_measurement(plugin_dir, sent): + client = FakeInfluxdb3Local(tables=("mem",)) + + plugin.process_writes(client, batch([{"host": "a", "temp": 40.0}]), WRITES_ARGS) + + assert sent == [] + assert any("not found in database" in m for m in client.messages("error")) + + +def test_writes_drop_alert_after_failed_delivery(monkeypatch, plugin_dir): + monkeypatch.setattr(plugin.requests, "post", lambda *a, **kw: FakeResponse(500)) + monkeypatch.setattr(plugin.time, "sleep", lambda seconds: None) + client = FakeInfluxdb3Local() + + plugin.process_writes(client, batch([{"host": "a", "temp": 40.0}]), WRITES_ARGS) + + assert len([m for m in client.messages("warn") if "Error sending alert" in m]) == 3 + assert any("after 3 attempts" in m for m in client.messages("error")) + + +# --- scheduled flow --------------------------------------------------------- + + +def test_scheduled_window_bounds_treat_call_time_as_utc( + plugin_dir, sent, host_timezone +): + host_timezone("Europe/Warsaw") + client = FakeInfluxdb3Local(rows=[]) + + plugin.process_scheduled_call( + client, datetime(2026, 8, 5, 12, 0), {**SCHEDULED_ARGS, "deadman_check": "true"} + ) + + assert any( + "from 2026-08-05 11:50:00+00:00 to 2026-08-05 12:00:00+00:00" in m + for m in client.messages("info") + ) + + +def test_scheduled_deadman_accumulates_then_resets_when_data_returns(plugin_dir, sent): + client = FakeInfluxdb3Local(rows=[]) + args = {**SCHEDULED_ARGS, "deadman_check": "true", "trigger_count": "2"} + + plugin.process_scheduled_call(client, datetime(2026, 8, 5, 12, 0), args) + assert sent == [] + + plugin.process_scheduled_call(client, datetime(2026, 8, 5, 12, 10), args) + assert sent[0]["payload"]["notification_text"].startswith( + "Deadman Alert: No data received" + ) + + client.rows = [ + {"_time": datetime(2026, 8, 5, 12, 15), "host": "a", "temp_avg": 1.0} + ] + plugin.process_scheduled_call(client, datetime(2026, 8, 5, 12, 20), args) + assert client.cache.get("cpu") == "0" + assert len(sent) == 1 + + +def test_scheduled_threshold_alert_reports_aggregation_and_row(plugin_dir, sent): + client = FakeInfluxdb3Local( + rows=[{"_time": datetime(2026, 8, 5, 12, 0), "host": "a", "temp_avg": 40.0}] + ) + + plugin.process_scheduled_call( + client, + datetime(2026, 8, 5, 12, 0), + {**SCHEDULED_ARGS, "field_aggregation_values": "temp:avg@>30-ERROR"}, + ) + + assert ( + sent[0]["payload"]["notification_text"] + == "[ERROR] Threshold Alert on table cpu: avg of temp > 30.0 (actual: 40.0) " + "— matched in row cpu:temp:avg:ERROR:host=a." + ) + + +def test_scheduled_uses_custom_template_and_counts_bins_of_one_run(plugin_dir, sent): + client = FakeInfluxdb3Local( + rows=[ + {"_time": datetime(2026, 8, 5, 12, 0), "host": "a", "temp_avg": 40.0}, + {"_time": datetime(2026, 8, 5, 12, 1), "host": "a", "temp_avg": 41.0}, + ] + ) + + plugin.process_scheduled_call( + client, + datetime(2026, 8, 5, 12, 2), + { + **SCHEDULED_ARGS, + "trigger_count": "2", + "field_aggregation_values": "temp:avg@>30-ERROR", + "notification_threshold_text": "S $aggregation $actual $row", + }, + ) + + assert [p["payload"]["notification_text"] for p in sent] == [ + "S avg 41.0 cpu:temp:avg:ERROR:host=a" + ] + + +def test_scheduled_skips_condition_when_aggregate_column_missing(plugin_dir, sent): + client = FakeInfluxdb3Local( + rows=[{"_time": datetime(2026, 8, 5, 12, 0), "host": "a", "temp_avg": 40.0}] + ) + + plugin.process_scheduled_call( + client, + datetime(2026, 8, 5, 12, 0), + {**SCHEDULED_ARGS, "field_aggregation_values": "temp:max@>30-ERROR"}, + ) + + assert sent == [] + assert any("'temp_max' not found" in m for m in client.messages("warn")) + + +def test_scheduled_requires_conditions_or_deadman(plugin_dir, sent): + client = FakeInfluxdb3Local(rows=[]) + + plugin.process_scheduled_call(client, datetime(2026, 8, 5, 12, 0), SCHEDULED_ARGS) + + assert sent == [] + assert any("deadman_check to True" in m for m in client.messages("error")) From e425f1c7fc8b084a47b9ddec721e8b753d75b646 Mon Sep 17 00:00:00 2001 From: Aliaksei Kharlap Date: Wed, 5 Aug 2026 22:16:24 +0300 Subject: [PATCH 4/4] fix version --- influxdata/threshold_deadman_checks/manifest.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/influxdata/threshold_deadman_checks/manifest.toml b/influxdata/threshold_deadman_checks/manifest.toml index c49be1d..4832fd6 100644 --- a/influxdata/threshold_deadman_checks/manifest.toml +++ b/influxdata/threshold_deadman_checks/manifest.toml @@ -2,7 +2,7 @@ manifest_schema_version = "1.3" [plugin] name = "threshold_deadman_checks" -version = "2.0.0" +version = "1.3.0" description = "Provides comprehensive monitoring capabilities including deadman alerts and aggregation-based threshold checks. Supports both scheduler and data write triggers with multi-channel notifications." triggers = ["process_writes", "process_scheduled_call"] homepage = "https://www.influxdata.com/"