From a6925ac7cac6a6fb3e8b8a6f0dc35235fad4fff1 Mon Sep 17 00:00:00 2001 From: Aliaksei Kharlap Date: Wed, 5 Aug 2026 20:37:49 +0300 Subject: [PATCH 1/3] 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 2/3] 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 3/3] 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/"