Cython中的systemd包装器

使用Cython实现的Python systemd包装器。

安装

所有包均可在github releases <https://github.com/mosquito/cysystemd/releases>上找到。

从二进制轮安装

• 目前为Python 3.8, 3.9, 3.10, 3.11, 3.12的x86_64和arm64提供轮。

python3.10 -m pip install \
  https://github.com/mosquito/cysystemd/releases/download/1.6.2/cysystemd-1.6.2-cp310-cp310-linux_x86_64.whl

从源安装

您必须安装systemd headers

对于Debian/Ubuntu用户

apt install build-essential libsystemd-dev

在Debian/Ubuntu的较旧版本中，您可能还需要安装

apt install libsystemd-daemon-dev libsystemd-journal-dev

对于CentOS/RHEL

yum install gcc systemd-devel

然后从pypi安装它

pip install cysystemd

使用示例

写入journald

python logger的日志处理器

from cysystemd import journal
import logging
import uuid

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger()
logger.addHandler(journal.JournaldLogHandler())

try:
    logger.info("Trying to do something")
    raise Exception('foo')
except:
    logger.exception("Test Exception %s", 1)

systemd守护进程通知

from cysystemd.daemon import notify, Notification

# Send READY=1
notify(Notification.READY)

# Send status
notify(Notification.STATUS, "I'm fine.")

# Send stopping
notify(Notification.STOPPING)

将消息写入systemd日志

from cysystemd import journal


journal.write("Hello Lennart")

# Or send structured data
journal.send(
    message="Hello Lennart",
    priority=journal.Priority.INFO,
    some_field='some value',
)

读取journald

读取所有systemd记录

from cysystemd.reader import JournalReader, JournalOpenMode

journal_reader = JournalReader()
journal_reader.open(JournalOpenMode.SYSTEM)
journal_reader.seek_head()

for record in journal_reader:
    print(record.data['MESSAGE'])

仅读取cron日志

from cysystemd.reader import JournalReader, JournalOpenMode, Rule


rules = (
  Rule("SYSLOG_IDENTIFIER", "CRON") &
  Rule("_SYSTEMD_UNIT", "crond.service") |
  Rule("_SYSTEMD_UNIT", "cron.service")
)

cron_reader = JournalReader()
cron_reader.open(JournalOpenMode.SYSTEM)
cron_reader.seek_head()
cron_reader.add_filter(rules)

for record in cron_reader:
    print(record.data['MESSAGE'])

轮询记录

from cysystemd.reader import JournalReader, JournalOpenMode


reader = JournalReader()
reader.open(JournalOpenMode.SYSTEM)
reader.seek_tail()

poll_timeout = 255

while True:
    reader.wait(poll_timeout)

    for record in reader:
       print(record.data['MESSAGE'])

journald打开模式

• CURRENT_USER
• LOCAL_ONLY
• RUNTIME_ONLY
• SYSTEM
• SYSTEM_ONLY - SYSTEM的已弃用别名

from cysystemd.reader import JournalReader, JournalOpenMode


reader = JournalReader()
reader.open(JournalOpenMode.CURRENT_USER)

journald条目

JournalEntry类有一些特殊的属性和方法

• data - 日志条目内容 (dict)
• date - 条目时间戳 (datetime实例)
• cursor - 系统d为此条目识别的字节
• boot_id() - 返回引导ID
• get_realtime_sec() - 条目纪元 (float)
• get_realtime_usec() - 条目纪元 (int 微秒)
• get_monotonic_sec() - 条目单调时间 (float)
• get_monotonic_usec() - 条目单调时间 (int 微秒)
• __getitem__(key) - entry.data[key] 的快捷方式

journald 读取器

JournalReader 类有一些特殊的属性和方法

• open(flags=JournalOpenMode.CURRENT_USER) - 以选定的模式打开journald
• open_directory(path) - 从路径打开journald
• open_files(*filename) - 从文件打开journald
• data_threshold - 可以用于获取或设置获取条目数据返回的数据字段大小阈值。
• closed - 当journald 读取器关闭时返回 True
• locked - 当journald 读取器锁定时返回 True
• idle - 当journald 读取器打开时返回 True
• seek_head - 将读取器指针移动到第一条目
• seek_tail - 将读取器指针移动到最后一条目
• seek_monotonic_usec - 跳转到具有指定单调时间戳的条目，即 CLOCK_MONOTONIC。由于单调时间在每次重启时都会重置，因此还需要指定引导ID。
• seek_realtime_usec - 跳转到具有指定实时时间戳的条目，即 CLOCK_REALTIME。注意，实时时钟不一定单调。如果实时时间戳模糊不清，则未定义要查找的位置。
• seek_cursor - 跳转到指定光标位置处的条目（见 JournalEntry.cursor）。
• wait(timeout) - 它将同步等待直到journald发生变化。此调用睡眠的最大时间可以通过 timeout_usec 参数控制。
• __iter__ - 返回 JournalReader 对象
• __next__ - 调用 next() 或引发 StopIteration
• next(skip=0) - 返回下一个 JournalEntry。skip 参数跳过一些条目。
• previous(skip=0) - 返回上一个 JournalEntry。skip 参数跳过一些条目。
• skip_next(skip) - 跳过后续条目。
• skip_previous(skip) - 跳过前序条目。
• add_filter(rule) - 添加过滤规则。请参阅 read-only-cron-logs 作为示例。
• clear_filter - 重置所有过滤器
• fd - 返回特殊文件描述符
• events - 返回 EPOLL 事件
• timeout - 返回内部超时
• process_events() - 在每次 poll() 唤醒后需要调用 process_events() 来处理事件。此调用还将指示检测到的更改类型。
• get_catalog() - 获取当前日志条目的消息目录条目。它将通过当前日志条目的 "MESSAGE_ID=" 字段在消息目录中查找条目。在返回条目之前，将用当前条目的相应字段值替换目录条目文本中包含的@符号的所有日志字段名。如果消息目录条目中引用的字段名在当前日志条目中不存在，则将移除@符号，但字段名其他部分保持不变。
• get_catalog_for_message_id(message_id: UUID) - 与 get_catalog() 类似，但通过指定的消息ID查找条目（不需要打开日志上下文），并且不执行字段替换。

异步支持

对读取日志的异步读取的初始 asyncio 支持。

AsyncJournalReader

阻塞方法被线程包装。方法 wait() 使用 journald 文件描述符的 epoll。

import asyncio
import json

from cysystemd.reader import JournalOpenMode
from cysystemd.async_reader import AsyncJournalReader


async def main():
    reader = AsyncJournalReader()
    await reader.open(JournalOpenMode.SYSTEM)
    await reader.seek_tail()

    while await reader.wait():
        async for record in reader:
            print(
                json.dumps(
                    record.data,
                    indent=1,
                    sort_keys=True
                )
            )

if __name__ == '__main__':
    asyncio.run(main())