0.2.8: structured JSON logging for production deployments

oikb daemon --log-format json (or LOG_FORMAT=json env var) outputs
one JSON object per line with ts, level, logger, msg, and optional
sync-specific fields. Compatible with Datadog, Splunk, ELK,
CloudWatch, and Loki.

Author: tjbck

CHANGELOG.md

@@ -4,6 +4,12 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
+## [0.2.8] - 2025-05-21
+
+### Added
+
+- Structured JSON logging for the daemon — `oikb daemon --log-format json` or `LOG_FORMAT=json` env var. Outputs one JSON object per line with `ts`, `level`, `logger`, `msg`, and optional `source`/`kb_id`/`duration_ms` fields. Compatible with Datadog, Splunk, ELK, CloudWatch, and Loki.
+
 ## [0.2.7] - 2025-05-21
 
 ### Added

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "oikb"
-version = "0.2.7"
+version = "0.2.8"
 description = "Sync anything to Open WebUI Knowledge Bases"
 readme = "README.md"
 authors = [

src/oikb/__init__.py

@@ -1,3 +1,3 @@
 """oikb — Open WebUI Knowledge Base CLI."""
 
-__version__ = "0.2.7"
+__version__ = "0.2.8"

src/oikb/cli.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import os
 import sys
 from pathlib import Path
 
@@ -760,12 +761,15 @@ def validate(config_file: str | None):
 @click.option("--port", default=8080, type=int, help="HTTP port for healthcheck and API (default: 8080).")
 @click.option("--no-server", is_flag=True, help="Run scheduler only, no HTTP server.")
 @click.option("--config", "config_file", default=None, type=click.Path(), help="Path to .oikb.yaml (default: ./.oikb.yaml).")
-def daemon(port: int, no_server: bool, config_file: str | None):
+@click.option("--log-format", default=None, type=click.Choice(["text", "json"]), help="Log output format (default: text, env: LOG_FORMAT).")
+def daemon(port: int, no_server: bool, config_file: str | None, log_format: str | None):
     """Run as a long-lived daemon with scheduled sync.
 
     Reads .oikb.yaml and syncs each source on its configured interval.
     Exposes /health, /history, and /sync endpoints.
     """
+    log_format = log_format or os.environ.get("LOG_FORMAT", "text")
+
     if config_file:
         import yaml
         with open(config_file) as f:
@@ -785,7 +789,7 @@ def daemon(port: int, no_server: bool, config_file: str | None):
             sys.exit(1)
 
     from oikb.daemon import start_daemon
-    start_daemon(entries=entries, port=port, no_server=no_server)
+    start_daemon(entries=entries, port=port, no_server=no_server, log_format=log_format)
 
 
 # ── history ─────────────────────────────────────────────────────

src/oikb/daemon.py

@@ -40,7 +40,7 @@ async def verify_api_key(
 app = FastAPI(
     title="oikb",
     description="Sync engine for Open WebUI Knowledge Bases. Trigger syncs, check status, and query history.",
-    version="0.2.7",
+    version="0.2.8",
 )
 
 # Runtime state populated by start_daemon().
@@ -305,9 +305,14 @@ def start_daemon(
     entries: list[dict],
     port: int = 8080,
     no_server: bool = False,
+    log_format: str = "text",
 ) -> None:
     """Start the daemon with scheduler + optional HTTP server."""
     global _history, _entries
+
+    from oikb.logging import configure_logging
+    configure_logging(log_format=log_format)
+
     _entries = entries
     _history = SyncHistory()
     set_build_info(__version__)

src/oikb/logging.py

@@ -0,0 +1,64 @@
+"""Structured JSON log formatter for production deployments."""
+
+from __future__ import annotations
+
+import json
+import logging
+import time
+from typing import Any
+
+
+class JsonFormatter(logging.Formatter):
+    """Emit log records as single-line JSON objects.
+
+    Output format (one JSON object per line):
+        {"ts":"2025-05-21T06:00:01Z","level":"INFO","logger":"oikb.daemon","msg":"Synced ..."}
+
+    Compatible with Datadog, Splunk, ELK, CloudWatch, Loki, and any
+    JSON-aware log aggregator.
+    """
+
+    def format(self, record: logging.LogRecord) -> str:
+        entry: dict[str, Any] = {
+            "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(record.created)),
+            "level": record.levelname,
+            "logger": record.name,
+            "msg": record.getMessage(),
+        }
+
+        if record.exc_info and record.exc_info[1] is not None:
+            entry["exception"] = self.formatException(record.exc_info)
+
+        # Merge any extra fields passed via `log.info("msg", extra={...})`.
+        for key in ("source", "kb_id", "status", "duration_ms",
+                     "files_added", "files_modified", "files_deleted"):
+            val = getattr(record, key, None)
+            if val is not None:
+                entry[key] = val
+
+        return json.dumps(entry, default=str, ensure_ascii=False)
+
+
+def configure_logging(log_format: str = "text", log_level: str = "INFO") -> None:
+    """Configure root logging for the daemon.
+
+    Args:
+        log_format: "text" for human-readable, "json" for structured JSON lines.
+        log_level:  Standard Python log level name.
+    """
+    root = logging.getLogger()
+    root.setLevel(getattr(logging, log_level.upper(), logging.INFO))
+
+    # Remove any existing handlers (uvicorn may add its own).
+    root.handlers.clear()
+
+    handler = logging.StreamHandler()
+
+    if log_format == "json":
+        handler.setFormatter(JsonFormatter())
+    else:
+        handler.setFormatter(
+            logging.Formatter("%(asctime)s %(levelname)-8s %(name)s  %(message)s")
+        )
+
+    root.addHandler(handler)