diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 00000000..2f20469b --- /dev/null +++ b/.dockerignore @@ -0,0 +1,30 @@ +# Version-control and local tooling +.git +.github +.venv +venv +.ruff_cache +.mypy_cache +.pytest_cache +.tox +.nox +.idea +.vscode + +# Python build and test artifacts +**/__pycache__ +**/*.py[cod] +**/*.egg-info +build +dist +site +htmlcov +.coverage +.coverage.* + +# Local configuration, credentials, and logs +.env +.env.* +.pypi_token +*.log +uv.lock diff --git a/.gitignore b/.gitignore index 17ff1004..e04d5f06 100644 --- a/.gitignore +++ b/.gitignore @@ -3,6 +3,7 @@ site/ docs/img/ /.* +!/.dockerignore !/.github !/.pre-commit-config.yaml diff --git a/config/event_logging.yml b/config/event_logging.yml index 4c1f2be3..34313104 100644 --- a/config/event_logging.yml +++ b/config/event_logging.yml @@ -18,10 +18,16 @@ groups: # Two intervals - 5 min and 2 hours for longer-term history in Munin/Icinga intervals: ["5m", "2h"] # Cache counts locally, push to Redis every second - sync-interval: 1 + sync_interval: 1 # Number of processed tasks by their "src" attribute tasks_by_src: events: [] auto_declare_events: true intervals: ["5m", "2h"] - sync-interval: 1 + sync_interval: 1 + # Execution statistics for callbacks registered with the task executor + secondary_hooks: + events: [] + auto_declare_events: true + intervals: ["5m", "2h"] + sync_interval: 1 diff --git a/docker/python/Dockerfile b/docker/python/Dockerfile index 426ae6d6..019b2cb6 100644 --- a/docker/python/Dockerfile +++ b/docker/python/Dockerfile @@ -1,15 +1,24 @@ # syntax=docker/dockerfile:1 -# Base interpreter with installed requirements FROM python:3.11-slim AS base -RUN apt-get update; apt-get install -y \ - gcc \ - git -WORKDIR /dp3/ -COPY requirements.txt requirements.txt -RUN pip install --upgrade pip; \ - pip install -r requirements.txt +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + UV_LINK_MODE=copy \ + UV_SYSTEM_PYTHON=1 -COPY . /dp3/ -RUN SETUPTOOLS_SCM_PRETEND_VERSION=0.0.0 pip install -e /dp3 +RUN pip install --no-cache-dir "uv==0.9.7" + +RUN apt-get update && \ + apt-get install -y --no-install-recommends git && \ + rm -rf /var/lib/apt/lists/* + +WORKDIR /dp3 + +COPY requirements.txt ./ +RUN uv pip install --no-cache "setuptools>=61" "setuptools_scm>=6.2" wheel && \ + uv pip install --no-cache -r requirements.txt + +COPY . . +RUN SETUPTOOLS_SCM_PRETEND_VERSION=0.0.0 \ + uv pip install --no-cache --no-deps --no-build-isolation -e . diff --git a/docker/rabbitmq/.dockerignore b/docker/rabbitmq/.dockerignore new file mode 100644 index 00000000..31a1c19e --- /dev/null +++ b/docker/rabbitmq/.dockerignore @@ -0,0 +1,3 @@ +* +!Dockerfile +!init-rmq.sh diff --git a/docker/rabbitmq/Dockerfile b/docker/rabbitmq/Dockerfile index a5b2e3bf..0b0286d4 100644 --- a/docker/rabbitmq/Dockerfile +++ b/docker/rabbitmq/Dockerfile @@ -5,7 +5,6 @@ FROM rabbitmq:3-management ENV RABBITMQ_PID_FILE=/var/lib/rabbitmq/mnesia/rabbitmq # Add custom configuration script -ADD init-rmq.sh /init-rmq.sh -RUN chmod +x /init-rmq.sh +COPY --chmod=755 init-rmq.sh /init-rmq.sh CMD ["/init-rmq.sh"] diff --git a/docs/api.md b/docs/api.md index 683abc9e..92cacb76 100644 --- a/docs/api.md +++ b/docs/api.md @@ -5,7 +5,7 @@ As the API is made using FastAPI, there is also an interactive documentation ava If you are wiring a new producer into a DP³ application, start with [How to add an input module](howto/add-input.md), then use this page as the endpoint reference. -For routine same-host reads and writes, prefer `dp3 sh` or the generated `sh` wrapper. They provide a shell-oriented interface for the common API workflows documented here. Use raw HTTP requests when you need to exercise the underlying endpoint behavior directly. +For routine same-host reads and writes, prefer [`dp3 sh`](cli.md) or the generated `sh` wrapper. They provide a shell-oriented interface for the common API workflows documented here. Use raw HTTP requests when you need to exercise the underlying endpoint behavior directly. For an operational walkthrough, see [How to inspect DP³ telemetry](howto/telemetry.md). There are several API endpoints: diff --git a/docs/cli.md b/docs/cli.md new file mode 100644 index 00000000..e8650198 --- /dev/null +++ b/docs/cli.md @@ -0,0 +1,7 @@ +# `dp3 sh` command reference + +`dp3 sh` is a shell-oriented client for reading from and writing to a running DP³ API. It formats API responses as JSON or newline-delimited JSON and provides commands for datapoints, entities, control actions, telemetry, and shell completion. + +On a deployment host, you can use the generated `sh` wrapper in place of `dp3 sh`. The wrapper supplies the application's configuration directory. Otherwise, select a configuration with `--config` or `DP3_CONFIG_DIR`, and use `--url` when the API is not available through the automatically probed localhost URLs. + +{{ dp3_sh_help() }} diff --git a/docs/configuration/event_logging.md b/docs/configuration/event_logging.md index a47bb735..7ae7c74e 100644 --- a/docs/configuration/event_logging.md +++ b/docs/configuration/event_logging.md @@ -21,13 +21,19 @@ groups: - task_processed - task_processing_error intervals: [ "5m", "2h" ] # (1)! - sync-interval: 1 # (2)! + sync_interval: 1 # (2)! # Number of processed tasks by their "src" attribute tasks_by_src: events: [ ] auto_declare_events: true intervals: [ "5s", "5m" ] - sync-interval: 1 + sync_interval: 1 + # Task-executor hook execution statistics + secondary_hooks: + events: [ ] + auto_declare_events: true + intervals: [ "5m", "2h" ] + sync_interval: 1 ``` 1. Two intervals - 5 min and 2 hours for longer-term history in Munin/Icinga @@ -45,8 +51,15 @@ This section describes Redis connection details: # Groups -The default configuration groups enables logging of events in task execution, namely -`task_processed` and `task_processing_error`. +The default groups record task execution, processed tasks by source, and secondary-module hook +statistics. Hook event names are declared dynamically and use the namespace +`///`. The callback is module-qualified but omits bound +`partial` arguments, while the single context component identifies the hook's entity, attribute, +or snapshot scope. +EventCountLogger buffers each group's increments in memory and flushes them according to its +`sync_interval` or `sync_limit` setting. See the +[telemetry guide](../howto/telemetry.md#5-check-secondary-module-hooks) for the hook metrics and +their interpretation. -To learn more about the group configuration for EventCountLogger, +To learn more about the group configuration for EventCountLogger, please refer to the official [documentation](https://github.com/CESNET/EventCountLogger#configuration). diff --git a/docs/hooks.md b/docs/hooks.md index e728b771..ccfdac02 100644 --- a/docs/hooks.md +++ b/docs/hooks.md @@ -26,6 +26,10 @@ The most important split is: - **Ingestion-time** hooks see the incoming task or datapoint. - **Periodic / already-in-system** hooks see persisted data (`master_record`) or snapshot-time derived values. +A hook can be registered only once in the same hook context. Registering the same callback again +for the same hook type and entity or attribute raises `ValueError`. The same callback can still be +bound to different entities, attributes, hook types, or dependency contexts. + ## Quick hook placement guide Use this as a fast way to choose the right hook family. diff --git a/docs/howto/add-input.md b/docs/howto/add-input.md index 3994c62f..5047195f 100644 --- a/docs/howto/add-input.md +++ b/docs/howto/add-input.md @@ -108,7 +108,7 @@ Before running the real input module, send one datapoint manually. This narrows === "HTTP (`curl`)" ```shell - curl -X POST 'http://localhost:5000/datapoints' \ + curl --fail-with-body -X POST 'http://localhost:5000/datapoints' \ -H 'Content-Type: application/json' \ --data '[ { @@ -178,11 +178,31 @@ If the attribute shows up for the manually sent datapoint but not for the real i ### Check the API response first +The input module must check the HTTP status code and response body for every +`POST /datapoints` request. A completed HTTP call only proves that the server responded; it +does not prove that DP³ accepted the datapoints. Treat every non-2xx response as a failed +delivery and log enough of its response body to diagnose it. + If `POST /datapoints` returns an error, fix that before inspecting workers or the database. Validation errors are often enough to tell you whether: -- the attribute id is wrong -- the value type does not match -- timestamps are missing +- the entity type or attribute id is wrong +- the entity id or value has the wrong type +- required timestamps are missing + +### Check the bad datapoint API log + +A producer that ignores response codes can appear healthy while every payload is rejected. +Check `api.datapoint_logger.bad_log` in [`api.yml`](../configuration/api.md). If it contains a +path, inspect that file for the rejected input and its validation error: + +```shell +grep -n 'bad_log' /path/to/config/api.yml +tail -n 100 /path/to/configured/bad_dp.json.log +``` + +Datapoints in this log failed API validation and never reached RabbitMQ or a worker. If +`bad_log` is `false`, logging is disabled. To enable it, configure an absolute path whose +parent directory exists and is writable by the API process, then restart the API. ### Check API and worker logs @@ -248,6 +268,7 @@ worker logs, and attribute definition again. ## Common failure modes - The producer sends to the wrong API URL. +- The producer ignores a non-2xx API response, and rejected datapoints are only visible in the configured bad datapoint log. - The payload shape does not match the configured attribute type. - The attribute was not added to `db_entities` before the producer started sending it. - The API accepted the request, but workers are not running or are using old configuration. diff --git a/docs/howto/extending.md b/docs/howto/extending.md index 70e9f814..7722d017 100644 --- a/docs/howto/extending.md +++ b/docs/howto/extending.md @@ -76,6 +76,10 @@ There are additional options that can be specified, which affect the way the doc Even if you create a duplicate code reference description, the mkdocstring-style link still leads to the code reference, as you can see [here][dp3.snapshots.snapshot_hooks.SnapshotTimeseriesHookContainer.register]. +## CLI reference generation + +The [`dp3 sh` command reference](../cli.md) is generated from the CLI's argparse parsers during each MkDocs build. The `dp3_sh_help` macro in `macros.py` traverses the parser returned by `dp3.bin.sh.build_parser` and renders each command's `format_help()` output. Update command descriptions and argument help in the parser definitions under `dp3/bin/shcmd/`; do not copy command sections into `docs/cli.md` manually. + ## Deployment The documentation is updated and deployed automatically with each push to selected branches thanks to the configured GitHub Action, which can be found in: `.github/workflows/deploy.yml`. diff --git a/docs/howto/index.md b/docs/howto/index.md index 430c9350..9391cf6b 100644 --- a/docs/howto/index.md +++ b/docs/howto/index.md @@ -8,6 +8,7 @@ These guides walk through common DP³ application-development tasks step by step - [How to add an attribute](add-attribute.md) - Define a new attribute in `db_entities`, roll the configuration out, send test data, and verify that DP³ accepts it. - [How to add an input module](add-input.md) - Connect an external producer to the DP³ API and verify that the incoming data is accepted. - [How to add a secondary module](add-module.md) - Add worker-side logic that reacts to incoming or stored data and emits derived results. +- [How to inspect DP³ telemetry](telemetry.md) - Check input activity, stored data, queues, snapshots, periodic processes, and secondary modules in a running application. - [How to deploy a DP³ application](deploy-app.md) - Install the backing services, generate the process-manager setup, and operate a live deployment. - [How to set up for DP³ platform development](develop-dp3.md) - Prepare the repository for working on DP³ itself, including tests and docs. - [How to extend the documentation](extending.md) - Build the docs locally, preview changes, and work with the documentation toolchain. diff --git a/docs/howto/telemetry.md b/docs/howto/telemetry.md new file mode 100644 index 00000000..78510752 --- /dev/null +++ b/docs/howto/telemetry.md @@ -0,0 +1,204 @@ +# How to inspect DP³ telemetry + +This guide is a beginner-friendly checklist for answering three questions about a running DP³ application: + +- Are input sources still sending data? +- Is DP³ keeping up with the data it receives? +- Are secondary modules running as expected? + +`dp3 sh telemetry` is the packaged client for the telemetry exposed by the DP³ API. The examples below use it for convenience. You can also call the corresponding [telemetry API endpoints](../api.md#other-endpoints) directly. + +You will end up with: + +- a current view of active and stale input sources +- counts of the data present in the database +- a quick assessment of RabbitMQ queues and snapshot processing +- commands for investigating periodic processes and secondary modules + +## Before you start + +You need: + +- access to a running DP³ API +- the `dp3` command and the application's configuration directory, or its generated `sh` wrapper +- access to worker logs for secondary-module diagnostics +- `jq` for the optional metadata examples + +On a deployment host, use `sh` in place of `dp3 sh` below. The wrapper already knows the application's production configuration. For another API URL, pass `-u`, for example `dp3 sh -u http://localhost:5000 telemetry sources-validity`. + +## 1. Check for datapoints rejected by the API + +When an input module reports successful sends but no data appears in DP³, first verify that +the module checks the HTTP response status and body. **Completing an HTTP request does not +mean that DP³ accepted its datapoints.** + +If `api.datapoint_logger.bad_log` is configured in +[`api.yml`](../configuration/api.md), inspect that file for payload validation errors: + +```shell +grep -n 'bad_log' /path/to/config/api.yml +tail -n 100 /path/to/configured/bad_dp.json.log +``` + +The bad datapoint log includes the rejected input and its validation error. Typical causes +include an unknown entity or attribute, an invalid entity id, a value of the wrong type, or +missing timestamps. A request rejected by the API never reaches a worker, so it will not +appear in source-validity telemetry, Redis task counters, raw datapoints, or entity data. + +If `bad_log` is `false`, this diagnostic log is disabled. To use it, configure an absolute +path whose parent directory exists and is writable by the API process, then restart the API. + +## 2. Check RabbitMQ queues + +Inspect queue totals, consumers, and rates: + +```shell +dp3 sh telemetry rabbitmq-queues +``` + +For a compact table of queue names, totals, and rates, format the result with `jq` and `column`: + +```shell +dp3 sh telemetry rabbitmq-queues | jq '.queues[] | [.name,.total, .incoming, .outgoing] | @csv' -cr | column -ts ',' +``` + +Read the output as follows: + +- `total`, `ready`, and `unacked` show queued work. A large or continuously growing total indicates a backlog. +- `incoming` and `outgoing` are message rates. Sustained `incoming` much greater than `outgoing` means the backlog is growing. +- `outgoing` equal to or greater than `incoming` normally means the workers are keeping up or draining a backlog. +- `consumers` should match the processes expected to consume that queue. A zero rate is only healthy when no work is arriving. + +Rates are a momentary view. Repeat the command before drawing conclusions from a single sample. + +## 3. Check whether workers receive inputs + +Start with either view of source activity: + +```shell +dp3 sh telemetry sources-validity +dp3 sh telemetry source-age +dp3 sh telemetry source-age -u seconds +``` + +`sources-validity` and `source-age` are two views of the same telemetry records. For each datapoint source tag, they show either the latest datapoint validity timestamp (`t2`, or `t1` when `t2` is absent) or its age. The timestamp is recorded by an `on_task_start` hook, before task validation, database insertion, and attribute hooks. + +Use this to find sources that have stopped sending data. Keep these limitations in mind: + +- It shows that a worker received a task carrying the source, not that the task completed successfully. Validation, database insertion, or a later callback can still fail. +- A source remains listed after it becomes inactive. It can therefore appear here even after HistoryManager has deleted all of its old entity data. +- It does not report whether a secondary module processed the datapoint successfully. + +## 4. Check how much data reached the database + +Get current entity counts at attribute granularity: + +```shell +dp3 sh telemetry entities-per-attr +``` + +This answers “how many entities currently have each attribute?”. It counts value presence in the database, rather than the number of datapoints received over time. + +To drill into a particular attribute, list each distinct latest value and its entity count: + +```shell +dp3 sh entity attr-values +``` + +The equivalent HTTP endpoint is [`GET /entity//_/distinct/`](../api.md#get-distinct-values). + +## 5. Check secondary-module hooks + +Task-executor, snapshot, and periodic updater hooks publish execution statistics to the `secondary_hooks` EventCountLogger group. Read the counters for the current and last intervals with: + +```shell +dp3 sh telemetry event-counts --group secondary_hooks --interval 5m --both +``` + +Counter names use `///`. The callback is module-qualified, while bound `partial` arguments are omitted so configuration values do not become part of the metric identity. Hooks with the same callback and context therefore share counters even when their bound arguments differ. The context is one parenthesized namespace component identifying the entity, attribute, or snapshot scope. Reserved separators inside components are URL-escaped, including `/` in context values. All tracked hooks report `executions`, `failures`, and `duration_ns`. Hooks that can return datapoint tasks also report `created_tasks` when they create at least one. `allow_entity_creation` hooks instead report `decisions_allowed` or `decisions_denied` for successful calls. + +Snapshot telemetry covers timeseries, correlation, snapshot-run initialization, and snapshot-run finalization hooks. Timeseries counter contexts include the entity and attribute. Correlation counter contexts include only the entity type; dependency and changed-attribute declarations are omitted. Correlation hooks run once for each applicable entity, so their execution and task counts can be higher than the number of snapshot runs. Initialization and finalization hooks run once per worker because snapshot-run messages are broadcast to all workers. + +Periodic updater telemetry uses separate `periodic_update` and `periodic_eid_update` families. Its counter context contains the entity type, configured hook ID, and update period. Each execution represents one callback invocation for one entity in an updater batch. + +An allowed decision only means that one hook returned a truthy value. A later hook can still deny creation, and a later processing error can prevent the entity from being stored. A hook that denies creation also stops subsequent allow hooks from running. + +Use `duration_ns / executions` to calculate the mean callback duration in an interval. The duration includes failed executions, while `created_tasks` counts tasks returned by the hook rather than tasks subsequently processed successfully. + +The counters identify a failing callback, but the worker log contains its exception details: + +```shell +grep -F '' /var/log//worker*.log +grep 'Exception\|Error\|Traceback\|File "' -B1 -A1 /var/log//worker*.log +``` + +Callback registration is also logged during application startup. If no counters exist for a callback, restart the affected workers and search the startup logs for the module class name. This confirms whether its hooks were registered and exposes import or configuration errors. + +## 6. Read Redis event counters + +Every standard DP³ deployment has Redis-backed event counters configured through [`event_logging.yml`](../configuration/event_logging.md). Read the last completed interval by selecting a configured group and interval: + +```shell +dp3 sh telemetry event-counts -g te -i 5m +dp3 sh telemetry event-counts --group tasks_by_src --interval 2h +dp3 sh telemetry event-counts -g secondary_hooks -i 5m +``` + +The command defaults to `--last`. Use `--current` for the incomplete interval or `--both` to return both periods: + +```shell +dp3 sh telemetry event-counts -g te -i 5m --both +``` + +The `te` group contains task-processing and error counters. The `tasks_by_src` group contains one counter per observed datapoint source. The `secondary_hooks` group contains task-executor, snapshot, and periodic updater hook statistics described above. Groups and intervals are application-configurable; check `event_logging.yml` for the available values. + +This command uses the Redis connection from the selected DP³ configuration directory and must run from a host that can reach that Redis instance. Current counts may lag behind workers by each group's configured EventCountLogger synchronization interval. Last counts cover the most recently completed interval. + +## 7. Check snapshot processing + +Use the summary for a quick view of recent snapshot runs: + +```shell +dp3 sh telemetry snapshot-summary +``` + +The result reports: + +- `latest_age`: seconds since the newest snapshot run started +- `finished_age`: seconds since the newest completed run started +- `entities`: entities handled by that completed run +- `total_s`: duration of that completed run + +Unexpectedly old ages or unusually long durations are a reason to inspect metadata and worker logs. The summary is a formatted view of `SnapShooter` records in the internal `#metadata` collection. + +## 8. Investigate periodic processes through metadata + +The metadata command gives a lower-level view of records produced by internal periodic processes: + +```shell +dp3 sh telemetry metadata -m -l 5 +``` + +It supports module filtering with `-m`/`--module`, time bounds with `-f`/`--from` and `-t`/`--to`, pagination with `-s`/`--skip` and `-l`/`--limit`, ordering with `-S`/`--sort newest|oldest`, and output selection with `-F`/`--format json|ndjson`. Its default newline-delimited JSON output works well with `jq`: + +```shell +dp3 sh telemetry metadata -m SnapShooter -l 5 -f "2026-07-30T12:00:00" \ + | jq '{entities: .entities, components: .components, w_done: .workers_finished, start: .task_creation_start, end: ."#last_update"}' -c + +dp3 sh telemetry metadata -m HistoryManager -l 4 -f "2026-07-30T12:00:00" \ + | jq '{id: ._id, entities: .entities, updated: .updated, rev_conflicts: .revision_conflicts, retries: .retries, retry_fail: .retry_exhausted, end: ."#last_update"}' -c + +dp3 sh telemetry metadata -m GarbageCollector -l 12 \ + | jq '{id: ._id, etype: .entity, seen: .entities, deleted: .deleted, end: ."#last_update"}' -c +``` + +Replace the example timestamps with the period you are investigating. For `SnapShooter`, `workers_finished` equal to the configured worker count means that all workers finished their part of the run. The snapshot summary treats a run as complete after linked-entity processing has also finished. + +Metadata is an internal diagnostic format, so fields differ by module and may change as the corresponding process evolves. + +## Related pages + +- [API reference](../api.md) +- [How to deploy a DP³ application](deploy-app.md) +- [Event logging configuration](../configuration/event_logging.md) +- [Processing core configuration](../configuration/processing_core.md) diff --git a/dp3/bin/sh.py b/dp3/bin/sh.py index e38ee17c..50a31122 100644 --- a/dp3/bin/sh.py +++ b/dp3/bin/sh.py @@ -9,7 +9,13 @@ from argcomplete.shell_integration import shellcode from dp3.bin.shcmd import control, datapoints, entities, entity, health, telemetry -from dp3.bin.shcmd.common import APIError, DP3APIClient, resolve_config_dir +from dp3.bin.shcmd.common import ( + APIError, + DP3APIClient, + MarkdownHelpArgumentParser, + command_description, + resolve_config_dir, +) from dp3.common.config import ModelSpec, read_config_dir @@ -32,9 +38,10 @@ def register_completion_parser(commands) -> None: completion_parser = commands.add_parser( "completion", help="Print shell completion scripts.", - description=( - "Print shell completion scripts backed by argcomplete. Evaluate the generated " - "output in your shell or source it from your shell startup file." + description=command_description( + "Print an argcomplete registration script for the selected shell. Evaluate the " + "output or source it from your shell startup file.", + "dp3 sh completion bash --command dp3", ), ) completion_parser.add_argument( @@ -50,7 +57,7 @@ def register_completion_parser(commands) -> None: default=None, help=( "Command name to register completion for. Repeat to register multiple " - "commands. Use 'dp3' for 'dp3 sh' and 'sh' for wrapper commands." + "commands. Use `dp3` for `dp3 sh` and `sh` for wrapper commands." ), ) completion_parser.set_defaults( @@ -58,6 +65,16 @@ def register_completion_parser(commands) -> None: ) +def build_parser() -> argparse.ArgumentParser: + """Build the standalone shell-oriented CLI parser.""" + parser = MarkdownHelpArgumentParser( + prog="dp3 sh", + description="Shell-oriented interface to a running DP3 API.", + ) + init_parser(parser) + return parser + + def init_parser(parser: argparse.ArgumentParser) -> None: """Initialize the shell-oriented CLI parser.""" config_action = parser.add_argument( @@ -84,7 +101,11 @@ def init_parser(parser: argparse.ArgumentParser) -> None: help="HTTP timeout in seconds.", ) - commands = parser.add_subparsers(dest="sh_command", required=True) + commands = parser.add_subparsers( + dest="sh_command", + required=True, + parser_class=MarkdownHelpArgumentParser, + ) health.register_parser(commands) datapoints.register_parser(commands) entities.register_parser(commands) @@ -96,8 +117,7 @@ def init_parser(parser: argparse.ArgumentParser) -> None: def run() -> None: """Run the shell-oriented CLI as a standalone script.""" - parser = argparse.ArgumentParser(prog="dp3 sh") - init_parser(parser) + parser = build_parser() argcomplete.autocomplete(parser) args = parser.parse_args() sys.exit(main(args)) diff --git a/dp3/bin/shcmd/common.py b/dp3/bin/shcmd/common.py index 266bde98..cbbb3784 100644 --- a/dp3/bin/shcmd/common.py +++ b/dp3/bin/shcmd/common.py @@ -1,6 +1,7 @@ #!/usr/bin/env python3 """Shared helpers for the shell-oriented DP3 CLI.""" +import argparse import json import os import sys @@ -18,10 +19,23 @@ class APIError(RuntimeError): JSON_LITERAL_HELP = ( - "'set' requires a JSON literal value, for example '\"hello\"', '42', 'true', or '{\"k\":1}'." + '`set` requires a JSON literal value, for example `"hello"`, `42`, `true`, or `{"k":1}`.' ) +def command_description(summary: str, example: str) -> str: + """Build a Markdown command description with a shell example.""" + return f"{summary}\n\n```shell\n{example}\n```" + + +class MarkdownHelpArgumentParser(argparse.ArgumentParser): + """Argument parser that preserves Markdown description formatting in terminal help.""" + + def __init__(self, *args, **kwargs): + kwargs.setdefault("formatter_class", argparse.RawDescriptionHelpFormatter) + super().__init__(*args, **kwargs) + + class DP3APIClient: """Small HTTP client for the DP3 API.""" diff --git a/dp3/bin/shcmd/control.py b/dp3/bin/shcmd/control.py index af793046..58454c1a 100644 --- a/dp3/bin/shcmd/control.py +++ b/dp3/bin/shcmd/control.py @@ -33,17 +33,24 @@ def handle_refresh_module_config(client, args) -> int: def register_parser(commands) -> None: """Register control commands on the root parser.""" - control_parser = commands.add_parser("control", help="Execute control actions.") + control_parser = commands.add_parser( + "control", + help="Execute control actions.", + description="Execute operational control actions against the DP³ API.", + ) control_commands = control_parser.add_subparsers(dest="control_command", required=True) make_snapshots_parser = control_commands.add_parser( - "make-snapshots", help="Trigger an out-of-order snapshot run." + "make-snapshots", + help="Trigger an out-of-order snapshot run.", + description="Trigger an out-of-order snapshot run.", ) make_snapshots_parser.set_defaults(handler=handle_make_snapshots) refresh_entity_creation_parser = control_commands.add_parser( "refresh-on-entity-creation", help="Re-run entity creation callbacks for an entity type.", + description="Re-run entity-creation callbacks for one entity type.", ) etype_action = refresh_entity_creation_parser.add_argument("etype") etype_action.completer = complete_entity_type_names @@ -52,6 +59,7 @@ def register_parser(commands) -> None: refresh_module_config_parser = control_commands.add_parser( "refresh-module-config", help="Reload module configuration.", + description="Reload configuration for one module.", ) refresh_module_config_parser.add_argument("module") refresh_module_config_parser.set_defaults(handler=handle_refresh_module_config) diff --git a/dp3/bin/shcmd/datapoints.py b/dp3/bin/shcmd/datapoints.py index ae62d6fe..75e834fa 100644 --- a/dp3/bin/shcmd/datapoints.py +++ b/dp3/bin/shcmd/datapoints.py @@ -15,7 +15,11 @@ def register_parser(commands) -> None: datapoints_parser = commands.add_parser( "datapoints", help="Post datapoints from JSON input.", - description="Post datapoints from JSON input.", + description=( + "Post a JSON array of datapoints from a file or standard input. Each datapoint " + "must contain `type`, `id`, `attr`, `v`, and `src`; observations and time series " + "also require ISO 8601 `t1` and `t2` timestamps." + ), ) datapoints_parser.add_argument( "path", diff --git a/dp3/bin/shcmd/entities.py b/dp3/bin/shcmd/entities.py index 5580b4c2..e26f904e 100644 --- a/dp3/bin/shcmd/entities.py +++ b/dp3/bin/shcmd/entities.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 """Entity catalog commands for the shell-oriented DP3 CLI.""" -from dp3.bin.shcmd.common import print_response_json +from dp3.bin.shcmd.common import command_description, print_response_json def handle_entities(client, _args) -> int: @@ -14,10 +14,10 @@ def register_parser(commands) -> None: entities_parser = commands.add_parser( "entities", help="Return the full entity-type map exposed by the API.", - description=( - "Return the full entity-type map exposed by the API. To print only entity " - "type names, use 'dp3 sh entities | jq keys' or 'sh entities | jq " - "keys'." + description=command_description( + "Return the full entity-type map exposed by the API. Pipe the result through " + "`jq keys` to print only entity type names.", + "dp3 sh entities | jq keys", ), ) entities_parser.set_defaults(handler=handle_entities) diff --git a/dp3/bin/shcmd/entity/__init__.py b/dp3/bin/shcmd/entity/__init__.py index 46de9e9a..629fecbf 100644 --- a/dp3/bin/shcmd/entity/__init__.py +++ b/dp3/bin/shcmd/entity/__init__.py @@ -3,16 +3,18 @@ import argparse +from dp3.bin.shcmd.common import MarkdownHelpArgumentParser + from . import etype from .common import complete_entity_rest, complete_entity_selector def _build_overview_parser() -> argparse.ArgumentParser: - parser = argparse.ArgumentParser( + parser = MarkdownHelpArgumentParser( prog="dp3 sh entity", description=( - "Inspect and modify entity data. Use 'dp3 sh entities' to list entity types, " - "then continue with a type-scope command or 'id' for a single entity." + "Inspect and modify data for a selected entity type. Use `dp3 sh entities` " + "to list the available entity types." ), ) selector_action = parser.add_argument( diff --git a/dp3/bin/shcmd/entity/attr.py b/dp3/bin/shcmd/entity/attr.py index 2628bf1d..9f97e1b7 100644 --- a/dp3/bin/shcmd/entity/attr.py +++ b/dp3/bin/shcmd/entity/attr.py @@ -1,8 +1,7 @@ #!/usr/bin/env python3 """Entity attribute-scope commands for the shell-oriented DP3 CLI.""" - -from dp3.bin.shcmd.common import common_time_params, print_response_json +from dp3.bin.shcmd.common import command_description, common_time_params, print_response_json from .common import ( add_entity_attr_set_args, @@ -25,16 +24,31 @@ def handle_get(client, args) -> int: def add_instance_attr_parser(commands, etype: str) -> None: """Register entity attribute commands under a single-entity parser.""" - attr_parser = commands.add_parser("attr", help="Get or modify an entity attribute value.") + attr_parser = commands.add_parser( + "attr", + help="Get or modify an entity attribute value.", + description="Get or modify one attribute value for one entity.", + ) attr_parser.set_defaults(etype=etype) attr_action = attr_parser.add_argument("attr", metavar="ATTR") attr_action.completer = complete_entity_attr_names attr_commands = attr_parser.add_subparsers(dest="entity_attr_command", required=True) - get_parser = attr_commands.add_parser("get", help="Get an entity attribute value.") + get_parser = attr_commands.add_parser( + "get", + help="Get an entity attribute value.", + description="Get one attribute value for one entity.", + ) add_time_range_args(get_parser) get_parser.set_defaults(handler=handle_get, etype=etype) - set_parser = attr_commands.add_parser("set", help="Set a current entity attribute value.") + set_parser = attr_commands.add_parser( + "set", + help="Set a current entity attribute value.", + description=command_description( + "Set the current value of one entity attribute from a JSON literal.", + f"dp3 sh entity {etype} id EID attr ATTR set '\"value\"'", + ), + ) add_entity_attr_set_args(set_parser) set_parser.set_defaults(handler=handle_attr_set_request, etype=etype) diff --git a/dp3/bin/shcmd/entity/common.py b/dp3/bin/shcmd/entity/common.py index 0b3eb9a0..53117ada 100644 --- a/dp3/bin/shcmd/entity/common.py +++ b/dp3/bin/shcmd/entity/common.py @@ -23,7 +23,7 @@ "Browse current raw datapoints for troubleshooting ingestion. " "Can be slow on large raw collections." ) -ATTR_VALUE_HELP = "JSON literal value, for example '\"hello\"', '42', 'true', or '{\"k\":1}'." +ATTR_VALUE_HELP = 'JSON literal value, for example `"hello"`, `42`, `true`, or `{"k":1}`.' ENTITY_ID_PLACEHOLDER = "" @@ -42,14 +42,14 @@ def add_time_range_args( "-f", "--from", dest="date_from", - help=f"Lower bound of the {scope}.", + help=f"ISO 8601 lower timestamp bound of the {scope}.", ) from_action.completer = suppress_completion to_action = parser.add_argument( "-t", "--to", dest="date_to", - help=f"Upper bound of the {scope}.", + help=f"ISO 8601 upper timestamp bound of the {scope}.", ) to_action.completer = suppress_completion @@ -117,14 +117,14 @@ def add_type_filter_args( "-q", "--fulltext-json", default=None, - help="JSON object with fulltext search filters.", + help='JSON object with fulltext search filters, for example `{"name":"router"}`.', ) fulltext_action.completer = suppress_completion filter_action = parser.add_argument( "-j", "--filter-json", default=None, - help="JSON object with additional generic filters.", + help="JSON object using MongoDB query syntax.", ) filter_action.completer = suppress_completion has_attr_action = parser.add_argument( @@ -262,7 +262,7 @@ def _entity_attr_descriptions( model_spec, etype: str, entity_catalog: dict[str, Any] | None = None ) -> dict[str, str]: attrs = _entity_attrs(model_spec, etype, entity_catalog) - descriptions = {attr: f"Attribute on entity type '{etype}'." for attr in attrs} + descriptions = dict.fromkeys(attrs, f"Attribute on entity type '{etype}'.") if model_spec is not None and etype in model_spec.entities: for attr in attrs: attr_spec = model_spec.attr(etype, attr) diff --git a/dp3/bin/shcmd/entity/etype.py b/dp3/bin/shcmd/entity/etype.py index 03180896..975add45 100644 --- a/dp3/bin/shcmd/entity/etype.py +++ b/dp3/bin/shcmd/entity/etype.py @@ -3,7 +3,12 @@ import argparse -from dp3.bin.shcmd.common import print_response_json, stream_json_pages +from dp3.bin.shcmd.common import ( + MarkdownHelpArgumentParser, + command_description, + print_response_json, + stream_json_pages, +) from . import instance from .common import ( @@ -43,23 +48,42 @@ def handle_distinct(client, args) -> int: def build_parser(etype: str) -> argparse.ArgumentParser: """Build the parser for entity type-scope commands.""" - parser = argparse.ArgumentParser( + parser = MarkdownHelpArgumentParser( prog=f"dp3 sh entity {etype}", - description=(f"Query entities of type '{etype}' or use 'id' to inspect one entity by id."), + description=f"Query entities of type `{etype}` or select one entity by id.", ) parser.set_defaults(etype=etype) commands = parser.add_subparsers(dest="entity_type_command", required=True) - list_parser = commands.add_parser("list", help="List latest entity snapshots.") + list_parser = commands.add_parser( + "list", + help="List latest entity snapshots.", + description=f"List latest snapshots for entities of type `{etype}`.", + ) add_type_filter_args(list_parser, include_paging=True) add_ndjson_format_arg(list_parser) list_parser.set_defaults(handler=handle_list, etype=etype) - count_parser = commands.add_parser("count", help="Count latest entity snapshots.") + count_parser = commands.add_parser( + "count", + help="Count latest entity snapshots.", + description=command_description( + f"Count latest snapshots for entities of type `{etype}`. JSON filters use " + "MongoDB query syntax.", + f"dp3 sh entity {etype} count --filter-json '{{\"last.active\":true}}'", + ), + ) add_type_filter_args(count_parser, include_paging=False) count_parser.set_defaults(handler=handle_count, etype=etype) - raw_parser = commands.add_parser("raw", help=RAW_HELP) + raw_parser = commands.add_parser( + "raw", + help=RAW_HELP, + description=command_description( + f"Browse current raw datapoints for entities of type `{etype}`.", + f"dp3 sh entity {etype} raw --attr ATTR --src SOURCE --limit 100 --format ndjson", + ), + ) add_raw_filter_args(raw_parser) add_page_args(raw_parser, default_limit=20, subject="raw datapoints") add_ndjson_format_arg(raw_parser) @@ -68,7 +92,9 @@ def build_parser(etype: str) -> argparse.ArgumentParser: attr_values_parser = commands.add_parser( "attr-values", help="Get distinct latest values of one attribute across the entity type.", - description="Get distinct latest values of one attribute across the entity type.", + description=( + f"Get distinct latest values of an attribute across entities of type `{etype}`." + ), ) attr_action = attr_values_parser.add_argument( "attr", metavar="ATTR", help="Attribute to query across the entity type." diff --git a/dp3/bin/shcmd/entity/instance.py b/dp3/bin/shcmd/entity/instance.py index 8f196c01..9c41b7bd 100644 --- a/dp3/bin/shcmd/entity/instance.py +++ b/dp3/bin/shcmd/entity/instance.py @@ -4,6 +4,7 @@ import argparse from dp3.bin.shcmd.common import ( + command_description, common_time_params, print_response_json, read_json_value, @@ -65,21 +66,46 @@ def _add_instance_commands(parser: argparse.ArgumentParser, etype: str) -> None: """Register single-entity commands under a parser with an `eid` argument.""" commands = parser.add_subparsers(dest="entity_instance_command", required=True) - get_parser = commands.add_parser("get", help="Get full entity data.") + get_parser = commands.add_parser( + "get", + help="Get full entity data.", + description="Get full data for one entity.", + ) add_time_range_args(get_parser) get_parser.set_defaults(handler=handle_get, etype=etype) - master_parser = commands.add_parser("master", help="Get an entity master record.") + master_parser = commands.add_parser( + "master", + help="Get an entity master record.", + description="Get the master record for one entity.", + ) add_time_range_args(master_parser) master_parser.set_defaults(handler=handle_master, etype=etype) - snapshots_parser = commands.add_parser("snapshots", help="Get snapshots of a single entity.") + snapshots_parser = commands.add_parser( + "snapshots", + help="Get snapshots of a single entity.", + description=command_description( + "Get snapshots for one entity. Time bounds are ISO 8601 timestamps.", + f"dp3 sh entity {etype} id EID snapshots " + "--from 2024-01-01T00:00:00Z --to 2024-02-01T00:00:00Z " + "--limit 100 --format ndjson", + ), + ) add_time_range_args(snapshots_parser, scope="snapshot time range") add_page_args(snapshots_parser, default_limit=0, subject="snapshots") add_ndjson_format_arg(snapshots_parser) snapshots_parser.set_defaults(handler=handle_snapshots, etype=etype) - raw_parser = commands.add_parser("raw", help=RAW_HELP) + raw_parser = commands.add_parser( + "raw", + help=RAW_HELP, + description=command_description( + "Browse current raw datapoints for one entity.", + f"dp3 sh entity {etype} id EID raw --attr ATTR --src SOURCE " + "--limit 100 --format ndjson", + ), + ) add_raw_filter_args(raw_parser) add_page_args(raw_parser, default_limit=20, subject="raw datapoints") add_ndjson_format_arg(raw_parser) @@ -87,7 +113,15 @@ def _add_instance_commands(parser: argparse.ArgumentParser, etype: str) -> None: attr.add_instance_attr_parser(commands, etype) - ttl_parser = commands.add_parser("ttl", help="Extend entity TTLs.") + ttl_parser = commands.add_parser( + "ttl", + help="Extend entity TTLs.", + description=command_description( + "Extend TTLs for one entity using a JSON request body.", + f"dp3 sh entity {etype} id EID ttl " + '--body-json \'{"manual":"2030-01-01T00:00:00Z"}\'', + ), + ) body_action = ttl_parser.add_argument( "-b", "--body-json", @@ -97,7 +131,11 @@ def _add_instance_commands(parser: argparse.ArgumentParser, etype: str) -> None: body_action.completer = suppress_completion ttl_parser.set_defaults(handler=handle_ttl, etype=etype) - delete_parser = commands.add_parser("delete", help="Delete entity data.") + delete_parser = commands.add_parser( + "delete", + help="Delete entity data.", + description="Delete data for one entity.", + ) delete_parser.set_defaults(handler=handle_delete, etype=etype) @@ -106,7 +144,7 @@ def add_id_parser(commands, etype: str) -> None: id_parser = commands.add_parser( "id", help="Inspect or modify one entity by id.", - description=f"Inspect or modify one entity of type '{etype}' by id.", + description=f"Inspect or modify one entity of type `{etype}` by id.", ) id_parser.set_defaults(etype=etype) eid_action = id_parser.add_argument("eid", metavar="EID", help="Entity id.") diff --git a/dp3/bin/shcmd/health.py b/dp3/bin/shcmd/health.py index 157e41a3..24bb5096 100644 --- a/dp3/bin/shcmd/health.py +++ b/dp3/bin/shcmd/health.py @@ -11,5 +11,9 @@ def handle_health(client, _args) -> int: def register_parser(commands) -> None: """Register health commands on the root parser.""" - health_parser = commands.add_parser("health", help="Check whether the API is reachable.") + health_parser = commands.add_parser( + "health", + help="Check whether the API is reachable.", + description="Check whether the configured DP³ API is reachable.", + ) health_parser.set_defaults(handler=handle_health) diff --git a/dp3/bin/shcmd/telemetry.py b/dp3/bin/shcmd/telemetry.py index dbf7c3b6..1194aaae 100644 --- a/dp3/bin/shcmd/telemetry.py +++ b/dp3/bin/shcmd/telemetry.py @@ -1,7 +1,13 @@ #!/usr/bin/env python3 """Telemetry commands for the shell-oriented DP3 CLI.""" -from dp3.bin.shcmd.common import print_response_json, stream_json_pages +import json +import sys + +from event_count_logger import EventCountLogger + +from dp3.bin.shcmd.common import command_description, print_response_json, stream_json_pages +from dp3.common.config import read_config_dir def handle_sources_validity(client, _args) -> int: @@ -51,17 +57,60 @@ def handle_rabbitmq_queues(client, _args) -> int: return print_response_json(client.request("GET", "/telemetry/rabbitmq/queues")) +def handle_event_counts(_client, args) -> int: + """Read EventCountLogger counters from Redis.""" + try: + config = read_config_dir(args.config_dir, recursive=True) + groups = config.get("event_logging.groups") + redis_config = config.get("event_logging.redis") + group_config = groups.get(args.group) if isinstance(groups, dict) else None + if group_config is None: + raise ValueError(f"Event counter group '{args.group}' is not configured") + + intervals = group_config.get("intervals", []) + if args.interval not in intervals: + configured = ", ".join(intervals) or "none" + raise ValueError( + f"Interval '{args.interval}' is not configured for group " + f"'{args.group}' (configured: {configured})" + ) + + group = EventCountLogger(groups, redis_config).get_group(args.group) + result = {"group": args.group, "interval": args.interval} + if not args.current: + result["last"] = group.get_counts(args.interval) + if args.current or args.both: + result["current"] = group.get_counts(args.interval, current=True) + except Exception as e: + print(f"Cannot read event counters: {e}", file=sys.stderr) + return 1 + + json.dump(result, sys.stdout, sort_keys=True) + sys.stdout.write("\n") + return 0 + + def register_parser(commands) -> None: """Register telemetry commands on the root parser.""" - telemetry_parser = commands.add_parser("telemetry", help="Read operational telemetry.") + telemetry_parser = commands.add_parser( + "telemetry", + help="Read operational telemetry.", + description="Read operational telemetry from DP³ services.", + ) telemetry_commands = telemetry_parser.add_subparsers(dest="telemetry_command", required=True) sources_validity_parser = telemetry_commands.add_parser( - "sources-validity", help="Show source validity timestamps." + "sources-validity", + help="Show source validity timestamps.", + description="Show the latest datapoint validity timestamp observed for each source.", ) sources_validity_parser.set_defaults(handler=handle_sources_validity) - source_age_parser = telemetry_commands.add_parser("source-age", help="Show source ages.") + source_age_parser = telemetry_commands.add_parser( + "source-age", + help="Show source ages.", + description="Show the age of each source in the selected unit.", + ) source_age_parser.add_argument( "-u", "--unit", @@ -71,28 +120,97 @@ def register_parser(commands) -> None: source_age_parser.set_defaults(handler=handle_source_age) entities_per_attr_parser = telemetry_commands.add_parser( - "entities-per-attr", help="Count entities with data present for each attribute." + "entities-per-attr", + help="Count entities with data present for each attribute.", + description="Count entities with data present for each configured attribute.", ) entities_per_attr_parser.set_defaults(handler=handle_entities_per_attr) snapshot_summary_parser = telemetry_commands.add_parser( - "snapshot-summary", help="Show recent snapshot activity summary." + "snapshot-summary", + help="Show recent snapshot activity summary.", + description="Show a summary of recent snapshot activity.", ) snapshot_summary_parser.set_defaults(handler=handle_snapshot_summary) metadata_parser = telemetry_commands.add_parser( - "metadata", help="Browse internal metadata records." - ) - metadata_parser.add_argument("-m", "--module") - metadata_parser.add_argument("-f", "--from", dest="date_from") - metadata_parser.add_argument("-t", "--to", dest="date_to") - metadata_parser.add_argument("-s", "--skip", type=int, default=0) - metadata_parser.add_argument("-l", "--limit", type=int, default=0) - metadata_parser.add_argument("-S", "--sort", choices=["newest", "oldest"], default="newest") - metadata_parser.add_argument("-F", "--format", choices=["json", "ndjson"], default="ndjson") + "metadata", + help="Browse internal metadata records.", + description=command_description( + "Browse diagnostic records produced by internal periodic processes. Time bounds " + "are ISO 8601 timestamps.", + "dp3 sh telemetry metadata --module SnapShooter " + "--from 2024-01-01T00:00:00Z --sort oldest --limit 100 --format ndjson", + ), + ) + metadata_parser.add_argument("-m", "--module", help="Limit records to one module.") + metadata_parser.add_argument( + "-f", "--from", dest="date_from", help="ISO 8601 lower timestamp bound." + ) + metadata_parser.add_argument( + "-t", "--to", dest="date_to", help="ISO 8601 upper timestamp bound." + ) + metadata_parser.add_argument( + "-s", "--skip", type=int, default=0, help="Skip this many records." + ) + metadata_parser.add_argument( + "-l", "--limit", type=int, default=0, help="Return at most this many records." + ) + metadata_parser.add_argument( + "-S", + "--sort", + choices=["newest", "oldest"], + default="newest", + help="Select record ordering.", + ) + metadata_parser.add_argument( + "-F", + "--format", + choices=["json", "ndjson"], + default="ndjson", + help="Choose JSON or NDJSON output.", + ) metadata_parser.set_defaults(handler=handle_metadata) rabbitmq_queues_parser = telemetry_commands.add_parser( - "rabbitmq-queues", help="Show RabbitMQ queue telemetry." + "rabbitmq-queues", + help="Show RabbitMQ queue telemetry.", + description="Show queue sizes, consumers, and message rates for the application.", ) rabbitmq_queues_parser.set_defaults(handler=handle_rabbitmq_queues) + + event_counts_parser = telemetry_commands.add_parser( + "event-counts", + help="Read EventCountLogger counters from Redis.", + description=command_description( + "Read EventCountLogger counters directly from the configured Redis instance.", + "dp3 sh telemetry event-counts --group te --interval 5m --both", + ), + ) + event_counts_parser.add_argument( + "-g", "--group", required=True, help="Configured event counter group." + ) + event_counts_parser.add_argument( + "-i", "--interval", required=True, help="Configured counter interval." + ) + counter_period = event_counts_parser.add_mutually_exclusive_group() + counter_period.add_argument( + "--last", + action="store_true", + help="Show the last completed interval (default).", + ) + counter_period.add_argument( + "--current", + action="store_true", + help="Show the current incomplete interval.", + ) + counter_period.add_argument( + "--both", + action="store_true", + help="Show both the last and current intervals.", + ) + event_counts_parser.set_defaults( + handler=handle_event_counts, + requires_api=False, + load_model_spec=False, + ) diff --git a/dp3/common/callback_registrar.py b/dp3/common/callback_registrar.py index a74a6bfd..79ced557 100644 --- a/dp3/common/callback_registrar.py +++ b/dp3/common/callback_registrar.py @@ -15,6 +15,7 @@ from dp3.common.state import SharedFlag from dp3.common.task import DataPointTask, task_context from dp3.common.types import ParsedTimedelta +from dp3.common.utils import get_func_name from dp3.core.updater import Updater from dp3.snapshots.snapshooter import SnapShooter from dp3.task_processing.task_executor import TaskExecutor @@ -210,7 +211,7 @@ def register_allow_entity_creation_hook( """ Registers passed hook to allow entity creation. - Binds hook to specified entity (though same hook can be bound multiple times). + Binds hook to the specified entity. A hook can only be bound once to each entity. Args: hook: `hook` callable should expect eid and Task as arguments and return a bool. @@ -228,7 +229,7 @@ def register_on_entity_creation_hook( """ Registers passed hook to be called on entity creation. - Binds hook to specified entity (though same hook can be bound multiple times). + Binds hook to the specified entity. A hook can only be bound once to each entity. Allows registration of refreshing on configuration changes, if `refresh` is specified. In that case, `may_change` must be specified. @@ -255,7 +256,9 @@ def register_on_entity_creation_hook( [], may_change, ) - self._snap_shooter.register_run_finalize_hook(partial(unset_flag, refresh)) + self._snap_shooter.register_run_finalize_hook( + partial(unset_flag, refresh), get_func_name(hook), entity + ) def register_entity_hook(self, hook_type: str, hook: Callable, entity: str): """Registers one of available task entity hooks @@ -313,7 +316,9 @@ def register_on_new_attr_hook( [[attr]], may_change, ) - self._snap_shooter.register_run_finalize_hook(partial(unset_flag, refresh)) + self._snap_shooter.register_run_finalize_hook( + partial(unset_flag, refresh), get_func_name(hook), entity, attr + ) def register_attr_hook(self, hook_type: str, hook: Callable, entity: str, attr: str): """ @@ -337,8 +342,8 @@ def register_timeseries_hook( """ Registers passed timeseries hook to be called during snapshot creation. - Binds hook to specified `entity_type` and `attr_type` (though same hook can be bound - multiple times). + Binds hook to the specified `entity_type` and `attr_type`. A hook can only be bound + once to each entity attribute. Args: hook: `hook` callable should expect entity_type, attr_type and attribute @@ -362,7 +367,8 @@ def register_correlation_hook( """ Registers passed hook to be called during snapshot creation. - Binds hook to specified entity_type (though same hook can be bound multiple times). + Binds hook to the specified entity type and dependency context. Duplicate hook IDs + are rejected. `entity_type` and attribute specifications are validated, `ValueError` is raised on failure. @@ -396,7 +402,8 @@ def register_correlation_hook_with_master_record( Identical to `register_correlation_hook`, but the hook also receives the master record. - Binds hook to specified entity_type (though same hook can be bound multiple times). + Binds hook to the specified entity type and dependency context. Duplicate hook IDs + are rejected. `entity_type` and attribute specifications are validated, `ValueError` is raised on failure. diff --git a/dp3/common/hook_telemetry.py b/dp3/common/hook_telemetry.py new file mode 100644 index 00000000..f25fb4e4 --- /dev/null +++ b/dp3/common/hook_telemetry.py @@ -0,0 +1,60 @@ +"""Telemetry helpers for registered callback hooks.""" + +from collections.abc import Callable +from dataclasses import dataclass +from functools import partial +from inspect import unwrap +from time import perf_counter_ns +from typing import Generic, ParamSpec, TypeVar +from urllib.parse import quote + +from dp3.common.types import EventGroupType +from dp3.common.utils import get_stable_func_name + +P = ParamSpec("P") +R = TypeVar("R") + + +@dataclass(frozen=True) +class TrackedHook(Generic[P, R]): + """A callable hook paired with its metric prefix and event counter group.""" + + callback: Callable[P, R] + event_group: EventGroupType + metric_prefix: str + + def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R: + """Invoke the callback and record execution, failure, and duration counters.""" + self.log("executions") + started = perf_counter_ns() + try: + return self.callback(*args, **kwargs) + except Exception: + self.log("failures") + raise + finally: + self.log("duration_ns", perf_counter_ns() - started) + + def log(self, metric: str, count: int = 1) -> None: + """Add a hook-specific metric when its value is nonzero.""" + if count: + self.event_group.log(f"{self.metric_prefix}/{metric}", count=count) + + +class HookTelemetry: + """Wrap hooks with execution telemetry and stable metric identities.""" + + def __init__(self, event_group: EventGroupType): + self.event_group = event_group + + def wrap(self, hook_type: str, hook: Callable[P, R], *context: str) -> TrackedHook[P, R]: + """Return a callable hook that records telemetry under a stable identity.""" + callback = unwrap(hook) + while isinstance(callback, partial): + callback = unwrap(callback.func) + callback_name = get_stable_func_name(callback) + context_name = f"({','.join(context)})" + prefix = "/".join( + quote(part, safe="._-(),[]") for part in (hook_type, callback_name, context_name) + ) + return TrackedHook(hook, self.event_group, prefix) diff --git a/dp3/common/utils.py b/dp3/common/utils.py index a70c2f7c..a463286a 100644 --- a/dp3/common/utils.py +++ b/dp3/common/utils.py @@ -161,31 +161,44 @@ def batched(iterable: Iterable, n: int) -> Iterator[list]: # *** pretty print *** -def get_func_name(func_or_method): +def _format_stable_func_arg(arg) -> str: + if callable(arg): + return get_stable_func_name(arg) + if type(arg).__str__ is object.__str__: + return f"{arg.__class__.__module__}.{arg.__class__.__qualname__}" + return str(arg) + + +def _get_func_name(func_or_method, *, arg_formatter): """Get name of function or method as pretty string.""" if isinstance(func_or_method, partial): wrapper = "partial({name}, {args})" - args = [str(a) for a in func_or_method.args] - args.extend(f"{k}={v}" for k, v in func_or_method.keywords.items()) + args = [arg_formatter(arg) for arg in func_or_method.args] + args.extend( + f"{key}={arg_formatter(value)}" + for key, value in sorted(func_or_method.keywords.items()) + ) args = ", ".join(args) func_or_method = func_or_method.func else: wrapper = "{name}{args}" args = "" - try: - fname = func_or_method.__func__.__qualname__ - except AttributeError: - try: - fname = func_or_method.__name__ - except AttributeError: - fname = str(func_or_method) - - try: - module = func_or_method.__module__ - except AttributeError: - return fname - return wrapper.format(name=f"{module}.{fname}", args=args) + func = getattr(func_or_method, "__func__", func_or_method) + module = getattr(func, "__module__", None) + fname = getattr(func, "__qualname__", getattr(func, "__name__", None)) + if fname is None and callable(func): + module = func.__class__.__module__ + fname = func.__class__.__qualname__ + elif fname is None: + fname = str(func) + + name = f"{module}.{fname}" if module else fname + return wrapper.format(name=name, args=args) + + +get_func_name = partial(_get_func_name, arg_formatter=str) +get_stable_func_name = partial(_get_func_name, arg_formatter=_format_stable_func_arg) DEPENDENCY_LOGGERS = ( diff --git a/dp3/core/collector.py b/dp3/core/collector.py index 31801f7f..28c07861 100644 --- a/dp3/core/collector.py +++ b/dp3/core/collector.py @@ -4,8 +4,9 @@ import logging from collections import defaultdict +from collections.abc import Callable from datetime import UTC, datetime, timedelta -from functools import partial +from functools import partial, wraps from pydantic import BaseModel @@ -20,6 +21,16 @@ DB_SEND_CHUNK = 1000 +def _bind_ttl_extension(callback: Callable, extend_by: timedelta) -> Callable: + """Bind a TTL duration while preserving the extension callback's identity.""" + + @wraps(callback) + def wrapped(eid: AnyEidT, datapoint: DataPointBase): + return callback(eid, datapoint, extend_by=extend_by) + + return wrapped + + class GarbageCollectorConfig(BaseModel): """The configuration of the Collector module. @@ -121,7 +132,7 @@ def _register_ttl_extensions( registrar.register_attr_hook( "on_new_ts_chunk", - partial(self.extend_timeseries_ttl, extend_by=ttl), + _bind_ttl_extension(self.extend_timeseries_ttl, ttl), entity, attr, ) @@ -138,7 +149,7 @@ def _register_ttl_extensions( registrar.register_attr_hook( "on_new_observation", - partial(self.extend_observations_ttl, extend_by=ttl), + _bind_ttl_extension(self.extend_observations_ttl, ttl), entity, attr, ) @@ -148,7 +159,7 @@ def _register_ttl_extensions( registrar.register_attr_hook( "on_new_plain", - partial(self.extend_plain_ttl, extend_by=attr_spec.ttl), + _bind_ttl_extension(self.extend_plain_ttl, attr_spec.ttl), entity, attr, ) diff --git a/dp3/core/updater.py b/dp3/core/updater.py index 8526ce3c..b109eda7 100644 --- a/dp3/core/updater.py +++ b/dp3/core/updater.py @@ -7,11 +7,13 @@ from functools import partial from typing import Literal +from event_count_logger import DummyEventGroup from pydantic import BaseModel, validate_call from pymongo.cursor import Cursor from pymongo.results import UpdateResult from dp3.common.config import CronExpression, PlatformConfig +from dp3.common.hook_telemetry import HookTelemetry from dp3.common.scheduler import Scheduler from dp3.common.task import DataPointTask, task_context from dp3.common.types import EventGroupType, ParsedTimedelta @@ -195,9 +197,13 @@ def __init__( platform_config: PlatformConfig, scheduler: Scheduler, elog: EventGroupType, + hook_elog: EventGroupType | None = None, ): self.log = logging.getLogger("Updater") self.elog = elog + self.hook_telemetry = HookTelemetry( + hook_elog if hook_elog is not None else DummyEventGroup() + ) self.model_spec = platform_config.model_spec self.config = UpdaterConfig.model_validate(platform_config.config.get("updater", {})) @@ -262,7 +268,14 @@ def _register_hook(self, hook, hook_id: str, entity_type: str, period: float, ei period, eid_only, ) - hooks[hook_id] = hook + hook_type = "periodic_eid_update" if eid_only else "periodic_update" + hooks[hook_id] = self.hook_telemetry.wrap( + hook_type, + hook, + entity_type, + hook_id, + f"period={period:g}s", + ) def start(self): """ @@ -496,9 +509,11 @@ def _run_hooks(self, hooks: dict[str, Callable], entity_type: str, record: dict) try: new_tasks = hook(entity_type, record["_id"], record) tasks.extend(new_tasks) + if isinstance(new_tasks, list): + hook.log("created_tasks", len(new_tasks)) except Exception as e: self.elog.log("module_error") - self.log.error(f"Error during running hook {hook}: {e}") + self.log.error(f"Error during running hook {hook.callback}: {e}") for task in tasks: self.task_queue_writer.put_task(task) @@ -511,9 +526,11 @@ def _run_hooks_eid(self, hooks: dict[str, Callable], entity_type: str, record: d try: new_tasks = hook(entity_type, record["_id"]) tasks.extend(new_tasks) + if isinstance(new_tasks, list): + hook.log("created_tasks", len(new_tasks)) except Exception as e: self.elog.log("module_error") - self.log.error(f"Error during running hook {hook}: {e}") + self.log.error(f"Error during running hook {hook.callback}: {e}") for task in tasks: self.task_queue_writer.put_task(task) diff --git a/dp3/snapshots/snapshooter.py b/dp3/snapshots/snapshooter.py index 535dd6f8..6c20ac90 100644 --- a/dp3/snapshots/snapshooter.py +++ b/dp3/snapshots/snapshooter.py @@ -34,6 +34,7 @@ ObservationsHistoryParams, ) from dp3.common.config import CronExpression, PlatformConfig, entity_type_context +from dp3.common.hook_telemetry import HookTelemetry, TrackedHook from dp3.common.scheduler import Scheduler from dp3.common.task import ( DataPointTask, @@ -54,6 +55,8 @@ DB_SEND_CHUNK = 100 +SnapshotRunHook = TrackedHook[[], list[DataPointTask]] + class SnapShooterConfig(BaseModel): creation_rate: CronExpression = CronExpression(minute="*/30") @@ -70,6 +73,7 @@ def __init__( platform_config: PlatformConfig, scheduler: Scheduler, elog: EventGroupType | None = None, + hook_elog: EventGroupType | None = None, ) -> None: self.log = logging.getLogger("SnapShooter") @@ -87,11 +91,16 @@ def __init__( self.config = SnapShooterConfig.model_validate(platform_config.config.get("snapshots")) self.elog = elog or DummyEventGroup() + self.hook_telemetry = HookTelemetry(hook_elog or DummyEventGroup()) - self._timeseries_hooks = SnapshotTimeseriesHookContainer(self.log, self.model_spec, elog) - self._correlation_hooks = SnapshotCorrelationHookContainer(self.log, self.model_spec, elog) - self._init_hooks: list[Callable[[], list[DataPointTask]]] = [] - self._finalize_hooks: list[Callable[[], list[DataPointTask]]] = [] + self._timeseries_hooks = SnapshotTimeseriesHookContainer( + self.log, self.model_spec, self.elog, hook_elog + ) + self._correlation_hooks = SnapshotCorrelationHookContainer( + self.log, self.model_spec, self.elog, hook_elog + ) + self._init_hooks: list[SnapshotRunHook] = [] + self._finalize_hooks: list[SnapshotRunHook] = [] queue = f"{platform_config.app_name}-worker-{platform_config.process_index}-snapshots" self.snapshot_queue_reader = TaskQueueReader( @@ -168,8 +177,8 @@ def register_timeseries_hook( """ Registers passed timeseries hook to be called during snapshot creation. - Binds hook to specified `entity_type` and `attr_type` (though same hook can be bound - multiple times). + Binds hook to the specified `entity_type` and `attr_type`. A hook can only be bound + once to each entity attribute. Args: hook: `hook` callable should expect entity_type, attr_type and attribute @@ -195,7 +204,8 @@ def register_correlation_hook( Common implementation for hooks with and without master record. - Binds hook to specified entity_type (though same hook can be bound multiple times). + Binds hook to the specified entity_type and dependency context. Duplicate hook IDs + are rejected. `entity_type` and attribute specifications are validated, `ValueError` is raised on failure. @@ -218,23 +228,32 @@ def register_correlation_hook( def register_run_init_hook(self, hook: Callable[[], list[DataPointTask]]): """ - Registers passed hook to be called before a run of snapshot creation begins. + Registers passed hook to be called before a run of snapshot creation begins. Args: hook: `hook` callable should expect no arguments and return a list of DataPointTask objects to perform. """ - self._init_hooks.append(hook) + if any(registered.callback == hook for registered in self._init_hooks): + raise ValueError(f"Snapshot init hook '{get_func_name(hook)}' is already registered.") + self._init_hooks.append(self.hook_telemetry.wrap("snapshot_run_init", hook)) - def register_run_finalize_hook(self, hook: Callable[[], list[DataPointTask]]): + def register_run_finalize_hook(self, hook: Callable[[], list[DataPointTask]], *context: str): """ - Registers passed hook to be called after a run of snapshot creation ends. + Registers passed hook to be called after a run of snapshot creation ends. Args: hook: `hook` callable should expect no arguments and return a list of DataPointTask objects to perform. + context: Values identifying generated finalizers with the same callback signature. """ - self._finalize_hooks.append(hook) + if any(registered.callback == hook for registered in self._finalize_hooks): + raise ValueError( + f"Snapshot finalize hook '{get_func_name(hook)}' is already registered." + ) + self._finalize_hooks.append( + self.hook_telemetry.wrap("snapshot_run_finalize", hook, *context) + ) def make_snapshots(self): """Creates snapshots for all entities currently active in database.""" @@ -378,17 +397,19 @@ def process_snapshot_task(self, msg_id, task: Snapshot): else: raise ValueError("Unknown SnapshotMessageType.") - def _run_hooks(self, hooks: list[Callable[[], list[DataPointTask]]]): + def _run_hooks(self, hooks: list[SnapshotRunHook]): tasks = [] with task_context(self.model_spec): for hook in hooks: - self.log.debug("Running hook: '%s'", get_func_name(hook)) + self.log.debug("Running hook: '%s'", get_func_name(hook.callback)) try: new_tasks = hook() tasks.extend(new_tasks) + if isinstance(new_tasks, list): + hook.log("created_tasks", len(new_tasks)) except Exception as e: self.elog.log("module_error") - self.log.error(f"Error during running hook {hook}: {e}") + self.log.error(f"Error during running hook {hook.callback}: {e}") for task in tasks: self.task_queue_writer.put_task(task) diff --git a/dp3/snapshots/snapshot_hooks.py b/dp3/snapshots/snapshot_hooks.py index 2ffe3c1e..0afce04e 100644 --- a/dp3/snapshots/snapshot_hooks.py +++ b/dp3/snapshots/snapshot_hooks.py @@ -7,22 +7,38 @@ from collections.abc import Callable, Hashable from dataclasses import dataclass, field +from event_count_logger import DummyEventGroup + from dp3.common.attrspec import AttrType from dp3.common.config import ModelSpec +from dp3.common.hook_telemetry import HookTelemetry, TrackedHook from dp3.common.task import DataPointTask, task_context +from dp3.common.types import EventGroupType from dp3.common.utils import get_func_name -from dp3.task_processing.task_hooks import EventGroupType + +SnapshotTimeseriesHook = TrackedHook[[str, str, list[dict]], list[DataPointTask]] +SnapshotCorrelationHook = TrackedHook[ + [str, dict, dict], + list[DataPointTask] | None, +] class SnapshotTimeseriesHookContainer: """Container for timeseries analysis hooks""" - def __init__(self, log: logging.Logger, model_spec: ModelSpec, elog: EventGroupType): + def __init__( + self, + log: logging.Logger, + model_spec: ModelSpec, + elog: EventGroupType | None = None, + hook_elog: EventGroupType | None = None, + ): self.log = log.getChild("TimeseriesHooks") - self.elog = elog + self.elog = elog or DummyEventGroup() + self.telemetry = HookTelemetry(hook_elog or DummyEventGroup()) self.model_spec = model_spec - self._hooks = defaultdict(list) + self._hooks: defaultdict[tuple[str, str], list[SnapshotTimeseriesHook]] = defaultdict(list) def register( self, @@ -33,8 +49,8 @@ def register( """ Registers passed timeseries hook to be called during snapshot creation. - Binds hook to specified entity_type and attr_type (though same hook can be bound - multiple times). + Binds hook to the specified entity_type and attr_type. A hook can only be bound + once to each entity attribute. If entity_type and attr_type do not specify a valid timeseries attribute, a ValueError is raised. Args: @@ -48,7 +64,13 @@ def register( spec = self.model_spec.attributes[entity_type, attr_type] if spec.t != AttrType.TIMESERIES: raise ValueError(f"'{entity_type}.{attr_type}' is not a timeseries, but '{spec.t}'") - self._hooks[entity_type, attr_type].append(hook) + hooks = self._hooks[entity_type, attr_type] + if any(registered.callback == hook for registered in hooks): + raise ValueError( + f"Hook '{get_func_name(hook)}' is already registered for " + f"attribute '{entity_type}/{attr_type}'." + ) + hooks.append(self.telemetry.wrap("snapshot_timeseries", hook, entity_type, attr_type)) self.log.debug(f"Added hook: '{get_func_name(hook)}'") def run( @@ -61,21 +83,30 @@ def run( try: new_tasks = hook(entity_type, attr_type, attr_history) tasks.extend(new_tasks) + if isinstance(new_tasks, list): + hook.log("created_tasks", len(new_tasks)) except Exception as e: self.elog.log("module_error") - self.log.error(f"Error during running hook {hook}: {e}") + self.log.error(f"Error during running hook {hook.callback}: {e}") return tasks class SnapshotCorrelationHookContainer: """Container for data fusion and correlation hooks.""" - def __init__(self, log: logging.Logger, model_spec: ModelSpec, elog: EventGroupType): + def __init__( + self, + log: logging.Logger, + model_spec: ModelSpec, + elog: EventGroupType | None = None, + hook_elog: EventGroupType | None = None, + ): self.log = log.getChild("CorrelationHooks") - self.elog = elog + self.elog = elog or DummyEventGroup() + self.telemetry = HookTelemetry(hook_elog or DummyEventGroup()) self.model_spec = model_spec - self._hooks: defaultdict[str, list[tuple[str, Callable]]] = defaultdict(list) + self._hooks: defaultdict[str, list[tuple[str, SnapshotCorrelationHook]]] = defaultdict(list) self._short_hook_ids: dict = {} self._dependency_graph = DependencyGraph(self.log) @@ -91,7 +122,8 @@ def register( """ Registers passed hook to be called during snapshot creation. - Binds hook to specified entity_type (though same hook can be bound multiple times). + Binds hook to the specified entity_type and dependency context. Duplicate hook IDs + are rejected. If entity_type and attribute specifications are validated and ValueError is raised on failure. @@ -124,7 +156,8 @@ def register( self._short_hook_ids[hook_id] = hook_args self._dependency_graph.add_hook_dependency(hook_id, depends_on, may_change) - self._hooks[entity_type].append((hook_id, hook)) + tracked_hook = self.telemetry.wrap("snapshot_correlation", hook, entity_type) + self._hooks[entity_type].append((hook_id, tracked_hook)) self._restore_hook_order(self._hooks[entity_type]) self.log.info(f"Added hook: '{hook_id}'") @@ -218,6 +251,8 @@ def run(self, entities: dict, entity_master_records: dict) -> list[DataPointTask tasks = hook(etype, entity_values, entity_master_record) if tasks is not None and tasks: created_tasks.extend(tasks) + if isinstance(tasks, list): + hook.log("created_tasks", len(tasks)) except Exception as e: self.elog.log("module_error") self.log.error(f"Error during running hook {hook_id}: {e}") @@ -225,7 +260,7 @@ def run(self, entities: dict, entity_master_records: dict) -> list[DataPointTask return created_tasks - def _restore_hook_order(self, hooks: list[tuple[str, Callable]]): + def _restore_hook_order(self, hooks: list[tuple[str, SnapshotCorrelationHook]]): topological_order = self._dependency_graph.topological_sort() hooks.sort(key=lambda x: topological_order.index(x[0])) diff --git a/dp3/task_processing/task_executor.py b/dp3/task_processing/task_executor.py index ff8a487d..1ec150d1 100644 --- a/dp3/task_processing/task_executor.py +++ b/dp3/task_processing/task_executor.py @@ -31,6 +31,7 @@ def __init__( platform_config: PlatformConfig, elog: EventGroupType, elog_by_src: EventGroupType, + hook_elog: EventGroupType | None = None, ) -> None: # initialize task distribution @@ -46,6 +47,7 @@ def __init__( # Event logging self.elog = elog self.elog_by_src = elog_by_src + self.hook_elog = hook_elog if hook_elog is not None else DummyEventGroup() # Print warning if some event group is not configured not_configured_groups = [] if isinstance(self.elog, DummyEventGroup): @@ -60,19 +62,25 @@ def __init__( ) # Hooks - self._task_generic_hooks = TaskGenericHooksContainer(self.log, self.elog) + self._task_generic_hooks = TaskGenericHooksContainer(self.log, self.elog, self.hook_elog) self._task_entity_hooks = {} self._task_attr_hooks = {} for entity in self.model_spec.entities: self._task_entity_hooks[entity] = TaskEntityHooksContainer( - entity, self.model_spec, self.log, self.elog + entity, self.model_spec, self.log, self.elog, self.hook_elog ) for entity, attr in self.model_spec.attributes: attr_type = self.model_spec.attributes[entity, attr].t self._task_attr_hooks[entity, attr] = TaskAttrHooksContainer( - entity, attr, attr_type, self.model_spec, self.log, self.elog + entity, + attr, + attr_type, + self.model_spec, + self.log, + self.elog, + self.hook_elog, ) def register_task_hook(self, hook_type: str, hook: Callable): diff --git a/dp3/task_processing/task_hooks.py b/dp3/task_processing/task_hooks.py index 9e015f37..cb14448a 100644 --- a/dp3/task_processing/task_hooks.py +++ b/dp3/task_processing/task_hooks.py @@ -1,15 +1,27 @@ import logging from collections.abc import Callable +from typing import Any + +from event_count_logger import DummyEventGroup from dp3.common.attrspec import AttrType from dp3.common.config import ModelSpec from dp3.common.datapoint import DataPointBase from dp3.common.datatype import AnyEidT +from dp3.common.hook_telemetry import HookTelemetry, TrackedHook from dp3.common.hook_types import ATTR_TYPE_TO_ON_NEW_HOOK from dp3.common.task import DataPointTask, task_context from dp3.common.types import EventGroupType from dp3.common.utils import get_func_name +TaskStartHook = TrackedHook[[DataPointTask], Any] +AllowEntityCreationHook = TrackedHook[[AnyEidT, DataPointTask], bool] +OnEntityCreationHook = TrackedHook[[AnyEidT, DataPointTask], list[DataPointTask]] +OnNewAttributeHook = TrackedHook[ + [AnyEidT, DataPointBase], + list[DataPointTask] | None, +] + class TaskGenericHooksContainer: """Container for generic hooks @@ -19,15 +31,23 @@ class TaskGenericHooksContainer: - `on_task_start`: receives Task, no return value requirements """ - def __init__(self, log: logging.Logger, elog: EventGroupType): + def __init__( + self, + log: logging.Logger, + elog: EventGroupType, + hook_elog: EventGroupType | None = None, + ): self.log = log.getChild("genericHooks") self.elog = elog + self.telemetry = HookTelemetry(hook_elog if hook_elog is not None else DummyEventGroup()) - self._on_start = [] + self._on_start: list[TaskStartHook] = [] def register(self, hook_type: str, hook: Callable): if hook_type == "on_task_start": - self._on_start.append(hook) + if any(registered.callback == hook for registered in self._on_start): + raise ValueError(f"Hook '{get_func_name(hook)}' is already registered.") + self._on_start.append(self.telemetry.wrap(hook_type, hook)) else: raise ValueError(f"Hook type '{hook_type}' doesn't exist.") @@ -35,12 +55,11 @@ def register(self, hook_type: str, hook: Callable): def run_on_start(self, task: DataPointTask): for hook in self._on_start: - # Run hook try: hook(task) except Exception as e: self.elog.log("module_error") - self.log.error(f"Error during running hook {hook}: {e}") + self.log.error(f"Error during running hook {hook.callback}: {e}") class TaskEntityHooksContainer: @@ -54,37 +73,54 @@ class TaskEntityHooksContainer: """ def __init__( - self, entity: str, model_spec: ModelSpec, log: logging.Logger, elog: EventGroupType + self, + entity: str, + model_spec: ModelSpec, + log: logging.Logger, + elog: EventGroupType, + hook_elog: EventGroupType | None = None, ): self.entity = entity self.log = log.getChild(f"entityHooks.{entity}") self.elog = elog + self.telemetry = HookTelemetry(hook_elog if hook_elog is not None else DummyEventGroup()) self.model_spec = model_spec - self._allow_creation = [] - self._on_creation = [] + self._allow_creation: list[AllowEntityCreationHook] = [] + self._on_creation: list[OnEntityCreationHook] = [] def register(self, hook_type: str, hook: Callable): if hook_type == "allow_entity_creation": - self._allow_creation.append(hook) + hooks = self._allow_creation elif hook_type == "on_entity_creation": - self._on_creation.append(hook) + hooks = self._on_creation else: raise ValueError(f"Hook type '{hook_type}' doesn't exist.") + if any(registered.callback == hook for registered in hooks): + raise ValueError( + f"Hook '{get_func_name(hook)}' is already registered for entity '{self.entity}'." + ) + hooks.append(self.telemetry.wrap(hook_type, hook, self.entity)) + self.log.debug(f"Added '{hook_type}' hook: {get_func_name(hook)}") def run_allow_creation(self, eid: AnyEidT, task: DataPointTask): for hook in self._allow_creation: try: - if not hook(eid, task): + if hook(eid, task): + hook.log("decisions_allowed") + else: + hook.log("decisions_denied") self.log.debug( - f"Creation of eid '{eid}' prevented because hook '{hook}' returned False." + "Creation of eid '%s' prevented because hook '%s' returned False.", + eid, + get_func_name(hook.callback), ) return False except Exception as e: self.elog.log("module_error") - self.log.error(f"Error during running hook {get_func_name(hook)}: {e}") + self.log.error(f"Error during running hook {get_func_name(hook.callback)}: {e}") return True @@ -94,15 +130,13 @@ def run_on_creation(self, eid: AnyEidT, task: DataPointTask): with task_context(self.model_spec): for hook in self._on_creation: try: - # Run hook hook_new_tasks = hook(eid, task) - - # Append new tasks to process if isinstance(hook_new_tasks, list): + hook.log("created_tasks", len(hook_new_tasks)) new_tasks += hook_new_tasks except Exception as e: self.elog.log("module_error") - self.log.error(f"Error during running hook {hook}: {e}") + self.log.error(f"Error during running hook {hook.callback}: {e}") return new_tasks @@ -124,11 +158,13 @@ def __init__( model_spec: ModelSpec, log: logging.Logger, elog: EventGroupType, + hook_elog: EventGroupType | None = None, ): self.entity = entity self.attr = attr self.log = log.getChild(f"attributeHooks.{entity}.{attr}") self.elog = elog + self.telemetry = HookTelemetry(hook_elog if hook_elog is not None else DummyEventGroup()) self.model_spec = model_spec try: @@ -136,15 +172,19 @@ def __init__( except KeyError as e: raise ValueError(f"Invalid attribute type '{attr_type}'") from e - self._on_new = [] + self._on_new: list[OnNewAttributeHook] = [] def register(self, hook_type: str, hook: Callable): - if hook_type == self.on_new_hook_type: - self._on_new.append(hook) - else: + if hook_type != self.on_new_hook_type: raise ValueError( f"Hook type '{hook_type}' doesn't exist for {self.entity}/{self.attr}." ) + if any(registered.callback == hook for registered in self._on_new): + raise ValueError( + f"Hook '{get_func_name(hook)}' is already registered for " + f"attribute '{self.entity}/{self.attr}'." + ) + self._on_new.append(self.telemetry.wrap(hook_type, hook, self.entity, self.attr)) self.log.debug(f"Added '{hook_type}' hook: {get_func_name(hook)}") @@ -154,14 +194,12 @@ def run_on_new(self, eid: AnyEidT, dp: DataPointBase): with task_context(self.model_spec): for hook in self._on_new: try: - # Run hook hook_new_tasks = hook(eid, dp) - - # Append new tasks to process if isinstance(hook_new_tasks, list): + hook.log("created_tasks", len(hook_new_tasks)) new_tasks += hook_new_tasks except Exception as e: self.elog.log("module_error") - self.log.error(f"Error during running hook {hook}: {e}") + self.log.error(f"Error during running hook {hook.callback}: {e}") return new_tasks diff --git a/dp3/template/app/.dockerignore b/dp3/template/app/.dockerignore new file mode 100644 index 00000000..22985035 --- /dev/null +++ b/dp3/template/app/.dockerignore @@ -0,0 +1,27 @@ +# Version-control and local tooling +.git +.venv +venv +.ruff_cache +.mypy_cache +.pytest_cache +.tox +.nox +.idea +.vscode + +# Python build and test artifacts +**/__pycache__ +**/*.py[cod] +**/*.egg-info +build +dist +site +htmlcov +.coverage +.coverage.* + +# Local configuration, credentials, and logs +.env +.env.* +*.log diff --git a/dp3/template/app/config/event_logging.yml b/dp3/template/app/config/event_logging.yml index 4c1f2be3..34313104 100644 --- a/dp3/template/app/config/event_logging.yml +++ b/dp3/template/app/config/event_logging.yml @@ -18,10 +18,16 @@ groups: # Two intervals - 5 min and 2 hours for longer-term history in Munin/Icinga intervals: ["5m", "2h"] # Cache counts locally, push to Redis every second - sync-interval: 1 + sync_interval: 1 # Number of processed tasks by their "src" attribute tasks_by_src: events: [] auto_declare_events: true intervals: ["5m", "2h"] - sync-interval: 1 + sync_interval: 1 + # Execution statistics for callbacks registered with the task executor + secondary_hooks: + events: [] + auto_declare_events: true + intervals: ["5m", "2h"] + sync_interval: 1 diff --git a/dp3/template/app/docker/python/Dockerfile b/dp3/template/app/docker/python/Dockerfile index 81c6589f..035e4923 100644 --- a/dp3/template/app/docker/python/Dockerfile +++ b/dp3/template/app/docker/python/Dockerfile @@ -1,15 +1,21 @@ # syntax=docker/dockerfile:1 -# Base interpreter with installed requirements -FROM python:3.11-slim as base -RUN apt-get update; apt-get install -y git - -# Install requirements -WORKDIR /{{DP3_APP}}/ -COPY requirements.txt requirements.txt -RUN pip install --upgrade pip; \ - pip install -r requirements.txt - -# When the application changes, only the COPY . /{{DP3_APP}}/ line will be re-run -# (this means changes in modules, config files, etc. will not trigger a re-install of requirements) -COPY . /{{DP3_APP}}/ \ No newline at end of file +FROM python:3.11-slim AS base + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + UV_LINK_MODE=copy \ + UV_SYSTEM_PYTHON=1 + +RUN pip install --no-cache-dir "uv==0.9.7" + +RUN apt-get update && \ + apt-get install -y --no-install-recommends git && \ + rm -rf /var/lib/apt/lists/* + +WORKDIR /{{DP3_APP}} + +COPY requirements.txt ./ +RUN uv pip install --no-cache -r requirements.txt + +COPY . . diff --git a/dp3/template/app/docker/rabbitmq/.dockerignore b/dp3/template/app/docker/rabbitmq/.dockerignore new file mode 100644 index 00000000..31a1c19e --- /dev/null +++ b/dp3/template/app/docker/rabbitmq/.dockerignore @@ -0,0 +1,3 @@ +* +!Dockerfile +!init-rmq.sh diff --git a/dp3/template/app/docker/rabbitmq/Dockerfile b/dp3/template/app/docker/rabbitmq/Dockerfile index 9991aaa3..0b0286d4 100644 --- a/dp3/template/app/docker/rabbitmq/Dockerfile +++ b/dp3/template/app/docker/rabbitmq/Dockerfile @@ -2,10 +2,9 @@ FROM rabbitmq:3-management # Define environment variable indicating file in which the process id is placed for use by `rabbitmqctl wait` -ENV RABBITMQ_PID_FILE /var/lib/rabbitmq/mnesia/rabbitmq +ENV RABBITMQ_PID_FILE=/var/lib/rabbitmq/mnesia/rabbitmq # Add custom configuration script -ADD init-rmq.sh /init-rmq.sh -RUN chmod +x /init-rmq.sh +COPY --chmod=755 init-rmq.sh /init-rmq.sh CMD ["/init-rmq.sh"] diff --git a/dp3/testing/registrar.py b/dp3/testing/registrar.py index 3d9fa902..ee899ac8 100644 --- a/dp3/testing/registrar.py +++ b/dp3/testing/registrar.py @@ -361,7 +361,7 @@ def run_timeseries_hook( tasks: list[DataPointTask] = [] with task_context(self.model_spec): for hook in self._timeseries_hooks._hooks[entity_type, attr_type]: - hook_tasks = hook(entity_type, attr_type, attr_history) + hook_tasks = hook.callback(entity_type, attr_type, attr_history) tasks.extend(hook_tasks) return tasks @@ -400,7 +400,7 @@ def run_correlation_hooks_for_entities( for entity_key, record in entities.items(): if entity_key[0] != etype: continue - hook_tasks = hook(etype, record, master_records.get(entity_key, {})) + hook_tasks = hook.callback(etype, record, master_records.get(entity_key, {})) if hook_tasks is not None and hook_tasks: tasks.extend(hook_tasks) return tasks diff --git a/dp3/worker.py b/dp3/worker.py index 32563f19..1f39abcb 100755 --- a/dp3/worker.py +++ b/dp3/worker.py @@ -3,6 +3,7 @@ Don't run directly. Import and run the main() function. """ + import contextlib import faulthandler import inspect @@ -12,11 +13,12 @@ import sys import threading import time +from collections.abc import Callable from functools import partial from importlib import import_module import yaml -from event_count_logger import DummyEventGroup, EventCountLogger +from event_count_logger import DummyEventGroup, EventCountLogger, EventGroup from pydantic import ValidationError from dp3.common.callback_registrar import CallbackRegistrar, reload_module_config @@ -61,6 +63,25 @@ def _force_worker_shutdown(log: logging.Logger, message: str, *args) -> None: os._exit(1) +def _run_shutdown_callback( + callback: Callable[[], None], timeout: float, thread_name: str +) -> tuple[bool, Exception | None]: + """Run shutdown work in a daemon thread for at most ``timeout`` seconds.""" + error = None + + def run() -> None: + nonlocal error + try: + callback() + except Exception as exc: + error = exc + + thread = threading.Thread(target=run, name=thread_name, daemon=True) + thread.start() + thread.join(timeout=timeout) + return not thread.is_alive(), error + + def _stop_module(module, timeout: float, log: logging.Logger) -> tuple[bool, bool]: """Stop a module with a deadline. @@ -68,22 +89,38 @@ def _stop_module(module, timeout: float, log: logging.Logger) -> tuple[bool, boo A tuple of (completed, failed). ``completed`` is false when the stop call is still blocked after the timeout. ``failed`` is true when stop raised. """ - failed = False + completed, error = _run_shutdown_callback( + module.stop, timeout, f"{module.__class__.__name__}Stop" + ) + if error is not None: + log.error( + "Error while stopping %s", + module.__class__.__name__, + exc_info=(type(error), error, error.__traceback__), + ) + return completed, error is not None - def stop_module() -> None: - nonlocal failed - try: - module.stop() - except Exception: - failed = True - log.exception("Error while stopping %s", module.__class__.__name__) - stopper = threading.Thread( - target=stop_module, name=f"{module.__class__.__name__}Stop", daemon=True +def _flush_hook_telemetry(hook_elog: EventGroup, deadline_ts: float, log: logging.Logger) -> bool: + """Flush buffered hook telemetry within the worker shutdown deadline.""" + remaining = _remaining_time(deadline_ts) + if remaining == 0: + log.warning("Skipping secondary-hook telemetry flush: shutdown deadline exhausted") + return True + + completed, error = _run_shutdown_callback( + hook_elog.sync, remaining, "SecondaryHookTelemetryFlush" ) - stopper.start() - stopper.join(timeout=timeout) - return not stopper.is_alive(), failed + if not completed: + log.error("Secondary-hook telemetry flush did not finish before shutdown deadline") + return False + if error is not None: + log.error( + "Failed to flush secondary-hook telemetry during shutdown", + exc_info=(type(error), error, error.__traceback__), + ) + return False + return True def load_modules( @@ -182,6 +219,7 @@ def main(app_name: str, config_dir: str, process_index: int, verbose: bool) -> i running_modules: list[BaseModule] = [] # plug-in modules whose start() was attempted running_core_modules = [] # core modules whose start() was attempted signal_handlers_installed = False + hook_elog = None try: ############################################## @@ -228,6 +266,7 @@ def main(app_name: str, config_dir: str, process_index: int, verbose: bool) -> i ) elog = ecl.get_group("te") or DummyEventGroup() elog_by_src = ecl.get_group("tasks_by_src") or DummyEventGroup() + hook_elog = ecl.get_group("secondary_hooks") db = EntityDatabase(config, model_spec, num_processes, process_index, elog) if process_index == 0: @@ -236,13 +275,14 @@ def main(app_name: str, config_dir: str, process_index: int, verbose: bool) -> i db.await_updated_schema() global_scheduler = scheduler.Scheduler() - task_executor = TaskExecutor(db, platform_config, elog, elog_by_src) + task_executor = TaskExecutor(db, platform_config, elog, elog_by_src, hook_elog) snap_shooter = SnapShooter( db, TaskQueueWriter(app_name, num_processes, config.get("processing_core.msg_broker")), platform_config, global_scheduler, elog, + hook_elog, ) updater = Updater( db, @@ -250,6 +290,7 @@ def main(app_name: str, config_dir: str, process_index: int, verbose: bool) -> i platform_config, global_scheduler, elog, + hook_elog, ) registrar = CallbackRegistrar(global_scheduler, task_executor, snap_shooter, updater) @@ -377,6 +418,11 @@ def sigint_handler(signum, frame): module.__class__.__name__, ) + if hook_elog is not None and not _flush_hook_telemetry( + hook_elog, shutdown_deadline_ts, log + ): + exit_code = 1 + log.info("***** Finished, main thread exiting with code %d. *****", exit_code) logging.shutdown() diff --git a/macros.py b/macros.py index 661f542f..4935a391 100644 --- a/macros.py +++ b/macros.py @@ -1,5 +1,9 @@ +import argparse from urllib.parse import quote_plus +from dp3.bin.sh import build_parser as build_sh_parser +from dp3.bin.shcmd.entity.etype import build_parser as build_entity_type_parser + CESNET_APP_REPOS = ( "CESNET/Amfora", "CESNET/ADiCT", @@ -18,7 +22,126 @@ def _github_code_search_link(label: str, query: str) -> str: return f"[{label}]({GITHUB_CODE_SEARCH_URL.format(query=encoded_query)})" +def _subcommand_parser_items( + parser: argparse.ArgumentParser, +) -> list[tuple[str, argparse.ArgumentParser]]: + """Return unique subcommand names and parsers in argparse display order.""" + subcommands = [] + seen = set() + for action in parser._actions: + if not isinstance(action, argparse._SubParsersAction): + continue + for name, subparser in action.choices.items(): + parser_id = id(subparser) + if parser_id not in seen: + subcommands.append((name, subparser)) + seen.add(parser_id) + return subcommands + + +def _subcommand_parsers( + parser: argparse.ArgumentParser, +) -> list[argparse.ArgumentParser]: + """Return unique subcommand parsers in their argparse display order.""" + return [subparser for _, subparser in _subcommand_parser_items(parser)] + + +def _format_parser_help(parser: argparse.ArgumentParser) -> str: + """Format argparse help without descriptions or the help-option entry.""" + formatter = parser._get_formatter() + formatter.add_usage(parser.usage, parser._actions, parser._mutually_exclusive_groups) + for action_group in parser._action_groups: + actions = [action for action in action_group._group_actions if action.dest != "help"] + if not actions: + continue + formatter.start_section(action_group.title) + formatter.add_text(action_group.description) + formatter.add_arguments(actions) + formatter.end_section() + return formatter.format_help().rstrip() + + +def _render_argparse_tree( + parser: argparse.ArgumentParser, + heading_level: int = 2, + extra_children: dict[str, list[argparse.ArgumentParser]] | None = None, + heading: str | None = None, + flattened_parsers: dict[str, str] | None = None, +) -> str: + """Render a parser and its subcommands as Markdown help sections.""" + extra_children = extra_children or {} + flattened_parsers = flattened_parsers or {} + children = _subcommand_parser_items(parser) + children.extend( + ( + child.prog.removeprefix(f"{parser.prog} "), + child, + ) + for child in extra_children.get(parser.prog, []) + ) + + if parser.prog in flattened_parsers: + heading_prefix = flattened_parsers[parser.prog] + return "\n\n".join( + _render_argparse_tree( + child, + heading_level, + extra_children, + heading=" ".join(filter(None, (heading_prefix, child_name))), + flattened_parsers=flattened_parsers, + ) + for child_name, child in children + ) + + markdown_heading_level = min(heading_level, 6) + sections = [f"{'#' * markdown_heading_level} `{heading or parser.prog}`", ""] + if parser.description: + sections.extend([parser.description, ""]) + sections.extend(["```text", _format_parser_help(parser), "```"]) + for child_name, child in children: + sections.extend( + [ + "", + _render_argparse_tree( + child, + heading_level + 1, + extra_children, + heading=child_name, + flattened_parsers=flattened_parsers, + ), + ] + ) + return "\n".join(sections) + + +def dp3_sh_help() -> str: + """Render the complete `dp3 sh` command tree from its argparse help.""" + root_parser = build_sh_parser() + entity_type_parser = build_entity_type_parser("") + extra_children = {"dp3 sh entity": [entity_type_parser]} + flattened_parsers = { + "dp3 sh entity ": "", + "dp3 sh entity id": "id", + } + sections = ["```text", _format_parser_help(root_parser), "```"] + for child_name, child in _subcommand_parser_items(root_parser): + sections.extend( + [ + "", + _render_argparse_tree( + child, + extra_children=extra_children, + heading=child_name, + flattened_parsers=flattened_parsers, + ), + ] + ) + return "\n".join(sections) + + def define_env(env): + env.macro(dp3_sh_help) + @env.macro def query_cesnet_apps(label: str, query: str) -> str: full_query = f"{_repo_query(CESNET_APP_REPOS)} {query}" diff --git a/mkdocs.yml b/mkdocs.yml index 31c57b12..2f2ca6ff 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -50,6 +50,7 @@ nav: - Add an input module: howto/add-input.md - Add a secondary module: howto/add-module.md - Test a secondary module: howto/test-module.md + - Inspect telemetry: howto/telemetry.md - Deploy an app: howto/deploy-app.md - Develop DP3 itself: howto/develop-dp3.md - Extend Docs: howto/extending.md @@ -60,6 +61,7 @@ nav: - Modules: modules.md - Hooks: hooks.md - API: api.md + - CLI: cli.md - Configuration: - configuration/index.md - API: configuration/api.md diff --git a/pyproject.toml b/pyproject.toml index 36553e6c..d4893a99 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -49,7 +49,11 @@ Documentation = "https://cesnet.github.io/dp3" include = ["dp3*"] [tool.setuptools.package-data] -dp3 = ["template/**/*"] +dp3 = [ + "template/**/*", + "template/app/.dockerignore", + "template/app/docker/rabbitmq/.dockerignore", +] [tool.setuptools.dynamic] dependencies = { file = ["requirements.txt"] } diff --git a/requirements.txt b/requirements.txt index faa473d8..af369cd8 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,7 +1,7 @@ AMQPStorm~=2.7 apscheduler~=3.10 argcomplete~=3.6 -event-count-logger>=1.1 +event-count-logger @ git+https://github.com/xsedla1o/EventCountLogger.git@fix-sync_interval fastapi>=0.109.1 pydantic>=2.4.0 pymongo~=4.6 diff --git a/tests/test_api/test_raw.py b/tests/test_api/test_raw.py index 2c9a5170..b5dd7e0f 100644 --- a/tests/test_api/test_raw.py +++ b/tests/test_api/test_raw.py @@ -70,7 +70,7 @@ def test_get_raw_datapoints(self): self.assertNotIn("etype", page.data[0]) self.assertNotIn("eid", page.data[0]) - def test_get_plain_raw_datapoints_newest_first(self): + def test_get_plain_raw_datapoints(self): payload = self.query_expected_value( lambda: self.get_request( "entity/A/raw/get", @@ -82,5 +82,7 @@ def test_get_plain_raw_datapoints_newest_first(self): msg="Timed out waiting for plain raw datapoints to appear.", ) page = EntityRawDataPage.model_validate(payload) - self.assertEqual(["plain-2", "plain-1"], [item["v"] for item in page.data]) - self.assertEqual([9102, 9101], [item["id"] for item in page.data]) + self.assertCountEqual( + [(9101, "plain-1"), (9102, "plain-2")], + [(item["id"], item["v"]) for item in page.data], + ) diff --git a/tests/test_common/test_sh_completion.py b/tests/test_common/test_sh_completion.py index 3877ab83..e2497793 100644 --- a/tests/test_common/test_sh_completion.py +++ b/tests/test_common/test_sh_completion.py @@ -1,12 +1,16 @@ import argparse +import io +import json import os import unittest -from unittest.mock import patch +from contextlib import redirect_stdout +from unittest.mock import MagicMock, patch from argcomplete.finders import CompletionFinder from dp3.bin.cli import init_parser as init_root_parser from dp3.bin.sh import init_parser, render_completion_shellcode +from dp3.bin.shcmd import telemetry from dp3.bin.shcmd.common import complete_entity_type_names @@ -145,6 +149,58 @@ def test_telemetry_metadata_default_format_is_ndjson(self): args = self._parse_args(["telemetry", "metadata"]) self.assertEqual("ndjson", args.format) + def test_telemetry_event_counts_options_parse(self): + args = self._parse_args(["telemetry", "event-counts", "-g", "te", "-i", "5m", "--both"]) + self.assertEqual("te", args.group) + self.assertEqual("5m", args.interval) + self.assertTrue(args.both) + self.assertFalse(args.current) + self.assertFalse(args.requires_api) + + def test_telemetry_event_counts_reads_last_and_current(self): + event_group = MagicMock() + event_group.get_counts.side_effect = [ + {"task_processed": 10}, + {"task_processed": 2}, + ] + event_logger = MagicMock() + event_logger.get_group.return_value = event_group + args = argparse.Namespace( + config_dir="config", + group="te", + interval="5m", + current=False, + both=True, + ) + + output = io.StringIO() + with ( + patch.object( + telemetry, + "read_config_dir", + return_value={ + "event_logging.groups": {"te": {"intervals": ["5m"]}}, + "event_logging.redis": {"host": "redis"}, + }, + ), + patch.object(telemetry, "EventCountLogger", return_value=event_logger), + redirect_stdout(output), + ): + exit_code = telemetry.handle_event_counts(None, args) + + self.assertEqual(0, exit_code) + self.assertEqual( + { + "group": "te", + "interval": "5m", + "last": {"task_processed": 10}, + "current": {"task_processed": 2}, + }, + json.loads(output.getvalue()), + ) + event_group.get_counts.assert_any_call("5m") + event_group.get_counts.assert_any_call("5m", current=True) + def test_snapshot_option_completion(self): values = self._get_completions( ["dpsh", "--config", "tests/test_config", "entity", "A", "id", "10", "snapshots"], @@ -171,7 +227,7 @@ def test_snapshot_option_completion_includes_descriptions(self): ) self.assertIn("--from", values) self.assertEqual( - "Lower bound of the snapshot time range.", + "ISO 8601 lower timestamp bound of the snapshot time range.", finder._display_completions.get("--from"), ) diff --git a/tests/test_common/test_sh_docs.py b/tests/test_common/test_sh_docs.py new file mode 100644 index 00000000..7278140a --- /dev/null +++ b/tests/test_common/test_sh_docs.py @@ -0,0 +1,100 @@ +import argparse +import unittest + +from dp3.bin.sh import build_parser +from dp3.bin.shcmd.entity.etype import build_parser as build_entity_type_parser +from macros import ( + _format_parser_help, + _render_argparse_tree, + _subcommand_parser_items, + dp3_sh_help, +) + + +class TestShDocs(unittest.TestCase): + def test_standalone_parser_uses_dp3_sh_program_name(self): + parser = build_parser() + + self.assertEqual("dp3 sh", parser.prog) + self.assertEqual("health", parser.parse_args(["health"]).sh_command) + + def test_generated_help_uses_argparse_output_without_root_heading(self): + rendered = dp3_sh_help() + + self.assertIn("usage: dp3 sh", rendered) + self.assertNotIn("## `dp3 sh`", rendered) + self.assertIn("## `telemetry`", rendered) + self.assertIn("### `metadata`", rendered) + self.assertNotIn("### `dp3 sh telemetry`", rendered) + + def test_generated_help_places_rich_example_before_help_block(self): + rendered = dp3_sh_help() + + self.assertIn( + "### `metadata`\n\nBrowse diagnostic records produced by internal periodic processes. " + "Time bounds are ISO 8601 timestamps.\n\n```shell\n" + "dp3 sh telemetry metadata --module SnapShooter " + "--from 2024-01-01T00:00:00Z --sort oldest --limit 100 --format ndjson\n" + "```\n\n```text\n", + rendered, + ) + + def test_every_command_has_sentence_description(self): + def assert_descriptions(parser): + for _name, child in _subcommand_parser_items(parser): + summary = (child.description or "").partition("\n\n")[0] + self.assertTrue(summary.endswith("."), child.prog) + assert_descriptions(child) + + assert_descriptions(build_parser()) + assert_descriptions(build_entity_type_parser("")) + + def test_examples_are_limited_to_commands_where_they_explain_formats_or_options(self): + rendered = dp3_sh_help() + + self.assertNotIn("dp3 sh health\n", rendered) + self.assertNotIn("dp3 sh entity list --limit", rendered) + self.assertIn( + "dp3 sh entity id EID snapshots --from 2024-01-01T00:00:00Z", + rendered, + ) + self.assertIn( + "dp3 sh telemetry event-counts --group te --interval 5m --both", + rendered, + ) + + def test_formatted_help_omits_help_option_and_empty_options_section(self): + parser = argparse.ArgumentParser(prog="example") + + rendered = _format_parser_help(parser) + + self.assertIn("usage: example [-h]", rendered) + self.assertNotIn("-h, --help", rendered) + self.assertNotIn("options:", rendered) + + def test_generated_help_flattens_dynamic_entity_selectors(self): + rendered = dp3_sh_help() + + self.assertNotIn("### ``", rendered) + self.assertNotIn("### `id`", rendered) + self.assertIn("### `list`", rendered) + self.assertIn("### `id snapshots`", rendered) + self.assertIn("### `id attr`", rendered) + self.assertIn("#### `get`", rendered) + self.assertNotIn("#####", rendered) + self.assertIn("usage: dp3 sh entity id EID snapshots", rendered) + self.assertIn("--body-json BODY_JSON", rendered) + + def test_renderer_deduplicates_subcommand_aliases(self): + parser = argparse.ArgumentParser(prog="example") + commands = parser.add_subparsers() + commands.add_parser("status", aliases=["s"]) + + rendered = _render_argparse_tree(parser) + + self.assertEqual(1, rendered.count("### `status`")) + self.assertNotIn("### `s`", rendered) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_common/test_snapshot_hooks_telemetry.py b/tests/test_common/test_snapshot_hooks_telemetry.py new file mode 100644 index 00000000..da9e48df --- /dev/null +++ b/tests/test_common/test_snapshot_hooks_telemetry.py @@ -0,0 +1,197 @@ +import logging +import os +import unittest +from collections import Counter +from functools import partial +from unittest.mock import patch + +from dp3.common.callback_registrar import _drop_master +from dp3.common.config import ModelSpec, read_config_dir +from dp3.common.hook_telemetry import HookTelemetry +from dp3.snapshots.snapshooter import SnapShooter +from dp3.snapshots.snapshot_hooks import ( + SnapshotCorrelationHookContainer, + SnapshotTimeseriesHookContainer, +) + + +class RecordingEventGroup: + def __init__(self): + self.counts = Counter() + + def log(self, event_id, count=1): + self.counts[event_id] += count + + +class RecordingTaskQueueWriter: + def __init__(self): + self.tasks = [] + + def put_task(self, task): + self.tasks.append(task) + + +class TestSnapshotHookTelemetry(unittest.TestCase): + def setUp(self): + config = read_config_dir( + os.path.join(os.path.dirname(__file__), "..", "test_config"), recursive=True + ) + self.model_spec = ModelSpec(config.get("db_entities")) + self.log = logging.getLogger(self.id()) + self.task_events = RecordingEventGroup() + self.hook_events = RecordingEventGroup() + + def test_timeseries_hooks_record_tasks_failures_and_duration(self): + hooks = SnapshotTimeseriesHookContainer( + self.log, self.model_spec, self.task_events, self.hook_events + ) + + def create_tasks(_entity_type, _attr_type, _history): + return [object(), object()] + + def fail(_entity_type, _attr_type, _history): + raise RuntimeError("hook failed") + + context = ("test_entity_type", "test_attr_timeseries") + hooks.register(create_tasks, *context) + hooks.register(fail, *context) + successful_prefix = hooks._hooks[context][0].metric_prefix + failing_prefix = hooks._hooks[context][1].metric_prefix + + with patch( + "dp3.common.hook_telemetry.perf_counter_ns", + side_effect=[100, 125, 200, 240], + ): + tasks = hooks.run(*context, []) + + self.assertEqual(2, len(tasks)) + self.assertEqual(1, self.hook_events.counts[f"{successful_prefix}/executions"]) + self.assertEqual(2, self.hook_events.counts[f"{successful_prefix}/created_tasks"]) + self.assertEqual(25, self.hook_events.counts[f"{successful_prefix}/duration_ns"]) + self.assertEqual(1, self.hook_events.counts[f"{failing_prefix}/executions"]) + self.assertEqual(1, self.hook_events.counts[f"{failing_prefix}/failures"]) + self.assertEqual(40, self.hook_events.counts[f"{failing_prefix}/duration_ns"]) + self.assertEqual(1, self.task_events.counts["module_error"]) + + def test_timeseries_hooks_reject_duplicate_registration(self): + hooks = SnapshotTimeseriesHookContainer( + self.log, self.model_spec, self.task_events, self.hook_events + ) + + def hook(_entity_type, _attr_type, _history): + return [] + + context = ("test_entity_type", "test_attr_timeseries") + hooks.register(hook, *context) + with self.assertRaisesRegex(ValueError, "already registered"): + hooks.register(hook, *context) + + def test_correlation_hook_records_each_entity_execution_and_created_task(self): + hooks = SnapshotCorrelationHookContainer( + self.log, self.model_spec, self.task_events, self.hook_events + ) + + def create_task(_entity_type, _values, _master_record): + return [object()] + + hooks.register(create_task, "A", [["data1"]], [["data2"]]) + registered_hook = hooks._hooks["A"][0][1] + + with patch( + "dp3.common.hook_telemetry.perf_counter_ns", + side_effect=[10, 15, 20, 27], + ): + tasks = hooks.run({("A", "a1"): {}, ("A", "a2"): {}}, {}) + + prefix = registered_hook.metric_prefix + self.assertEqual(2, len(tasks)) + self.assertEqual(2, self.hook_events.counts[f"{prefix}/executions"]) + self.assertEqual(2, self.hook_events.counts[f"{prefix}/created_tasks"]) + self.assertEqual(12, self.hook_events.counts[f"{prefix}/duration_ns"]) + self.assertNotIn("depends_on", prefix) + self.assertNotIn("may_change", prefix) + self.assertTrue(prefix.endswith("/(A)")) + + def test_correlation_hooks_reject_duplicate_registration(self): + hooks = SnapshotCorrelationHookContainer( + self.log, self.model_spec, self.task_events, self.hook_events + ) + + def hook(_entity_type, _values, _master_record): + return [] + + hooks.register(hook, "A", [["data1"]], [["data2"]]) + with self.assertRaisesRegex(ValueError, "already present"): + hooks.register(hook, "A", [["data1"]], [["data2"]]) + + def test_correlation_dependencies_are_omitted_from_metric_identity(self): + def hook(_entity_type, _values, _master_record): + return [] + + first = SnapshotCorrelationHookContainer( + self.log, self.model_spec, self.task_events, self.hook_events + ) + second = SnapshotCorrelationHookContainer( + self.log, self.model_spec, self.task_events, self.hook_events + ) + first.register(hook, "A", [["data1"]], []) + second.register(hook, "A", [["data2"]], []) + + self.assertEqual( + first._hooks["A"][0][1].metric_prefix, + second._hooks["A"][0][1].metric_prefix, + ) + + def test_wrapped_partial_correlation_hooks_omit_bound_arguments(self): + def hook(_context, _entity_type, _values): + return [] + + telemetry = HookTelemetry(self.hook_events) + unbound = telemetry.wrap("snapshot_correlation", hook, "A") + first = telemetry.wrap("snapshot_correlation", _drop_master(partial(hook, "first")), "A") + second = telemetry.wrap("snapshot_correlation", _drop_master(partial(hook, "second")), "A") + + self.assertEqual(unbound.metric_prefix, first.metric_prefix) + self.assertEqual(unbound.metric_prefix, second.metric_prefix) + self.assertNotIn("partial(", first.metric_prefix) + + def test_snapshot_run_hooks_have_separate_families(self): + snapshooter = object.__new__(SnapShooter) + snapshooter.log = self.log + snapshooter.elog = self.task_events + snapshooter.model_spec = ModelSpec({}) + snapshooter.task_queue_writer = RecordingTaskQueueWriter() + snapshooter.hook_telemetry = HookTelemetry(self.hook_events) + snapshooter._init_hooks = [] + snapshooter._finalize_hooks = [] + + def create_task(): + return [object()] + + snapshooter.register_run_init_hook(create_task) + snapshooter.register_run_finalize_hook(create_task) + with self.assertRaisesRegex(ValueError, "already registered"): + snapshooter.register_run_init_hook(create_task) + with self.assertRaisesRegex(ValueError, "already registered"): + snapshooter.register_run_finalize_hook(create_task) + init_hook = snapshooter._init_hooks[0] + finalize_hook = snapshooter._finalize_hooks[0] + + with patch( + "dp3.common.hook_telemetry.perf_counter_ns", + side_effect=[100, 110, 200, 215], + ): + snapshooter._run_hooks(snapshooter._init_hooks) + snapshooter._run_hooks(snapshooter._finalize_hooks) + + self.assertTrue(init_hook.metric_prefix.startswith("snapshot_run_init/")) + self.assertTrue(finalize_hook.metric_prefix.startswith("snapshot_run_finalize/")) + self.assertEqual(1, self.hook_events.counts[f"{init_hook.metric_prefix}/created_tasks"]) + self.assertEqual(1, self.hook_events.counts[f"{finalize_hook.metric_prefix}/created_tasks"]) + self.assertEqual(10, self.hook_events.counts[f"{init_hook.metric_prefix}/duration_ns"]) + self.assertEqual(15, self.hook_events.counts[f"{finalize_hook.metric_prefix}/duration_ns"]) + self.assertEqual(2, len(snapshooter.task_queue_writer.tasks)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_common/test_task_hooks_telemetry.py b/tests/test_common/test_task_hooks_telemetry.py new file mode 100644 index 00000000..149af62a --- /dev/null +++ b/tests/test_common/test_task_hooks_telemetry.py @@ -0,0 +1,315 @@ +import logging +import unittest +from collections import Counter +from datetime import timedelta +from functools import partial +from unittest.mock import Mock, patch + +from event_count_logger import DummyEventGroup + +from dp3.common.attrspec import AttrType +from dp3.common.config import ModelSpec +from dp3.common.hook_telemetry import HookTelemetry +from dp3.common.utils import get_func_name, get_stable_func_name +from dp3.core.collector import _bind_ttl_extension +from dp3.task_processing.task_executor import TaskExecutor +from dp3.task_processing.task_hooks import ( + TaskAttrHooksContainer, + TaskEntityHooksContainer, + TaskGenericHooksContainer, +) + + +class RecordingEventGroup: + def __init__(self): + self.counts = Counter() + + def log(self, event_id, count=1): + self.counts[event_id] += count + + +class TestTaskHookTelemetry(unittest.TestCase): + def setUp(self): + self.log = logging.getLogger(self.id()) + self.task_events = RecordingEventGroup() + self.hook_events = RecordingEventGroup() + self.model_spec = ModelSpec({}) + + @staticmethod + def successful_task_hook(_task): + return None + + @staticmethod + def failing_task_hook(_task): + raise RuntimeError("hook failed") + + def test_generic_hook_records_execution_failure_and_duration(self): + hooks = TaskGenericHooksContainer(self.log, self.task_events, self.hook_events) + hooks.register("on_task_start", self.successful_task_hook) + hooks.register("on_task_start", self.failing_task_hook) + successful_prefix = hooks._on_start[0].metric_prefix + failing_prefix = hooks._on_start[1].metric_prefix + + with patch( + "dp3.common.hook_telemetry.perf_counter_ns", + side_effect=[100, 125, 200, 240], + ): + hooks.run_on_start(object()) + + self.assertEqual(1, self.hook_events.counts[f"{successful_prefix}/executions"]) + self.assertEqual(25, self.hook_events.counts[f"{successful_prefix}/duration_ns"]) + self.assertEqual(1, self.hook_events.counts[f"{failing_prefix}/executions"]) + self.assertEqual(1, self.hook_events.counts[f"{failing_prefix}/failures"]) + self.assertEqual(40, self.hook_events.counts[f"{failing_prefix}/duration_ns"]) + self.assertEqual(1, self.task_events.counts["module_error"]) + + def test_allow_creation_records_each_decision_and_short_circuits(self): + hooks = TaskEntityHooksContainer( + "device", self.model_spec, self.log, self.task_events, self.hook_events + ) + later_calls = [] + + def allow(_eid, _task): + return True + + def deny(_eid, _task): + return False + + def later(_eid, _task): + later_calls.append(True) + return True + + hooks.register("allow_entity_creation", allow) + hooks.register("allow_entity_creation", deny) + hooks.register("allow_entity_creation", later) + allowed_prefix = hooks._allow_creation[0].metric_prefix + denied_prefix = hooks._allow_creation[1].metric_prefix + later_prefix = hooks._allow_creation[2].metric_prefix + + with patch( + "dp3.common.hook_telemetry.perf_counter_ns", + side_effect=[10, 20, 30, 50], + ): + result = hooks.run_allow_creation("device-1", object()) + + self.assertFalse(result) + self.assertEqual(1, self.hook_events.counts[f"{allowed_prefix}/decisions_allowed"]) + self.assertEqual(1, self.hook_events.counts[f"{denied_prefix}/decisions_denied"]) + self.assertEqual(0, self.hook_events.counts[f"{later_prefix}/executions"]) + self.assertEqual([], later_calls) + + def test_task_creating_hooks_record_returned_task_count(self): + entity_hooks = TaskEntityHooksContainer( + "device", self.model_spec, self.log, self.task_events, self.hook_events + ) + attr_hooks = TaskAttrHooksContainer( + "device", + "hostname", + AttrType.PLAIN, + self.model_spec, + self.log, + self.task_events, + self.hook_events, + ) + entity_hooks.register("on_entity_creation", lambda _eid, _task: [object(), object()]) + attr_hooks.register("on_new_plain", lambda _eid, _dp: [object()]) + entity_prefix = entity_hooks._on_creation[0].metric_prefix + attr_prefix = attr_hooks._on_new[0].metric_prefix + + with patch( + "dp3.common.hook_telemetry.perf_counter_ns", + side_effect=[100, 110, 200, 215], + ): + entity_tasks = entity_hooks.run_on_creation("device-1", object()) + attr_tasks = attr_hooks.run_on_new("device-1", object()) + + self.assertEqual(2, len(entity_tasks)) + self.assertEqual(1, len(attr_tasks)) + self.assertEqual(2, self.hook_events.counts[f"{entity_prefix}/created_tasks"]) + self.assertEqual(1, self.hook_events.counts[f"{attr_prefix}/created_tasks"]) + + def test_task_hook_containers_reject_duplicate_registration(self): + containers = ( + ( + TaskGenericHooksContainer(self.log, self.task_events, self.hook_events), + "on_task_start", + ), + ( + TaskEntityHooksContainer( + "device", self.model_spec, self.log, self.task_events, self.hook_events + ), + "allow_entity_creation", + ), + ( + TaskEntityHooksContainer( + "device", self.model_spec, self.log, self.task_events, self.hook_events + ), + "on_entity_creation", + ), + ( + TaskAttrHooksContainer( + "device", + "hostname", + AttrType.PLAIN, + self.model_spec, + self.log, + self.task_events, + self.hook_events, + ), + "on_new_plain", + ), + ) + + for hooks, hook_type in containers: + with self.subTest(hook_type=hook_type): + hooks.register(hook_type, self.successful_task_hook) + with self.assertRaisesRegex(ValueError, "already registered"): + hooks.register(hook_type, self.successful_task_hook) + + def test_colliding_telemetry_identities_share_metric_prefix(self): + telemetry = HookTelemetry(self.hook_events) + + first = telemetry.wrap("on_task_start", self.successful_task_hook) + second = telemetry.wrap("on_task_start", self.successful_task_hook) + + self.assertEqual(first.metric_prefix, second.metric_prefix) + + def test_wrapped_hook_is_callable_and_forwards_keyword_arguments(self): + def callback(*, value): + return value + + hook = HookTelemetry(self.hook_events).wrap("on_task_start", callback) + + self.assertEqual("result", hook(value="result")) + self.assertEqual(1, self.hook_events.counts[f"{hook.metric_prefix}/executions"]) + + def test_partial_arguments_are_omitted_from_identity(self): + def callback(_context, _task): + return None + + telemetry = HookTelemetry(self.hook_events) + unbound = telemetry.wrap("on_task_start", callback) + first = telemetry.wrap("on_task_start", partial(callback, "a")) + second = telemetry.wrap("on_task_start", partial(callback, "b")) + + self.assertEqual(unbound.metric_prefix, first.metric_prefix) + self.assertEqual(unbound.metric_prefix, second.metric_prefix) + self.assertNotIn("partial(", first.metric_prefix) + + def test_partial_keyword_arguments_are_omitted_from_identity(self): + def callback(*, option): + return option + + telemetry = HookTelemetry(self.hook_events) + tracked = telemetry.wrap("on_task_start", partial(callback, option="value")) + + self.assertEqual( + telemetry.wrap("on_task_start", callback).metric_prefix, tracked.metric_prefix + ) + self.assertNotIn("option", tracked.metric_prefix) + self.assertNotIn("value", tracked.metric_prefix) + + def test_garbage_collector_ttl_binding_preserves_callback_identity(self): + calls = [] + + def callback(eid, datapoint, *, extend_by): + calls.append((eid, datapoint, extend_by)) + + telemetry = HookTelemetry(self.hook_events) + bound = _bind_ttl_extension(callback, timedelta(days=7)) + tracked = telemetry.wrap("on_new_observation", bound, "device", "activity") + unbound = telemetry.wrap("on_new_observation", callback, "device", "activity") + + self.assertEqual(unbound.metric_prefix, tracked.metric_prefix) + tracked("device-1", "datapoint") + self.assertEqual( + [("device-1", "datapoint", timedelta(days=7))], + calls, + ) + + def test_callable_partial_arguments_have_stable_names(self): + def callback(_bound, *, fallback): + return None + + def bound(): + return None + + class BoundObject: + pass + + name = get_stable_func_name(partial(callback, bound, BoundObject(), fallback=bound)) + + self.assertNotIn("0x", name) + self.assertEqual(2, name.count(get_stable_func_name(bound))) + self.assertIn(f"{BoundObject.__module__}.{BoundObject.__qualname__}", name) + + def test_default_partial_names_distinguish_bound_instances(self): + def callback(_bound): + return None + + class BoundObject: + pass + + first_bound = BoundObject() + second_bound = BoundObject() + first = get_func_name(partial(callback, first_bound)) + second = get_func_name(partial(callback, second_bound)) + + self.assertNotEqual(first, second) + + def test_hook_context_uses_one_namespace_component(self): + tracked = HookTelemetry(self.hook_events).wrap( + "on_new_plain", self.successful_task_hook, "device/site", "hostname" + ) + + parts = tracked.metric_prefix.split("/") + self.assertEqual(3, len(parts)) + self.assertEqual("on_new_plain", parts[0]) + self.assertIn("device%2Fsite", parts[2]) + + def test_context_components_are_unambiguously_encoded(self): + hooks = TaskEntityHooksContainer( + "device/site", self.model_spec, self.log, self.task_events, self.hook_events + ) + + hooks.register("allow_entity_creation", lambda _eid, _task: True) + + prefix = hooks._allow_creation[0].metric_prefix + self.assertIn("device%2Fsite", prefix) + + def test_task_hook_telemetry_is_optional(self): + generic = TaskGenericHooksContainer(self.log, self.task_events) + entity = TaskEntityHooksContainer("device", self.model_spec, self.log, self.task_events) + attr = TaskAttrHooksContainer( + "device", "hostname", AttrType.PLAIN, self.model_spec, self.log, self.task_events + ) + executor = TaskExecutor( + Mock(), Mock(model_spec=self.model_spec), self.task_events, self.task_events + ) + + self.assertIsInstance(generic.telemetry.event_group, DummyEventGroup) + self.assertIsInstance(entity.telemetry.event_group, DummyEventGroup) + self.assertIsInstance(attr.telemetry.event_group, DummyEventGroup) + self.assertIsInstance(executor.hook_elog, DummyEventGroup) + + def test_falsy_task_hook_event_group_is_preserved(self): + class FalsyEventGroup(RecordingEventGroup): + def __bool__(self): + return False + + hook_events = FalsyEventGroup() + generic = TaskGenericHooksContainer(self.log, self.task_events, hook_events) + executor = TaskExecutor( + Mock(), + Mock(model_spec=self.model_spec), + self.task_events, + self.task_events, + hook_events, + ) + + self.assertIs(generic.telemetry.event_group, hook_events) + self.assertIs(executor.hook_elog, hook_events) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_common/test_updater_hooks_telemetry.py b/tests/test_common/test_updater_hooks_telemetry.py new file mode 100644 index 00000000..24033e2e --- /dev/null +++ b/tests/test_common/test_updater_hooks_telemetry.py @@ -0,0 +1,110 @@ +import logging +import unittest +from collections import Counter, defaultdict +from datetime import timedelta +from types import SimpleNamespace +from unittest.mock import patch + +from dp3.common.config import ModelSpec +from dp3.common.hook_telemetry import HookTelemetry +from dp3.core.updater import Updater + + +class RecordingEventGroup: + def __init__(self): + self.counts = Counter() + + def log(self, event_id, count=1): + self.counts[event_id] += count + + +class RecordingTaskQueueWriter: + def __init__(self): + self.tasks = [] + + def put_task(self, task): + self.tasks.append(task) + + +class TestUpdaterHookTelemetry(unittest.TestCase): + def setUp(self): + self.task_events = RecordingEventGroup() + self.hook_events = RecordingEventGroup() + self.updater = object.__new__(Updater) + self.updater.log = logging.getLogger(self.id()) + self.updater.elog = self.task_events + self.updater.hook_telemetry = HookTelemetry(self.hook_events) + self.updater.model_spec = ModelSpec({}) + self.updater.task_queue_writer = RecordingTaskQueueWriter() + self.updater.enabled = True + self.updater.config = SimpleNamespace(update_batch_period=timedelta(seconds=5)) + self.updater.update_thread_hooks = defaultdict(dict) + + def test_record_hooks_record_tasks_failures_and_duration(self): + def create_tasks(_entity_type, _eid, _record): + return [object(), object()] + + def fail(_entity_type, _eid, _record): + raise RuntimeError("hook failed") + + self.updater._register_hook(create_tasks, "create", "device", 60, eid_only=False) + self.updater._register_hook(fail, "fail", "device", 60, eid_only=False) + hooks = self.updater.update_thread_hooks[60, "device", False] + successful_hook = hooks["create"] + failing_hook = hooks["fail"] + + with patch( + "dp3.common.hook_telemetry.perf_counter_ns", + side_effect=[100, 125, 200, 240], + ): + self.updater._run_hooks(hooks, "device", {"_id": "device-1"}) + + successful_prefix = successful_hook.metric_prefix + failing_prefix = failing_hook.metric_prefix + self.assertTrue(successful_prefix.startswith("periodic_update/")) + self.assertIn("(device,create,period%3D60s)", successful_prefix) + self.assertEqual(1, self.hook_events.counts[f"{successful_prefix}/executions"]) + self.assertEqual(2, self.hook_events.counts[f"{successful_prefix}/created_tasks"]) + self.assertEqual(25, self.hook_events.counts[f"{successful_prefix}/duration_ns"]) + self.assertEqual(1, self.hook_events.counts[f"{failing_prefix}/executions"]) + self.assertEqual(1, self.hook_events.counts[f"{failing_prefix}/failures"]) + self.assertEqual(40, self.hook_events.counts[f"{failing_prefix}/duration_ns"]) + self.assertEqual(1, self.task_events.counts["module_error"]) + self.assertEqual(2, len(self.updater.task_queue_writer.tasks)) + + def test_eid_hooks_use_a_separate_family(self): + def create_task(_entity_type, _eid): + return [object()] + + self.updater._register_hook(create_task, "refresh", "device", 120, eid_only=True) + hooks = self.updater.update_thread_hooks[120, "device", True] + hook = hooks["refresh"] + + with patch( + "dp3.common.hook_telemetry.perf_counter_ns", + side_effect=[10, 25], + ): + self.updater._run_hooks_eid(hooks, "device", {"_id": "device-1"}) + + prefix = hook.metric_prefix + self.assertTrue(prefix.startswith("periodic_eid_update/")) + self.assertIn("(device,refresh,period%3D120s)", prefix) + self.assertEqual(1, self.hook_events.counts[f"{prefix}/executions"]) + self.assertEqual(1, self.hook_events.counts[f"{prefix}/created_tasks"]) + self.assertEqual(15, self.hook_events.counts[f"{prefix}/duration_ns"]) + self.assertEqual(1, len(self.updater.task_queue_writer.tasks)) + + def test_period_is_part_of_the_metric_context(self): + def hook(_entity_type, _eid): + return [] + + self.updater._register_hook(hook, "refresh", "device", 60, eid_only=True) + self.updater._register_hook(hook, "refresh", "device", 120, eid_only=True) + + first = self.updater.update_thread_hooks[60, "device", True]["refresh"] + second = self.updater.update_thread_hooks[120, "device", True]["refresh"] + self.assertNotEqual(first.metric_prefix, second.metric_prefix) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_common/test_worker_lifecycle.py b/tests/test_common/test_worker_lifecycle.py index 83516793..4ab26852 100644 --- a/tests/test_common/test_worker_lifecycle.py +++ b/tests/test_common/test_worker_lifecycle.py @@ -39,6 +39,50 @@ def make_task_queue_reader(**overrides): class TestWorkerLifecycle(unittest.TestCase): + def test_shutdown_callback_is_bounded(self): + release = threading.Event() + started = time.monotonic() + + completed, error = worker._run_shutdown_callback(release.wait, 0.02, "BlockingShutdownTest") + + self.assertFalse(completed) + self.assertIsNone(error) + self.assertLess(time.monotonic() - started, 0.2) + release.set() + + def test_hook_telemetry_flush_uses_remaining_shutdown_deadline(self): + release = threading.Event() + hook_elog = Mock(sync=release.wait) + started = time.monotonic() + + flushed = worker._flush_hook_telemetry( + hook_elog, time.monotonic() + 0.02, logging.getLogger(self.id()) + ) + + self.assertFalse(flushed) + self.assertLess(time.monotonic() - started, 0.2) + release.set() + + def test_hook_telemetry_flush_skips_exhausted_deadline(self): + hook_elog = Mock() + + flushed = worker._flush_hook_telemetry( + hook_elog, time.monotonic() - 1, logging.getLogger(self.id()) + ) + + self.assertTrue(flushed) + hook_elog.sync.assert_not_called() + + def test_hook_telemetry_flush_reports_failure(self): + hook_elog = Mock() + hook_elog.sync.side_effect = RuntimeError("flush failed") + + flushed = worker._flush_hook_telemetry( + hook_elog, time.monotonic() + 1, logging.getLogger(self.id()) + ) + + self.assertFalse(flushed) + def test_worker_main_returns_failure_on_redis_startup_error(self): original_thread_name = threading.current_thread().name try: diff --git a/tests/test_config/event_logging.yml b/tests/test_config/event_logging.yml index b300b3fa..38549797 100644 --- a/tests/test_config/event_logging.yml +++ b/tests/test_config/event_logging.yml @@ -18,10 +18,16 @@ groups: # Two intervals - 5 min and 2 hours for longer-term history in Munin/Icinga intervals: ["5m", "2h"] # Cache counts locally, push to Redis every second - sync-interval: 1 + sync_interval: 1 # Number of processed tasks by their "src" attribute tasks_by_src: events: [] auto_declare_events: true intervals: ["5m", "2h"] - sync-interval: 1 + sync_interval: 1 + # Execution statistics for callbacks registered with the task executor + secondary_hooks: + events: [] + auto_declare_events: true + intervals: ["5m", "2h"] + sync_interval: 1 diff --git a/tests/test_example/config/event_logging.yml b/tests/test_example/config/event_logging.yml index b300b3fa..38549797 100644 --- a/tests/test_example/config/event_logging.yml +++ b/tests/test_example/config/event_logging.yml @@ -18,10 +18,16 @@ groups: # Two intervals - 5 min and 2 hours for longer-term history in Munin/Icinga intervals: ["5m", "2h"] # Cache counts locally, push to Redis every second - sync-interval: 1 + sync_interval: 1 # Number of processed tasks by their "src" attribute tasks_by_src: events: [] auto_declare_events: true intervals: ["5m", "2h"] - sync-interval: 1 + sync_interval: 1 + # Execution statistics for callbacks registered with the task executor + secondary_hooks: + events: [] + auto_declare_events: true + intervals: ["5m", "2h"] + sync_interval: 1