sdp_structured_log.py #1

#!/usr/bin/env python3
# ==============================================================================
# Copyright and license info is available in the LICENSE file included with
# the Server Deployment Package (SDP), and also available online:
# https://swarm.workshop.perforce.com/projects/perforce-software-sdp/view/main/LICENSE
# ------------------------------------------------------------------------------
#
# sdp_structured_log.py -- shared structured-logging helper for SDP Python
# scripts, both cron-invoked (p4verify.py, update_limits.py) and trigger
# scripts fired directly by p4d from the trigger table (triggers/*.py).
#
# stdlib only, deliberately: trigger scripts must run with minimal external
# requirements at trigger time (documented project constraint), so this
# module takes no dependency beyond the standard library -- no sdputils.py,
# no P4Python, nothing third-party.
#
# NEVER WRITE TO STDOUT. p4d relays a trigger's stdout to the connected
# client, so a stray print() here becomes a user-visible message on, e.g.,
# `p4 group -i`. Every public entry point below is wrapped in a broad
# try/except so a logging bug can never fail the caller's real work, and
# diagnostics (if SDP_LOG_DEBUG=1) go to stderr, never stdout.
#
# One mode at a time (SDP_LOG_FORMAT env var, default "jsonl"), matching the
# bash helper's semantics exactly: "jsonl" appends one JSON object per run
# alongside the caller's own log; "prom" writes Prometheus textfile-collector
# gauges to $SDP_METRICS_DIR instead; "off" disables structured output.
# is_trigger=True callers always skip prom mode -- a trigger firing
# concurrently from many p4d processes rewriting one gauge file is
# meaningless (last-writer-wins, no counters).
#
# Usage (cron script):
#   import sdp_structured_log as sdplog
#   with sdplog.run("p4verify.py", logfile=summary_log_path) as ev:
#       ev.add(depots_verified=12)
#       ...
#       if problems:
#           ev.set_status("warning")
#
# Usage (trigger -- optional import, must degrade silently if this module
# hasn't shipped yet, e.g. mid-rollout when the Ansible-templated trigger
# lands before the tarball that carries this file):
#   sys.path.insert(0, "/p4/common/bin")
#   try:
#       import sdp_structured_log as sdplog
#   except Exception:
#       sdplog = None
#   ev = sdplog.init("keep_group_unset.py", is_trigger=True) if sdplog else None
#   ...
#   if ev:
#       ev.add(user=user, action="rewrote")
#       ev.emit(0)
# ------------------------------------------------------------------------------

import atexit
import datetime
import json
import os
import sys
import time

_PIPE_BUF_SAFE_LINE = 4000
_MAX_FIELD_LEN = 256
_MAX_ERROR_LEN = 900
_MAX_LABEL_LEN = 128


def _debug(msg):
    if os.environ.get("SDP_LOG_DEBUG") == "1":
        try:
            sys.stderr.write("sdp_structured_log: %s\n" % msg)
        except Exception:
            pass


def _truncate(s, n):
    s = str(s)
    return s if len(s) <= n else s[:n]


class SDPEvent:
    """One in-flight terminal record. emit() is idempotent -- called at
    most once regardless of how many of atexit/excepthook/run() reach it."""

    def __init__(self, script, logfile=None, instance=None, mode=None, is_trigger=False):
        self.script = script
        self.logfile = logfile
        self.instance = instance if instance is not None else os.environ.get("SDP_INSTANCE", "")
        self.mode = mode if mode is not None else os.environ.get("SDP_LOG_FORMAT", "jsonl")
        self.is_trigger = is_trigger
        self.status = "success"
        self.error = None
        self.fields = {}
        self.start = time.time()
        self._emitted = False

    def add(self, **fields):
        try:
            for k, v in fields.items():
                self.fields[k] = v
        except Exception as exc:  # noqa: BLE001 - logging must never raise
            _debug("add() failed: %s" % exc)

    def set_status(self, status):
        # Monotonic downgrade only, same rule as the bash helper: a later
        # success can never mask an earlier warning/failure.
        order = {"success": 0, "warning": 1, "failure": 2}
        try:
            if order.get(status, 0) > order.get(self.status, 0):
                self.status = status
        except Exception as exc:  # noqa: BLE001
            _debug("set_status() failed: %s" % exc)

    def set_error(self, message):
        try:
            self.error = _truncate(message, _MAX_ERROR_LEN)
            if self.status == "success":
                self.status = "failure"
        except Exception as exc:  # noqa: BLE001
            _debug("set_error() failed: %s" % exc)

    def emit(self, exit_code=None):
        if self._emitted:
            return
        self._emitted = True
        try:
            duration = time.time() - self.start
            rc = 0 if exit_code is None else exit_code
            if self.status == "success" and rc not in (0, None):
                self.status = "failure"
            if self.mode == "off":
                return
            if self.mode == "prom":
                if not self.is_trigger:
                    _write_prom(self, rc, duration)
                return
            _write_jsonl(self, rc, duration)
        except Exception as exc:  # noqa: BLE001 - logging must never raise
            _debug("emit() failed: %s" % exc)


def _structured_file(event):
    if event.logfile:
        base, ext = os.path.splitext(event.logfile)
        if ext == ".log":
            return base + ".jsonl"
        return event.logfile + ".jsonl"
    logs_dir = os.environ.get("LOGS", "/tmp")
    name = event.script
    if name.endswith(".py") or name.endswith(".sh"):
        name = name.rsplit(".", 1)[0]
    return os.path.join(logs_dir, name + ".jsonl")


def _write_jsonl(event, rc, duration):
    path = _structured_file(event)
    directory = os.path.dirname(path) or "."
    if not os.path.isdir(directory) or not os.access(directory, os.W_OK):
        return

    record = {
        "timestamp": datetime.datetime.now().astimezone().isoformat(timespec="seconds"),
        "script": event.script,
        "instance": str(event.instance),
        "status": event.status,
        "exit_code": rc,
        "duration_seconds": round(duration, 3),
        "error": event.error,
    }
    for k, v in event.fields.items():
        if isinstance(v, (int, float)):
            record[k] = v
        else:
            record[k] = _truncate(v, _MAX_FIELD_LEN)

    line = json.dumps(record, separators=(",", ":"))
    if len(line) > _PIPE_BUF_SAFE_LINE:
        # Same PIPE_BUF safety rule as the bash helper: a single line over
        # ~4KB is not guaranteed to append atomically. Drop fields (not the
        # fixed prefix) until it fits, and say so.
        minimal = dict(record)
        minimal["truncated"] = True
        for k in list(event.fields.keys()):
            minimal.pop(k, None)
            line = json.dumps(minimal, separators=(",", ":"))
            if len(line) <= _PIPE_BUF_SAFE_LINE:
                break

    with open(path, "a") as f:  # noqa: PTH123 - append, O_APPEND is the point
        f.write(line + "\n")


def _metric_base(script):
    name = "p4_sdp_" + os.path.splitext(script)[0]
    return "".join(c if (c.isalnum() or c == "_") else "_" for c in name)


def _prom_label_escape(s):
    s = _truncate(s, _MAX_LABEL_LEN)
    return s.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n")


def _write_prom(event, rc, duration):
    metrics_dir = os.environ.get("SDP_METRICS_DIR", "/p4/metrics")
    if not os.path.isdir(metrics_dir) or not os.access(metrics_dir, os.W_OK):
        return

    base = _metric_base(event.script)
    instance = _prom_label_escape(event.instance)
    final_path = os.path.join(metrics_dir, base + ".prom")
    tmp_path = "%s.%d" % (final_path, os.getpid())
    success = 1 if event.status == "success" else 0

    lines = [
        "# HELP %s_success Whether the last run of this script succeeded (1) or not (0)." % base,
        "# TYPE %s_success gauge" % base,
        '%s_success{instance="%s"} %d' % (base, instance, success),
        "# HELP %s_exit_code Exit code of the last run." % base,
        "# TYPE %s_exit_code gauge" % base,
        '%s_exit_code{instance="%s"} %s' % (base, instance, rc),
        "# HELP %s_duration_seconds Duration of the last run, in seconds." % base,
        "# TYPE %s_duration_seconds gauge" % base,
        '%s_duration_seconds{instance="%s"} %.3f' % (base, instance, duration),
        "# HELP %s_last_run_timestamp_seconds Unix time the last run finished." % base,
        "# TYPE %s_last_run_timestamp_seconds gauge" % base,
        '%s_last_run_timestamp_seconds{instance="%s"} %d' % (base, instance, time.time()),
    ]
    if event.status != "success":
        lines += [
            "# HELP %s_status Non-success status of the last run (info-style; 1 when present)." % base,
            "# TYPE %s_status gauge" % base,
            '%s_status{instance="%s",status="%s",error="%s"} 1'
            % (base, instance, _prom_label_escape(event.status), _prom_label_escape(event.error or "")),
        ]
    for k, v in event.fields.items():
        if isinstance(v, (int, float)):
            key = "".join(c if (c.isalnum() or c == "_") else "_" for c in str(k))
            lines.append('%s_%s{instance="%s"} %s' % (base, key, instance, v))

    try:
        with open(tmp_path, "w") as f:  # noqa: PTH123
            f.write("\n".join(lines) + "\n")
        os.replace(tmp_path, final_path)
    finally:
        try:
            os.unlink(tmp_path)
        except OSError:
            pass  # already replaced, or never created


def init(script=None, logfile=None, instance=None, mode=None,
         install_signal_handlers=False, is_trigger=False):
    """One-shot setup: creates the event, wires atexit + excepthook, returns
    the event so the caller can add()/set_status()/set_error() during the
    run. emit() fires automatically at process exit; call it directly first
    if an explicit exit code is known (SystemExit doesn't always carry one
    cleanly through excepthook)."""
    name = script or os.path.basename(sys.argv[0])
    event = SDPEvent(name, logfile=logfile, instance=instance, mode=mode, is_trigger=is_trigger)

    try:
        atexit.register(event.emit)
    except Exception as exc:  # noqa: BLE001
        _debug("atexit.register failed: %s" % exc)

    previous_hook = sys.excepthook

    def _hook(exc_type, exc_value, exc_tb):
        try:
            if exc_type is not None and exc_type is not SystemExit:
                event.set_error("%s: %s" % (exc_type.__name__, exc_value))
                event.set_status("failure")
        except Exception:  # noqa: BLE001
            pass
        return previous_hook(exc_type, exc_value, exc_tb)

    try:
        sys.excepthook = _hook
    except Exception as exc:  # noqa: BLE001
        _debug("excepthook install failed: %s" % exc)

    if install_signal_handlers:
        try:
            import signal

            def _signal_handler(signum, _frame):
                event.set_status("failure")
                event.set_error("received signal %d" % signum)
                event.emit(128 + signum)
                signal.signal(signum, signal.SIG_DFL)
                os.kill(os.getpid(), signum)

            for sig in (getattr(signal, "SIGTERM", None), getattr(signal, "SIGINT", None)):
                if sig is not None:
                    signal.signal(sig, _signal_handler)
        except Exception as exc:  # noqa: BLE001
            _debug("signal handler install failed: %s" % exc)

    return event


def log_event(script, status, exit_code, duration_seconds, error=None, **fields):
    """One-shot emit, matching the bash helper's sdp_log_event signature --
    for a caller that manages its own error handling and just wants to
    write one record without the init()/atexit machinery."""
    try:
        event = SDPEvent(script)
        event.status = status
        event.error = _truncate(error, _MAX_ERROR_LEN) if error else None
        event.fields = dict(fields)
        event.start = time.time() - float(duration_seconds or 0)
        event.emit(exit_code)
    except Exception as exc:  # noqa: BLE001
        _debug("log_event() failed: %s" % exc)


class run:
    """Context manager form: with sdp_structured_log.run("script.py") as ev:
    ... . Records a failure status + formatted exception on any exception
    other than SystemExit (whose .code becomes the exit_code), then
    re-raises / lets SystemExit propagate unchanged either way."""

    def __init__(self, script, logfile=None, **kw):
        self._event = init(script, logfile=logfile, **kw)

    def __enter__(self):
        return self._event

    def __exit__(self, exc_type, exc_value, _exc_tb):
        try:
            if exc_type is None:
                self._event.emit(0)
            elif exc_type is SystemExit:
                code = exc_value.code if isinstance(exc_value.code, int) else (1 if exc_value.code else 0)
                self._event.emit(code)
            else:
                self._event.set_status("failure")
                self._event.set_error("%s: %s" % (exc_type.__name__, exc_value))
                self._event.emit(1)
        except Exception:  # noqa: BLE001
            pass
        return False  # never swallow the caller's exception