| Server IP : 74.208.127.88 / Your IP : 3.135.185.96 Web Server : Apache/2.4.41 (Ubuntu) System : Linux ubuntu 5.4.0-163-generic #180-Ubuntu SMP Tue Sep 5 13:21:23 UTC 2023 x86_64 User : www-data ( 33) PHP Version : 7.4.3-4ubuntu2.29 Disable Function : pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare, MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : ON | Pkexec : ON Directory : /proc/self/root/usr/lib/python3/dist-packages/cloudinit/reporting/ |
Upload File : |
# Copyright (C) 2015 Canonical Ltd.
#
# This file is part of cloud-init. See LICENSE file for license information.
"""
cloud-init reporting framework
The reporting framework is intended to allow all parts of cloud-init to
report events in a structured manner.
"""
from typing import Type
from cloudinit.registry import DictRegistry
from cloudinit.reporting.handlers import HandlerType, available_handlers
DEFAULT_CONFIG = {
"logging": {"type": "log"},
}
def update_configuration(config):
"""Update the instantiated_handler_registry.
:param config:
The dictionary containing changes to apply. If a key is given
with a False-ish value, the registered handler matching that name
will be unregistered.
"""
for handler_name, handler_config in config.items():
if not handler_config:
instantiated_handler_registry.unregister_item(
handler_name, force=True
)
continue
handler_config = handler_config.copy()
cls: Type[HandlerType] = available_handlers.registered_items[
handler_config.pop("type")
]
instantiated_handler_registry.unregister_item(handler_name)
instance = cls(**handler_config) # pyright: ignore
instantiated_handler_registry.register_item(handler_name, instance)
def flush_events():
handler: HandlerType
for handler in instantiated_handler_registry.registered_items.values():
handler.flush()
instantiated_handler_registry = DictRegistry()
update_configuration(DEFAULT_CONFIG)