diff --git a/influxdata/library/plugin_library.json b/influxdata/library/plugin_library.json index b089733..327158b 100644 --- a/influxdata/library/plugin_library.json +++ b/influxdata/library/plugin_library.json @@ -244,12 +244,12 @@ { "name": "Signal Generator", "path": "influxdata/signal_generator/signal_generator.py", - "description": "[BETA] Requires InfluxDB 3.8.2+. Generates configurable waveform-based time-series data on a schedule for demos, testing, dashboards, alerts, and downstream plugin validation without requiring an external data source.", + "description": "[BETA] Requires InfluxDB 3.8.2+. Generates configurable waveform-based time-series data on a schedule for demos, testing, dashboards, alerts, and downstream plugin validation without requiring an external data source. Simulates data-source outage gaps by default.", "author": "InfluxData", "docs_file_link": "https://github.com/influxdata/influxdb3_plugins/blob/main/influxdata/signal_generator/README.md", "required_plugins": [], "required_libraries": [], - "last_update": "2026-05-15", + "last_update": "2026-07-09", "trigger_types_supported": ["scheduler"] }, { diff --git a/influxdata/signal_generator/README.md b/influxdata/signal_generator/README.md index 2bc8b9f..4d16418 100644 --- a/influxdata/signal_generator/README.md +++ b/influxdata/signal_generator/README.md @@ -10,7 +10,8 @@ The Signal Generator Plugin lets new users produce realistic time-series data fo - **Composable waveforms**: Mix sine, square, triangle, sawtooth, noise, and spike signals by stacking them - **Realistic signals**: Default preset produces a signal centered at 30 with a slow sine trend, Gaussian noise, and occasional large spikes — immediately useful for alert testing - **Timestamp jitter**: Offsets point timestamps by default to simulate sensors that do not sample at perfectly uniform intervals -- **Gap-filling**: Generates continuous, gapless data between executions regardless of trigger schedule +- **Simulated gaps**: Simulates temporary data-source outages by default — bounded windows with no written points, useful for testing deadman alerts and gap handling. Set `gap_enabled=false` for continuous output +- **Catch-up generation**: Generates data for the full span between executions regardless of trigger schedule — the only missing spans are the simulated gaps - **Flexible output**: Configure measurement name, field name, and custom tags per trigger ## Important CLI limitation @@ -44,6 +45,12 @@ This plugin has no required parameters. | `points_per_second`| float | `1.0` | Data point resolution in points per second. Controls how many points are generated per second of elapsed time. | | `jitter_amplitude_seconds` | float | 10% of point interval | Maximum timestamp offset in seconds. Set to `0` to disable timestamp jitter. | | `jitter_seed` | integer | *(none)* | Optional seed for reproducible per-timestamp jitter, mainly useful for tests. | +| `gap_enabled` | boolean | `true` | Enables simulated data-source outage gaps. Set to `false` for continuous output. | +| `gap_min_duration_seconds` | float | `10.0` | Shortest simulated outage duration, in seconds. | +| `gap_max_duration_seconds` | float | `30.0` | Longest simulated outage duration, in seconds. | +| `gap_min_interval_seconds` | float | `120.0` | Shortest time between consecutive outage starts, in seconds. Must be at least `gap_max_duration_seconds`. | +| `gap_max_interval_seconds` | float | `240.0` | Longest time between consecutive outage starts, in seconds. | +| `gap_seed` | integer | *(none)* | Optional seed for reproducible gap timing. | | `target_database` | string | *(none)* | Optional target database. If omitted, writes to the trigger's own database. | ### Timestamp jitter @@ -59,6 +66,28 @@ Each generated timestamp receives an independent random offset from `[-jitter_am Set `jitter_amplitude_seconds=0` to preserve perfectly regular timestamps. The plugin rejects jitter settings that could produce duplicate or reordered timestamps. +### Simulated gaps + +Simulated gaps are enabled by default. The plugin periodically simulates a data-source outage: points whose final (jittered) timestamps fall inside an outage window are not written, leaving realistic missing spans in the output. + +With the default configuration, each outage lasts 10–30 seconds, consecutive outages start 2–4 minutes apart, and about 11% of generated points are removed. The min/max bounds are hard guarantees: + +- Every outage duration lies in `[gap_min_duration_seconds, gap_max_duration_seconds]`. +- The time between consecutive outage starts lies in `[gap_min_interval_seconds, gap_max_interval_seconds]`. +- At least `gap_min_interval_seconds - gap_max_duration_seconds` seconds of continuous data separate consecutive outages (90 seconds at defaults). + +Gap windows are anchored to absolute time, so the same outage pattern is produced whether each scheduled execution writes one point or thousands, and catch-up after downtime contains the same gaps an uninterrupted run would have. When `gap_seed` is set, gap timing is fully reproducible. When omitted, the plugin generates a seed during first-run initialization and caches it, so the schedule stays coherent across executions; a restart that clears the trigger cache starts a new schedule. + +Missing spans are never backfilled: the generation boundary advances even when every point in a window was removed. Runs that removed points log a summary, for example: + +```text +Wrote 585 points to signal.value (600 generated, 15 removed by 1 gap windows) +``` + +Set `gap_enabled=false` to disable simulated gaps entirely. + +**Upgrade note:** existing triggers pick up default gaps on upgrade with no configuration change. Downstream demos that use deadman or no-data alerts will begin firing on the simulated outages — useful for exercising those alerts, but set `gap_enabled=false` if you need continuous data. + ### Waveform types Waveform configuration is supplied as a JSON array. Each object requires a `type` key; all other parameters are optional and fall back to defaults. @@ -350,6 +379,7 @@ Example with `tags={"host": "server01", "region": "us-west"}`: - Nanosecond precision Unix epoch timestamps. Each point's timestamp reflects simulated sample time, not the wall-clock time of the write. - Timestamp jitter is enabled by default, so adjacent timestamps are usually close to, but not exactly on, the nominal cadence grid. Set `jitter_amplitude_seconds=0` for regular intervals. +- Simulated gaps are enabled by default, so output contains periodic missing spans of 10–30 seconds. Set `gap_enabled=false` for continuous data. ### Line protocol examples @@ -428,14 +458,15 @@ Entry point for scheduled triggers. Orchestrates the full execution: parses conf Key operations: -1. Parses waveform, output, resolution, and timestamp jitter configuration from trigger arguments +1. Parses waveform, output, resolution, timestamp jitter, and gap configuration from trigger arguments 2. Reads `last_time` from the trigger-local cache -3. On first run: stores current time and returns without writing (initialization) +3. On first run: stores current time (and the auto-generated gap seed when gaps are enabled and unseeded) and returns without writing (initialization) 4. Generates timestamps in the half-open interval `(last_time, now]` 5. Evaluates the combined waveform at each nominal timestamp 6. Applies timestamp jitter without changing field values -7. Writes all points using `write_sync` with `no_sync=True` -8. Updates `last_time` in the cache +7. Removes points whose final timestamps fall inside simulated gap windows (see [Simulated gaps](#simulated-gaps)) +8. Writes the remaining points using `write_sync` with `no_sync=True` +9. Updates `last_time` in the cache #### Waveform factories @@ -453,6 +484,10 @@ Generates timestamps in the half-open interval `(start, end]`. The interval is e Offsets generated point timestamps after signal evaluation. Values are unchanged; only timestamps are moved. +#### `gap_window(gap_config, n)` / `in_gap(gap_config, t)` / `apply_gaps(points, gap_config)` + +Implements the simulated gap schedule: absolute time is divided into fixed strata, and stratum `n` contains one gap whose start offset and duration are derived from a `blake2b` hash of `(seed, n)`. Every window is a pure function of seed, configuration, and gap index, so results are identical regardless of how many points each scheduled call generates. `apply_gaps` filters final (jittered) timestamps and reports how many points were removed. See `GAP_DESIGN.md` for the full design. + ## Troubleshooting ### Common issues @@ -482,6 +517,12 @@ Look for a log entry containing `"Signal generator initializing"` — this confi echo '[{"type": "sine"}, {"type": "noise"}]' | python3 -m json.tool ``` +#### Issue: Missing spans in query results + +**Cause**: Simulated gaps are enabled by default. The plugin periodically simulates a data-source outage and writes no points inside the outage window. + +**Solution**: This is expected behavior. To confirm a missing span is a simulated gap rather than a real failure, check the logs — runs that removed points report `... (N generated, M removed by K gap windows)`. Set `gap_enabled=false` in the trigger arguments for continuous output. + #### Issue: No points generated (interval too short) **Cause**: The trigger fires more frequently than `1 / points_per_second` seconds, so no timestamps fall in the interval. @@ -510,7 +551,7 @@ influxdb3 query \ **Cause**: Deterministic waveforms (sine, square, triangle, sawtooth) are anchored to absolute Unix time, so they are always at the correct phase for a given wall-clock time. If the signal appears discontinuous, check that the `frequency` parameter is the same before and after the restart. -**Cause of gaps in data**: If the trigger was disabled or InfluxDB was stopped, the cache retains `last_time` from the last successful execution. On restart, the plugin generates all missing points from `last_time` to the current time, filling the gap automatically. +**Cause of gaps in data**: Short gaps (10–30 seconds by default) are simulated outages; see [Simulated gaps](#simulated-gaps). For downtime-related gaps: if the trigger was disabled or InfluxDB was stopped, the cache retains `last_time` from the last successful execution, and on restart the plugin generates all missing points from `last_time` to the current time — the caught-up span contains the same simulated gap windows an uninterrupted run would have. ### Viewing logs diff --git a/influxdata/signal_generator/manifest.toml b/influxdata/signal_generator/manifest.toml index e64ca79..5607dd6 100644 --- a/influxdata/signal_generator/manifest.toml +++ b/influxdata/signal_generator/manifest.toml @@ -2,8 +2,8 @@ manifest_schema_version = "1.2" [plugin] name = "signal_generator" -version = "0.2.0" -description = "Generates configurable waveform-based time-series data on a schedule for demos, testing, dashboards, alerts, and downstream plugin validation without requiring an external data source." +version = "0.3.0" +description = "Generates configurable waveform-based time-series data on a schedule for demos, testing, dashboards, alerts, and downstream plugin validation without requiring an external data source. Simulates data-source outage gaps by default." triggers = ["process_scheduled_call"] homepage = "https://www.influxdata.com/" repository = "https://github.com/influxdata/influxdb3_plugins/tree/main/influxdata/signal_generator" diff --git a/influxdata/signal_generator/signal_generator.py b/influxdata/signal_generator/signal_generator.py index 80804db..ec69192 100644 --- a/influxdata/signal_generator/signal_generator.py +++ b/influxdata/signal_generator/signal_generator.py @@ -44,6 +44,42 @@ "description": "Optional integer seed for reproducible per-timestamp jitter.", "required": false }, + { + "name": "gap_enabled", + "example": "false", + "description": "Enables simulated data-source outage gaps. Enabled by default; set to 'false' for continuous output.", + "required": false + }, + { + "name": "gap_min_duration_seconds", + "example": "10", + "description": "Shortest simulated outage duration in seconds. Defaults to 10.", + "required": false + }, + { + "name": "gap_max_duration_seconds", + "example": "30", + "description": "Longest simulated outage duration in seconds. Defaults to 30.", + "required": false + }, + { + "name": "gap_min_interval_seconds", + "example": "120", + "description": "Shortest time between consecutive outage starts in seconds. Must be at least gap_max_duration_seconds. Defaults to 120.", + "required": false + }, + { + "name": "gap_max_interval_seconds", + "example": "240", + "description": "Longest time between consecutive outage starts in seconds. Defaults to 240.", + "required": false + }, + { + "name": "gap_seed", + "example": "123", + "description": "Optional integer seed for reproducible gap timing.", + "required": false + }, { "name": "target_database", "example": "my_db", @@ -181,6 +217,11 @@ def _validate_params(params): DEFAULT_JITTER_INTERVAL_FRACTION = 0.10 MIN_JITTER_GAP_SECONDS = 1e-6 +DEFAULT_GAP_MIN_DURATION_SECONDS = 10.0 +DEFAULT_GAP_MAX_DURATION_SECONDS = 30.0 +DEFAULT_GAP_MIN_INTERVAL_SECONDS = 120.0 +DEFAULT_GAP_MAX_INTERVAL_SECONDS = 240.0 + # --------------------------------------------------------------------------- # Config parsing @@ -306,6 +347,101 @@ def parse_jitter_config(args, points_per_second): return ((amplitude, seed), None) +def _parse_bool(raw): + """Parse a boolean trigger argument. Returns None on error.""" + if isinstance(raw, bool): + return raw + if isinstance(raw, str): + lowered = raw.strip().lower() + if lowered == "true": + return True + if lowered == "false": + return False + return None + + +def parse_gap_config(args): + """Parse simulated gap config from trigger args. + Returns (config_dict, None) on success, or (None, error_message) on error. + All provided arguments are validated even when gaps are disabled.""" + if args is None: + args = {} + + raw_enabled = args.get("gap_enabled") + if raw_enabled is None: + enabled = True + else: + enabled = _parse_bool(raw_enabled) + if enabled is None: + return (None, "Invalid gap config: gap_enabled must be 'true' or 'false'") + + bounds = {} + for name, default in ( + ("gap_min_duration_seconds", DEFAULT_GAP_MIN_DURATION_SECONDS), + ("gap_max_duration_seconds", DEFAULT_GAP_MAX_DURATION_SECONDS), + ("gap_min_interval_seconds", DEFAULT_GAP_MIN_INTERVAL_SECONDS), + ("gap_max_interval_seconds", DEFAULT_GAP_MAX_INTERVAL_SECONDS), + ): + raw = args.get(name) + if raw is None: + bounds[name] = default + continue + try: + value = float(raw) + except (ValueError, TypeError): + return (None, f"Invalid gap config: {name} must be a float") + if not math.isfinite(value): + return (None, f"Invalid gap config: {name} must be finite") + bounds[name] = value + + min_duration = bounds["gap_min_duration_seconds"] + max_duration = bounds["gap_max_duration_seconds"] + min_interval = bounds["gap_min_interval_seconds"] + max_interval = bounds["gap_max_interval_seconds"] + + if min_duration < 0: + return (None, "Invalid gap config: gap_min_duration_seconds must be >= 0") + if max_duration < min_duration: + return (None, "Invalid gap config: gap_max_duration_seconds must be >= gap_min_duration_seconds") + if min_interval <= 0: + return (None, "Invalid gap config: gap_min_interval_seconds must be > 0") + if max_interval < min_interval: + return (None, "Invalid gap config: gap_max_interval_seconds must be >= gap_min_interval_seconds") + if min_interval < max_duration: + return ( + None, + f"Invalid gap config: gap_min_interval_seconds ({min_interval}s) must be >= " + f"gap_max_duration_seconds ({max_duration}s) so outages cannot overlap", + ) + + raw_seed = args.get("gap_seed") + if raw_seed is None: + seed = None + else: + try: + if isinstance(raw_seed, bool): + raise ValueError + if isinstance(raw_seed, float) and not raw_seed.is_integer(): + raise ValueError + seed = int(raw_seed) + except (ValueError, TypeError): + return (None, "Invalid gap config: gap_seed must be an integer") + + return ( + { + "enabled": enabled, + "min_duration": min_duration, + "max_duration": max_duration, + "min_interval": min_interval, + "max_interval": max_interval, + "pitch": (min_interval + max_interval) / 2, + "offset": (max_interval - min_interval) / 4, + "seed": seed, + }, + None, + ) + + # --------------------------------------------------------------------------- # Time series generation # --------------------------------------------------------------------------- @@ -356,6 +492,77 @@ def apply_timestamp_jitter(points, amplitude_seconds, seed=None): return jittered +# --------------------------------------------------------------------------- +# Simulated gaps (see GAP_DESIGN.md) +# --------------------------------------------------------------------------- + +def _rng_for_gap_index(seed, n): + payload = f"{seed}:gap:{n}".encode("ascii") + digest = hashlib.blake2b(payload, digest_size=8).digest() + return random.Random(int.from_bytes(digest, "big")) + + +def gap_window(gap_config, n): + """Gap n's half-open [start, end) window in epoch seconds. + The draw order (start offset, then duration) and the payload format in + _rng_for_gap_index are a reproducibility contract: changing either + reshuffles every seeded schedule.""" + rng = _rng_for_gap_index(gap_config["seed"], n) + start = n * gap_config["pitch"] + rng.uniform(-gap_config["offset"], gap_config["offset"]) + duration = rng.uniform(gap_config["min_duration"], gap_config["max_duration"]) + return (start, start + duration) + + +def in_gap(gap_config, t, window_cache=None): + """Return True if timestamp t falls inside a gap window. + Under the min_interval >= max_duration validation rule, only two gap + indices can contain t, so this is O(1).""" + n_hi = int((t + gap_config["offset"]) // gap_config["pitch"]) + for n in (n_hi, n_hi - 1): + if n < 0: + continue + if window_cache is None: + window = gap_window(gap_config, n) + else: + window = window_cache.get(n) + if window is None: + window = gap_window(gap_config, n) + window_cache[n] = window + start, end = window + if start <= t < end: + return True + return False + + +def apply_gaps(points, gap_config): + """Remove points whose final timestamp falls inside a gap window. + Returns (kept_points, removed_count, gap_window_count) where + gap_window_count is the number of gap windows intersecting the time span + of the given points.""" + if not points: + return ([], 0, 0) + + window_cache = {} + kept = [p for p in points if not in_gap(gap_config, p[0], window_cache)] + removed = len(points) - len(kept) + + timestamps = [t for t, _ in points] + t_min, t_max = min(timestamps), max(timestamps) + pitch = gap_config["pitch"] + offset = gap_config["offset"] + n_lo = int((t_min - offset - gap_config["max_duration"]) // pitch) + n_hi = int((t_max + offset) // pitch) + gap_window_count = 0 + for n in range(max(0, n_lo), n_hi + 1): + window = window_cache.get(n) + if window is None: + window = gap_window(gap_config, n) + start, end = window + if start <= t_max and end > t_min: + gap_window_count += 1 + return (kept, removed, gap_window_count) + + # --------------------------------------------------------------------------- # Batch write helper # --------------------------------------------------------------------------- @@ -426,6 +633,7 @@ def write_points(influxdb3_local, points, measurement, field, tags, target_datab # --------------------------------------------------------------------------- CACHE_KEY = "signal_gen:last_time" +GAP_SEED_CACHE_KEY = "signal_gen:gap_seed" def get_last_time(cache): @@ -436,6 +644,20 @@ def set_last_time(cache, t): cache.put(CACHE_KEY, t) +def resolve_gap_seed(cache, configured_seed): + """Return the gap seed: the configured value, else the cached auto-seed, + else a newly generated auto-seed persisted to the cache. Cache loss means + a new schedule (documented in GAP_DESIGN.md).""" + if configured_seed is not None: + return configured_seed + cached = cache.get(GAP_SEED_CACHE_KEY) + if cached is not None: + return int(cached) + seed = random.getrandbits(63) + cache.put(GAP_SEED_CACHE_KEY, seed) + return seed + + # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- @@ -468,10 +690,17 @@ def process_scheduled_call(influxdb3_local, call_time, args=None): influxdb3_local.error(f"[{task_id}] {jitter_error}") return + gap_config, gap_error = parse_gap_config(args) + if gap_config is None: + influxdb3_local.error(f"[{task_id}] {gap_error}") + return + # --- Check cache for last execution time --- last_time = get_last_time(influxdb3_local.cache) if last_time is None: influxdb3_local.info(f"[{task_id}] Signal generator initializing, first data on next execution") + if gap_config["enabled"]: + resolve_gap_seed(influxdb3_local.cache, gap_config["seed"]) set_last_time(influxdb3_local.cache, now) return @@ -488,9 +717,22 @@ def process_scheduled_call(influxdb3_local, call_time, args=None): jitter_amplitude_seconds, jitter_seed = jitter_config points = apply_timestamp_jitter(points, jitter_amplitude_seconds, jitter_seed) + generated = len(points) + removed = 0 + gap_window_count = 0 + if gap_config["enabled"]: + gap_config["seed"] = resolve_gap_seed(influxdb3_local.cache, gap_config["seed"]) + points, removed, gap_window_count = apply_gaps(points, gap_config) + try: written = write_points(influxdb3_local, points, measurement, field, tags, target_database) - influxdb3_local.info(f"[{task_id}] Wrote {written} points to {measurement}.{field}") + if removed: + influxdb3_local.info( + f"[{task_id}] Wrote {written} points to {measurement}.{field} " + f"({generated} generated, {removed} removed by {gap_window_count} gap windows)" + ) + else: + influxdb3_local.info(f"[{task_id}] Wrote {written} points to {measurement}.{field}") except Exception as e: influxdb3_local.error(f"[{task_id}] Batch write failed: {e}") diff --git a/influxdata/signal_generator/test_signal_generator.py b/influxdata/signal_generator/test_signal_generator.py index 1130003..10d0f9c 100644 --- a/influxdata/signal_generator/test_signal_generator.py +++ b/influxdata/signal_generator/test_signal_generator.py @@ -748,6 +748,8 @@ def test_process_scheduled_call_first_run(monkeypatch): # First run: no writes, only cache initialization assert len(mock.written_lines) == 0 assert mock.cache.get("signal_gen:last_time") is not None + # Gaps are enabled by default and unseeded, so init stores an auto-seed + assert mock.cache.get("signal_gen:gap_seed") is not None def test_process_scheduled_call_second_run(monkeypatch): @@ -755,12 +757,13 @@ def test_process_scheduled_call_second_run(monkeypatch): monkeypatch.setattr(signal_generator, "LineBuilder", MockLineBuilder) from signal_generator import process_scheduled_call mock = MockInfluxDB3Local() + args = {"gap_enabled": "false"} # Simulate first run first_time = datetime(2026, 4, 10, 12, 0, 0) - process_scheduled_call(mock, first_time) + process_scheduled_call(mock, first_time, args) # Second run 5 seconds later — should produce 5 points at 1 pps second_time = datetime(2026, 4, 10, 12, 0, 5) - process_scheduled_call(mock, second_time) + process_scheduled_call(mock, second_time, args) assert count_written_points(mock) == 5 @@ -769,7 +772,7 @@ def test_process_scheduled_call_custom_pps(monkeypatch): monkeypatch.setattr(signal_generator, "LineBuilder", MockLineBuilder) from signal_generator import process_scheduled_call mock = MockInfluxDB3Local() - args = {"points_per_second": "2"} + args = {"points_per_second": "2", "gap_enabled": "false"} first_time = datetime(2026, 4, 10, 12, 0, 0) process_scheduled_call(mock, first_time, args) # Second run 5 seconds later — 2 pps * 5s = 10 points @@ -783,7 +786,7 @@ def test_process_scheduled_call_target_database(monkeypatch): monkeypatch.setattr(signal_generator, "LineBuilder", MockLineBuilder) from signal_generator import process_scheduled_call mock = MockInfluxDB3Local() - args = {"target_database": "other_db"} + args = {"target_database": "other_db", "gap_enabled": "false"} first_time = datetime(2026, 4, 10, 12, 0, 0) process_scheduled_call(mock, first_time, args) second_time = datetime(2026, 4, 10, 12, 0, 5) @@ -831,6 +834,7 @@ def test_process_scheduled_call_jitter_zero_keeps_regular_timestamps(monkeypatch args = { "waveforms": '[{"type":"constant","value":1.0}]', "jitter_amplitude_seconds": "0", + "gap_enabled": "false", } first_time = datetime(2026, 4, 10, 12, 0, 0, tzinfo=timezone.utc) process_scheduled_call(mock, first_time, args) @@ -853,6 +857,7 @@ def test_process_scheduled_call_seeded_jitter_offsets_timestamps(monkeypatch): "waveforms": '[{"type":"constant","value":1.0}]', "jitter_amplitude_seconds": "0.1", "jitter_seed": "7", + "gap_enabled": "false", } first_time = datetime(2026, 4, 10, 12, 0, 0, tzinfo=timezone.utc) process_scheduled_call(mock, first_time, args) @@ -874,6 +879,7 @@ def test_process_scheduled_call_seeded_jitter_varies_across_one_point_calls(monk "waveforms": '[{"type":"constant","value":1.0}]', "jitter_amplitude_seconds": "0.1", "jitter_seed": "7", + "gap_enabled": "false", } first_time = datetime(2026, 4, 10, 12, 0, 0, tzinfo=timezone.utc) process_scheduled_call(mock, first_time, args) @@ -910,7 +916,7 @@ def test_process_scheduled_call_default_jitter_offsets_timestamps(monkeypatch): monkeypatch.setattr(signal_generator, "LineBuilder", MockLineBuilder) from signal_generator import process_scheduled_call mock = MockInfluxDB3Local() - args = {"waveforms": '[{"type":"constant","value":1.0}]'} + args = {"waveforms": '[{"type":"constant","value":1.0}]', "gap_enabled": "false"} first_time = datetime(2026, 4, 10, 12, 0, 0, tzinfo=timezone.utc) process_scheduled_call(mock, first_time, args) second_time = datetime(2026, 4, 10, 12, 0, 3, tzinfo=timezone.utc) @@ -928,3 +934,449 @@ def test_process_scheduled_call_default_jitter_offsets_timestamps(monkeypatch): earlier < later for earlier, later in zip(timestamps, timestamps[1:]) ) + + +# --------------------------------------------------------------------------- +# Gap config parsing tests +# --------------------------------------------------------------------------- + +def make_gap_config(seed=123, **args): + """Build a parsed gap config for schedule-function tests.""" + from signal_generator import parse_gap_config + config, error = parse_gap_config({k: str(v) for k, v in args.items()}) + assert error is None, error + config["seed"] = seed + return config + + +def test_parse_gap_config_defaults(): + from signal_generator import parse_gap_config + config, error = parse_gap_config(None) + assert error is None + assert config["enabled"] is True + assert config["min_duration"] == 10.0 + assert config["max_duration"] == 30.0 + assert config["min_interval"] == 120.0 + assert config["max_interval"] == 240.0 + assert config["pitch"] == 180.0 + assert config["offset"] == 30.0 + assert config["seed"] is None + + +def test_parse_gap_config_custom_values(): + from signal_generator import parse_gap_config + config, error = parse_gap_config({ + "gap_min_duration_seconds": "5", + "gap_max_duration_seconds": "15", + "gap_min_interval_seconds": "60", + "gap_max_interval_seconds": "100", + "gap_seed": "42", + }) + assert error is None + assert config["min_duration"] == 5.0 + assert config["max_duration"] == 15.0 + assert config["min_interval"] == 60.0 + assert config["max_interval"] == 100.0 + assert config["pitch"] == 80.0 + assert config["offset"] == 10.0 + assert config["seed"] == 42 + + +def test_parse_gap_config_enabled_false(): + from signal_generator import parse_gap_config + config, error = parse_gap_config({"gap_enabled": "false"}) + assert error is None + assert config["enabled"] is False + + +def test_parse_gap_config_enabled_case_insensitive(): + from signal_generator import parse_gap_config + config, error = parse_gap_config({"gap_enabled": "TRUE"}) + assert error is None + assert config["enabled"] is True + config, error = parse_gap_config({"gap_enabled": "False"}) + assert error is None + assert config["enabled"] is False + + +def test_parse_gap_config_enabled_garbage(): + from signal_generator import parse_gap_config + config, error = parse_gap_config({"gap_enabled": "yes"}) + assert config is None + assert "Invalid gap config" in error + + +def test_parse_gap_config_non_numeric_duration(): + from signal_generator import parse_gap_config + config, error = parse_gap_config({"gap_min_duration_seconds": "abc"}) + assert config is None + assert "Invalid gap config" in error + + +def test_parse_gap_config_nan_rejected(): + from signal_generator import parse_gap_config + config, error = parse_gap_config({"gap_max_interval_seconds": "nan"}) + assert config is None + assert "Invalid gap config" in error + + +def test_parse_gap_config_inf_rejected(): + from signal_generator import parse_gap_config + config, error = parse_gap_config({"gap_max_duration_seconds": "inf"}) + assert config is None + assert "Invalid gap config" in error + + +def test_parse_gap_config_negative_min_duration(): + from signal_generator import parse_gap_config + config, error = parse_gap_config({"gap_min_duration_seconds": "-1"}) + assert config is None + assert "Invalid gap config" in error + + +def test_parse_gap_config_max_duration_below_min(): + from signal_generator import parse_gap_config + config, error = parse_gap_config({ + "gap_min_duration_seconds": "20", + "gap_max_duration_seconds": "10", + }) + assert config is None + assert "Invalid gap config" in error + + +def test_parse_gap_config_zero_min_interval(): + from signal_generator import parse_gap_config + config, error = parse_gap_config({ + "gap_min_interval_seconds": "0", + "gap_max_interval_seconds": "10", + }) + assert config is None + assert "Invalid gap config" in error + + +def test_parse_gap_config_max_interval_below_min(): + from signal_generator import parse_gap_config + config, error = parse_gap_config({ + "gap_min_interval_seconds": "200", + "gap_max_interval_seconds": "100", + }) + assert config is None + assert "Invalid gap config" in error + + +def test_parse_gap_config_min_interval_below_max_duration(): + from signal_generator import parse_gap_config + config, error = parse_gap_config({ + "gap_max_duration_seconds": "30", + "gap_min_interval_seconds": "25", + "gap_max_interval_seconds": "240", + }) + assert config is None + assert "Invalid gap config" in error + + +def test_parse_gap_config_min_interval_equal_max_duration_valid(): + from signal_generator import parse_gap_config + config, error = parse_gap_config({ + "gap_max_duration_seconds": "30", + "gap_min_interval_seconds": "30", + "gap_max_interval_seconds": "90", + }) + assert error is None + assert config["min_interval"] == 30.0 + + +def test_parse_gap_config_non_integer_seed(): + from signal_generator import parse_gap_config + config, error = parse_gap_config({"gap_seed": "1.5"}) + assert config is None + assert "Invalid gap config" in error + + +def test_parse_gap_config_disabled_still_validates(): + from signal_generator import parse_gap_config + config, error = parse_gap_config({ + "gap_enabled": "false", + "gap_min_duration_seconds": "abc", + }) + assert config is None + assert "Invalid gap config" in error + + +# --------------------------------------------------------------------------- +# Gap schedule function tests +# --------------------------------------------------------------------------- + +def test_gap_window_bounded_spacing_and_no_overlap(): + from signal_generator import gap_window + config = make_gap_config(seed=123) + starts = [] + prev_end = None + for n in range(5000): + start, end = gap_window(config, n) + assert 10.0 <= end - start <= 30.0 + if prev_end is not None: + assert start >= prev_end + starts.append(start) + prev_end = end + spacings = [b - a for a, b in zip(starts, starts[1:])] + assert min(spacings) >= 120.0 + assert max(spacings) <= 240.0 + + +def test_gap_window_no_overlap_at_equality_boundary(): + # min_interval == max_duration: gaps may touch but never overlap + from signal_generator import gap_window + config = make_gap_config( + seed=7, + gap_min_duration_seconds=5, + gap_max_duration_seconds=30, + gap_min_interval_seconds=30, + gap_max_interval_seconds=90, + ) + prev_end = None + for n in range(5000): + start, end = gap_window(config, n) + if prev_end is not None: + assert start >= prev_end - 1e-9 + prev_end = end + + +def test_in_gap_matches_brute_force(): + import random as pyrandom + from signal_generator import gap_window, in_gap + config = make_gap_config(seed=123) + rng = pyrandom.Random(9) + for _ in range(5000): + t = rng.uniform(0, 5000 * config["pitch"]) + n_center = int(t // config["pitch"]) + brute = any( + gap_window(config, n)[0] <= t < gap_window(config, n)[1] + for n in range(max(0, n_center - 3), n_center + 4) + ) + assert in_gap(config, t) == brute + + +def test_gap_window_reproducible_across_seeds(): + from signal_generator import gap_window + config_a = make_gap_config(seed=123) + config_b = make_gap_config(seed=123) + config_c = make_gap_config(seed=124) + windows_a = [gap_window(config_a, n) for n in range(100)] + windows_b = [gap_window(config_b, n) for n in range(100)] + windows_c = [gap_window(config_c, n) for n in range(100)] + assert windows_a == windows_b + assert windows_a != windows_c + + +def test_gap_window_pinned_values(): + # Freezes the hash payload format and draw order (offset, then duration). + # If this fails, every seeded gap schedule in the wild has changed. + from signal_generator import gap_window + config = make_gap_config(seed=7) + start, end = gap_window(config, 1000) + assert abs(start - 180011.7885872335) < 1e-9 + assert abs(end - 180037.7907425121) < 1e-9 + + +def test_apply_gaps_empty_points(): + from signal_generator import apply_gaps + config = make_gap_config(seed=123) + assert apply_gaps([], config) == ([], 0, 0) + + +def test_apply_gaps_half_open_boundaries(): + from signal_generator import apply_gaps, gap_window + config = make_gap_config(seed=123) + start, end = gap_window(config, 50) + points = [(start, 1.0), ((start + end) / 2, 2.0), (end, 3.0)] + kept, removed, window_count = apply_gaps(points, config) + # start and midpoint are inside [start, end); the point exactly at end is kept + assert kept == [(end, 3.0)] + assert removed == 2 + assert window_count >= 1 + + +def test_apply_gaps_window_count(): + from signal_generator import apply_gaps, gap_window + config = make_gap_config(seed=123) + t0 = 1775822400.0 + points = [(t0 + i, 1.0) for i in range(1, 601)] + _, _, window_count = apply_gaps(points, config) + expected = 0 + n_center = int(t0 // config["pitch"]) + for n in range(max(0, n_center - 3), n_center + 8): + start, end = gap_window(config, n) + if start <= t0 + 600 and end > t0 + 1: + expected += 1 + assert window_count == expected + assert window_count >= 2 # a 600s span always intersects multiple windows + + +# --------------------------------------------------------------------------- +# Entry point gap tests +# --------------------------------------------------------------------------- + +GAP_TEST_ARGS = { + "waveforms": '[{"type":"constant","value":1.0}]', + "jitter_amplitude_seconds": "0", + "gap_seed": "123", +} + + +def test_process_scheduled_call_default_gaps_remove_points(monkeypatch): + import signal_generator + monkeypatch.setattr(signal_generator, "LineBuilder", MockLineBuilder) + from signal_generator import process_scheduled_call + mock = MockInfluxDB3Local() + first_time = datetime(2026, 4, 10, 12, 0, 0, tzinfo=timezone.utc) + process_scheduled_call(mock, first_time, GAP_TEST_ARGS) + second_time = datetime(2026, 4, 10, 12, 10, 0, tzinfo=timezone.utc) + process_scheduled_call(mock, second_time, GAP_TEST_ARGS) + written = count_written_points(mock) + # A 600s span always contains at least one full gap window (10-30s) + assert 0 < written < 600 + assert any("removed by" in log for log in mock.logs["info"]) + + +def test_process_scheduled_call_gap_batch_invariance(monkeypatch): + import signal_generator + monkeypatch.setattr(signal_generator, "LineBuilder", MockLineBuilder) + from signal_generator import process_scheduled_call + first_time = datetime(2026, 4, 10, 12, 0, 0, tzinfo=timezone.utc) + + mock_one = MockInfluxDB3Local() + process_scheduled_call(mock_one, first_time, GAP_TEST_ARGS) + process_scheduled_call( + mock_one, datetime(2026, 4, 10, 12, 10, 0, tzinfo=timezone.utc), GAP_TEST_ARGS + ) + + mock_many = MockInfluxDB3Local() + process_scheduled_call(mock_many, first_time, GAP_TEST_ARGS) + for seconds in range(10, 601, 10): + call_time = datetime(2026, 4, 10, 12, 0, 0, tzinfo=timezone.utc) + call_time = datetime.fromtimestamp( + call_time.timestamp() + seconds, tz=timezone.utc + ) + process_scheduled_call(mock_many, call_time, GAP_TEST_ARGS) + + assert written_timestamps_ns(mock_one) == written_timestamps_ns(mock_many) + assert 0 < count_written_points(mock_one) < 600 + + +def test_process_scheduled_call_gap_spans_call_boundary(monkeypatch): + import math + import signal_generator + monkeypatch.setattr(signal_generator, "LineBuilder", MockLineBuilder) + from signal_generator import gap_window, process_scheduled_call + config = make_gap_config(seed=123) + t0 = datetime(2026, 4, 10, 12, 0, 0, tzinfo=timezone.utc).timestamp() + + # Find a gap window comfortably inside (t0+60, t0+540) + n = int(t0 // config["pitch"]) + while True: + start, end = gap_window(config, n) + if start > t0 + 60 and end < t0 + 540: + break + n += 1 + boundary = int(start + (end - start) / 2) # call boundary inside the gap + + mock = MockInfluxDB3Local() + process_scheduled_call(mock, datetime.fromtimestamp(t0, tz=timezone.utc), GAP_TEST_ARGS) + + process_scheduled_call( + mock, datetime.fromtimestamp(boundary, tz=timezone.utc), GAP_TEST_ARGS + ) + written_first = count_written_points(mock) + generated_first = boundary - int(t0) + assert written_first < generated_first # gap removed points before the boundary + + process_scheduled_call( + mock, datetime.fromtimestamp(t0 + 600, tz=timezone.utc), GAP_TEST_ARGS + ) + written_second = count_written_points(mock) - written_first + generated_second = int(t0 + 600) - boundary + assert written_second < generated_second # and after the boundary + + # No written timestamp falls inside the gap window + for ts_ns in written_timestamps_ns(mock): + ts = ts_ns / 1_000_000_000 + assert not (start <= ts < end) + + +def test_process_scheduled_call_gap_covers_entire_call(monkeypatch): + import math + import signal_generator + monkeypatch.setattr(signal_generator, "LineBuilder", MockLineBuilder) + from signal_generator import gap_window, process_scheduled_call + config = make_gap_config(seed=123) + t0 = datetime(2026, 4, 10, 12, 0, 0, tzinfo=timezone.utc).timestamp() + n = int(t0 // config["pitch"]) + 1 + start, end = gap_window(config, n) + + init_time = math.ceil(start) + call_time = init_time + 4 # entire (init_time, call_time] sits inside the gap + assert call_time < end + + mock = MockInfluxDB3Local() + process_scheduled_call(mock, datetime.fromtimestamp(init_time, tz=timezone.utc), GAP_TEST_ARGS) + process_scheduled_call(mock, datetime.fromtimestamp(call_time, tz=timezone.utc), GAP_TEST_ARGS) + + assert count_written_points(mock) == 0 + assert any("removed by" in log for log in mock.logs["info"]) + # Generation boundary still advances: the missing span is never backfilled + assert mock.cache.get("signal_gen:last_time") == call_time + + +def test_process_scheduled_call_gap_catch_up_matches_schedule(monkeypatch): + import signal_generator + monkeypatch.setattr(signal_generator, "LineBuilder", MockLineBuilder) + from signal_generator import in_gap, process_scheduled_call + config = make_gap_config(seed=123) + t0 = int(datetime(2026, 4, 10, 12, 0, 0, tzinfo=timezone.utc).timestamp()) + + mock = MockInfluxDB3Local() + process_scheduled_call(mock, datetime.fromtimestamp(t0, tz=timezone.utc), GAP_TEST_ARGS) + # Single catch-up call one hour later + process_scheduled_call( + mock, datetime.fromtimestamp(t0 + 3600, tz=timezone.utc), GAP_TEST_ARGS + ) + + expected = [ + t * 1_000_000_000 + for t in range(t0 + 1, t0 + 3601) + if not in_gap(config, float(t)) + ] + assert written_timestamps_ns(mock) == expected + assert len(expected) < 3600 + + +def test_process_scheduled_call_unseeded_gap_seed_is_stable(monkeypatch): + import signal_generator + monkeypatch.setattr(signal_generator, "LineBuilder", MockLineBuilder) + from signal_generator import process_scheduled_call + mock = MockInfluxDB3Local() + args = {"waveforms": '[{"type":"constant","value":1.0}]'} + first_time = datetime(2026, 4, 10, 12, 0, 0, tzinfo=timezone.utc) + process_scheduled_call(mock, first_time, args) + seed_after_init = mock.cache.get("signal_gen:gap_seed") + assert isinstance(seed_after_init, int) + for seconds in (10, 20, 30): + call_time = datetime.fromtimestamp( + first_time.timestamp() + seconds, tz=timezone.utc + ) + process_scheduled_call(mock, call_time, args) + assert mock.cache.get("signal_gen:gap_seed") == seed_after_init + + +def test_process_scheduled_call_gap_config_error_writes_nothing(monkeypatch): + import signal_generator + monkeypatch.setattr(signal_generator, "LineBuilder", MockLineBuilder) + from signal_generator import process_scheduled_call + mock = MockInfluxDB3Local() + args = {"gap_min_duration_seconds": "abc"} + call_time = datetime(2026, 4, 10, 12, 0, 0) + process_scheduled_call(mock, call_time, args) + assert len(mock.written_lines) == 0 + assert mock.cache.get("signal_gen:last_time") is None + assert any("Invalid gap config" in log for log in mock.logs["error"])