16388870099

Committed 19 Jul 2025 12:51PM UTC coverage: 72.024% (-0.1%) from 72.129%

Build # 16388870099

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1825 from pbiering/add-trace-logging-feature

Add trace logging feature

Coverage Stats

2024 of 2959 branches covered (68.4%)

Branch coverage included in aggregate %.

16 of 31 new or added lines in 4 files covered. (51.61%)

4685 of 6356 relevant lines covered (73.71%)

12.12 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

64.41

/radicale/log.py

# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2011-2017 Guillaume Ayoub
# Copyright © 2017-2023 Unrud <unrud@outlook.com>
# Copyright © 2024-2024 Peter Bieringer <pb@bieringer.de>
#
# This library 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.
#
# This library 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 Radicale.  If not, see <http://www.gnu.org/licenses/>.

"""
Functions to set up Python's logging facility for Radicale's WSGI application.

Log messages are sent to the first available target of:

  - Error stream specified by the WSGI server in "wsgi.errors"
  - ``sys.stderr``

"""

import contextlib
import io
import logging
import os
import socket
import struct
import sys
import threading
import time
from typing import (Any, Callable, ClassVar, Dict, Iterator, Mapping, Optional,
                    Tuple, Union, cast)

from radicale import types

LOGGER_NAME: str = "radicale"
LOGGER_FORMATS: Mapping[str, str] = {
    "verbose": "[%(asctime)s] [%(ident)s] [%(levelname)s] %(message)s",
    "journal": "[%(ident)s] [%(levelname)s] %(message)s",
}
DATE_FORMAT: str = "%Y-%m-%d %H:%M:%S %z"

logger: logging.Logger = logging.getLogger(LOGGER_NAME)


class RemoveTracebackFilter(logging.Filter):

    def filter(self, record: logging.LogRecord) -> bool:
        record.exc_info = None
        return True


class RemoveTRACEFilter(logging.Filter):

    def filter(self, record: logging.LogRecord) -> bool:
        if record.msg.startswith("TRACE"):
            return False
        else:
            return True


class PassTRACETOKENFilter(logging.Filter):
    def __init__(self, trace_filter: str):
        super().__init__()
        self.trace_filter = trace_filter
        self.prefix = "TRACE/" + self.trace_filter

    def filter(self, record: logging.LogRecord) -> bool:
        if record.msg.startswith("TRACE"):
            if record.msg.startswith(self.prefix):
                return True
            else:
                return False
        else:
            return True


REMOVE_TRACEBACK_FILTER: logging.Filter = RemoveTracebackFilter()

REMOVE_TRACE_FILTER: logging.Filter = RemoveTRACEFilter()


class IdentLogRecordFactory:
    """LogRecordFactory that adds ``ident`` attribute."""

    def __init__(self, upstream_factory: Callable[..., logging.LogRecord]
                 ) -> None:
        self._upstream_factory = upstream_factory

    def __call__(self, *args: Any, **kwargs: Any) -> logging.LogRecord:
        record = self._upstream_factory(*args, **kwargs)
        ident = ("%d" % record.process if record.process is not None
                 else record.processName or "unknown")
        tid = None
        if record.thread is not None:
            if record.thread != threading.main_thread().ident:
                ident += "/%s" % (record.threadName or "unknown")
            if (sys.version_info >= (3, 8) and
                    record.thread == threading.get_ident()):
                try:
                    tid = threading.get_native_id()
                except AttributeError:
                    # so far function not existing e.g. on SunOS
                    # see also https://docs.python.org/3/library/threading.html#threading.get_native_id
                    tid = None

        record.ident = ident  # type:ignore[attr-defined]
        record.tid = tid  # type:ignore[attr-defined]
        return record


class ThreadedStreamHandler(logging.Handler):
    """Sends logging output to the stream registered for the current thread or
       ``sys.stderr`` when no stream was registered."""

    terminator: ClassVar[str] = "\n"

    _streams: Dict[int, types.ErrorStream]
    _journal_stream_id: Optional[Tuple[int, int]]
    _journal_socket: Optional[socket.socket]
    _journal_socket_failed: bool
    _formatters: Mapping[str, logging.Formatter]
    _formatter: Optional[logging.Formatter]

    def __init__(self, format_name: Optional[str] = None) -> None:
        super().__init__()
        self._streams = {}
        self._journal_stream_id = None
        with contextlib.suppress(TypeError, ValueError):
            dev, inode = os.environ.get("JOURNAL_STREAM", "").split(":", 1)
            self._journal_stream_id = (int(dev), int(inode))
        self._journal_socket = None
        self._journal_socket_failed = False
        self._formatters = {name: logging.Formatter(fmt, DATE_FORMAT)
                            for name, fmt in LOGGER_FORMATS.items()}
        self._formatter = (self._formatters[format_name]
                           if format_name is not None else None)

    def _get_formatter(self, default_format_name: str) -> logging.Formatter:
        return self._formatter or self._formatters[default_format_name]

    def _detect_journal(self, stream: types.ErrorStream) -> bool:
        if not self._journal_stream_id or not isinstance(stream, io.IOBase):
            return False
        try:
            stat = os.fstat(stream.fileno())
        except OSError:
            return False
        return self._journal_stream_id == (stat.st_dev, stat.st_ino)

    @staticmethod
    def _encode_journal(data: Mapping[str, Optional[Union[str, int]]]
                        ) -> bytes:
        msg = b""
        for key, value in data.items():
            if value is None:
                continue
            keyb = key.encode()
            valueb = str(value).encode()
            if b"\n" in valueb:
                msg += (keyb + b"\n" +
                        struct.pack("<Q", len(valueb)) + valueb + b"\n")
            else:
                msg += keyb + b"=" + valueb + b"\n"
        return msg

    def _try_emit_journal(self, record: logging.LogRecord) -> bool:
        if not self._journal_socket:
            # Try to connect to systemd journal socket
            if self._journal_socket_failed or not hasattr(socket, "AF_UNIX"):
                return False
            journal_socket = None
            try:
                journal_socket = socket.socket(
                    socket.AF_UNIX, socket.SOCK_DGRAM)
                journal_socket.connect("/run/systemd/journal/socket")
            except OSError as e:
                self._journal_socket_failed = True
                if journal_socket:
                    journal_socket.close()
                # Log after setting `_journal_socket_failed` to prevent loop!
                logger.error("Failed to connect to systemd journal: %s",
                             e, exc_info=True)
                return False
            self._journal_socket = journal_socket

        priority = {"DEBUG": 7,
                    "INFO": 6,
                    "WARNING": 4,
                    "ERROR": 3,
                    "CRITICAL": 2}.get(record.levelname, 4)
        timestamp = time.strftime("%Y-%m-%dT%H:%M:%S.%%03dZ",
                                  time.gmtime(record.created)) % record.msecs
        data = {"PRIORITY": priority,
                "TID": cast(Optional[int], getattr(record, "tid", None)),
                "SYSLOG_IDENTIFIER": record.name,
                "SYSLOG_FACILITY": 1,
                "SYSLOG_PID": record.process,
                "SYSLOG_TIMESTAMP": timestamp,
                "CODE_FILE": record.pathname,
                "CODE_LINE": record.lineno,
                "CODE_FUNC": record.funcName,
                "MESSAGE": self._get_formatter("journal").format(record)}
        self._journal_socket.sendall(self._encode_journal(data))
        return True

    def emit(self, record: logging.LogRecord) -> None:
        try:
            stream = self._streams.get(threading.get_ident(), sys.stderr)
            if self._detect_journal(stream) and self._try_emit_journal(record):
                return
            msg = self._get_formatter("verbose").format(record)
            stream.write(msg + self.terminator)
            stream.flush()
        except Exception:
            self.handleError(record)

    @types.contextmanager
    def register_stream(self, stream: types.ErrorStream) -> Iterator[None]:
        """Register stream for logging output of the current thread."""
        key = threading.get_ident()
        self._streams[key] = stream
        try:
            yield
        finally:
            del self._streams[key]


@types.contextmanager
def register_stream(stream: types.ErrorStream) -> Iterator[None]:
    """Register stream for logging output of the current thread."""
    yield


def setup() -> None:
    """Set global logging up."""
    global register_stream
    format_name = os.environ.get("RADICALE_LOG_FORMAT") or None
    sane_format_name = format_name if format_name in LOGGER_FORMATS else None
    handler = ThreadedStreamHandler(sane_format_name)
    logging.basicConfig(handlers=[handler])
    register_stream = handler.register_stream
    log_record_factory = IdentLogRecordFactory(logging.getLogRecordFactory())
    logging.setLogRecordFactory(log_record_factory)
    set_level(logging.INFO, True)
    if format_name != sane_format_name:
        logger.error("Invalid RADICALE_LOG_FORMAT: %r", format_name)


logger_display_backtrace_disabled: bool = False
logger_display_backtrace_enabled: bool = False


def set_level(level: Union[int, str], backtrace_on_debug: bool, trace_on_debug: bool = False, trace_filter: str = "") -> None:
    """Set logging level for global logger."""
    global logger_display_backtrace_disabled
    global logger_display_backtrace_enabled
    if isinstance(level, str):
        level = getattr(logging, level.upper())
        assert isinstance(level, int)
    logger.setLevel(level)
    if level > logging.DEBUG:
        if logger_display_backtrace_disabled is False:
            logger.info("Logging of backtrace is disabled in this loglevel")
            logger_display_backtrace_disabled = True
        logger.addFilter(REMOVE_TRACEBACK_FILTER)
    else:
        if not backtrace_on_debug:
            if logger_display_backtrace_disabled is False:
                logger.debug("Logging of backtrace is disabled by option in this loglevel")
                logger_display_backtrace_disabled = True
            logger.addFilter(REMOVE_TRACEBACK_FILTER)
        else:
            if logger_display_backtrace_enabled is False:
                logger.debug("Logging of backtrace is enabled by option in this loglevel")
                logger_display_backtrace_enabled = True
            logger.removeFilter(REMOVE_TRACEBACK_FILTER)
        if trace_on_debug:
            if trace_filter != "":
                logger.debug("Logging messages starting with 'TRACE/%s' enabled", trace_filter)
                logger.addFilter(PassTRACETOKENFilter(trace_filter))
                logger.removeFilter(REMOVE_TRACE_FILTER)
            else:
                logger.debug("Logging messages starting with 'TRACE' enabled")
                logger.removeFilter(REMOVE_TRACE_FILTER)
        else:
            logger.debug("Logging messages starting with 'TRACE' disabled")
            logger.addFilter(REMOVE_TRACE_FILTER)

1	# This file is part of Radicale - CalDAV and CardDAV server
2	# Copyright © 2011-2017 Guillaume Ayoub
3	# Copyright © 2017-2023 Unrud <unrud@outlook.com>
4	# Copyright © 2024-2024 Peter Bieringer <pb@bieringer.de>
5	#
6	# This library is free software: you can redistribute it and/or modify
7	# it under the terms of the GNU General Public License as published by
8	# the Free Software Foundation, either version 3 of the License, or
9	# (at your option) any later version.
10	#
11	# This library is distributed in the hope that it will be useful,
12	# but WITHOUT ANY WARRANTY; without even the implied warranty of
13	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14	# GNU General Public License for more details.
15	#
16	# You should have received a copy of the GNU General Public License
17	# along with Radicale. If not, see <http://www.gnu.org/licenses/>.
18
19	"""	3✔
20	Functions to set up Python's logging facility for Radicale's WSGI application.
21
22	Log messages are sent to the first available target of:
23
24	- Error stream specified by the WSGI server in "wsgi.errors"
25	- ``sys.stderr``
26
27	"""
28
29	import contextlib	17✔
30	import io	17✔
31	import logging	17✔
32	import os	17✔
33	import socket	17✔
34	import struct	17✔
35	import sys	17✔
36	import threading	17✔
37	import time	17✔
38	from typing import (Any, Callable, ClassVar, Dict, Iterator, Mapping, Optional,	17✔
39	Tuple, Union, cast)
40
41	from radicale import types	17✔
42
43	LOGGER_NAME: str = "radicale"	17✔
44	LOGGER_FORMATS: Mapping[str, str] = {	17✔
45	"verbose": "[%(asctime)s] [%(ident)s] [%(levelname)s] %(message)s",
46	"journal": "[%(ident)s] [%(levelname)s] %(message)s",
47	}
48	DATE_FORMAT: str = "%Y-%m-%d %H:%M:%S %z"	17✔
49
50	logger: logging.Logger = logging.getLogger(LOGGER_NAME)	17✔
51
52
53	class RemoveTracebackFilter(logging.Filter):	17✔
54
55	def filter(self, record: logging.LogRecord) -> bool:	17✔
56	record.exc_info = None	12✔
57	return True	12✔
58
59
60	class RemoveTRACEFilter(logging.Filter):	17✔
61
62	def filter(self, record: logging.LogRecord) -> bool:	17✔
63	if record.msg.startswith("TRACE"):	12✔
64	return False	12✔
65	else:
66	return True	12✔
67
68
69	class PassTRACETOKENFilter(logging.Filter):	17✔
70	def __init__(self, trace_filter: str):	17✔
NEW 71	super().__init__()	×
NEW 72	self.trace_filter = trace_filter	×
NEW 73	self.prefix = "TRACE/" + self.trace_filter	×
74
75	def filter(self, record: logging.LogRecord) -> bool:	17✔
NEW 76	if record.msg.startswith("TRACE"):	×
NEW 77	if record.msg.startswith(self.prefix):	×
NEW 78	return True	×
79	else:
NEW 80	return False	×
81	else:
NEW 82	return True	×
83
84
85	REMOVE_TRACEBACK_FILTER: logging.Filter = RemoveTracebackFilter()	17✔
86
87	REMOVE_TRACE_FILTER: logging.Filter = RemoveTRACEFilter()	17✔
88
89
90	class IdentLogRecordFactory:	17✔
91	"""LogRecordFactory that adds ``ident`` attribute."""
92
93	def __init__(self, upstream_factory: Callable[..., logging.LogRecord]	17✔
94	) -> None:
95	self._upstream_factory = upstream_factory	12✔
96
97	def __call__(self, args: Any, *kwargs: Any) -> logging.LogRecord:	17✔
98	record = self._upstream_factory(args, *kwargs)	12✔
99	ident = ("%d" % record.process if record.process is not None	12✔
100	else record.processName or "unknown")
101	tid = None	12✔
102	if record.thread is not None:	12!
103	if record.thread != threading.main_thread().ident:	12✔
104	ident += "/%s" % (record.threadName or "unknown")	12✔
105	if (sys.version_info >= (3, 8) and	12!
106	record.thread == threading.get_ident()):
107	try:	12✔
108	tid = threading.get_native_id()	12✔
109	except AttributeError:	×
110	# so far function not existing e.g. on SunOS
111	# see also https://docs.python.org/3/library/threading.html#threading.get_native_id
112	tid = None	×
113
114	record.ident = ident # type:ignore[attr-defined]	12✔
115	record.tid = tid # type:ignore[attr-defined]	12✔
116	return record	12✔
117
118
119	class ThreadedStreamHandler(logging.Handler):	17✔
120	"""Sends logging output to the stream registered for the current thread or
121	``sys.stderr`` when no stream was registered."""
122
123	terminator: ClassVar[str] = "\n"	17✔
124
125	_streams: Dict[int, types.ErrorStream]	17✔
126	_journal_stream_id: Optional[Tuple[int, int]]	17✔
127	_journal_socket: Optional[socket.socket]	17✔
128	_journal_socket_failed: bool	17✔
129	_formatters: Mapping[str, logging.Formatter]	17✔
130	_formatter: Optional[logging.Formatter]	17✔
131
132	def __init__(self, format_name: Optional[str] = None) -> None:	17✔
133	super().__init__()	12✔
134	self._streams = {}	12✔
135	self._journal_stream_id = None	12✔
136	with contextlib.suppress(TypeError, ValueError):	12✔
137	dev, inode = os.environ.get("JOURNAL_STREAM", "").split(":", 1)	12✔
138	self._journal_stream_id = (int(dev), int(inode))	4✔
139	self._journal_socket = None	12✔
140	self._journal_socket_failed = False	12✔
141	self._formatters = {name: logging.Formatter(fmt, DATE_FORMAT)	12!
142	for name, fmt in LOGGER_FORMATS.items()}
143	self._formatter = (self._formatters[format_name]	12✔
144	if format_name is not None else None)
145
146	def _get_formatter(self, default_format_name: str) -> logging.Formatter:	17✔
147	return self._formatter or self._formatters[default_format_name]	12✔
148
149	def _detect_journal(self, stream: types.ErrorStream) -> bool:	17✔
150	if not self._journal_stream_id or not isinstance(stream, io.IOBase):	12!
151	return False	12✔
152	try:	×
153	stat = os.fstat(stream.fileno())	×
154	except OSError:	×
155	return False	×
156	return self._journal_stream_id == (stat.st_dev, stat.st_ino)	×
157
158	@staticmethod	17✔
159	def _encode_journal(data: Mapping[str, Optional[Union[str, int]]]	17✔
160	) -> bytes:
161	msg = b""	×
162	for key, value in data.items():	×
163	if value is None:	×
164	continue	×
165	keyb = key.encode()	×
166	valueb = str(value).encode()	×
167	if b"\n" in valueb:	×
168	msg += (keyb + b"\n" +	×
169	struct.pack("<Q", len(valueb)) + valueb + b"\n")
170	else:
171	msg += keyb + b"=" + valueb + b"\n"	×
172	return msg	×
173
174	def _try_emit_journal(self, record: logging.LogRecord) -> bool:	17✔
175	if not self._journal_socket:	×
176	# Try to connect to systemd journal socket
177	if self._journal_socket_failed or not hasattr(socket, "AF_UNIX"):	×
178	return False	×
179	journal_socket = None	×
180	try:	×
181	journal_socket = socket.socket(	×
182	socket.AF_UNIX, socket.SOCK_DGRAM)
183	journal_socket.connect("/run/systemd/journal/socket")	×
184	except OSError as e:	×
185	self._journal_socket_failed = True	×
186	if journal_socket:	×
187	journal_socket.close()	×
188	# Log after setting `_journal_socket_failed` to prevent loop!
189	logger.error("Failed to connect to systemd journal: %s",	×
190	e, exc_info=True)
191	return False	×
192	self._journal_socket = journal_socket	×
193
194	priority = {"DEBUG": 7,	×
195	"INFO": 6,
196	"WARNING": 4,
197	"ERROR": 3,
198	"CRITICAL": 2}.get(record.levelname, 4)
199	timestamp = time.strftime("%Y-%m-%dT%H:%M:%S.%%03dZ",	×
200	time.gmtime(record.created)) % record.msecs
201	data = {"PRIORITY": priority,	×
202	"TID": cast(Optional[int], getattr(record, "tid", None)),
203	"SYSLOG_IDENTIFIER": record.name,
204	"SYSLOG_FACILITY": 1,
205	"SYSLOG_PID": record.process,
206	"SYSLOG_TIMESTAMP": timestamp,
207	"CODE_FILE": record.pathname,
208	"CODE_LINE": record.lineno,
209	"CODE_FUNC": record.funcName,
210	"MESSAGE": self._get_formatter("journal").format(record)}
211	self._journal_socket.sendall(self._encode_journal(data))	×
212	return True	×
213
214	def emit(self, record: logging.LogRecord) -> None:	17✔
215	try:	12✔
216	stream = self._streams.get(threading.get_ident(), sys.stderr)	12✔
217	if self._detect_journal(stream) and self._try_emit_journal(record):	12!
218	return	×
219	msg = self._get_formatter("verbose").format(record)	12✔
220	stream.write(msg + self.terminator)	12✔
221	stream.flush()	12✔
222	except Exception:	×
223	self.handleError(record)	×
224
225	@types.contextmanager	17✔
226	def register_stream(self, stream: types.ErrorStream) -> Iterator[None]:	17✔
227	"""Register stream for logging output of the current thread."""
228	key = threading.get_ident()	12✔
229	self._streams[key] = stream	12✔
230	try:	12✔
231	yield	12✔
232	finally:
233	del self._streams[key]	12✔
234
235
236	@types.contextmanager	17✔
237	def register_stream(stream: types.ErrorStream) -> Iterator[None]:	17✔
238	"""Register stream for logging output of the current thread."""
239	yield	17✔
240
241
242	def setup() -> None:	17✔
243	"""Set global logging up."""
244	global register_stream
245	format_name = os.environ.get("RADICALE_LOG_FORMAT") or None	12✔
246	sane_format_name = format_name if format_name in LOGGER_FORMATS else None	12✔
247	handler = ThreadedStreamHandler(sane_format_name)	12✔
248	logging.basicConfig(handlers=[handler])	12✔
249	register_stream = handler.register_stream	12✔
250	log_record_factory = IdentLogRecordFactory(logging.getLogRecordFactory())	12✔
251	logging.setLogRecordFactory(log_record_factory)	12✔
252	set_level(logging.INFO, True)	12✔
253	if format_name != sane_format_name:	12!
254	logger.error("Invalid RADICALE_LOG_FORMAT: %r", format_name)	×
255
256
257	logger_display_backtrace_disabled: bool = False	17✔
258	logger_display_backtrace_enabled: bool = False	17✔
259
260
261	def set_level(level: Union[int, str], backtrace_on_debug: bool, trace_on_debug: bool = False, trace_filter: str = "") -> None:	17✔
262	"""Set logging level for global logger."""
263	global logger_display_backtrace_disabled
264	global logger_display_backtrace_enabled
265	if isinstance(level, str):	12✔
266	level = getattr(logging, level.upper())	12✔
267	assert isinstance(level, int)	12✔
268	logger.setLevel(level)	12✔
269	if level > logging.DEBUG:	12✔
270	if logger_display_backtrace_disabled is False:	12!
271	logger.info("Logging of backtrace is disabled in this loglevel")	12✔
272	logger_display_backtrace_disabled = True	12✔
273	logger.addFilter(REMOVE_TRACEBACK_FILTER)	12✔
274	else:
275	if not backtrace_on_debug:	12✔
276	if logger_display_backtrace_disabled is False:	12!
277	logger.debug("Logging of backtrace is disabled by option in this loglevel")	×
278	logger_display_backtrace_disabled = True	×
279	logger.addFilter(REMOVE_TRACEBACK_FILTER)	12✔
280	else:
281	if logger_display_backtrace_enabled is False:	12!
282	logger.debug("Logging of backtrace is enabled by option in this loglevel")	12✔
283	logger_display_backtrace_enabled = True	12✔
284	logger.removeFilter(REMOVE_TRACEBACK_FILTER)	12✔
285	if trace_on_debug:	12!
NEW 286	if trace_filter != "":	×
NEW 287	logger.debug("Logging messages starting with 'TRACE/%s' enabled", trace_filter)	×
NEW 288	logger.addFilter(PassTRACETOKENFilter(trace_filter))	×
NEW 289	logger.removeFilter(REMOVE_TRACE_FILTER)	×
290	else:
NEW 291	logger.debug("Logging messages starting with 'TRACE' enabled")	×
NEW 292	logger.removeFilter(REMOVE_TRACE_FILTER)	×
293	else:
294	logger.debug("Logging messages starting with 'TRACE' disabled")	12✔
295	logger.addFilter(REMOVE_TRACE_FILTER)	12✔

Kozea / Radicale / 16388870099

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous