Posted on
• Artificial Intelligence

Artificial Intelligence Python Logging

Author
  • User
    linuxbash
    Posts by this author
    Posts by this author

Artificial Intelligence Python Logging on Linux: From “It Works” to “We Can Prove It”

Your model is smart, but your logs are smarter. In AI systems, failures are often non-deterministic, expensive, and hard to reproduce. Good logging turns “it looked slow” into “p95 inference latency increased 37% after model upgrade.” This article shows you how to build production-grade logging for Python AI workloads on Linux—complete with structured JSON logs, journald integration, redaction, and rotation—using tools you already have.

What you’ll get:

  • Why AI workloads demand better logging

  • A minimal, reusable Python logging setup with JSON + redaction

  • Options to ship logs to systemd-journald or files (with logrotate)

  • Practical commands for apt, dnf, and zypper

  • Real-world patterns for latency, correlation IDs, and safe prompt logging

Why AI logging is different (and critical)

  • Non-determinism and drift: Outputs change with temperature, prompt tweaks, or model updates. Logs anchor what happened and when.

  • Latency and cost: Tokenization and remote calls introduce variable latency and usage costs. Timed, structured logs quantify impact.

  • Safety and compliance: You must avoid leaking PII, API keys, or customer prompts in plaintext logs. Redaction and hashing help.

  • Operations and SRE: Journald and JSON logs make it easy to filter, alert, and investigate issues fast on Linux.

0) Prerequisites and installation

Install Python, pip, journald bindings, logrotate, and jq (optional, for pretty-printing JSON):

On Debian/Ubuntu (apt):

sudo apt update
sudo apt install -y python3 python3-pip python3-venv python3-systemd logrotate jq

On Fedora/RHEL/CentOS (dnf):

sudo dnf install -y python3 python3-pip python3-systemd logrotate jq

On openSUSE/SLE (zypper):

sudo zypper refresh
sudo zypper install -y python3 python3-pip python3-systemd logrotate jq

Optional: Create and activate a virtual environment (Debian/Ubuntu require python3-venv above):

python3 -m venv .venv
. .venv/bin/activate

1) Structured JSON logging with safe defaults (drop-in module)

Save this as logging_setup.py. It emits JSON logs to console by default, or to journald/file based on environment variables. It includes a redaction filter for common secrets.

# logging_setup.py
import json, logging, os, sys, uuid, traceback
from datetime import datetime

REDACT_KEYS = {"api_key", "authorization", "password", "token", "secret"}

class RedactFilter(logging.Filter):
    def filter(self, record):
        # Scrub sensitive fields if users pass them via "extra={}"
        if hasattr(record, "__dict__"):
            for k in list(record.__dict__.keys()):
                if k.lower() in REDACT_KEYS and record.__dict__[k] is not None:
                    record.__dict__[k] = "***REDACTED***"
        # Light message scrub: common key=... patterns
        if isinstance(record.msg, str):
            scrub = record.msg
            for k in REDACT_KEYS:
                # naive pattern: key=<anything without whitespace>
                scrub = scrub.replace(f"{k}=", f"{k}=***REDACTED***")
            record.msg = scrub
        return True

class JsonFormatter(logging.Formatter):
    def format(self, record):
        base = {
            "ts": datetime.utcnow().isoformat(timespec="milliseconds") + "Z",
            "level": record.levelname,
            "logger": record.name,
            "pid": os.getpid(),
            "message": record.getMessage(),
        }
        # Merge extras safely
        for k, v in record.__dict__.items():
            if k in ("args", "msg", "levelname", "levelno", "pathname", "filename",
                     "module", "exc_info", "exc_text", "stack_info", "lineno", "funcName",
                     "created", "msecs", "relativeCreated", "thread", "threadName",
                     "processName", "process", "name"):
                continue
            # Best-effort JSON-serializable
            try:
                json.dumps(v)
                base[k] = v
            except Exception:
                base[k] = str(v)
        # exception serialization
        if record.exc_info:
            base["exc_type"] = record.exc_info[0].__name__
            base["exc_message"] = str(record.exc_info[1])
            base["exc_traceback"] = "".join(traceback.format_exception(*record.exc_info))
        return json.dumps(base, ensure_ascii=False)

def get_logger(
    name="myai",
    level=os.getenv("LOG_LEVEL", "INFO"),
    dest=os.getenv("LOG_DEST", "console"),  # console | file | journald
    logfile=os.getenv("LOG_FILE", "/tmp/myai/app.jsonl"),
    syslog_id=os.getenv("SYSLOG_IDENTIFIER", "myai"),
):
    logger = logging.getLogger(name)
    if logger.handlers:
        return logger  # already configured

    logger.setLevel(getattr(logging, level.upper(), logging.INFO))
    formatter = JsonFormatter()
    filt = RedactFilter()

    handler = None
    if dest == "journald":
        try:
            from systemd.journal import JournalHandler
            handler = JournalHandler(SYSLOG_IDENTIFIER=syslog_id)
        except Exception:
            # Fallback to console if journald unavailable
            handler = logging.StreamHandler(sys.stdout)
    elif dest == "file":
        # Ensure directory exists or fall back to console
        try:
            os.makedirs(os.path.dirname(logfile), exist_ok=True)
            handler = logging.FileHandler(logfile, encoding="utf-8")
        except Exception:
            handler = logging.StreamHandler(sys.stdout)
    else:
        handler = logging.StreamHandler(sys.stdout)

    handler.addFilter(filt)
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    logger.propagate = False
    return logger

def new_correlation_id():
    return os.getenv("CORRELATION_ID") or str(uuid.uuid4())

Usage notes:

  • LOG_DEST=console|file|journald controls output.

  • LOG_FILE=/path/to/app.jsonl sets the file path if using file logs.

  • SYSLOG_IDENTIFIER influences journald tagging (-t in journalctl).

  • LOG_LEVEL=DEBUG for verbose debugging.

2) Instrument AI inference: latency, correlation IDs, and safe prompt logging

This example wraps an “inference” call, logs request/response events, tracks latency, and avoids logging raw prompts by default.

Save as app.py:

# app.py
import os, time, hashlib, random
from logging_setup import get_logger, new_correlation_id

logger = get_logger(dest=os.getenv("LOG_DEST", "console"))

def sha256_hex(s: str) -> str:
    return hashlib.sha256(s.encode("utf-8")).hexdigest()

def simulate_model_inference(prompt: str, model="toy-1"):
    # Simulate variable latency (e.g., network + generation)
    time.sleep(random.uniform(0.05, 0.30))
    # Mock "AI response"
    return f"Echo: {prompt[:50]}..." if len(prompt) > 50 else f"Echo: {prompt}"

def run_inference(prompt: str, model="toy-1", user_id=None):
    corr_id = new_correlation_id()
    t0 = time.perf_counter()

    # Log request WITHOUT raw prompt to protect privacy by default
    logger.info(
        "ai.request",
        extra={
            "event": "ai.request",
            "correlation_id": corr_id,
            "model": model,
            "prompt_sha256": sha256_hex(prompt),
            "prompt_tokens_est": len(prompt.split()),
            "user_id": user_id,
        },
    )

    try:
        result = simulate_model_inference(prompt, model=model)
        dt_ms = round((time.perf_counter() - t0) * 1000, 2)

        # Log response (again avoid raw content by default)
        logger.info(
            "ai.response",
            extra={
                "event": "ai.response",
                "correlation_id": corr_id,
                "model": model,
                "latency_ms": dt_ms,
                "response_len": len(result),
                "ok": True,
            },
        )
        return result
    except Exception as e:
        dt_ms = round((time.perf_counter() - t0) * 1000, 2)
        logger.error(
            "ai.error",
            extra={
                "event": "ai.error",
                "correlation_id": new_correlation_id(),
                "model": model,
                "latency_ms": dt_ms,
                "ok": False,
            },
            exc_info=True,
        )
        raise

if __name__ == "__main__":
    prompt = os.getenv("PROMPT", "Explain the benefits of structured logging.")
    out = run_inference(prompt, model=os.getenv("MODEL", "toy-1"), user_id=os.getenv("USER_ID"))
    print(out)

Run it:

# Console logs (default)
python3 app.py

# Journald logs
LOG_DEST=journald SYSLOG_IDENTIFIER=myai python3 app.py

# File logs
LOG_DEST=file LOG_FILE=/tmp/myai/app.jsonl python3 app.py

Tip: If you absolutely must log raw prompts or responses, add a feature flag and a redaction pass first. A safer compromise is logging only hashes, sizes, and verdicts.

3) Redaction patterns you should adopt on day one

  • Strip or hash raw prompts/responses; log a hash + token/length estimates instead.

  • Never log secrets: API keys, OAuth tokens, passwords, or Authorization headers.

  • Prefer correlation IDs over PII. If you must link to a user, use a pseudonymous ID and store the mapping elsewhere with stricter access controls.

  • Use a dedicated redaction filter (already included above) and expand REDACT_KEYS to fit your org.

If you need to pass secrets through extra={}, they’ll be replaced with REDACTED.

4) Send logs to journald or files (and rotate safely)

A) Journald (systemd)

  • Ensure python3-systemd is installed (see package installs above).

  • Run with LOG_DEST=journald. View logs:

# Follow all logs from this identifier
journalctl -t myai -f

# Pretty JSON
journalctl -t myai -o json-pretty | jq .

B) File logs with rotation

  • By default, the sample writes to /tmp. For a persistent path:
sudo install -d -m 0755 -o "$USER" -g "$USER" /var/log/myai
LOG_DEST=file LOG_FILE=/var/log/myai/app.jsonl python3 app.py
  • Add a logrotate policy at /etc/logrotate.d/myai:
/var/log/myai/app.jsonl {
    daily
    rotate 14
    compress
    delaycompress
    missingok
    notifempty
    copytruncate
}
  • Test logrotate:
sudo logrotate -d /etc/logrotate.conf

Notes:

  • copytruncate helps rotate files without stopping your app.

  • For high-throughput services, consider writing to stdout and letting journald handle rotation.

5) Fast queries and debugging on Linux

  • Tail live journald logs for your app:
journalctl -t myai -f
  • Show only errors:
journalctl -t myai -p err
  • Pretty-print JSON and filter by correlation_id:
journalctl -t myai -o json | jq -r 'select(.MESSAGE | test("ai\\.request|ai\\.response")) | select(.correlation_id=="YOUR_ID_HERE")'
  • For file logs:
tail -f /tmp/myai/app.jsonl | jq .

Real-world tip: Standardize fields like event, correlation_id, model, latency_ms across services. This makes ad-hoc analysis trivial and charts easy.

Common enhancements

  • Add environment-based sampling to reduce log volume for very high traffic.

  • Emit counters/aggregates periodically (e.g., total calls/min, errors/min).

  • Push structured logs to a centralized stack (Loki, Elastic, OpenSearch) using journald as the source of truth.

  • Attach deployment metadata (git_sha, model_version) to each log event via extra.

Conclusion and next step (CTA)

Observability is not a postscript to AI—it's the backbone of safe, fast iteration. You now have:

  • Structured JSON logs with redaction

  • Journald and file outputs (with rotation)

  • Actionable fields for latency, correlation, and model tracking

Next step: instrument one AI path in your codebase today. Drop in logging_setup.get_logger, wrap your inference calls, and run with LOG_DEST=journald. Use journalctl and jq to validate. Once you see latency and outcomes in your terminal, you’re ready to scale to dashboards and alerts.

If you want a follow-up post on exporting these logs to Loki/OpenSearch with systemd-journald as the source, let me know.

Related posts