Pipelines

Pipelines that enable running the various components within the library, simulating the final behavior delivered by the service.

Main pipeline

The main pipeline functions like a standard inference pipeline. To initialize the entire system, the user only needs to provide either a configuration dictionary or the configuration classes themselves. Since the pipeline inherits from CoreComponent, all CoreComponent methods are also available in the pipeline class.

If do_parsing=False, the pipeline will take already-parsed logs as input, and the parsing step will be skipped. Likewise, if do_aggregation=False, the pipeline will output the detectors’ alerts directly, and the alert aggregation step will be skipped.

The pipeline allows you to pass either a Config class instance or the dictionary used to initialize that Config class as arguments to the various components. For the detectors list, both methods can be combined, as shown in the example.

Example

from detectmatelibrary.alert_aggregation.logid_count import LogIDCountAggegationConfig

from detectmatelibrary.parsers.tree_matcher import TemplateCppTreeMatcherConfig

from detectmatelibrary.detectors.charset_detector import CharsetDetectorConfig

from detectmatelibrary.pipeline import PipelineConfig, Pipeline
import detectmatelibrary.schemas as schemas

config = PipelineConfig(
        do_parsing=True,  # Use the parser
        do_aggregation=True,  # Use the aggregator
        parser=TemplateCppTreeMatcherConfig(
            path_templates="docs/examples/data/tutorial_templates.txt",
            log_format="type=<Type> msg=audit(<Time>): <Content>"
        ),
        n_workers=1,  # You can add parallelization for the detectors
        detectors=[
            {
                "detectors": {
                    "ECVCDetector": {
                        "method_type": "ecvc_detector",
                        "data_use_configure": 100,
                        "window_size": 10,
                        "seed": 0,
                        "validation_per": 0.,
                        "threshold_method": "mean"
                    }
                }
            },
            CharsetDetectorConfig(
                data_use_configure=100,
                data_use_training=1,
                auto_config=True,
            )
        ],
        aggregator=LogIDCountAggegationConfig(max_count=1)
    )

pipeline = Pipeline(config=config)
pipeline.process(schemas.LogSchema({
    "logID": "0",
    "log": "pid=9699 uid=0 auid=4294967295 ses=4294967295 msg='op=PAM:accounting acct=\"root\"'",
}))