repository·master·Indexed 21 days ago
https://github.com/logpai/drain3A persistent and streaming log template miner that extracts structured templates from unstructured log streams in real-time using a fixed-depth parse tree. It supports real-time learning, masking of variable parts (like IPs and emails), and state persistence via Kafka, Redis, File, or Memory. The library provides both Training mode for continuous learning and Inference mode for matching logs against fixed templates.
Drain3 provides two distinct modes depending on whether you want to learn new templates or just match against existing ones.
Use this when you want the model to continuously learn from a stream. It will match logs to existing clusters or create/update clusters as needed.
template_miner.add_log_message(log_line)Use this when you want to match logs against a fixed set of already-learned templates without modifying the model. No new clusters are created, and existing templates are not changed. A match must be perfect; otherwise, it returns None.
template_miner.match(log_line)Masking allows you to replace specific variable parts of a log message (like IPs, numbers, or emails) with keywords before they reach the Drain core. This prevents the miner from creating too many unique templates for similar messages.
Custom masking is defined using a list of regular expressions in the format {'regex_pattern', 'mask_with'}. Any parameter that does not match a custom mask is replaced by the default <*> by the Drain core.
[MASKING]
masking = [
{"regex_pattern":"((?<=[^A-Za-z0-9])|^)(\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\.\\d{1,3})((?=[^A-Za-z0-9])|$)", "mask_with": "IP"},
{"regex_pattern":"((?<=[^A-Za-z0-9])|^)([\\-\+]?\\d+)((?=[^A-Za-z0-9])|$)", "mask_with": "NUM"},
]Drain3 can save and load snapshots of its state (the search tree and all identified clusters) in JSON format. This allows the miner to continue learning from where it left off after a restart.
Snapshots are triggered by:
cluster_created: When a new template is identified.cluster_template_changed: When an existing template is updated.periodic: After the interval specified by snapshot_interval_minutes.topic_name and supports standard Kafka kwargs (e.g., bootstrap_servers).Developers can extend persistence to other databases by inheriting from the PersistenceHandler class.
You can install the core Drain3 package using pip3.
Note on Persistence: If you intend to use Kafka or Redis for persistence, you must install their respective client libraries explicitly as they are optional dependencies.
pip3 install drain3
# If using Kafka persistence:
pip3 install kafka-python
# If using Redis persistence:
pip3 install redisTo run the provided examples from the repository, you must first install the dependencies using pipenv:
python3 -m pipenv syncRun examples/drain_stdin_demo.py to test Drain3 using input from stdin. This demo supports various persistence modes (Kafka, file, or none). You can switch modes by changing the persistence_type variable in the script.
q to exit 'online learn-and-match' mode.q again to finish.Run examples/drain_bigfile_demo.py to process a real-world SSH server log file. This demo prints the resulting clusters, the prefix tree, and performance statistics.
# Install dependencies
python3 -m pipenv sync
# Run stdin demo
python3 -m pipenv run python -m examples.drain_stdin_demo
# Run big file demo
python3 -m pipenv run python -m examples.drain_bigfile_demoThe similarity between a log sequence (seq2) and a template sequence (seq1) is calculated using the Jaccard index: len(set(seq1) & set(seq2)) / len(set(seq1) | set(seq2)).
Key implementation details:
include_params is True, the param_str (wildcard tokens) are excluded from the set intersection and union calculation.1.3 is applied to the result (capped at 1.0) to adjust the similarity score.1.0.# Internal logic representation of the similarity calculation
ret_val = len(set(seq1) & set(seq2)) / len(set(seq1) | set(seq2))
ret_val = ret_val * 1.3 if ret_val * 1.3 < 1 else 1Drain3 uses configparser for configuration. By default, it looks for a drain3.ini file in the working directory. Alternatively, you can pass a TemplateMinerConfig object directly to the TemplateMiner constructor.
sim_th: Similarity threshold. If the percentage of similar tokens is below this value, a new cluster is created. (Default: 0.4)depth: Maximum depth levels of log clusters. Minimum is 3. (Default: 4)max_children: Maximum number of children for an internal node. (Default: 100)max_clusters: Maximum number of tracked clusters. When reached, the model uses an LRU (Least Recently Used) eviction policy to replace old clusters. (Default: unlimited)extra_delimiters: Additional delimiters used when splitting log messages into words (e.g., ['_', ':']).masking: Parameters for masking, provided in JSON format.mask_prefix: The prefix used for identified parameters in templates. (Default: <)mask_suffix: The suffix used for identified parameters in templates. (Default: >)snapshot_interval_minutes: Time interval for creating new snapshots. (Default: 1)compress_state: Whether to compress the state before saving (useful for Kafka persistence).Drain3 can automatically treat tokens containing digits as parameters (wildcards). This is controlled by the parametrize_numeric_tokens flag in the Drain constructor.
When enabled, if a token contains a digit, the parser may use the param_str (default <*>) to represent it in the template, allowing for more generalized matching of logs containing IDs, IP addresses, or timestamps.
When SimpleProfiler.report() is called, it prints a summary of all tracked sections. The output for each section follows this pattern:
{section_name}: took {total_time} s ({% of total}), {sample_count} samples, {ms_per_1k} ms / 1000 samples, {hz} hz
12.34 s). If an enclosing_section_name is provided, it also shows the percentage of that section's time.If reset_after_sample_count is used, batch statistics are also shown in parentheses next to the total statistics.
Drain3 allows you to retrieve an ordered list of variables from a log message after its template has been mined. Each extracted parameter is an ExtractedParameter object containing the value and the mask_name (the name of the mask that matched, or * for the catch-all mask).
By default, exact_matching=True uses the regular expressions defined in your masking instructions to find variables. If you disable exact matching, every variable is matched against a non-whitespace character sequence, which can improve performance at the cost of accuracy.
To optimize performance, regexes generated per template are cached. You can adjust the cache size using the MASKING/parameter_extraction_cache_capacity configuration parameter.
result = template_miner.add_log_message(log_line)
params = template_miner.extract_parameters(
result["template_mined"],
log_line,
exact_matching=True
)The following attributes define the behavior of the template mining engine. When loading from a file, these correspond to specific sections and keys in a .ini formatted configuration file.
engine: The mining engine to use (e.g., Drain).extra_delimiters: A collection of strings used as additional delimiters.sim_th: Similarity threshold for matching log messages to existing templates.depth: The maximum depth of the parsing tree.max_children: The maximum number of children allowed per node in the tree.max_clusters: The maximum number of clusters (templates) to maintain. If None, there is no limit.parametrize_numeric_tokens: Boolean indicating whether numeric tokens should be automatically parameterized.enabled: Boolean to enable or disable profiling.report_sec: Interval in seconds for generating profiling reports.snapshot_interval_minutes: Interval in minutes for taking state snapshots.compress_state: Boolean indicating whether to compress the state during snapshots.masking: A JSON-formatted list of masking instructions. Each instruction is a dictionary containing regex_pattern and mask_with.mask_prefix: The prefix string used for masked parameters.mask_suffix: The suffix string used for masked parameters.parameter_extraction_cache_capacity: The capacity of the cache used for parameter extraction.[DRAIN]
engine = Drain
sim_th = 0.4
depth = 4
max_children = 100
max_clusters = 1000
parametrize_numeric_tokens = True
extra_delimiters = [' ', ' ', ':']
[PROFILING]
enabled = True
report_sec = 60
[SNAPSHOT]
snapshot_interval_minutes = 5
compress_state = True
[MASKING]
mask_prefix = <
mask_suffix = >
parameter_extraction_cache_capacity = 3000
masking = [{"regex_pattern": "\\d+", "mask_with": "<NUM>"}]