anti
035499f255
- Modify Rfc5424Formatter to read decnet_component from LogRecord
and use it as RFC 5424 APP-NAME field (falls back to 'decnet')
- Add get_logger(component) factory in decnet/logging/__init__.py
with _ComponentFilter that injects decnet_component on each record
- Wire all five layers to their component tag:
cli -> 'cli', engine -> 'engine', api -> 'api' (api.py, ingester,
routers), mutator -> 'mutator', collector -> 'collector'
- Add structured INFO/DEBUG/WARNING/ERROR log calls throughout each
layer per the defined vocabulary; DEBUG calls are suppressed unless
DECNET_DEVELOPER=true
- Add tests/test_logging.py covering factory, filter, formatter
component-awareness, fallback behaviour, and level gating
107 lines
3.5 KiB
Python
107 lines
3.5 KiB
Python
"""
|
|
Pydantic models for DECNET configuration and runtime state.
|
|
State is persisted to decnet-state.json in the working directory.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import socket as _socket
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from decnet.models import DeckyConfig, DecnetConfig # noqa: F401
|
|
|
|
from decnet.distros import random_hostname as _random_hostname
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# RFC 5424 syslog formatter
|
|
# ---------------------------------------------------------------------------
|
|
# Severity mapping: Python level → syslog severity (RFC 5424 §6.2.1)
|
|
_SYSLOG_SEVERITY: dict[int, int] = {
|
|
logging.CRITICAL: 2, # Critical
|
|
logging.ERROR: 3, # Error
|
|
logging.WARNING: 4, # Warning
|
|
logging.INFO: 6, # Informational
|
|
logging.DEBUG: 7, # Debug
|
|
}
|
|
_FACILITY_LOCAL0 = 16 # local0 (RFC 5424 §6.2.1 / POSIX)
|
|
|
|
|
|
class Rfc5424Formatter(logging.Formatter):
|
|
"""Formats log records as RFC 5424 syslog messages.
|
|
|
|
Output:
|
|
<PRIVAL>1 TIMESTAMP HOSTNAME APP-NAME PROCID MSGID STRUCTURED-DATA MSG
|
|
|
|
Example:
|
|
<134>1 2026-04-12T21:48:03.123456+00:00 host decnet 1234 decnet.config - Dev mode active
|
|
"""
|
|
|
|
_hostname: str = _socket.gethostname()
|
|
_app: str = "decnet"
|
|
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
severity = _SYSLOG_SEVERITY.get(record.levelno, 6)
|
|
prival = (_FACILITY_LOCAL0 * 8) + severity
|
|
ts = datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat(timespec="microseconds")
|
|
msg = record.getMessage()
|
|
if record.exc_info:
|
|
msg += "\n" + self.formatException(record.exc_info)
|
|
app = getattr(record, "decnet_component", self._app)
|
|
return (
|
|
f"<{prival}>1 {ts} {self._hostname} {app}"
|
|
f" {os.getpid()} {record.name} - {msg}"
|
|
)
|
|
|
|
|
|
def _configure_logging(dev: bool) -> None:
|
|
"""Install the RFC 5424 handler on the root logger (idempotent)."""
|
|
root = logging.getLogger()
|
|
# Avoid adding duplicate handlers on re-import (e.g. during testing)
|
|
if any(isinstance(h, logging.StreamHandler) and isinstance(h.formatter, Rfc5424Formatter)
|
|
for h in root.handlers):
|
|
return
|
|
handler = logging.StreamHandler()
|
|
handler.setFormatter(Rfc5424Formatter())
|
|
root.setLevel(logging.DEBUG if dev else logging.INFO)
|
|
root.addHandler(handler)
|
|
|
|
|
|
_dev = os.environ.get("DECNET_DEVELOPER", "").lower() == "true"
|
|
_configure_logging(_dev)
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
if _dev:
|
|
log.debug("Developer mode: debug logging active")
|
|
|
|
# Calculate absolute path to the project root (where the config file resides)
|
|
_ROOT: Path = Path(__file__).parent.parent.absolute()
|
|
STATE_FILE: Path = _ROOT / "decnet-state.json"
|
|
DEFAULT_MUTATE_INTERVAL: int = 30 # default rotation interval in minutes
|
|
|
|
|
|
def random_hostname(distro_slug: str = "debian") -> str:
|
|
return _random_hostname(distro_slug)
|
|
|
|
|
|
def save_state(config: DecnetConfig, compose_path: Path) -> None:
|
|
payload = {
|
|
"config": config.model_dump(),
|
|
"compose_path": str(compose_path),
|
|
}
|
|
STATE_FILE.write_text(json.dumps(payload, indent=2))
|
|
|
|
|
|
def load_state() -> tuple[DecnetConfig, Path] | None:
|
|
if not STATE_FILE.exists():
|
|
return None
|
|
data = json.loads(STATE_FILE.read_text())
|
|
return DecnetConfig(**data["config"]), Path(data["compose_path"])
|
|
|
|
|
|
def clear_state() -> None:
|
|
if STATE_FILE.exists():
|
|
STATE_FILE.unlink()
|