diff --git a/influxdata/library/plugin_library.json b/influxdata/library/plugin_library.json index d30d7ec..e769879 100644 --- a/influxdata/library/plugin_library.json +++ b/influxdata/library/plugin_library.json @@ -106,8 +106,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-17", "trigger_types_supported": ["scheduler", "data_writes"] }, { diff --git a/influxdata/state_change/README.md b/influxdata/state_change/README.md index be507e4..125f484 100644 --- a/influxdata/state_change/README.md +++ b/influxdata/state_change/README.md @@ -19,20 +19,20 @@ 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 field changes | -| `field_change_count` | string | required | Dot-separated field thresholds (for example, "temp:3.load:2"). Supports count-based conditions | -| `senders` | string | required | Dot-separated notification channels with multi-channel alert support (Slack, Discord, etc.) | -| `window` | string | required | Time window for analysis. Format: `` (for example, "10m", "1h") | +| Parameter | Type | Default | Description | +|----------------------|--------|----------|------------------------------------------------------------------------------------------------------------------------| +| `measurement` | string | required | Measurement to monitor for field changes | +| `field_change_count` | string | required | Dot-separated field thresholds (for example, "temp:3.load:2" or "temp:3.disk.used:2"). Each count must be 1 or greater | +| `senders` | string | required | Dot-separated notification channels with multi-channel alert support (Slack, Discord, etc.) | +| `window` | string | required | Time window for analysis. Format: ``, units: `us`, `ms`, `s`, `min`, `h`, `d`, `w`. Must be positive | ### Data write trigger parameters -| Parameter | Type | Default | Description | -|--------------------|--------|----------|----------------------------------------------------------------------------------------------------------------| -| `measurement` | string | required | Measurement to monitor for threshold conditions | -| `field_thresholds` | string | required | Flexible threshold conditions with count-based and duration-based support (e.g., "temp:30:10@status:ok:1h") | -| `senders` | string | required | Dot-separated notification channels with multi-channel alert support (Slack, Discord, HTTP, SMS, WhatsApp) | +| Parameter | Type | Default | Description | +|--------------------|--------|----------|-------------------------------------------------------------------------------------------------------------------------------------------------------------| +| `measurement` | string | required | Measurement to monitor for threshold conditions | +| `field_thresholds` | string | required | Threshold conditions with count-based and duration-based support (e.g., "temp:30:10@status:ok:1h"). Counts must be 1 or greater; durations must be positive | +| `senders` | string | required | Dot-separated notification channels with multi-channel alert support (Slack, Discord, HTTP, SMS, WhatsApp) | ### Notification parameters @@ -49,8 +49,10 @@ This plugin includes a JSON metadata schema in its docstring that defines suppor | Parameter | Type | Default | Description | |-----------------------|--------|---------|-------------------------------------------------------------------------------------------| -| `state_change_window` | number | 1 | Recent values to check for stability (configurable state change detection to reduce noise) | -| `state_change_count` | number | 1 | Max changes allowed within stability window (configurable state change detection) | +| `state_change_window` | number | 1 | Recent values to check for stability (reduces noise from flapping fields) | +| `state_change_count` | number | 1 | Changes within the stability window at which notifications start being suppressed | + +The stability check applies only when `state_change_window` is 2 or greater; the default of 1 leaves it off. Notifications are suppressed once the window contains `state_change_count` changes, so `state_change_count=3` is the setting that tolerates two flips. ### TOML configuration @@ -58,7 +60,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_thresholds`, and `field_change_count` use native structures (list, list of entries, table) instead of the inline string formats, though the inline strings are also accepted. + +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: @@ -80,6 +86,7 @@ The plugin assumes that the table schema is already defined in the database, as - **InfluxDB 3 Core/Enterprise**: with the Processing Engine enabled. - **Notification Sender Plugin for InfluxDB 3**: Required for sending notifications. See the [influxdata/notifier plugin](../notifier/README.md). - **Python packages**: + - `influxdata-plugin-utils>=0.3.0` (configuration loading, parsing, and schema introspection) - `requests` (for HTTP notifications) ### Installation steps @@ -97,6 +104,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 ``` @@ -113,7 +121,7 @@ influxdb3 create trigger \ --database mydb \ --path "gh:influxdata/state_change/state_change_check_plugin.py" \ --trigger-spec "every:10m" \ - --trigger-arguments "measurement=cpu,field_change_count=temp:3.load:2,window=10m,senders=slack,slack_webhook_url=$SLACK_WEBHOOK_URL" \ + --trigger-arguments "measurement=cpu,field_change_count=temp:3.load:2,window=10min,senders=slack,slack_webhook_url=$SLACK_WEBHOOK_URL" \ state_change_scheduler ``` @@ -186,7 +194,7 @@ Set `SLACK_WEBHOOK_URL` to your Slack incoming webhook URL. **Expected output** -When the field changes more than 5 times within 1 hour, a notification is sent: "Temperature sensor value changed 6 times in 1h for tags location=office" +When the field changes 5 or more times within 1 hour, a notification is sent: "Field value in table temperature changed 6 times in window 1:00:00 for tags location=office" ### Example 2: Advanced scheduled field change monitoring @@ -240,6 +248,9 @@ Set `SLACK_WEBHOOK_URL` to your Slack incoming webhook URL. - `state_change_check_plugin.py`: The main plugin code containing handlers for scheduled and data write triggers - `state_change_config_scheduler.toml`: Example TOML configuration for scheduled triggers - `state_change_config_data_writes.toml`: Example TOML configuration for data write triggers +- `test_state_change.py`: Pytest suite, 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 @@ -280,17 +291,21 @@ Handles real-time threshold monitoring on data writes. Evaluates incoming data a **Count-based thresholds** - Format: `field_name:"value":count` -- Example: `temp:"30.5":10` (10 occurrences of temperature = 30.5) +- Example: `temp:"30.5":10` (10 consecutive occurrences of temperature = 30.5) +- The count must be an integer of 1 or greater **Time-based thresholds** - Format: `field_name:"value":duration` - Example: `status:"error":5min` (status = error for 5 minutes) -- Supported units: `s`, `min`, `h`, `d`, `w` +- Supported units: `us`, `ms`, `s`, `min`, `h`, `d`, `w`; the duration must be positive **Multiple conditions** - Separate with `@`: `temp:"30":5@humidity:"high":10min` +- Segments that fail to parse are skipped with a warning; if none remain, the run stops with an error + +In TOML, the same thresholds are written as entries: `field_thresholds = [["temp", 30.5, 10], ["status", "error", "5min"]]`. ### Message template variables diff --git a/influxdata/state_change/manifest.toml b/influxdata/state_change/manifest.toml index b2a5b86..3de3969 100644 --- a/influxdata/state_change/manifest.toml +++ b/influxdata/state_change/manifest.toml @@ -2,7 +2,7 @@ manifest_schema_version = "1.3" [plugin] name = "state_change" -version = "1.2.0" +version = "1.3.0" description = "Provides field change and threshold monitoring capabilities through scheduler and data writes plugins. Detects changes in field values or threshold conditions with customizable notification templates." 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/state_change/requirements-dev.txt b/influxdata/state_change/requirements-dev.txt new file mode 100644 index 0000000..57bfbef --- /dev/null +++ b/influxdata/state_change/requirements-dev.txt @@ -0,0 +1,3 @@ +pytest +influxdata-plugin-utils>=0.3.0 +requests \ No newline at end of file diff --git a/influxdata/state_change/requirements.txt b/influxdata/state_change/requirements.txt index 663bd1f..3349a3f 100644 --- a/influxdata/state_change/requirements.txt +++ b/influxdata/state_change/requirements.txt @@ -1 +1,2 @@ +influxdata-plugin-utils>=0.3.0 requests \ No newline at end of file diff --git a/influxdata/state_change/state_change_check_plugin.py b/influxdata/state_change/state_change_check_plugin.py index 88dbf35..9fc7518 100644 --- a/influxdata/state_change/state_change_check_plugin.py +++ b/influxdata/state_change/state_change_check_plugin.py @@ -10,8 +10,8 @@ }, { "name": "field_change_count", - "example": "temp:3.load:2", - "description": "Dot-separated list of field thresholds (e.g., field:count).", + "example": "temp:3.disk.used:2", + "description": "Dot-separated list of field thresholds (e.g., field:count). Each count must be 1 or greater.", "required": true }, { @@ -23,7 +23,7 @@ { "name": "window", "example": "1h", - "description": "Time window for data analysis (e.g., '1h' for 1 hour). Units: 's', 'min', 'h', 'd', 'w'.", + "description": "Time window for data analysis (e.g., '1h' for 1 hour). Units: 'us', 'ms', 's', 'min', 'h', 'd', 'w'. Must be a positive duration.", "required": true }, { @@ -113,7 +113,7 @@ { "name": "config_file_path", "example": "config.toml", - "description": "Path to config file to override args. Format: 'config.toml'.", + "description": "Path to a TOML config file that replaces the trigger arguments entirely. Format: 'config.toml'.", "required": false } ], @@ -127,7 +127,7 @@ { "name": "field_thresholds", "example": "temp:'30.1':10@humidity:'true':2h", - "description": "Threshold conditions (e.g., field:value:count or field:value:time). Multiple conditions separated by '@'.", + "description": "Threshold conditions (e.g., field:value:count or field:value:time). Multiple conditions separated by '@'. Count must be 1 or greater; duration units: 'us', 'ms', 's', 'min', 'h', 'd', 'w'.", "required": true }, { @@ -145,13 +145,13 @@ { "name": "state_change_window", "example": "5", - "description": "Number of recent values to check for stability. Default: 1.", + "description": "Number of recent values to check for stability. The stability check applies only when this is 2 or greater. Default: 1.", "required": false }, { "name": "state_change_count", "example": "2", - "description": "Maximum allowed changes within state_change_window to allow notifications. Default: 1.", + "description": "Number of changes within state_change_window at which notifications start being suppressed. Default: 1.", "required": false }, { @@ -241,7 +241,7 @@ { "name": "config_file_path", "example": "config.toml", - "description": "Path to config file to override args. Format: 'config.toml'.", + "description": "Path to a TOML config file that replaces the trigger arguments entirely. Format: 'config.toml'.", "required": false } ] @@ -253,15 +253,24 @@ import random import re import time -import tomllib import uuid from collections import defaultdict, deque 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, + query_window, +) +from influxdata_plugin_utils.parsing import ( + parse_delimited_list, + parse_int, + parse_timedelta, +) # Supported sender types with their required arguments AVAILABLE_SENDERS = { @@ -280,72 +289,109 @@ # List of keywords to exclude from argument validation in AVAILABLE_SENDERS EXCLUDED_KEYWORDS = ["headers", "token", "sid"] - -def get_all_measurements(influxdb3_local) -> list[str]: +_DEFAULT_NOTIFICATION_TEXT = ( + "Field $field in table $table changed $changes times in window $window " + "for tags $tags" +) +_DEFAULT_COUNT_TEXT = ( + "State change detected: Field $field in table $table changed to $value " + "during last $duration times. Row: $row" +) +_DEFAULT_TIME_TEXT = ( + "State change detected: Field $field in table $table changed to $value " + "during $duration. Row: $row" +) + +_CHANGE_COUNT_PAIR_RE = re.compile(r"(?P[^:]+):\s*(?P-?\d+)\s*(?:\.|$)") + + +def parse_window(raw) -> timedelta: + """Parse the analysis 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( + "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_thresholds", required=True), + Validator( + "state_change_window", default=1, cast=lambda raw: parse_int(raw, minimum=0) + ), + Validator( + "state_change_count", default=1, cast=lambda raw: parse_int(raw, minimum=0) + ), + Validator("notification_count_text", default=_DEFAULT_COUNT_TEXT, cast=str), + Validator("notification_time_text", default=_DEFAULT_TIME_TEXT, cast=str), +] + +_SCHEDULED_VALIDATORS = _COMMON_VALIDATORS + [ + Validator("field_change_count", required=True), + Validator("window", required=True, cast=parse_window), + Validator("notification_text", default=_DEFAULT_NOTIFICATION_TEXT, cast=str), +] + +_WRITES_CONFIG_CACHE_KEY = "state_change: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. 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" - ] - - # cache the result for 1 hour - influxdb3_local.cache.put(f"measurements", measurements, 60 * 60) - - return measurements + 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 -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. + return {key.lower(): value for key, value in loaded.as_dict().items()} - 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 = """ - 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: +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: + 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 [] - - 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 + return tags def generate_cache_key( @@ -366,65 +412,55 @@ def generate_cache_key( return cache_key -def parse_senders(influxdb3_local, args: dict, task_id: str) -> dict: +def read_counter(influxdb3_local, cache_key: str) -> int: + """Read a breach counter, treating a missing or non-numeric entry as zero.""" + try: + return int(influxdb3_local.cache.get(cache_key)) + except (TypeError, ValueError): + return 0 + + +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": dot-separated list of sender types (e.g., "slack.http"). - - For each sender, its own required keys (see AVAILABLE_SENDERS). - task_id (str): Unique task identifier used for logging context. + config (dict): Loaded config containing "senders" and related settings. + task_id (str): Unique task identifier. Returns: dict: A mapping `{sender_type: {key: value}}` for each valid sender. - For example: - { - "slack": { - "slack_webhook_url": "https://hooks.slack.com/...", - "slack_headers": "..." - }, - "sms": { ... } - } Raises: Exception: If no valid senders are found after parsing. """ 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}] Required key '{key}' missing for sender '{sender}'" ) 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 @@ -443,8 +479,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 = { @@ -480,36 +516,6 @@ def send_notification( ) -def parse_port_override(args: dict, task_id: str) -> int: - """ - Parse and validate the 'port_override' argument, converting it from string to int. - - Args: - args (dict): Runtime arguments containing 'port_override'. - task_id (str): Unique task identifier for logging context. - - Returns: - int: Parsed port number (1–65535), or 8181 if not provided. - - 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) - - try: - port = int(raw) - except (TypeError, ValueError): - raise Exception(f"[{task_id}] Invalid port_override, not an integer: {raw!r}") - - # Validate port range - if not (1 <= port <= 65535): - raise Exception( - f"[{task_id}] Invalid port_override, must be between 1 and 65535: {port}" - ) - - return port - - def validate_webhook_url(influxdb3_local, service: str, url: str, task_id: str) -> bool: """ Validate webhook URL format. @@ -575,115 +581,68 @@ def _coerce_value(raw: str) -> str | int | float | bool: return raw -def parse_field_thresholds( - influxdb3_local, - args: dict, - task_id: str, -) -> list: +def _parse_threshold_param( + influxdb3_local, raw, task_id: str +) -> int | timedelta | None: """ - Extracts and parses field threshold definitions from args or use values from config file. + Parse the third part of a threshold into a consecutive count or a duration. - Args: - influxdb3_local: InfluxDB client instance. - args (dict): Dictionary with the 'field_thresholds' key. - task_id (str): Unique identifier for the current task, used for logging. + A bare integer is a count of consecutive matching points; anything else is a + duration such as '10s' or '2h'. - The args dict must contain the key "field_thresholds" with a value like: - 'field_name-1:"value 2":10@field_name-2:"value 1":1s' + Returns: + int | timedelta | None: The parsed threshold, or None when it is invalid. + """ + if isinstance(raw, bool): + influxdb3_local.warn(f"[{task_id}] Invalid threshold parameter: {raw!r}") + return None - Each '@'-separated segment must contain exactly two ':' characters, - producing three parts: field_name, raw_value, and raw_third. + if isinstance(raw, int) or re.fullmatch(r"-?\d+", str(raw).strip()): + count = int(raw) + if count < 1: + influxdb3_local.warn( + f"[{task_id}] Invalid threshold count {count}, must be 1 or greater" + ) + return None + return count - - raw_value is coerced into int, float, bool, or str. - - raw_third is either: - • A plain integer (e.g., "10") - • A duration with unit suffix: , where unit ∈ {s, min, h, d, w} + try: + duration: timedelta = parse_timedelta(raw) + except ValueError as e: + influxdb3_local.warn(f"[{task_id}] Invalid threshold duration {raw!r}: {e}") + return None + + if duration <= timedelta(0): + influxdb3_local.warn( + f"[{task_id}] Invalid threshold duration {raw!r}, must be positive" + ) + return None + return duration - Valid units and their corresponding timedelta keyword: - "s" → "seconds" - "min" → "minutes" - "h" → "hours" - "d" → "days" - "w" → "weeks" - Returns: - A list of lists [field_name, coerced_value, converted_third]: - - coerced_value is int, float, bool, or str - - converted_third is int or timedelta +def _thresholds_from_entries(influxdb3_local, entries: list, task_id: str) -> list: + """Parse thresholds given as [field, value, count_or_duration] entries.""" + thresholds: list = [] - Example: - args = { - ... "field_thresholds": 'temp:"30":60@humidity:"true":2h' - ... } - parse_field_thresholds(args) - [ - ["temp", 30, 60], - ["humidity", True, datetime.timedelta(hours=2)] - ] - """ - valid_units: dict[str, str] = { - "s": "seconds", - "min": "minutes", - "h": "hours", - "d": "days", - "w": "weeks", - } + for entry in entries: + if not isinstance(entry, (list, tuple)) or len(entry) != 3: + influxdb3_local.warn( + f"[{task_id}] Invalid threshold '{entry}', expected [field, value, count_or_duration]" + ) + continue + threshold_param = _parse_threshold_param(influxdb3_local, entry[2], task_id) + if threshold_param is None: + continue + thresholds.append((str(entry[0]), entry[1], threshold_param)) - raw_input: str | list = args.get("field_thresholds") - results: list = [] + return thresholds - if args["use_config_file"]: - if not isinstance(raw_input, list): - raise Exception( - "[{task_id}] field_thresholds must be a list while using config file" - ) - for threshold in raw_input: - try: - field_name = str(threshold[0]) - value = threshold[1] - duration: str | int = threshold[2] - if isinstance(duration, str): - num_part, unit_part = "", "" - for unit in sorted(valid_units.keys(), key=len, reverse=True): - if duration.endswith(unit): - num_part = duration[: -len(unit)] - unit_part = unit - break - if not num_part or unit_part not in valid_units: - influxdb3_local.warn( - f"[{task_id}] Invalid duration format '{duration}'" - ) - continue - try: - num = int(num_part) - except ValueError: - influxdb3_local.warn( - f"[{task_id}] Invalid number in duration '{duration}'" - ) - continue - threshold_param: timedelta | int = timedelta( - **{valid_units[unit_part]: num} - ) - elif isinstance(duration, int): - threshold_param = duration - else: - influxdb3_local.warn( - f"[{task_id}] Invalid duration format '{duration}'" - ) - continue - results.append([field_name, value, threshold_param]) - except Exception: - influxdb3_local.warn( - f"[{task_id}] Invalid duration definition: {threshold}, skipping" - ) - if not results: - raise Exception( - f"[{task_id}] No valid field threshold segments found in {raw_input}" - ) - return results - segments: list = [seg.strip() for seg in raw_input.split("@") if seg.strip()] - for segment in segments: +def _thresholds_from_string(influxdb3_local, raw: str, task_id: str) -> list: + """Parse thresholds given as '::' joined by '@'.""" + thresholds: list = [] + + for segment in parse_delimited_list(raw, sep="@"): # Each segment must contain exactly two ':' characters if segment.count(":") != 2: influxdb3_local.warn( @@ -691,51 +650,132 @@ def parse_field_thresholds( ) continue - # Split into three parts: field_name, raw_value, raw_third - field_name, raw_value, raw_third = segment.split(":", 2) - field_name = field_name.strip() - raw_value = raw_value.strip() - raw_third = raw_third.strip() - - # Coerce raw_value into int, float, bool, or str - value = _coerce_value(raw_value) - - # Parse raw_third: integer or duration - if re.fullmatch(r"-?\d+", raw_third): - third_converted: int | timedelta = int(raw_third) - else: - # Attempt duration parsing: - num_part: str = "" - unit_part: str = "" - for unit in sorted(valid_units.keys(), key=len, reverse=True): - if raw_third.endswith(unit): - num_part = raw_third[: -len(unit)] - unit_part = unit - break - - if not num_part or unit_part not in valid_units: - influxdb3_local.warn( - f"[{task_id}] Invalid duration format: {raw_third}" - ) - continue + field_name, raw_value, raw_param = segment.split(":", 2) + threshold_param = _parse_threshold_param( + influxdb3_local, raw_param.strip(), task_id + ) + if threshold_param is None: + continue + thresholds.append( + (field_name.strip(), _coerce_value(raw_value), threshold_param) + ) - try: - num = int(num_part) - except ValueError: - influxdb3_local.warn(f"[{task_id}] Invalid duration number: {num_part}") - continue + return thresholds - kw: str = valid_units[unit_part] - third_converted = timedelta(**{kw: num}) - results.append((field_name, value, third_converted)) +def parse_field_thresholds(influxdb3_local, config: dict, task_id: str) -> list: + """ + Parse the field thresholds used by the data write trigger. - if not results: + Thresholds come either as entries of [field, value, count_or_duration] (TOML) or + as a string of '::' expressions separated by '@'. + + Args: + influxdb3_local: InfluxDB client instance. + config (dict): Loaded config containing "field_thresholds". + task_id (str): Unique task identifier. + + Returns: + list[tuple]: Tuples of (field_name, target_value, count_or_duration). + + Example: + 'temp:"30":60@humidity:"true":2h' + [ + ("temp", 30, 60), + ("humidity", True, datetime.timedelta(hours=2)), + ] + """ + raw: str | list = config["field_thresholds"] + + if isinstance(raw, (list, tuple)): + thresholds = _thresholds_from_entries(influxdb3_local, raw, task_id) + elif isinstance(raw, str): + thresholds = _thresholds_from_string(influxdb3_local, raw, task_id) + else: raise Exception( - f"[{task_id}] No valid field threshold segments found in {raw_input}" + "'field_thresholds' must be a list of entries or a string, " + f"got {type(raw).__name__}" + ) + + if not thresholds: + raise Exception("No valid field thresholds provided.") + return thresholds + + +def _change_counts_from_string(influxdb3_local, raw: str, task_id: str) -> list: + """ + Split a string of 'field:count' pairs joined by '.' into (field, count) tuples. + + Example: + 'temp:3.disk.used:2' -> [('temp', '3'), ('disk.used', '2')] + """ + pairs: list = [] + text: str = raw.strip() + position: int = 0 + + while position < len(text): + match = _CHANGE_COUNT_PAIR_RE.match(text, position) + if match: + pairs.append((match.group("field"), match.group("count"))) + position = match.end() + continue + + next_dot: int = text.find(".", position) + skipped: str = text[position:] if next_dot == -1 else text[position:next_dot] + influxdb3_local.warn( + f"[{task_id}] Invalid format of field_change_count, expected 'field:count' in: {skipped}" ) + if next_dot == -1: + break + position = next_dot + 1 - return results + return pairs + + +def parse_field_change_count( + influxdb3_local, config: dict, task_id: str +) -> dict[str, int]: + """ + Parse the per-field change thresholds used by the scheduled trigger. + + Thresholds come either as a mapping of {field: count} (TOML) or as a string of + 'field:count' pairs separated by '.'. + + Args: + influxdb3_local: InfluxDB client instance. + config (dict): Loaded config containing "field_change_count". + task_id (str): Unique task identifier. + + Returns: + dict[str, int]: Field names mapped to their change count thresholds. + + Raises: + Exception: If the value has an unsupported type or no valid fields are found. + """ + raw: str | dict = config["field_change_count"] + + if isinstance(raw, dict): + pairs: list = list(raw.items()) + elif isinstance(raw, str): + pairs = _change_counts_from_string(influxdb3_local, raw, task_id) + else: + raise Exception( + "'field_change_count' must be a mapping or a string, " + f"got {type(raw).__name__}" + ) + + field_counts: dict = {} + for field, raw_count in pairs: + try: + field_counts[str(field).strip()] = parse_int(raw_count, minimum=1) + except ValueError as e: + influxdb3_local.warn( + f"[{task_id}] Invalid change count for field '{field}': {e}" + ) + + if not field_counts: + raise Exception("No valid entries found in field_change_count.") + return field_counts def check_state_changes(cached_values: deque, state_change_count: int) -> bool: @@ -744,12 +784,12 @@ def check_state_changes(cached_values: deque, state_change_count: int) -> bool: Args: cached_values (deque): A deque of recent field values (size = state_change_window). - state_change_count (int): Maximum allowed number of changes within the window. + state_change_count (int): Number of changes at which notifications are suppressed. Returns: bool: - True if the number of value changes in cached_values is <= state_change_count, - False if it exceeds state_change_count. + True while the number of value changes in cached_values stays below + state_change_count, False once it reaches it. """ # If fewer than 2 values, there can be no change if len(cached_values) < 2: @@ -774,154 +814,86 @@ def check_state_changes(cached_values: deque, state_change_count: int) -> bool: return True -def process_writes(influxdb3_local, table_batches: list, args: dict | None = None): +def process_writes(influxdb3_local, table_batches: list, args: dict): """ - Data write trigger entry point implementing field‐level thresholds with “count” and “duration” logic, - while also suppressing notifications if the field value has flipped too many times recently. + Data write trigger entry point implementing field-level thresholds with "count" and + "duration" logic, while also suppressing notifications if the field value has flipped + too many times recently. - When you create a Data Write trigger, point to this file and the function name must be `process_writes`. - Other names are not supported. + When you create a Data Write trigger, point to this file and the function name must be + `process_writes`. Other names are not supported. - The trigger fires on each WAL flush. All newly written rows within that flush—optionally filtered by - a configured measurement—are grouped into `table_batches`. + The trigger fires on each WAL flush. All newly written rows within that flush are grouped + into `table_batches`; only batches of the configured measurement are processed. Args: - influxdb3_local: - InfluxDB client instance (for logging, SQL queries, writing, and cache). - table_batches (list): - A list of dicts, each with: - - "table_name": str - - "rows": list[dict] # Each dict is one row of data, containing fields and tags. - args (dict, optional): - Must include: - - "measurement": measurement (table) name to monitor (str). - - "field_thresholds": string defining thresholds, parsed by `parse_field_thresholds`. - - "senders": dot-separated list of notification channels (e.g., "slack.sms"). - May also include: - - "config_file_path": path to config file to override args (str). - - "state_change_window": integer count of last values to consider for flip detection. - - "state_change_count": integer threshold of flips to suppress notifications. - - "port_override": HTTP port for notification plugin (default 8181). - - "influxdb3_auth_token": API v3 token (or provided via ENV var INFLUXDB3_AUTH_TOKEN). - - "notification_path": path on engine (default "notify"). - - "notification_text": template for alert text, with placeholders: - $table, $field, $value, $duration, $row. - - Raises: - Exception: Captures and logs any unexpected error (with `influxdb3_local.error`). + influxdb3_local: InfluxDB client instance (for logging, SQL queries, and cache). + table_batches (list): Dicts with "table_name" (str) and "rows" (list[dict]). + args (dict): Runtime arguments of the trigger. """ - task_id: str = str(uuid.uuid4()) - influxdb3_local.info(f"[{task_id}] Starting writes process with args: {args}") + if not table_batches: + return - # 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 args - or "measurement" not in args - or "field_thresholds" not in args - or "senders" not in args - ): - influxdb3_local.error( - f"[{task_id}] Missing required arguments: measurement, field_thresholds, 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 - # Parse configuration - 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 + 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: - field_thresholds: list = parse_field_thresholds(influxdb3_local, args, task_id) + field_thresholds: list = parse_field_thresholds( + influxdb3_local, config, task_id + ) influxdb3_local.info(f"[{task_id}] Field thresholds: {field_thresholds}") - senders_config: dict = parse_senders(influxdb3_local, args, task_id) - tags: list = get_tag_names(influxdb3_local, measurement, task_id) - port_override: int = parse_port_override(args, task_id) - state_change_window: int = int(args.get("state_change_window", 1)) - state_change_count: int = int(args.get("state_change_count", 1)) - notification_path: str = args.get("notification_path", "notify") - influxdb3_auth_token: str = args.get("influxdb3_auth_token") or os.getenv( - "INFLUXDB3_AUTH_TOKEN" + senders_config: dict = parse_senders(influxdb3_local, config, task_id) + port_override: int = config["port_override"] + state_change_window: int = config["state_change_window"] + state_change_count: int = config["state_change_count"] + 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_count_tpl = args.get( - "notification_count_text", - "State change detected: Field $field in table $table changed to $value during last $duration times. Row: $row", - ) - notification_time_tpl = args.get( - "notification_time_text", - "State change detected: Field $field in table $table changed to $value during $duration. Row: $row", - ) + notification_count_tpl: str = config["notification_count_text"] + notification_time_tpl: str = config["notification_time_text"] - # Process incoming data - for table_batch in table_batches: - # Skip non-matching tables - if table_batch["table_name"] != measurement: - continue + tags: list = get_measurement_tags(influxdb3_local, measurement, task_id) - # Process rows in this batch + for table_batch in monitored_batches: for row in table_batch["rows"]: for field_name, target_value, threshold_param in field_thresholds: - # Get cache keys - if isinstance(threshold_param, timedelta): - duration_suffix: str = "time" - else: - duration_suffix = "count" + is_duration: bool = isinstance(threshold_param, timedelta) + duration_suffix: str = "time" if is_duration else "count" + reset_value: str = "" if is_duration else "0" duration_cache_key: str = generate_cache_key( measurement=measurement, @@ -940,13 +912,12 @@ def process_writes(influxdb3_local, table_batches: list, args: dict | None = Non influxdb3_local.info( f"[{task_id}] Field '{field_name}' not present in row. Cache key: {duration_cache_key}. Resetting state." ) - influxdb3_local.cache.put(duration_cache_key, "") + influxdb3_local.cache.put(duration_cache_key, reset_value) continue # Check if the condition is satisfied: row[field_name] == target_value condition_met: bool = current_val == target_value - # Get cache keys values_cache_key: str = generate_cache_key( measurement=measurement, field=field_name, @@ -955,7 +926,6 @@ def process_writes(influxdb3_local, table_batches: list, args: dict | None = Non tags=tags, row=row, ) - # Get cached values cached_values = influxdb3_local.cache.get( values_cache_key, default=deque(maxlen=state_change_window) ) @@ -971,9 +941,9 @@ def process_writes(influxdb3_local, table_batches: list, args: dict | None = Non ) cached_values.append(current_val) - if duration_suffix == "count": - cached_state: int = int( - influxdb3_local.cache.get(duration_cache_key, default=0) + if not is_duration: + cached_state: int = read_counter( + influxdb3_local, duration_cache_key ) if condition_met: @@ -983,7 +953,6 @@ def process_writes(influxdb3_local, table_batches: list, args: dict | None = Non influxdb3_local.error( f"[{task_id}] State change detected: {field_name} in table {measurement} changed to {target_value} during last {threshold_param} values. Row: {duration_cache_key}, sending alert" ) - # Send notification payload: dict = { "notification_text": interpolate_notification_text( notification_count_tpl, @@ -1026,22 +995,19 @@ def process_writes(influxdb3_local, table_batches: list, args: dict | None = Non # Condition failed → reset count influxdb3_local.cache.put(duration_cache_key, "0") - else: # duration_suffix == "time" + else: required_duration: timedelta = threshold_param - cached_state: str = influxdb3_local.cache.get( + prev_start_iso: str = influxdb3_local.cache.get( duration_cache_key, default="" ) if condition_met: - # Parse cached start time, if any - prev_start_iso: str = cached_state + start_time = None if prev_start_iso: try: start_time = datetime.fromisoformat(prev_start_iso) except Exception: start_time = None - else: - start_time = None # Use current UTC time rather than row's "time" field now = datetime.now(timezone.utc) @@ -1060,7 +1026,6 @@ def process_writes(influxdb3_local, table_batches: list, args: dict | None = Non influxdb3_local.error( f"[{task_id}] Threshold duration reached for row: {duration_cache_key}, target_value={target_value} (required {required_duration})" ) - # Send notification payload: dict = { "notification_text": interpolate_notification_text( notification_time_tpl, @@ -1093,15 +1058,15 @@ def process_writes(influxdb3_local, table_batches: list, args: dict | None = Non influxdb3_local.cache.put(duration_cache_key, "") else: - # Update elapsed (keep original start in cache) + # Keep the original start in cache and wait influxdb3_local.warn( - f"[{task_id}] Threshold duration reached for row: {row}, target_value={target_value} with elapsed={elapsed} (required {required_duration})" + f"[{task_id}] Condition still holding for row: {duration_cache_key}, target_value={target_value} with elapsed={elapsed} (required {required_duration})" ) else: # Condition failed → reset any stored start time - if cached_state: + if prev_start_iso: influxdb3_local.info( - f"[{task_id}] Condition failed for row: {row}, clearing duration cache" + f"[{task_id}] Condition failed for row: {duration_cache_key}, clearing duration cache" ) influxdb3_local.cache.put(duration_cache_key, "") @@ -1111,211 +1076,29 @@ def process_writes(influxdb3_local, table_batches: list, args: dict | None = Non influxdb3_local.error(f"[{task_id}] Error: {str(e)}") -def parse_field_change_count( - influxdb3_local, args: dict, task_id: str -) -> dict[str, int]: - """ - Parses the 'field_change_count' parameter into a dictionary of field names and their change thresholds or use values from config file. - - Args: - influxdb3_local: InfluxDB client instance. - args (dict): Dictionary with the 'field_change_count' key. - task_id (str): Unique task identifier for logging. - - Returns: - dict[str, int]: Dictionary mapping field names to change counts. - - Raises: - Exception: If the format is invalid or no valid fields are found. - """ - raw_input: str | dict = args.get("field_change_count") - field_counts: dict = {} - - if args["use_config_file"]: - if not isinstance(raw_input, dict): - raise Exception( - f"[{task_id}] field_change_count must be a dictionary when using config file" - ) - return raw_input - - pairs: list = raw_input.split(".") - for pair in pairs: - if ":" not in pair: - influxdb3_local.warn( - f"[{task_id}] Invalid format of field_change_count, missing ':' in pair: {pair}" - ) - continue - field, count_str = pair.split(":", 1) - try: - count: int = int(count_str) - field_counts[field.strip()] = count - except ValueError: - influxdb3_local.warn( - f"[{task_id}] Invalid format of field_change_count, invalid count: {count_str} in pair: {pair}" - ) - continue - - if not field_counts: - raise Exception(f"[{task_id}] No valid entries found in field_change_count.") - - return field_counts - - -def parse_window(args: dict, task_id: str) -> timedelta: +def process_scheduled_call(influxdb3_local, call_time: datetime, args: dict) -> None: """ - Parses the 'window' argument from args and converts it into a timedelta object. + Scheduled trigger entry point that counts how often fields change within a time window + and sends a notification when a field exceeds its configured change threshold. Args: - args (dict): Dictionary with the 'window' key (e.g., {"window": "2h"}). - task_id (str): Unique task identifier. - - Returns: - timedelta: Parsed time interval. - - Raises: - Exception: If window is missing or has an invalid format. - """ - valid_units: dict = { - "s": "seconds", - "min": "minutes", - "h": "hours", - "d": "days", - "w": "weeks", - } - - window: str | None = args.get("window") - - match = re.fullmatch(r"(\d+)([a-zA-Z]+)", window) - if match: - number, unit = match.groups() - number = int(number) - if number >= 1 and unit in valid_units: - return timedelta(**{valid_units[unit]: number}) - - raise Exception(f"[{task_id}] Invalid interval format: {window}.") - - -def build_query(measurement: str, start_time: datetime, end_time: datetime) -> str: - """ - Builds an SQL query to select all data from a measurement within a time range. - - Args: - measurement (str): Name of the measurement/table. - start_time (datetime): Start time (inclusive). - end_time (datetime): End time (exclusive). - - Returns: - str: SQL query string selecting all data between start_time and end_time. - """ - 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") - - query = f""" - SELECT * - FROM '{measurement}' - WHERE time >= '{start_iso}' - AND time < '{end_iso}' - ORDER BY time ASC - """ - return query - - -def process_scheduled_call( - influxdb3_local, call_time: datetime, args: dict | None = None -) -> None: - """ - Entry point for the InfluxDB scheduler plugin that monitors field changes - in a measurement and triggers notifications if a threshold is exceeded. - - It queries a specified measurement within a time window, detects how many times certain - fields have changed, and sends notifications if those changes exceed - predefined thresholds. - - Args: - influxdb3_local: Instance of the InfluxDB client used for querying - and logging. - call_time (datetime): The UTC timestamp at which the scheduler triggers - this function. This defines the end of the time window. - args (dict, optional): Dictionary containing the following required keys: - - config_file_path (str): path to config file to override args. - - measurement (str): The name of the measurement to query. - - field_change_count (dict): Mapping of field names to change thresholds. - - senders (dict): Configuration for notification senders. - - window (str | int): Duration to look back from `call_time`. - - influxdb3_auth_token (str, optional): Token for authentication. - - notification_path (str, optional): Endpoint path for notifications. - - notification_text (str, optional): Template for the alert message. - - port_override (int, optional): Custom port for notification endpoint. - - Raises: - No exceptions are raised directly; all errors are caught and logged. + influxdb3_local: InfluxDB client instance used for querying and logging. + call_time (datetime): UTC timestamp of the scheduled run; the end of the window. + args (dict): Runtime arguments of the trigger. """ task_id: str = str(uuid.uuid4()) - influxdb3_local.info(f"[{task_id}] Starting scheduled field change check at {call_time} 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 - - # Check for required arguments - if ( - not args - or "measurement" not in args - or "field_change_count" not in args - or "senders" not in args - or "window" not in args - ): - influxdb3_local.error( - f"[{task_id}] Missing required arguments: measurement, field_change_count, senders, or window" - ) + influxdb3_local.info( + f"[{task_id}] Starting scheduled field change check at {call_time}" + ) + + 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" @@ -1323,53 +1106,55 @@ def process_scheduled_call( return try: - # Extract and validate parameters - field_counts: dict = parse_field_change_count(influxdb3_local, args, task_id) + field_counts: dict = parse_field_change_count(influxdb3_local, config, task_id) influxdb3_local.info(f"[{task_id}] Field change counts: {field_counts}") - senders_config: dict = parse_senders(influxdb3_local, args, task_id) - tags: list = get_tag_names(influxdb3_local, measurement, task_id) - window: timedelta = parse_window(args, task_id) - notification_path: str = args.get("notification_path", "notify") - port_override: int = parse_port_override(args, task_id) - influxdb3_auth_token: str = args.get("influxdb3_auth_token") or os.getenv( - "INFLUXDB3_AUTH_TOKEN" + senders_config: dict = parse_senders(influxdb3_local, config, task_id) + 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", - "Field $field in table $table changed $changes times in window $window for tags $tags", - ) + notification_tpl: str = config["notification_text"] - # Calculate time range + tags: list = get_measurement_tags(influxdb3_local, measurement, task_id) + window: timedelta = config["window"] end_time: datetime = call_time.replace(tzinfo=timezone.utc) start_time: datetime = end_time - window - influxdb3_local.info(f"[{task_id}] Querying '{measurement}' from {start_time} to {end_time}") + influxdb3_local.info( + f"[{task_id}] Querying '{measurement}' from {start_time} to {end_time}" + ) - # Build query to get data - query: str = build_query(measurement, start_time, end_time) - results: list = influxdb3_local.query(query) + results: list = query_window( + influxdb3_local, + measurement, + start=start_time.strftime("%Y-%m-%dT%H:%M:%SZ"), + end=end_time.strftime("%Y-%m-%dT%H:%M:%SZ"), + ) if not results: influxdb3_local.info( f"[{task_id}] No data found in '{measurement}' from {start_time} to {end_time}." ) return - influxdb3_local.info(f"[{task_id}] Retrieved {len(results)} records from {measurement}") + influxdb3_local.info( + f"[{task_id}] Retrieved {len(results)} records from {measurement}" + ) # Group data by unique tag combinations tag_combinations = defaultdict(list) for row in results: - tag_values = tuple(row[tag] if tag in row else "None" for tag in tags) + tag_values = tuple(row.get(tag, "None") for tag in tags) tag_combinations[tag_values].append(row) - # Process each tag combination for tag_values, rows in tag_combinations.items(): for field, count_threshold in field_counts.items(): - # Count changes changes: int = 0 prev_value = None for row in rows: @@ -1382,9 +1167,8 @@ def process_scheduled_call( if changes >= count_threshold: influxdb3_local.error( - f"[{task_id}] Found {count_threshold} changes in field '{field}' for tags {tag_values}, sending alert..." + f"[{task_id}] Found {changes} changes (threshold {count_threshold}) in field '{field}' for tags {tag_values}, sending alert..." ) - # Send notification tag_str = ", ".join( f"{tag}={value}" for tag, value in zip(tags, tag_values) ) @@ -1411,4 +1195,4 @@ def process_scheduled_call( ) except Exception as e: - influxdb3_local.error(f"[{task_id}] Error: {str(e)}") + influxdb3_local.error(f"[{task_id}] Error: {str(e)}") \ No newline at end of file diff --git a/influxdata/state_change/state_change_config_data_writes.toml b/influxdata/state_change/state_change_config_data_writes.toml index 5162293..dcde7f7 100644 --- a/influxdata/state_change/state_change_config_data_writes.toml +++ b/influxdata/state_change/state_change_config_data_writes.toml @@ -14,8 +14,9 @@ measurement = "your_measurement" # e.g., "cpu", "temperature", "home" # Format: [[field, value, count_or_duration], ...] # - field: field name (string) # - value: threshold value (number or boolean) -# - count_or_duration: integer (count-based) or duration string (e.g., "10s") for duration-based triggering -# Supported units for duration: s (seconds), min (minutes), h (hours), d (days), w (weeks) +# - count_or_duration: integer >= 1 (count-based) or duration string (e.g., "10s") for duration-based triggering +# Supported units for duration: us (microseconds), ms (milliseconds), s (seconds), +# min (minutes), h (hours), d (days), w (weeks) field_thresholds = [["field1", 0.0, "your_count_or_duration"]] # e.g., [["temp", 30, 1], ["temp", 15, "10s"]] # Notification channels @@ -29,10 +30,10 @@ senders = ["your_channel"] # e.g., ["slack"], ["sms", "http"] # Stability check parameters # Number of recent values to assess stability -# Specify an integer ≥ 1; default is 1 +# The check applies only when this is 2 or greater; default is 1 #state_change_window = 1 # e.g., 5 -# Maximum allowed changes in that window +# Number of changes in that window at which notifications start being suppressed # Specify an integer ≥ 1; default is 1 #state_change_count = 1 # e.g., 2 diff --git a/influxdata/state_change/state_change_config_scheduler.toml b/influxdata/state_change/state_change_config_scheduler.toml index 6d7c90d..cb607c6 100644 --- a/influxdata/state_change/state_change_config_scheduler.toml +++ b/influxdata/state_change/state_change_config_scheduler.toml @@ -12,16 +12,17 @@ measurement = "your_measurement" # e.g., "cpu", "temperature", "home" # Field change thresholds per field # Format: {field = count, ...} # - field: field name (string) -# - count: number of changes to trigger notification (integer) -field_change_count = {field1 = 0} # e.g., {temp = 3, hum = 2} +# - count: number of changes to trigger notification (integer >= 1) +field_change_count = {field1 = 1} # e.g., {temp = 3, hum = 2, "disk.used" = 1} # Notification channels # Specify a list of notification channels (strings) senders = ["your_channel"] # e.g., ["slack"], ["slack", "http"] # Analysis window duration -# Format: , where unit is s (seconds), min (minutes), h (hours), d (days), w (weeks) -window = "your_window" # e.g., "10m", "24h" +# Format: , where unit is us (microseconds), ms (milliseconds), s (seconds), +# min (minutes), h (hours), d (days), w (weeks). Must be a positive duration. +window = "your_window" # e.g., "10min", "24h" ########## Optional Parameters ########## # InfluxDB 3 API token for notifications diff --git a/influxdata/state_change/test_state_change.py b/influxdata/state_change/test_state_change.py new file mode 100644 index 0000000..c4a9398 --- /dev/null +++ b/influxdata/state_change/test_state_change.py @@ -0,0 +1,665 @@ +import json +from collections import deque +from datetime import datetime, timedelta, timezone + +import pytest + +import state_change_check_plugin as plugin + +TOKEN = "apiv3_secret_token_value" +WEBHOOK = "https://example.com/hook" + + +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=("home",), 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 raise_for_status(self): + return None + + 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 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": "home", + "field_thresholds": "temp:30:2", + "senders": "http", + "http_webhook_url": WEBHOOK, + "influxdb3_auth_token": TOKEN, +} +SCHEDULED_ARGS = { + "measurement": "home", + "field_change_count": "value:2", + "senders": "http", + "http_webhook_url": WEBHOOK, + "influxdb3_auth_token": TOKEN, + "window": "1h", +} + + +def batch(rows, table="home"): + return [{"table_name": table, "rows": rows}] + + +def count_key(field="temp", value=30, host="a"): + return plugin.generate_cache_key( + "home", field, value, "count", ["host"], {"host": host} + ) + + +def time_key(field="temp", value=30, host="a"): + return plugin.generate_cache_key( + "home", field, value, "time", ["host"], {"host": host} + ) + + +# --- parsing ---------------------------------------------------------------- + + +@pytest.mark.parametrize( + "raw, expected", + [ + ("10", 10), + (10, 10), + ("2h", timedelta(hours=2)), + ("500ms", timedelta(milliseconds=500)), + ("5min", timedelta(minutes=5)), + ], +) +def test_threshold_param_accepts_counts_and_durations(raw, expected): + assert plugin._parse_threshold_param(FakeInfluxdb3Local(), raw, "tid") == expected + + +@pytest.mark.parametrize("raw", ["0", "-5", "0s", "abc", True, "2x"]) +def test_threshold_param_rejects_invalid_values(raw): + influxdb3_local = FakeInfluxdb3Local() + + assert plugin._parse_threshold_param(influxdb3_local, raw, "tid") is None + assert influxdb3_local.messages("warn") + + +def test_field_thresholds_from_string(): + thresholds = plugin.parse_field_thresholds( + FakeInfluxdb3Local(), + {"field_thresholds": "temp:'30.1':10@humidity:'true':2h"}, + "tid", + ) + + assert thresholds == [ + ("temp", 30.1, 10), + ("humidity", True, timedelta(hours=2)), + ] + + +def test_field_thresholds_from_toml_entries(): + thresholds = plugin.parse_field_thresholds( + FakeInfluxdb3Local(), + {"field_thresholds": [["temp", 30, 1], ["status", "error", "10s"]]}, + "tid", + ) + + assert thresholds == [ + ("temp", 30, 1), + ("status", "error", timedelta(seconds=10)), + ] + + +def test_field_thresholds_skips_malformed_segments(): + influxdb3_local = FakeInfluxdb3Local() + + thresholds = plugin.parse_field_thresholds( + influxdb3_local, {"field_thresholds": "temp:30@humidity:5:1"}, "tid" + ) + + assert thresholds == [("humidity", 5, 1)] + assert any("must have exactly 2 colons" in m for m in influxdb3_local.messages("warn")) + + +def test_field_thresholds_without_valid_entries_raises(): + with pytest.raises(Exception, match="No valid field thresholds"): + plugin.parse_field_thresholds( + FakeInfluxdb3Local(), {"field_thresholds": "temp:30"}, "tid" + ) + + +def test_field_thresholds_rejects_unsupported_type(): + with pytest.raises(Exception, match="must be a list of entries or a string"): + plugin.parse_field_thresholds( + FakeInfluxdb3Local(), {"field_thresholds": 42}, "tid" + ) + + +def test_field_change_count_from_string(): + assert plugin.parse_field_change_count( + FakeInfluxdb3Local(), {"field_change_count": "temp:3.load:2"}, "tid" + ) == {"temp": 3, "load": 2} + + +def test_field_change_count_from_toml_mapping(): + assert plugin.parse_field_change_count( + FakeInfluxdb3Local(), {"field_change_count": {"temp": 3}}, "tid" + ) == {"temp": 3} + + +@pytest.mark.parametrize( + "raw, expected", + [ + ("disk.used:2", {"disk.used": 2}), + ("temp:3.disk.used:2.a.b.c:10", {"temp": 3, "disk.used": 2, "a.b.c": 10}), + ("sensor.1.value:5", {"sensor.1.value": 5}), + ("temp : 3", {"temp": 3}), + ], +) +def test_field_change_count_accepts_dotted_field_names(raw, expected): + """A pair ends at the dot after the count, so field names may contain dots.""" + influxdb3_local = FakeInfluxdb3Local() + + assert ( + plugin.parse_field_change_count( + influxdb3_local, {"field_change_count": raw}, "tid" + ) + == expected + ) + assert influxdb3_local.messages("warn") == [] + + +@pytest.mark.parametrize( + "raw, expected", + [("temp:abc.load:2", {"load": 2}), ("temp:0.load:2", {"load": 2})], +) +def test_field_change_count_skips_invalid_pairs(raw, expected): + influxdb3_local = FakeInfluxdb3Local() + + assert ( + plugin.parse_field_change_count( + influxdb3_local, {"field_change_count": raw}, "tid" + ) + == expected + ) + assert influxdb3_local.messages("warn") + + +def test_field_change_count_without_valid_entries_raises(): + with pytest.raises(Exception, match="No valid entries"): + plugin.parse_field_change_count( + FakeInfluxdb3Local(), {"field_change_count": "temp:0"}, "tid" + ) + + +@pytest.mark.parametrize( + "raw, expected", + [("1h", timedelta(hours=1)), ("30s", timedelta(seconds=30))], +) +def test_parse_window(raw, expected): + assert plugin.parse_window(raw) == expected + + +@pytest.mark.parametrize("raw", ["0s", "10m", "abc"]) +def test_parse_window_rejects_invalid(raw): + with pytest.raises(ValueError): + plugin.parse_window(raw) + + +# --- senders ---------------------------------------------------------------- + + +def test_senders_from_string_and_list(): + config = {"senders": "http.slack", "http_webhook_url": WEBHOOK, "slack_webhook_url": WEBHOOK} + + from_string = plugin.parse_senders(FakeInfluxdb3Local(), config, "tid") + from_list = plugin.parse_senders( + FakeInfluxdb3Local(), {**config, "senders": ["http", "slack"]}, "tid" + ) + + assert set(from_string) == {"http", "slack"} == set(from_list) + + +def test_senders_skips_unknown_channel(): + influxdb3_local = FakeInfluxdb3Local() + + senders = plugin.parse_senders( + influxdb3_local, + {"senders": "telegram.http", "http_webhook_url": WEBHOOK}, + "tid", + ) + + assert set(senders) == {"http"} + assert any("Invalid sender type: telegram" in m for m in influxdb3_local.messages("warn")) + + +def test_senders_requires_webhook_url(): + with pytest.raises(Exception, match="No valid senders"): + plugin.parse_senders(FakeInfluxdb3Local(), {"senders": "http"}, "tid") + + +def test_senders_rejects_non_http_scheme(): + influxdb3_local = FakeInfluxdb3Local() + + with pytest.raises(Exception, match="No valid senders"): + plugin.parse_senders( + influxdb3_local, + {"senders": "http", "http_webhook_url": "ftp://example.com/hook"}, + "tid", + ) + assert any("must start with" in m for m in influxdb3_local.messages("error")) + + +# --- configuration ---------------------------------------------------------- + + +def test_load_config_applies_defaults(plugin_dir): + config = plugin._load_config( + FakeInfluxdb3Local(), dict(WRITES_ARGS), plugin._WRITES_VALIDATORS, "tid" + ) + + assert config["port_override"] == 8181 + assert config["notification_path"] == "notify" + assert config["state_change_window"] == 1 + assert config["state_change_count"] == 1 + + +def test_load_config_reads_toml(plugin_dir): + (plugin_dir / "writes.toml").write_text( + "measurement = 'home'\n" + "senders = ['http']\n" + "http_webhook_url = 'https://example.com/hook'\n" + "influxdb3_auth_token = 'tok'\n" + "field_thresholds = [['temp', 30, 1]]\n" + "state_change_window = 4\n" + ) + + config = plugin._load_config( + FakeInfluxdb3Local(), + {"config_file_path": "writes.toml"}, + plugin._WRITES_VALIDATORS, + "tid", + ) + + assert config["senders"] == ["http"] + assert config["field_thresholds"] == [["temp", 30, 1]] + assert config["state_change_window"] == 4 + + +def test_load_config_rejects_non_toml_path(plugin_dir): + influxdb3_local = FakeInfluxdb3Local() + + assert ( + plugin._load_config( + influxdb3_local, + {"config_file_path": "config.yaml"}, + plugin._WRITES_VALIDATORS, + "tid", + ) + is None + ) + assert any("expected a .toml file" in m for m in influxdb3_local.messages("error")) + + +@pytest.mark.parametrize( + "override", + [{"measurement": None}, {"state_change_window": "-1"}, {"port_override": "70000"}], +) +def test_load_config_reports_validation_failures(plugin_dir, override): + args = {**WRITES_ARGS, **override} + args = {key: value for key, value in args.items() if value is not None} + influxdb3_local = FakeInfluxdb3Local() + + assert ( + plugin._load_config( + influxdb3_local, args, plugin._WRITES_VALIDATORS, "tid" + ) + is None + ) + assert any("Failed to load configuration" in m for m in influxdb3_local.messages("error")) + + +def test_load_config_accepts_zero_stability_settings(plugin_dir): + """Existing triggers may pass 0; it behaves like the default of 1.""" + args = {**WRITES_ARGS, "state_change_window": "0", "state_change_count": "0"} + + config = plugin._load_config( + FakeInfluxdb3Local(), args, plugin._WRITES_VALIDATORS, "tid" + ) + + assert config["state_change_window"] == 0 + assert config["state_change_count"] == 0 + + +def test_writes_with_zero_window_keeps_alerting(plugin_dir, sent): + influxdb3_local = FakeInfluxdb3Local() + args = {**WRITES_ARGS, "field_thresholds": "temp:30:1", "state_change_window": "0"} + + plugin.process_writes(influxdb3_local, batch([{"host": "a", "temp": 30}]), args) + + assert len(sent) == 1 + + +def test_load_config_uses_token_from_environment(plugin_dir, monkeypatch): + monkeypatch.setenv("INFLUXDB3_AUTH_TOKEN", "env-token") + args = {key: value for key, value in WRITES_ARGS.items() if key != "influxdb3_auth_token"} + + config = plugin._load_config( + FakeInfluxdb3Local(), args, plugin._WRITES_VALIDATORS, "tid" + ) + + assert config["influxdb3_auth_token"] == "env-token" + + +# --- stability and counters ------------------------------------------------- + + +@pytest.mark.parametrize( + "values, allowed, stable", + [ + ([], 1, True), + ([1], 1, True), + ([1, 1, 1], 1, True), + ([1, 2, 1], 2, False), + ([1, 2, 2], 2, True), + ], +) +def test_check_state_changes(values, allowed, stable): + assert plugin.check_state_changes(deque(values), allowed) is stable + + +@pytest.mark.parametrize("stored, expected", [("3", 3), ("", 0), (None, 0), ("x", 0)]) +def test_read_counter_tolerates_unusable_values(stored, expected): + influxdb3_local = FakeInfluxdb3Local() + if stored is not None: + influxdb3_local.cache.put("key", stored) + + assert plugin.read_counter(influxdb3_local, "key") == expected + + +# --- process_writes --------------------------------------------------------- + + +def test_writes_count_threshold_alerts_after_enough_matches(plugin_dir, sent): + influxdb3_local = FakeInfluxdb3Local() + + plugin.process_writes( + influxdb3_local, batch([{"host": "a", "temp": 30}]), dict(WRITES_ARGS) + ) + assert sent == [] + assert influxdb3_local.cache.get(count_key()) == "1" + + plugin.process_writes( + influxdb3_local, batch([{"host": "a", "temp": 30}]), dict(WRITES_ARGS) + ) + assert len(sent) == 1 + assert "changed to 30" in sent[0]["payload"]["notification_text"] + assert sent[0]["payload"]["senders_config"] == {"http": {"http_webhook_url": WEBHOOK}} + assert influxdb3_local.cache.get(count_key()) == "0" + + +def test_writes_count_resets_when_condition_fails(plugin_dir, sent): + influxdb3_local = FakeInfluxdb3Local() + + plugin.process_writes( + influxdb3_local, + batch([{"host": "a", "temp": 30}, {"host": "a", "temp": 25}]), + dict(WRITES_ARGS), + ) + + assert sent == [] + assert influxdb3_local.cache.get(count_key()) == "0" + + +def test_writes_missing_field_does_not_break_the_batch(plugin_dir, sent): + """Regression: an absent field used to store '' and crash the next int() read.""" + influxdb3_local = FakeInfluxdb3Local() + + plugin.process_writes( + influxdb3_local, + batch([{"host": "a", "hum": 5}, {"host": "a", "temp": 30}]), + dict(WRITES_ARGS), + ) + + assert influxdb3_local.messages("error") == [] + assert influxdb3_local.cache.get(count_key()) == "1" + + +def test_writes_duration_threshold_alerts_once_elapsed(plugin_dir, sent): + influxdb3_local = FakeInfluxdb3Local() + args = {**WRITES_ARGS, "field_thresholds": "temp:30:1h"} + started = datetime.now(timezone.utc) - timedelta(hours=2) + influxdb3_local.cache.put(time_key(), started.isoformat()) + + plugin.process_writes(influxdb3_local, batch([{"host": "a", "temp": 30}]), args) + + assert len(sent) == 1 + assert influxdb3_local.cache.get(time_key()) == "" + + +def test_writes_duration_threshold_waits_and_keeps_start(plugin_dir, sent): + influxdb3_local = FakeInfluxdb3Local() + args = {**WRITES_ARGS, "field_thresholds": "temp:30:1h"} + started = (datetime.now(timezone.utc) - timedelta(minutes=5)).isoformat() + influxdb3_local.cache.put(time_key(), started) + + plugin.process_writes(influxdb3_local, batch([{"host": "a", "temp": 30}]), args) + + assert sent == [] + assert influxdb3_local.cache.get(time_key()) == started + assert any("Condition still holding" in m for m in influxdb3_local.messages("warn")) + + +def test_writes_unstable_data_suppresses_notification(plugin_dir, sent): + influxdb3_local = FakeInfluxdb3Local() + args = { + **WRITES_ARGS, + "field_thresholds": "temp:30:1", + "state_change_window": "3", + "state_change_count": "2", + } + values_key = plugin.generate_cache_key( + "home", "temp", 30, "values", ["host"], {"host": "a"} + ) + influxdb3_local.cache.put(values_key, deque([25, 30, 25], maxlen=3)) + + plugin.process_writes(influxdb3_local, batch([{"host": "a", "temp": 30}]), args) + + assert sent == [] + assert any("unstable data state" in m for m in influxdb3_local.messages("warn")) + + +def test_writes_ignores_batches_of_other_tables(plugin_dir, sent): + influxdb3_local = FakeInfluxdb3Local() + + plugin.process_writes( + influxdb3_local, batch([{"host": "a", "temp": 30}], table="cpu"), dict(WRITES_ARGS) + ) + + assert sent == [] + assert not any("Starting writes process" in m for m in influxdb3_local.messages()) + assert not any("information_schema" in q for q in influxdb3_local.queries) + + +def test_writes_reports_unknown_measurement(plugin_dir, sent): + influxdb3_local = FakeInfluxdb3Local(tables=("cpu",)) + + plugin.process_writes( + influxdb3_local, batch([{"host": "a", "temp": 30}]), dict(WRITES_ARGS) + ) + + assert any("not found in database" in m for m in influxdb3_local.messages("error")) + + +def test_writes_caches_configuration(plugin_dir, sent): + influxdb3_local = FakeInfluxdb3Local() + + plugin.process_writes( + influxdb3_local, batch([{"host": "a", "temp": 30}]), dict(WRITES_ARGS) + ) + + assert influxdb3_local.cache.get(plugin._WRITES_CONFIG_CACHE_KEY) is not None + assert ( + influxdb3_local.cache.ttls[plugin._WRITES_CONFIG_CACHE_KEY] + == plugin._WRITES_CONFIG_TTL_SECONDS + ) + + +def test_writes_never_logs_credentials(plugin_dir, sent): + influxdb3_local = FakeInfluxdb3Local() + + plugin.process_writes( + influxdb3_local, batch([{"host": "a", "temp": 30}]), dict(WRITES_ARGS) + ) + + logged = " ".join(influxdb3_local.messages()) + assert TOKEN not in logged + assert WEBHOOK not in logged + + +# --- process_scheduled_call ------------------------------------------------- + + +def test_scheduled_alerts_when_changes_reach_threshold(plugin_dir, sent): + rows = [ + {"host": "a", "value": 1}, + {"host": "a", "value": 2}, + {"host": "a", "value": 1}, + ] + influxdb3_local = FakeInfluxdb3Local(rows=rows) + + plugin.process_scheduled_call( + influxdb3_local, datetime(2026, 8, 16, 12, 0, 0), dict(SCHEDULED_ARGS) + ) + + assert len(sent) == 1 + assert "changed 2 times" in sent[0]["payload"]["notification_text"] + assert any( + "Found 2 changes (threshold 2)" in m for m in influxdb3_local.messages("error") + ) + + +def test_scheduled_stays_quiet_below_threshold(plugin_dir, sent): + rows = [{"host": "a", "value": 1}, {"host": "a", "value": 2}] + influxdb3_local = FakeInfluxdb3Local(rows=rows) + args = {**SCHEDULED_ARGS, "field_change_count": "value:5"} + + plugin.process_scheduled_call( + influxdb3_local, datetime(2026, 8, 16, 12, 0, 0), args + ) + + assert sent == [] + + +def test_scheduled_counts_changes_per_tag_combination(plugin_dir, sent): + rows = [ + {"host": "a", "value": 1}, + {"host": "a", "value": 2}, + {"host": "a", "value": 3}, + {"host": "b", "value": 7}, + ] + influxdb3_local = FakeInfluxdb3Local(rows=rows) + + plugin.process_scheduled_call( + influxdb3_local, datetime(2026, 8, 16, 12, 0, 0), dict(SCHEDULED_ARGS) + ) + + assert len(sent) == 1 + assert "host=a" in sent[0]["payload"]["notification_text"] + + +def test_scheduled_handles_empty_window(plugin_dir, sent): + influxdb3_local = FakeInfluxdb3Local(rows=[]) + + plugin.process_scheduled_call( + influxdb3_local, datetime(2026, 8, 16, 12, 0, 0), dict(SCHEDULED_ARGS) + ) + + assert sent == [] + assert any("No data found" in m for m in influxdb3_local.messages("info")) + + +def test_scheduled_queries_the_configured_window(plugin_dir, sent): + influxdb3_local = FakeInfluxdb3Local(rows=[]) + + plugin.process_scheduled_call( + influxdb3_local, datetime(2026, 8, 16, 12, 0, 0), dict(SCHEDULED_ARGS) + ) + + data_query = influxdb3_local.queries[-1] + assert '"home"' in data_query + assert "time >= $start AND time < $end" in data_query + assert any("from 2026-08-16 11:00:00+00:00" in m for m in influxdb3_local.messages("info")) + + +def test_scheduled_never_logs_credentials(plugin_dir, sent): + influxdb3_local = FakeInfluxdb3Local(rows=[{"host": "a", "value": 1}]) + + plugin.process_scheduled_call( + influxdb3_local, datetime(2026, 8, 16, 12, 0, 0), dict(SCHEDULED_ARGS) + ) + + logged = " ".join(influxdb3_local.messages()) + assert TOKEN not in logged + assert WEBHOOK not in logged