Structured Logging skill for Codex
JSON logs you can query, correlate and trust. Codex replaces ad hoc printf logs with one JSON event per line, a documented field schema, constant messages with variables in fields, and log levels that mean something to on-call. It binds a request or trace ID into context so every line correlates, redacts tokens and PII at the logger, and checks captured output with a bundled validator script.
curl -fsSL codexskills.dev/install.sh \ | sh -s -- structured-loggingInstalls to ~/.agents/skills/structured-logging/ for every project on this machine.
curl -fsSL codexskills.dev/cli.mjs -o codex-skills.mjsnode codex-skills.mjs install structured-loggingThe CLI also searches, updates and removes skills. Run it with no arguments to pick from a list. Needs Node 20.12+.
curl -fsSL codexskills.dev/install.sh \ | sh -s -- --project structured-loggingInstalls to .agents/skills/structured-logging/ in the current repository. Commit it and everyone who runs Codex there gets it.
curl -fsSLO codexskills.dev/downloads/structured-logging.zipunzip -o structured-logging.zip -d ~/.agents/skills/The archive holds the skill folder with its generated agents/openai.yaml. Unzip into .agents/skills/ to scope it to a repo.
& ([scriptblock]::Create((irm codexskills.dev/install.ps1))) structured-loggingPowerShell 5.1 or 7. Installs to %USERPROFILE%\.agents\skills\. Add -Project for the current repo.
104-line POSIX script. No sudo, no telemetry. Read install.sh
Use it in Codex
Start a prompt with $structured-logging, or pick it from /skills.
You can also just describe the task: Codex reads the description below and loads the playbook when your request matches.
What Codex reads before loading it
Designs and retrofits structured application logs - one JSON object per line, a stable field schema, honest severity levels, request and trace correlation IDs, and redaction of secrets and personal data - using pino, structlog, Go log/slog, Logback with logstash-logback-encoder or Serilog. Use when logs are unsearchable printf strings, when adding request IDs or trace_id to logs, when choosing log levels, when log volume or cost is too high, or when reviewing logging code for leaked tokens or PII.
Use it when
- Logs are free text and cannot be filtered by field
- Requests cannot be followed across services or into background jobs
- Log levels are inconsistent and errors are drowned in noise
- A review or incident found secrets or PII in logs
Use something else when
- You need full tracing with spans and context propagation: use opentelemetry
- The task is querying logs already in Grafana Loki: use loki-logs
- You are designing security audit trails and detection logging: use security-logging
How to check the result
Capture a few hundred lines of real output and run scripts/check_logs.py with the required fields; it must report zero non-JSON lines, missing fields, type conflicts or secret-like values.
The mistake it prevents
Interpolating IDs into the message text, which makes every line unique and defeats grouping, counting and alerting.
Done well means: every line is one machine-parseable event with the same core fields, any request can be followed across services by one ID, levels mean something an on-call engineer can filter on, and no secret or personal identifier ever reaches the log pipeline.
Before changing anything
- Read
AGENTS.mdand find the existing logger:rg -n "getLogger|pino\(|structlog|slog\.|LoggerFactory|Log\.Logger" -g '!**/node_modules/**'. - Find where logs go: stdout collected by an agent (Kubernetes, ECS, systemd), files, or a vendor SDK. Containers should log to stdout; the platform ships it.
- Find who reads them: Loki, Elasticsearch/OpenSearch, CloudWatch, Datadog. Field names should match what that backend indexes or maps (ECS for Elastic, OTel names for OTLP pipelines).
- Grep for existing correlation:
rg -n "request_id|requestId|trace_id|X-Request-ID". Extend what exists rather than adding a second ID.
Field schema
Pick one schema per organisation and write it down (in AGENTS.md or a logging doc).
Default core fields:
| Field | Example | Notes |
|---|---|---|
timestamp | 2026-03-14T09:26:53.589Z | RFC 3339 UTC with milliseconds; never local time |
level | error | lowercase string from the fixed set below |
message | payment declined | constant text; no interpolated IDs |
service | checkout-api | matches OTel service.name |
env / version | prod, 2026.03.1 | from deploy metadata, not code |
trace_id, span_id | 32 and 16 hex chars | from the active OTel span when tracing exists |
request_id | 01HV... | generated at the edge if no trace ID exists |
event | payment.declined | dot-namespaced, stable name for queries and alerts |
error.type, error.message, error.stack | only on failures | |
duration_ms | 184 | numbers as numbers, never "184ms" |
Rules: keep the message constant and put variables in fields
({"message":"order created","order_id":"o_123"}, not "order o_123 created"), so
grouping and counting work. Use snake_case or dotted keys consistently; never mix
userId and user_id. Never reuse a key with different types (status as both
200 and "ok" breaks Elasticsearch mappings).
Levels
| Level | Means | Who acts |
|---|---|---|
error | A request or job failed and the user/caller saw it, or data is at risk | Someone should look; often counted by alerts |
warn | Degraded but handled: retry succeeded, fallback used, limit approaching | Reviewed in aggregate |
info | Business-significant state change: started, order placed, job finished | Default production level |
debug | Developer detail for diagnosis | Off in production; enable per service or per request |
Hard rules: log an exception once, at the boundary that handles it, not at every layer
that rethrows. A 404 or a validation failure from a client is info or warn, not
error. Never log at error inside a retry loop that may still succeed.
Implementation by stack
Node (pino, JSON to stdout by default):
import pino from 'pino';
export const log = pino({
level: process.env.LOG_LEVEL ?? 'info',
base: { service: 'checkout-api', env: process.env.APP_ENV, version: process.env.APP_VERSION },
timestamp: pino.stdTimeFunctions.isoTime,
formatters: { level: (label) => ({ level: label }) },
redact: { paths: ['req.headers.authorization', 'req.headers.cookie', '*.password', '*.token'], censor: '[REDACTED]' },
});
// per request: const reqLog = log.child({ request_id, trace_id });Python (structlog over stdlib):
import logging, structlog
structlog.configure(
processors=[
structlog.contextvars.merge_contextvars,
structlog.processors.add_log_level,
structlog.processors.TimeStamper(fmt="iso", utc=True),
structlog.processors.dict_tracebacks,
structlog.processors.JSONRenderer(),
],
wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
)
log = structlog.get_logger(service="checkout-api")
# middleware: structlog.contextvars.bind_contextvars(request_id=rid, trace_id=tid)Go (standard library log/slog):
logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo}))
logger = logger.With("service", "checkout-api", "version", version)
logger.InfoContext(ctx, "order created", "order_id", id, "duration_ms", ms)Java: Logback with net.logstash.logback.encoder.LogstashEncoder and MDC for
request_id/trace_id. .NET: Serilog with CompactJsonFormatter, or
builder.Logging.AddJsonConsole(). Adding a dependency needs network; ask before
installing in a sandbox with network off.
Correlation IDs
- At the edge (gateway or first service), accept an incoming
traceparent(W3C) orX-Request-ID; generate one if absent. Do not trust arbitrary client-supplied IDs for anything except correlation; cap length and charset. - Bind the ID into request-scoped context (
AsyncLocalStorage,contextvars,context.Context, MDC) so every log line in that request carries it without passing it by hand. - Propagate on every outbound call and message: HTTP headers, queue message attributes, job payload metadata. Background jobs log the ID of the request that enqueued them.
- Return the ID in the response header so a user’s bug report can quote it.
- When OpenTelemetry is present, log
trace_id/span_idfrom the active span instead of inventing a parallel ID; the OTel logging bridges for each language do this automatically.
Redaction and safety
- Deny by default: log explicit fields, never whole request bodies, headers, ORM
objects or
config. AtoJSONon a user object will eventually include a new sensitive column. - Redact at the logger (pino
redact, a structlog processor, a slogReplaceAttr) so one place enforces it. Pipeline-side scrubbing is a second line, not the first. - Never log: passwords, tokens, API keys, session cookies, full card numbers, national IDs, full email bodies. Hash or truncate user identifiers if the logs are widely accessible and the use case allows.
- Escape or encode user input; JSON encoding prevents forged log lines from embedded newlines, plain text logging does not.
Volume and cost
- Do not log per-iteration inside hot loops; aggregate and log once with a count.
- Drop health-check and readiness-probe access logs at the source.
- Sample high-volume success logs (for example 1 in 100
infoaccess lines) but never samplewarn/error. - A metric answers “how many”; a log answers “what happened to this one”. If a log exists only to be counted, emit a counter instead.
Checklist
- One logger configured once; JSON to stdout in production, pretty output only in dev
- Core fields present on every line; message text constant
- Correlation ID bound per request and propagated to outbound calls and jobs
- Levels follow the table; each exception logged once
- Redaction configured and tested with a fake secret
- Noise sources (probes, loops) removed or sampled
Verification
Capture real output and check it mechanically:
LOG_LEVEL=debug npm start 2>&1 | head -200 > /tmp/sample.jsonl # or the service's run command
python3 scripts/check_logs.py /tmp/sample.jsonl --require timestamp,level,message,service
python3 scripts/check_logs.py /tmp/sample.jsonl --require timestamp,level,message --require-one trace_id,request_idThe script (standard library only) reports lines that are not JSON, missing required fields, unknown level values, fields whose type changes between lines, and values that look like secrets (bearer tokens, AWS keys, JWTs, card-like numbers, emails). It exits 1 on any finding. Also add a unit test that logs a fake password through the real logger and asserts it is redacted.
Output
When reporting a logging change, include: the schema table as adopted, an example line
before and after, where correlation is bound and propagated, the redaction list, and
the result of check_logs.py on captured output.
Gotchas
- Logging libraries that buffer asynchronously (pino transports, async appenders) lose the last lines on crash unless flushed on exit; pino writes synchronously to stdout by default, keep it that way unless throughput demands otherwise.
- Multi-line stack traces in plain text become many separate events in the collector. JSON keeps the stack in one field.
print/console.logcalls bypass the logger and its redaction; grep for them.- Child loggers created per request are cheap; global mutable logger context is a race in concurrent servers.
What the zip contains
Codex reads bundled files only when a step points to them. Every one is shown here, so you can review it before installing.
agents/openai.yaml
Gives the Codex app a display name, a one-line description and a starter prompt for this skill.
interface:
display_name: "Structured Logging"
short_description: "JSON logs you can query, correlate and trust"
default_prompt: "$structured-logging Our logs are plain strings and impossible to search in Loki, convert this service to structured JSON logging"scripts/check_logs.py
#!/usr/bin/env python3
"""Check a JSON-lines log sample for structure, schema and leaked secrets.
Usage:
python3 scripts/check_logs.py sample.jsonl --require timestamp,level,message
python3 scripts/check_logs.py sample.jsonl --require-one trace_id,request_id
cat sample.jsonl | python3 scripts/check_logs.py -
Exits 0 when no findings, 1 when any finding, 2 on usage errors. Standard library only.
"""
import argparse
import json
import re
import sys
DEFAULT_LEVELS = {"trace", "debug", "info", "warn", "warning", "error", "fatal", "critical"}
LEVEL_KEYS = ("level", "severity", "log.level", "severity_text", "levelname")
SECRET_PATTERNS = {
"bearer token": re.compile(r"\bBearer\s+[A-Za-z0-9\-._~+/]{16,}=*", re.I),
"jwt": re.compile(r"\beyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}"),
"aws access key": re.compile(r"\b(AKIA|ASIA)[0-9A-Z]{16}\b"),
"private key": re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----"),
"password field": re.compile(r'"(password|passwd|secret|api_key|apikey)"\s*:\s*"(?!\[REDACTED\]|\*+")[^"]+"', re.I),
"email": re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b"),
}
CARD = re.compile(r"\b(?:\d[ -]?){13,19}\b")
def luhn_ok(digits: str) -> bool:
total, alt = 0, False
for ch in reversed(digits):
d = int(ch)
if alt:
d *= 2
if d > 9:
d -= 9
total += d
alt = not alt
return total % 10 == 0
def flatten(obj, prefix=""):
out = {}
if isinstance(obj, dict):
for k, v in obj.items():
key = f"{prefix}.{k}" if prefix else str(k)
if isinstance(v, dict):
out[key] = v # keep the parent so "error" as an object still counts
out.update(flatten(v, key))
else:
out[key] = v
return out
def type_name(v):
if v is None:
return "null"
if isinstance(v, bool):
return "bool"
if isinstance(v, (int, float)):
return "number"
if isinstance(v, str):
return "string"
if isinstance(v, list):
return "array"
return "object"
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("path", help="JSON-lines file, or - for stdin")
ap.add_argument("--require", default="", help="comma-separated fields every line must have (dotted for nested)")
ap.add_argument("--require-one", default="", help="comma-separated fields; each line needs at least one")
ap.add_argument("--levels", default=",".join(sorted(DEFAULT_LEVELS)), help="allowed level values")
ap.add_argument("--allow-email", action="store_true", help="do not flag email addresses")
ap.add_argument("--max-report", type=int, default=20, help="max findings printed per category")
args = ap.parse_args()
required = [f for f in args.require.split(",") if f]
one_of = [f for f in args.require_one.split(",") if f]
levels = {x.strip().lower() for x in args.levels.split(",") if x.strip()}
try:
fh = sys.stdin if args.path == "-" else open(args.path, encoding="utf-8", errors="replace")
except OSError as exc:
print(f"error: {exc}", file=sys.stderr)
return 2
findings: dict[str, list[str]] = {}
field_types: dict[str, dict[str, int]] = {}
total = parsed = 0
def add(cat, msg):
findings.setdefault(cat, []).append(msg)
with fh:
for n, line in enumerate(fh, 1):
line = line.strip()
if not line:
continue
total += 1
try:
obj = json.loads(line)
except json.JSONDecodeError:
add("not JSON", f"line {n}: {line[:100]}")
continue
if not isinstance(obj, dict):
add("not a JSON object", f"line {n}")
continue
parsed += 1
flat = flatten(obj)
for f in required:
if f not in flat:
add("missing required field", f"line {n}: {f}")
if one_of and not any(f in flat for f in one_of):
add("missing correlation field", f"line {n}: none of {', '.join(one_of)}")
for key in LEVEL_KEYS:
if key in flat and isinstance(flat[key], str):
if flat[key].lower() not in levels:
add("unknown level", f"line {n}: {key}={flat[key]!r}")
break
for k, v in flat.items():
if isinstance(v, dict):
continue
field_types.setdefault(k, {}).setdefault(type_name(v), n)
for name, rx in SECRET_PATTERNS.items():
if name == "email" and args.allow_email:
continue
if rx.search(line):
add("possible secret or PII", f"line {n}: {name}")
for m in CARD.finditer(line):
digits = re.sub(r"\D", "", m.group())
if 13 <= len(digits) <= 19 and luhn_ok(digits):
add("possible secret or PII", f"line {n}: card-like number")
break
for k, types in field_types.items():
real = {t: ln for t, ln in types.items() if t != "null"}
if len(real) > 1:
desc = ", ".join(f"{t} (first line {ln})" for t, ln in sorted(real.items()))
add("field type conflict", f"{k}: {desc}")
print(f"lines: {total}, parsed JSON objects: {parsed}")
if not findings:
print("OK: no findings")
return 0
for cat, items in findings.items():
print(f"\n{cat}: {len(items)}")
for item in items[: args.max_report]:
print(f" {item}")
if len(items) > args.max_report:
print(f" ... {len(items) - args.max_report} more")
return 1
if __name__ == "__main__":
sys.exit(main())
About this skill
Written and maintained by the Codex Skills editors to the same editorial standard as every skill here: a selective description, a real workflow, known pitfalls and a way to verify the work. First published , last updated . MIT licensed. Something wrong or out of date? Tell us.