Logging & Observability

Warning

Most of this page still describes proposed logging, metrics, and provider-observability contracts. The redaction boundary in Redaction at implemented egress points is implemented and tested; do not infer that the surrounding proposed logging integrations are available.

How the framework logs, what it logs at each layer, and how to observe Spark / Vertex Batch jobs in production. Also covers the error-retry policy and the structured-error model.

Redaction at implemented egress points

rfgen.redaction provides the implemented, fail-closed secret-redaction boundary. It is applied by the core structured log serializers, CLI JSON output paths, and shard-error rendering. It is not a general data-loss prevention system and does not make arbitrary application-owned sinks safe. Callers that add a new diagnostic, trace, manifest, or log egress path must apply the default policy before serialization.

The immutable RedactionPolicyV1 policy accepts only exact mapping keys, RE2-compatible regular-expression patterns, and the fixed replacement value [REDACTED]. Policy construction rejects unknown fields, unsupported regular expression syntax, more than 100 patterns, patterns larger than 1 KiB, and empty exact-key names. The default policy matches these case-insensitive mapping keys: access_key, api_key, authorization, credential, password, secret, and token. It also detects common assignment forms, AWS access-key IDs, Google API-key prefixes, and Bearer credentials.

redact(value, policy) preserves mappings, lists, and tuples while replacing matching values. It replaces the complete input if JSON serialization is unsafe, the serialized input exceeds 1 MiB, or nesting reaches the supported depth of 32. URL-encoded strings are checked after percent decoding, so a matching encoded secret is replaced as a whole. These fail-closed limits are intentional: a diagnostic that cannot be inspected safely is not emitted in its original form.

redact_exception(error) renders the exception type and a redacted message for each cause or context in the chain. It does not render object dictionaries or tracebacks. Shard errors separately redact both their message and captured traceback before they enter the structured error record.

Scope and limits

• Redaction is pattern and key based. A secret that is neither under a matched key nor matched by a configured pattern can remain visible.

• The default policy protects framework egress points that invoke it. It does not intercept direct writes to stdout, third-party logging handlers, network clients, or application-defined sinks.

• Manifest configuration is durable provenance, not a redaction sink. A resolved manifest configuration containing default-policy material is rejected rather than rewritten, because rewriting would invalidate its canonical content hash.

• Redaction does not replace access control, retention controls, or the annotation metadata whitelist.

Logging strategy

Three principles:

1. Structured by default. JSON lines on stdout. Human-readable mode is opt-in via --log-format=text.

2. One config knob. RFGEN_LOG_LEVEL env var or --log-level=INFO. No per-module knobs in the public surface (set them via Python’s standard logging.getLogger(...).setLevel() in user code if needed).

3. GCP Cloud Logging is automatic when running on Spark serverless or Vertex Workbench. The structured JSON is parsed by Cloud Logging out of the box; no setup required.

# rfgen/core/logging.py
import logging
import json
import sys
from datetime import datetime, timezone


class JsonFormatter(logging.Formatter):
    def format(self, record):
        out = {
            "ts": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "msg": record.getMessage(),
        }
        if record.exc_info:
            out["exc"] = self.formatException(record.exc_info)
        # Pull through any structured extras passed via logger.info(..., extra={...})
        for key in ("sample_id", "shard_id", "preset", "phase",
                    "emitter_family", "channel_backend", "elapsed_ms"):
            if hasattr(record, key):
                out[key] = getattr(record, key)
        return json.dumps(out, default=str)


def setup_logging(level: str = "INFO", fmt: str = "json") -> None:
    handler = logging.StreamHandler(sys.stdout)
    if fmt == "json":
        handler.setFormatter(JsonFormatter())
    else:
        handler.setFormatter(
            logging.Formatter("%(asctime)s %(levelname)-7s %(name)s: %(message)s")
        )
    root = logging.getLogger()
    root.setLevel(level.upper())
    root.handlers[:] = [handler]

setup_logging() is called once at CLI startup. Library users (importing rfgen from a notebook) get whatever logger the host configured; the framework does not call setup_logging itself unless invoked through the CLI.

Log levels per layer

Layer	DEBUG	INFO	WARNING	ERROR
rfgen.core	RNG seed derivation, registry lookups	(none)	(none)	unknown plugin
rfgen.emitters.*	per-sample synthesis params	family + class on emitter init	param out of typical range	synthesis failed
rfgen.channels.*	per-realization channel meta	channel name on init	unsupported config; falls back	channel failed
rfgen.scene.*	per-slot placement decisions	scene size, density on compose	overlap-policy retry budget exhausted	composition failed
rfgen.labels.*	per-cell rasterization counts	(none)	bbox-seg ±1 hop tolerance hit	cross-modality consistency violation
rfgen.annotators.*	rendered prompt	inference job submission, completion	JSON validation retry	inference job failed; record retried
rfgen.storage.*	per-chunk write	shard open/close	partial-write recovery	shard write failed
rfgen.orchestration.*	per-task seed flow	shard counts, autoscale events	spot termination + resume	job failed

DEBUG is verbose enough to be useful only when investigating a specific bug, so it is disabled in normal runs to keep logs scannable.

Structured fields

Every log line for a generation run carries a stable set of structured fields when applicable:

Field	Type	Where set
ts	ISO 8601 UTC	always
level	str	always
logger	str	always
msg	str	always
run_id	str (UUID)	once per rfgen generate invocation, propagated to every log
preset	str	when generating from a preset
phase	1 | 2	Phase 1 (generation) vs Phase 2 (annotation)
shard_id	str	inside a shard worker
sample_id	str	inside a per-sample loop
emitter_family	str	within emitter logs
channel_backend	str	within channel logs
elapsed_ms	float	on completion logs
error_class	str	on ERROR-level lines

Cloud Logging automatically indexes these as labels; you can filter with jsonPayload.shard_id="abc..." in the Logs Explorer.

Standard log lines

A complete generation invocation produces these reference log lines (level INFO, simplified):

{"ts": "2026-06-04T13:15:02Z", "level": "INFO", "msg": "rfgen.generate starting", "run_id": "5e2c...", "preset": "wideband_baseline_md", "num_samples": 100000}
{"ts": "...", "level": "INFO", "msg": "config validated", "config_hash": "sha256:..."}
{"ts": "...", "level": "INFO", "msg": "submitting Spark batch", "batch_id": "1234abcd-..."}
{"ts": "...", "level": "INFO", "msg": "shard worker started", "shard_id": "abc..."}
{"ts": "...", "level": "DEBUG", "msg": "emitter generated", "emitter_family": "comms", "class": "qpsk", "elapsed_ms": 4.2}
{"ts": "...", "level": "INFO", "msg": "shard complete", "shard_id": "abc...", "samples": 1000, "elapsed_ms": 47230}
{"ts": "...", "level": "INFO", "msg": "all shards complete", "samples": 100000, "elapsed_ms": 1872210}
{"ts": "...", "level": "INFO", "msg": "statistics audit: PASS"}
{"ts": "...", "level": "INFO", "msg": "rfgen.generate done", "output_path": "gs://...", "wall_clock_ms": 1875300, "cost_estimate_usd": 142.40}

What does NOT get logged

By policy:

• Raw IQ samples or any sample-level tensor data. Logs would balloon by orders of magnitude. Use rfgen inspect sample --first N for sample-level inspection.

• Full prompts or inference responses at INFO. They appear at DEBUG only. They contain whitelist-filtered metadata (no raw labels), but they’re verbose.

• Provider API keys, bearer credentials, and configured secret fields at implemented framework egress points. The core structured log serializer applies the default redaction policy after it flattens standard fields and caller attributes. CLI structured output and shard-error message and traceback capture use the same policy.

• PII-shaped strings are not covered by the default redaction policy. Do not represent email addresses, IP addresses, or other personal data as automatically redacted unless a policy explicitly matches them.

Observability for distributed jobs

Spark serverless

Each Spark batch run is one logical Cloud Logging stream:

# Tail logs of a running batch
gcloud dataproc batches logs $BATCH_ID --region us-central1 --tail

# Query in Cloud Logging
gcloud logging read \
    'resource.type="cloud_dataproc_batch" AND
     resource.labels.batch_id="'"$BATCH_ID"'" AND
     jsonPayload.level="ERROR"' \
    --format=json --limit=50

The Spark image’s entrypoint.sh calls rfgen.core.logging.setup_logging(level="INFO", fmt="json") before invoking the worker so all per-shard logs flow through Cloud Logging structured-field parsing.

Vertex AI Batch Prediction

Vertex Batch jobs log per-request status and per-job summary. The framework’s coordinator polls the job and emits its own INFO log lines for major transitions:

{"level": "INFO", "msg": "Vertex Batch submitted", "job_id": "...", "annotation_type": "caption", "model_tier": "bulk"}
{"level": "INFO", "msg": "Vertex Batch progress", "job_id": "...", "completed": 78213, "total": 100000}
{"level": "INFO", "msg": "Vertex Batch complete", "job_id": "...", "elapsed_ms": 1834000}
{"level": "INFO", "msg": "validating responses", "job_id": "..."}
{"level": "WARNING", "msg": "JSON validation failure", "sample_id": "...", "attempt": 1}
{"level": "INFO", "msg": "appended to Zarr", "job_id": "...", "annotation_type": "caption"}

Local runs

Logs go to stdout. Pipe to jq for human-friendly viewing:

rfgen generate +preset=narrowband_xs run.num_samples=100 2>&1 | jq -r '"\(.ts) \(.level) \(.msg)"'

Or use the text formatter:

rfgen generate ... --log-format=text --log-level=DEBUG

Metrics emission

Beyond logs, the framework emits Prometheus-style counters / histograms for the orchestrator. These are exposed on localhost:9090 during local runs and to Cloud Monitoring on Spark.

Metric	Type	Labels
rfgen_samples_generated_total	counter	preset, shard_id, phase
rfgen_sample_duration_seconds	histogram	preset, phase
rfgen_shard_duration_seconds	histogram	preset
rfgen_errors_total	counter	error_class, phase
rfgen_emitter_synth_seconds	histogram	family, class
rfgen_channel_apply_seconds	histogram	backend
rfgen_label_consistency_failures_total	counter	invariant (1/2/3/4)
rfgen_annotation_paes	gauge	annotation_type, template_id
rfgen_annotation_hallucination_count	gauge	annotation_type, template_id
rfgen_llm_requests_total	counter	provider, model, outcome
rfgen_llm_tokens_total	counter	provider, model, direction (in/out)
rfgen_llm_cost_usd_total	counter	provider, model

Implemented via prometheus_client. The HTTP endpoint is started by the CLI when --metrics-port=N is passed; otherwise metrics are silently collected and emitted at run end as a summary log line.

Error model and retry policy

Defined in rfgen.core.errors:

RfgenError                       # base
├── ConfigurationError           # invalid config (caught before generation starts); fail fast
├── PluginNotFoundError          # registry miss; fail fast
├── UnsupportedConfiguration     # plugin can't fulfil this combo (e.g. multi-RX on AWGN); fail fast
├── GenerationError              # runtime fault during generate/apply/compose
│   ├── EmitterError             # retry once, then exclude sample
│   ├── ChannelError             # retry once, then exclude sample
│   └── SceneError               # retry once, then exclude sample
├── LabelError
│   ├── LabelInconsistencyError  # exclude sample; do not retry (data-driven, not transient)
│   └── LabelComputeError        # retry once
├── AnnotationError
│   ├── LLMError                 # retry up to 3× with exponential backoff
│   ├── LLMRateLimitError        # honor Retry-After; cap at 60s; retry until config.max_retries
│   ├── LLMRefusalError          # do not retry (model declined to answer); log and skip
│   └── HallucinationError       # do not retry; flag in audit
└── StorageError
    ├── StorageTransientError    # retry up to 3× with exponential backoff
    └── StoragePermanentError    # fail the shard

Retry policy: concrete numbers

Error class	Max retries	Backoff	Strategy
EmitterError / ChannelError / SceneError	1	none	retry with temperature × 0.5 if the original used randomness; else fail to errors.jsonl
LabelComputeError	1	none	bug-shaped; usually indicates code error
LLMError (transient)	3	exponential 1s/2s/4s + 0–500 ms jitter	retry the same prompt
LLMRateLimitError	unlimited (until config.max_retries cap)	honor Retry-After header; cap at 60 s	the framework respects per-provider tier limits proactively, but transient bursts still trigger this
LLMRefusalError	0	(none)	exclude sample; record in errors.jsonl
HallucinationError	0	(none)	exclude sample; flag in audit
StorageTransientError	3	exponential 0.5 s / 1 s / 2 s	retry write
StoragePermanentError	0	(none)	fail shard; let orchestrator re-run shard if idempotent

Errors output sidecar

Per-sample errors land in errors.jsonl next to the dataset:

gs://rf-fm-datasets-synth/dense_urban/v0.5.0/
├── manifest.json
├── samples.zarr/
├── stats.json
└── errors.jsonl              # one line per excluded sample

{"sample_id": "f9a3b7c2-...", "shard_id": "abc...", "error_class": "EmitterError",
 "error_msg": "LoRa SF12 BW125 demands sample_rate >= 1 MHz; got 500 kHz",
 "retry_count": 1, "ts": "2026-06-04T13:15:14Z"}

A run is rejected if errors.jsonl has > 0.1 % of num_samples (per Reference / Metrics § Schema validation).

Audit logging

Distinct from the operational logs above. The audit log is written once per run to manifest.json["audit_log"]:

{
  "audit_log": {
    "schema_validation": {
      "samples_total": 100000,
      "samples_excluded": 24,
      "exclusion_rate": 0.00024,
      "passed": true
    },
    "statistics": [
      {"name": "snr_distribution_ks", "value": 0.034, "threshold": 0.05, "passed": true},
      {"name": "class_balance",       "value": null,   "passed": true},
      {"name": "density",             "value": 0.018, "threshold": 0.05, "passed": true},
      ...
    ],
    "annotation_audit": {
      "verifier_subset_size": 10000,
      "dataset_paes": 0.91,
      "dataset_hallucination_count": 0.31,
      "per_template_paes": {...},
      "per_class_paes": {...},
      "passed": true
    }
  }
}

This block is the release gate. CI reads it and blocks promotion if any check fails.

Local-development tips

# Tail your local generation in human-readable form
rfgen generate +preset=narrowband_xs --log-format=text 2>&1 | grep -E "INFO|WARNING|ERROR"

# Filter to errors only, JSON
rfgen generate +preset=narrowband_xs 2>&1 | jq 'select(.level=="ERROR")'

# Per-shard timing
rfgen generate +preset=narrowband_xs 2>&1 | jq 'select(.msg=="shard complete") | {shard_id, elapsed_ms}'

# Compute estimate after a run
jq '.audit_log' out/run-*/manifest.json

See Also

• Reference / Metrics: what the audit log measures

• Reference / Build and CI § Pre-commit hooks: no-non-deterministic-seeds lint rule

• Reference / Determinism: what run_id and shard_id mean

• How-to / Run on Spark: Cloud Logging integration