#!/usr/bin/python3
"""aep-logs -- render the log forwarding of an AEP node (rsyslog -> GELF).

Usage:
  aep-logs render            print the rsyslog snippet to stdout, change nothing
  aep-logs apply [--quiet]   render, install /etc/rsyslog.d/50-aep-logs.conf when
                             it changed (checked with rsyslogd -N1, rolled back on
                             error), restart rsyslog on a config change, HUP it on
                             a lifecycle change
  aep-logs watch             apply every AEP_LOGS_INTERVAL seconds (longrun)

Inputs (ticket #98, interface logs.json v1 agreed with ADL #99):
  /etc/aep/logs.d/*.json      declarations "aep.logs/v1", written by artie in
                              setup_node (one file per service)
  /var/lib/aep/podinfo/       downward API of the pod: labels, annotations,
                              namespace -> fixed fields instance, service,
                              node_uid, cust, project
  /var/lib/aep/status/lifecycle.json
                              last lifecycle state (aep.status/v1) -> field
                              lifecycle, via an rsyslog lookup table
  /etc/aep/logging.json       optional: the logging endpoint of the contract
                              ({"fqdn","port","protocol"}); without it the
                              values of /etc/default/aep-logs apply

Without any declaration the node still forwards every s6 logger file
(/var/log/*/current) and syslog with sensitivity internal and retention class
default (platform clause P1). An entry with sensitivity "secret" is never
forwarded, and its s6 unit is taken out of the default forwarding as well.

Exit codes: 0 ok, 1 error (an invalid declaration is skipped, the rest is
applied, and the exit code is 1 so that the caller sees it).
"""

import fcntl
import glob
import hashlib
import json
import os
import re
import socket
import subprocess
import sys
import tempfile
import time

SCHEMA = "aep.logs/v1"
SOURCE_TYPES = ("file", "s6", "journal")
FORMATS = ("json", "syslog", "plain")
SENSITIVITY = ("public", "internal", "secret")
RETENTION = ("short", "default", "long", "audit")
SEV_NAMES = {"emerg": 0, "alert": 1, "crit": 2, "err": 3, "warning": 4, "notice": 5, "info": 6, "debug": 7}

DEFAULTS = {
    "AEP_LOGS_DEFAULTS_FILE": "/etc/default/aep-logs",
    "AEP_LOGS_DIR": "/etc/aep/logs.d",
    "AEP_LOGS_CONF": "/etc/rsyslog.d/50-aep-logs.conf",
    "AEP_LOGS_STATE": "/var/lib/aep-logs",
    "AEP_LOGS_PODINFO": "/var/lib/aep/podinfo",
    "AEP_LOGS_LIFECYCLE": "/var/lib/aep/status/lifecycle.json",
    "AEP_LOGS_ENDPOINT": "/etc/aep/logging.json",
    "AEP_LOGS_TARGET": "graylog.scale42.com",
    "AEP_LOGS_PORT": "12201",
    "AEP_LOGS_PROTOCOL": "gelf-tcp-tls",
    "AEP_LOGS_CA_FILE": "/etc/ssl/certs/ca-certificates.crt",
    "AEP_LOGS_PERMITTED_PEER": "",
    "AEP_LOGS_S6_ROOT": "/var/log",
    "AEP_LOGS_FORWARD_SYSLOG": "yes",
    "AEP_LOGS_S6_FRESHSTART": "off",
    "AEP_LOGS_QUEUE_MAX_DISK": "512m",
    "AEP_LOGS_INSTANCE": "",
    "AEP_LOGS_INTERVAL": "30",
    "AEP_LOGS_RSYSLOG_SERVICEDIR": "/run/service/rsyslog",
    "AEP_LOGS_RSYSLOG_CHECK": "rsyslogd -N1",
    "AEP_LOGS_RESTART_CMD": "",
    "AEP_LOGS_HUP_CMD": "",
}


def log(msg):
    print("aep-logs: " + msg, flush=True)


# --------------------------------------------------------------------- config
def load_settings():
    """Defaults, then /etc/default/aep-logs (KEY=VALUE), then the environment."""
    cfg = dict(DEFAULTS)
    path = os.environ.get("AEP_LOGS_DEFAULTS_FILE", cfg["AEP_LOGS_DEFAULTS_FILE"])
    try:
        with open(path) as f:
            for line in f:
                line = line.strip()
                if not line or line.startswith("#") or "=" not in line:
                    continue
                k, v = line.split("=", 1)
                k = k.strip()
                if k.startswith("export "):
                    k = k[len("export "):].strip()
                if k in cfg:
                    cfg[k] = v.strip().strip('"').strip("'")
    except FileNotFoundError:
        pass
    for k in cfg:
        if k in os.environ:
            cfg[k] = os.environ[k]
    return cfg


def endpoint(cfg):
    """Target of the GELF output: /etc/aep/logging.json wins over the defaults."""
    ep = {"fqdn": cfg["AEP_LOGS_TARGET"], "port": cfg["AEP_LOGS_PORT"], "protocol": cfg["AEP_LOGS_PROTOCOL"]}
    try:
        with open(cfg["AEP_LOGS_ENDPOINT"]) as f:
            doc = json.load(f)
        for k in ("fqdn", "port", "protocol"):
            if doc.get(k) not in (None, ""):
                ep[k] = str(doc[k])
        ep["from"] = cfg["AEP_LOGS_ENDPOINT"]
    except FileNotFoundError:
        ep["from"] = "defaults"
    return ep


# ------------------------------------------------------------------- identity
_KV = re.compile(r'^([^=]+)="(.*)"$')


def read_kv(path):
    """Downward API file: key="value" per line, value escaped like a Go string."""
    out = {}
    try:
        with open(path) as f:
            for line in f:
                m = _KV.match(line.rstrip("\n"))
                if m:
                    try:
                        out[m.group(1)] = json.loads('"' + m.group(2) + '"')
                    except ValueError:
                        out[m.group(1)] = m.group(2)
    except (FileNotFoundError, PermissionError):
        pass
    return out


def read_text(path):
    try:
        with open(path) as f:
            return f.read().strip()
    except (FileNotFoundError, PermissionError):
        return ""


def identity(cfg):
    """Fixed fields of every message. None of them can be set by a service."""
    pi = cfg["AEP_LOGS_PODINFO"]
    labels = read_kv(os.path.join(pi, "labels"))
    ann = read_kv(os.path.join(pi, "annotations"))
    namespace = read_text(os.path.join(pi, "namespace"))
    fqdn = labels.get("dns.service.aep.asag.io/fqdn") or socket.getfqdn()
    # instance = the Trick instance (tick1), not the service instance: the
    # Graylog stream rule of the instance matches on it (stream tick1, rule
    # instance == tick1). svc-ctrl passes it to the pod as the service flag
    # v1_customer_env; the domain of the FQDN is the fallback.
    instance = cfg["AEP_LOGS_INSTANCE"] or ann.get("serviceflag.aep.asag.io/v1_customer_env", "")
    if not instance and fqdn.count(".") >= 2:
        instance = fqdn.split(".")[1]
    # project = namespace <cust>-<project> (N35); cust = part before the first
    # hyphen (the rule of svc-ctrl getDefinitionEnv).
    return {
        "fqdn": fqdn,
        "instance": instance,
        "service": labels.get("service.aep.asag.io/definitionname") or labels.get("serviceflag.aep.asag.io/v1_role", ""),
        "node_uid": labels.get("service.aep.asag.io/node", ""),
        "provision": labels.get("service.aep.asag.io/provisionname", ""),
        "project": namespace,
        "cust": namespace.split("-", 1)[0] if namespace else "",
        "role": labels.get("serviceflag.aep.asag.io/v1_role", ""),
        "dc": labels.get("serviceflag.aep.asag.io/v1_datacenter", ""),
    }


def lifecycle_state(cfg):
    try:
        with open(cfg["AEP_LOGS_LIFECYCLE"]) as f:
            st = json.load(f).get("state")
        return st if isinstance(st, str) and st else "unknown"
    except (FileNotFoundError, PermissionError, ValueError, AttributeError):
        return "unknown"


# ---------------------------------------------------------------- declarations
def check_entry(e, where):
    errs = []
    if not isinstance(e, dict):
        return None, [where + ": entry is not an object"]
    name = e.get("name")
    if not isinstance(name, str) or not re.match(r"^[a-z0-9_]+$", name or ""):
        errs.append(where + ": name must match [a-z0-9_]")
    src = e.get("source") or {}
    st = src.get("type")
    if st not in SOURCE_TYPES:
        errs.append("%s.%s: source.type %r, expected file, s6 or journal" % (where, name, st))
    elif st == "file":
        p = src.get("path", "")
        if not isinstance(p, str) or not p.startswith("/") or p.endswith("/") or re.search(r"[\s\"'\\{}]", p):
            errs.append("%s.%s: source.path %r must be an absolute file path (glob * ? allowed)" % (where, name, p))
    else:
        u = src.get("unit", "")
        if not isinstance(u, str) or not re.match(r"^[A-Za-z0-9][A-Za-z0-9._@:-]*$", u or ""):
            errs.append("%s.%s: source.unit %r invalid" % (where, name, u))
    fmt = e.get("format", "plain")
    if fmt not in FORMATS:
        errs.append("%s.%s: format %r, expected json, syslog or plain" % (where, name, fmt))
    sens = e.get("sensitivity", "internal")
    if sens not in SENSITIVITY:
        errs.append("%s.%s: sensitivity %r, expected public, internal or secret" % (where, name, sens))
    ret = e.get("retention_class", "default")
    if ret not in RETENTION:
        errs.append("%s.%s: retention_class %r, expected short, default, long or audit" % (where, name, ret))
    sevmap = {}
    for k, v in (e.get("severity_map") or {}).items():
        if isinstance(v, str):
            v = SEV_NAMES.get(v)
        if not isinstance(v, int) or isinstance(v, bool) or not 0 <= v <= 7:
            errs.append("%s.%s: severity_map.%s must be 0-7 or emerg..debug" % (where, name, k))
            continue
        sevmap[str(k)] = v
    ml = e.get("multiline") or None
    start = ml.get("start") if isinstance(ml, dict) else None
    if ml is not None and not (isinstance(start, str) and start):
        errs.append("%s.%s: multiline.start must be a non-empty regular expression" % (where, name))
    # tags: a list of strings (artie 0.11.0) or an object (first draft of the
    # interface); both end up as one comma separated field.
    tags = e.get("tags") or []
    if isinstance(tags, dict):
        tags = ["%s=%s" % (k, tags[k]) for k in sorted(tags)]
    elif not isinstance(tags, list):
        tags = []
    if errs:
        return None, errs
    return {
        "name": name, "type": st, "path": src.get("path", ""), "unit": src.get("unit", ""),
        "format": fmt, "sensitivity": sens, "retention": ret, "severity_map": sevmap,
        "multiline": start, "tags": ",".join(str(t) for t in tags),
    }, []


def load_declarations(cfg):
    """All declaration files, sorted. Returns (entries, errors, files)."""
    entries, errors, files = [], [], []
    for path in sorted(glob.glob(os.path.join(cfg["AEP_LOGS_DIR"], "*.json"))):
        try:
            with open(path) as f:
                doc = json.load(f)
        except (OSError, ValueError) as ex:
            errors.append("%s: unreadable (%s), skipped" % (path, ex))
            continue
        if not isinstance(doc, dict) or doc.get("schema") != SCHEMA:
            errors.append("%s: schema is not %s, skipped" % (path, SCHEMA))
            continue
        svc = doc.get("service") or os.path.basename(path)[:-5]
        good, bad = [], []
        for i, e in enumerate(doc.get("entries") or []):
            ent, errs = check_entry(e, "%s entries[%d]" % (os.path.basename(path), i))
            if errs:
                bad.extend(errs)
            else:
                ent["service"] = svc
                good.append(ent)
        if bad:
            errors.extend("%s, file skipped" % m for m in bad)
            continue
        files.append(path)
        entries.extend(good)
    return entries, errors, files


# ------------------------------------------------------------------- rendering
def rs(s):
    """rsyslog string literal content (backslash and double quote escaped)."""
    return str(s).replace("\\", "\\\\").replace('"', '\\"')


def js(s):
    """JSON string content, then escaped for an rsyslog string literal."""
    return rs(json.dumps(str(s))[1:-1])


def rx(s):
    """rsyslog single-quoted regex literal content."""
    return str(s).replace("\\", "\\\\").replace("'", "\\'")


def glob_re(pattern):
    """fnmatch-like glob (* ? within one path segment) to an anchored ERE."""
    out = []
    for ch in pattern:
        if ch == "*":
            out.append("[^/]*")
        elif ch == "?":
            out.append("[^/]")
        elif ch in ".^$+()[]{}|\\":
            out.append("\\" + ch)
        else:
            out.append(ch)
    return "^" + "".join(out) + "$"


def readable_by_rsyslog(path):
    """Whether user syslog (groups syslog, adm) may read path, by mode bits."""
    import grp
    import pwd
    try:
        st = os.stat(path)
        uid = pwd.getpwnam("syslog").pw_uid
        gids = {grp.getgrnam(g).gr_gid for g in ("syslog", "adm")}
    except (OSError, KeyError):
        return True  # cannot tell (no such user here): do not warn
    if st.st_uid == uid:
        return bool(st.st_mode & 0o400)
    if st.st_gid in gids:
        return bool(st.st_mode & 0o040)
    return bool(st.st_mode & 0o004)


def ident_name(prefix, s):
    return prefix + re.sub(r"[^A-Za-z0-9_]", "_", s)


def sev_block(entry, indent="  "):
    lines = ['%sset $.sev = "6";' % indent]
    first = True
    # most severe first: a line that contains "error" and "info" is an error
    for key, lvl in sorted(entry["severity_map"].items(), key=lambda kv: (kv[1], kv[0])):
        lines.append('%s%s ($msg contains "%s") then { set $.sev = "%d"; }' % (indent, "if" if first else "else if", rs(key), lvl))
        first = False
    return lines


def source_ruleset(entry):
    rsn = ident_name("aep_logs_src_", entry["service"] + "__" + entry["name"])
    out = ['ruleset(name="%s") {' % rsn]
    out += ['  set $.log_name = "%s";' % rs(entry["name"]),
            '  set $.svc = "%s";' % rs(entry["service"]),
            '  set $.sens = "%s";' % rs(entry["sensitivity"]),
            '  set $.ret = "%s";' % rs(entry["retention"]),
            '  set $.tags = "%s";' % rs(entry["tags"]),
            '  set $.app = "%s";' % rs(entry["unit"] or entry["name"])]
    out += sev_block(entry)
    out += ["  call aep_logs_out", "}"]
    return rsn, out


def render(cfg):
    """Returns (text, summary dict, errors, warnings)."""
    entries, errors, files = load_declarations(cfg)
    warnings = []
    ident = identity(cfg)
    ep = endpoint(cfg)
    root = cfg["AEP_LOGS_S6_ROOT"].rstrip("/") or "/var/log"
    unit_field = len(root.split("/")) + 1  # /var/log/<unit>/current -> field 4
    state = cfg["AEP_LOGS_STATE"]
    proto = ep["protocol"]
    if proto not in ("gelf-tcp-tls", "gelf-tcp", "gelf-udp"):
        errors.append("endpoint protocol %r unsupported (gelf-tcp-tls, gelf-tcp, gelf-udp); using gelf-tcp-tls" % proto)
        proto = "gelf-tcp-tls"
    tls = proto == "gelf-tcp-tls"
    transport = "udp" if proto == "gelf-udp" else "tcp"

    secret, s6_entries, merged_files, own_files, journal = [], [], [], [], []
    for e in entries:
        if e["sensitivity"] == "secret":
            secret.append(e)
            continue
        if e["type"] == "s6":
            s6_entries.append(e)
        elif e["type"] == "file":
            d = os.path.dirname(e["path"])
            # a file directly in <root>/<dir>/ is in a directory the wildcard
            # input watches already: a second input there would be a second
            # inotify watch on it ("wd N already in wdmap", lost rotations,
            # tick-r10k profiles::logging) -- it is read by the wildcard input.
            if os.path.dirname(d) == root and not re.search(r"[*?]", d):
                merged_files.append(e)
            else:
                own_files.append(e)
        else:
            journal.append(e)
    have_journal = os.path.exists("/run/systemd/journal/socket")
    # rsyslog drops its privileges to user syslog (groups syslog, adm) on
    # Ubuntu: a declared file it cannot read is silently not forwarded.
    for e in own_files + merged_files:
        for path in glob.glob(e["path"]) or []:
            if not readable_by_rsyslog(path):
                warnings.append("%s/%s: %s is not readable by rsyslog (user syslog, groups syslog/adm); not forwarded until its mode or group allows it" % (e["service"], e["name"], path))

    L = []
    L.append("# Generated by aep-logs (package aep-logs-s6) -- do not edit, it is")
    L.append("# overwritten. Inputs: %s/*.json, %s, %s." % (cfg["AEP_LOGS_DIR"], cfg["AEP_LOGS_PODINFO"], cfg["AEP_LOGS_LIFECYCLE"]))
    L.append("# Declarations: %s" % (", ".join(os.path.basename(f) for f in files) or "none (default forwarding P1 only)"))
    L.append("")
    if tls:
        L.append("# TLS driver (package rsyslog-gnutls)")
        L.append('global(DefaultNetstreamDriver="gtls" DefaultNetstreamDriverCAFile="%s")' % rs(cfg["AEP_LOGS_CA_FILE"]))
        L.append("")
    L.append("# Lifecycle of the node (aep.status/v1 lifecycle.json), reloaded on HUP:")
    L.append("# aep-logs rewrites the table and sends HUP when the state changes.")
    L.append('lookup_table(name="aep_logs_lifecycle" file="%s/lifecycle.lookup.json" reloadOnHUP="on")' % rs(state))
    L.append("")
    # GELF 1.1; fields as in tick-r10k aep-gelf.conf.erb (host, level from
    # $.sev, _tag, _app, _instance, _role, _dc, _fqdn) plus the fields of the
    # observability contract (#98).
    f = ident
    L.append('template(name="aep_logs_gelf" type="list") {')
    L.append('  constant(value="{\\"version\\":\\"1.1\\",\\"host\\":\\"%s\\"")' % js(f["fqdn"]))
    L.append('  constant(value=",\\"short_message\\":\\"")   property(name="msg" format="json")')
    L.append('  constant(value="\\",\\"timestamp\\":")       property(name="timegenerated" dateformat="unixtimestamp")')
    L.append('  constant(value=".")                       property(name="timegenerated" dateformat="subseconds")')
    L.append('  constant(value=",\\"level\\":")             property(name="$.sev")')
    L.append('  constant(value=",\\"_tag\\":\\"")            property(name="syslogtag" format="json")')
    L.append('  constant(value="\\",\\"_app\\":\\"")          property(name="$.app" format="json")')
    L.append('  constant(value="\\",\\"_service\\":\\"")      property(name="$.svc" format="json")')
    L.append('  constant(value="\\",\\"_log_name\\":\\"")     property(name="$.log_name" format="json")')
    L.append('  constant(value="\\",\\"_sensitivity\\":\\"")  property(name="$.sens" format="json")')
    L.append('  constant(value="\\",\\"_retention_class\\":\\"") property(name="$.ret" format="json")')
    L.append('  constant(value="\\",\\"_log_tags\\":\\"")     property(name="$.tags" format="json")')
    L.append('  constant(value="\\",\\"_lifecycle\\":\\"")    property(name="$.lifecycle" format="json")')
    for k in ("instance", "node_uid", "cust", "project", "provision", "role", "dc", "fqdn"):
        L.append('  constant(value="\\",\\"_%s\\":\\"%s")' % (k, js(f[k])))
    L.append('  constant(value="\\"}")')
    L.append("}")
    L.append("")
    L.append('ruleset(name="aep_logs_out") {')
    L.append('  set $.lifecycle = lookup("aep_logs_lifecycle", "state");')
    L.append("  action(")
    L.append('    type="omfwd" target="%s" port="%s" protocol="%s"' % (rs(ep["fqdn"]), rs(ep["port"]), transport))
    L.append('    template="aep_logs_gelf"')
    if transport == "tcp":
        L.append('    TCP_FrameDelimiter="0" KeepAlive="on"')
    if tls:
        L.append('    StreamDriver="gtls" StreamDriverMode="1" StreamDriverAuthMode="x509/name"')
        L.append('    StreamDriverPermittedPeers="%s"' % rs(cfg["AEP_LOGS_PERMITTED_PEER"] or ep["fqdn"]))
    L.append('    queue.type="LinkedList" queue.filename="aep_logs_fwd"')
    L.append('    queue.maxDiskSpace="%s" queue.saveOnShutdown="on"' % rs(cfg["AEP_LOGS_QUEUE_MAX_DISK"]))
    L.append('    action.resumeRetryCount="-1"')
    L.append("  )")
    L.append("}")
    L.append("")

    # ---- declared sources: one ruleset each
    for e in s6_entries + merged_files + own_files:
        _, block = source_ruleset(e)
        L += block
    if s6_entries or merged_files or own_files:
        L.append("")

    # ---- inputs. mode="inotify": s6-log rotates by RENAMING current to
    # @<tai64n>.s and creates a new current; inotify follows the directory,
    # polling would stay on the old inode. reopenOnTruncate covers truncation.
    # State: imfile keeps one state file per inode in the rsyslog work
    # directory (written every persistStateInterval lines and on shutdown), so
    # a restart of rsyslog neither loses nor repeats lines -- unless s6-log
    # renames current away WHILE rsyslog is down: the rest of that file is
    # lost (measured: 93 of 200 lines with s4096 and 2 s downtime; none
    # without a rotation in the window).
    # freshStartTail: "off" for the s6 input (AEP_LOGS_S6_FRESHSTART): on
    # first activation the current of every unit is read from its start --
    # bounded by the s6-log size, and it carries the start messages of the
    # units. "on" for declared files: they can be large and unrotated.
    L.append('module(load="imfile" mode="inotify")')
    L.append("")
    mf_bases = sorted(set(os.path.basename(e["path"]) for e in merged_files))
    # Archives (@<tai64n>.s) are NOT in the pattern: measured 2026-09-30 with
    # tests/rotation-test.sh, reading them duplicated lines (the renamed
    # current is tracked by inode already) and let archives of a secret unit
    # through. imfile follows current across the rename by itself.
    names = ["current"] + mf_bases
    pattern = names[0] if len(names) == 1 else "{" + ",".join(names) + "}"
    L.append("# All s6 logger files %s/*/current (default forwarding P1) and the" % root)
    L.append("# declared files directly in %s/<dir>/: ONE input, one inotify watch." % root)
    L.append('ruleset(name="aep_logs_s6") {')
    L.append('  set $.unit = field($!metadata!filename, 47, %d);' % unit_field)
    for e in secret:
        if e["type"] == "s6":
            L.append('  # %s/%s: sensitivity secret -- never forwarded' % (e["service"], e["name"]))
            L.append('  if ($!metadata!filename == "%s/%s/current") then { stop }' % (rs(root), rs(e["unit"])))
    for e in s6_entries:
        rsn = ident_name("aep_logs_src_", e["service"] + "__" + e["name"])
        L.append('  if ($!metadata!filename == "%s/%s/current") then { call %s stop }' % (rs(root), rs(e["unit"]), rsn))
    for e in merged_files:
        rsn = ident_name("aep_logs_src_", e["service"] + "__" + e["name"])
        L.append("  if re_match($!metadata!filename, '%s') then { call %s stop }" % (rx(glob_re(e["path"])), rsn))
    L.append("  if not re_match($!metadata!filename, '/current$') then { stop }")
    L.append('  set $.log_name = $.unit;')
    L.append('  set $.app = $.unit;')
    L.append('  set $.svc = "%s";' % rs(ident["service"]))
    L.append('  set $.sens = "internal";')
    L.append('  set $.ret = "default";')
    L.append('  set $.tags = "";')
    L.append('  set $.sev = "6";')
    L.append("  # level position of known formats (as in tick-r10k aep-gelf.conf.erb)")
    L.append("  if re_match($msg, '(^| )level=(ERROR|error|FATAL|fatal)( |$)') or re_match($msg, '\\\\{\"level\":\"(error|fatal|panic|dpanic)\"') then { set $.sev = \"3\"; }")
    L.append("  else if re_match($msg, '(^| )level=(WARN|WARNING|warn|warning)( |$)') or re_match($msg, '\\\\{\"level\":\"warn\"') then { set $.sev = \"4\"; }")
    L.append("  call aep_logs_out")
    L.append("}")
    L.append('input(type="imfile" File="%s/*/%s" Tag="s6-unit:" Severity="info" Facility="local6"' % (rs(root), pattern))
    L.append('      ruleset="aep_logs_s6" addMetadata="on"')
    L.append('      reopenOnTruncate="on" freshStartTail="%s" persistStateInterval="100")' % cfg["AEP_LOGS_S6_FRESHSTART"])
    L.append("")

    # ---- declared files elsewhere: one input per directory
    groups = {}
    for e in own_files:
        groups.setdefault(os.path.dirname(e["path"]), []).append(e)
    for d in sorted(groups):
        es = sorted(groups[d], key=lambda x: x["path"])
        if len(es) == 1:
            e = es[0]
            rsn = ident_name("aep_logs_src_", e["service"] + "__" + e["name"])
            ml = ""
            if e["multiline"]:
                ml = " startmsg.regex=\"%s\"" % rs(e["multiline"])
            L.append('input(type="imfile" File="%s" Tag="%s:" Severity="info" Facility="local6"' % (rs(e["path"]), rs(e["name"])))
            L.append('      ruleset="%s"%s' % (rsn, ml))
            L.append('      reopenOnTruncate="on" freshStartTail="on" persistStateInterval="100")')
        else:
            drs = ident_name("aep_logs_dir_", d)
            L.append("# %d files in %s: ONE input (one inotify watch per directory)." % (len(es), d))
            L.append('ruleset(name="%s") {' % drs)
            for i, e in enumerate(es):
                rsn = ident_name("aep_logs_src_", e["service"] + "__" + e["name"])
                L.append("  %s re_match($!metadata!filename, '%s') then { call %s }" % ("if" if i == 0 else "else if", rx(glob_re(e["path"])), rsn))
                if e["multiline"]:
                    warnings.append("%s/%s: multiline not applied, %s holds several declared files (one input per directory)" % (e["service"], e["name"], d))
            L.append("}")
            L.append('input(type="imfile" File="%s/{%s}" Tag="aep-file:" Severity="info" Facility="local6"' % (rs(d), ",".join(os.path.basename(e["path"]) for e in es)))
            L.append('      ruleset="%s" addMetadata="on"' % drs)
            L.append('      reopenOnTruncate="on" freshStartTail="on" persistStateInterval="100")')
        L.append("")
    for e in merged_files + s6_entries:
        if e["multiline"]:
            warnings.append("%s/%s: multiline not applied, the source is read by the shared s6 input" % (e["service"], e["name"]))

    # ---- journal sources: only where a journald runs (not in s6 pods)
    if journal:
        if have_journal:
            L.append('module(load="imjournal" StateFile="aep_logs_journal.state")')
            L.append('ruleset(name="aep_logs_journal") {')
            for e in journal:
                rsn = ident_name("aep_logs_src_", e["service"] + "__" + e["name"])
                L.append('  if ($!_SYSTEMD_UNIT == "%s" or $!_SYSTEMD_UNIT == "%s.service") then { call %s }' % (rs(e["unit"]), rs(e["unit"]), rsn))
            L.append("}")
            for e in journal:
                _, block = source_ruleset(e)
                L += block
        else:
            for e in journal:
                warnings.append("%s/%s: source type journal, but no journald on this node; not forwarded" % (e["service"], e["name"]))
        L.append("")

    # ---- syslog(3) of the node, main ruleset
    if cfg["AEP_LOGS_FORWARD_SYSLOG"].lower() in ("yes", "on", "true", "1"):
        L.append("# syslog(3) of the node (main ruleset), severity from the sender.")
        L.append("set $.sev = $syslogseverity;")
        L.append("set $.app = $programname;")
        L.append('set $.log_name = "syslog";')
        L.append('set $.svc = "%s";' % rs(ident["service"]))
        L.append('set $.sens = "internal";')
        L.append('set $.ret = "default";')
        L.append('set $.tags = "";')
        L.append("# rsyslog start messages this configuration causes itself, on every")
        L.append("# restart (measured on tick1 2026-09-30, mailrelay-vf7rg): the wildcard")
        L.append("# input names plain files in /var/log (level 4) and the TLS client has no")
        L.append("# own certificate (two lines level 3). A configuration must not report")
        L.append("# itself as a permanent alarm. Not with `stop`: /var/log/syslog is written")
        L.append("# by 50-default.conf AFTER this file and keeps the lines.")
        L.append('if not ($programname == "rsyslogd" and ($msg contains "is FILE but DIRECTORY expected" or '
                 '$msg contains "certificate file is not set" or $msg contains "key file is not set")) then {')
        L.append("  call aep_logs_out")
        L.append("}")
    text = "\n".join(L) + "\n"
    summary = {
        "declarations": len(files), "entries": len(entries), "secret_skipped": len(secret),
        "target": "%s:%s %s (%s)" % (ep["fqdn"], ep["port"], proto, ep["from"]),
        "instance": ident["instance"], "service": ident["service"], "project": ident["project"],
    }
    return text, summary, errors, warnings


# ---------------------------------------------------------------------- apply
def write_atomic(path, content, mode=0o644):
    d = os.path.dirname(path)
    os.makedirs(d, exist_ok=True)
    fd, tmp = tempfile.mkstemp(dir=d, prefix=".aep-logs.")
    with os.fdopen(fd, "w") as f:
        f.write(content)
    os.chmod(tmp, mode)
    os.replace(tmp, path)


def run(cmd):
    p = subprocess.run(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, universal_newlines=True)
    return p.returncode, p.stdout.strip()


def signal_rsyslog(cfg, restart):
    cmd = cfg["AEP_LOGS_RESTART_CMD"] if restart else cfg["AEP_LOGS_HUP_CMD"]
    if not cmd:
        sd = cfg["AEP_LOGS_RSYSLOG_SERVICEDIR"]
        if not os.path.isdir(sd):
            return 1, "no s6 service directory %s" % sd
        # s6-svc, not svctl: this runs inside an s6-rc transition (oneshot up),
        # and s6-rc holds its lock there. -r = restart by the supervisor (TERM,
        # rsyslog saves queue and imfile positions on shutdown), -h = HUP.
        cmd = "s6-svc %s %s" % ("-r" if restart else "-h", sd)
    return run(cmd)


def apply(cfg, quiet=False):
    state = cfg["AEP_LOGS_STATE"]
    os.makedirs(state, exist_ok=True)
    lockf = open(os.path.join(state, "lock"), "w")
    fcntl.flock(lockf, fcntl.LOCK_EX)
    try:
        text, summary, errors, warnings = render(cfg)
        # the watcher calls this every few seconds: report a set of errors
        # when it first appears, not in every round
        esig = hashlib.sha256("\n".join(errors + warnings).encode()).hexdigest()
        epath = os.path.join(state, "errors.sha256")
        if not quiet or read_text(epath) != esig:
            for e in errors:
                log("ERROR " + e)
            for w in warnings:
                log("WARNING " + w)
            write_atomic(epath, esig + "\n")
        # lifecycle lookup table first: the config refers to it, rsyslog must
        # find it when it (re)starts.
        lc = lifecycle_state(cfg)
        table = json.dumps({"version": 1, "nomatch": "unknown", "type": "string",
                            "table": [{"index": "state", "value": lc}]}, sort_keys=True) + "\n"
        lpath = os.path.join(state, "lifecycle.lookup.json")
        lc_changed = read_text(lpath) != table.strip()
        if lc_changed:
            write_atomic(lpath, table)
        conf = cfg["AEP_LOGS_CONF"]
        try:
            with open(conf) as f:
                old = f.read()
        except FileNotFoundError:
            old = None
        digest = hashlib.sha256(text.encode()).hexdigest()[:16]
        changed = old != text
        if changed:
            write_atomic(conf, text)
            rc, out = run(cfg["AEP_LOGS_RSYSLOG_CHECK"])
            if rc != 0:
                if old is None:
                    os.unlink(conf)
                else:
                    write_atomic(conf, old)
                log("ERROR rsyslog rejects the rendered config (rolled back): " + " | ".join(out.splitlines()[-5:]))
                write_atomic(os.path.join(state, "rejected.conf"), text)
                return 1
            rc, out = signal_rsyslog(cfg, restart=True)
            if rc != 0:
                log("ERROR restart of rsyslog failed: " + out)
                return 1
        elif lc_changed:
            rc, out = signal_rsyslog(cfg, restart=False)
            if rc != 0:
                log("ERROR HUP to rsyslog failed: " + out)
                return 1
        if changed or lc_changed or not quiet:
            log("config %s sha256:%s, lifecycle %s%s, declarations %d, entries %d, secret skipped %d, instance %s, service %s, project %s, target %s" % (
                "written" if changed else "unchanged", digest, lc, " (HUP)" if lc_changed and not changed else "",
                summary["declarations"], summary["entries"], summary["secret_skipped"], summary["instance"] or "-",
                summary["service"] or "-", summary["project"] or "-", summary["target"]))
        return 1 if errors else 0
    finally:
        fcntl.flock(lockf, fcntl.LOCK_UN)
        lockf.close()


def notify_ready():
    """s6 readiness: ready once the first round ran. The run script names the
    notification fd in AEP_LOGS_NOTIFY_FD (unit file notification-fd)."""
    fd = os.environ.get("AEP_LOGS_NOTIFY_FD", "")
    if not fd.isdigit():
        return
    try:
        os.write(int(fd), b"\n")
        os.close(int(fd))
    except OSError:
        pass


def main(argv):
    cfg = load_settings()
    cmd = argv[1] if len(argv) > 1 else "apply"
    if cmd == "render":
        text, _, errors, warnings = render(cfg)
        sys.stdout.write(text)
        for e in errors:
            print("aep-logs: ERROR " + e, file=sys.stderr)
        for w in warnings:
            print("aep-logs: WARNING " + w, file=sys.stderr)
        return 1 if errors else 0
    if cmd == "apply":
        return apply(cfg, quiet="--quiet" in argv)
    if cmd == "watch":
        interval = max(5, int(cfg["AEP_LOGS_INTERVAL"] or 30))
        log("watching %s, %s and %s every %ds" % (cfg["AEP_LOGS_DIR"], cfg["AEP_LOGS_PODINFO"], cfg["AEP_LOGS_LIFECYCLE"], interval))
        last_rc = None
        while True:
            try:
                # the first round logs its summary line even without a change:
                # it reaches Graylog (log_name aep-logs-watch) and tells what
                # the node forwards (read by tools/verify-order)
                rc = apply(cfg, quiet=last_rc is not None)
            except Exception as ex:  # keep watching; the next round retries
                log("ERROR %s" % ex)
                rc = 1
            if rc != last_rc and rc == 0 and last_rc is not None:
                log("ok again")
            if last_rc is None:
                notify_ready()
            last_rc = rc
            time.sleep(interval)
    print(__doc__, file=sys.stderr)
    return 2


if __name__ == "__main__":
    sys.exit(main(sys.argv))
