Source code
Revision control
Copy as Markdown
Other Tools
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
import json
import sys
import time
import traceback
from multiprocessing import current_process
from threading import Lock, current_thread
import six
from .logtypes import (
Any,
Boolean,
Dict,
Int,
List,
Nullable,
Status,
SubStatus,
TestId,
TestList,
Tuple,
Unicode,
convertor_registry,
log_action,
)
"""Structured Logging for recording test results.
Allowed actions, and subfields:
suite_start
tests - List of test names
name - Name for the suite
run_info - Dictionary of run properties
add_subsuite
name - Name for the subsuite (must be unique)
run_info - Updates to the suite run_info (optional)
group_start
name - Name for the test group
group_end
name - Name for the test group
suite_end
test_start
test - ID for the test
path - Relative path to test (optional)
subsuite - Name of the subsuite to which test belongs (optional)
group - Name of the test group for the incoming test (optional)
test_end
test - ID for the test
status [PASS | FAIL | OK | ERROR | TIMEOUT | CRASH |
ASSERT PRECONDITION_FAILED | SKIP] - test status
expected [As for status] - Status that the test was expected to get,
or absent if the test got the expected status
extra - Dictionary of harness-specific extra information e.g. debug info
known_intermittent - List of known intermittent statuses that should
not fail a test. eg. ['FAIL', 'TIMEOUT']
subsuite - Name of the subsuite to which test belongs (optional)
group - Name of the test group for the incoming test (optional)
test_status
test - ID for the test
subtest - Name of the subtest
status [PASS | FAIL | TIMEOUT |
PRECONDITION_FAILED | NOTRUN | SKIP] - test status
expected [As for status] - Status that the subtest was expected to get,
or absent if the subtest got the expected status
known_intermittent - List of known intermittent statuses that should
not fail a test. eg. ['FAIL', 'TIMEOUT']
subsuite - Name of the subsuite to which test belongs (optional)
group - Name of the test group for the incoming test (optional)
process_output
process - PID of the process
command - Command line of the process
data - Output data from the process
test - ID of the test that the process was running (optional)
subsuite - Name of the subsuite that the process was running (optional)
assertion_count
count - Number of assertions produced
min_expected - Minimum expected number of assertions
max_expected - Maximum expected number of assertions
subsuite - Name of the subsuite for the tests that ran (optional)
lsan_leak
frames - List of stack frames from the leak report
scope - An identifier for the set of tests run during the browser session
(e.g. a directory name)
allowed_match - A stack frame in the list that matched a rule meaning the
leak is expected
subsuite - Name of the subsuite for the tests that ran (optional)
lsan_summary
bytes - Number of bytes leaked
allocations - Number of allocations
allowed - Boolean indicating whether all detected leaks matched allow rules
subsuite - Name of the subsuite for the tests that ran (optional)
mozleak_object
process - Process that leaked
bytes - Number of bytes that leaked
name - Name of the object that leaked
scope - An identifier for the set of tests run during the browser session
(e.g. a directory name)
allowed - Boolean indicating whether the leak was permitted
subsuite - Name of the subsuite for the tests that ran (optional)
log
level [CRITICAL | ERROR | WARNING |
INFO | DEBUG] - level of the logging message
message - Message to log
Subfields for all messages:
action - the action type of the current message
time - the timestamp in ms since the epoch of the log message
thread - name for the thread emitting the message
pid - id of the python process in which the logger is running
source - name for the source emitting the message
component - name of the subcomponent emitting the message
"""
_default_logger_name = None
def get_default_logger(component=None):
"""Gets the default logger if available, optionally tagged with component
name. Will return None if not yet set
:param component: The component name to tag log messages with
"""
global _default_logger_name
if not _default_logger_name:
return None
return StructuredLogger(_default_logger_name, component=component)
def set_default_logger(default_logger):
"""Sets the default logger to logger.
It can then be retrieved with :py:func:`get_default_logger`
Note that :py:func:`~mozlog.commandline.setup_logging` will
set a default logger for you, so there should be no need to call this
function if you're using setting up logging that way (recommended).
:param default_logger: The logger to set to default.
"""
global _default_logger_name
_default_logger_name = default_logger.name
log_levels = dict(
(k.upper(), v)
for v, k in enumerate(["critical", "error", "warning", "info", "debug"])
)
lint_levels = ["ERROR", "WARNING"]
def log_actions():
"""Returns the set of actions implemented by mozlog."""
return set(convertor_registry.keys())
class LoggerShutdownError(Exception):
"""Raised when attempting to log after logger.shutdown() has been called."""
class LoggerState(object):
def __init__(self):
self.reset()
def reset(self):
self.handlers = []
self.subsuites = set()
self.running_tests = set()
self.suite_started = False
self.component_states = {}
self.has_shutdown = False
class ComponentState(object):
def __init__(self):
self.filter_ = None
class StructuredLogger(object):
_lock = Lock()
_logger_states = {}
"""Create a structured logger with the given name
:param name: The name of the logger.
:param component: A subcomponent that the logger belongs to (typically a library name)
"""
def __init__(self, name, component=None):
self.name = name
self.component = component
with self._lock:
if name not in self._logger_states:
self._logger_states[name] = LoggerState()
if component not in self._logger_states[name].component_states:
self._logger_states[name].component_states[component] = ComponentState()
self._state = self._logger_states[name]
self._component_state = self._state.component_states[component]
def add_handler(self, handler):
"""Add a handler to the current logger"""
self._state.handlers.append(handler)
def remove_handler(self, handler):
"""Remove a handler from the current logger"""
self._state.handlers.remove(handler)
def reset_state(self):
"""Resets the logger to a brand new state. This means all handlers
are removed, running tests are discarded and components are reset.
"""
self._state.reset()
self._component_state = self._state.component_states[
self.component
] = ComponentState()
def send_message(self, topic, command, *args):
"""Send a message to each handler configured for this logger. This
part of the api is useful to those users requiring dynamic control
of a handler's behavior.
:param topic: The name used by handlers to subscribe to a message.
:param command: The name of the command to issue.
:param args: Any arguments known to the target for specialized
behavior.
"""
rv = []
for handler in self._state.handlers:
if hasattr(handler, "message_handler"):
rv += handler.message_handler.handle_message(topic, command, *args)
return rv
@property
def has_shutdown(self):
"""Property indicating whether the logger has been shutdown"""
return self._state.has_shutdown
@property
def handlers(self):
"""A list of handlers that will be called when a
message is logged from this logger"""
return self._state.handlers
@property
def component_filter(self):
return self._component_state.filter_
@component_filter.setter
def component_filter(self, value):
self._component_state.filter_ = value
def log_raw(self, raw_data):
if "action" not in raw_data:
raise ValueError
action = raw_data["action"]
converted_data = convertor_registry[action].convert_known(**raw_data)
for k, v in six.iteritems(raw_data):
if (
k not in converted_data
and k not in convertor_registry[action].optional_args
):
converted_data[k] = v
data = self._make_log_data(action, converted_data)
if action in ("test_status", "test_end"):
if (
data["expected"] == data["status"]
or data["status"] == "SKIP"
or "expected" not in raw_data
):
del data["expected"]
if not self._ensure_suite_state(action, data):
return
self._handle_log(data)
def _log_data(self, action, data=None):
if data is None:
data = {}
if data.get("subsuite") and data["subsuite"] not in self._state.subsuites:
self.error(f"Unrecognised subsuite {data['subsuite']}")
return
log_data = self._make_log_data(action, data)
self._handle_log(log_data)
def _handle_log(self, data):
if self._state.has_shutdown:
raise LoggerShutdownError(
"{} action received after shutdown.".format(data["action"])
)
with self._lock:
if self.component_filter:
data = self.component_filter(data)
if data is None:
return
for handler in self.handlers:
try:
handler(data)
except Exception:
# Write the exception details directly to stderr because
# log() would call this method again which is currently locked.
print(
"%s: Failure calling log handler:" % __name__,
file=sys.__stderr__,
)
print(traceback.format_exc(), file=sys.__stderr__)
def _make_log_data(self, action, data):
all_data = {
"action": action,
"time": int(time.time() * 1000),
"thread": current_thread().name,
"pid": current_process().pid,
"source": self.name,
}
if self.component:
all_data["component"] = self.component
all_data.update(data)
return all_data
def _ensure_suite_state(self, action, data):
if action == "suite_start":
if self._state.suite_started:
# limit data to reduce unnecessary log bloat
self.error(
"Got second suite_start message before suite_end. "
+ "Logged with data: {}".format(json.dumps(data)[:100])
)
return False
self._state.suite_started = True
elif action == "suite_end":
if not self._state.suite_started:
self.error(
"Got suite_end message before suite_start. "
+ "Logged with data: {}".format(json.dumps(data))
)
return False
self._state.suite_started = False
return True
@log_action(
TestList("tests"),
Unicode("name", default=None, optional=True),
Dict(Any, "run_info", default=None, optional=True),
Dict(Any, "version_info", default=None, optional=True),
Dict(Any, "device_info", default=None, optional=True),
Dict(Any, "extra", default=None, optional=True),
)
def suite_start(self, data):
"""Log a suite_start message
:param dict tests: Test identifiers that will be run in the suite, keyed by group name.
:param str name: Optional name to identify the suite.
:param dict run_info: Optional information typically provided by mozinfo.
:param dict version_info: Optional target application version information provided
by mozversion.
:param dict device_info: Optional target device information provided by mozdevice.
"""
if not self._ensure_suite_state("suite_start", data):
return
self._log_data("suite_start", data)
@log_action(
Unicode("name"),
Dict(Any, "run_info", default=None, optional=True),
)
def add_subsuite(self, data):
"""Log a add_subsuite message
:param str name: Name to identify the subsuite.
:param dict run_info: Optional information about the subsuite. This updates the suite run_info.
"""
if data["name"] in self._state.subsuites:
return
run_info = data.get("run_info", {"subsuite": data["name"]})
if "subsuite" not in run_info:
run_info = run_info.copy()
run_info["subsuite"] = data["name"]
data["run_info"] = run_info
self._state.subsuites.add(data["name"])
self._log_data("add_subsuite", data)
@log_action(
Unicode("name"),
)
def group_start(self, data):
"""Log a group_start message
:param str name: Name to identify the test group.
"""
self._log_data("group_start", data)
@log_action(
Unicode("name"),
)
def group_end(self, data):
"""Log a group_end message
:param str name: Name to identify the test group.
"""
self._log_data("group_end", data)
@log_action(Dict(Any, "extra", default=None, optional=True))
def suite_end(self, data):
"""Log a suite_end message"""
if not self._ensure_suite_state("suite_end", data):
return
self._state.subsuites.clear()
self._log_data("suite_end", data)
@log_action(
TestId("test"),
Unicode("path", default=None, optional=True),
Unicode("subsuite", default=None, optional=True),
Unicode("group", default=None, optional=True),
)
def test_start(self, data):
"""Log a test_start message
:param test: Identifier of the test that will run.
:param path: Path to test relative to some base (typically the root of
the source tree).
:param subsuite: Optional name of the subsuite to which the test belongs.
:param group: Optional name of the test group or manifest name (useful
when running in paralle)
"""
if not self._state.suite_started:
self.error(
"Got test_start message before suite_start for test %s" % data["test"]
)
return
test_key = (data.get("subsuite"), data["test"])
if test_key in self._state.running_tests:
self.error("test_start for %s logged while in progress." % data["test"])
return
self._state.running_tests.add(test_key)
self._log_data("test_start", data)
@log_action(
TestId("test"),
Unicode("subtest"),
SubStatus("status"),
SubStatus("expected", default="PASS"),
Unicode("message", default=None, optional=True),
Unicode("stack", default=None, optional=True),
Dict(Any, "extra", default=None, optional=True),
List(SubStatus, "known_intermittent", default=None, optional=True),
Unicode("subsuite", default=None, optional=True),
Unicode("group", default=None, optional=True),
)
def test_status(self, data):
"""
Log a test_status message indicating a subtest result. Tests that
do not have subtests are not expected to produce test_status messages.
:param test: Identifier of the test that produced the result.
:param subtest: Name of the subtest.
:param status: Status string indicating the subtest result
:param expected: Status string indicating the expected subtest result.
:param message: Optional string containing a message associated with the result.
:param stack: Optional stack trace encountered during test execution.
:param extra: Optional suite-specific data associated with the test result.
:param known_intermittent: Optional list of string expected intermittent statuses
:param subsuite: Optional name of the subsuite to which the test belongs.
:param group: Optional name of the test group or manifest name (useful
when running in paralle)
"""
if data["expected"] == data["status"] or data["status"] == "SKIP":
del data["expected"]
test_key = (data.get("subsuite"), data["test"])
if test_key not in self._state.running_tests:
self.error(
"test_status for %s logged while not in progress. "
"Logged with data: %s" % (data["test"], json.dumps(data))
)
return
self._log_data("test_status", data)
@log_action(
TestId("test"),
Status("status"),
Status("expected", default="OK"),
Unicode("message", default=None, optional=True),
Unicode("stack", default=None, optional=True),
Dict(Any, "extra", default=None, optional=True),
List(Status, "known_intermittent", default=None, optional=True),
Unicode("subsuite", default=None, optional=True),
Unicode("group", default=None, optional=True),
)
def test_end(self, data):
"""
Log a test_end message indicating that a test completed. For tests
with subtests this indicates whether the overall test completed without
errors. For tests without subtests this indicates the test result
directly.
:param test: Identifier of the test that produced the result.
:param status: Status string indicating the test result
:param expected: Status string indicating the expected test result.
:param message: Optonal string containing a message associated with the result.
:param stack: Optional stack trace encountered during test execution.
:param extra: Optional suite-specific data associated with the test result.
:param subsuite: Optional name of the subsuite to which the test belongs.
:param group: Optional name of the test group or manifest name (useful
when running in paralle)
"""
if data["expected"] == data["status"] or data["status"] == "SKIP":
del data["expected"]
test_key = (data.get("subsuite"), data["test"])
if test_key not in self._state.running_tests:
self.error(
"test_end for %s logged while not in progress. "
"Logged with data: %s" % (data["test"], json.dumps(data))
)
else:
self._state.running_tests.remove(test_key)
self._log_data("test_end", data)
@log_action(
Unicode("process"),
Unicode("data"),
Unicode("command", default=None, optional=True),
TestId("test", default=None, optional=True),
Unicode("subsuite", default=None, optional=True),
)
def process_output(self, data):
"""Log output from a managed process.
:param process: A unique identifier for the process producing the output
(typically the pid)
:param data: The output to log
:param command: Optional string representing the full command line used to start
the process.
:param test: Optional ID of the test which the process was running.
:param subsuite: Optional name of the subsuite which the process was running.
"""
self._log_data("process_output", data)
@log_action(
Unicode("process", default=None),
Unicode("signature", default="[Unknown]"),
TestId("test", default=None, optional=True),
Unicode("minidump_path", default=None, optional=True),
Unicode("minidump_extra", default=None, optional=True),
Int("stackwalk_retcode", default=None, optional=True),
Unicode("stackwalk_stdout", default=None, optional=True),
Unicode("stackwalk_stderr", default=None, optional=True),
Unicode("reason", default=None, optional=True),
Unicode("java_stack", default=None, optional=True),
Unicode("process_type", default=None, optional=True),
List(Unicode, "stackwalk_errors", default=None),
Unicode("subsuite", default=None, optional=True),
)
def crash(self, data):
if data["stackwalk_errors"] is None:
data["stackwalk_errors"] = []
self._log_data("crash", data)
@log_action(Unicode("group", default=None), Unicode("message", default=None))
def shutdown_failure(self, data):
self._log_data("shutdown_failure", data)
@log_action(
Unicode("primary", default=None), List(Unicode, "secondary", default=None)
)
def valgrind_error(self, data):
self._log_data("valgrind_error", data)
@log_action(
Unicode("process"),
Unicode("command", default=None, optional=True),
Unicode("subsuite", default=None, optional=True),
)
def process_start(self, data):
"""Log start event of a process.
:param process: A unique identifier for the process producing the
output (typically the pid)
:param command: Optional string representing the full command line used to
start the process.
:param subsuite: Optional name of the subsuite using the process.
"""
self._log_data("process_start", data)
@log_action(
Unicode("process"),
Int("exitcode"),
Unicode("command", default=None, optional=True),
Unicode("subsuite", default=None, optional=True),
)
def process_exit(self, data):
"""Log exit event of a process.
:param process: A unique identifier for the process producing the
output (typically the pid)
:param exitcode: the exit code
:param command: Optional string representing the full command line used to
start the process.
:param subsuite: Optional name of the subsuite using the process.
"""
self._log_data("process_exit", data)
@log_action(
TestId("test"),
Int("count"),
Int("min_expected"),
Int("max_expected"),
Unicode("subsuite", default=None, optional=True),
)
def assertion_count(self, data):
"""Log count of assertions produced when running a test.
:param count: Number of assertions produced
:param min_expected: Minimum expected number of assertions
:param max_expected: Maximum expected number of assertions
:param subsuite: Optional name of the subsuite for the tests that ran
"""
self._log_data("assertion_count", data)
@log_action(
List(Unicode, "frames"),
Unicode("scope", optional=True, default=None),
Unicode("allowed_match", optional=True, default=None),
Unicode("subsuite", default=None, optional=True),
)
def lsan_leak(self, data):
self._log_data("lsan_leak", data)
@log_action(
Int("bytes"),
Int("allocations"),
Boolean("allowed", optional=True, default=False),
Unicode("subsuite", default=None, optional=True),
)
def lsan_summary(self, data):
self._log_data("lsan_summary", data)
@log_action(
Unicode("process"),
Int("bytes"),
Unicode("name"),
Unicode("scope", optional=True, default=None),
Boolean("allowed", optional=True, default=False),
Unicode("subsuite", default=None, optional=True),
)
def mozleak_object(self, data):
self._log_data("mozleak_object", data)
@log_action(
Unicode("process"),
Nullable(Int, "bytes"),
Int("threshold"),
List(Unicode, "objects"),
Unicode("scope", optional=True, default=None),
Boolean("induced_crash", optional=True, default=False),
Boolean("ignore_missing", optional=True, default=False),
Unicode("subsuite", default=None, optional=True),
)
def mozleak_total(self, data):
self._log_data("mozleak_total", data)
@log_action()
def shutdown(self, data):
"""Shutdown the logger.
This logs a 'shutdown' action after which any further attempts to use
the logger will raise a :exc:`LoggerShutdownError`.
This is also called implicitly from the destructor or
when exiting the context manager.
"""
self._log_data("shutdown", data)
self._state.has_shutdown = True
def __enter__(self):
return self
def __exit__(self, exc, val, tb):
self.shutdown()
def _log_func(level_name):
@log_action(Unicode("message"), Any("exc_info", default=False))
def log(self, data):
exc_info = data.pop("exc_info", None)
if exc_info:
if not isinstance(exc_info, tuple):
exc_info = sys.exc_info()
if exc_info != (None, None, None):
bt = traceback.format_exception(*exc_info)
data["stack"] = "\n".join(bt)
data["level"] = level_name
self._log_data("log", data)
log.__doc__ = (
"""Log a message with level %s
:param message: The string message to log
:param exc_info: Either a boolean indicating whether to include a traceback
derived from sys.exc_info() or a three-item tuple in the
same format as sys.exc_info() containing exception information
to log.
"""
% level_name
)
log.__name__ = str(level_name).lower()
return log
def _lint_func(level_name):
@log_action(
Unicode("path"),
Unicode("message", default=""),
Int("lineno", default=0),
Int("column", default=None, optional=True),
Unicode("hint", default=None, optional=True),
Unicode("source", default=None, optional=True),
Unicode("rule", default=None, optional=True),
Tuple((Int, Int), "lineoffset", default=None, optional=True),
Unicode("linter", default=None, optional=True),
)
def lint(self, data):
data["level"] = level_name
self._log_data("lint", data)
lint.__doc__ = """Log an error resulting from a failed lint check
:param linter: name of the linter that flagged this error
:param path: path to the file containing the error
:param message: text describing the error
:param lineno: line number that contains the error
:param column: column containing the error
:param hint: suggestion for fixing the error (optional)
:param source: source code context of the error (optional)
:param rule: name of the rule that was violated (optional)
:param lineoffset: denotes an error spans multiple lines, of the form
(<lineno offset>, <num lines>) (optional)
"""
lint.__name__ = str("lint_%s" % level_name)
return lint
# Create all the methods on StructuredLog for log/lint levels
for level_name in log_levels:
setattr(StructuredLogger, level_name.lower(), _log_func(level_name))
for level_name in lint_levels:
level_name = level_name.lower()
name = "lint_%s" % level_name
setattr(StructuredLogger, name, _lint_func(level_name))
class StructuredLogFileLike(object):
"""Wrapper for file-like objects to redirect writes to logger
instead. Each call to `write` becomes a single log entry of type `log`.
When using this it is important that the callees i.e. the logging
handlers do not themselves try to write to the wrapped file as this
will cause infinite recursion.
:param logger: `StructuredLogger` to which to redirect the file write operations.
:param level: log level to use for each write.
:param prefix: String prefix to prepend to each log entry.
"""
def __init__(self, logger, level="info", prefix=None):
self.logger = logger
self.log_func = getattr(self.logger, level)
self.prefix = prefix
def write(self, data):
if data.endswith("\n"):
data = data[:-1]
if data.endswith("\r"):
data = data[:-1]
if self.prefix is not None:
data = "%s: %s" % (self.prefix, data)
self.log_func(data)
def flush(self):
pass