kiwi.logger

  1# Copyright (c) 2015 SUSE Linux GmbH.  All rights reserved.
  2#
  3# This file is part of kiwi.
  4#
  5# kiwi is free software: you can redistribute it and/or modify
  6# it under the terms of the GNU General Public License as published by
  7# the Free Software Foundation, either version 3 of the License, or
  8# (at your option) any later version.
  9#
 10# kiwi is distributed in the hope that it will be useful,
 11# but WITHOUT ANY WARRANTY; without even the implied warranty of
 12# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 13# GNU General Public License for more details.
 14#
 15# You should have received a copy of the GNU General Public License
 16# along with kiwi.  If not, see <http://www.gnu.org/licenses/>
 17#
 18from typing import (
 19    Dict, Optional, List
 20)
 21import logging
 22from kiwi.logger_socket import PlainTextSocketHandler
 23import sys
 24
 25# project
 26from kiwi.logger_color_formatter import ColorFormatter
 27from kiwi.logger_filter import (
 28    LoggerSchedulerFilter,
 29    InfoFilter,
 30    DebugFilter,
 31    ErrorFilter,
 32    WarningFilter
 33)
 34
 35from kiwi.exceptions import (
 36    KiwiLogFileSetupFailed,
 37    KiwiLogSocketSetupFailed
 38)
 39
 40
 41class Logger(logging.Logger):
 42    """
 43    **Extended logging facility based on Python logging**
 44
 45    :param str name: name of the logger
 46    """
 47    def __init__(self, name: str):
 48        logging.Logger.__init__(self, name)
 49        self.log_handlers: Dict = {}
 50        self.logfile: Optional[str] = None
 51        # log INFO to stdout
 52        self._add_stream_handler(
 53            'info',
 54            '[ %(levelname)-8s]: %(asctime)-8s | %(message)s',
 55            [InfoFilter(), LoggerSchedulerFilter()]
 56        )
 57        # log WARNING messages to stdout
 58        self._add_stream_handler(
 59            'warning',
 60            '[ %(levelname)-8s]: %(asctime)-8s | %(message)s',
 61            [WarningFilter()]
 62        )
 63        # log DEBUG messages to stdout
 64        self._add_stream_handler(
 65            'debug',
 66            '[ %(levelname)-8s]: %(asctime)-8s | %(message)s',
 67            [DebugFilter()]
 68        )
 69        # log ERROR messages to stderr
 70        self._add_stream_handler(
 71            'error',
 72            '[ %(levelname)-8s]: %(asctime)-8s | %(message)s',
 73            [ErrorFilter()],
 74            sys.__stderr__
 75        )
 76        self.log_level = self.level
 77        self.log_flags: Dict[str, bool] = {}
 78
 79    def getLogLevel(self) -> int:
 80        """
 81        Return currently used log level
 82
 83        :return: log level number
 84
 85        :rtype: int
 86        """
 87        return self.log_level
 88
 89    def getLogFlags(self) -> Dict[str, bool]:
 90        """
 91        Return logging flags
 92
 93        :return:
 94            Dictionary with flags and their activation status
 95
 96        :rtype: dict
 97        """
 98        return self.log_flags
 99
100    def setLogLevel(
101        self, level: int, except_for: List[str] = [], only_for: List[str] = []
102    ) -> None:
103        """
104        Set custom log level for all console handlers
105
106        :param int level: log level number
107        :param list except_for:
108            set log level to all handlers except for the given list
109        :param list only_for:
110            set log level to the given handlers only
111
112        if both except_for and only_for handlers are specified,
113        the except_for list will be ignored
114        """
115        self.log_level = level
116        if only_for:
117            for handler_type in self.log_handlers:
118                if handler_type in only_for:
119                    self.log_handlers[handler_type].setLevel(level)
120            return
121        for handler_type in self.log_handlers:
122            if handler_type not in except_for:
123                self.log_handlers[handler_type].setLevel(level)
124
125    def setLogFlag(self, flag: str, value: bool = True) -> None:
126        """
127        Set logging flag for further properties of the logging facility
128        Available flags are:
129
130        * run-scripts-in-screen
131
132        :param str flag: name
133        """
134        self.log_flags[flag] = value
135
136    def set_color_format(self) -> None:
137        """
138        Set color format for all console handlers
139        """
140        for handler_type in self.log_handlers:
141            message_format = None
142            if handler_type == 'debug':
143                message_format = \
144                    '$LIGHTCOLOR[ %(levelname)-8s]: %(asctime)-8s | %(message)s'
145            elif handler_type == 'warning' or handler_type == 'error':
146                message_format = \
147                    '$COLOR[ %(levelname)-8s]: %(asctime)-8s | %(message)s'
148
149            if message_format:
150                self.log_handlers[handler_type].setFormatter(
151                    ColorFormatter(message_format, '%H:%M:%S')
152                )
153
154    def set_logfile(self, filename: str) -> None:
155        """
156        Set logfile handler
157
158        :param str filename: logfile file path
159        """
160        try:
161            if filename == 'stdout':
162                # special case, log usual log file contents to stdout
163                handler = logging.StreamHandler(sys.__stdout__)
164            else:
165                handler = logging.FileHandler(
166                    filename=filename, encoding='utf-8'
167                )
168                self.logfile = filename
169            handler.setFormatter(
170                logging.Formatter(
171                    '[ %(levelname)-8s]: %(asctime)-8s | %(message)s',
172                    '%H:%M:%S'
173                )
174            )
175            handler.addFilter(LoggerSchedulerFilter())
176            self.addHandler(handler)
177            self.log_handlers['file'] = handler
178        except Exception as e:
179            raise KiwiLogFileSetupFailed(
180                f'{type(e).__name__}: {format(e)}'
181            )
182
183    def set_log_socket(self, filename: str) -> None:
184        """
185        Set log socket handler
186
187        :param str filename:
188            UDS socket file path. Note if there is no server
189            listening on the socket the log handler setup
190            will fail
191        """
192        try:
193            handler = PlainTextSocketHandler(filename, None)
194            handler.makeSocket()
195            handler.setFormatter(
196                logging.Formatter(
197                    '[ %(levelname)-8s]: %(asctime)-8s | %(message)s',
198                    '%H:%M:%S'
199                )
200            )
201            handler.addFilter(LoggerSchedulerFilter())
202            self.addHandler(handler)
203            self.log_handlers['socket'] = handler
204        except Exception as e:
205            raise KiwiLogSocketSetupFailed(
206                'UDS socket: {0}:{1}: {2}'.format(filename, type(e).__name__, e)
207            )
208
209    def get_logfile(self) -> Optional[str]:
210        """
211        Return file path name of logfile
212
213        :return: file path
214
215        :rtype: str
216        """
217        return self.logfile
218
219    @staticmethod
220    def progress(
221        current: int, total: int, prefix: str, bar_length: int = 40
222    ) -> None:
223        """
224        Custom progress log information. progress information is
225        intentionally only logged to stdout and will bypass any
226        handlers. We don't want this information to show up in
227        the log file
228
229        :param int current: current item
230        :param int total: total number of items
231        :param string prefix: prefix name
232        :param int bar_length: length of progress bar
233        """
234        try:
235            percent = float(current) / total
236        except Exception:
237            # we don't want the progress to raise an exception
238            # In case of any error e.g division by zero the current
239            # way out is to skip the progress update
240            return
241        hashes = '#' * int(round(percent * bar_length))
242        spaces = ' ' * (bar_length - len(hashes))
243        sys.stdout.write('\r{0}: [{1}] {2}%'.format(
244            prefix, hashes + spaces, int(round(percent * 100))
245        ))
246        if current == 100:
247            sys.stdout.write('\n')
248        sys.stdout.flush()
249
250    def _add_stream_handler(
251        self, handler_type, message_format, message_filter,
252        channel=sys.__stdout__
253    ):
254        handler = logging.StreamHandler(channel)
255        handler.setFormatter(
256            logging.Formatter(message_format, '%H:%M:%S')
257        )
258        for rule in message_filter:
259            handler.addFilter(rule)
260        self.addHandler(handler)
261        self.log_handlers[handler_type] = handler