Drain parser

The parser is derived from the official Drain publication.

It also wraps functionality from the DetectMatePerformance project: https://github.com/ait-detectmate/DetectMatePerformance. When parsing large numbers of log lines in non-stream (batch) mode, it is recommended to use the performance-oriented implementation.

Schema Description
Input LogSchema Unstructured log
Output ParserSchema Structured log

Configuration

Drain parser parameters:

  • method_type (string): identifier for the parser type (e.g., "tree_matcher").
  • depth (int): number of token/word levels.
  • max_childs (int): maximum number of children allowed in the given layer.
  • sim_thres (float): threshold used for similarity.
  • reset_in_post_train (bool): if enabled, clears logs from the training buffer once templates are created; otherwise, it keeps them for the next training cycle.
  • auto_config (bool): indicates whether to run an optional auto-configuration step (not mandatory).

Example YAML fragment: 

parsers:
  DrainParser:
    method_type: drain_parser
    auto_config: False
    params:
      depth: 2

Usage example

Simple usage (Reset = False):

from detectmatelibrary.parsers.drain import DrainParser
from detectmatelibrary import schemas

# instantiate parser (config can be a dict or a config object)
config_dict = {
    "parsers": {
        "DrainParser": {
            "method_type": "drain_parser",
            "data_use_training": 2,
            "reset_in_post_train": False,
        }
    }
}

parser = DrainParser(config=config_dict)

parsed = parser.process(schemas.LogSchema({"log": "hello there, general kenobi!"}))
print(parsed["template"])  # "templates not yet generated"

parsed = parser.process(schemas.LogSchema({"log": "hello there, captain kenobi!"}))
print(parsed["template"])  # "templates not yet generated"

parsed = parser.process(schemas.LogSchema({"log": "hello there, sargent kenobi!"}))
print(parsed["template"])  # "hello there <*> kenobi"

parser.update_state("keep_training")
parser.process(schemas.LogSchema({"log": "bella ciao bella ciao"}))
parser.update_state("stop_training")

parsed = parser.process(schemas.LogSchema({"log": "hello there, sargent kenobi!"}))
print(parsed["template"])  # "hello there <*> kenobi"

Simple usage (Reset = True):

from detectmatelibrary.parsers.drain import DrainParser
from detectmatelibrary import schemas

# instantiate parser (config can be a dict or a config object)
config_dict = {
    "parsers": {
        "DrainParser": {
            "method_type": "drain_parser",
            "data_use_training": 2,
            "reset_in_post_train": False,
        }
    }
}

parser = DrainParser(config=config_dict)

parsed = parser.process(schemas.LogSchema({"log": "hello there, general kenobi!"}))
print(parsed["template"])  # "templates not yet generated"

parsed = parser.process(schemas.LogSchema({"log": "hello there, captain kenobi!"}))
print(parsed["template"])  # "templates not yet generated"

parsed = parser.process(schemas.LogSchema({"log": "hello there, sargent kenobi!"}))
print(parsed["template"])  # "hello there <*> kenobi"

parser.update_state("keep_training")
parser.process(schemas.LogSchema({"log": "bella ciao bella ciao"}))
parser.update_state("stop_training")

parsed = parser.process(schemas.LogSchema({"log": "hello there, sargent kenobi!"}))
print(parsed["template"])  # "template not found"

Go back to Index