diff --git a/docker/compose.all5-daemon.yaml b/docker/compose.all5-daemon.yaml new file mode 100644 index 0000000..d87d7ae --- /dev/null +++ b/docker/compose.all5-daemon.yaml @@ -0,0 +1,73 @@ +services: + validator: + build: + context: .. + dockerfile: docker/Dockerfile + image: "${PANDA_RPKI_IMAGE:-panda-rpki:v0.1.0}" + user: "${PUID:-1000}:${PGID:-1000}" + read_only: true + init: true + restart: unless-stopped + stop_grace_period: 40s + tmpfs: + - /tmp + volumes: + - ${INPUT_DIR:?set INPUT_DIR to the TAL/TA directory}:/input:ro + - ${DAEMON_DIR:?set DAEMON_DIR to a writable dedicated directory}:/data + environment: + PANDA_RPKI_LOG_LEVEL: "${PANDA_RPKI_LOG_LEVEL:-info}" + PANDA_RPKI_LOG_FORMAT: "${PANDA_RPKI_LOG_FORMAT:-text}" + PANDA_RPKI_LOG_PROGRESS_INTERVAL_SECS: "${PANDA_RPKI_LOG_PROGRESS_INTERVAL_SECS:-10}" + command: + - daemon + - --state-root + - /data + - --interval-secs + - "${PANDA_RPKI_INTERVAL_SECS:-60}" + - --retain-runs + - "${PANDA_RPKI_RETAIN_RUNS:-10}" + - --run-timeout-secs + - "${PANDA_RPKI_RUN_TIMEOUT_SECS:-0}" + - --shutdown-grace-secs + - "${PANDA_RPKI_SHUTDOWN_GRACE_SECS:-30}" + - -- + - --tal + - /input/${AFRINIC_TAL_FILE:?set AFRINIC_TAL_FILE} + - --ta + - /input/${AFRINIC_TA_FILE:?set AFRINIC_TA_FILE} + - --tal-id + - afrinic + - --tal + - /input/${APNIC_TAL_FILE:?set APNIC_TAL_FILE} + - --ta + - /input/${APNIC_TA_FILE:?set APNIC_TA_FILE} + - --tal-id + - apnic + - --tal + - /input/${ARIN_TAL_FILE:?set ARIN_TAL_FILE} + - --ta + - /input/${ARIN_TA_FILE:?set ARIN_TA_FILE} + - --tal-id + - arin + - --tal + - /input/${LACNIC_TAL_FILE:?set LACNIC_TAL_FILE} + - --ta + - /input/${LACNIC_TA_FILE:?set LACNIC_TA_FILE} + - --tal-id + - lacnic + - --tal + - /input/${RIPE_TAL_FILE:?set RIPE_TAL_FILE} + - --ta + - /input/${RIPE_TA_FILE:?set RIPE_TA_FILE} + - --tal-id + - ripe + - --parallel-phase2-object-workers + - "${PANDA_RPKI_WORKERS:-8}" + - --parallel-max-repo-sync-workers-global + - "${PANDA_RPKI_REPO_WORKERS:-8}" + - --parallel-phase2-worker-queue-capacity + - "${PANDA_RPKI_WORKER_QUEUE_CAPACITY:-256}" + - --parallel-repo-worker-queue-capacity + - "${PANDA_RPKI_REPO_QUEUE_CAPACITY:-256}" + - --http-timeout-secs + - "${PANDA_RPKI_HTTP_TIMEOUT_SECS:-600}" diff --git a/docker/compose.daemon.yaml b/docker/compose.daemon.yaml index 0c122de..344ecc5 100644 --- a/docker/compose.daemon.yaml +++ b/docker/compose.daemon.yaml @@ -16,7 +16,7 @@ services: - ${DAEMON_DIR:?set DAEMON_DIR to a writable dedicated directory}:/data environment: PANDA_RPKI_LOG_LEVEL: "${PANDA_RPKI_LOG_LEVEL:-info}" - PANDA_RPKI_LOG_FORMAT: "${PANDA_RPKI_LOG_FORMAT:-json}" + PANDA_RPKI_LOG_FORMAT: "${PANDA_RPKI_LOG_FORMAT:-text}" command: - daemon - --state-root diff --git a/docs/command-line-reference.md b/docs/command-line-reference.md index 36469ca..6b42c9c 100644 --- a/docs/command-line-reference.md +++ b/docs/command-line-reference.md @@ -73,6 +73,7 @@ can increase memory consumption and does not guarantee proportional speedup. | Option | Default | Values | | --- | --- | --- | | `--log-level ` | `info` | `off`, `error`, `warn`, `info`, `debug`, `trace` | +| `--log-progress-interval-secs ` | `10` | Periodic progress; `0` disables periodic events, not stage events. CLI overrides `PANDA_RPKI_LOG_PROGRESS_INTERVAL_SECS`. | | `--log-format ` | `text` | `text`, `json` | Precedence: explicit option, then `PANDA_RPKI_LOG_LEVEL` or @@ -125,7 +126,7 @@ has stopped before restarting. Stop the entire container for Docker deployments. Daemon adds `daemon.lock`, `daemon-status.json`, `lifecycle.json`, `run-summary.jsonl`, `state/repository-db/`, and `runs/run_000001/` etc. Each run includes validator outputs, `run-meta.json`, `run-summary.json`, -`stdout.log`, `stderr.log` and optional `process-time.txt`. `/usr/bin/time` +`process-output.txt`, `validator.log` and optional `process-time.txt`. `/usr/bin/time` provides user/system CPU and peak RSS when installed; Docker includes it. Controller states: `starting`, `running`, `sleeping`, `exited`, `failed`. diff --git a/docs/docker.md b/docs/docker.md index e10651b..95758c4 100644 --- a/docs/docker.md +++ b/docs/docker.md @@ -47,6 +47,7 @@ docker compose -f docker/compose.all5.yaml run --build --rm validator | `PANDA_RPKI_WORKERS` | 8 | all5 and daemon object workers | | `PANDA_RPKI_REPO_WORKERS` | 8 | all5 and daemon transport workers | | `PANDA_RPKI_WORKER_QUEUE_CAPACITY` | 256 | all5 object queue | +| `PANDA_RPKI_REPO_QUEUE_CAPACITY` | 256 | all5 daemon repository queue | | `PANDA_RPKI_HTTP_TIMEOUT_SECS` | 600 | all5 HTTP timeout (native CLI default is 300) | | `PANDA_RPKI_RRDP_SYNC_MODE` | auto | single-anchor and all5 | @@ -54,7 +55,121 @@ The all5 CCR is `OUTPUT_DIR/all5.ccr`; the single-anchor CCR is `OUTPUT_DIR/result.ccr`. Reuse `STATE_DIR` and select new `OUTPUT_DIR` values to retain multiple cycles. These templates take no bundled RIR input files. -## Daemon service +## All-five daemon service + +The all-five daemon template combines the five regional TAL/TA pairs with the +continuous daemon controller. Prepare the same `input/` directory used by the +all-five one-shot example, then run: + +```bash +mkdir -p data/all5-daemon +export INPUT_DIR="$PWD/input" DAEMON_DIR="$PWD/data/all5-daemon" +export AFRINIC_TAL_FILE=afrinic.tal AFRINIC_TA_FILE=afrinic.cer +export APNIC_TAL_FILE=apnic.tal APNIC_TA_FILE=apnic.cer +export ARIN_TAL_FILE=arin.tal ARIN_TA_FILE=arin.cer +export LACNIC_TAL_FILE=lacnic.tal LACNIC_TA_FILE=lacnic.cer +export RIPE_TAL_FILE=ripe.tal RIPE_TA_FILE=ripe.cer +export PUID="$(id -u)" PGID="$(id -g)" +docker compose -f docker/compose.all5-daemon.yaml up -d --build +docker compose -f docker/compose.all5-daemon.yaml logs -f validator +``` + +The template uses a 60-second interval by default, keeps 10 completed runs, +and leaves the daemon running until it is stopped. It writes each run below +`DAEMON_DIR/runs/` and keeps shared protocol state below `DAEMON_DIR/state/`. +The default diagnostic format is `text`, with a source file and line appended +to each diagnostic record; select `json` explicitly for machine consumers. +Set `PANDA_RPKI_INTERVAL_SECS` only when a different interval is intended. Stop +the service without deleting its data with: + +```bash +docker compose -f docker/compose.all5-daemon.yaml stop +``` + +## Upgrade a daemon deployment + +An upgrade replaces the image and recreates the container. It does not replace +the host data directory. Keep the same `DAEMON_DIR`, Compose project, TAL/TA +inputs and controller settings so the new image can resume the existing +protocol state. Do not run two writable daemons against the same state root. + +Run the following from the deployment directory that owns the Compose file. Use +a unique image tag for every candidate so the image used by a run is auditable: + +```bash +export COMPOSE_FILE="${COMPOSE_FILE:-docker/compose.all5-daemon.yaml}" +export COMPOSE_PROJECT_NAME=panda-rpki-all5 +export PANDA_RPKI_IMAGE=panda-rpki:2026-09-11-logging +export SOURCE_DIR=/path/to/panda-rpki-oss + +# Keep the existing INPUT_DIR, DAEMON_DIR, TAL/TA variables and settings. +docker compose -p "$COMPOSE_PROJECT_NAME" -f "$COMPOSE_FILE" ps +test "$(jq -r .state "$DAEMON_DIR/daemon-status.json")" = sleeping + +# Build from the source tree before stopping the running service. +docker build -f "$SOURCE_DIR/docker/Dockerfile" -t "$PANDA_RPKI_IMAGE" "$SOURCE_DIR" +docker image inspect "$PANDA_RPKI_IMAGE" --format '{{.Id}}' + +# Save configuration and create a consistent stopped-state backup. +backup="upgrade-backup-$(date -u +%Y%m%dT%H%M%SZ)" +mkdir -p "$backup" +cp "$COMPOSE_FILE" "$backup/compose.yaml" +if test -f .env; then cp .env "$backup/env"; fi +docker compose -p "$COMPOSE_PROJECT_NAME" -f "$COMPOSE_FILE" stop +cp -a "$DAEMON_DIR/state" "$backup/state" +cp -a "$DAEMON_DIR/daemon-status.json" "$DAEMON_DIR/lifecycle.json" "$backup/" + +# Recreate the same service with the new image; bind-mounted data is retained. +docker compose -p "$COMPOSE_PROJECT_NAME" -f "$COMPOSE_FILE" up -d --no-build --force-recreate +docker inspect "$(docker compose -p "$COMPOSE_PROJECT_NAME" -f "$COMPOSE_FILE" ps -q validator)" \ + --format 'image={{.Config.Image}} running={{.State.Running}}' +``` + +After the first upgraded run, verify that the run sequence advanced, the run +completed, and the new log/outputs exist. The exact run directory is printed by +the daemon status and can be selected with `find`: + +```bash +latest_run="$(find "$DAEMON_DIR/runs" -mindepth 1 -maxdepth 1 -type d -name 'run_*' | sort | tail -1)" +test -s "$latest_run/run-summary.json" +test -s "$latest_run/validator.log" +test -s "$latest_run/result.ccr" +grep -Eq '\[src/[^]]+\.rs:[0-9]+\]$' "$latest_run/validator.log" +jq . "$latest_run/run-summary.json" +``` + +The upgrade process preserves completed run directories until normal retention +removes them. Archive important runs separately if they must outlive +`PANDA_RPKI_RETAIN_RUNS`. + +### Roll back + +Stop the upgraded service before changing the image. Restore the previous +Compose file and `.env`, or set `PANDA_RPKI_IMAGE` to the previously verified +tag, then recreate the service with the same project and data directory: + +```bash +docker compose -p "$COMPOSE_PROJECT_NAME" -f "$COMPOSE_FILE" stop +cp "$backup/compose.yaml" "$COMPOSE_FILE" +if test -f "$backup/env"; then cp "$backup/env" .env; fi +unset PANDA_RPKI_IMAGE +docker compose -p "$COMPOSE_PROJECT_NAME" -f "$COMPOSE_FILE" up -d --no-build --force-recreate +``` + +If the new binary changed the state schema or wrote incompatible state, stop the +service and move the current state aside before restoring the matching +stopped-state backup: + +```bash +mv "$DAEMON_DIR/state" "$DAEMON_DIR/state-incompatible-$(date -u +%Y%m%dT%H%M%SZ)" +cp -a "$backup/state" "$DAEMON_DIR/state" +``` + +Never delete `DAEMON_DIR` or use `docker compose down -v` as part of a routine +upgrade. Check [state compatibility](command-line-reference.md#state-and-network) +before deploying a binary with storage changes. + +## Single-anchor daemon service ```bash mkdir -p data/daemon @@ -69,7 +184,7 @@ docker compose -f docker/compose.daemon.yaml stop ``` Defaults: interval 60 seconds, retain 10, `PANDA_RPKI_RUN_TIMEOUT_SECS=0` -(disabled), log level `info`, log format `json`. The daemon template passes +(disabled), log level `info`, log format `text`. The daemon template passes `PANDA_RPKI_LOG_LEVEL` and `PANDA_RPKI_LOG_FORMAT` into the container. It restarts unless stopped and allows 40 seconds for shutdown, exceeding the controller's default 30-second grace plus termination wait. diff --git a/docs/logging.md b/docs/logging.md new file mode 100644 index 0000000..ecaee80 --- /dev/null +++ b/docs/logging.md @@ -0,0 +1,124 @@ +# Diagnostic logging + +The validator writes diagnostics to stderr. Under daemon, that stream is saved +as `runs/run_XXXXXX/validator.log`; it contains all enabled levels, not only errors. +`process-output.txt` captures ordinary output and is normally empty. The daemon's +own lifecycle diagnostics remain in the container log. No duplicate log forwarding +is performed. Historical `stderr.log` files are not renamed. + +CLI and both daemon Compose files default to `info` and `text`. +Text is written for people, with independently authored English sentences for +run lifecycle, validation stages, progress, output files and failures. It does not +enumerate event fields or translate event identifiers into prose. Progress reports +elapsed time, publication points processed when known, and active repository sync +attempts. Old progress snapshots are explicitly described as the last reported +update. UUIDs, worker queues and unrelated counters are not printed. +The synthetic initializing-stage completion is omitted in text only. Control +characters are escaped to keep each record on one line. Diagnostic filenames do +not select the format. Text is not a stable machine-parsing interface. + +```text +2026-09-11T10:44:14Z INFO [run 230] Starting validation with delta synchronization across 5 trust anchors. +2026-09-11T10:44:24Z INFO [run 230] Synchronizing repositories and validating objects; 10.0 seconds elapsed. Processed 32000 publication points so far. 3 repository sync attempts are still in progress. +2026-09-11T10:45:36Z INFO [run 230] Validation completed in 82.0 seconds. +``` + +Use `--log-format json` or `PANDA_RPKI_LOG_FORMAT=json` for machine consumers; +the JSON event names and fields are unchanged. Both formats redact sensitive data. +Text error reasons are limited to 240 Unicode characters with an explicit notice +when shortened. Unknown extension/debug events retain a diagnostic identifier and +their message, not an arbitrary field dump. Use JSON (and debug/trace as needed) +for full technical details, category examples and counter snapshots. +JSON is one complete object per line: `timestamp`, `level`, `event`, `fields`, `source`. +Timestamps are UTC RFC3339. Run context is in `fields.run_id` and `fields.run_seq` +(null for standalone validation). Logging failures do not change validation results. + +Every emitted text line ends with `[source/file.rs:line]`. JSON records include +`source: {"file": "source/file.rs", "line": 123}` at the top level. These are +compile-time source paths and one-based line numbers from the logger call site, +not the shared output sink; progress forwarding preserves the caller location. +Logs emitted inside artifact/stage helpers point to the emit call inside that +helper. Match locations against the source revision used to build the binary. + +```text +2026-09-11T10:45:36Z INFO [run 230] Validation completed in 82.0 seconds. [src/cli/validate.rs:123] +``` + +The source line in this example is illustrative; actual logs report the compiled location. + +## Event contract + +| Events | Level | Meaning / frequency | +| --- | --- | --- | +| validation_started/completed/summary | info | Run boundaries and exact final business counts | +| validation_failed, command_failed | error | Run/command failure, with error and elapsed time where available | +| validation_stage_started/completed/failed | info/error | State store, setup, TA discovery, online validation, output, CSV conversion | +| validation_progress | info | Default every 10s, independent of blocked network/store calls | +| trust_anchor_started/completed/failed | info/error | One pair per TAL, fatal failure includes TAL ID | +| validation_warnings | warn | Existing final total, not a declaration that the run failed | +| validation_warning_category | info | Category count and up to three redacted 1KiB examples | +| validation_warning_detail | debug | Full warning message and context | +| artifact_write_started/completed/failed | info/error | Type, path, wall milliseconds and bytes; CCR, CSV, summary, timings | +| rrdp_sync_attempt/success | debug | Transport attempt and successful object writes/duration | +| rrdp_sync_failed, http_fetch_failed | warn | Recoverable attempt failures; retryable does not imply will_retry | +| rrdp_sync_selected | debug | Actual snapshot/delta/unchanged selection, reason, serial/session | +| rrdp_download_completed | debug | Snapshot/delta decoded payload bytes, not wire traffic | +| rrdp_cross_origin_rejected, rrdp_delta_fallback | warn | Protocol security rejection / delta application fallback | +| repo_terminal_failure | warn | Failed publication point, not a fatal command error | +| http_fetch_slow, rsync_sync_slow, phase2_*_slow | debug | Detailed slow operations, aggregated into info progress | +| phase1_repo_* | debug | Repository task scheduling and worker decisions | +| phase2 batch/queue/object events, publication_point_* | trace | Detailed high-volume processing; not recommended for continuous all5 | +| daemon_started/state_recovered/run_started/child_started | info | Effective configuration, recovery and per-run log path | +| daemon_run_completed/sleeping/retention_applied/exited | info/warn | Outcome, next scheduled time and retention decisions | +| daemon_stop_requested | info | Graceful stop request observed outside the signal handler | +| daemon_child_terminate/child_kill | warn | One event per signal escalation, with reason | + +`src/logging/progress.rs` explicitly registers progress event levels, including +the dynamic finalize-worker label; a source-inventory regression detects new +unregistered literal call sites. Unknown library-supplied names default to debug. + +## Progress and volume + +`--log-progress-interval-secs` overrides `PANDA_RPKI_LOG_PROGRESS_INTERVAL_SECS`, +default `10`. `0` suppresses periodic events, not stage events. `off/error/warn` +do not create a progress thread. Queue snapshots are refreshed at most every 250ms. +`statistics_age_ms` is the age of the queue/PP snapshot, not the repository event +counters. `progress_stalled` means no observed completion for 30s, not proof of deadlock. +Progress does not predict the total tree size or invent completion percentages. + +- Repository counters count attempts, not publication points or unique repositories. +- `repository_objects_written` counts writes, including updates; it is not unique validated objects. +- `roa_tasks_completed` counts returned parallel ROA tasks, not all object types. +- `downloaded_snapshot_delta_bytes` excludes notifications and measures decoded payload. +- The serial path reports PP completed/failed counts; absent parallel queue fields are not zeroes. +- Slow-operation counts are cumulative per run. Detailed slow events stay debug. + +Warning categories are assigned at typed source sites. Opaque legacy string +errors remain `unclassified`; they are not classified by substring guesses. +The category total equals the existing warning total, which may contain both tree +and audit warnings. Examples are redacted before UTF-8-safe truncation. + +| Category | Assignment at currently instrumented sites | Meaning / example | +| --- | --- | --- | +| unclassified | No typed classification retained | Opaque repository-sync error; do not guess its cause | +| ber_compatible_cms_encoding | Existing BER-compatible CMS warning | Accepted non-DER encoding | +| transport | Typed fetch error in RRDP fallback warning | Download failure | +| rrdp_protocol | Typed RRDP error in fallback warning | Invalid RRDP response | +| security_policy | Cross-origin fallback, manifest location/hash/rollback rejection | Integrity or anti-rollback check | +| certificate | Child discovery or typed manifest EE/issuer certificate errors | Certificate/path/resource failure | +| manifest_crl | Typed manifest errors or missing required CRL | Stale manifest or missing CRL | +| object_validation | Parallel ROA/ASPA rejection | Invalid signed routing object | + +This is incremental typed coverage, not a promise that every historical warning +site is already classified. Uninstrumented sites intentionally remain unclassified. + +```bash +tail -f /path/to/data/runs/run_000001/validator.log +jq -c 'select(.event == "validation_progress")' /path/to/validator.log +jq -c 'select(.level == "warn" or .level == "error")' /path/to/validator.log +``` + +The `jq` examples require JSON logging to have been explicitly enabled. + +Retention removes completed run directories including their logs. Archive needed +runs before the configured retention window expires. Summary history is append-only. diff --git a/docs/usage.md b/docs/usage.md index b90096c..1b98f98 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -76,6 +76,15 @@ The append-only summary history is not trimmed by retention. ## Outputs and diagnostics +`validator.log` is the per-run diagnostic log, not an error-only file. +`process-output.txt` captures ordinary child output and is normally empty. +Old runs keep their historical filenames; collectors support `stderr.log` as a fallback. +Every ten seconds, info logs report the current stage and available progress counters. +Use `--log-progress-interval-secs 0` to disable periodic events; stage events remain. +`statistics_age_ms` identifies stale statistics; `progress_stalled` means no observed +completion for thirty seconds, not a proven deadlock. Totals and percentages are not guessed. +Events include `run_id` and, under daemon, `run_seq`. Controller logs remain separate. + `vrps.csv` contains `asn,prefix,max_length`. CCR contains canonical state; duplicate routing payloads can collapse into a single CCR entry. Count raw rows and normalized unique payloads separately when comparing outputs. @@ -83,7 +92,7 @@ ASPA canonicalization groups providers by customer ASN. Use `--log-level debug` for diagnosis, or `trace` for detailed object/worker events. Logs are written to stderr and can be redirected independently from -outputs. In daemon mode, inspect per-run `stderr.log` and the controller's +outputs. In daemon mode, inspect per-run `validator.log` and the controller's `daemon-status.json` and `run-summary.jsonl`. | Symptom | Action | diff --git a/src/cli/mod.rs b/src/cli/mod.rs index 7d0dd4c..7306720 100644 --- a/src/cli/mod.rs +++ b/src/cli/mod.rs @@ -34,7 +34,10 @@ pub fn usage() -> String { "panda-rpki {}\n\nUSAGE:\n panda-rpki validate --tal --ta [--tal --ta ...] --out [--rrdp-state-dir ] [--rrdp-sync-mode auto|snapshot|delta] [--ccr-out ] [--parallel-phase2-object-workers ] [--parallel-max-repo-sync-workers-global ] [--parallel-phase2-worker-queue-capacity ] [--parallel-repo-worker-queue-capacity ] [--max-ca-depth ] [--http-timeout-secs ] [--http-root-cert ]... [--log-level off|error|warn|info|debug|trace] [--log-format text|json]\n\nTAL/TA pairs are positional and may be repeated. RRDP state is an explicit protocol-resume directory, not a validation cache. The public v0.1.0 fast path accepts HTTPS RRDP notification/snapshot/delta URLs only.", crate::VERSION ); - format!("{validation}\n\n{}", crate::daemon::usage()) + format!( + "{validation}\n --log-progress-interval-secs : periodic info progress (default 10; 0 disables).\n\n{}", + crate::daemon::usage() + ) } pub fn run(args: I) -> Result<(), String> diff --git a/src/cli/validate.rs b/src/cli/validate.rs index 941e51d..11a9661 100644 --- a/src/cli/validate.rs +++ b/src/cli/validate.rs @@ -15,6 +15,7 @@ pub(crate) const VALIDATE_OPTIONS: &[&str] = &[ "tal", "log-level", "log-format", + "log-progress-interval-secs", "ta", "tal-id", "ta-constraints", @@ -46,6 +47,21 @@ pub(super) fn run(args: &[String]) -> Result<(), String> { "text", ); crate::logging::configure(&level, &format)?; + let progress_interval = resolve_setting( + options.values("log-progress-interval-secs"), + std::env::var("PANDA_RPKI_LOG_PROGRESS_INTERVAL_SECS").ok(), + "10", + ) + .parse::() + .map_err(|_| { + "invalid log-progress-interval-secs: expected a nonnegative integer".to_string() + })?; + let run_id = + std::env::var("PANDA_RPKI_RUN_ID").unwrap_or_else(|_| uuid::Uuid::new_v4().to_string()); + let run_seq = std::env::var("PANDA_RPKI_RUN_SEQ") + .ok() + .and_then(|value| value.parse().ok()); + crate::logging::set_context(&run_id, run_seq); let tal_paths = options.values("tal"); let ta_paths = options.values("ta"); if tal_paths.is_empty() || ta_paths.is_empty() || tal_paths.len() != ta_paths.len() { @@ -156,22 +172,30 @@ pub(super) fn run(args: &[String]) -> Result<(), String> { }; let started = std::time::Instant::now(); + let observer = crate::logging::observer::Observer::start(progress_interval)?; crate::logging::emit(crate::logging::Level::Info, "validation_started", || { serde_json::json!({ "tal_count": tal_paths.len(), "tal_ids": tal_ids, "phase2_workers": workers, "repo_workers": repo_workers, "sync_mode": format!("{sync_mode:?}").to_lowercase(), }) }); - crate::runtime::run_config(config).inspect_err(|_| { + let result = crate::runtime::run_config(config).and_then(|()| { + crate::logging::observer::phase("vrp_csv_conversion"); + let fast_vrps = out.join(".vrps-source.csv"); + crate::logging::artifact::write("vrps", &out.join("vrps.csv"), || { + convert_vrp_csv(&fast_vrps, &out.join("vrps.csv")) + }) + }); + observer.finish(result.is_ok()); + result.inspect_err(|error| { crate::logging::emit(crate::logging::Level::Error, "validation_failed", || { serde_json::json!({ "elapsed_ms": started.elapsed().as_millis(), + "error": error, }) }); })?; - let fast_vrps = out.join(".vrps-source.csv"); - convert_vrp_csv(&fast_vrps, &out.join("vrps.csv"))?; crate::logging::emit(crate::logging::Level::Info, "validation_completed", || { serde_json::json!({ "elapsed_ms": started.elapsed().as_millis(), "phase2_workers": workers, "repo_workers": repo_workers, diff --git a/src/daemon/mod.rs b/src/daemon/mod.rs index fe24b8c..d24e2ea 100644 --- a/src/daemon/mod.rs +++ b/src/daemon/mod.rs @@ -219,7 +219,19 @@ pub(crate) fn run(args: &[String]) -> Result<(), String> { return Err(crate::cli::usage()); } let mut config = parse(args)?; + let progress_interval = + std::env::var("PANDA_RPKI_LOG_PROGRESS_INTERVAL_SECS").unwrap_or_else(|_| "10".into()); + Options::parse(&config.child_args, VALIDATE_OPTIONS, VALIDATE_REPEATABLE)? + .values("log-progress-interval-secs") + .first() + .unwrap_or(&progress_interval) + .parse::() + .map_err(|_| "invalid log-progress-interval-secs".to_string())?; crate::logging::configure(&config.log_level, &config.log_format)?; + emit(Level::Info, "daemon_started", || { + json!({"interval_secs": config.interval, + "retain_runs": config.retain, "run_timeout_secs": config.timeout, "state_root": config.root}) + }); regular_directory(&config.root)?; config.root = fs::canonicalize(&config.root).map_err(|e| e.to_string())?; let lock = OpenOptions::new() @@ -269,8 +281,18 @@ fn controller(config: &Config) -> Result<(), String> { let mut completed = 0u64; let mut failures = 0u64; status(root, "starting", lifecycle.last_seq, completed)?; + emit( + Level::Info, + "daemon_state_recovered", + || json!({"last_run_seq": lifecycle.last_seq, "last_success": lifecycle.last_success}), + ); while !process::stopping() && config.max_runs.is_none_or(|n| completed < n) { let mode = sync_mode(&state, lifecycle.last_success)?; + let mode_reason = match mode { + "snapshot" => "empty_state", + "delta" => "previous_run_succeeded", + _ => "previous_run_incomplete_or_failed", + }; lifecycle.last_seq = lifecycle .last_seq .checked_add(1) @@ -278,18 +300,20 @@ fn controller(config: &Config) -> Result<(), String> { lifecycle.last_success = false; write_json_pretty(&lifecycle_path, &lifecycle)?; let seq = lifecycle.last_seq; + let run_id = uuid::Uuid::new_v4().to_string(); + crate::logging::set_context(&run_id, Some(seq)); let out = runs.join(format!("run_{seq:06}")); fs::create_dir(&out).map_err(|e| format!("create run: {e}"))?; let started_at = now(); let start = Instant::now(); let mut meta = json!({"controller": "panda-rpki-daemon-v1", "run_seq": seq, - "status": "running", "sync_mode": mode, "started_at": started_at}); + "status": "running", "sync_mode": mode, "started_at": started_at, "run_id": run_id, "sync_mode_reason": mode_reason}); write_json_pretty(&out.join("run-meta.json"), &meta)?; status(root, "running", seq, completed)?; emit( Level::Info, "daemon_run_started", - || json!({"run_seq": seq, "sync_mode": mode}), + || json!({"run_seq": seq, "sync_mode": mode, "reason": mode_reason, "validator_log": out.join("validator.log")}), ); // Child-per-run execution and stdout/stderr redirection; paths // are supplied as arguments, never rendered into shell command text. @@ -307,6 +331,9 @@ fn controller(config: &Config) -> Result<(), String> { Command::new(&executable) }; command.env("LC_ALL", "C"); + command + .env("PANDA_RPKI_RUN_ID", &run_id) + .env("PANDA_RPKI_RUN_SEQ", seq.to_string()); command .arg("validate") .args(&config.child_args) @@ -319,8 +346,8 @@ fn controller(config: &Config) -> Result<(), String> { .arg("--rrdp-sync-mode") .arg(mode) .stdin(Stdio::null()) - .stdout(File::create(out.join("stdout.log")).map_err(|e| e.to_string())?) - .stderr(File::create(out.join("stderr.log")).map_err(|e| e.to_string())?); + .stdout(File::create(out.join("process-output.txt")).map_err(|e| e.to_string())?) + .stderr(File::create(out.join("validator.log")).map_err(|e| e.to_string())?); let result = process::execute(&mut command, config.timeout, config.grace); let (exit_code, failure) = match result { Ok((exit, reason)) => ( @@ -351,18 +378,34 @@ fn controller(config: &Config) -> Result<(), String> { completed += 1; failures += u64::from(!success); let removed = apply_retention(&runs, config.retain)?; + if !removed.is_empty() { + emit( + Level::Info, + "daemon_retention_applied", + || json!({"removed_runs": removed, "retain_runs": config.retain}), + ); + } emit( if success { Level::Info } else { Level::Warn }, "daemon_run_completed", - || json!({"run_seq": seq, "success": success, "wall_ms": meta["wall_ms"], "removed_runs": removed}), + || json!({"run_seq": seq, "success": success, "wall_ms": meta["wall_ms"], "removed_runs": removed, "error": meta["error"], "exit_code": exit_code}), ); if process::stopping() || config.max_runs.is_some_and(|n| completed >= n) { break; } status(root, "sleeping", seq, completed)?; + emit(Level::Info, "daemon_sleeping", || { + json!({"interval_secs": config.interval, + "next_run_at_unix": time::OffsetDateTime::now_utc().unix_timestamp().checked_add(i64::try_from(config.interval).unwrap_or(i64::MAX))}) + }); process::sleep(config.interval); } status(root, "exited", lifecycle.last_seq, completed)?; + emit( + Level::Info, + "daemon_exited", + || json!({"runs_completed": completed, "failed_runs": failures, "stop_requested": process::stopping()}), + ); if failures > 0 { Err(format!("daemon completed with {failures} failed run(s)")) } else { diff --git a/src/daemon/process.rs b/src/daemon/process.rs index ed88244..ebc5904 100644 --- a/src/daemon/process.rs +++ b/src/daemon/process.rs @@ -68,9 +68,15 @@ pub(super) fn execute( .map_err(|e| format!("spawn validator: {e}"))?, ); let start = Instant::now(); + crate::logging::emit( + crate::logging::Level::Info, + "daemon_child_started", + || serde_json::json!({"pid": child.0.id()}), + ); let mut stop_at = None; let mut terminated_at = None; let mut reason = None; + let mut killed = false; loop { if let Some(status) = child .0 @@ -80,17 +86,35 @@ pub(super) fn execute( return Ok((status, reason)); } if stopping() { + if stop_at.is_none() { + crate::logging::emit( + crate::logging::Level::Info, + "daemon_stop_requested", + || serde_json::json!({"grace_secs": grace}), + ); + } stop_at.get_or_insert_with(Instant::now); } let timed_out = timeout > 0 && start.elapsed().as_secs() >= timeout; let grace_expired = stop_at.is_some_and(|at: Instant| at.elapsed().as_secs() >= grace); if terminated_at.is_none() && (timed_out || grace_expired) { reason = Some(if timed_out { "timeout" } else { "interrupted" }); + crate::logging::emit( + crate::logging::Level::Warn, + "daemon_child_terminate", + || serde_json::json!({"reason": reason, "signal": "SIGTERM", "elapsed_ms": start.elapsed().as_millis()}), + ); child.signal_group(libc::SIGTERM); terminated_at = Some(Instant::now()); } - if terminated_at.is_some_and(|at| at.elapsed() >= Duration::from_secs(2)) { + if !killed && terminated_at.is_some_and(|at| at.elapsed() >= Duration::from_secs(2)) { + crate::logging::emit( + crate::logging::Level::Warn, + "daemon_child_kill", + || serde_json::json!({"reason": reason, "signal": "SIGKILL"}), + ); child.signal_group(libc::SIGKILL); + killed = true; } std::thread::sleep(Duration::from_millis(50)); } diff --git a/src/logging.rs b/src/logging.rs index d35ace8..1ccd291 100644 --- a/src/logging.rs +++ b/src/logging.rs @@ -2,6 +2,17 @@ use std::io::Write; use std::sync::Mutex; use std::sync::atomic::{AtomicU8, Ordering}; +pub(crate) mod artifact; +pub(crate) mod observer; +mod text; + +static CONTEXT: Mutex> = Mutex::new(None); + +pub(crate) fn set_context(run_id: &str, run_seq: Option) { + *CONTEXT.lock().unwrap_or_else(|error| error.into_inner()) = Some(serde_json::json!({ + "run_id": run_id, "run_seq": run_seq, + })); +} #[derive(Clone, Copy, Debug, PartialEq, Eq)] #[repr(u8)] @@ -47,7 +58,7 @@ pub fn enabled(level: Level) -> bool { level as u8 <= FILTER.load(Ordering::Relaxed) } -fn redact_text(text: &str) -> String { +pub(crate) fn redact_text(text: &str) -> String { let mut output = String::with_capacity(text.len()); let lower = text.to_ascii_lowercase(); let mut cursor = 0; @@ -123,6 +134,7 @@ fn redact(value: &mut serde_json::Value) { } /// Payload creation is lazy, so disabled diagnostics do not allocate JSON. +#[track_caller] pub fn emit(level: Level, event: &str, payload: impl FnOnce() -> serde_json::Value) { let _ = emit_to( FILTER.load(Ordering::Relaxed), @@ -139,6 +151,7 @@ pub fn emit(level: Level, event: &str, payload: impl FnOnce() -> serde_json::Val ); } +#[track_caller] fn emit_to( filter: u8, json: bool, @@ -154,22 +167,36 @@ fn emit_to( .format(&time::format_description::well_known::Rfc3339) .unwrap_or_default(); let mut fields = payload(); + if let Some(context) = CONTEXT + .lock() + .unwrap_or_else(|error| error.into_inner()) + .as_ref() + && let Some(fields) = fields.as_object_mut() + && let Some(context) = context.as_object() + { + fields.extend(context.clone()); + } redact(&mut fields); + let source = std::panic::Location::caller(); let record = serde_json::json!({ "timestamp": timestamp, "level": format!("{level:?}").to_lowercase(), "event": event, "fields": fields, + "source": {"file": source.file(), "line": source.line()}, }); let line = if json { format!("{record}\n") } else { + if event == "validation_stage_completed" && record["fields"]["phase"] == "initializing" { + return Ok(()); + } + let message = text::render(×tamp, level, event, &record["fields"]); format!( - "{} {} {} {}\n", - timestamp, - record["level"].as_str().unwrap_or(""), - event, - record["fields"] + "{} [{}:{}]\n", + message.trim_end_matches('\n'), + source.file(), + source.line() ) }; sink(line.as_bytes()) @@ -179,6 +206,76 @@ fn emit_to( mod tests { use super::*; + #[track_caller] + fn forwarded_record(json: bool) -> String { + let mut output = String::new(); + emit_to( + 3, + json, + Level::Info, + "validation_completed", + || serde_json::json!({"elapsed_ms":1000}), + |line| { + output = String::from_utf8(line.to_vec()).unwrap(); + Ok(()) + }, + ) + .unwrap(); + output + } + + #[test] + fn source_location_survives_forwarding_and_matches_both_formats() { + let expected_line = line!() + 1; + let raw = forwarded_record(true); + let record: serde_json::Value = serde_json::from_str(&raw).unwrap(); + assert_eq!(record["source"]["file"], file!()); + assert_eq!(record["source"]["line"], expected_line); + let expected_line = line!() + 1; + let text = forwarded_record(false); + assert!(text.ends_with(&format!("[{}:{expected_line}]\n", file!()))); + } + + #[test] + fn text_redacts_before_rendering_and_json_keeps_initializing() { + emit_to( + 3, + false, + Level::Warn, + "http_fetch_failed", + || { + serde_json::json!({ + "error": "fetch https://alice:pw@example.test/a?private=yes\nfailed" + }) + }, + |line| { + let line = std::str::from_utf8(line).unwrap(); + assert!(line.contains("WARN HTTP download failed")); + assert!(!line.contains("alice")); + assert!(!line.contains("private=yes")); + assert_eq!(line.lines().count(), 1); + Ok(()) + }, + ) + .unwrap(); + for json in [false, true] { + let written = std::cell::Cell::new(false); + emit_to( + 3, + json, + Level::Info, + "validation_stage_completed", + || serde_json::json!({"phase": "initializing", "elapsed_ms": 0}), + |_| { + written.set(true); + Ok(()) + }, + ) + .unwrap(); + assert_eq!(written.get(), json); + } + } + #[test] fn mixed_case_urls_and_opaque_credentials_are_redacted() { assert_eq!( @@ -320,13 +417,6 @@ mod tests { } pub(crate) mod progress; -macro_rules! info { - ($($args:tt)*) => { - crate::logging::emit(crate::logging::Level::Info, "runtime", || serde_json::json!({"message": format!($($args)*)})) - }; -} -pub(crate) use info; - macro_rules! warning { ($($args:tt)*) => { crate::logging::emit(crate::logging::Level::Warn, "runtime_warning", || serde_json::json!({"message": format!($($args)*)})) diff --git a/src/logging/artifact.rs b/src/logging/artifact.rs new file mode 100644 index 0000000..525a44a --- /dev/null +++ b/src/logging/artifact.rs @@ -0,0 +1,34 @@ +use super::{Level, emit}; +use serde_json::json; +use std::path::Path; +use std::time::Instant; + +pub(crate) fn write( + kind: &str, + path: &Path, + operation: impl FnOnce() -> Result, +) -> Result { + let started = Instant::now(); + emit( + Level::Info, + "artifact_write_started", + || json!({"artifact_type": kind, "path": path}), + ); + match operation() { + Ok(value) => { + emit(Level::Info, "artifact_write_completed", || { + json!({"artifact_type": kind, + "path": path, "bytes": std::fs::metadata(path).ok().map(|meta| meta.len()), + "elapsed_ms": started.elapsed().as_millis()}) + }); + Ok(value) + } + Err(error) => { + emit(Level::Error, "artifact_write_failed", || { + json!({"artifact_type": kind, + "path": path, "error": error, "elapsed_ms": started.elapsed().as_millis()}) + }); + Err(error) + } + } +} diff --git a/src/logging/observer.rs b/src/logging/observer.rs new file mode 100644 index 0000000..f4f0422 --- /dev/null +++ b/src/logging/observer.rs @@ -0,0 +1,297 @@ +use super::{Level, emit}; +use serde_json::{Value, json}; +use std::collections::HashMap; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex, mpsc}; +use std::time::{Duration, Instant}; + +static ACTIVE: Mutex>>> = Mutex::new(None); +static ENABLED: AtomicBool = AtomicBool::new(false); + +struct Snapshot { + started: Instant, + phase_started: Instant, + updated: Instant, + advanced: Instant, + phase: &'static str, + completed: u64, + counters: Value, + repositories: HashMap, + repositories_completed: u64, + repositories_failed: u64, + objects_written: u64, + downloaded_bytes: u64, + roa_tasks_completed: u64, + slow_operations: u64, + slowest_ms: u64, + slowest_event: String, +} + +impl Snapshot { + fn record(&self, now: Instant) -> Value { + json!({ + "phase": self.phase, + "elapsed_ms": now.duration_since(self.started).as_millis(), + "phase_elapsed_ms": now.duration_since(self.phase_started).as_millis(), + "statistics_age_ms": now.duration_since(self.updated).as_millis(), + "progress_stalled": now.duration_since(self.advanced) >= Duration::from_secs(30), + "counters": self.counters, + "repository_attempts_inflight": self.repositories.len(), + "repository_attempts_completed": self.repositories_completed, + "repository_attempts_failed": self.repositories_failed, + "repository_objects_written": self.objects_written, + "downloaded_snapshot_delta_bytes": self.downloaded_bytes, + "roa_tasks_completed": self.roa_tasks_completed, + "slow_operations": self.slow_operations, + "slowest_ms": self.slowest_ms, + "slowest_event": self.slowest_event, + "longest_repository_wait_ms": self.repositories.values().map(|started| now.duration_since(*started).as_millis()).max(), + }) + } +} + +pub(crate) struct Observer { + stop: mpsc::Sender<()>, + worker: Option>, +} + +impl Observer { + pub(crate) fn finish(mut self, success: bool) { + let _ = self.stop.send(()); + if let Some(worker) = self.worker.take() { + let _ = worker.join(); + } + if let Some(shared) = ACTIVE + .lock() + .unwrap_or_else(|error| error.into_inner()) + .as_ref() + { + let record = shared + .lock() + .unwrap_or_else(|error| error.into_inner()) + .record(Instant::now()); + emit( + if success { Level::Info } else { Level::Error }, + if success { + "validation_stage_completed" + } else { + "validation_stage_failed" + }, + || record, + ); + } + } + + pub(crate) fn start(seconds: u64) -> Result { + let now = Instant::now(); + let snapshot = Arc::new(Mutex::new(Snapshot { + started: now, + phase_started: now, + updated: now, + advanced: now, + phase: "initializing", + completed: 0, + counters: json!({}), + repositories: HashMap::new(), + repositories_completed: 0, + repositories_failed: 0, + objects_written: 0, + downloaded_bytes: 0, + roa_tasks_completed: 0, + slow_operations: 0, + slowest_ms: 0, + slowest_event: String::new(), + })); + let (stop, receiver) = mpsc::channel(); + let worker = if seconds > 0 && super::enabled(Level::Info) { + let shared = Arc::clone(&snapshot); + Some( + std::thread::Builder::new() + .name("validation-progress".into()) + .spawn(move || { + while receiver.recv_timeout(Duration::from_secs(seconds)) + == Err(mpsc::RecvTimeoutError::Timeout) + { + let record = shared + .lock() + .unwrap_or_else(|error| error.into_inner()) + .record(Instant::now()); + emit(Level::Info, "validation_progress", || record); + } + }) + .map_err(|error| format!("start progress observer: {error}"))?, + ) + } else { + None + }; + if super::enabled(Level::Info) { + *ACTIVE.lock().unwrap_or_else(|error| error.into_inner()) = Some(snapshot); + ENABLED.store(true, Ordering::Relaxed); + } + Ok(Self { stop, worker }) + } +} + +impl Drop for Observer { + fn drop(&mut self) { + ENABLED.store(false, Ordering::Relaxed); + let _ = self.stop.send(()); + if let Some(worker) = self.worker.take() { + let _ = worker.join(); + } + *ACTIVE.lock().unwrap_or_else(|error| error.into_inner()) = None; + } +} + +pub(crate) fn phase(name: &'static str) { + if let Some(shared) = ACTIVE + .lock() + .unwrap_or_else(|error| error.into_inner()) + .as_ref() + { + let mut snapshot = shared.lock().unwrap_or_else(|error| error.into_inner()); + let now = Instant::now(); + emit(Level::Info, "validation_stage_completed", || { + snapshot.record(now) + }); + snapshot.phase = name; + snapshot.phase_started = now; + snapshot.updated = now; + snapshot.advanced = now; + snapshot.completed = 0; + snapshot.counters = json!({}); + emit(Level::Info, "validation_stage_started", || { + snapshot.record(now) + }); + } +} + +pub(crate) fn update(completed: u64, counters: impl FnOnce() -> Value) { + if !ENABLED.load(Ordering::Relaxed) { + return; + } + if let Some(shared) = ACTIVE + .lock() + .unwrap_or_else(|error| error.into_inner()) + .as_ref() + { + let mut snapshot = shared.lock().unwrap_or_else(|error| error.into_inner()); + let now = Instant::now(); + if now.duration_since(snapshot.updated) < Duration::from_millis(250) { + return; + } + if completed != snapshot.completed { + snapshot.advanced = now; + } + snapshot.completed = completed; + snapshot.updated = now; + snapshot.counters = counters(); + } +} + +pub(crate) fn observes(kind: &str) -> bool { + ENABLED.load(Ordering::Relaxed) + && matches!( + kind, + "rrdp_sync_attempt" + | "rrdp_sync_success" + | "rrdp_sync_failed" + | "phase2_object_results_drain" + | "rrdp_download_completed" + | "http_fetch_slow" + | "rsync_sync_slow" + | "phase2_control_loop_slow" + | "phase2_stage_fresh_slow" + | "phase2_ready_publication_point_control_slow" + ) +} + +pub(crate) fn event(kind: &str, fields: &Value) { + if let Some(shared) = ACTIVE + .lock() + .unwrap_or_else(|error| error.into_inner()) + .as_ref() + { + let mut snapshot = shared.lock().unwrap_or_else(|error| error.into_inner()); + let now = Instant::now(); + match kind { + "rrdp_download_completed" => { + snapshot.downloaded_bytes += fields["bytes"].as_u64().unwrap_or(0); + snapshot.advanced = now; + } + "rrdp_sync_attempt" => { + if let Some(uri) = fields["notify_uri"].as_str() { + snapshot.repositories.insert(uri.into(), now); + } + } + "rrdp_sync_success" | "rrdp_sync_failed" => { + if let Some(uri) = fields["notify_uri"].as_str() { + snapshot.repositories.remove(uri); + } + snapshot.repositories_completed += 1; + snapshot.repositories_failed += u64::from(kind == "rrdp_sync_failed"); + snapshot.objects_written += fields["objects_written"].as_u64().unwrap_or(0); + snapshot.advanced = now; + } + "phase2_object_results_drain" => { + let completed = fields["results_drained"].as_u64().unwrap_or(0); + snapshot.roa_tasks_completed += completed; + if completed > 0 { + snapshot.advanced = now; + } + } + _ => { + snapshot.slow_operations += 1; + let elapsed = [ + "duration_ms", + "total_duration_ms", + "stage_fresh_ms", + "elapsed_ms", + "control_ms", + ] + .iter() + .filter_map(|key| fields[key].as_u64()) + .max() + .unwrap_or(0); + if elapsed > snapshot.slowest_ms { + snapshot.slowest_ms = elapsed; + snapshot.slowest_event = kind.into(); + } + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn stalled_and_statistics_age_are_explicit() { + let now = Instant::now(); + let snapshot = Snapshot { + started: now, + phase_started: now, + updated: now, + advanced: now, + phase: "online_validation", + completed: 0, + counters: json!({"publication_points_completed": 0}), + repositories: HashMap::new(), + repositories_completed: 0, + repositories_failed: 0, + objects_written: 0, + downloaded_bytes: 0, + roa_tasks_completed: 0, + slow_operations: 0, + slowest_ms: 0, + slowest_event: String::new(), + }; + assert_eq!( + snapshot.record(now + Duration::from_secs(29))["progress_stalled"], + false + ); + let record = snapshot.record(now + Duration::from_secs(30)); + assert_eq!(record["progress_stalled"], true); + assert_eq!(record["statistics_age_ms"], 30000); + } +} diff --git a/src/logging/progress.rs b/src/logging/progress.rs index e65125c..5ec496d 100644 --- a/src/logging/progress.rs +++ b/src/logging/progress.rs @@ -1,41 +1,102 @@ // Event sites construct detailed payloads lazily. macro_rules! emit { ($kind:expr, $payload:expr $(,)?) => { - crate::logging::emit(crate::logging::progress::event_level($kind), $kind, || { - $payload - }) + crate::logging::progress::emit_event($kind, || $payload) }; } pub(crate) use emit; -pub(crate) fn event_level(kind: &str) -> crate::logging::Level { - if matches!(kind, "http_fetch_failed" | "rrdp_sync_failed") { - crate::logging::Level::Warn - } else if kind == "validation_failed" || kind.ends_with("_failed") || kind.starts_with("error_") - { - crate::logging::Level::Error - } else if kind.ends_with("_warning") - || kind.ends_with("_warnings") - || kind.starts_with("warning_") - { - crate::logging::Level::Warn - } else if kind.starts_with("publication_point") - || kind.starts_with("phase2_") - || kind.starts_with("object_") - { - crate::logging::Level::Trace - } else if kind.starts_with("validation_") - || kind.starts_with("run_") - || kind.starts_with("output_") - || kind.starts_with("ccr_") - || kind.ends_with("_summary") - { - crate::logging::Level::Info +#[track_caller] +pub(crate) fn emit_event(kind: &str, payload: impl FnOnce() -> serde_json::Value) { + let level = event_level(kind); + if super::observer::observes(kind) { + let fields = payload(); + super::observer::event(kind, &fields); + super::emit(level, kind, || fields); } else { - crate::logging::Level::Debug + super::emit(level, kind, payload); } } +pub(crate) fn event_level(kind: &str) -> crate::logging::Level { + registered_event_level(kind).unwrap_or(crate::logging::Level::Debug) +} + +fn registered_event_level(kind: &str) -> Option { + Some( + if matches!( + kind, + "http_fetch_failed" + | "rrdp_sync_failed" + | "rrdp_cross_origin_rejected" + | "repo_terminal_failure" + | "rrdp_failed_fallback_rsync" + ) { + crate::logging::Level::Warn + } else if matches!(kind, "validation_failed" | "object_failed") { + crate::logging::Level::Error + } else if matches!(kind, "validation_warning" | "validation_warnings") { + crate::logging::Level::Warn + } else if matches!( + kind, + "publication_point_start" + | "publication_point_repo_sync_done" + | "publication_point_finish" + | "publication_point_object" + | "phase2_repo_events_drain" + | "phase2_ready_queue_stage_fresh_breakdown" + | "phase2_stage_pool_stats" + | "phase2_publication_point_reduced" + | "phase2_build_tree_output" + | "phase2_roa_dispatch_batch" + | "phase2_object_results_drain" + | "phase2_finalize_task_submit" + | "phase2_finalize_results_drain" + | "phase2_ready_queue_batch" + | "phase2_ready_queue_control_breakdown" + | "phase2_ready_publication_point_control_snapshot_breakdown" + | "phase2_finalize_worker_breakdown" + ) { + crate::logging::Level::Trace + } else if matches!( + kind, + "validation_started" | "validation_completed" | "validation_summary" + ) { + crate::logging::Level::Info + } else if matches!( + kind, + "repository_sync_selected" + | "repo_sync_rrdp_ok" + | "rrdp_fallback_rsync" + | "repo_sync_rsync_direct" + | "rrdp_sync_attempt" + | "rrdp_sync_success" + | "rsync_sync_start" + | "rsync_sync_fetch_done" + | "rsync_sync_done" + | "rsync_sync_slow" + | "http_fetch_slow" + | "rrdp_redirect_checked" + | "phase2_control_loop_slow" + | "phase2_stage_fresh_slow" + | "phase2_ready_publication_point_control_slow" + | "phase1_repo_task_enqueued" + | "phase1_repo_task_waiting" + | "phase1_repo_task_reused" + | "phase1_repo_task_dispatched" + | "phase1_repo_task_result" + | "phase1_repo_worker_execute_start" + | "phase1_repo_worker_execute_finish" + | "rrdp_sync_selected" + | "rrdp_download_completed" + ) { + crate::logging::Level::Debug + } else { + return None; + }, + ) +} + pub fn slow_threshold_secs() -> f64 { 30.0 } @@ -57,6 +118,37 @@ mod tests { use super::event_level; use crate::logging::Level; + #[test] + fn all_literal_progress_sites_are_registered() { + fn visit(path: &std::path::Path) { + for entry in std::fs::read_dir(path).unwrap() { + let path = entry.unwrap().path(); + if path.is_dir() { + visit(&path); + } else if path.extension().is_some_and(|extension| extension == "rs") { + let source = std::fs::read_to_string(&path).unwrap(); + for site in source.split(concat!("progress::", "emit!(")).skip(1) { + if let Some(literal) = site.trim_start().strip_prefix('"') { + let event = literal.split('"').next().unwrap(); + assert!( + super::registered_event_level(event).is_some(), + "unregistered event {event} in {}", + path.display() + ); + } + } + } + } + } + visit(&std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src")); + assert_eq!( + super::registered_event_level("phase2_finalize_worker_breakdown"), + Some(Level::Trace) + ); + assert_eq!(event_level("rrdp_cross_origin_rejected"), Level::Warn); + assert_eq!(event_level("phase2_control_loop_slow"), Level::Debug); + } + #[test] fn event_levels_keep_default_output_quiet() { assert_eq!(event_level("validation_started"), Level::Info); diff --git a/src/logging/text.rs b/src/logging/text.rs new file mode 100644 index 0000000..9c9bc2a --- /dev/null +++ b/src/logging/text.rs @@ -0,0 +1,506 @@ +use super::Level; +use serde_json::Value; + +fn string<'a>(fields: &'a Value, key: &str, fallback: &'a str) -> &'a str { + fields[key] + .as_str() + .filter(|value| !value.is_empty()) + .unwrap_or(fallback) +} + +fn number(fields: &Value, key: &str) -> String { + fields[key] + .as_u64() + .map(|value| value.to_string()) + .unwrap_or_else(|| "an unknown number of".into()) +} + +fn duration(fields: &Value, key: &str) -> String { + fields[key] + .as_u64() + .map(|value| format!("{:.1} seconds", value as f64 / 1000.0)) + .unwrap_or_else(|| "an unknown duration".into()) +} + +fn reason(fields: &Value) -> String { + let raw = string( + fields, + "error", + string( + fields, + "message", + string(fields, "rrdp_error", "No further details were reported."), + ), + ); + let shortened: String = raw.chars().take(240).collect(); + if raw.chars().count() > 240 { + format!("{shortened}… (use JSON logging for the full diagnostic)") + } else { + shortened + } +} + +fn repository(fields: &Value) -> &str { + string( + fields, + "notify_uri", + string( + fields, + "manifest_rsync_uri", + string(fields, "uri", "the repository"), + ), + ) +} + +fn phase(fields: &Value) -> (&'static str, &'static str) { + match fields["phase"].as_str() { + Some("state_store_open") => ( + "Opening the saved repository state", + "Opened the saved repository state", + ), + Some("runtime_setup") => ("Preparing the validator", "Prepared the validator"), + Some("trust_anchor_discovery") => ("Loading trust anchors", "Loaded trust anchors"), + Some("online_validation") => ( + "Synchronizing repositories and validating objects", + "Finished synchronizing repositories and validating objects", + ), + Some("result_output") => ("Writing validation results", "Wrote validation results"), + Some("vrp_csv_conversion") => ("Exporting VRPs to CSV", "Exported VRPs to CSV"), + _ => ("Initializing validation", "Finished initialization"), + } +} + +fn progress(fields: &Value) -> String { + let mut sentence = format!( + "{}; {} elapsed.", + phase(fields).0, + duration(fields, "elapsed_ms") + ); + if fields["phase"] == "online_validation" { + if let Some(completed) = fields["counters"]["publication_points_completed"].as_u64() { + let stale = fields["statistics_age_ms"] + .as_u64() + .is_some_and(|age| age > 2000); + sentence.push_str(&if stale { + format!(" The last update reported {completed} publication points processed.") + } else { + format!(" Processed {completed} publication points so far.") + }); + } + if let Some(inflight) = fields["repository_attempts_inflight"] + .as_u64() + .filter(|count| *count > 0) + { + sentence.push_str(&if inflight == 1 { + " 1 repository sync attempt is still in progress.".into() + } else { + format!(" {inflight} repository sync attempts are still in progress.") + }); + } + } + if fields["progress_stalled"] == true { + sentence.push_str(" No completed work has been observed for at least 30 seconds; this does not necessarily mean the validator is stuck."); + } + sentence +} + +fn category(fields: &Value) -> &'static str { + match fields["category"].as_str() { + Some("transport") => "network access", + Some("rrdp_protocol") => "RRDP protocol checks", + Some("security_policy") => "security policy checks", + Some("certificate") => "certificate validation", + Some("manifest_crl") => "manifest and CRL checks", + Some("object_validation") => "signed object validation", + Some("ber_compatible_cms_encoding") => "accepted BER-compatible CMS encoding", + _ => "issues without a recorded category", + } +} + +fn message(event: &str, fields: &Value) -> String { + match event { + "validation_started" => format!( + "Starting validation with {} synchronization across {} trust anchors.", + string(fields, "sync_mode", "the configured"), + number(fields, "tal_count") + ), + "validation_completed" => format!( + "Validation completed in {}.", + duration(fields, "elapsed_ms") + ), + "validation_failed" => format!( + "Validation failed after {}: {}", + duration(fields, "elapsed_ms"), + reason(fields) + ), + "command_failed" => format!("Command failed: {}", reason(fields)), + "validation_summary" => format!( + "Validation produced {} VRPs and {} ASPA records from {} publication points.", + number(fields, "vrps"), + number(fields, "aspas"), + number(fields, "publication_points") + ), + "validation_progress" => progress(fields), + "validation_stage_started" => format!("{}.", phase(fields).0), + "validation_stage_completed" => format!( + "{} in {}.", + phase(fields).1, + duration(fields, "phase_elapsed_ms") + ), + "validation_stage_failed" => format!( + "Validation stopped while {} after {}.", + phase(fields).0.to_ascii_lowercase(), + duration(fields, "phase_elapsed_ms") + ), + "trust_anchor_started" => format!( + "Loading trust anchor {}.", + string(fields, "tal_id", "with an unknown ID") + ), + "trust_anchor_completed" => format!( + "Trust anchor {} is ready.", + string(fields, "tal_id", "with an unknown ID") + ), + "trust_anchor_failed" => format!( + "Could not load trust anchor {}: {}", + string(fields, "tal_id", "with an unknown ID"), + reason(fields) + ), + "artifact_write_started" => { + format!("Writing {}.", string(fields, "path", "the output file")) + } + "artifact_write_completed" => format!( + "Saved {} in {}.", + string(fields, "path", "the output file"), + duration(fields, "elapsed_ms") + ), + "artifact_write_failed" => format!( + "Could not write {}: {}", + string(fields, "path", "the output file"), + reason(fields) + ), + "validation_warnings" => format!( + "Validation reported {} warnings. Review the affected repositories and objects before interpreting the results.", + number(fields, "count") + ), + "validation_warning_category" => format!( + "Recorded {} warnings about {}.", + number(fields, "count"), + category(fields) + ), + "validation_warning_detail" | "validation_warning" | "runtime_warning" => reason(fields), + "object_failed" => format!("Object validation failed: {}", reason(fields)), + "rrdp_sync_attempt" => format!("Fetching repository updates from {}.", repository(fields)), + "rrdp_sync_success" => format!( + "Updated {} in {}.", + repository(fields), + duration(fields, "duration_ms") + ), + "rrdp_sync_failed" => format!( + "Could not update {} using RRDP: {}", + repository(fields), + reason(fields) + ), + "repo_terminal_failure" => format!( + "Could not process {}: {}", + repository(fields), + reason(fields) + ), + "http_fetch_failed" => format!( + "HTTP download failed for {}: {}", + repository(fields), + reason(fields) + ), + "rrdp_delta_fallback" => format!( + "Could not apply incremental updates from {}; trying a snapshot instead. {}", + repository(fields), + reason(fields) + ), + "rrdp_cross_origin_rejected" => format!( + "Rejected the RRDP {} because its origin ({}) differs from the notification origin ({}).", + string(fields, "resource", "resource"), + string(fields, "resource_origin", "unknown"), + string(fields, "notification_origin", "unknown") + ), + "rrdp_failed_fallback_rsync" => format!( + "RRDP failed for {}; trying rsync instead. {}", + repository(fields), + reason(fields) + ), + "rrdp_sync_selected" => format!( + "Using {} synchronization for {}.", + string(fields, "mode", "the selected"), + repository(fields) + ), + "rrdp_download_completed" => format!( + "Finished downloading repository data from {}.", + repository(fields) + ), + "daemon_started" => format!( + "Daemon started. Runs will be stored in {}; waiting {} seconds between runs.", + string(fields, "state_root", "the configured directory"), + number(fields, "interval_secs") + ), + "daemon_state_recovered" => format!( + "Resuming saved daemon state after run {}.", + number(fields, "last_run_seq") + ), + "daemon_run_started" => format!( + "Starting the next validation run. Follow its progress in {}.", + string(fields, "validator_log", "validator.log") + ), + "daemon_child_started" => "The validator process has started.".into(), + "daemon_run_completed" if fields["success"] == true => format!( + "Run completed successfully in {}.", + duration(fields, "wall_ms") + ), + "daemon_run_completed" => format!( + "Run failed after {}: {}", + duration(fields, "wall_ms"), + reason(fields) + ), + "daemon_sleeping" => format!( + "The next run will start in {} seconds.", + number(fields, "interval_secs") + ), + "daemon_retention_applied" => format!( + "Removed {} old run directories; keeping the latest {} runs.", + fields["removed_runs"] + .as_array() + .map(|runs| runs.len().to_string()) + .unwrap_or_else(|| "an unknown number of".into()), + number(fields, "retain_runs") + ), + "daemon_exited" => format!( + "Daemon stopped after {} runs, including {} failed runs.", + number(fields, "runs_completed"), + number(fields, "failed_runs") + ), + "daemon_stop_requested" => "Shutdown requested. Waiting for the validator to stop.".into(), + "daemon_child_terminate" => "Asking the validator process to terminate.".into(), + "daemon_child_kill" => { + "The validator did not stop within the grace period; forcing it to exit.".into() + } + _ => { + let description = if ["error", "message", "rrdp_error"] + .iter() + .any(|key| fields[*key].as_str().is_some()) + { + reason(fields) + } else { + "An internal diagnostic was recorded; use JSON logging for details.".into() + }; + format!("{description} [diagnostic: {event}]") + } + } +} + +pub(super) fn render(timestamp: &str, level: Level, event: &str, fields: &Value) -> String { + let context = fields["run_seq"] + .as_u64() + .map(|sequence| format!(" [run {sequence}]")) + .unwrap_or_default(); + let sentence = message(event, fields); + let escaped: String = sentence + .chars() + .flat_map(|character| { + if character.is_control() { + character.escape_default().collect::>() + } else { + vec![character] + } + }) + .collect(); + format!( + "{timestamp} {}{context} {escaped}\n", + format!("{level:?}").to_uppercase() + ) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn starting_a_run_is_a_sentence_not_a_field_dump() { + let line = render( + "now", + Level::Info, + "validation_started", + &json!({ + "run_seq":230,"sync_mode":"delta","tal_count":5,"run_id":"hidden-uuid", + "phase2_workers":8,"unexpected":{"secret_internal_detail":55} + }), + ); + assert_eq!( + line, + "now INFO [run 230] Starting validation with delta synchronization across 5 trust anchors.\n" + ); + } + + #[test] + fn progress_selects_only_useful_facts() { + let fields = json!({"phase":"online_validation","elapsed_ms":10005, + "counters":{"publication_points_completed":32000,"ready_queue":999}, + "repository_attempts_inflight":3,"repository_objects_written":456, + "statistics_age_ms":200,"slowest_event":"hidden","run_id":"hidden"}); + assert_eq!( + message("validation_progress", &fields), + "Synchronizing repositories and validating objects; 10.0 seconds elapsed. Processed 32000 publication points so far. 3 repository sync attempts are still in progress." + ); + } + + #[test] + fn stale_and_stalled_progress_does_not_claim_deadlock_or_completion() { + let fields = json!({"phase":"online_validation","elapsed_ms":40000, + "counters":{"publication_points_completed":0},"statistics_age_ms":5000,"progress_stalled":true}); + let sentence = progress(&fields); + assert!(sentence.contains("last update reported 0 publication points")); + assert!(sentence.contains("does not necessarily mean")); + assert!(!sentence.contains("sync attempts")); + assert!( + !progress(&json!({"phase":"state_store_open","elapsed_ms":10000})) + .contains("publication points") + ); + } + + #[test] + fn errors_are_bounded_and_cannot_inject_lines() { + let line = render( + "now", + Level::Warn, + "repo_terminal_failure", + &json!({"notify_uri":"https://example.test/notification.xml","error":format!("bad\n\r\u{1b}{}", "证书".repeat(300))}), + ); + assert_eq!(line.lines().count(), 1); + assert!(line.contains("bad\\n\\r\\u{1b}")); + assert!(line.contains("use JSON logging")); + assert!(!line.contains("cached")); + } + + #[test] + fn stages_outputs_and_results_have_independent_wording() { + assert_eq!( + message( + "validation_stage_completed", + &json!({"phase":"state_store_open","phase_elapsed_ms":339,"repository_objects_written":0}) + ), + "Opened the saved repository state in 0.3 seconds." + ); + assert_eq!( + message( + "artifact_write_completed", + &json!({"path":"/data/vrps.csv","elapsed_ms":133,"bytes":999}) + ), + "Saved /data/vrps.csv in 0.1 seconds." + ); + assert_eq!( + message( + "validation_summary", + &json!({"vrps":1012510,"aspas":3007,"publication_points":54463}) + ), + "Validation produced 1012510 VRPs and 3007 ASPA records from 54463 publication points." + ); + assert_eq!( + message( + "daemon_sleeping", + &json!({"interval_secs":60,"next_run_at_unix":123456}) + ), + "The next run will start in 60 seconds." + ); + } + + #[test] + fn unknown_diagnostics_keep_the_message_not_arbitrary_fields() { + assert_eq!( + message( + "extension_notice", + &json!({"message":"A custom check failed.","arbitrary":99}) + ), + "A custom check failed. [diagnostic: extension_notice]" + ); + assert!(message("validation_completed", &json!({})).contains("unknown duration")); + } + + #[test] + fn common_events_never_fall_back_to_technical_identifiers() { + let fields = json!({"error":"Connection refused.","notify_uri":"https://example.test/n.xml", + "duration_ms":1000,"elapsed_ms":1000,"phase_elapsed_ms":1000,"wall_ms":1000, + "tal_id":"ripe","path":"/data/result.ccr","count":2,"last_run_seq":229, + "validator_log":"/data/validator.log","removed_runs":["run_1"],"retain_runs":10, + "runs_completed":3,"failed_runs":0,"success":true,"state_root":"/data","interval_secs":60}); + for event in [ + "validation_failed", + "command_failed", + "validation_stage_started", + "validation_stage_failed", + "trust_anchor_started", + "trust_anchor_completed", + "trust_anchor_failed", + "artifact_write_started", + "artifact_write_failed", + "validation_warnings", + "validation_warning_detail", + "runtime_warning", + "rrdp_sync_attempt", + "rrdp_sync_success", + "rrdp_sync_failed", + "repo_terminal_failure", + "http_fetch_failed", + "rrdp_delta_fallback", + "rrdp_cross_origin_rejected", + "rrdp_sync_selected", + "rrdp_download_completed", + "daemon_started", + "daemon_state_recovered", + "daemon_run_started", + "daemon_child_started", + "daemon_run_completed", + "daemon_retention_applied", + "daemon_exited", + "daemon_stop_requested", + "daemon_child_terminate", + "daemon_child_kill", + ] { + let sentence = message(event, &fields); + assert!(!sentence.contains("diagnostic:"), "{event}: {sentence}"); + assert!(!sentence.contains("run_id")); + } + assert!( + message( + "daemon_run_completed", + &json!({"success":false,"wall_ms":1000,"error":"Stopped."}) + ) + .starts_with("Run failed") + ); + for name in [ + "transport", + "rrdp_protocol", + "security_policy", + "certificate", + "manifest_crl", + "object_validation", + "ber_compatible_cms_encoding", + "unclassified", + ] { + assert!( + message( + "validation_warning_category", + &json!({"category":name,"count":2}) + ) + .starts_with("Recorded 2 warnings about ") + ); + } + for name in [ + "state_store_open", + "runtime_setup", + "trust_anchor_discovery", + "online_validation", + "result_output", + "vrp_csv_conversion", + "initializing", + ] { + assert!(!message("validation_stage_started", &json!({"phase":name})).contains('_')); + } + } +} diff --git a/src/output/report.rs b/src/output/report.rs index 942511d..8295ece 100644 --- a/src/output/report.rs +++ b/src/output/report.rs @@ -11,6 +11,12 @@ pub enum WarningCategory { #[default] Unclassified, BerCompatibleCmsEncoding, + Transport, + RrdpProtocol, + SecurityPolicy, + Certificate, + ManifestCrl, + ObjectValidation, } impl WarningCategory { @@ -18,6 +24,12 @@ impl WarningCategory { match self { Self::Unclassified => "unclassified", Self::BerCompatibleCmsEncoding => "ber_compatible_cms_encoding", + Self::Transport => "transport", + Self::RrdpProtocol => "rrdp_protocol", + Self::SecurityPolicy => "security_policy", + Self::Certificate => "certificate", + Self::ManifestCrl => "manifest_crl", + Self::ObjectValidation => "object_validation", } } } diff --git a/src/repository/sync/repo.rs b/src/repository/sync/repo.rs index 71460af..9c9b044 100644 --- a/src/repository/sync/repo.rs +++ b/src/repository/sync/repo.rs @@ -123,6 +123,20 @@ pub fn sync_publication_point( } let warnings = vec![ Warning::new(format!("RRDP failed; falling back to rsync: {err}")) + .with_category(match &err { + RrdpSyncError::Fetch(_) => { + crate::output::report::WarningCategory::Transport + } + RrdpSyncError::Rrdp(error) if error.is_cross_origin_violation() => { + crate::output::report::WarningCategory::SecurityPolicy + } + RrdpSyncError::Rrdp(_) => { + crate::output::report::WarningCategory::RrdpProtocol + } + RrdpSyncError::Storage(_) => { + crate::output::report::WarningCategory::Unclassified + } + }) .with_rfc_refs(&rfc_refs) .with_context(notification_uri), ]; @@ -281,6 +295,7 @@ fn try_rrdp_sync_with_retry( download_log: Option<&DownloadLogHandle>, ) -> Result { let attempt = 1usize; + let started = std::time::Instant::now(); crate::logging::progress::emit!( "rrdp_sync_attempt", serde_json::json!({ @@ -307,6 +322,7 @@ fn try_rrdp_sync_with_retry( "notify_uri": notification_uri, "attempt": attempt, "objects_written": written, + "duration_ms": started.elapsed().as_millis(), }), ); Ok(written) @@ -322,6 +338,8 @@ fn try_rrdp_sync_with_retry( "notify_uri": notification_uri, "attempt": attempt, "retryable": retryable, + "will_retry": false, + "duration_ms": started.elapsed().as_millis(), "error": err.to_string(), }), ); diff --git a/src/repository/sync/rrdp/notification_sync.rs b/src/repository/sync/rrdp/notification_sync.rs index 743a0d3..8dd8fc7 100644 --- a/src/repository/sync/rrdp/notification_sync.rs +++ b/src/repository/sync/rrdp/notification_sync.rs @@ -105,9 +105,20 @@ fn sync_from_notification_inner( let same_session_state = state .as_ref() .filter(|s| s.session_id == notif.session_id.to_string()); + let log_selection = |mode: &str, reason: &str| { + crate::logging::emit(crate::logging::Level::Debug, "rrdp_sync_selected", || serde_json::json!({ + "notify_uri": notification_uri, "mode": mode, "reason": reason, + "session_id": notif.session_id.to_string(), "serial": notif.serial, + "previous_serial": state.as_ref().map(|previous| previous.serial), + "previous_session_id": state.as_ref().map(|previous| &previous.session_id), + })); + }; + let mut snapshot_reason = if state.is_none() { "no_local_state" } + else if same_session_state.is_none() { "session_changed" } else { "delta_chain_unavailable" }; if let Some(s) = same_session_state { if s.serial == notif.serial { + log_selection("unchanged", "same_session_and_serial"); let _hydrate_step = timing .as_ref() .map(|t| t.span_rrdp_repo_step(notification_uri, "hydrate_current_index")); @@ -147,6 +158,7 @@ fn sync_from_notification_inner( // `parse_notification` guarantees contiguity and max==notif.serial when deltas exist. if max_serial == notif.serial && want_first >= min_serial { + log_selection("delta", "contiguous_delta_chain"); // Fetch all required delta files first so a network failure doesn't leave us with // partially applied deltas and no snapshot fallback. let _fetch_d_step = timing @@ -179,6 +191,9 @@ fn sync_from_notification_inner( ¬ification_origin, ) { Ok(bytes) => { + crate::logging::progress::emit!("rrdp_download_completed", serde_json::json!({ + "resource": "delta", "bytes": bytes.len(), "notify_uri": notification_uri, + })); if let Some(t) = timing.as_ref() { t.record_count("rrdp_delta_fetch_ok_total", 1); t.record_count("rrdp_delta_bytes_total", bytes.len() as u64); @@ -190,6 +205,7 @@ fn sync_from_notification_inner( fetched.push((serial, dref.hash_sha256, bytes)) } Err(RrdpFetchError::Fetch(e)) => { + snapshot_reason = "delta_fetch_failed"; if let Some(t) = timing.as_ref() { t.record_count("rrdp_delta_fetch_fail_total", 1); } @@ -225,7 +241,9 @@ fn sync_from_notification_inner( *serial, ) { Ok(n) => applied_total += n, - Err(_) => { + Err(error) => { + snapshot_reason = "delta_apply_failed"; + crate::logging::emit(crate::logging::Level::Warn, "rrdp_delta_fallback", || serde_json::json!({"notify_uri": notification_uri, "error": error.to_string(), "serial": serial})); ok = false; break; } @@ -286,6 +304,7 @@ fn sync_from_notification_inner( } // Snapshot fallback (RFC 8182 §3.4.3). + log_selection("snapshot", snapshot_reason); let _fetch_step = timing .as_ref() .map(|t| t.span_rrdp_repo_step(notification_uri, "fetch_snapshot")); @@ -301,6 +320,9 @@ fn sync_from_notification_inner( ¬ification_origin, ) { Ok(v) => { + crate::logging::progress::emit!("rrdp_download_completed", serde_json::json!({ + "resource": "snapshot", "bytes": v.1, "notify_uri": notification_uri, + })); if let Some(t) = timing.as_ref() { t.record_count("rrdp_snapshot_fetch_ok_total", 1); t.record_count("rrdp_snapshot_bytes_total", v.1); diff --git a/src/repository/sync/rrdp/snapshot_sync.rs b/src/repository/sync/rrdp/snapshot_sync.rs index 2c86de7..b2500d5 100644 --- a/src/repository/sync/rrdp/snapshot_sync.rs +++ b/src/repository/sync/rrdp/snapshot_sync.rs @@ -80,6 +80,10 @@ fn sync_from_notification_snapshot_inner( }; drop(_parse_step); drop(_parse_total); + crate::logging::emit(crate::logging::Level::Debug, "rrdp_sync_selected", || serde_json::json!({ + "notify_uri": notification_uri, "mode": "snapshot", "reason": "snapshot_requested", + "session_id": notif.session_id.to_string(), "serial": notif.serial, + })); let _fetch_step = timing .as_ref() @@ -96,6 +100,9 @@ fn sync_from_notification_snapshot_inner( ¬ification_origin, ) { Ok(v) => { + crate::logging::progress::emit!("rrdp_download_completed", serde_json::json!({ + "resource": "snapshot", "bytes": v.1, "notify_uri": notification_uri, + })); if let Some(t) = timing.as_ref() { t.record_count("rrdp_snapshot_fetch_ok_total", 1); t.record_count("rrdp_snapshot_bytes_total", v.1); diff --git a/src/runtime/output.rs b/src/runtime/output.rs index 556691e..c2260d7 100644 --- a/src/runtime/output.rs +++ b/src/runtime/output.rs @@ -320,14 +320,13 @@ pub(super) fn run_compare_view_task( if let (Some(vrps_path), Some(vaps_path)) = (vrps_csv_out_path, vaps_csv_out_path) { let started = std::time::Instant::now(); build_ms = Some(0); - write_direct_vrp_csv(vrps_path, shared.vrps.as_ref(), trust_anchor)?; - write_direct_vap_csv(vaps_path, shared.aspas.as_ref(), trust_anchor)?; + crate::logging::artifact::write("vrps_source", vrps_path, || { + write_direct_vrp_csv(vrps_path, shared.vrps.as_ref(), trust_anchor) + })?; + crate::logging::artifact::write("vaps", vaps_path, || { + write_direct_vap_csv(vaps_path, shared.aspas.as_ref(), trust_anchor) + })?; write_ms = Some(started.elapsed().as_millis() as u64); - crate::logging::info!( - "wrote compare views: vrps={} vaps={}", - vrps_path.display(), - vaps_path.display() - ); } Ok(CompareViewTaskOutput { build_ms, write_ms }) } @@ -403,17 +402,18 @@ pub(super) fn write_stage_timing( && let Some(parent) = path.parent() { let stage_timing_path = parent.join("stage-timing.json"); - std::fs::write( - &stage_timing_path, - serde_json::to_vec_pretty(stage_timing).map_err(|e| e.to_string())?, - ) - .map_err(|e| { - format!( - "write stage timing failed: {}: {e}", - stage_timing_path.display() + crate::logging::artifact::write("stage_timing", &stage_timing_path, || { + std::fs::write( + &stage_timing_path, + serde_json::to_vec_pretty(stage_timing).map_err(|e| e.to_string())?, ) + .map_err(|e| { + format!( + "write stage timing failed: {}: {e}", + stage_timing_path.display() + ) + }) })?; - crate::logging::info!("analysis: wrote {}", stage_timing_path.display()); } Ok(()) } diff --git a/src/runtime/post_validation.rs b/src/runtime/post_validation.rs index 28c543f..feeb34d 100644 --- a/src/runtime/post_validation.rs +++ b/src/runtime/post_validation.rs @@ -40,6 +40,27 @@ fn print_summary_from_shared(validation_time: time::OffsetDateTime, shared: &Pos }) }); if warning_count > 0 { + let mut categories: std::collections::BTreeMap<&str, (usize, Vec)> = std::collections::BTreeMap::new(); + let warnings = shared.tree_warnings.iter().map(|warning| (warning.category.as_str(), warning.message.as_str(), warning.context.as_deref())) + .chain(shared.publication_points.iter().flat_map(|pp| pp.warnings.iter()).map(|warning| (warning.category.as_str(), warning.message.as_str(), warning.context.as_deref()))); + for (category, message, context) in warnings { + let entry = categories.entry(category).or_default(); + entry.0 += 1; + let mut text = crate::logging::redact_text(message); + let truncated = text.len() > 1024; + let mut boundary = text.len().min(1024); + while !text.is_char_boundary(boundary) { boundary -= 1; } + text.truncate(boundary); + if entry.1.len() < 3 { + entry.1.push(serde_json::json!({"message": text, "truncated": truncated})); + } + crate::logging::emit(crate::logging::Level::Debug, "validation_warning_detail", || serde_json::json!({ + "category": category, "message": message, "context": context, + })); + } + for (category, (count, examples)) in categories { + crate::logging::emit(crate::logging::Level::Info, "validation_warning_category", || serde_json::json!({"category": category, "count": count, "examples": examples})); + } crate::logging::emit(crate::logging::Level::Warn, "validation_warnings", || { serde_json::json!({ "count": warning_count, "failed_publication_points": shared.instances_failed, diff --git a/src/runtime/report_tasks.rs b/src/runtime/report_tasks.rs index 7e11e9e..540bd9f 100644 --- a/src/runtime/report_tasks.rs +++ b/src/runtime/report_tasks.rs @@ -80,9 +80,10 @@ pub(super) fn run_ccr_task( .map_err(|e| e.to_string())?; ccr_build_ms = Some(started.elapsed().as_millis() as u64); let started = std::time::Instant::now(); - write_ccr_file(path, &ccr).map_err(|e| e.to_string())?; + crate::logging::artifact::write("ccr", path, || { + write_ccr_file(path, &ccr).map_err(|e| e.to_string()) + })?; ccr_write_ms = Some(started.elapsed().as_millis() as u64); - crate::logging::info!("wrote CCR: {}", path.display()); } Ok(CcrTaskOutput { diff --git a/src/runtime/run.rs b/src/runtime/run.rs index c68363c..23c665c 100644 --- a/src/runtime/run.rs +++ b/src/runtime/run.rs @@ -1,6 +1,7 @@ // Top-level CLI execution pipeline. pub(crate) fn run_config(args: RunConfig) -> Result<(), String> { + crate::logging::observer::phase("state_store_open"); let mut policy = read_policy(args.policy_path.as_deref())?; if let Some(strict_policy) = args.strict_policy { policy.strict = strict_policy; @@ -109,6 +110,7 @@ pub(crate) fn run_config(args: RunConfig) -> Result<(), String> { store.as_ref(), ); let validation_started = std::time::Instant::now(); + crate::logging::observer::phase("runtime_setup"); let collect_current_repo_objects = false; let out = if let Some(dir) = args.rsync_local_dir.as_ref() { let http = BlockingHttpFetcher::new(HttpFetcherConfig { @@ -166,6 +168,7 @@ pub(crate) fn run_config(args: RunConfig) -> Result<(), String> { || serde_json::json!({"elapsed_ms": validation_ms}), ); let shared = PostValidationShared::from_run_output(out); + crate::logging::observer::phase("result_output"); record_memory_checkpoint( &mut memory_checkpoints, "after_validation", @@ -366,8 +369,7 @@ pub(crate) fn run_config(args: RunConfig) -> Result<(), String> { shared.publication_points.len() as u64, ); let timing_json_path = out_dir.join("timing.json"); - t.write_json(&timing_json_path, 20)?; - crate::logging::info!("analysis: wrote {}", timing_json_path.display()); + crate::logging::artifact::write("timing", &timing_json_path, || t.write_json(&timing_json_path, 20))?; } // Write the compatibility summary after analysis timing has been flushed. @@ -376,13 +378,13 @@ pub(crate) fn run_config(args: RunConfig) -> Result<(), String> { // point audit, so the final summary can use those authoritative counts. if let Some(path) = args.summary_out_path.as_deref() { let analysis_counts = timing.as_ref().map(|(_, handle)| handle.counts_snapshot()); - write_summary( + crate::logging::artifact::write("summary", path, || write_summary( path, &shared, &args, validation_ms, analysis_counts.as_ref(), - )?; + ))?; } print_summary_from_shared(validation_time, &shared); diff --git a/src/validation/manifest/models_and_process.rs b/src/validation/manifest/models_and_process.rs index 71b6675..47ac23f 100644 --- a/src/validation/manifest/models_and_process.rs +++ b/src/validation/manifest/models_and_process.rs @@ -207,6 +207,19 @@ pub enum ManifestFreshError { IssuerCaLoadFailed { detail: String }, } +impl ManifestFreshError { + pub(crate) fn warning_category(&self) -> crate::output::report::WarningCategory { + use crate::output::report::WarningCategory; + match self { + Self::Storage(_) | Self::RepoSyncFailed { .. } => WarningCategory::Unclassified, + Self::IssuerCaLoadFailed { .. } | Self::EeCertPath(_) | Self::EeResources(_) => WarningCategory::Certificate, + Self::ManifestOutsidePublicationPoint { .. } | Self::ManifestNumberNotIncreasing { .. } + | Self::ThisUpdateNotIncreasing { .. } | Self::HashMismatch { .. } => WarningCategory::SecurityPolicy, + _ => WarningCategory::ManifestCrl, + } + } +} + #[derive(Debug, thiserror::Error)] pub enum ManifestProcessError { #[error("manifest processing failed: {0}")] diff --git a/src/validation/objects/parallel_stage.rs b/src/validation/objects/parallel_stage.rs index 6444b7f..df46955 100644 --- a/src/validation/objects/parallel_stage.rs +++ b/src/validation/objects/parallel_stage.rs @@ -145,6 +145,7 @@ pub(crate) fn prepare_publication_point_for_parallel_roa_and_ta_constraints< stats.publication_point_dropped = true; warnings.push( Warning::new("dropping publication point: no CRL files in validated publication point") + .with_category(WarningCategory::ManifestCrl) .with_rfc_refs(&[RfcRef("RFC 6487 §4.8.6"), RfcRef("RFC 9286 §7")]) .with_context(manifest_uri), ); @@ -246,6 +247,7 @@ pub(crate) fn reduce_parallel_roa_stage( refs.extend_from_slice(extra_rfc_refs_for_crl_selection(&error)); warnings.push( Warning::new(format!("dropping invalid ROA: {}: {error}", file.rsync_uri)) + .with_category(WarningCategory::ObjectValidation) .with_rfc_refs(&refs) .with_context(&file.rsync_uri), ); @@ -296,6 +298,7 @@ pub(crate) fn reduce_parallel_roa_stage( "dropping invalid ASPA: {}: {error}", file.rsync_uri )) + .with_category(WarningCategory::ObjectValidation) .with_context(&file.rsync_uri), ); } diff --git a/src/validation/objects/serial_processing.rs b/src/validation/objects/serial_processing.rs index 66444b3..8aea251 100644 --- a/src/validation/objects/serial_processing.rs +++ b/src/validation/objects/serial_processing.rs @@ -142,6 +142,7 @@ pub fn process_publication_point_for_issuer_with_ta_constraints Result, RunTreeFromTalError> { let mut roots = Vec::with_capacity(tal_inputs.len()); + crate::logging::observer::phase("trust_anchor_discovery"); for tal_input in tal_inputs { + crate::logging::emit(crate::logging::Level::Info, "trust_anchor_started", || serde_json::json!({"tal_id": tal_input.tal_id})); let discovery = match root_discovery_from_tal_input( tal_input, http_fetcher, @@ -292,7 +294,10 @@ fn discover_multiple_roots_from_tal_inputs( ); continue; } - Err(error) => return Err(error.into()), + Err(error) => { + crate::logging::emit(crate::logging::Level::Error, "trust_anchor_failed", || serde_json::json!({"tal_id": tal_input.tal_id, "error": error.to_string()})); + return Err(error.into()); + } }; let root_handle = root_handle_from_trust_anchor( &discovery.trust_anchor, @@ -305,6 +310,7 @@ fn discover_multiple_roots_from_tal_inputs( discovery, root_handle, }); + crate::logging::emit(crate::logging::Level::Info, "trust_anchor_completed", || serde_json::json!({"tal_id": tal_input.tal_id})); } if roots.is_empty() { return Err(RunTreeFromTalError::Setup( @@ -312,6 +318,7 @@ fn discover_multiple_roots_from_tal_inputs( .to_string(), )); } + crate::logging::observer::phase("online_validation"); Ok(roots) } diff --git a/src/validation/tree.rs b/src/validation/tree.rs index 0646455..8719c6a 100644 --- a/src/validation/tree.rs +++ b/src/validation/tree.rs @@ -308,6 +308,12 @@ pub fn run_tree_serial_audit_multi_root( }; instances_processed += 1; + crate::logging::observer::update(instances_processed as u64, || { + serde_json::json!({ + "publication_points_completed": instances_processed, + "publication_points_failed": instances_failed, + }) + }); warnings.extend(res.warnings); warnings.extend(res.objects.warnings); vrps.extend(res.objects.vrps); diff --git a/src/validation/tree_parallel/phase2.rs b/src/validation/tree_parallel/phase2.rs index 8b854b7..a9037b6 100644 --- a/src/validation/tree_parallel/phase2.rs +++ b/src/validation/tree_parallel/phase2.rs @@ -164,6 +164,17 @@ pub fn run_tree_parallel_phase2_audit_multi_root( config, ); + crate::logging::observer::update(finished.len() as u64, || serde_json::json!({ + "publication_points_completed": finished.len(), + "publication_points_started": instances_started, + "ca_queue": ca_queue.len(), "ready_queue": ready_queue.len(), + "waiting_repository_identities": ca_waiting_repo_by_identity.len(), + "publication_points_inflight": inflight_publication_points.len(), + "pending_object_dispatch": pending_roa_dispatch.len(), + "pending_finalization": pending_finalization.len(), + "finalization_inflight": finalize_inflight, + "staging_inflight": staging_inflight, + })); let repo_poll_timeout = event_poll_timeout( &ca_queue, &ready_queue, diff --git a/src/validation/tree_runner/audit_projection.rs b/src/validation/tree_runner/audit_projection.rs index 1f63b90..07a6dae 100644 --- a/src/validation/tree_runner/audit_projection.rs +++ b/src/validation/tree_runner/audit_projection.rs @@ -127,7 +127,7 @@ fn build_publication_point_audit_from_failed_fetch( .map(AuditWarning::from) .collect::>(); warnings.push(AuditWarning::from( - &Warning::new(fresh_error.to_string()).with_context(&ca.manifest_rsync_uri), + &Warning::new(fresh_error.to_string()).with_category(fresh_error.warning_category()).with_context(&ca.manifest_rsync_uri), )); PublicationPointAudit { node_id: None, diff --git a/src/validation/tree_runner/fresh_pipeline.rs b/src/validation/tree_runner/fresh_pipeline.rs index 02a9fa0..2ecccc4 100644 --- a/src/validation/tree_runner/fresh_pipeline.rs +++ b/src/validation/tree_runner/fresh_pipeline.rs @@ -160,6 +160,7 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> { Vec::new(), vec![ Warning::new(format!("child CA discovery failed: {e}")) + .with_category(crate::output::report::WarningCategory::Certificate) .with_rfc_refs(&[RfcRef("RFC 6487 §7.2")]) .with_context(&ca.manifest_rsync_uri), ], diff --git a/src/validation/tree_runner/publication_point_runner.rs b/src/validation/tree_runner/publication_point_runner.rs index 192f377..4f62c3c 100644 --- a/src/validation/tree_runner/publication_point_runner.rs +++ b/src/validation/tree_runner/publication_point_runner.rs @@ -103,6 +103,7 @@ impl<'a> PublicationPointRunner for Rpkiv1PublicationPointRunner<'a> { "publication point processing failed: {}", stage_error.error )) + .with_category(stage_error.error.warning_category()) .with_rfc_refs(&[RfcRef("RFC 9286 §6.6")]) .with_context(&ca.manifest_rsync_uri), ); diff --git a/tests/logging_contract.rs b/tests/logging_contract.rs new file mode 100644 index 0000000..82eb841 --- /dev/null +++ b/tests/logging_contract.rs @@ -0,0 +1,70 @@ +use std::process::Command; + +#[test] +fn default_format_is_human_readable_text() { + let result = Command::new(env!("CARGO_BIN_EXE_panda-rpki")) + .env_remove("PANDA_RPKI_LOG_FORMAT") + .env_remove("PANDA_RPKI_LOG_LEVEL") + .args(["validate"]) + .output() + .unwrap(); + let diagnostic = String::from_utf8(result.stderr).unwrap(); + assert_eq!(result.status.code(), Some(2)); + assert!(diagnostic.contains("ERROR Command failed")); + assert!(!diagnostic.contains("command_failed")); + let call_line = include_str!("../src/main.rs") + .lines() + .position(|line| line.contains("panda_rpki::logging::emit(")) + .unwrap() + + 1; + assert!(diagnostic.ends_with(&format!("[src/main.rs:{call_line}]\n"))); + assert!(result.stdout.is_empty()); +} + +#[test] +fn invalid_progress_interval_is_rejected_without_outputs() { + for value in ["-1", "wrong", "18446744073709551616"] { + let result = Command::new(env!("CARGO_BIN_EXE_panda-rpki")) + .args(["validate", "--log-progress-interval-secs", value]) + .output() + .unwrap(); + assert_eq!(result.status.code(), Some(2)); + assert!(result.stdout.is_empty()); + assert!(String::from_utf8_lossy(&result.stderr).contains("log-progress-interval-secs")); + } +} + +#[test] +fn progress_cli_overrides_invalid_environment() { + let result = Command::new(env!("CARGO_BIN_EXE_panda-rpki")) + .env("PANDA_RPKI_LOG_PROGRESS_INTERVAL_SECS", "invalid") + .args([ + "validate", + "--log-progress-interval-secs", + "0", + "--log-format", + "json", + ]) + .output() + .unwrap(); + let record: serde_json::Value = serde_json::from_slice(&result.stderr).unwrap(); + let call_line = include_str!("../src/main.rs") + .lines() + .position(|line| line.contains("panda_rpki::logging::emit(")) + .unwrap() + + 1; + assert_eq!(record["source"]["file"], "src/main.rs"); + assert_eq!(record["source"]["line"], call_line); + assert!( + record["fields"]["error"] + .as_str() + .unwrap() + .contains("--tal") + ); + assert!( + !record["fields"]["error"] + .as_str() + .unwrap() + .contains("progress") + ); +} diff --git a/tests/support/test_daemon.py b/tests/support/test_daemon.py index b9bf59d..8b358e6 100644 --- a/tests/support/test_daemon.py +++ b/tests/support/test_daemon.py @@ -86,6 +86,16 @@ def run(binary): assert gap >= 1.9, gap for seq in (1, 2): out = root / f"runs/run_{seq:06}" + assert not (out / "stderr.log").exists() + assert not (out / "stdout.log").exists() + assert (out / "process-output.txt").read_bytes() == b"" + records = [json.loads(line) for line in (out / "validator.log").read_text().splitlines()] + assert records[-1]["event"] == "validation_completed", records[-1] + assert all(record["source"]["line"] > 0 and record["source"]["file"].endswith(".rs") for record in records) + assert all(record["source"]["file"] not in ("src/logging.rs", "src/logging/progress.rs") for record in records) + assert all(record["fields"]["run_seq"] == seq for record in records) + assert len({record["fields"]["run_id"] for record in records}) == 1 + assert any(record["event"] == "artifact_write_completed" for record in records) assert len((out / "vrps.csv").read_text().splitlines()) - 1 == seq subprocess.run(["openssl", "asn1parse", "-inform", "DER", "-in", str(out / "result.ccr"), "-noout"], check=True) @@ -106,6 +116,15 @@ def run(binary): assert state_is(root, "exited") # A real validator failure is recorded, retried, and returned as nonzero at max-runs. + output_failure = subprocess.run( + [binary, "validate", *child_args, "--out", str(base / "bad-output"), "--ccr-out", str(base)], + capture_output=True, text=True, timeout=30, + ) + assert output_failure.returncode == 2 and not output_failure.stdout + failure_records = [json.loads(line) for line in output_failure.stderr.splitlines()] + assert any(record["event"] == "artifact_write_failed" and record["fields"]["artifact_type"] == "ccr" for record in failure_records) + assert any(record["event"] == "validation_stage_failed" for record in failure_records) + assert not any(record["event"] == "validation_completed" for record in failure_records) failed = base / "failed" bad_inputs = ["--tal", str(base / "missing.tal"), "--ta", str(base / "missing.cer")] failure = launch(failed, ["--max-runs", "2", "--interval-secs", "0"], bad_inputs) @@ -115,10 +134,33 @@ def run(binary): # Suspend the HTTPS server so the actual validator cannot complete a request. server.send_signal(signal.SIGSTOP) + slow_inputs = list(child_args) + slow_inputs[slow_inputs.index("--http-timeout-secs") + 1] = "30" + slow_inputs += ["--log-progress-interval-secs", "1"] + progress_root = base / "progress" + progress = launch(progress_root, ["--max-runs", "1", "--run-timeout-secs", "20"], slow_inputs) + progress_log = progress_root / "runs/run_000001/validator.log" + + def has_progress(): + if not progress_log.exists(): + return False + lines = progress_log.read_text().splitlines() + return sum('"event":"validation_progress"' in line for line in lines) >= 2 + + wait_for(has_progress, "periodic progress during blocked network", timeout=15) + server.send_signal(signal.SIGCONT) + assert progress.wait(timeout=30) == 0 + progress_records = [json.loads(line) for line in progress_log.read_text().splitlines()] + assert progress_records[-1]["event"] == "validation_completed" + assert any(record["event"] == "validation_progress" and record["fields"]["repository_attempts_inflight"] > 0 for record in progress_records) + server.send_signal(signal.SIGSTOP) hung = base / "timeout" - timeout = launch(hung, ["--max-runs", "1", "--run-timeout-secs", "1"]) + disabled_inputs = list(slow_inputs) + disabled_inputs[-1] = "0" + timeout = launch(hung, ["--max-runs", "1", "--run-timeout-secs", "2"], disabled_inputs) assert timeout.wait(timeout=10) == 2 assert read(hung / "runs/run_000001/run-summary.json")["error"] == "timeout" + assert '"event":"validation_progress"' not in (hung / "runs/run_000001/validator.log").read_text() interrupted = base / "interrupted" active_proc = launch(interrupted, ["--shutdown-grace-secs", "0"]) wait_for(lambda: state_is(interrupted, "running"), "active validator")