diff --git a/README.md b/README.md index 72a284da..20ae0647 100644 --- a/README.md +++ b/README.md @@ -1,106 +1,24 @@ -# DetectMateLibrary - -Main library to run the different components in DetectMate. - -## Main structure - -The library contains the next components: - -* **Parsers**: parse the logs received from the reader. -* **Detectors**: return alerts if anomalies are detected. -* **Alert Aggregation**: aggregate the alerts produced by the detectors. -* **Schemas**: standard data classes used in DetectMate. -``` - +--------+ +-----------+ +-------------------+ - | Parser | --> | Detector | -> | Alert Aggregation | - +--------+ +-----------+ +-------------------+ -``` -## Developer setup: - -### Step 1: Install python dependencies -Set up the dev environment and install pre-commit hooks: - -```bash -uv sync --dev -uv run prek install -``` - -### Step 2: Install Protobuf dependencies - -To install in Linux do: - -```bash -sudo apt install -y protobuf-compiler -protoc --version -``` - -This dependency is only needed if a proto file is modified. To compile the proto file do: -```bash -protoc --proto_path=src/detectmatelibrary/schemas/ --python_out=src/detectmatelibrary/schemas/ src/detectmatelibrary/schemas/schemas.proto -``` -### Step 3: Run unit tests -Run the tests: - -```bash -uv run --dev pytest -q -``` - -Run the tests with coverage (add --cov-report=html to generate an HTML report): - -```bash -uv run --dev pytest --cov=. --cov-report=term-missing -``` - -## Workspace generator (`mate create`) - -DetectMateLibrary includes a small CLI helper to bootstrap standalone workspaces -for custom parsers and detectors. This is useful if you want to develop and test -components in isolation while still using the same library and schemas. - -### Usage - -The CLI entry point is `mate` with a `create` command: - -```bash -mate create --type --name --dir -``` +# DetectMateLibrary -| Option | Description | -|----------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| -| `--type` | Component type to generate:
- `parser`: CoreParser-based template
- `detector`: CoreDetector-based template | -| `--name` | Name of the component and package:
- Creates package dir: `//`
- Creates main file: `.py`
- Derives class names: `` and `Config` | -| `--dir` | Directory where the workspace will be created | +![UML](docs/img/logo.png) +DetectMate is a flexible, component-based library for parsing and log anomaly detection. -### What gets generated +## Installation -For example: +Download us from `Pip`!! ```bash -mate create --type parser --name custom_parser --dir ./workspaces/custom_parser -``` - -will create: - -```text -workspaces/custom_parser/ # workspace root -├── custom_parser/ # Python package -│ ├── __init__.py -│ └── custom_parser.py # CoreParser-based template -├── tests/ -│ └── test_custom_parser.py # generated from template to test custom_parser -├── data.json # example data to run the code -├── LICENSE.md # copied from main project -├── .gitignore # copied from main project -├── .pre-commit-config.yaml # copied from main project -├── pyproject.toml # minimal project + dev extras -└── README.md # setup instructions +pip install detectmatelibrary ``` ## Documentation -- [Project Documentation](https://ait-detectmate.github.io/DetectMateLibrary/latest/) +Main documentations: + +- [Library Documentation](https://ait-detectmate.github.io/DetectMateLibrary/latest/) (this repo) +- [Service Documentation](https://github.com/ait-detectmate/DetectMateService/tree/main) ## Contribution diff --git a/docs/alert_aggregator.md b/docs/alert_aggregator.md index cc177559..c0deb337 100644 --- a/docs/alert_aggregator.md +++ b/docs/alert_aggregator.md @@ -19,6 +19,7 @@ This document explains the expected API and how to implement an alert aggregator With the default `buffer_size` of 1, it is called for every alert with a list of one. - Use a typed `Config` class (subclass of `CoreAlertAggregatorConfig`) to hold runtime parameters. + ## CoreAlertAggregator: minimal API ```python @@ -62,6 +63,35 @@ fields left for `aggregate_alerts()` to fill are `description` and `alertDetails unless your `aggregate_alerts()` keeps it, including their `detectedTimestamp` and `receivedTimestamp`. -## Available alert aggregators +## Alert aggregator methods + +The alert aggregator are numbered from simplest to most complex, and the sidebar lists them in the same order. + +| # | Detector | What it detects | +|---|---|---| +| 00 | [Basic Concat Aggregator](alert_aggregators/basic_concatenation.md) | Concat a set of alerts. | +| 01 | [LogID Count Aggregator](alert_aggregators/logid_count.md) | Count the amount of times a logID is in alerts. | + + +### Common parameters (all alert aggregators) + +There are some parameters, that **every** alert aggregator inherits from `CoreAlertAggregatorConfig`/`CoreConfigCoreAlertAggregator`/`BasicConfig`, regardless of what it does. The other parameters, that are **specific** for the respective detector, are explained right at the detectors documentation page, later on. + + +???+ note "Top level" + + | Field | Type | Default | Description | + |---|---|---|---| + | `auto_config` | boolean | False | Runs the configuration step before the training process. | + +???+ note "params" -- [Basic Concat](alert_aggregators/basic_concatenation.md) + | Field | Type | Default | Description | + |---|---|---|---| + | `start_id` | integer | 10 | Number used to start the unique ID generator. | + | `data_use_training` | integer, null | None | Data used for training, if None, training is not done. | + | `data_use_configure` | integer, null | None | Data used for configuration, if None, configuration is not done. | + | `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. | + diff --git a/docs/alert_aggregators/basic_concatenation.md b/docs/alert_aggregators/basic_concatenation.md index d7c60c1c..6334c7ef 100644 --- a/docs/alert_aggregators/basic_concatenation.md +++ b/docs/alert_aggregators/basic_concatenation.md @@ -2,25 +2,59 @@ The basic concat alert aggregation approach aggregates x number of alerts into one output schema. +## In/out + | | Schema | Description | |------------|----------------------------|--------------------| | **Input** | [DetectorSchema](../schemas.md) | Alerts from detectors | | **Output** | [AggregateSchema](../schemas.md) | Aggregated alerts | +## At a glance -# Configuration example +| Learns from training data | Auto-configuration | Needs `events` | Federation | +|---|---|---|---| +| ❌ | ❌ | ❌ | ❌ | -```yaml -alert_aggregators: - BasicConcatAggregator: - method_type: "basic_concat_aggregator" - buffer_size: 3 - deduplicate_values: False - auto_config: False -``` +## Description + +A basic approach that simply concatenates the alerts without verifying anything specific. -## Example usage + +## Example ```python --8<-- "docs/examples/alert_aggregators/basic_concat.py:example_1" ``` + +## Configuration file + +The configuration used by the example above. It sets only what this use case needs; every other parameter keeps its default (see [Configuration arguments](#configuration-arguments)). + +```yaml +--8<-- "docs/examples/alert_aggregators/basic_concat.yaml" +``` + +## Configuration arguments + +All parameters this detector accepts, grouped by the YAML block they go in. **Scope** tells whether a parameter is `specific` to this detector or `shared` with other detectors (see the [Alert Aggregation overview](../alert_aggregator.md#common-parameters-all-alert-aggregators)). Where this detector changes a shared default, the shared value is shown in brackets. + + +??? note "Top level" + + | Field | Type | Default | Scope | Description | + |---|---|---|---|---| + | `method_type` | string | basic_concat_aggregator | shared | Indicates what type of method it is. | + | `auto_config` | boolean | False | shared | Runs the configuration step before the training process. | + +???+ note "params" + + | Field | Type | Default | Scope | Description | + |---|---|---|---|---| + | `buffer_size` | integer | 3 | specific | Buffer size use in the aggregation. | + | `start_id` | integer | 10 | shared | Number used to start the unique ID generator. | + | `data_use_training` | integer, null | None | shared | Data used for training, if None, training is not done. | + | `data_use_configure` | integer, null | None | shared | Data used for configuration, if None, configuration is not done. | + | `use_config_data_as_training` | boolean | True | shared | Combine the configured data in the training process if True. | + | `train_buffer_max_records` | integer | 100000 | shared | 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 | shared | 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. | + diff --git a/docs/alert_aggregators/logid_count.md b/docs/alert_aggregators/logid_count.md new file mode 100644 index 00000000..1a868e22 --- /dev/null +++ b/docs/alert_aggregators/logid_count.md @@ -0,0 +1,60 @@ +# LogID Count Alert Aggregation + +The logID count alert aggregation tracks how many times each log ID appears in the alerts, and if it exceeds a specified threshold, it triggers the aggregation. + +## In/out + +| | Schema | Description | +|------------|----------------------------|--------------------| +| **Input** | [DetectorSchema](../schemas.md) | Alerts from detectors | +| **Output** | [AggregateSchema](../schemas.md) | Aggregated alerts | + +## At a glance + +| Learns from training data | Auto-configuration | Needs `events` | Federation | +|---|---|---|---| +| ❌ | ❌ | ❌ | ❌ | + +## Description + +A log ID counter stores, in a fixed-size table, the counts for different log IDs. If the count for any log ID from the Detector Schemas exceeds a defined threshold, it returns an alert aggregation. + +The fixed-size table operates as a FIFO: once it is full, it removes the oldest entry and inserts the new log ID. +## Example + +```python +--8<-- "docs/examples/alert_aggregators/logid_count.py:example_1" +``` + +## Configuration file + +The configuration used by the example above. It sets only what this use case needs; every other parameter keeps its default (see [Configuration arguments](#configuration-arguments)). + +```yaml +--8<-- "docs/examples/alert_aggregators/logid_count.yaml" +``` + +## Configuration arguments + +All parameters this detector accepts, grouped by the YAML block they go in. **Scope** tells whether a parameter is `specific` to this detector or `shared` with other detectors (see the [Alert Aggregation overview](../alert_aggregator.md#common-parameters-all-alert-aggregators)). Where this detector changes a shared default, the shared value is shown in brackets. + + +??? note "Top level" + +| Field | Type | Default | Scope | Description | +|---|---|---|---|---| +| `method_type` | string | basic_concat_aggregator | shared | Indicates what type of method it is. | +| `auto_config` | boolean | False | shared | Runs the configuration step before the training process. | + +???+ note "params" + +| Field | Type | Default | Scope | Description | +|---|---|---|---|---| +| `buffer_size` | integer | 3 | specific | Buffer size use in the aggregation. | +| `start_id` | integer | 10 | shared | Number used to start the unique ID generator. | +| `data_use_training` | integer, null | None | shared | Data used for training, if None, training is not done. | +| `data_use_configure` | integer, null | None | shared | Data used for configuration, if None, configuration is not done. | +| `use_config_data_as_training` | boolean | True | shared | Combine the configured data in the training process if True. | +| `train_buffer_max_records` | integer | 100000 | shared | 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 | shared | 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. | + diff --git a/docs/examples/alert_aggregators/basic_concat.py b/docs/examples/alert_aggregators/basic_concat.py index 27cb326e..101c311e 100644 --- a/docs/examples/alert_aggregators/basic_concat.py +++ b/docs/examples/alert_aggregators/basic_concat.py @@ -4,18 +4,11 @@ import detectmatelibrary.schemas as schemas +import yaml -aggregations_config = { - "alert_aggregators": { - "BasicConcatAggregator": { - "method_type": "basic_concat_aggregator", - "buffer_size": 3, - "deduplicate_values": False, - "auto_config": False, - } - } -} +with open("docs/examples/alert_aggregators/basic_concat.yaml") as f: + config = yaml.safe_load(f) input_ = schemas.DetectorSchema({ "detectorID": "1", @@ -30,9 +23,7 @@ "alertDetails": {"99 problems": "but logs aint one"} }) -alert_aggregator = BasicConcatAggregation( - "BasicConcatAggregator", aggregations_config -) +alert_aggregator = BasicConcatAggregation("BasicConcatAggregator", config) alert_aggregator.process(input_) # --8<-- [end:example_1] diff --git a/docs/examples/alert_aggregators/basic_concat.yaml b/docs/examples/alert_aggregators/basic_concat.yaml new file mode 100644 index 00000000..8eef83ca --- /dev/null +++ b/docs/examples/alert_aggregators/basic_concat.yaml @@ -0,0 +1,5 @@ +alert_aggregators: + BasicConcatAggregator: + method_type: "basic_concat_aggregator" + buffer_size: 3 + auto_config: False diff --git a/docs/examples/alert_aggregators/logid_count.py b/docs/examples/alert_aggregators/logid_count.py new file mode 100644 index 00000000..cb9d012a --- /dev/null +++ b/docs/examples/alert_aggregators/logid_count.py @@ -0,0 +1,35 @@ +# --8<-- [start:example_1] +from detectmatelibrary.alert_aggregation.logid_count import LogIDCountAggegation + +import detectmatelibrary.schemas as schemas + +import yaml + + +with open("docs/examples/alert_aggregators/logid_count.yaml") as f: + config = yaml.safe_load(f) + + +def generate_schema(log_id: str | list[str]) -> schemas.DetectorSchema: + log_id = log_id if isinstance(log_id, list) else [log_id] + return schemas.DetectorSchema({ + "detectorID": "1", + "detectorType": "dummy", + "alertID": "2", + "detectedTimestamp": 0, + "logIDs": log_id, + "score": 0.2, + "logTimestamps": [-1], + "description": "hello there", + "receivedTimestamp": 1, + "alertDetails": {"99 problems": "but logs aint one"} + }) + + +alert_aggregator = LogIDCountAggegation("LogIDCountAggregator", config=config) + +print(alert_aggregator.process(generate_schema("a")) is None) # True +print(alert_aggregator.process(generate_schema("b")) is None) # True +print(alert_aggregator.process(generate_schema("a")) is None) # False + +# --8<-- [end:example_1] diff --git a/docs/examples/alert_aggregators/logid_count.yaml b/docs/examples/alert_aggregators/logid_count.yaml new file mode 100644 index 00000000..bc646f55 --- /dev/null +++ b/docs/examples/alert_aggregators/logid_count.yaml @@ -0,0 +1,5 @@ +alert_aggregators: + LogIDCountAggregator: + method_type: "logid_count_aggregator" + max_count: 2 + max_memory: 5000 diff --git a/docs/examples/config/update.py b/docs/examples/config/update.py index 12b8d8aa..c6479a2b 100644 --- a/docs/examples/config/update.py +++ b/docs/examples/config/update.py @@ -25,11 +25,14 @@ from detectmatelibrary.common.core import CoreConfig from detectmatelibrary.common.detector import CoreDetectorConfig +from detectmatelibrary.common.alert_aggregator import CoreAlertAggregatorConfig from detectmatelibrary.common.parser import CoreParserConfig from detectmatelibrary.base_detectors import TrackerDetectorConfig from detectmatelibrary.base_detectors import VariableDetectorConfig from detectmatelibrary.base_detectors import DeepLearningDetectorConfig +from detectmatelibrary.alert_aggregation.basic_concat import BasicConcatAggregationConfig + from typing import Any @@ -184,6 +187,25 @@ def update_shared_args_parsers(doc_path: str) -> None: raise Exception(f"While updating {doc_path} -> {str(e)}") +def update_shared_args_aggregation(doc_path: str) -> None: + """Fill docs/alert_aggregation.md's tables of parameters shared across several + alert_aggregator: fields every parser has (CoreAlertAggregatorConfig).""" + try: + with open(doc_path, "r") as f: + docs = f.readlines() + + docs = append_docs( + docs=docs, + start_cmd="\n", + end_cmd="\n", + add=get_arguments(CoreAlertAggregatorConfig().get_docs(), with_scope=False), + ) + with open(doc_path, "w") as f: + f.writelines(docs) + except Exception as e: + raise Exception(f"While updating {doc_path} -> {str(e)}") + + # %% Documentation update # # Every config below drives its own doc page: whenever a Field is added, @@ -244,11 +266,23 @@ def update_shared_args_parsers(doc_path: str) -> None: ), ] +AGGREGATOR_DOCS: list[tuple[CoreConfig, type[CoreConfig], str]] = [ + ( + BasicConcatAggregationConfig(), + CoreAlertAggregatorConfig, + "docs/alert_aggregators/basic_concatenation.md" + ) +] + update_shared_args_detectors("docs/detectors.md") update_shared_args_parsers("docs/parsers.md") +update_shared_args_aggregation("docs/alert_aggregator.md") for detector_config, shared_base, doc_path in DETECTOR_DOCS: update_docs(detector_config, doc_path=doc_path, shared_base=shared_base) for parser_config, shared_base, doc_path in PARSER_DOCS: update_docs(parser_config, doc_path=doc_path, shared_base=shared_base) + +for aggr_config, shared_base, doc_path in AGGREGATOR_DOCS: + update_docs(aggr_config, doc_path=doc_path, shared_base=shared_base) diff --git a/docs/img/logo.png b/docs/img/logo.png new file mode 100644 index 00000000..a1e6491d Binary files /dev/null and b/docs/img/logo.png differ diff --git a/docs/index.md b/docs/index.md index 97977898..ad490e77 100644 --- a/docs/index.md +++ b/docs/index.md @@ -1,5 +1,7 @@ # DetectMate Library +![UML](img/logo.png) + Welcome to the documentation for the DetectMate Library. DetectMate is a flexible, component-based library for parsing and log anomaly detection. It can be used in research and production, and together with the diff --git a/mkdocs.yml b/mkdocs.yml index b5b8fa43..ffe48996 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -66,6 +66,7 @@ nav: - 12_LogBERT Detector: detectors/logbert.md - Alert Aggregation methods: - 00_Basic Concat: alert_aggregators/basic_concatenation.md + - 01_LogID Count: alert_aggregators/logid_count.md - Helpers: - FromTo: helper/from_to.md - Auxiliary: diff --git a/src/detectmatelibrary/alert_aggregation/basic_concat.py b/src/detectmatelibrary/alert_aggregation/basic_concat.py index 2ca1ec32..b863af3f 100644 --- a/src/detectmatelibrary/alert_aggregation/basic_concat.py +++ b/src/detectmatelibrary/alert_aggregation/basic_concat.py @@ -1,12 +1,20 @@ -from detectmatelibrary.common.alert_aggregator import (CoreAlertAggregatorConfig, CoreAlertAggregator) +from detectmatelibrary.common.alert_aggregator import ( + CoreAlertAggregatorConfig, CoreAlertAggregator +) + from detectmatelibrary.schemas import DetectorSchema, AggregateSchema + from typing import Any, Optional +from pydantic import Field class BasicConcatAggregationConfig(CoreAlertAggregatorConfig): - method_type: str = "basic_concat_aggregator" - buffer_size: int = 3 - deduplicate_values: bool = False + method_type: str = Field( + default="basic_concat_aggregator", description="Indicates what type of method it is." + ) + buffer_size: int = Field( + default=3, description="Buffer size use in the aggregation." + ) class BasicConcatAggregation(CoreAlertAggregator): @@ -17,18 +25,13 @@ def __init__( ) -> None: if isinstance(config, dict): config = BasicConcatAggregationConfig.from_dict(config, name) + buffer_size: int = config.buffer_size # type: ignore - self.deduplicate_values: bool = config.deduplicate_values # type: ignore super().__init__(name=name, buffer_size=buffer_size, config=config) def aggregate_alerts( - self, input_: list[DetectorSchema] | DetectorSchema, output_: AggregateSchema + self, input_: list[DetectorSchema], output_: AggregateSchema # type: ignore ) -> bool: + output_["description"] = "Basic aggregation by alert concatenation" - if self.deduplicate_values: - output_["detectorIDs"] = list(dict.fromkeys(output_["detectorIDs"])) - output_["detectorTypes"] = list(dict.fromkeys(output_["detectorTypes"])) - output_["alertIDs"] = list(dict.fromkeys(output_["alertIDs"])) - output_["logIDs"] = list(dict.fromkeys(output_["logIDs"])) - output_["logTimestamps"] = list(dict.fromkeys(output_["logTimestamps"])) return True diff --git a/src/detectmatelibrary/alert_aggregation/logid_count.py b/src/detectmatelibrary/alert_aggregation/logid_count.py new file mode 100644 index 00000000..deea9db2 --- /dev/null +++ b/src/detectmatelibrary/alert_aggregation/logid_count.py @@ -0,0 +1,96 @@ +from detectmatelibrary.common.alert_aggregator import ( + CoreAlertAggregatorConfig, CoreAlertAggregator +) + +from detectmatelibrary.schemas import DetectorSchema, AggregateSchema +from detectmatelibrary.utils.data_buffer import BufferMode + +from typing import Any, Optional +from functools import lru_cache +from pydantic import Field +import polars as pl + + +class Counter: + def __init__(self, max_count: int, overall_size: int) -> None: + self.max_count = max_count + + self._df = pl.DataFrame() + self._df = self._df.insert_column(0, pl.repeat("", n=overall_size).alias("LogID")) + self._df = self._df.insert_column(1, pl.zeros(overall_size, dtype=pl.UInt8).alias("Count")) + self._df = self._df.insert_column(2, pl.zeros(overall_size, dtype=pl.Boolean).alias("WasAlert")) + self._df = self._df.with_row_index(name="Index") + + self._cached_search = lru_cache(maxsize=overall_size)(self.__get_idx) + + self.idx = 0 + + def __is_new_alert(self, idx: int) -> bool: + if not self._df[idx, "WasAlert"] and self._df[idx, "Count"] >= self.max_count: + self._df[idx, "WasAlert"] = True + return True + return False + + def __get_idx(self, logId: str) -> int: + indices = self._df.filter(pl.col("LogID") == logId).select("Index").to_series() + + if indices.shape[0] == 0: + new_row = pl.DataFrame( + data=[[self.idx, logId, 0, False]], schema=self._df.schema, orient="row" + ) + self._df = self._df.update(new_row, on="Index") + idx = self.idx + self.idx = self.idx + 1 if self.idx < self._df.shape[0] - 1 else 0 + else: + idx = indices[0] + + return idx + + def add_and_check(self, logId: str) -> bool: + idx = self._cached_search(logId) + self._df[idx, "Count"] += 1 + return self.__is_new_alert(idx) + + +class LogIDCountAggegationConfig(CoreAlertAggregatorConfig): + method_type: str = Field( + default="logid_count_aggregator", description="Indicates what type of method it is." + ) + max_count: int = Field(default=4, description="Max count of alerts in a specific logID.") + max_memory: int = Field( + default=5000, description="Max number of logIDs store in the count memory." + ) + + +class LogIDCountAggegation(CoreAlertAggregator): + def __init__( + self, + name: str = "LogIDCountAggregator", + config: Optional[LogIDCountAggegationConfig | dict[str, Any]] = LogIDCountAggegationConfig(), + ) -> None: + if isinstance(config, dict): + config = LogIDCountAggegationConfig.from_dict(config, name) + + super().__init__(name=name, config=config, buffer_mode=BufferMode.NO_BUF, buffer_size=None) + self.config: LogIDCountAggegationConfig + + self.counter = Counter( + max_count=self.config.max_count, overall_size=self.config.max_memory + ) + + def aggregate_alerts( + self, input_: DetectorSchema, output_: AggregateSchema # type: ignore + ) -> bool: + + alert_logs = [] + for logID in input_["logIDs"]: + if self.counter.add_and_check(logID): + alert_logs.append(logID) + + if (aggreated_alerts := len(alert_logs) > 0): + output_["description"] = ( + "LogID count how many times a log was count as alert." + f"Logs that pass the limit count: {", ".join(alert_logs)}" + ) + + return aggreated_alerts diff --git a/src/detectmatelibrary/common/alert_aggregator.py b/src/detectmatelibrary/common/alert_aggregator.py index 28045ead..e589bceb 100644 --- a/src/detectmatelibrary/common/alert_aggregator.py +++ b/src/detectmatelibrary/common/alert_aggregator.py @@ -6,8 +6,9 @@ from detectmatelibrary.schemas import DetectorSchema, AggregateSchema -from typing import Any, Optional from typing_extensions import override +from typing import Any, Optional +from pydantic import Field def _extract_field( @@ -16,23 +17,31 @@ def _extract_field( if not isinstance(input_, list): input_ = [input_] - return [str(i[field]) for i in input_] + return list(set([str(i[field]) for i in input_])) -def _extract_list_field(input_: list[DetectorSchema] | DetectorSchema, field: str) -> list[str]: +def _extract_list_fields( + input_: list[DetectorSchema] | DetectorSchema, field1: str, field2: str +) -> tuple[list[str], list[str]]: + if not isinstance(input_, list): input_ = [input_] - final = [] - for i in input_: - final.extend(i[field]) + final1, final2 = [], [] + already_checked = set() + for in_ in input_: + for i_1, i_2 in zip(in_[field1], in_[field2]): + if i_1 not in already_checked: + final1.append(i_1) + final2.append(i_2) + already_checked.add(i_1) - return final + return final1, final2 class CoreAlertAggregatorConfig(CoreConfig): - component_type: str = "alert_aggregators" - method_type: str = "core_alert_aggregator" + component_type: str = Field(default="alert_aggregators", description="<$IGNORE$>") + method_type: str = Field(default="core_alert_aggregator", description="<$IGNORE$>") class CoreAlertAggregator(CoreComponent): @@ -63,8 +72,9 @@ def run( output_["detectorIDs"] = _extract_field(input_, "detectorID") output_["detectorTypes"] = _extract_field(input_, "detectorType") output_["alertIDs"] = _extract_field(input_, "alertID") - output_["logIDs"] = _extract_list_field(input_, "logIDs") - output_["logTimestamps"] = _extract_list_field(input_, "logTimestamps") + output_["logIDs"], output_["logTimestamps"] = _extract_list_fields( + input_, "logIDs", "logTimestamps" + ) if (alert := self.aggregate_alerts(input_=input_, output_=output_)): output_["aggregatedTimestamp"] = get_timestamp() diff --git a/src/detectmatelibrary/tools/workspace/templates/test_templates/test_CustomAlertAggregator.py b/src/detectmatelibrary/tools/workspace/templates/test_templates/test_CustomAlertAggregator.py index 1853be3f..ac7ceaa2 100644 --- a/src/detectmatelibrary/tools/workspace/templates/test_templates/test_CustomAlertAggregator.py +++ b/src/detectmatelibrary/tools/workspace/templates/test_templates/test_CustomAlertAggregator.py @@ -39,11 +39,12 @@ def test_run_aggregate_alerts_method(self) -> None: output = alert_aggregator.process(data) if i % 3 == 2: assert getattr(output, "description") == "Custom alert aggregation" - assert getattr(output, "detectorIDs") == ["MyCoolThing"]*3 + assert getattr(output, "detectorIDs") == ["MyCoolThing"] assert getattr(output, "alertDetails")["type"] == \ "Anomalies aggregated by CustomAlertAggregator" - assert getattr(output, "detectorTypes") == ["MyCoolThing_detector"]*3 - assert getattr(output, "alertIDs") == [ - f"MyCoolThing_{i + 8}", f"MyCoolThing_{i + 9}", f"MyCoolThing_{i + 10}"] - assert getattr(output, "logIDs") == [str(i-2), str(i-1), str(i)] + assert getattr(output, "detectorTypes") == ["MyCoolThing_detector"] + assert set(getattr(output, "alertIDs")) == set([ + f"MyCoolThing_{i + 8}", f"MyCoolThing_{i + 9}", f"MyCoolThing_{i + 10}" + ]) + assert set(getattr(output, "logIDs")) == set([str(i-2), str(i-1), str(i)]) i += 1 diff --git a/tests/test_alert_aggregators/test_basic_concat.py b/tests/test_alert_aggregators/test_basic_concat.py index 08d60c60..46a7c52f 100644 --- a/tests/test_alert_aggregators/test_basic_concat.py +++ b/tests/test_alert_aggregators/test_basic_concat.py @@ -1,96 +1,58 @@ from detectmatelibrary.alert_aggregation.basic_concat import BasicConcatAggregation -from detectmatelibrary.schemas import AggregateSchema + + from detectmatelibrary.utils.aux import time_test_mode + import detectmatelibrary.schemas as schemas -from datetime import datetime time_test_mode() aggregations_config = { "alert_aggregators": { - "Buffer_4": { + "Buffer_3": { "method_type": "basic_concat_aggregator", - "buffer_size": 4, - "auto_config": False, - "params": {}, + "buffer_size": 3, }, - "Buffer_6": { + "Buffer_5": { "method_type": "basic_concat_aggregator", - "buffer_size": 6, - "deduplicate_values": True, - "auto_config": False, - "params": {}, + "buffer_size": 5, }, - }, - "auto_config": False + } } -data_global = [schemas.DetectorSchema({ - "detectorID": str(i), - "detectorType": "dummy", - "alertID": str(i), - "detectedTimestamp": 0, - "logIDs": [f"logID{i}"], - "score": 0.2 + i / 10, - "logTimestamps": [int(datetime.now().timestamp() + i)], - "description": "hello there", - "receivedTimestamp": 1, - "alertDetails": {"99 problems": "but logs aint one"} -}) for i in range(1, 4)] - -data_global += [schemas.DetectorSchema({ +dummy_schema = { "detectorID": "1", "detectorType": "dummy", - "alertID": "1", + "alertID": "2", "detectedTimestamp": 0, - "logIDs": [f"logID{1}"], - "score": 0.3, - "logTimestamps": [int(datetime.now().timestamp() + 1)], + "logIDs": ["logID1"], + "score": 0.2, + "logTimestamps": [-1], "description": "hello there", "receivedTimestamp": 1, "alertDetails": {"99 problems": "but logs aint one"} -}) for i in range(1, 4)] +} class TestBasicAggregation: - def test_with_buffer_4(self): - alert_aggregator = BasicConcatAggregation("Buffer_4", aggregations_config) - data = data_global - - assert alert_aggregator.process(data[0]) is None - assert alert_aggregator.process(data[1]) is None - assert alert_aggregator.process(data[2]) is None - result = alert_aggregator.process(data[3]) - assert result is not None - assert isinstance(result, AggregateSchema) - res_dict = result.as_dict() - assert res_dict["detectorIDs"] == ["1", "2", "3", "1"] - assert res_dict["alertIDs"] == ["1", "2", "3", "1"] - assert res_dict["aggregatedTimestamp"] == 0 - assert res_dict["logIDs"] == ["logID1", "logID2", "logID3", "logID1"] - assert len(res_dict["logTimestamps"]) == 4 - assert res_dict["description"] == "Basic aggregation by alert concatenation" - assert res_dict["detectorTypes"] == ["dummy"]*4 - - def test_with_buffer_6(self): - alert_aggregator = BasicConcatAggregation("Buffer_6", aggregations_config) - data = data_global - - assert alert_aggregator.process(data[0]) is None - assert alert_aggregator.process(data[1]) is None - assert alert_aggregator.process(data[2]) is None - assert alert_aggregator.process(data[3]) is None - assert alert_aggregator.process(data[4]) is None - result = alert_aggregator.process(data[5]) - assert result is not None - assert isinstance(result, AggregateSchema) - res_dict = result.as_dict() - assert res_dict["detectorIDs"] == ["1", "2", "3"] - assert res_dict["alertIDs"] == ["1", "2", "3"] - assert res_dict["aggregatedTimestamp"] == 0 - assert res_dict["logIDs"] == ["logID1", "logID2", "logID3"] - assert len(res_dict["logTimestamps"]) == 3 - assert res_dict["description"] == "Basic aggregation by alert concatenation" - assert res_dict["detectorTypes"] == ["dummy"] + def test_with_buffer_3(self): + alert_aggregator = BasicConcatAggregation("Buffer_3", aggregations_config) + data = schemas.DetectorSchema(dummy_schema).serialize() + + assert alert_aggregator.process(data) is None + assert alert_aggregator.process(data) is None + assert (result := alert_aggregator.process(data)) is not None + assert isinstance(result, bytes) + + def test_with_buffer_5(self): + alert_aggregator = BasicConcatAggregation("Buffer_5", aggregations_config) + data = schemas.DetectorSchema(dummy_schema).serialize() + + assert alert_aggregator.process(data) is None + assert alert_aggregator.process(data) is None + assert alert_aggregator.process(data) is None + assert alert_aggregator.process(data) is None + assert (result := alert_aggregator.process(data)) is not None + assert isinstance(result, bytes) diff --git a/tests/test_alert_aggregators/test_logid_count.py b/tests/test_alert_aggregators/test_logid_count.py new file mode 100644 index 00000000..7fbd6b93 --- /dev/null +++ b/tests/test_alert_aggregators/test_logid_count.py @@ -0,0 +1,96 @@ + +from detectmatelibrary.alert_aggregation.logid_count import ( + Counter, LogIDCountAggegation, LogIDCountAggegationConfig +) +import detectmatelibrary.schemas as schemas + + +class TestCounter: + def test_basic_count(self) -> None: + counter = Counter(max_count=2, overall_size=10) + + assert not counter.add_and_check("a") + assert not counter.add_and_check("b") + assert counter.add_and_check("a") + assert counter.add_and_check("b") + + assert not counter.add_and_check("a") # should not generate new ones + assert not counter.add_and_check("b") # should not generate new ones + + def test_surpass_memory(self) -> None: + counter = Counter(max_count=2, overall_size=2) + + assert not counter.add_and_check("a") + assert not counter.add_and_check("b") + assert not counter.add_and_check("c") + + assert counter.add_and_check("b") + assert not counter.add_and_check("a") # was rename by c + assert counter.add_and_check("c") + assert counter.add_and_check("a") + + assert not counter.add_and_check("a") # should not generate new ones + assert not counter.add_and_check("c") # should not generate new ones + + +def generate_schema(log_id: str | list[str]) -> schemas.DetectorSchema: + log_id = log_id if isinstance(log_id, list) else [log_id] + return schemas.DetectorSchema({ + "detectorID": "1", + "detectorType": "dummy", + "alertID": "2", + "detectedTimestamp": 0, + "logIDs": log_id, + "score": 0.2, + "logTimestamps": [-1], + "description": "hello there", + "receivedTimestamp": 1, + "alertDetails": {"99 problems": "but logs aint one"} + }) + + +class TestLogIDCount: + def test_basic_count(self): + aggregator = LogIDCountAggegation(config=LogIDCountAggegationConfig(max_count=2)) + + assert aggregator.process(generate_schema("a")) is None + assert aggregator.process(generate_schema("b")) is None + + assert aggregator.process(generate_schema("a")) is not None + assert aggregator.process(generate_schema("b")) is not None + + assert aggregator.process(generate_schema("a")) is None + assert aggregator.process(generate_schema("b")) is None + + def test_basic_count_as_list(self): + aggregator = LogIDCountAggegation(config=LogIDCountAggegationConfig(max_count=2)) + + assert aggregator.process(generate_schema(["a", "b"])) is None + assert aggregator.process(generate_schema(["a", "b"])) is not None + assert aggregator.process(generate_schema(["a", "b"])) is None + + def test_surpass_memory(self): + aggregator = LogIDCountAggegation( + config=LogIDCountAggegationConfig(max_count=2, max_memory=2) + ) + + assert aggregator.process(generate_schema("a")) is None + assert aggregator.process(generate_schema("b")) is None + assert aggregator.process(generate_schema("c")) is None + + assert aggregator.process(generate_schema("b")) is not None + assert aggregator.process(generate_schema("a")) is None # was rename by c + assert aggregator.process(generate_schema("c")) is not None + assert aggregator.process(generate_schema("a")) is not None + + assert aggregator.process(generate_schema("a")) is None # should not generate new ones + assert aggregator.process(generate_schema("c")) is None # should not generate new ones + + def test_description(self): + aggregator = LogIDCountAggegation(config=LogIDCountAggegationConfig(max_count=1)) + alert = aggregator.process(generate_schema(["a", "b"])) + + expected = "LogID count how many times a log was count as alert." + expected += "Logs that pass the limit count: a, b" + + assert alert["description"] == expected \ No newline at end of file diff --git a/tests/test_common/test_core_alert_aggregator.py b/tests/test_common/test_core_alert_aggregator.py index 88d09fdf..a9756273 100644 --- a/tests/test_common/test_core_alert_aggregator.py +++ b/tests/test_common/test_core_alert_aggregator.py @@ -1,5 +1,5 @@ from detectmatelibrary.common.alert_aggregator import ( - CoreAlertAggregatorConfig, CoreAlertAggregator, BufferMode + CoreAlertAggregatorConfig, CoreAlertAggregator, BufferMode, _extract_field, _extract_list_fields ) from detectmatelibrary.utils.aux import time_test_mode @@ -74,6 +74,59 @@ def aggregate_alerts(self, input_, output_): time_test_mode() +class TestAuxiliar: + def test_get_field_single(self) -> None: + expected = {"dummy"} + input_ = schemas.DetectorSchema(dummy_schema) + + assert len(field := _extract_field(input_, "detectorType")) == 1 + assert expected == set(field) + + def test_get_field_multiple(self) -> None: + expected = {"dummy2", "dummy"} + + dummy_schema2 = dummy_schema.copy() + dummy_schema2["detectorType"] = "dummy2" + input_ = [dummy_schema, dummy_schema2] + input_ = [schemas.DetectorSchema(in_) for in_ in input_] + + assert len(field := _extract_field(input_, "detectorType")) == 2 + assert expected == set(field) + + def test_get_logID_timestamp_single(self) -> None: + expected_logID = ["logID1", "logID2", "logID3"] + expected_times = [-1, 0, 1] + + dummy_schema2 = dummy_schema.copy() + dummy_schema2["logIDs"] = ["logID1", "logID2", "logID3", "logID1"] + dummy_schema2["logTimestamps"] = [-1, 0, 1, 3] + input_ = schemas.DetectorSchema(dummy_schema2) + + logids, times = _extract_list_fields( + input_, field1="logIDs", field2="logTimestamps" + ) + + assert expected_logID == logids + assert times == expected_times + + def test_get_logID_timestamp_multiple(self) -> None: + expected_logID = ["logID1", "logID2", "logID3"] + expected_times = [-1, 0, 1] + + dummy_schema2 = dummy_schema.copy() + dummy_schema2["logIDs"] = ["logID2", "logID3", "logID1"] + dummy_schema2["logTimestamps"] = [0, 1, 3] + input_ = [dummy_schema, dummy_schema2] + input_ = [schemas.DetectorSchema(in_) for in_ in input_] + + logids, times = _extract_list_fields( + input_, field1="logIDs", field2="logTimestamps" + ) + + assert expected_logID == logids + assert times == expected_times + + class TestCoreAlertAggregator: def test_initialize_default(self) -> None: detector = MockupAlertAggregator(name="TestAAG", config=dummy_config) @@ -124,14 +177,14 @@ def test_process_input_schema_not_serialized(self) -> None: def test_process_input_schema_not_serialized_window_3(self) -> None: alert_aggregator = MockupAlertAggregator(name="TestDetector", config=MockupConfig(), buffer_size=3) expected_result = schemas.AggregateSchema({ - "__version__": "2.0.0", - "detectorIDs": ["1", "1", "1"], - "detectorTypes": ["dummy", "dummy", "dummy"], - "alertIDs": ["2", "2", "2"], + "__version__": "1.0.0", + "detectorIDs": ["1"], + "detectorTypes": ["dummy"], + "alertIDs": ["2"], "aggregatedTimestamp": 0, - "logIDs": ["logID1", "logID1", "logID1"], + "logIDs": ["logID1"], "description": "Mockup aggregator", - "logTimestamps": [-1, -1, -1], + "logTimestamps": [-1], "alertDetails": {} }) data = schemas.DetectorSchema(dummy_schema) @@ -147,14 +200,14 @@ def test_process_input_schema_not_serialized_buffer_3(self) -> None: name="TestDetector", config=MockupConfig(), buffer_size=3 ) expected_result = schemas.AggregateSchema({ - "__version__": "2.0.0", - "detectorIDs": ["1", "1", "1"], - "detectorTypes": ["dummy", "dummy", "dummy"], - "alertIDs": ["2", "2", "2"], + "__version__": "1.0.0", + "detectorIDs": ["1"], + "detectorTypes": ["dummy"], + "alertIDs": ["2"], "aggregatedTimestamp": 0, - "logIDs": ["logID1", "logID1", "logID1"], + "logIDs": ["logID1"], "description": "Mockup aggregator", - "logTimestamps": [-1, -1, -1], + "logTimestamps": [-1], "alertDetails": {} }) data = schemas.DetectorSchema(dummy_schema)