MR K IS HERE

MRKShell
Server IP : 172.67.193.120 / Your IP : 172.71.28.168
Web Server : nginx/1.14.1
System : Linux comtuc2-s-2vcpu-8gb-160gb-intel-nyc3 4.18.0-348.7.1.el8_5.x86_64 #1 SMP Wed Dec 22 13:25:12 UTC 2021 x86_64
User : nginx ( 991)
PHP Version : 7.2.34
Disable Function : NONE
MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : ON | Pkexec : ON
Directory : /usr/lib/python3.6/site-packages/cloudinit/reporting/
Upload File :
[ HOME SHELL ]
Current File : /usr/lib/python3.6/site-packages/cloudinit/reporting//events.py
# Copyright (C) 2015 Canonical Ltd.
#
# This file is part of cloud-init. See LICENSE file for license information.

"""
events for reporting.

The events here are designed to be used with reporting.
They can be published to registered handlers with report_event.
"""
import base64
import os.path
import time

from . import instantiated_handler_registry, available_handlers

FINISH_EVENT_TYPE = 'finish'
START_EVENT_TYPE = 'start'

DEFAULT_EVENT_ORIGIN = 'cloudinit'


class _nameset(set):
    def __getattr__(self, name):
        if name in self:
            return name
        raise AttributeError("%s not a valid value" % name)


status = _nameset(("SUCCESS", "WARN", "FAIL"))


class ReportingEvent(object):
    """Encapsulation of event formatting."""

    def __init__(self, event_type, name, description,
                 origin=DEFAULT_EVENT_ORIGIN, timestamp=None):
        self.event_type = event_type
        self.name = name
        self.description = description
        self.origin = origin
        if timestamp is None:
            timestamp = time.time()
        self.timestamp = timestamp

    def as_string(self):
        """The event represented as a string."""
        return '{0}: {1}: {2}'.format(
            self.event_type, self.name, self.description)

    def as_dict(self):
        """The event represented as a dictionary."""
        return {'name': self.name, 'description': self.description,
                'event_type': self.event_type, 'origin': self.origin,
                'timestamp': self.timestamp}


class FinishReportingEvent(ReportingEvent):

    def __init__(self, name, description, result=status.SUCCESS,
                 post_files=None):
        super(FinishReportingEvent, self).__init__(
            FINISH_EVENT_TYPE, name, description)
        self.result = result
        if post_files is None:
            post_files = []
        self.post_files = post_files
        if result not in status:
            raise ValueError("Invalid result: %s" % result)

    def as_string(self):
        return '{0}: {1}: {2}: {3}'.format(
            self.event_type, self.name, self.result, self.description)

    def as_dict(self):
        """The event represented as json friendly."""
        data = super(FinishReportingEvent, self).as_dict()
        data['result'] = self.result
        if self.post_files:
            data['files'] = _collect_file_info(self.post_files)
        return data


def report_event(event, excluded_handler_types=None):
    """Report an event to all registered event handlers
    except those whose type is in excluded_handler_types.

    This should generally be called via one of the other functions in
    the reporting module.

    :param excluded_handler_types:
         List of handlers types to exclude from reporting the event to.
    :param event_type:
        The type of the event; this should be a constant from the
        reporting module.
    """

    if not excluded_handler_types:
        excluded_handler_types = {}
    excluded_handler_classes = {
        hndl_cls
        for hndl_type, hndl_cls in available_handlers.registered_items.items()
        if hndl_type in excluded_handler_types
    }

    handlers = instantiated_handler_registry.registered_items.items()
    for _, handler in handlers:
        if type(handler) in excluded_handler_classes:
            continue  # skip this excluded handler
        handler.publish_event(event)


def report_finish_event(event_name, event_description,
                        result=status.SUCCESS, post_files=None):
    """Report a "finish" event.

    See :py:func:`.report_event` for parameter details.
    """
    event = FinishReportingEvent(event_name, event_description, result,
                                 post_files=post_files)
    return report_event(event)


def report_start_event(event_name, event_description):
    """Report a "start" event.

    :param event_name:
        The name of the event; this should be a topic which events would
        share (e.g. it will be the same for start and finish events).

    :param event_description:
        A human-readable description of the event that has occurred.
    """
    event = ReportingEvent(START_EVENT_TYPE, event_name, event_description)
    return report_event(event)


class ReportEventStack(object):
    """Context Manager for using :py:func:`report_event`

    This enables calling :py:func:`report_start_event` and
    :py:func:`report_finish_event` through a context manager.

    :param name:
        the name of the event

    :param description:
        the event's description, passed on to :py:func:`report_start_event`

    :param message:
        the description to use for the finish event. defaults to
        :param:description.

    :param parent:
    :type parent: :py:class:ReportEventStack or None
        The parent of this event.  The parent is populated with
        results of all its children.  The name used in reporting
        is <parent.name>/<name>

    :param reporting_enabled:
        Indicates if reporting events should be generated.
        If not provided, defaults to the parent's value, or True if no parent
        is provided.

    :param result_on_exception:
        The result value to set if an exception is caught. default
        value is FAIL.
    """
    def __init__(self, name, description, message=None, parent=None,
                 reporting_enabled=None, result_on_exception=status.FAIL,
                 post_files=None):
        self.parent = parent
        self.name = name
        self.description = description
        self.message = message
        self.result_on_exception = result_on_exception
        self.result = status.SUCCESS
        if post_files is None:
            post_files = []
        self.post_files = post_files

        # use parents reporting value if not provided
        if reporting_enabled is None:
            if parent:
                reporting_enabled = parent.reporting_enabled
            else:
                reporting_enabled = True
        self.reporting_enabled = reporting_enabled

        if parent:
            self.fullname = '/'.join((parent.fullname, name,))
        else:
            self.fullname = self.name
        self.children = {}

    def __repr__(self):
        return ("ReportEventStack(%s, %s, reporting_enabled=%s)" %
                (self.name, self.description, self.reporting_enabled))

    def __enter__(self):
        self.result = status.SUCCESS
        if self.reporting_enabled:
            report_start_event(self.fullname, self.description)
        if self.parent:
            self.parent.children[self.name] = (None, None)
        return self

    def _childrens_finish_info(self):
        for cand_result in (status.FAIL, status.WARN):
            for _name, (value, _msg) in self.children.items():
                if value == cand_result:
                    return (value, self.message)
        return (self.result, self.message)

    @property
    def result(self):
        return self._result

    @result.setter
    def result(self, value):
        if value not in status:
            raise ValueError("'%s' not a valid result" % value)
        self._result = value

    @property
    def message(self):
        if self._message is not None:
            return self._message
        return self.description

    @message.setter
    def message(self, value):
        self._message = value

    def _finish_info(self, exc):
        # return tuple of description, and value
        if exc:
            return (self.result_on_exception, self.message)
        return self._childrens_finish_info()

    def __exit__(self, exc_type, exc_value, traceback):
        (result, msg) = self._finish_info(exc_value)
        if self.parent:
            self.parent.children[self.name] = (result, msg)
        if self.reporting_enabled:
            report_finish_event(self.fullname, msg, result,
                                post_files=self.post_files)


def _collect_file_info(files):
    if not files:
        return None
    ret = []
    for fname in files:
        if not os.path.isfile(fname):
            content = None
        else:
            with open(fname, "rb") as fp:
                content = base64.b64encode(fp.read()).decode()
        ret.append({'path': fname, 'content': content,
                    'encoding': 'base64'})
    return ret

# vi: ts=4 expandtab
Anon7 - 2022
AnonSec Team