Python API

This page is generated from Python docstrings.

log_parser

octoprint_logmonitor.log_parser

Shared log-line parsing for the Log Monitor plugin.

Single source of truth for the log formats understood by both the live tailer (:mod:.log_tailer) and the search backend (:mod:.log_searcher), so that streaming and searching always classify lines identically.

Supported formats (tried in order):

1. Standard OctoPrint format::

   YYYY-MM-DD HH:MM:SS[,ms] - LOGGER - LEVEL - MESSAGE

2. Simple serial-log format (level inferred from message keywords)::

   YYYY-MM-DD HH:MM:SS[,ms] - MESSAGE

3. Compact format without " - " separators::

   YYYY-MM-DD HH:MM:SS[,ms]LEVEL LOGGER MESSAGE

4. Serial I/O format from the virtual printer::

   YYYY-MM-DD HH:MM:SS[,ms] >>> MESSAGE

Anything else is returned with level UNKNOWN and the raw line as the message.

parse_line(line)

Parse a raw log line into a structured dictionary.

Parameters:

Name	Type	Description	Default
line	str	Raw log line (trailing newline is stripped, tabs are normalised to four spaces).	required

Returns:

Type	Description
dict[str, Any]	Dictionary with the keys timestamp, logger, level,
dict[str, Any]	message and raw. Lines that match no known format get
dict[str, Any]	level UNKNOWN with empty timestamp/logger fields.

log_searcher

octoprint_logmonitor.log_searcher

Log Searcher Module.

Provides efficient log file searching with pagination and severity filtering. Memory-efficient implementation that streams through large log files line by line instead of loading them into memory.

LogSearcher

Efficient log file searcher with pagination support.

Features: - Memory-efficient line-by-line streaming (never loads the whole file) - Free-text search (case-insensitive) - Severity level filtering - Pagination support - Context lines (lines before/after match) - Regex search mode (optional)

__init__(logger=None)

Initialize the log searcher.

Parameters:

Name	Type	Description	Default
logger	Optional[Any]	Optional logger instance for debugging	None

export_to_csv(results)

Export search results to CSV format.

Parameters:

Name	Type	Description	Default
results	list[dict[str, Any]]	List of search result dictionaries	required

Returns:

Type	Description
str	CSV string

export_to_txt(results)

Export search results to plain text format.

Parameters:

Name	Type	Description	Default
results	list[dict[str, Any]]	List of search result dictionaries	required

Returns:

Type	Description
str	Plain text string

get_file_stats(filepath)

Get statistics about a log file.

Parameters:

Name	Type	Description	Default
filepath	str	Path to the log file	required

Returns:

Type	Description
dict[str, Any]	Dictionary with file statistics

search(filepath, query='', levels=None, offset=0, limit=50, case_sensitive=False, use_regex=False, context_lines=0)

Search log file for matching entries.

The file is streamed line by line; memory usage is bounded by limit and context_lines, not by file size. Scanning stops early once the requested page (plus one look-ahead match) is complete, so total is exact only up to that point and should be read as "at least this many matches".

Parameters:

Name	Type	Description	Default
filepath	str	Path to the log file	required
query	str	Search query (free text or regex)	''
levels	Optional[list[str]]	List of severity levels to filter by (None = all levels)	None
offset	int	Number of matches to skip (for pagination)	0
limit	int	Maximum number of results to return	50
case_sensitive	bool	Whether search should be case-sensitive	False
use_regex	bool	Whether to treat query as regex pattern	False
context_lines	int	Number of lines to include before/after each match	0

Returns:

Type	Description
dict[str, Any]	Dictionary with: - results: List of matching log entries - total: Number of matches found before scanning stopped - offset: Current offset - limit: Current limit

log_tailer

octoprint_logmonitor.log_tailer

Log Tailer Module.

Provides background thread-based log file tailing functionality. Similar to 'tail -f' behavior for live log streaming.

LogTailer

Background thread that continuously monitors a log file for new lines.

Features: - Thread-safe start/stop - Handles file rotation (rename/recreate and copy-truncate) - Parses OctoPrint log format - Calls callback for each new line

__init__(filepath, callback, poll_interval=0.5, logger=None)

Initialize the log tailer.

Parameters:

Name	Type	Description	Default
filepath	str	Path to the log file to tail	required
callback	Callable[[dict[str, Any]], None]	Function to call for each new log line (receives parsed dict)	required
poll_interval	float	Polling interval in seconds (default: 0.5)	0.5
logger	Optional[Any]	Optional logger instance for debugging	None

get_last_n_lines(n=100)

Read the last N lines from the log file.

Reads the file backwards in blocks, so only roughly the requested amount of data is loaded regardless of total file size.

Parameters:

Name	Type	Description	Default
n	int	Number of lines to read	100

Returns:

Type	Description
list	List of parsed log line dictionaries

is_running()

Check if the tailer is currently running.

start()

Start the log tailer in a background thread.

Returns:

Type	Description
bool	True if started successfully, False otherwise

stop(timeout=5.0)

Stop the log tailer gracefully.

Parameters:

Name	Type	Description	Default
timeout	float	Maximum time to wait for thread to stop	5.0

Returns:

Type	Description
bool	True if stopped successfully, False if timeout

security

octoprint_logmonitor.security

Security Utilities Module.

Centralized security helpers for the OctoPrint Log Monitor plugin. Provides path validation, input sanitization, rate limiting, and sensitive data masking to protect against common web security threats.

RateLimiter

Simple thread-safe sliding-window rate limiter.

Tracks call timestamps per client key string (e.g. an IP address or user identifier). Calls that exceed the configured quota within the sliding time window are rejected.

Example::

limiter = RateLimiter(max_calls=10, period=60.0)  # 10 req / min

if not limiter.is_allowed(client_ip):
    return flask.jsonify({"error": "Rate limit exceeded"}), 429

__init__(max_calls, period)

Initialise the rate limiter.

Parameters:

Name	Type	Description	Default
max_calls	int	Maximum number of calls allowed within period.	required
period	float	Sliding window duration in seconds.	required

cleanup()

Remove all fully-expired client entries to free memory.

is_allowed(client_key)

Check whether client_key is within the rate limit.

Calling this method counts as one request attempt regardless of the return value. Fully-expired client entries are pruned automatically once per sliding-window period, so the tracking dict cannot grow without bound.

Parameters:

Name	Type	Description	Default
client_key	str	An opaque string identifying the caller (e.g. IP).	required

Returns:

Type	Description
bool	True if the request is within quota, False if it should
bool	be rejected.

check_file_size(filepath, max_bytes=MAX_FILE_SIZE_BYTES)

Return True if filepath exists and is within max_bytes.

Files that cannot be stat-ed (missing, permissions, etc.) return False so the caller can handle the error appropriately.

Parameters:

Name	Type	Description	Default
filepath	str	Absolute path to the file.	required
max_bytes	int	Size limit in bytes. Defaults to :data:MAX_FILE_SIZE_BYTES.	MAX_FILE_SIZE_BYTES

Returns:

Type	Description
bool	True if the file is within the size limit.

is_safe_path(base_dir, filename)

Determine whether filename resolves to a path inside base_dir.

This prevents path-traversal attacks (../, absolute paths, symlink escapes, etc.). Both base_dir and the resolved candidate path are canonicalised with :func:os.path.realpath before comparison so that symbolic links cannot be used to escape the allowed directory.

Parameters:

Name	Type	Description	Default
base_dir	str	The directory that all file access must be confined to.	required
filename	str	A bare filename (no directory components expected) provided by the user / API caller.	required

Returns:

Type	Description
bool	True if and only if the resolved path is strictly inside
bool	(or equal to) base_dir.

mask_sensitive_data(text)

Apply all sensitive-data masking patterns to text.

Returns the sanitised result. This can be used for optional log-output sanitisation where administrators want to prevent passwords, API keys, or e-mail addresses from appearing in streamed log content sent to the browser.

Parameters:

Name	Type	Description	Default
text	str	The raw log text to sanitise.	required

Returns:

Type	Description
str	The text with sensitive values replaced.

validate_filename(filename)

Check that filename is a plain filename with no path components.

Rejects: * Empty strings * Names containing path separators (/ or backslash) * Names starting with . (hidden / relative traversal start) * Names that differ from os.path.basename(filename)

Parameters:

Name	Type	Description	Default
filename	str	The filename string to validate.	required

Returns:

Type	Description
bool	True if the filename is considered safe.

validate_pagination(offset, limit, max_limit=MAX_SEARCH_LIMIT)

Validate offset and limit pagination parameters.

Parameters:

Name	Type	Description	Default
offset	int	The requested result offset.	required
limit	int	The requested result count.	required
max_limit	int	Maximum permitted limit value.	MAX_SEARCH_LIMIT

Returns:

Type	Description
bool	A (valid, error_message) tuple. error_message is empty when
str	valid is True.

validate_severity_levels(levels)

Partition levels into valid and invalid severity level strings.

Parameters:

Name	Type	Description	Default
levels	Optional[list[str]]	User-supplied list of level strings (may be None).	required

Returns:

Type	Description
tuple[list[str], list[str]]	(valid_levels, invalid_levels) where each is a list of strings.

plugin core

octoprint_logmonitor

OctoPrint Log Monitor Plugin.

Provides live log streaming and searching capabilities with severity-based alerting.

LogmonitorPlugin

Bases: StartupPlugin, TemplatePlugin, SettingsPlugin, AssetPlugin, BlueprintPlugin

Main plugin class implementing OctoPrint Log Monitor functionality.

Features:

• Live log streaming via WebSocket
• Full-text log search with severity filtering
• Navbar and Sidebar indicators for severity alerts
• Configurable severity triggers

__init__()

Initialize plugin state.

clear_alert_history()

Clear alert history.

download_log_file(filename)

Download a log file directly.

export_results()

Export search results to CSV or TXT format.

frontend_debug_log()

Write frontend debug events into OctoPrint server logs.

get_active_streams()

Get list of active streaming files (multi-file support).

get_alert_history()

Get alert history.

get_alert_monitor_status()

Return current alert-monitor configuration and active files.

get_assets()

Define asset files to include in the UI.

get_log_files()

Get list of available log files.

get_settings_defaults()

Define default plugin settings.

get_template_configs()

Define template configurations for tab, navbar, and sidebar.

get_template_vars()

Provide additional template variables for rendering settings UI.

get_update_information()

Configure Software Update Plugin integration.

is_blueprint_csrf_protected()

Explicitly require CSRF protection for blueprint routes.

is_template_autoescaped()

Enable Jinja autoescaping for all plugin templates.

on_after_startup()

Initialize plugin after OctoPrint startup.

on_settings_save(data)

Persist settings and refresh background alert monitoring.

on_shutdown()

Clean shutdown of background threads.

reset_alerts()

Reset severity alert counters.

search_logs()

Search logs with pagination and severity filtering.

start_multi_stream()

Start streaming multiple log files simultaneously.

start_stream()

Start or switch log streaming.

stop_multi_stream()

Stop streaming specific log files.

stop_stream()

Stop log streaming.

write_debug_test_entries()

Write one test entry per severity category into OctoPrint logs.

__plugin_load__()

Load the plugin.