Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695"""systemd from Python: units, their state, start/stop, and the journal.
from pyrepl import systemd systemd.status("pyrepld", user=True) like systemctl status: state, PID, memory, recent log systemd.units("pyrepl*", user=True) units by glob (state="failed", "active" ...) systemd.failed() failed units systemd.props("sshd") every property, as Python values systemd.start/stop/restart/reload(unit) waits for the job; returns the new status systemd.enable(unit, now=False), systemd.disable(unit), systemd.daemon_reload() systemd.run(["sleep", "60"], user=True) a command as a transient service (like systemd-run) systemd.journal("pyrepld", user=True, n=50, since="-1h", grep="error", priority=4) systemd.call(user, dest, path, iface, method, sig, *args) any D-Bus call (raw: no root check, so a system write may make polkit ask)
user=True talks to your own service manager (systemctl --user); the defaultis the system one, where reading is open to everyone but changing thingsneeds root (a root session: session("x").sudo(), or sudo=True on exec).
It binds the machine's libsystemd through ctypes: sd-bus for the servicemanager and sd-journal for logs, so nothing runs a shell or parses text. Ona host without libsystemd it falls back to systemctl and journalctl.
Must stay runnable on python 3.6 (sessions can run on any 3.6+)."""import ctypesimport ctypes.utilimport datetimeimport jsonimport osimport reimport shutilimport subprocessimport threadingimport time
try: from tk import Outexcept ImportError: # a session started with PYREPL_NO_TK class Out(str): def __pyrepl_show__(self): return str(self)
__all__ = ["status", "units", "failed", "props", "start", "stop", "restart", "reload", "enable", "disable", "daemon_reload", "run", "journal", "call", "SystemdError"]
DEST = "org.freedesktop.systemd1"MANAGER = "/org/freedesktop/systemd1"MGR_IFACE = "org.freedesktop.systemd1.Manager"UNSET = 2 ** 64 - 1 # systemd's "infinity / not set" for uint64 properties
class SystemdError(RuntimeError): pass
# -- sd-bus through ctypes --------------------------------------------------------class _BusError(ctypes.Structure): _fields_ = [("name", ctypes.c_char_p), ("message", ctypes.c_char_p), ("_need_free", ctypes.c_int)]
def _load(): name = ctypes.util.find_library("systemd") or "libsystemd.so.0" try: lib = ctypes.CDLL(name, use_errno=True) lib.sd_bus_open_user # old libsystemd without sd-bus: no use to us except (OSError, AttributeError): return None P, vp, cp, ch, i = ctypes.POINTER, ctypes.c_void_p, ctypes.c_char_p, ctypes.c_char, ctypes.c_int sigs = { "sd_bus_open_user": ([P(vp)], i), "sd_bus_open_system": ([P(vp)], i), "sd_bus_flush_close_unref": ([vp], vp), "sd_bus_message_new_method_call": ([vp, P(vp), cp, cp, cp, cp], i), "sd_bus_message_append_basic": ([vp, ch, vp], i), "sd_bus_message_open_container": ([vp, ch, cp], i), "sd_bus_message_close_container": ([vp], i), "sd_bus_call": ([vp, vp, ctypes.c_uint64, P(_BusError), P(vp)], i), "sd_bus_error_free": ([P(_BusError)], None), "sd_bus_message_unref": ([vp], vp), "sd_bus_message_peek_type": ([vp, P(ch), P(cp)], i), "sd_bus_message_read_basic": ([vp, ch, vp], i), "sd_bus_message_enter_container": ([vp, ch, cp], i), "sd_bus_message_exit_container": ([vp], i), "sd_journal_open": ([P(vp), i], i), "sd_journal_close": ([vp], None), "sd_journal_add_match": ([vp, cp, ctypes.c_size_t], i), "sd_journal_add_disjunction": ([vp], i), "sd_journal_add_conjunction": ([vp], i), "sd_journal_seek_tail": ([vp], i), "sd_journal_seek_realtime_usec": ([vp, ctypes.c_uint64], i), "sd_journal_previous": ([vp], i), "sd_journal_next": ([vp], i), "sd_journal_get_data": ([vp, cp, P(vp), P(ctypes.c_size_t)], i), "sd_journal_get_realtime_usec": ([vp, P(ctypes.c_uint64)], i), "sd_journal_restart_data": ([vp], None), "sd_journal_has_runtime_files": ([vp], i), "sd_journal_has_persistent_files": ([vp], i), "sd_journal_enumerate_data": ([vp, P(vp), P(ctypes.c_size_t)], i), } for fn, (args, res) in sigs.items(): f = getattr(lib, fn) f.argtypes, f.restype = args, res return lib
LIB = _load()_LOCK = threading.RLock() # sd-bus objects are not thread-safe_BUSES = {}
# D-Bus basic types: ctypes holder for each_BASIC = {"y": ctypes.c_uint8, "b": ctypes.c_int, "n": ctypes.c_int16, "q": ctypes.c_uint16, "i": ctypes.c_int32, "u": ctypes.c_uint32, "x": ctypes.c_int64, "t": ctypes.c_uint64, "d": ctypes.c_double, "h": ctypes.c_int}_STRINGS = "sog"
def _check(r, what): if r < 0: raise SystemdError("%s: %s" % (what, os.strerror(-r))) return r
def _bus(user): """The connection to the user or system bus, one per process (sd-bus connections don't survive a fork).""" key = (bool(user), os.getpid()) b = _BUSES.get(key) if b is None: p = ctypes.c_void_p() _check((LIB.sd_bus_open_user if user else LIB.sd_bus_open_system)(ctypes.byref(p)), "connecting to the %s bus" % ("user" if user else "system")) b = _BUSES[key] = p return b
def _split(sig): """'sa{sv}(ii)' -> ['s', 'a{sv}', '(ii)']: complete types.""" out, i = [], 0 while i < len(sig): j = i while sig[j] == "a": j += 1 if sig[j] in "({": depth, k = 0, j while True: depth += sig[k] in "({" depth -= sig[k] in ")}" k += 1 if not depth: break j = k else: j += 1 out.append(sig[i:j]) i = j return out
def _guess(v): """A variant's signature from a Python value; (sig, value) says it outright.""" if isinstance(v, bool): return "b" if isinstance(v, int): return "t" # systemd's numeric properties are mostly uint64 if isinstance(v, float): return "d" if isinstance(v, str): return "s" if isinstance(v, (list, tuple)) and all(isinstance(x, str) for x in v): return "as" raise SystemdError("can't tell the D-Bus type of %r; pass (signature, value)" % (v,))
def _append(m, sig, v, keep): t = sig[0] if t in _BASIC: h = _BASIC[t](int(v) if t != "d" else float(v)) keep.append(h) _check(LIB.sd_bus_message_append_basic(m, t.encode(), ctypes.cast(ctypes.pointer(h), ctypes.c_void_p)), "append") elif t in _STRINGS: s = ctypes.c_char_p(str(v).encode("utf-8")) keep.append(s) _check(LIB.sd_bus_message_append_basic(m, t.encode(), ctypes.cast(s, ctypes.c_void_p)), "append") elif t == "a": inner = sig[1:] _check(LIB.sd_bus_message_open_container(m, b"a", inner.encode()), "append") if inner[0] == "{": ks, vs = _split(inner[1:-1]) for k, x in (v.items() if isinstance(v, dict) else v): _check(LIB.sd_bus_message_open_container(m, b"e", inner[1:-1].encode()), "append") _append(m, ks, k, keep) _append(m, vs, x, keep) _check(LIB.sd_bus_message_close_container(m), "append") else: for x in v: _append(m, inner, x, keep) _check(LIB.sd_bus_message_close_container(m), "append") elif t == "(": fields = _split(sig[1:-1]) _check(LIB.sd_bus_message_open_container(m, b"r", sig[1:-1].encode()), "append") for fs, x in zip(fields, v): _append(m, fs, x, keep) _check(LIB.sd_bus_message_close_container(m), "append") elif t == "v": vs, x = v if isinstance(v, tuple) and len(v) == 2 and isinstance(v[0], str) \ and v[0] and _is_sig(v[0]) else (_guess(v), v) _check(LIB.sd_bus_message_open_container(m, b"v", vs.encode()), "append") _append(m, vs, x, keep) _check(LIB.sd_bus_message_close_container(m), "append") else: raise SystemdError("D-Bus type %r is not supported" % t)
def _is_sig(s): return bool(re.match(r"^[ybnqiuxtdhsogav(){}]+$", s))
def _read_all(m): out = [] while True: t, contents = ctypes.c_char(), ctypes.c_char_p() r = _check(LIB.sd_bus_message_peek_type(m, ctypes.byref(t), ctypes.byref(contents)), "read") if r == 0: return out out.append(_read_one(m, t.value.decode(), contents.value))
def _read_one(m, t, contents): if t in _BASIC: h = _BASIC[t]() _check(LIB.sd_bus_message_read_basic(m, t.encode(), ctypes.cast(ctypes.pointer(h), ctypes.c_void_p)), "read") return bool(h.value) if t == "b" else h.value if t in _STRINGS: h = ctypes.c_char_p() _check(LIB.sd_bus_message_read_basic(m, t.encode(), ctypes.cast(ctypes.pointer(h), ctypes.c_void_p)), "read") return h.value.decode("utf-8", "replace") kind = {"a": b"a", "r": b"r", "v": b"v", "e": b"e"}[t] _check(LIB.sd_bus_message_enter_container(m, kind, contents), "read") items = _read_all(m) _check(LIB.sd_bus_message_exit_container(m), "read") if t == "a": if contents.startswith(b"{"): return dict(items) if contents == b"y": return bytes(items) return items if t == "v": return items[0] return tuple(items) # struct or dict entry
def call(user, dest, path, iface, method, sig="", *args, **kw): """One D-Bus method call; the reply as a list of Python values. sig is the argument signature ("ss" ...); a variant takes a value, or (signature, value) when its type can't be guessed.""" if LIB is None: raise SystemdError("no libsystemd here") with _LOCK: bus = _bus(user) m, reply = ctypes.c_void_p(), ctypes.c_void_p() err = _BusError() keep = [] _check(LIB.sd_bus_message_new_method_call(bus, ctypes.byref(m), dest.encode(), path.encode(), iface.encode(), method.encode()), "new call") try: for s, a in zip(_split(sig), args): _append(m, s, a, keep) r = LIB.sd_bus_call(bus, m, int(kw.get("timeout", 60) * 1e6), ctypes.byref(err), ctypes.byref(reply)) if r < 0: name = (err.name or b"").decode() msg = (err.message or b"").decode() or os.strerror(-r) LIB.sd_bus_error_free(ctypes.byref(err)) hint = "" if name.endswith(("AccessDenied", "InteractiveAuthorizationRequired")) or ( name.endswith("NoReply") and not user and os.geteuid()): hint = (" (changing system units needs root: run it in a root session, " "or pass user=True for your own units%s)" % ( "; no answer usually means polkit is waiting for an agent" if name.endswith("NoReply") else "")) raise SystemdError("%s: %s%s" % (name or "error", msg, hint)) try: return _read_all(reply) finally: LIB.sd_bus_message_unref(reply) finally: LIB.sd_bus_message_unref(m)
_WRITES = ("StartUnit", "StopUnit", "RestartUnit", "ReloadUnit", "EnableUnitFiles", "DisableUnitFiles", "Reload", "StartTransientUnit")
def _need_root(user, what): """Changes to system units are refused here, before any call, unless we are root. Never ask: on some systems (seen on RHEL 8, systemd 239) polkit pops an authentication dialog on the user's desktop even though the call doesn't allow interactive authorization, and systemctl would too.""" if not user and os.geteuid() != 0: raise SystemdError("%s on a system unit needs root: run it in a root session " "(session(\"x\").sudo(), or exec with sudo=True), or pass user=True " "for your own units" % what)
def _mgr(user, method, sig="", *args): if method in _WRITES: _need_root(user, method) return call(user, DEST, MANAGER, MGR_IFACE, method, sig, *args)
# -- units ----------------------------------------------------------------------------_TYPES = {"service": "Service", "socket": "Socket", "timer": "Timer", "mount": "Mount", "automount": "Automount", "path": "Path", "scope": "Scope", "slice": "Slice", "target": "Target", "swap": "Swap", "device": "Device"}
def _name(unit): """"pyrepld" -> "pyrepld.service": a bare name means a service, like systemctl.""" unit = str(unit) if "." not in unit or unit.rsplit(".", 1)[1] not in _TYPES: return unit + ".service" return unit
def _usec(v): if not v or v == UNSET: return None return datetime.datetime.fromtimestamp(v / 1e6).astimezone()
def _size(v): if v is None or v == UNSET: return None for unit in ("B", "K", "M", "G", "T"): if v < 1024 or unit == "T": return ("%.1f%s" % (v, unit)) if unit != "B" else "%dB" % v v /= 1024.0
def _ago(dt): s = int(time.time() - dt.timestamp()) for n, u in ((86400, "d"), (3600, "h"), (60, "min")): if s >= n: return "%d%s ago" % (s // n, u) return "%ds ago" % s
class Unit(dict): """A unit's properties (name -> value), shown like systemctl status."""
journal_tail = None
def __getattr__(self, k): try: return self[k] except KeyError: raise AttributeError(k)
def __pyrepl_show__(self): g = self.get active, sub = g("ActiveState", "?"), g("SubState", "?") dot = {"active": "●", "failed": "×", "inactive": "○"}.get(active, "●") out = ["%s %s - %s" % (dot, g("Id"), g("Description", ""))] where = g("FragmentPath") or g("SourcePath") or "" state = g("UnitFileState") or ("transient" if g("Transient") else "") out.append(" Loaded: %s (%s)" % (g("LoadState"), "; ".join(x for x in (where, state) if x))) since = _usec(g("ActiveEnterTimestamp") if active == "active" else g("InactiveEnterTimestamp")) line = " Active: %s (%s)" % (active, sub) if since: line += " since %s; %s" % (since.strftime("%a %Y-%m-%d %H:%M:%S %Z"), _ago(since)) out.append(line) if g("Result") and g("Result") != "success": out.append(" Result: %s" % g("Result")) if g("MainPID"): out.append(" Main PID: %d" % g("MainPID")) if g("TasksCurrent") not in (None, UNSET): out.append(" Tasks: %d%s" % (g("TasksCurrent"), " (limit: %d)" % g("TasksMax") if g("TasksMax") not in (None, UNSET) else "")) if g("MemoryCurrent") not in (None, UNSET): lim = ", ".join("%s: %s" % (k[6:].lower(), _size(g(k))) for k in ("MemoryHigh", "MemoryMax") if g(k) not in (None, UNSET)) out.append(" Memory: %s%s" % (_size(g("MemoryCurrent")), " (%s)" % lim if lim else "")) if g("ControlGroup"): out.append(" CGroup: %s" % g("ControlGroup")) if self.journal_tail: out.append("") out.append(self.journal_tail.__pyrepl_show__()) return "\n".join(out)
def __repr__(self): return "<Unit %s %s/%s>" % (self.get("Id"), self.get("ActiveState"), self.get("SubState"))
class UnitList(list): def __pyrepl_show__(self): if not self: return "(no units)" w = max(len(u["unit"]) for u in self) return "\n".join("%-*s %-9s %-9s %-10s %s" % (w, u["unit"], u["load"], u["active"], u["sub"], u["description"]) for u in self)
def props(unit, user=False): """Every property of a unit (the Unit interface and its type's, like Service), as Python values. Times are µs, sizes bytes; 2**64-1 = unset.""" name = _name(unit) if LIB is None: return _cli_props(name, user) path = _mgr(user, "LoadUnit", "s", name)[0] out = Unit() for iface in ("org.freedesktop.systemd1.Unit", "org.freedesktop.systemd1." + _TYPES.get(name.rsplit(".", 1)[1], "Unit")): out.update(call(user, DEST, path, "org.freedesktop.DBus.Properties", "GetAll", "s", iface)[0]) return out
def status(unit, user=False, lines=10): """A unit's state, like systemctl status (with its last log lines).""" u = props(unit, user) if lines: try: u.journal_tail = journal(u.get("Id") or _name(unit), user=user, n=lines) except SystemdError: pass return u
def units(pattern=None, user=False, state=None, all=False): """Units, like systemctl list-units: by glob (a list of globs works too) and state ("active", "failed", "inactive", a sub-state...). all=True also lists inactive units and aliases (systemctl --all).""" pats = [pattern] if isinstance(pattern, str) else list(pattern or []) states = [state] if isinstance(state, str) else list(state or []) if LIB is None: rows = _cli_units(pats, states, user, all) else: rows = [dict(unit=r[0], description=r[1], load=r[2], active=r[3], sub=r[4]) for r in _mgr(user, "ListUnitsByPatterns", "asas", states, pats)[0] if all or states or not (r[3] == "inactive" or r[5])] # r[5]: follows another return UnitList(sorted(rows, key=lambda r: r["unit"]))
def failed(user=False): return units(user=user, state="failed")
def _wait_job(name, user, timeout): """Until the unit has no job pending, or timeout.""" deadline = time.time() + timeout while time.time() < deadline: p = props(name, user) job = p.get("Job") if not job or job[0] == 0: return p time.sleep(0.1) raise SystemdError("%s still has a job pending after %ss" % (name, timeout))
def _job(method, unit, user, wait, timeout): name = _name(unit) _need_root(user, method) if LIB is None: _cli(["systemctl"] + (["--user"] if user else []) + [method.lower().replace("unit", ""), name] + ([] if wait else ["--no-block"])) return status(name, user) _mgr(user, method, "ss", name, "replace") if not wait: return props(name, user) _wait_job(name, user, timeout) return status(name, user)
def start(unit, user=False, wait=True, timeout=90): """Start a unit; waits for its job, then returns its status.""" return _job("StartUnit", unit, user, wait, timeout)
def stop(unit, user=False, wait=True, timeout=90): return _job("StopUnit", unit, user, wait, timeout)
def restart(unit, user=False, wait=True, timeout=90): return _job("RestartUnit", unit, user, wait, timeout)
def reload(unit, user=False, wait=True, timeout=90): return _job("ReloadUnit", unit, user, wait, timeout)
def daemon_reload(user=False): _need_root(user, "daemon-reload") if LIB is None: return _cli(["systemctl"] + (["--user"] if user else []) + ["daemon-reload"]) _mgr(user, "Reload") return Out("reloaded")
def _changes(rows): return Out("\n".join("%s %s%s" % (k, f, " -> " + d if d else "") for k, f, d in rows) or "(no changes)")
def enable(unit, user=False, now=False, runtime=False): _need_root(user, "enable") name = _name(unit) if LIB is None: return _cli(["systemctl"] + (["--user"] if user else []) + ["enable", name] + (["--now"] if now else []) + (["--runtime"] if runtime else [])) _, rows = _mgr(user, "EnableUnitFiles", "asbb", [name], runtime, False) _mgr(user, "Reload") if now: start(name, user) return _changes(rows)
def disable(unit, user=False, now=False, runtime=False): _need_root(user, "disable") name = _name(unit) if LIB is None: return _cli(["systemctl"] + (["--user"] if user else []) + ["disable", name] + (["--now"] if now else []) + (["--runtime"] if runtime else [])) rows = _mgr(user, "DisableUnitFiles", "asb", [name], runtime)[0] _mgr(user, "Reload") if now: stop(name, user) return _changes(rows)
def run(cmd, user=False, unit=None, props=None, wait=False, timeout=90): """Run a command as a transient service, like systemd-run: it outlives this session and logs to the journal (journal(unit)). cmd is an argv list, its first word looked up on PATH. props are extra unit properties ({"MemoryMax": 2 << 30, "Environment": ["A=1"]} ...); wait=True waits for it to start. Returns the unit's status.""" _need_root(user, "run") argv = [str(a) for a in cmd] exe = argv[0] if "/" in argv[0] else shutil.which(argv[0]) if not exe: raise SystemdError("%s: not found on PATH" % argv[0]) name = _name(unit or "pyrepl-run-%d-%d" % (os.getpid(), int(time.time() * 1000) % 10 ** 9)) p = {"Description": " ".join(argv), "ExecStart": ("a(sasb)", [(exe, argv, False)]), "WorkingDirectory": os.getcwd(), "RemainAfterExit": False} p.update(props or {}) if LIB is None: _cli(["systemd-run"] + (["--user"] if user else []) + ["--unit", name, "--working-directory", os.getcwd(), "--", exe] + argv[1:]) return status(name, user) _mgr(user, "StartTransientUnit", "ssa(sv)a(sa(sv))", name, "fail", list(p.items()), []) if wait: _wait_job(name, user, timeout) return status(name, user, lines=0)
# -- the journal ----------------------------------------------------------------------class Entry(dict): """One journal entry: its fields (MESSAGE, _PID, ...) plus .time."""
time = None
def line(self): when = self.time.strftime("%b %d %H:%M:%S") if self.time else "?" ident = self.get("SYSLOG_IDENTIFIER") or self.get("_COMM") or "?" pid = self.get("_PID") or self.get("SYSLOG_PID") return "%s %s %s%s: %s" % (when, self.get("_HOSTNAME", ""), ident, "[%s]" % pid if pid else "", self.get("MESSAGE", ""))
class Journal(list): def __pyrepl_show__(self): return "\n".join(e.line() for e in self) or "-- No entries --"
def _since(v): """datetime, epoch seconds, or "-1h"/"-30min"/"-2d" -> µs since the epoch.""" if v is None: return None if isinstance(v, datetime.datetime): return int(v.timestamp() * 1e6) if isinstance(v, (int, float)): return int(v * 1e6) m = re.match(r"^-?(\d+)\s*(s|sec|m|min|h|d)$", str(v).strip()) if not m: raise SystemdError("since: a datetime, epoch seconds, or like '-2h' (got %r)" % (v,)) mult = {"s": 1, "sec": 1, "m": 60, "min": 60, "h": 3600, "d": 86400}[m.group(2)] return int((time.time() - int(m.group(1)) * mult) * 1e6)
def journal(unit=None, user=False, n=50, since=None, grep=None, priority=None, fields=None): """The last n entries (oldest first), for one unit or all you can read. since: a datetime, epoch seconds, or "-1h"; grep: regex on MESSAGE; priority: 0-7, that and more urgent; fields: extra FIELD=value matches.""" if LIB is None: return _cli_journal(unit, user, n, since, grep, priority, fields) j = ctypes.c_void_p() _check(LIB.sd_journal_open(ctypes.byref(j), 1), "opening the journal") # LOCAL_ONLY try: if not (LIB.sd_journal_has_runtime_files(j) > 0 or LIB.sd_journal_has_persistent_files(j) > 0): raise SystemdError("no journal files here are readable by uid %d (journalctl says the " "same): reading the journal needs the systemd-journal group, or " "a root session" % os.geteuid()) def match(s): b = s.encode() _check(LIB.sd_journal_add_match(j, b, len(b)), "journal match")
extra = ["%s=%s" % kv for kv in (fields or {}).items()] if unit: name = _name(unit) # Like journalctl -u / --user-unit: the unit's own processes, or # systemd's messages about it. groups = ([["_SYSTEMD_USER_UNIT=" + name, "_UID=%d" % os.getuid()], ["USER_UNIT=" + name, "_UID=%d" % os.getuid()]] if user else [["_SYSTEMD_UNIT=" + name], ["UNIT=" + name, "_PID=1"]]) for k, g in enumerate(groups): if k: _check(LIB.sd_journal_add_disjunction(j), "journal match") for s in g + extra: match(s) else: for s in extra: match(s) pat = re.compile(grep) if grep else None out = [] start = _since(since) _check(LIB.sd_journal_seek_tail(j), "journal seek") data, size, usec = ctypes.c_void_p(), ctypes.c_size_t(), ctypes.c_uint64() while len(out) < n and _check(LIB.sd_journal_previous(j), "journal read") > 0: _check(LIB.sd_journal_get_realtime_usec(j, ctypes.byref(usec)), "journal read") if start is not None and usec.value < start: break e = Entry() LIB.sd_journal_restart_data(j) while LIB.sd_journal_enumerate_data(j, ctypes.byref(data), ctypes.byref(size)) > 0: k, _, v = ctypes.string_at(data, size.value).partition(b"=") e[k.decode("ascii", "replace")] = v.decode("utf-8", "replace") if priority is not None and int(e.get("PRIORITY", 6)) > priority: continue if pat and not pat.search(e.get("MESSAGE", "")): continue e.time = datetime.datetime.fromtimestamp(usec.value / 1e6) out.append(e) return Journal(reversed(out)) finally: LIB.sd_journal_close(j)
# -- without libsystemd: the command-line tools ---------------------------------------def _cli(argv): p = subprocess.run(argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, env=dict(os.environ, SYSTEMD_PAGER="", SYSTEMD_COLORS="0")) if p.returncode: raise SystemdError("%s: %s" % (" ".join(argv), (p.stderr or p.stdout).strip())) return Out(p.stdout)
def _cli_props(name, user): out = Unit() for line in _cli(["systemctl"] + (["--user"] if user else []) + ["show", name]).splitlines(): k, _, v = line.partition("=") out[k] = int(v) if v.isdigit() else v return out
def _cli_units(pats, states, user, all): argv = ["systemctl"] + (["--user"] if user else []) + ["list-units", "--plain", "--no-legend"] argv += (["--all"] if all else []) + ["--state=" + s for s in states] + pats rows = [] for line in _cli(argv).splitlines(): f = line.split(None, 4) if len(f) >= 4: rows.append(dict(unit=f[0], load=f[1], active=f[2], sub=f[3], description=f[4] if len(f) > 4 else "")) return rows
def _cli_journal(unit, user, n, since, grep, priority, fields): argv = ["journalctl", "-o", "json", "--no-pager", "-n", str(n)] + (["--user"] if user else []) if unit: argv += ["--user-unit" if user else "-u", _name(unit)] if since is not None: argv += ["--since", "@%d" % (_since(since) // 10 ** 6)] if priority is not None: argv += ["-p", str(priority)] if grep: argv += ["-g", grep] argv += ["%s=%s" % kv for kv in (fields or {}).items()] out = Journal() for line in _cli(argv).splitlines(): e = Entry(json.loads(line)) e.time = datetime.datetime.fromtimestamp(int(e.get("__REALTIME_TIMESTAMP", 0)) / 1e6) out.append(e) return out