# Copyright (c) 2015 SUSE Linux GmbH. All rights reserved.
#
# This file is part of kiwi.
#
# kiwi is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# kiwi is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with kiwi. If not, see <http://www.gnu.org/licenses/>
#
from typing import (
Dict, Optional, List
)
import logging
from kiwi.logger_socket import PlainTextSocketHandler
import sys
# project
from kiwi.logger_color_formatter import ColorFormatter
from kiwi.logger_filter import (
LoggerSchedulerFilter,
InfoFilter,
DebugFilter,
ErrorFilter,
WarningFilter
)
from kiwi.exceptions import (
KiwiLogFileSetupFailed,
KiwiLogSocketSetupFailed
)
[docs]
class Logger(logging.Logger):
"""
**Extended logging facility based on Python logging**
:param str name: name of the logger
"""
def __init__(self, name: str):
logging.Logger.__init__(self, name)
self.log_handlers: Dict = {}
self.logfile: Optional[str] = None
# log INFO to stdout
self._add_stream_handler(
'info',
'[ %(levelname)-8s]: %(asctime)-8s | %(message)s',
[InfoFilter(), LoggerSchedulerFilter()]
)
# log WARNING messages to stdout
self._add_stream_handler(
'warning',
'[ %(levelname)-8s]: %(asctime)-8s | %(message)s',
[WarningFilter()]
)
# log DEBUG messages to stdout
self._add_stream_handler(
'debug',
'[ %(levelname)-8s]: %(asctime)-8s | %(message)s',
[DebugFilter()]
)
# log ERROR messages to stderr
self._add_stream_handler(
'error',
'[ %(levelname)-8s]: %(asctime)-8s | %(message)s',
[ErrorFilter()],
sys.__stderr__
)
self.log_level = self.level
self.log_flags: Dict[str, bool] = {}
[docs]
def getLogLevel(self) -> int:
"""
Return currently used log level
:return: log level number
:rtype: int
"""
return self.log_level
[docs]
def getLogFlags(self) -> Dict[str, bool]:
"""
Return logging flags
:return:
Dictionary with flags and their activation status
:rtype: dict
"""
return self.log_flags
[docs]
def setLogLevel(self, level: int, except_for: List[str] = []) -> None:
"""
Set custom log level for all console handlers
:param int level: log level number
"""
self.log_level = level
for handler_type in self.log_handlers:
if handler_type not in except_for:
self.log_handlers[handler_type].setLevel(level)
[docs]
def setLogFlag(self, flag: str, value: bool = True) -> None:
"""
Set logging flag for further properties of the logging facility
Available flags are:
* run-scripts-in-screen
:param str flag: name
"""
self.log_flags[flag] = value
[docs]
def set_logfile(self, filename: str) -> None:
"""
Set logfile handler
:param str filename: logfile file path
"""
try:
if filename == 'stdout':
# special case, log usual log file contents to stdout
handler = logging.StreamHandler(sys.__stdout__)
else:
handler = logging.FileHandler(
filename=filename, encoding='utf-8'
)
self.logfile = filename
handler.setFormatter(
logging.Formatter(
'[ %(levelname)-8s]: %(asctime)-8s | %(message)s',
'%H:%M:%S'
)
)
handler.addFilter(LoggerSchedulerFilter())
self.addHandler(handler)
self.log_handlers['file'] = handler
except Exception as e:
raise KiwiLogFileSetupFailed(
'%s: %s' % (type(e).__name__, format(e))
)
[docs]
def set_log_socket(self, filename: str) -> None:
"""
Set log socket handler
:param str filename:
UDS socket file path. Note if there is no server
listening on the socket the log handler setup
will fail
"""
try:
handler = PlainTextSocketHandler(filename, None)
handler.makeSocket()
handler.setFormatter(
logging.Formatter(
'[ %(levelname)-8s]: %(asctime)-8s | %(message)s',
'%H:%M:%S'
)
)
handler.addFilter(LoggerSchedulerFilter())
self.addHandler(handler)
self.log_handlers['socket'] = handler
except Exception as e:
raise KiwiLogSocketSetupFailed(
'UDS socket: {0}:{1}: {2}'.format(filename, type(e).__name__, e)
)
[docs]
def get_logfile(self) -> Optional[str]:
"""
Return file path name of logfile
:return: file path
:rtype: str
"""
return self.logfile
[docs]
@staticmethod
def progress(
current: int, total: int, prefix: str, bar_length: int = 40
) -> None:
"""
Custom progress log information. progress information is
intentionally only logged to stdout and will bypass any
handlers. We don't want this information to show up in
the log file
:param int current: current item
:param int total: total number of items
:param string prefix: prefix name
:param int bar_length: length of progress bar
"""
try:
percent = float(current) / total
except Exception:
# we don't want the progress to raise an exception
# In case of any error e.g division by zero the current
# way out is to skip the progress update
return
hashes = '#' * int(round(percent * bar_length))
spaces = ' ' * (bar_length - len(hashes))
sys.stdout.write('\r{0}: [{1}] {2}%'.format(
prefix, hashes + spaces, int(round(percent * 100))
))
if current == 100:
sys.stdout.write('\n')
sys.stdout.flush()
def _add_stream_handler(
self, handler_type, message_format, message_filter,
channel=sys.__stdout__
):
handler = logging.StreamHandler(channel)
handler.setFormatter(
logging.Formatter(message_format, '%H:%M:%S')
)
for rule in message_filter:
handler.addFilter(rule)
self.addHandler(handler)
self.log_handlers[handler_type] = handler