blob: 1d18ad81875386181c25d73966a0c7703cec57d8 [file] [log] [blame]
#!/usr/bin/env python3
#
# Copyright 2022 The Fuchsia Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import datetime
import logging
import os
import re
from copy import copy
from antlion import tracelogger
from antlion.libs.logging import log_stream
from antlion.libs.logging.log_stream import LogStyles
log_line_format = "%(asctime)s.%(msecs).03d %(levelname)s %(message)s"
# The micro seconds are added by the format string above,
# so the time format does not include ms.
log_line_time_format = "%Y-%m-%d %H:%M:%S"
log_line_timestamp_len = 23
logline_timestamp_re = re.compile("\d\d\d\d-\d\d-\d\d \d\d:\d\d:\d\d.\d\d\d")
# yapf: disable
class Style:
RESET = '\033[0m'
BRIGHT = '\033[1m'
DIM = '\033[2m'
NORMAL = '\033[22m'
class Fore:
BLACK = '\033[30m'
RED = '\033[31m'
GREEN = '\033[32m'
YELLOW = '\033[33m'
BLUE = '\033[34m'
MAGENTA = '\033[35m'
CYAN = '\033[36m'
WHITE = '\033[37m'
RESET = '\033[39m'
class Back:
BLACK = '\033[40m'
RED = '\033[41m'
GREEN = '\033[42m'
YELLOW = '\033[43m'
BLUE = '\033[44m'
MAGENTA = '\033[45m'
CYAN = '\033[46m'
WHITE = '\033[47m'
RESET = '\033[49m'
LOG_LEVELS = {
'DEBUG': {'level': 10, 'style': Fore.GREEN + Style.BRIGHT},
'CASE': {'level': 11, 'style': Back.BLUE + Fore.WHITE + Style.BRIGHT},
'SUITE': {'level': 12, 'style': Back.MAGENTA + Fore.WHITE + Style.BRIGHT},
'INFO': {'level': 20, 'style': Style.NORMAL},
'STEP': {'level': 15, 'style': Fore.WHITE + Style.BRIGHT},
'WARNING': {'level': 30, 'style': Fore.YELLOW + Style.BRIGHT},
'ERROR': {'level': 40, 'style': Fore.RED + Style.BRIGHT},
'EXCEPTION': {'level': 45, 'style': Back.RED + Fore.WHITE + Style.BRIGHT},
'DEVICE': {'level': 51, 'style': Fore.CYAN + Style.BRIGHT},
}
# yapf: enable
class ColoredLogFormatter(logging.Formatter):
def format(self, record):
colored_record = copy(record)
level_name = colored_record.levelname
style = LOG_LEVELS[level_name]["style"]
formatted_level_name = "%s%s%s" % (style, level_name, Style.RESET)
colored_record.levelname = formatted_level_name
return super().format(colored_record)
def _parse_logline_timestamp(t):
"""Parses a logline timestamp into a tuple.
Args:
t: Timestamp in logline format.
Returns:
An iterable of date and time elements in the order of month, day, hour,
minute, second, microsecond.
"""
date, time = t.split(" ")
year, month, day = date.split("-")
h, m, s = time.split(":")
s, ms = s.split(".")
return year, month, day, h, m, s, ms
def is_valid_logline_timestamp(timestamp):
if len(timestamp) == log_line_timestamp_len:
if logline_timestamp_re.match(timestamp):
return True
return False
def logline_timestamp_comparator(t1, t2):
"""Comparator for timestamps in logline format.
Args:
t1: Timestamp in logline format.
t2: Timestamp in logline format.
Returns:
-1 if t1 < t2; 1 if t1 > t2; 0 if t1 == t2.
"""
dt1 = _parse_logline_timestamp(t1)
dt2 = _parse_logline_timestamp(t2)
for u1, u2 in zip(dt1, dt2):
if u1 < u2:
return -1
elif u1 > u2:
return 1
return 0
def _get_timestamp(time_format, delta=None):
t = datetime.datetime.now()
if delta:
t = t + datetime.timedelta(seconds=delta)
return t.strftime(time_format)[:-3]
def epoch_to_log_line_timestamp(epoch_time):
"""Converts an epoch timestamp in ms to log line timestamp format, which
is readable for humans.
Args:
epoch_time: integer, an epoch timestamp in ms.
Returns:
A string that is the corresponding timestamp in log line timestamp
format.
"""
s, ms = divmod(epoch_time, 1000)
d = datetime.datetime.fromtimestamp(s)
return d.strftime("%Y-%m-%d %H:%M:%S.") + str(ms)
def get_log_line_timestamp(delta=None):
"""Returns a timestamp in the format used by log lines.
Default is current time. If a delta is set, the return value will be
the current time offset by delta seconds.
Args:
delta: Number of seconds to offset from current time; can be negative.
Returns:
A timestamp in log line format with an offset.
"""
return _get_timestamp("%Y-%m-%d %H:%M:%S.%f", delta)
def get_log_file_timestamp(delta=None):
"""Returns a timestamp in the format used for log file names.
Default is current time. If a delta is set, the return value will be
the current time offset by delta seconds.
Args:
delta: Number of seconds to offset from current time; can be negative.
Returns:
A timestamp in log file name format with an offset.
"""
return _get_timestamp("%Y-%m-%d_%H-%M-%S-%f", delta)
def _setup_test_logger(log_path, prefix=None):
"""Customizes the root logger for a test run.
The logger object has a stream handler and a file handler. The stream
handler logs INFO level to the terminal, the file handler logs DEBUG
level to files.
Args:
log_path: Location of the log file.
prefix: A prefix for each log line in terminal.
"""
logging.log_path = log_path
log_styles = [
LogStyles.LOG_INFO + LogStyles.TO_STDOUT,
LogStyles.DEFAULT_LEVELS + LogStyles.TESTCASE_LOG,
]
terminal_format = log_line_format
if prefix:
terminal_format = "[{}] {}".format(prefix, log_line_format)
stream_formatter = ColoredLogFormatter(terminal_format, log_line_time_format)
file_formatter = logging.Formatter(log_line_format, log_line_time_format)
log = log_stream.create_logger(
"test_run",
"",
log_styles=log_styles,
stream_format=stream_formatter,
file_format=file_formatter,
)
log.setLevel(logging.DEBUG)
_enable_additional_log_levels()
def _enable_additional_log_levels():
"""Enables logging levels used for tracing tests and debugging devices."""
for log_type, log_data in LOG_LEVELS.items():
logging.addLevelName(log_data["level"], log_type)
def kill_test_logger(logger):
"""Cleans up a test logger object by removing all of its handlers.
Args:
logger: The logging object to clean up.
"""
for h in list(logger.handlers):
logger.removeHandler(h)
if isinstance(h, logging.FileHandler):
h.close()
def create_latest_log_alias(actual_path):
"""Creates a symlink to the latest test run logs.
Args:
actual_path: The source directory where the latest test run's logs are.
"""
link_path = os.path.join(os.path.dirname(actual_path), "latest")
if os.path.islink(link_path):
os.remove(link_path)
try:
os.symlink(actual_path, link_path)
except OSError:
logging.warning("Failed to create symlink to latest logs dir.", exc_info=True)
def setup_test_logger(log_path, prefix=None):
"""Customizes the root logger for a test run.
Args:
log_path: Location of the report file.
prefix: A prefix for each log line in terminal.
filename: Name of the files. The default is the time the objects
are requested.
"""
os.makedirs(log_path, exist_ok=True)
_setup_test_logger(log_path, prefix)
create_latest_log_alias(log_path)
def normalize_log_line_timestamp(log_line_timestamp):
"""Replace special characters in log line timestamp with normal characters.
Args:
log_line_timestamp: A string in the log line timestamp format. Obtained
with get_log_line_timestamp.
Returns:
A string representing the same time as input timestamp, but without
special characters.
"""
norm_tp = log_line_timestamp.replace(" ", "_")
norm_tp = norm_tp.replace(":", "-")
return norm_tp
class LoggerAdapter(logging.LoggerAdapter):
"""A LoggerAdapter class that takes in a lambda for transforming logs."""
def __init__(self, logging_lambda):
self.logging_lambda = logging_lambda
super(LoggerAdapter, self).__init__(logging.getLogger(), {})
def process(self, msg, kwargs):
return self.logging_lambda(msg), kwargs
def create_logger(logging_lambda=lambda message: message):
"""Returns a logger with logging defined by a given lambda.
Args:
logging_lambda: A lambda of the form:
>>> lambda log_message: return 'string'
"""
return tracelogger.TraceLogger(LoggerAdapter(logging_lambda))
def create_tagged_trace_logger(tag=""):
"""Returns a logger that logs each line with the given prefix.
Args:
tag: The tag of the log line, E.g. if tag == tag123, the output
line would be:
<TESTBED> <TIME> <LOG_LEVEL> [tag123] logged message
"""
def logging_lambda(msg):
return "[%s] %s" % (tag, msg)
return create_logger(logging_lambda)