MAJOR=1 MINOR=8 DEVNAME=random DEVMODE=0666 1ˆ¾Ë 0ÊÁÍ$3ˆl–Ðz¾”ji?u4 \¸î*bÑ¿Ì „^?ËÂÎ./3# Copyright (C) 2009-2010 Canonical Ltd. # Copyright (C) 2012 Hewlett-Packard Development Company, L.P. # Copyright (C) 2023 FreeBSD Foundation # # Author: Scott Moser # Author: Juerg Haefliger # Author: Mina Galić # # This file is part of cloud-init. See LICENSE file for license information. """Rsyslog: Configure system logging via rsyslog""" import contextlib import copy import logging import os import re from cloudinit import lifecycle, subp, util from cloudinit.cloud import Cloud from cloudinit.config import Config from cloudinit.config.schema import MetaSchema from cloudinit.distros import ALL_DISTROS, Distro from cloudinit.log import loggers from cloudinit.settings import PER_INSTANCE meta: MetaSchema = { "id": "cc_rsyslog", "distros": [ALL_DISTROS], "frequency": PER_INSTANCE, "activate_by_schema_keys": ["rsyslog"], } RSYSLOG_CONFIG = { "config_dir": "/etc/rsyslog.d", "config_filename": "20-cloud-config.conf", "service_reload_command": "auto", "remotes": {}, "configs": {}, "check_exe": "rsyslogd", "packages": ["rsyslog"], "install_rsyslog": False, } DISTRO_OVERRIDES = { "freebsd": { "config_dir": "/usr/local/etc/rsyslog.d", "packages": ["rsyslog"], }, "openbsd": { "config_dir": "/usr/local/etc/rsyslog.d", "packages": ["sysutils/rsyslog"], }, "netbsd": { "config_dir": "/usr/pkg/etc/rsyslog.d", "packages": ["sysutils/rsyslog"], }, } LOG = logging.getLogger(__name__) COMMENT_RE = re.compile(r"[ ]*[#]+[ ]*") HOST_PORT_RE = re.compile( r"^(?P[@]{0,2})" r"(([\[](?P[^\]]*)[\]])|(?P[^:]*))" r"([:](?P[0-9]+))?$" ) def distro_default_rsyslog_config(distro: Distro): """Construct a distro-specific rsyslog config dictionary by merging distro specific changes into base config. @param distro: String providing the distro class name. @returns: Dict of distro configurations for ntp clients. """ dcfg = DISTRO_OVERRIDES cfg = copy.copy(RSYSLOG_CONFIG) if distro.osfamily in dcfg: cfg = util.mergemanydict([cfg, dcfg[distro.name]], reverse=True) return cfg def install_rsyslog(install_func, packages=None, check_exe="rsyslog"): """Install rsyslog package if not already installed. @param install_func: function. This parameter is invoked with the contents of the packages parameter. @param packages: list. This parameter defaults to ['rsyslog']. @param check_exe: string. The name of a binary that indicates the package the specified package is already installed. """ if subp.which(check_exe): return if packages is None: packages = ["rsyslog"] install_func(packages) def reload_syslog(distro, command="auto"): if command == "auto": service = distro.get_option("rsyslog_svcname", "rsyslog") return distro.manage_service("try-reload", service) return subp.subp(command, capture=True) def load_config(cfg: dict, distro: Distro) -> dict: """Return an updated config. Support converting the old top level format into new format. Raise a `ValueError` if some top level entry has an incorrect type. """ mycfg = cfg.get("rsyslog", {}) distro_config = distro_default_rsyslog_config(distro) if isinstance(cfg.get("rsyslog"), list): lifecycle.deprecate( deprecated="The rsyslog key with value of type 'list'", deprecated_version="22.2", ) mycfg = {"configs": cfg.get("rsyslog")} if "rsyslog_filename" in cfg: mycfg["config_filename"] = cfg["rsyslog_filename"] if "rsyslog_dir" in cfg: mycfg["config_dir"] = cfg["rsyslog_dir"] fillup: tuple = ( ("configs", [], list), ("config_dir", distro_config["config_dir"], str), ("config_filename",