Parsers

Parsers convert unstructured raw logs into structured ParserSchema objects that downstream detectors consume.

	Schema	Description
Input	LogSchema	Unstructured log
Output	ParserSchema	Structured log

This document explains expected APIs, how to implement a parser, testing tips and common pitfalls.

Overview

• Parsers must inherit from CoreParser and provide a parse() implementation.
• CoreParser.run() handles lifecycle and calls parse() for each input; implement pure parsing logic inside parse() where possible.
• Use a typed Config class (subclass of CoreParserConfig) to hold runtime parameters.

CoreParser: minimal API

Recommended signatures and behavior:

class CoreParser:
    def run(self, input_: schemas.LogSchema, output_: schemas.ParserSchema) -> bool:
        """Top-level runner. Calls parse() and performs pre/post processing.
        Return True when a parsed output was produced, False otherwise.
        """

    def parse(self, input_: schemas.LogSchema, output_: schemas.ParserSchema) -> bool | None:
        """Implement parsing here.
        - Fill the required output_ fields (see "ParserSchema: what to populate" below).
        - Return True, or None, to emit output_ as the parsed log.
        - Return False to drop the log: process() then returns None for it.
        """

    def train(self, input_: schemas.LogSchema) -> None:
        """Optional: train internal models. Can be a no-op for stateless parsers."""

ParserSchema: what to populate

parse() only has to fill the fields that describe the match:

• eventID (int): identifier for the matched template/event
• template (string): event template text
• variables (repeated string): extracted parameters (extend the list)

CoreParser.run(), which calls parse(), fills in everything else for you:

• parserID, parserType: the parser's name and method_type
• parsedLogID: a new unique ID for the parsed record
• logID, log: copied from the input LogSchema
• headerVariables: the header fields extracted with log_format
• receivedTimestamp, parsedTimestamp: set just before and just after parse() runs

Don't set these in parse(): run() writes parsedTimestamp after parse() returns, so a value set there is overwritten.

Creating a new parser: step by step

1. Create a Config class inheriting CoreParserConfig.
2. Create parser class inheriting CoreParser.
3. Implement parse() to populate output_.
4. Add unit tests for parse() behavior.

Example:

# filepath: src/detectmatelibrary/parsers/my_parser.py
from typing import Any

from detectmatelibrary.common.parser import CoreParser, CoreParserConfig
from detectmatelibrary import schemas


class MyParserConfig(CoreParserConfig):
    method_type: str = "my_parser"
    # add parser-specific settings here
    pattern: str | None = None


class MyParser(CoreParser):
    def __init__(
        self,
        name: str = "MyParser",
        config: MyParserConfig | dict[str, Any] = MyParserConfig(),
    ):
        if isinstance(config, dict):
            config = MyParserConfig.from_dict(config, name)
        super().__init__(name=name, config=config)

    def parse(self, input_: schemas.LogSchema, output_: schemas.ParserSchema) -> bool:
        text = input_.log or ""
        # simple example: split on whitespace and treat as variables
        tokens = text.split()
        output_["eventID"] = 1
        output_["template"] = " ".join(["<*>"] * len(tokens))
        output_["variables"].extend(tokens)
        return True

Testing parsers

Unit test parse() directly using ParserSchema objects:

def test_my_parser_parse():
    from detectmatelibrary.parsers.my_parser import MyParser
    from detectmatelibrary import schemas

    parser = MyParser()
    raw = schemas.LogSchema({"log": "a b c", "logID": "1"})
    out = schemas.ParserSchema()
    ok = parser.parse(raw, out)
    assert ok is True
    assert out["eventID"] == 1
    assert out["variables"] == ["a", "b", "c"]

Available parsers

• JSON Parser: extracts structured fields from JSON-formatted logs.
• Template Matcher: matches logs against a predefined set of <*> templates.
• Template Tree Matcher: matches logs against a predefined set of <*> templates using a tree structure.
• LogBatcher Parser: LLM-based parser that infers templates from raw logs with no training data.
• Drain parser: Parser inspired by Drain Publication.
• Auto parser: The auto parser uses a brute-force strategy: it iterates through every log-type record in the internal dataset and chooses the regex and templates that best matches the provided logs.

Common parameters (all parsers)

There are some parameters, that every parser inherits from CoreParserConfig/CoreConfig/BasicConfig, regardless of what it does. The other parameters, that are specific for the respective parser, are explained right at the parsers documentation page, later on.

Top level

Field	Type	Default	Description
auto_config	boolean	False	Runs the configuration step before the training process.

params

Field	Type	Default	Description
start_id	integer	10	Number used to start the unique ID generator.
data_use_training	integer, null	None	Number of logs used for training. None skips training.
data_use_configure	integer, null	None	Number of logs used for the configuration step. None skips it.
use_config_data_as_training	boolean	True	Combine the configured data in the training process if True.
train_buffer_max_records	integer	100000	Configure records kept in memory for training (use_config_data_as_training) before the buffer spills to Parquet files on disk, in parts of this many records.
train_buffer_dir	string, null	None	Local directory for the spilled training buffer. None uses the system temp directory (TMPDIR). Each spill goes to a private detectmate-train-* directory, removed after training reads it; a killed process leaves it behind.
log_format	string, null	None	fitting description yet to find
time_format	string, null	None	fitting description yet to find