diff --git a/influxdata/library/plugin_library.json b/influxdata/library/plugin_library.json index 9d9129f..d30d7ec 100644 --- a/influxdata/library/plugin_library.json +++ b/influxdata/library/plugin_library.json @@ -122,8 +122,8 @@ "path": "influxdata/notifier/notifier_plugin.py" } ], - "required_libraries": ["requests", "adtk", "pandas"], - "last_update": "2025-06-16", + "required_libraries": ["influxdata-plugin-utils>=0.3.0", "requests", "adtk", "pandas<3"], + "last_update": "2026-08-16", "trigger_types_supported": ["scheduler"] }, { diff --git a/influxdata/stateless_adtk_detector/README.md b/influxdata/stateless_adtk_detector/README.md index 2318482..4cbaeb7 100644 --- a/influxdata/stateless_adtk_detector/README.md +++ b/influxdata/stateless_adtk_detector/README.md @@ -25,15 +25,33 @@ This plugin includes a JSON metadata schema in its docstring that defines suppor | `field` | string | required | Numeric field to evaluate | | `detectors` | string | required | Dot-separated list of advanced ADTK detectors for different anomaly types | | `detector_params` | string | required | Base64-encoded JSON parameters for each detector | -| `window` | string | required | Data analysis window with flexible scheduling. Format: `` (e.g., "1h", "30m") | +| `window` | string | required | Data analysis window. Format: `` (e.g., "1h", "30min"). Must be positive | | `senders` | string | required | Dot-separated notification channels with multi-channel notification support | +Duration units: `us`, `ms`, `s`, `min`, `h`, `d`, `w`. + ### Advanced parameters -| Parameter | Type | Default | Description | -|--------------------------|--------|---------|----------------------------------------------------------------------------------------------| -| `min_consensus` | number | 1 | Minimum detectors required to agree for consensus-based filtering to reduce false positives | -| `min_condition_duration` | string | "0s" | Minimum duration for configurable anomaly persistence before alerting | +| Parameter | Type | Default | Description | +|-----------------------------|---------|----------|------------------------------------------------------------------------------------------------------------| +| `min_consensus` | number | 1 | Minimum detectors required to agree for consensus-based filtering to reduce false positives (1 or greater) | +| `min_condition_duration` | string | "0s" | Minimum duration for configurable anomaly persistence before alerting | +| `group_by_tags` | bool | false | Analyze every tag combination as its own time series | +| `max_notifications_per_run` | number | 20 | Maximum number of notifications a single run may send | + +#### Analyzing tagged measurements + +By default the whole window forms a single time series. When a measurement holds several tag combinations (for example `host=server1` and `host=server2`), those rows share timestamps, and ADTK keeps only the first value of each timestamp — so one arbitrary series is analyzed and the rest of the data is ignored. + +Set `group_by_tags=true` to analyze each tag combination separately. Every series then gets its own detector run, its own consensus evaluation, and its own debounce state, so a measurement with N tag combinations can produce up to N notifications per run. + +#### Notification behavior + +The timestamp of the last alerted point is remembered per series, so a `window` longer than the trigger interval does not resend anomalies that earlier runs already reported. + +A single run sends at most `max_notifications_per_run` notifications. Anomalies beyond the limit are counted in a warning and are not resent by later runs — raise the limit if a run legitimately produces more alerts. + +Points that have no value for `field` are dropped before detection and reported in the log; without this a single NULL makes every detector fail. ### Notification parameters @@ -52,6 +70,8 @@ This plugin includes a JSON metadata schema in its docstring that defines suppor *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. +When a config file is given, it replaces the inline trigger arguments entirely. In TOML, `detectors` and `senders` accept either a list (`["QuantileAD", "PersistAD"]`) or a dot-separated string, and detector parameters may be given as a `[detector_params]` table or as a base64-encoded JSON string. + #### Example TOML configuration [adtk_anomaly_config_scheduler.toml](adtk_anomaly_config_scheduler.toml) @@ -74,10 +94,17 @@ For more information on using TOML configuration files, see the Using TOML Confi ## Software Requirements - **InfluxDB 3 Core/Enterprise**: with the Processing Engine enabled. +- **Python 3.11+** - **Python packages**: + - `influxdata-plugin-utils>=0.3.0` (for configuration loading, parsing, and schema introspection) - `adtk` (for anomaly detection) - - `pandas` (for data manipulation) + - `pandas<3` (for data manipulation) - `requests` (for HTTP notifications) + +`pandas` must stay below 3.0. Window-based detectors (`LevelShiftAD`, `VolatilityShiftAD`, `PersistAD`) +return `NaN` for the first `window` points, which pandas 3 refuses to store in a boolean result. With +pandas 3 installed those three detectors fail with `Invalid value 'nan' for dtype 'bool'`, are skipped +with a warning, and stop contributing to the consensus. - **Notification Sender Plugin** *(optional)*: Required if using the `senders` parameter. See the [influxdata/notifier plugin](../notifier/README.md). ### Installation steps @@ -95,9 +122,10 @@ For more information on using TOML configuration files, see the Using TOML Confi 2. Install required Python packages: ```bash + influxdb3 install package "influxdata-plugin-utils>=0.3.0" influxdb3 install package requests influxdb3 install package adtk - influxdb3 install package pandas + influxdb3 install package "pandas<3" ``` 3. *(Optional)* For notifications, install the [influxdata/notifier plugin](../notifier/README.md) and create an HTTP trigger for it. @@ -113,7 +141,7 @@ influxdb3 create trigger \ --database mydb \ --path "gh:influxdata/stateless_adtk_detector/adtk_anomaly_detection_plugin.py" \ --trigger-spec "every:10m" \ - --trigger-arguments "measurement=cpu,field=usage,detectors=QuantileAD.LevelShiftAD,detector_params=eyJRdWFudGlsZUFKIjogeyJsb3ciOiAwLjA1LCAiaGlnaCI6IDAuOTV9LCAiTGV2ZWxTaGlmdEFKIjogeyJ3aW5kb3ciOiA1fX0=,window=10m,senders=slack,slack_webhook_url=$SLACK_WEBHOOK_URL" \ + --trigger-arguments "measurement=cpu,field=usage,detectors=QuantileAD.LevelShiftAD,detector_params=eyJRdWFudGlsZUFKIjogeyJsb3ciOiAwLjA1LCAiaGlnaCI6IDAuOTV9LCAiTGV2ZWxTaGlmdEFKIjogeyJ3aW5kb3ciOiA1fX0=,window=10min,senders=slack,slack_webhook_url=$SLACK_WEBHOOK_URL" \ anomaly_detector ``` @@ -157,7 +185,7 @@ influxdb3 create trigger \ --database monitoring \ --path "gh:influxdata/stateless_adtk_detector/adtk_anomaly_detection_plugin.py" \ --trigger-spec "every:15m" \ - --trigger-arguments "measurement=cpu_metrics,field=utilization,detectors=QuantileAD.LevelShiftAD,detector_params=eyJRdWFudGlsZUFEIjogeyJsb3ciOiAwLjEsICJoaWdoIjogMC45fSwgIkxldmVsU2hpZnRBRCI6IHsid2luZG93IjogMTB9fQ==,min_consensus=2,window=30m,senders=discord,discord_webhook_url=$DISCORD_WEBHOOK_URL" \ + --trigger-arguments "measurement=cpu_metrics,field=utilization,detectors=QuantileAD.LevelShiftAD,detector_params=eyJRdWFudGlsZUFEIjogeyJsb3ciOiAwLjEsICJoaWdoIjogMC45fSwgIkxldmVsU2hpZnRBRCI6IHsid2luZG93IjogMTB9fQ==,min_consensus=2,window=30min,senders=discord,discord_webhook_url=$DISCORD_WEBHOOK_URL" \ cpu_consensus_detector ``` @@ -175,7 +203,7 @@ influxdb3 create trigger \ --database trading \ --path "gh:influxdata/stateless_adtk_detector/adtk_anomaly_detection_plugin.py" \ --trigger-spec "every:1m" \ - --trigger-arguments "measurement=stock_prices,field=price,detectors=VolatilityShiftAD,detector_params=eyJWb2xhdGlsaXR5U2hpZnRBRCI6IHsid2luZG93IjogMjB9fQ==,window=1h,min_condition_duration=5m,senders=sms,twilio_from_number=+1234567890,twilio_to_number=+0987654321" \ + --trigger-arguments "measurement=stock_prices,field=price,detectors=VolatilityShiftAD,detector_params=eyJWb2xhdGlsaXR5U2hpZnRBRCI6IHsid2luZG93IjogMjB9fQ==,window=1h,min_condition_duration=5min,senders=sms,twilio_from_number=+1234567890,twilio_to_number=+0987654321" \ volatility_detector ``` @@ -186,6 +214,9 @@ influxdb3 create trigger \ - `adtk_anomaly_detection_plugin.py`: The main plugin code containing the scheduled handler for anomaly detection - `adtk_anomaly_config_scheduler.toml`: Example TOML configuration file +- `test_adtk_anomaly_detection.py`: Pytest suite (49 tests, runs without a live InfluxDB 3 server) +- `requirements.txt`: Runtime dependencies (`influxdata-plugin-utils>=0.3.0`, `requests`, `adtk`, `pandas<3`) +- `requirements-dev.txt`: Development dependencies (`pytest`) ### Logging @@ -209,6 +240,14 @@ Key operations: 4. Evaluates consensus across detectors 5. Sends notifications when anomalies are confirmed +#### `parse_detectors(influxdb3_local, config, task_id)` + +Resolves the detectors to apply together with their parameters. Detectors that are unknown, have no entry in `detector_params`, or miss a parameter required to construct them are skipped with a warning and do not count toward `min_consensus`. + +#### `split_by_tags(df, tags, group_by_tags)` + +Splits query results into one frame per tag combination when `group_by_tags` is enabled, so detectors never mix values written under different tag sets. + ## Troubleshooting ### Common issues @@ -221,9 +260,25 @@ Key operations: **Solution**: Increase `min_consensus` to require more detectors to agree. Add `min_condition_duration` to require anomalies to persist. Adjust detector-specific thresholds in `detector_params`. +#### Issue: A newly created measurement is reported as not found + +**Solution**: Table and tag names are cached for one hour per trigger. Wait for the cache to expire, or recreate the trigger to clear it. + +#### Issue: Anomalies of some tag combinations are never detected + +**Solution**: Set `group_by_tags=true`. Without it, rows of different tag combinations share timestamps and only the first series survives. + +#### Issue: A detector is skipped with a warning + +**Solution**: The warning names the reason: the detector is not in the supported list (check the spelling), has no entry in `detector_params`, or misses a required parameter (`window` for `LevelShiftAD` and `VolatilityShiftAD`). `Invalid value 'nan' for dtype 'bool'` means pandas 3 is installed — downgrade to `pandas<3`. + +#### Issue: No anomalies are ever reported + +**Solution**: Check the warnings. `min_consensus` must not exceed the number of detectors that were actually applied — skipped detectors reduce that count. `min_condition_duration` must be shorter than `window`, otherwise no anomaly can persist long enough within a single query window. + #### Issue: Missing dependencies -**Solution**: Install required packages: `adtk`, `pandas`, `requests`. Ensure the Notifier Plugin is installed for notifications. +**Solution**: Install required packages: `influxdata-plugin-utils`, `adtk`, `pandas`, `requests`. Ensure the Notifier Plugin is installed for notifications. #### Issue: Data quality issues diff --git a/influxdata/stateless_adtk_detector/adtk_anomaly_config_scheduler.toml b/influxdata/stateless_adtk_detector/adtk_anomaly_config_scheduler.toml index be16b92..7c789b1 100644 --- a/influxdata/stateless_adtk_detector/adtk_anomaly_config_scheduler.toml +++ b/influxdata/stateless_adtk_detector/adtk_anomaly_config_scheduler.toml @@ -16,15 +16,16 @@ field = "your_field" # e.g., "usage", "value", "temp" # ADTK detectors to use # Supported detectors: GeneralizedESDTestAD, InterQuartileRangeAD, ThresholdAD, QuantileAD, LevelShiftAD, VolatilityShiftAD, PersistAD, SeasonalAD -# Specify a list of detector names (strings) +# Specify a list of detector names (strings), or a dot-separated string detectors = ["your_detector"] # e.g., ["QuantileAD"], ["GeneralizedESDTestAD", "InterQuartileRangeAD"] # Time window for analysis -# Format: , where unit is s (seconds), min (minutes), h (hours), d (days), w (weeks) -window = "your_window" # e.g., "15m", "24h" +# Format: , where unit is us, ms, s (seconds), min (minutes), h (hours), d (days), w (weeks) +# Must be a positive duration +window = "your_window" # e.g., "15min", "24h" # Notification channels -# Specify a list of notification channels (strings) +# Specify a list of notification channels (strings), or a dot-separated string senders = ["your_channel"] # e.g., ["slack"], ["http", "sms"] ########## Optional Parameters ########## @@ -32,9 +33,17 @@ senders = ["your_channel"] # e.g., ["slack"], ["http", "sms"] # Specify an integer ≥ 1; default is 1 #min_consensus = 1 # e.g., 2 +# Analyze every tag combination as its own time series +# Without it the whole window is one series and rows sharing a timestamp are collapsed +#group_by_tags = true # default is false + +# Maximum number of notifications a single run may send +# Anomalies beyond the limit are reported in a warning and not resent later +#max_notifications_per_run = 20 # e.g., 50 + # Minimum duration an anomaly condition must persist before notifying -# Format: , where unit is s (seconds), min (minutes), h (hours), d (days), w (weeks) -#min_condition_duration = "your_duration" # e.g., "1m", "10m" +# Format: , where unit is us, ms, s (seconds), min (minutes), h (hours), d (days), w (weeks) +#min_condition_duration = "your_duration" # e.g., "1min", "10min" # InfluxDB 3 API token # Specify the token (string); can also be provided via INFLUXDB3_AUTH_TOKEN environment variable @@ -84,14 +93,12 @@ senders = ["your_channel"] # e.g., ["slack"], ["http", "sms"] #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" +# WhatsApp uses the same twilio_* settings as SMS above # Detector parameters (Required) # Format: {"DetectorName": {param1: value, ...}, ...} # Specify parameters for each detector listed in detectors +# A base64-encoded JSON string is also accepted: detector_params = "eyJRdWFudGlsZUFEIjogey4uLn19" [detector_params] your_detector = { param1 = "value1", param2 = "value2" } # e.g., QuantileAD = { low = 0.05, high = 0.95 } diff --git a/influxdata/stateless_adtk_detector/adtk_anomaly_detection_plugin.py b/influxdata/stateless_adtk_detector/adtk_anomaly_detection_plugin.py index 03548e7..ef1a050 100644 --- a/influxdata/stateless_adtk_detector/adtk_anomaly_detection_plugin.py +++ b/influxdata/stateless_adtk_detector/adtk_anomaly_detection_plugin.py @@ -17,31 +17,37 @@ { "name": "detectors", "example": "QuantileAD.LevelShiftAD", - "description": "Dot-separated list of ADTK detectors. Supported: GeneralizedESDTestAD, InterQuartileRangeAD, ThresholdAD, QuantileAD, LevelShiftAD, VolatilityShiftAD, PersistAD, SeasonalAD.", + "description": "Dot-separated list of ADTK detectors (a TOML config may use a list instead). Supported: GeneralizedESDTestAD, InterQuartileRangeAD, ThresholdAD, QuantileAD, LevelShiftAD, VolatilityShiftAD, PersistAD, SeasonalAD.", "required": true }, { "name": "detector_params", "example": "eyJRdWFudGlsZUFKIjogeyJsb3dfcXVhbnRpbGUiOiA...", - "description": "Base64-encoded JSON string specifying parameters for each detector.", + "description": "Base64-encoded JSON string specifying parameters for each detector (a TOML config may use a [detector_params] table instead).", "required": true }, { "name": "min_consensus", "example": "2", - "description": "Minimum number of detectors that must agree to flag a point as anomalous. Default: 1.", + "description": "Minimum number of detectors that must agree to flag a point as anomalous. Must be 1 or greater. Default: 1.", + "required": false + }, + { + "name": "group_by_tags", + "example": "true", + "description": "Analyze every tag combination as its own time series. When disabled (default), all rows of the window form a single series and rows sharing a timestamp are collapsed to the first one. Default: false.", "required": false }, { "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). Must be a positive duration. Units: `us`, `ms`, `s`, `min`, `h`, `d`, `w`.", "required": true }, { "name": "senders", "example": "slack.discord", - "description": "Dot-separated list of notification channels. Supported channels: slack, discord, http, sms, whatsapp.", + "description": "Dot-separated list of notification channels (a TOML config may use a list instead). Supported channels: slack, discord, http, sms, whatsapp.", "required": true }, { @@ -52,14 +58,20 @@ }, { "name": "min_condition_duration", - "example": "5m", - "description": "Minimum duration for an anomaly condition to persist before triggering a notification (e.g., `5m`). Units: `s`, `min`, `h`, `d`, `w`. Default: `0s`.", + "example": "5min", + "description": "Minimum duration for an anomaly condition to persist before triggering a notification (e.g., `5min`). Units: `us`, `ms`, `s`, `min`, `h`, `d`, `w`. Default: `0s`.", + "required": false + }, + { + "name": "max_notifications_per_run", + "example": "20", + "description": "Maximum number of notifications sent by a single run. Anomalies beyond the limit are counted in a warning and not resent later. Default: 20.", "required": false }, { "name": "notification_text", "example": "Anomaly detected in $table.$field with value $value by $detectors. Tags: $tags", - "description": "Template for notification message with variables `$table`, `$field`, `$value`, `$detectors`, `$tags`.", + "description": "Template for notification message with variables `$table`, `$field`, `$value`, `$detectors`, `$tags`, `$timestamp`.", "required": false }, { @@ -149,11 +161,9 @@ import os import random import time -import tomllib import uuid from collections import defaultdict -from datetime import datetime, timedelta, timezone -from pathlib import Path +from datetime import datetime, timedelta from string import Template from urllib.parse import urlparse @@ -170,6 +180,18 @@ ThresholdAD, VolatilityShiftAD, ) +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_bool, + parse_delimited_list, + parse_int, + parse_timedelta, +) # Supported sender types with their required arguments AVAILABLE_SENDERS = { @@ -200,51 +222,105 @@ "SeasonalAD": SeasonalAD, } +# Detectors that cannot be constructed without these parameters +REQUIRED_DETECTOR_PARAMS = { + "LevelShiftAD": ["window"], + "VolatilityShiftAD": ["window"], +} -def get_all_measurements(influxdb3_local) -> list[str]: - """ - Retrieves a list of all tables of type 'BASE TABLE' from the current InfluxDB database. +# Detectors that classify points on their own and must not be fitted +UNFITTED_DETECTORS = ("ThresholdAD",) - Args: - influxdb3_local: InfluxDB client instance. - - Returns: - list[str]: List of table names (e.g., ["cpu", "memory", "disk"]). - """ - result: list = influxdb3_local.query("SHOW TABLES") - return [ - row["table_name"] for row in result if row.get("table_type") == "BASE TABLE" - ] +_DEFAULT_NOTIFICATION_TEXT = ( + "Anomaly detected in $table.$field with value $value by $detectors. Tags: $tags" +) -def get_tag_names(influxdb3_local, measurement: str, task_id: str) -> list[str]: +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 + + +_VALIDATORS = [ + Validator("measurement", required=True, cast=str), + Validator("field", required=True, cast=str), + Validator( + "detectors", + required=True, + cast=lambda raw: parse_delimited_list(raw, sep="."), + ), + Validator("detector_params", required=True), + Validator("senders", required=True), + Validator("window", required=True, cast=parse_window), + Validator("min_consensus", default=1, cast=lambda raw: parse_int(raw, minimum=1)), + Validator("group_by_tags", default=False, cast=parse_bool), + Validator("min_condition_duration", default="0s", cast=parse_timedelta), + Validator( + "max_notifications_per_run", + default=20, + cast=lambda raw: parse_int(raw, minimum=1), + ), + Validator("notification_text", default=_DEFAULT_NOTIFICATION_TEXT, cast=str), + Validator("notification_path", default="notify", cast=str), + Validator( + "port_override", + default=8181, + cast=lambda raw: parse_int(raw, minimum=1, maximum=65535), + ), +] + + +def _load_config(influxdb3_local, args: dict, task_id: str) -> dict | None: """ - Retrieves the list of tag names for a measurement. + 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. - measurement (str): Name of the measurement to query. - task_id (str): The task ID. + args (dict): Runtime arguments of the trigger. + task_id (str): Unique task identifier. Returns: - list[str]: List of tag names with 'Dictionary(Int32, Utf8)' data type. + dict | None: Config values keyed by lower-case name, or None if loading failed. """ - 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}) + 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()} - 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: + # 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 [] - - tag_names: list[str] = [tag["column_name"] for tag in res] - return tag_names + return tags def generate_cache_key( @@ -269,15 +345,14 @@ def generate_cache_key( return base -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": dot-separated list of sender types (e.g., "slack.http"). - - For each sender, its own required keys (see AVAILABLE_SENDERS). + config (dict): Loaded config containing "senders" and the sender-specific + keys listed in AVAILABLE_SENDERS. task_id (str): Unique task identifier used for logging context. Returns: @@ -295,45 +370,37 @@ def parse_senders(influxdb3_local, args: dict, task_id: str) -> dict: 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 def send_notification( influxdb3_local, port: int, path: str, token: str, payload: dict, task_id: str -) -> None: +) -> bool: """ Send a JSON POST to the given InfluxDB 3 webhook endpoint, with up to 3 retry attempts and randomized backoff delays between attempts. @@ -346,8 +413,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. + Returns: + bool: True when the alert was accepted, False when every attempt failed. """ url: str = f"http://localhost:{port}/api/v3/engine/{path}" headers: dict = { @@ -366,7 +433,7 @@ def send_notification( influxdb3_local.info( f"[{task_id}] Alert sent to notification plugin with results: {resp.json()['results']}" ) - break + return True except requests.RequestException as e: influxdb3_local.warn( f"[{task_id}] [Attempt {attempt}/{max_retries}] Error sending alert to notification plugin: {e}" @@ -382,35 +449,7 @@ def send_notification( f"[{task_id}] Failed to send alert to notification plugin after {max_retries} attempts: {e}" ) - -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 + return False def validate_webhook_url(influxdb3_local, service: str, url: str, task_id: str) -> bool: @@ -455,130 +494,137 @@ def interpolate_notification_text(text: str, row_data: dict) -> str: return Template(text).safe_substitute(row_data) -def parse_time_duration(raw: str, task_id: str) -> timedelta: +def decode_detector_params(raw: str | dict) -> dict: """ - Parse a time duration string into a timedelta. - - Args: - raw (str): Duration string (e.g., "5m", "1h"). - task_id (str): Unique task identifier for logging. + Decode 'detector_params' from a mapping or a base64-encoded JSON string. Returns: - timedelta: Parsed duration. + dict: Mapping of detector names to their parameter dictionaries. Raises: - Exception: If the duration format is invalid. + Exception: If the value is not valid base64 or not a JSON object. """ - valid_units: dict = { - "s": "seconds", - "min": "minutes", - "h": "hours", - "d": "days", - "w": "weeks", - } - num_part, unit_part = "", "" - for unit in sorted(valid_units.keys(), key=len, reverse=True): - if raw.endswith(unit): - num_part = raw[: -len(unit)] - unit_part = unit - break - if not num_part or unit_part not in valid_units: - raise Exception(f"[{task_id}] Invalid duration format: {raw}") + if isinstance(raw, dict): + return raw + try: - num = int(num_part) - except ValueError: - raise Exception(f"[{task_id}] Invalid number in duration: {raw}") - return timedelta(**{valid_units[unit_part]: num}) + decoded: str = base64.b64decode(raw).decode("utf-8") + except Exception: + raise Exception("Invalid base64 encoding in detector_params") + try: + params = json.loads(decoded) + except json.JSONDecodeError: + raise Exception(f"Invalid JSON in decoded detector_params: {decoded}") -def parse_detector_params( - influxdb3_local, args: dict, detectors: list, task_id: str -) -> dict: - """ - Parse and validate detector parameters from args, expecting detector_params as a base64-encoded JSON string or use values from config file. + if not isinstance(params, dict): + raise Exception("detector_params must decode to a JSON object") + return params - Args: - influxdb3_local: InfluxDB client instance. - args (dict): Must include "detector_params" as a base64-encoded JSON string. - detectors (list): List of detector names. - task_id (str): Unique task identifier. + +def parse_detectors(influxdb3_local, config: dict, task_id: str) -> tuple[list, dict]: + """ + Resolve the detectors to apply together with their parameters. Returns: - dict: Mapping of detector names to their parameter dictionaries. + tuple[list, dict]: Applicable detector names and their parameters. Raises: - Exception: If detector_params is not valid base64, contains invalid JSON, or is missing required parameters. + Exception: If no detector is applicable. """ - input_params: str | dict = args["detector_params"] - - if args["use_config_file"]: - if isinstance(input_params, dict): - params: dict = input_params - else: - raise Exception( - f"[{task_id}] detector_params must be a dict when using config file" - ) - else: - try: - # Decode base64-encoded string - decoded_bytes: bytes = base64.b64decode(input_params) - decoded_str: str = decoded_bytes.decode("utf-8") - except Exception: - raise Exception( - f"[{task_id}] Invalid base64 encoding in detector_params: {input_params}" - ) - - try: - # Parse JSON from decoded string - params: dict = json.loads(decoded_str) - except json.JSONDecodeError: - raise Exception( - f"[{task_id}] Invalid JSON in decoded detector_params: {decoded_str}" - ) + params: dict = decode_detector_params(config["detector_params"]) - valid_params: dict = {} - for detector in detectors[:]: + detectors: list = [] + detector_params: dict = {} + for detector in config["detectors"]: + if detector not in AVAILABLE_DETECTORS: + influxdb3_local.warn(f"[{task_id}] Unknown detector: {detector}") + continue if detector not in params: influxdb3_local.warn( f"[{task_id}] Missing parameters for detector: {detector}" ) continue - valid_params[detector] = params[detector] + if not isinstance(params[detector], dict): + influxdb3_local.warn( + f"[{task_id}] Parameters for detector {detector} must be a mapping" + ) + continue - # Validate required parameters - if detector == "LevelShiftAD": - if "window" not in valid_params[detector]: - influxdb3_local.warn( - f"[{task_id}] LevelShiftAD requires 'window' parameter" - ) - del valid_params[detector] - detectors.remove(detector) - continue - elif detector == "VolatilityShiftAD": - if "window" not in valid_params[detector]: - influxdb3_local.warn( - f"[{task_id}] VolatilityShiftAD requires 'window' parameter" - ) - del valid_params[detector] - detectors.remove(detector) - continue + missing: list = [ + name + for name in REQUIRED_DETECTOR_PARAMS.get(detector, []) + if name not in params[detector] + ] + if missing: + influxdb3_local.warn( + f"[{task_id}] {detector} requires the '{', '.join(missing)}' parameter" + ) + continue - if not valid_params: - raise Exception( - f"[{task_id}] No valid detector parameters found in detector_params: {params}" - ) + detectors.append(detector) + detector_params[detector] = params[detector] - return valid_params + if not detectors: + raise Exception(f"No applicable detectors in {config['detectors']}") + return detectors, detector_params -def parse_min_consensus(min_consensus: str | int, task_id: str) -> int: - """Validate and convert min_consensus to an integer.""" - try: - return int(min_consensus) - except (TypeError, ValueError): - raise Exception( - f"[{task_id}] Invalid min_consensus, not an integer: {min_consensus!r}" - ) + +def format_tags(row: pd.Series, tags: list) -> str: + """Render the tag values of a row as 'tag=value' pairs.""" + return ", ".join(f"{tag}={row.get(tag, 'None')}" for tag in tags) + + +def split_by_tags(df: pd.DataFrame, tags: list, group_by_tags: bool) -> list: + """ + Split query results into one frame per tag combination. + """ + if not group_by_tags or not tags: + return [df] + return [group for _, group in df.groupby(tags, dropna=False, sort=False)] + + +def detect_anomalies( + influxdb3_local, + series: pd.Series, + detectors: list, + detector_params: dict, + min_consensus: int, + task_id: str, +) -> pd.Series | None: + """ + Apply every detector to the series and combine their verdicts by consensus. + + Returns: + pd.Series | None: True for every point flagged by at least 'min_consensus' + detectors, or None if no detector could be applied. + """ + anomaly_results: list = [] + for detector_name in detectors: + try: + params: dict = detector_params[detector_name] + influxdb3_local.info( + f"[{task_id}] Applying detector {detector_name} with params {params}" + ) + detector = AVAILABLE_DETECTORS[detector_name](**params) + if detector_name not in UNFITTED_DETECTORS: + detector.fit(series) + anomalies: pd.Series = detector.detect(series) + anomaly_results.append(anomalies) + influxdb3_local.info( + f"[{task_id}] Detector {detector_name} found {anomalies.sum()} anomalies" + ) + except Exception as e: + influxdb3_local.warn( + f"[{task_id}] Failed to apply detector {detector_name}: {e}" + ) + + if not anomaly_results: + return None + + anomaly_df = pd.concat(anomaly_results, axis=1).fillna(False) + return (anomaly_df.sum(axis=1) >= min_consensus).astype(bool) def process_scheduled_call( @@ -589,23 +635,24 @@ def process_scheduled_call( Queries a specified measurement and field within a time window, applies one or more ADTK detectors, and sends notifications for anomalies. Supports consensus-based detection - (all detectors must agree) and optional debounce logic. + (a configurable number of detectors must agree) and optional debounce logic. Args: influxdb3_local: InfluxDB client for querying, caching, and logging. call_time (datetime): UTC timestamp at which the scheduler triggers this function. args (dict): Required: - - table (str): Measurement name to query. + - measurement (str): Measurement name to query. - field (str): Numeric field to evaluate. - detectors (str): Dot-separated list of ADTK detectors. - - detector_params (str): JSON string of detector parameters. - - min_consensus (int): Minimum number of detectors required to flag anomaly. + - detector_params (str): Base64-encoded JSON of detector parameters. - window (str): Time window for data query (e.g., "1h"). - senders (str): Dot-separated notification channels. Optional: - config_file_path (str): path to config file to override args. - - min_condition_duration (str): Minimum anomaly duration (e.g., "5m"). + - min_consensus (int): Detectors required to flag an anomaly (default: 1). + - min_condition_duration (str): Minimum anomaly duration (e.g., "5min"). + - max_notifications_per_run (int): Notification cap per run (default: 20). - notification_text (str): Message template. - notification_path (str): Path for notification plugin (default: "notify"). - port_override (int): HTTP port (default: 8181). @@ -615,135 +662,79 @@ def process_scheduled_call( All exceptions are caught and logged via influxdb3_local.error. """ task_id: str = str(uuid.uuid4()) - influxdb3_local.info(f"[{task_id}] Starting anomaly detection scheduled call at {call_time} with args: {args}") + influxdb3_local.info( + f"[{task_id}] Starting anomaly detection scheduled call at {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 - - if ( - not args - or "measurement" not in args - or "field" not in args - or "detectors" not in args - or "detector_params" not in args - or "window" not in args - or "senders" not in args - ): - influxdb3_local.error( - f"[{task_id}] Missing required arguments: measurement, field, detectors, detector_params, window, or senders" - ) + config: dict | None = _load_config(influxdb3_local, args, task_id) + if config is None: return try: - # Parse configuration - influxdb3_local.info(f"[{task_id}] Starting configuration parsing") - measurement: str = args["measurement"] - # Validate measurement - all_measurements: list = get_all_measurements(influxdb3_local) - if measurement not in all_measurements: + measurement: str = config["measurement"] + if measurement not in get_table_names(influxdb3_local): influxdb3_local.error(f"[{task_id}] Measurement '{measurement}' not found") return - influxdb3_local.info(f"[{task_id}] Validated measurement: {measurement}") - field: str = args["field"] - detectors: list = ( - args["detectors"].split(".") - if not args.get("use_config_file") - else args["detectors"] - ) - detector_params: dict = parse_detector_params( - influxdb3_local, args, detectors, task_id - ) - influxdb3_local.info(f"[{task_id}] Retrieved detector_params: {detector_params}") - - min_consensus: int = parse_min_consensus(args.get("min_consensus", 1), task_id) - window: timedelta = parse_time_duration(args["window"], task_id) - senders_config: dict = parse_senders(influxdb3_local, args, task_id) - port_override: int = parse_port_override(args, task_id) - min_condition_duration: timedelta = parse_time_duration( - args.get("min_condition_duration", "0s"), task_id - ) - notification_path: str = args.get("notification_path", "notify") - notification_text: str = args.get( - "notification_text", - "Anomaly detected in $table.$field with value $value by $detectors. Tags: $tags", + field: str = config["field"] + detectors, detector_params = parse_detectors(influxdb3_local, config, task_id) + influxdb3_local.info( + f"[{task_id}] Retrieved detector_params: {detector_params}" ) - influxdb3_auth_token: str = args.get("influxdb3_auth_token") or os.getenv( - "INFLUXDB3_AUTH_TOKEN" + + min_consensus: int = config["min_consensus"] + if min_consensus > len(detectors): + influxdb3_local.warn( + f"[{task_id}] min_consensus={min_consensus} exceeds the {len(detectors)} applicable detectors, no point can reach consensus" + ) + + group_by_tags: bool = config["group_by_tags"] + max_notifications_per_run: int = config["max_notifications_per_run"] + window: timedelta = config["window"] + senders_config: dict = parse_senders(influxdb3_local, config, task_id) + port_override: int = config["port_override"] + min_condition_duration: timedelta = config["min_condition_duration"] + if min_condition_duration >= window: + influxdb3_local.warn( + f"[{task_id}] min_condition_duration={min_condition_duration} is not shorter than window={window}, an anomaly can never persist long enough to alert" + ) + notification_path: str = config["notification_path"] + notification_text: str = config["notification_text"] + influxdb3_auth_token: str = ( + config.get("influxdb3_auth_token") + or os.getenv("INFLUXDB3_AUTH_TOKEN") + or "" ) if not influxdb3_auth_token: influxdb3_local.error(f"[{task_id}] Missing influxdb3_auth_token") return - influxdb3_local.info(f"[{task_id}] Configuration completed: field={field}, detectors={len(detectors)}, min_consensus={min_consensus}, window={window}") + influxdb3_local.info( + f"[{task_id}] Configuration completed: field={field}, detectors={len(detectors)}, min_consensus={min_consensus}, window={window}" + ) # Query data - tags: list = get_tag_names(influxdb3_local, measurement, task_id) - tags_clause: str = ", ".join([f'"{tag}"' for tag in tags]) + tags: list = get_measurement_tags(influxdb3_local, measurement, task_id) end_time: datetime = call_time start_time: datetime = end_time - window - influxdb3_local.info(f"[{task_id}] Querying {measurement}.{field} from {start_time} to {end_time}") - query: str = f""" - SELECT "{field}", "time", {tags_clause} - FROM "{measurement}" - WHERE time >= $start_time AND time < $end_time - ORDER BY time - """ - result: list = influxdb3_local.query( - query, - {"start_time": start_time.isoformat(), "end_time": end_time.isoformat()}, + influxdb3_local.info( + f"[{task_id}] Querying {measurement}.{field} from {start_time} to {end_time}" + ) + result: list = query_window( + influxdb3_local, + measurement, + start=start_time.isoformat(), + end=end_time.isoformat(), + columns=[field, "time", *tags], ) if not result: influxdb3_local.info( f"[{task_id}] No data found for {measurement}.{field} from {start_time} to {end_time}" ) return - influxdb3_local.info(f"[{task_id}] Retrieved {len(result)} records from {measurement}") + influxdb3_local.info( + f"[{task_id}] Retrieved {len(result)} records from {measurement}" + ) # Convert to pandas Series df: pd.DataFrame = pd.DataFrame(result) @@ -752,72 +743,114 @@ def process_scheduled_call( f"[{task_id}] Field '{field}' or 'time' not found in query results" ) return - series: pd.Series = pd.Series( - df[field].values, index=pd.to_datetime(df["time"], unit="ns") - ) - series = validate_series(series) # Ensure regular sampling and time order - influxdb3_local.info(f"[{task_id}] Prepared time series data with {len(series)} points") + groups: list = split_by_tags(df, tags, group_by_tags) # Apply detectors - influxdb3_local.info(f"[{task_id}] Starting anomaly detection with {len(detectors)} detectors") - anomaly_results: list = [] - for detector_name in detectors: - try: - params: dict = detector_params[detector_name] + influxdb3_local.info( + f"[{task_id}] Starting anomaly detection with {len(detectors)} detectors on {len(groups)} series" + ) + # Process anomalies with debounce logic + influxdb3_local.info( + f"[{task_id}] Processing anomalies with debounce logic (min_condition_duration={min_condition_duration})" + ) + processed_anomalies = 0 + sent_notifications = 0 + failed_notifications = 0 + suppressed_notifications = 0 + for group in groups: + series_label: str = ( + f" (tags: {format_tags(group.iloc[0], tags)})" + if group_by_tags and tags + else "" + ) + # a time index keeps the per-point row lookup below out of a full scan + rows: pd.DataFrame = group.drop_duplicates(subset="time") + rows.index = pd.to_datetime(rows["time"], unit="ns") + + series: pd.Series = rows[field].dropna() + missing_values: int = len(rows) - len(series) + if missing_values: + # detectors raise on NaN, which would drop the whole series influxdb3_local.info( - f"[{task_id}] Applying detector {detector_name} with params {params}" + f"[{task_id}] Skipped {missing_values} points without a '{field}' value{series_label}" ) - detector_class = AVAILABLE_DETECTORS[detector_name] - detector = detector_class(**params) - if detector_name not in ("ThresholdAD",): - detector.fit(series) - anomalies: pd.Series = detector.detect(series) - anomaly_results.append(anomalies) - anomaly_count = anomalies.sum() - influxdb3_local.info(f"[{task_id}] Detector {detector_name} found {anomaly_count} anomalies") - except Exception as e: - influxdb3_local.warn( - f"[{task_id}] Failed to apply detector {detector_name}: {e}" + if series.empty: + influxdb3_local.info( + f"[{task_id}] No values to analyze for {measurement}.{field}{series_label}" ) + continue - if not anomaly_results: - influxdb3_local.error( - f"[{task_id}] No valid detectors applied to {measurement}.{field}" + series = validate_series(series) # Ensure regular sampling and time order + influxdb3_local.info( + f"[{task_id}] Prepared time series data with {len(series)} points{series_label}" ) - return - # Consensus: point is anomaly if >= min_consensus detectors agree - anomaly_df = pd.concat(anomaly_results, axis=1).fillna(False) - anomaly_count = anomaly_df.sum(axis=1) - consensus_anomalies = anomaly_count >= min_consensus - consensus_anomalies = consensus_anomalies.astype(bool) - total_consensus_anomalies = consensus_anomalies.sum() - influxdb3_local.info(f"[{task_id}] Consensus analysis: {total_consensus_anomalies} anomalies detected with min_consensus={min_consensus}") + consensus_anomalies = detect_anomalies( + influxdb3_local, + series, + detectors, + detector_params, + min_consensus, + task_id, + ) + if consensus_anomalies is None: + influxdb3_local.error( + f"[{task_id}] No valid detectors applied to {measurement}.{field}{series_label}" + ) + continue + influxdb3_local.info( + f"[{task_id}] Consensus analysis: {consensus_anomalies.sum()} anomalies detected with min_consensus={min_consensus}{series_label}" + ) - # Process anomalies with debounce logic - influxdb3_local.info(f"[{task_id}] Processing anomalies with debounce logic (min_condition_duration={min_condition_duration})") - processed_anomalies = 0 - sent_notifications = 0 - for idx, is_anomaly in consensus_anomalies.items(): - row: pd.Series = df[df["time"] == pd.Timestamp(idx.isoformat()).value].iloc[ - 0 - ] - time_datetime: datetime = pd.to_datetime(row["time"], unit="ns") - cache_key: str = generate_cache_key(measurement, field, tags, row) - tag_str: str = ", ".join(f"{t}={row.get(t, 'None')}" for t in tags) - start_time_str: str = influxdb3_local.cache.get(cache_key, default="") - - if is_anomaly: - if not start_time_str: - if min_condition_duration > timedelta(0): - # Start of a new anomaly - influxdb3_local.cache.put(cache_key, time_datetime.isoformat()) - influxdb3_local.info( - f"[{task_id}] Anomaly started for {measurement}.{field} (tags: {tag_str}), waiting for duration {min_condition_duration}" - ) - continue + for timestamp, is_anomaly in consensus_anomalies.items(): + row: pd.Series = rows.loc[timestamp] + cache_key: str = generate_cache_key(measurement, field, tags, row) + alert_key: str = f"{cache_key}:last_alert" + tag_str: str = format_tags(row, tags) + + last_alert_str: str = influxdb3_local.cache.get(alert_key, default="") + if last_alert_str and timestamp <= pd.Timestamp(last_alert_str): + continue + + processed_anomalies += 1 + start_time_str: str = influxdb3_local.cache.get(cache_key, default="") + alert_reason: str | None = None + + if is_anomaly: + if not start_time_str: + if min_condition_duration > timedelta(0): + # Start of a new anomaly + influxdb3_local.cache.put(cache_key, timestamp.isoformat()) + influxdb3_local.info( + f"[{task_id}] Anomaly started for {measurement}.{field} (tags: {tag_str}), waiting for duration {min_condition_duration}" + ) + continue + alert_reason = f"Anomaly detected for {measurement}.{field} (tags: {tag_str}), sending alert" + else: + # Check duration + elapsed: timedelta = timestamp - pd.Timestamp(start_time_str) + if elapsed < min_condition_duration: + influxdb3_local.info( + f"[{task_id}] Anomaly ongoing for {elapsed} < {min_condition_duration} for {measurement}.{field} (tags: {tag_str})" + ) + continue + alert_reason = f"Anomaly persisted for {elapsed} for {measurement}.{field} (tags: {tag_str}), sending alert" + elif start_time_str: + # Reset cache if anomaly stops + influxdb3_local.cache.delete(cache_key) + influxdb3_local.info( + f"[{task_id}] Anomaly cleared for {measurement}.{field} (tags: {tag_str})" + ) + + if alert_reason is None: + continue - # Send notification + if ( + sent_notifications + failed_notifications + >= max_notifications_per_run + ): + suppressed_notifications += 1 + else: payload: dict = { "notification_text": interpolate_notification_text( notification_text, @@ -827,14 +860,13 @@ def process_scheduled_call( "value": row[field], "detectors": ".".join(detectors), "tags": tag_str, + "timestamp": timestamp.isoformat(), }, ), "senders_config": senders_config, } - influxdb3_local.error( - f"[{task_id}] Anomaly detected for {measurement}.{field} (tags: {tag_str}), sending alert" - ) - send_notification( + influxdb3_local.error(f"[{task_id}] {alert_reason}") + delivered: bool = send_notification( influxdb3_local, port_override, notification_path, @@ -842,57 +874,26 @@ def process_scheduled_call( payload, task_id, ) + if not delivered: + # leave the state untouched so a later run can alert again + failed_notifications += 1 + continue sent_notifications += 1 - else: - # Check duration - duration_start_time: datetime = datetime.fromisoformat( - start_time_str - ) - elapsed: timedelta = time_datetime - duration_start_time - if elapsed >= min_condition_duration: - # Send notification - payload: dict = { - "notification_text": interpolate_notification_text( - notification_text, - { - "table": measurement, - "field": field, - "value": row[field], - "detectors": ".".join(detectors), - "tags": tag_str, - }, - ), - "senders_config": senders_config, - } - influxdb3_local.error( - f"[{task_id}] Anomaly persisted for {elapsed} for {measurement}.{field} (tags: {tag_str}), sending alert" - ) - send_notification( - influxdb3_local, - port_override, - notification_path, - influxdb3_auth_token, - payload, - task_id, - ) - influxdb3_local.cache.put( - cache_key, "" - ) # Reset cache after sending - sent_notifications += 1 - else: - influxdb3_local.info( - f"[{task_id}] Anomaly ongoing for {elapsed} < {min_condition_duration} for {measurement}.{field} (tags: {tag_str})" - ) - else: - # Reset cache if anomaly stops - if start_time_str: - influxdb3_local.cache.put(cache_key, "") - influxdb3_local.info( - f"[{task_id}] Anomaly cleared for {measurement}.{field} (tags: {tag_str})" - ) - processed_anomalies += 1 - influxdb3_local.info(f"[{task_id}] Anomaly processing completed: {processed_anomalies} points processed, {sent_notifications} notifications sent") + influxdb3_local.cache.delete(cache_key) + influxdb3_local.cache.put(alert_key, timestamp.isoformat()) + + if failed_notifications: + influxdb3_local.warn( + f"[{task_id}] {failed_notifications} notifications could not be delivered, the next run will alert on them again" + ) + if suppressed_notifications: + influxdb3_local.warn( + f"[{task_id}] Suppressed {suppressed_notifications} notifications after reaching max_notifications_per_run={max_notifications_per_run}" + ) + influxdb3_local.info( + f"[{task_id}] Anomaly processing completed: {processed_anomalies} points processed, {sent_notifications} notifications sent" + ) except Exception as e: influxdb3_local.error(f"[{task_id}] Error: {e}") diff --git a/influxdata/stateless_adtk_detector/manifest.toml b/influxdata/stateless_adtk_detector/manifest.toml index 210f5ef..d563f9e 100644 --- a/influxdata/stateless_adtk_detector/manifest.toml +++ b/influxdata/stateless_adtk_detector/manifest.toml @@ -2,7 +2,7 @@ manifest_schema_version = "1.3" [plugin] name = "stateless_adtk_detector" -version = "1.2.0" +version = "1.3.0" description = "Provides anomaly detection capabilities for time series data using the ADTK library. Supports multiple stateless detectors with consensus-based detection and customizable notification messages." triggers = ["process_scheduled_call"] homepage = "https://www.influxdata.com/" @@ -16,7 +16,7 @@ exclude = [ [dependencies] database_version = ">=3.0.0" -python = ["requests", "adtk", "pandas"] +python = ["influxdata-plugin-utils>=0.3.0", "requests", "adtk", "pandas<3"] [[dependencies.plugins]] index_url = "https://github.com/influxdata/influxdb3_plugins/releases/download/registry/index.json" diff --git a/influxdata/stateless_adtk_detector/requirements-dev.txt b/influxdata/stateless_adtk_detector/requirements-dev.txt new file mode 100644 index 0000000..f145028 --- /dev/null +++ b/influxdata/stateless_adtk_detector/requirements-dev.txt @@ -0,0 +1,5 @@ +pytest +influxdata-plugin-utils>=0.3.0 +requests +adtk +pandas<3 diff --git a/influxdata/stateless_adtk_detector/requirements.txt b/influxdata/stateless_adtk_detector/requirements.txt index 4592bd5..559c4f4 100644 --- a/influxdata/stateless_adtk_detector/requirements.txt +++ b/influxdata/stateless_adtk_detector/requirements.txt @@ -1,3 +1,4 @@ +influxdata-plugin-utils>=0.3.0 requests adtk -pandas \ No newline at end of file +pandas<3 diff --git a/influxdata/stateless_adtk_detector/test_adtk_anomaly_detection.py b/influxdata/stateless_adtk_detector/test_adtk_anomaly_detection.py new file mode 100644 index 0000000..257da81 --- /dev/null +++ b/influxdata/stateless_adtk_detector/test_adtk_anomaly_detection.py @@ -0,0 +1,738 @@ +import base64 +import json +from datetime import datetime, timedelta, timezone +from itertools import chain + +import pandas as pd +import pytest + +import adtk_anomaly_detection_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, params)) + 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] + + def logged(self, fragment, level=None): + return any(fragment in message for message in self.messages(level)) + + def window_query(self): + return next( + (q, p) + for q, p in self.queries + if "information_schema" not in q and "SHOW" not in q + ) + + +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 plugin_dir(monkeypatch, tmp_path): + monkeypatch.setenv("PLUGIN_DIR", str(tmp_path)) + monkeypatch.delenv("INFLUXDB3_PLUGIN_DIR", raising=False) + monkeypatch.delenv("INFLUXDB3_AUTH_TOKEN", raising=False) + return tmp_path + + +START = datetime(2026, 8, 13, 12, 0, tzinfo=timezone.utc) +CALL_TIME = START + timedelta(hours=1) + + +def encode(params): + return base64.b64encode(json.dumps(params).encode()).decode() + + +ARGS = { + "measurement": "cpu", + "field": "usage", + "detectors": "ThresholdAD", + "detector_params": encode({"ThresholdAD": {"high": 200}}), + "window": "2h", + "senders": "http", + "http_webhook_url": "https://example.com/hook", + "influxdb3_auth_token": "tok", +} + + +def rows(values, host="server1", start=START, step=timedelta(minutes=1)): + """Build query results: one row per value, one step apart.""" + return [ + { + "usage": value, + "time": int((start + step * index).timestamp() * 1_000_000_000), + "host": host, + } + for index, value in enumerate(values) + ] + + +def merged(*row_lists): + """Interleave several series the way a time-ordered query returns them.""" + return sorted(chain(*row_lists), key=lambda row: row["time"]) + + +def run(args=None, local=None, call_time=CALL_TIME): + local = local or FakeInfluxdb3Local(rows=rows([10.0, 999.0, 10.0])) + plugin.process_scheduled_call(local, call_time, {**ARGS, **(args or {})}) + return local + + +def series_of(values, start=START): + index = pd.to_datetime( + [ + int((start + timedelta(minutes=i)).timestamp() * 1e9) + for i in range(len(values)) + ], + unit="ns", + ) + return pd.Series(values, index=index) + + +# --- configuration ---------------------------------------------------------- + + +def test_config_applies_defaults(plugin_dir): + config = plugin._load_config(FakeInfluxdb3Local(), dict(ARGS), "tid") + + assert config["min_consensus"] == 1 + assert config["group_by_tags"] is False + assert config["max_notifications_per_run"] == 20 + assert config["min_condition_duration"] == timedelta(0) + assert config["port_override"] == 8181 + assert config["notification_path"] == "notify" + assert config["notification_text"] == plugin._DEFAULT_NOTIFICATION_TEXT + assert config["window"] == timedelta(hours=2) + assert config["detectors"] == ["ThresholdAD"] + + +def test_config_reports_missing_required_argument(plugin_dir): + local = FakeInfluxdb3Local() + args = {key: value for key, value in ARGS.items() if key != "field"} + + assert plugin._load_config(local, args, "tid") is None + assert local.logged("field is required", "error") + + +@pytest.mark.parametrize( + "override", + [ + {"min_consensus": "0"}, + {"window": "0s"}, + {"window": "10m"}, + {"port_override": "70000"}, + {"group_by_tags": "maybe"}, + {"max_notifications_per_run": "0"}, + {"min_condition_duration": "-5min"}, + ], +) +def test_config_rejects_invalid_values(plugin_dir, override): + local = FakeInfluxdb3Local() + + assert plugin._load_config(local, {**ARGS, **override}, "tid") is None + assert local.logged("Failed to load configuration", "error") + + +def test_config_rejects_non_toml_path(plugin_dir): + local = FakeInfluxdb3Local() + + assert ( + plugin._load_config(local, {"config_file_path": "config.yaml"}, "tid") is None + ) + assert local.logged("expected a .toml file", "error") + + +def test_config_reports_missing_toml_file(plugin_dir): + local = FakeInfluxdb3Local() + + assert ( + plugin._load_config(local, {"config_file_path": "absent.toml"}, "tid") is None + ) + assert local.logged("Failed to load configuration", "error") + + +def test_config_from_toml_uses_native_structures(plugin_dir, sent): + (plugin_dir / "adtk.toml").write_text( + 'measurement = "cpu"\n' + 'field = "usage"\n' + 'detectors = ["ThresholdAD"]\n' + 'window = "2h"\n' + 'senders = ["http"]\n' + 'http_webhook_url = "https://example.com/hook"\n' + 'influxdb3_auth_token = "from-toml"\n' + "port_override = 8182\n" + 'notification_path = "custom/notify"\n' + "\n[detector_params]\n" + "ThresholdAD = { high = 200 }\n" + ) + local = FakeInfluxdb3Local(rows=rows([10.0, 999.0])) + + plugin.process_scheduled_call(local, CALL_TIME, {"config_file_path": "adtk.toml"}) + + assert len(sent) == 1 + assert sent[0]["url"] == "http://localhost:8182/api/v3/engine/custom/notify" + assert sent[0]["headers"]["Authorization"] == "Bearer from-toml" + + +def test_config_from_toml_accepts_inline_spellings(plugin_dir, sent): + (plugin_dir / "adtk.toml").write_text( + 'measurement = "cpu"\n' + 'field = "usage"\n' + 'detectors = "ThresholdAD"\n' + f'detector_params = "{encode({"ThresholdAD": {"high": 200}})}"\n' + 'window = "2h"\n' + 'senders = "http.slack"\n' + 'http_webhook_url = "https://example.com/hook"\n' + 'slack_webhook_url = "https://hooks.slack.com/services/T"\n' + 'influxdb3_auth_token = "tok"\n' + ) + local = FakeInfluxdb3Local(rows=rows([10.0, 999.0])) + + plugin.process_scheduled_call(local, CALL_TIME, {"config_file_path": "adtk.toml"}) + + assert set(sent[0]["payload"]["senders_config"]) == {"http", "slack"} + + +def test_token_falls_back_to_environment(monkeypatch, plugin_dir, sent): + monkeypatch.setenv("INFLUXDB3_AUTH_TOKEN", "from-env") + args = {key: value for key, value in ARGS.items() if key != "influxdb3_auth_token"} + local = FakeInfluxdb3Local(rows=rows([10.0, 999.0])) + + plugin.process_scheduled_call(local, CALL_TIME, args) + + assert sent[0]["headers"]["Authorization"] == "Bearer from-env" + + +def test_missing_token_stops_the_run(plugin_dir, sent): + local = FakeInfluxdb3Local(rows=rows([999.0])) + args = dict(ARGS) + args["influxdb3_auth_token"] = "" + + plugin.process_scheduled_call(local, CALL_TIME, args) + + assert local.logged("Missing influxdb3_auth_token", "error") + assert sent == [] + + +# --- detector parameters ---------------------------------------------------- + + +def test_decode_detector_params_accepts_mapping_and_base64(): + params = {"ThresholdAD": {"high": 200}} + + assert plugin.decode_detector_params(params) == params + assert plugin.decode_detector_params(encode(params)) == params + + +@pytest.mark.parametrize( + "raw, message", + [ + ("!!!not base64!!!", "Invalid base64 encoding"), + (base64.b64encode(b"{broken").decode(), "Invalid JSON"), + (encode(["ThresholdAD"]), "must decode to a JSON object"), + ], +) +def test_decode_detector_params_rejects_invalid(raw, message): + with pytest.raises(Exception, match=message): + plugin.decode_detector_params(raw) + + +def test_parse_detectors_skips_detectors_it_cannot_apply(): + local = FakeInfluxdb3Local() + config = { + "detectors": ["ThresholdAD", "Nonsense", "PersistAD", "LevelShiftAD"], + "detector_params": { + "ThresholdAD": {"high": 200}, + "LevelShiftAD": {}, + "Nonsense": {}, + }, + } + + detectors, params = plugin.parse_detectors(local, config, "tid") + + assert detectors == ["ThresholdAD"] + assert set(params) == {"ThresholdAD"} + assert local.logged("Unknown detector: Nonsense", "warn") + assert local.logged("Missing parameters for detector: PersistAD", "warn") + assert local.logged("LevelShiftAD requires the 'window' parameter", "warn") + + +def test_parse_detectors_rejects_non_mapping_parameters(): + local = FakeInfluxdb3Local() + config = {"detectors": ["ThresholdAD"], "detector_params": {"ThresholdAD": [200]}} + + with pytest.raises(Exception, match="No applicable detectors"): + plugin.parse_detectors(local, config, "tid") + assert local.logged("must be a mapping", "warn") + + +# --- detection and consensus ------------------------------------------------ + + +def test_detect_anomalies_requires_min_consensus_agreement(): + local = FakeInfluxdb3Local() + series = plugin.validate_series(series_of([10.0, 11.0, 12.0, 300.0, 999.0])) + detectors = ["QuantileAD", "ThresholdAD"] + params = {"QuantileAD": {"high": 0.6}, "ThresholdAD": {"high": 500}} + + lenient = plugin.detect_anomalies(local, series, detectors, params, 1, "tid") + strict = plugin.detect_anomalies(local, series, detectors, params, 2, "tid") + + assert lenient.sum() == 2 + assert strict.sum() == 1 + assert strict[strict].index == series.index[-1:] + + +def test_detect_anomalies_returns_none_when_every_detector_fails(): + local = FakeInfluxdb3Local() + series = plugin.validate_series(series_of([1.0, 2.0, 3.0])) + + result = plugin.detect_anomalies( + local, series, ["ThresholdAD"], {"ThresholdAD": {"nonsense": 1}}, 1, "tid" + ) + + assert result is None + assert local.logged("Failed to apply detector ThresholdAD", "warn") + + +def test_only_trainable_detectors_are_fitted(monkeypatch): + fitted = [] + + class Recording: + def __init__(self, **params): + self.name = params["name"] + + def fit(self, series): + fitted.append(self.name) + + def detect(self, series): + return pd.Series(False, index=series.index) + + monkeypatch.setitem(plugin.AVAILABLE_DETECTORS, "ThresholdAD", Recording) + monkeypatch.setitem(plugin.AVAILABLE_DETECTORS, "PersistAD", Recording) + series = series_of([1.0, 2.0]) + + plugin.detect_anomalies( + FakeInfluxdb3Local(), + series, + ["ThresholdAD", "PersistAD"], + {"ThresholdAD": {"name": "threshold"}, "PersistAD": {"name": "persist"}}, + 1, + "tid", + ) + + assert fitted == ["persist"] + + +def test_unreachable_min_consensus_warns(plugin_dir, sent): + local = run({"min_consensus": "3"}) + + assert local.logged("min_consensus=3 exceeds the 1 applicable detectors", "warn") + assert sent == [] + + +# --- series preparation ----------------------------------------------------- + + +def test_split_by_tags_groups_only_when_enabled(): + frame = pd.DataFrame(merged(rows([1.0, 2.0]), rows([3.0, 4.0], host="server2"))) + + assert len(plugin.split_by_tags(frame, ["host"], False)) == 1 + assert len(plugin.split_by_tags(frame, ["host"], True)) == 2 + assert len(plugin.split_by_tags(frame, [], True)) == 1 + + +def test_null_values_are_dropped_before_detection(plugin_dir, sent): + local = run(local=FakeInfluxdb3Local(rows=rows([10.0, None, 999.0, 10.0]))) + + assert len(sent) == 1 + assert local.logged("Skipped 1 points without a 'usage' value", "info") + assert local.messages("warn") == [] + + +def test_series_without_any_value_is_skipped(plugin_dir, sent): + local = run(local=FakeInfluxdb3Local(rows=rows([None, None]))) + + assert sent == [] + assert local.logged("No values to analyze", "info") + assert local.messages("error") == [] + + +def test_duplicate_timestamps_keep_the_first_row(plugin_dir, sent): + duplicated = rows([10.0, 999.0]) + duplicated.append({**duplicated[1], "usage": 10.0}) + + local = run(local=FakeInfluxdb3Local(rows=duplicated)) + + assert local.logged("Prepared time series data with 2 points", "info") + assert len(sent) == 1 + assert ( + sent[0]["payload"]["notification_text"] + == "Anomaly detected in cpu.usage with value 999.0 by ThresholdAD. Tags: host=server1" + ) + + +def test_cache_key_sorts_tags_and_marks_missing_ones(): + row = pd.Series({"region": "eu", "host": "server1"}) + + key = plugin.generate_cache_key("cpu", "usage", ["region", "host", "rack"], row) + + assert key == "cpu:usage:host=server1:rack=None:region=eu" + assert plugin.format_tags(row, ["host", "rack"]) == "host=server1, rack=None" + + +# --- senders ---------------------------------------------------------------- + + +def test_senders_collect_channel_arguments(): + config = { + "senders": "http.slack", + "http_webhook_url": "https://example.com/hook", + "slack_webhook_url": "https://hooks.slack.com/services/T", + "slack_headers": "eyJhIjogMX0=", + } + + senders = plugin.parse_senders(FakeInfluxdb3Local(), config, "tid") + + assert senders["http"] == {"http_webhook_url": "https://example.com/hook"} + assert senders["slack"] == { + "slack_webhook_url": "https://hooks.slack.com/services/T", + "slack_headers": "eyJhIjogMX0=", + } + + +def test_senders_drop_channel_without_required_argument(): + local = FakeInfluxdb3Local() + config = {"senders": "http.sms", "http_webhook_url": "https://example.com/hook"} + + senders = plugin.parse_senders(local, config, "tid") + + assert set(senders) == {"http"} + assert local.logged( + "Required key 'twilio_to_number' missing for sender 'sms'", "warn" + ) + + +def test_senders_reject_unusable_configuration(): + local = FakeInfluxdb3Local() + + with pytest.raises(Exception, match="No valid senders configured"): + plugin.parse_senders(local, {"senders": "carrier_pigeon"}, "tid") + assert local.logged("Invalid sender type: carrier_pigeon", "warn") + + with pytest.raises(Exception, match="No valid senders configured"): + plugin.parse_senders( + FakeInfluxdb3Local(), + {"senders": "http", "http_webhook_url": "ftp://x"}, + "tid", + ) + + +# --- notifications ---------------------------------------------------------- + + +def test_notification_payload_carries_template_variables(plugin_dir, sent): + run( + { + "notification_text": "$table.$field=$value at $timestamp by $detectors ($tags) $unknown" + }, + local=FakeInfluxdb3Local(rows=rows([10.0, 999.0])), + ) + + text = sent[0]["payload"]["notification_text"] + assert text == ( + "cpu.usage=999.0 at 2026-08-13T12:01:00 by ThresholdAD (host=server1) $unknown" + ) + assert sent[0]["payload"]["senders_config"] == { + "http": {"http_webhook_url": "https://example.com/hook"} + } + + +def test_notification_cap_suppresses_the_rest(plugin_dir, sent): + local = run( + {"max_notifications_per_run": "2"}, + local=FakeInfluxdb3Local(rows=rows([999.0] * 5)), + ) + + assert len(sent) == 2 + assert local.logged("Suppressed 3 notifications", "warn") + + sent.clear() + plugin.process_scheduled_call( + local, CALL_TIME, {**ARGS, "max_notifications_per_run": "2"} + ) + assert sent == [] + + +@pytest.fixture +def failing_delivery(monkeypatch): + """Make every notification attempt fail and count the attempts.""" + attempts = [] + + def failing_post(url, headers=None, data=None, timeout=None): + attempts.append(url) + raise plugin.requests.ConnectionError("refused") + + monkeypatch.setattr(plugin.requests, "post", failing_post) + monkeypatch.setattr(plugin.time, "sleep", lambda seconds: None) + return attempts + + +def test_failed_delivery_is_retried_by_the_next_run(plugin_dir, failing_delivery): + local = run(local=FakeInfluxdb3Local(rows=rows([999.0]))) + + assert len(failing_delivery) == 3 + assert local.logged( + "Failed to send alert to notification plugin after 3 attempts", "error" + ) + assert local.logged("1 notifications could not be delivered", "warn") + assert local.logged("0 notifications sent", "info") + # the point stays unhandled, so a later run alerts on it again + assert not any(key.endswith(":last_alert") for key in local.cache.store) + + plugin.process_scheduled_call(local, CALL_TIME, dict(ARGS)) + assert len(failing_delivery) == 6 + + +def test_failed_deliveries_count_towards_the_cap(plugin_dir, failing_delivery): + run( + {"max_notifications_per_run": "2"}, + local=FakeInfluxdb3Local(rows=rows([999.0] * 6)), + ) + + assert len(failing_delivery) == 6 # two alerts, three attempts each + + +# --- debounce --------------------------------------------------------------- + + +def test_anomaly_alerts_immediately_without_debounce(plugin_dir, sent): + local = run(local=FakeInfluxdb3Local(rows=rows([10.0, 999.0, 999.0]))) + + assert len(sent) == 2 + assert local.logged("Anomaly detected for cpu.usage", "error") + + +def test_debounce_waits_for_the_configured_duration(plugin_dir, sent): + local = run( + {"min_condition_duration": "3min"}, + local=FakeInfluxdb3Local(rows=rows([10.0] + [999.0] * 5)), + ) + + assert len(sent) == 1 + assert local.logged("Anomaly started for cpu.usage", "info") + assert local.logged("Anomaly ongoing for 0 days 00:01:00", "info") + assert local.logged("Anomaly persisted for 0 days 00:03:00", "error") + + +def test_debounce_state_is_cleared_when_the_anomaly_stops(plugin_dir, sent): + local = run( + {"min_condition_duration": "1h"}, + local=FakeInfluxdb3Local(rows=rows([999.0, 999.0, 10.0])), + ) + + assert sent == [] + assert local.logged("Anomaly cleared for cpu.usage", "info") + assert "cpu:usage:host=server1" not in local.cache.store + + +def test_debounce_longer_than_the_window_warns(plugin_dir, sent): + local = run({"window": "10min", "min_condition_duration": "1h"}) + + assert local.logged("is not shorter than window", "warn") + + +# --- overlapping windows ---------------------------------------------------- + + +def test_same_anomaly_is_reported_once_across_runs(plugin_dir, sent): + local = FakeInfluxdb3Local(rows=rows([10.0, 999.0, 10.0])) + + for _ in range(3): + plugin.process_scheduled_call(local, CALL_TIME, dict(ARGS)) + + assert len(sent) == 1 + assert ( + local.cache.store["cpu:usage:host=server1:last_alert"] == "2026-08-13T12:01:00" + ) + + +def test_nanosecond_timestamps_are_not_realerted(plugin_dir, sent): + # datetime.fromisoformat truncates to microseconds, which would let a point + # with nanosecond precision pass the "already alerted" check on every run + nanos = [ + { + "usage": 999.0 if index else 10.0, + "time": int(START.timestamp() * 1_000_000_000) + + index * 60_000_000_000 + + 678_851_840, + "host": "server1", + } + for index in range(2) + ] + local = FakeInfluxdb3Local(rows=nanos) + + plugin.process_scheduled_call(local, CALL_TIME, dict(ARGS)) + assert len(sent) == 1 + + plugin.process_scheduled_call(local, CALL_TIME, dict(ARGS)) + assert len(sent) == 1 + + +def test_anomaly_after_the_last_alert_is_reported(plugin_dir, sent): + local = FakeInfluxdb3Local(rows=rows([10.0, 999.0])) + plugin.process_scheduled_call(local, CALL_TIME, dict(ARGS)) + + local.rows = rows([10.0, 999.0, 10.0, 999.0]) + sent.clear() + plugin.process_scheduled_call(local, CALL_TIME, dict(ARGS)) + + assert len(sent) == 1 + assert ( + local.cache.store["cpu:usage:host=server1:last_alert"] == "2026-08-13T12:03:00" + ) + + +# --- tag grouping ----------------------------------------------------------- + + +def test_without_grouping_only_the_first_series_is_analyzed(plugin_dir, sent): + local = run( + local=FakeInfluxdb3Local( + rows=merged(rows([10.0, 10.0]), rows([20.0, 999.0], host="server2")) + ) + ) + + assert sent == [] + assert local.logged("Prepared time series data with 2 points", "info") + + +def test_grouping_analyzes_every_tag_combination(plugin_dir, sent): + local = run( + {"group_by_tags": "true"}, + local=FakeInfluxdb3Local( + rows=merged(rows([10.0, 10.0]), rows([20.0, 999.0], host="server2")) + ), + ) + + assert len(sent) == 1 + assert "host=server2" in sent[0]["payload"]["notification_text"] + assert local.logged("on 2 series", "info") + assert local.logged("(tags: host=server1)", "info") + + +def test_grouping_keeps_debounce_state_per_series(plugin_dir, sent): + local = run( + {"group_by_tags": "true", "min_condition_duration": "2min"}, + local=FakeInfluxdb3Local( + rows=merged(rows([999.0] * 4), rows([999.0] * 4, host="server2")) + ), + ) + + assert len(sent) == 2 + assert sorted(k for k in local.cache.store if k.endswith(":last_alert")) == [ + "cpu:usage:host=server1:last_alert", + "cpu:usage:host=server2:last_alert", + ] + + +# --- scheduled flow guards -------------------------------------------------- + + +def test_unknown_measurement_is_reported(plugin_dir, sent): + local = run(local=FakeInfluxdb3Local(tables=("memory",), rows=rows([999.0]))) + + assert local.logged("Measurement 'cpu' not found", "error") + assert sent == [] + + +def test_empty_window_is_reported(plugin_dir, sent): + local = run(local=FakeInfluxdb3Local(rows=[])) + + assert local.logged("No data found for cpu.usage", "info") + assert sent == [] + + +def test_missing_field_column_is_reported(plugin_dir, sent): + local = run( + local=FakeInfluxdb3Local(rows=[{"other": 1.0, "time": 0, "host": "server1"}]) + ) + + assert local.logged("Field 'usage' or 'time' not found", "error") + assert sent == [] + + +def test_query_covers_the_window_and_quotes_identifiers(plugin_dir, sent): + local = run(local=FakeInfluxdb3Local(rows=rows([10.0]))) + query, params = local.window_query() + + assert '"usage", "time", "host"' in query + assert 'FROM "cpu"' in query + assert params["start"] == (CALL_TIME - timedelta(hours=2)).isoformat() + assert params["end"] == CALL_TIME.isoformat()