diff --git a/docs/alert_aggregator.md b/docs/alert_aggregator.md index 0c2edd83..d22d1799 100644 --- a/docs/alert_aggregator.md +++ b/docs/alert_aggregator.md @@ -71,6 +71,7 @@ The alert aggregator are numbered from simplest to most complex, and the sidebar |---|---|---| | 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. | +| 02 | [Density Cluster](alert_aggregators/density_cluster.md) | Base class for clustering alerts using a distance function. | ### Common parameters (all alert aggregators) diff --git a/docs/alert_aggregators/density_cluster.md b/docs/alert_aggregators/density_cluster.md new file mode 100644 index 00000000..017c55d3 --- /dev/null +++ b/docs/alert_aggregators/density_cluster.md @@ -0,0 +1,70 @@ +# Density Cluster Alert Aggregation + +The density-clustering-based alert aggregator is a base class to implement alert aggregation by clustering alerts according a given distance function of alerts. + +## 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 + +This class implements alert aggregation by density clustering. + +### Implementation Details + +Creating a custom density clustering aggregator requires the implementation of 2 methods in subclasses: + +1. One has to implement either the single_metric or the metric function. +The single_metric function computes the distance between 2 alerts, +while the metric function computes the distance matrix between a list of alerts. +If both functions are implemented, the metric function will be used. + +2. The discard_alert_fn function must be implemented to determine which alerts +should no longer be considered for aggregation. +To this end, after each aggregation step, the discard_alert_fn +function is called with the list of relevant alerts. +It should return a list of booleans indicating which alerts should be discarded (True) +and which should be kept (False). + +### Labelling of Alerts + +The group labels assigned to alerts are published in the output schema as a +dictionary mapping alert IDs to group IDs in the "alertDetails" field. +Every alert is assigned a group ID immediately after it is received, +but as the aggregation progresses, groups may be merged. +This is published in the output schema as a list of messages in the "alertDetails" field. +Likewise, when a group is finished and no longer reachable from any active alert, +this is also published in the output schema as a list of messages in the "alertDetails" field. + +## Example + +```python +--8<-- "docs/examples/alert_aggregators/density_cluster.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/density_cluster.yaml" +``` + +Load it with `yaml.safe_load` and pass the dict as `config=`, as in the example. The key under `alert_aggregators:` must match the aggregator's `name`. + +## Configuration arguments + +All parameters this alert aggregator accepts, grouped by the YAML block they go in. **Scope** tells whether a parameter is `specific` to this aggregator or `shared` with other alert aggregators. + + + + diff --git a/docs/alert_aggregators/logid_count.md b/docs/alert_aggregators/logid_count.md index 1a868e22..94e178e3 100644 --- a/docs/alert_aggregators/logid_count.md +++ b/docs/alert_aggregators/logid_count.md @@ -41,20 +41,21 @@ All parameters this detector accepts, grouped by the YAML block they go in. **Sc ??? 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. | + | Field | Type | Default | Scope | Description | + |---|---|---|---|---| + | `method_type` | string | logid_count_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. | + | Field | Type | Default | Scope | Description | + |---|---|---|---|---| + | `max_count` | integer | 4 | specific | Max count of alerts in a specific logID. | + | `max_memory` | integer | 5000 | specific | Max number of logIDs store in the count memory. | + | `start_id` | integer | 10 | shared | Number used to start the unique ID generator. | + | `data_use_training` | integer, null | None | shared | Number of logs used for training. None skips training. | + | `data_use_configure` | integer, null | None | shared | Number of logs used for the configuration step. None skips it. | + | `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/density_cluster.py b/docs/examples/alert_aggregators/density_cluster.py new file mode 100644 index 00000000..299accbb --- /dev/null +++ b/docs/examples/alert_aggregators/density_cluster.py @@ -0,0 +1,63 @@ +# --8<-- [start:example_1] +import yaml + +from detectmatelibrary.alert_aggregation.density_clustering import ( + DensityClusteringAggregation, +) +from detectmatelibrary.schemas import AggregateSchema, DetectorSchema + +with open("docs/examples/alert_aggregators/density_cluster.yaml") as f: + config = yaml.safe_load(f) + + +def make_alert(alert_id: str, score: float) -> DetectorSchema: + return DetectorSchema( + { + "detectorID": "detector", + "detectorType": "dummy", + "alertID": alert_id, + "detectedTimestamp": 0, + "logIDs": [f"log-{alert_id}"], + "score": score, + "logTimestamps": [0], + "description": "test alert", + "receivedTimestamp": 0, + "alertDetails": {}, + } + ) + + +class ConcreteDensityClusteringAggregation(DensityClusteringAggregation): + def single_metric( + self, + alert1: DetectorSchema, + alert2: DetectorSchema + ) -> float: + return abs(alert1["score"] - alert2["score"]) + + def discard_alert_fn(self, alerts: list[DetectorSchema]) -> list[bool]: + return [False for alert in alerts] + + +aggregator = ConcreteDensityClusteringAggregation(config) +first = make_alert("a", 0.0) +second = make_alert("b", 6.0) +bridge = make_alert("c", 3.0) + +first_output = AggregateSchema() +aggregator.aggregate_alerts(first, first_output) + +second_output = AggregateSchema() +aggregator.aggregate_alerts(second, second_output) + +bridge_output = AggregateSchema() +aggregator.aggregate_alerts(bridge, bridge_output) + +print(first_output["alertDetails"]["a"] == "0") # True +print(second_output["alertDetails"]["b"] == "1") # True +print(bridge_output["alertDetails"]["c"] == "0") # True + +# group labels 0 and 1 are merged into a single group +print(aggregator.active_groups[1] == {0, 1}) # True + +# --8<-- [end:example_1] diff --git a/docs/examples/alert_aggregators/density_cluster.yaml b/docs/examples/alert_aggregators/density_cluster.yaml new file mode 100644 index 00000000..6a5ed1bb --- /dev/null +++ b/docs/examples/alert_aggregators/density_cluster.yaml @@ -0,0 +1,4 @@ +alert_aggregators: + DensityClusteringAggregator: + method_type: density_clustering_aggregator + delta: 4.0 diff --git a/docs/examples/config/update.py b/docs/examples/config/update.py index 6c104a2f..924035cf 100644 --- a/docs/examples/config/update.py +++ b/docs/examples/config/update.py @@ -28,6 +28,8 @@ from detectmatelibrary.common.parser import CoreParserConfig from detectmatelibrary.common.alert_aggregator import CoreAlertAggregatorConfig from detectmatelibrary.alert_aggregation.basic_concat import BasicConcatAggregationConfig +from detectmatelibrary.alert_aggregation.density_clustering import DensityClusteringAggregationConfig +from detectmatelibrary.alert_aggregation.logid_count import LogIDCountAggegationConfig from detectmatelibrary.base_detectors import TrackerDetectorConfig from detectmatelibrary.base_detectors import VariableDetectorConfig from detectmatelibrary.base_detectors import DeepLearningDetectorConfig @@ -271,7 +273,17 @@ def update_shared_args_aggregation(doc_path: str) -> None: BasicConcatAggregationConfig(), CoreAlertAggregatorConfig, "docs/alert_aggregators/basic_concatenation.md" - ) + ), + ( + LogIDCountAggegationConfig(), + CoreAlertAggregatorConfig, + "docs/alert_aggregators/logid_count.md" + ), + ( + DensityClusteringAggregationConfig(), + CoreAlertAggregatorConfig, + "docs/alert_aggregators/density_cluster.md" + ), ] update_shared_args_detectors("docs/detectors.md") diff --git a/mkdocs.yml b/mkdocs.yml index 8b0464f7..b0e19095 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -68,6 +68,7 @@ nav: - Alert Aggregation methods: - 00_Basic Concat: alert_aggregators/basic_concatenation.md - 01_LogID Count: alert_aggregators/logid_count.md + - 02_Density Cluster: alert_aggregators/density_cluster.md - Helpers: - FromTo: helper/from_to.md - Select Component: helper/select_component.md diff --git a/src/detectmatelibrary/alert_aggregation/__init__.py b/src/detectmatelibrary/alert_aggregation/__init__.py index d1b82f94..b858e98b 100644 --- a/src/detectmatelibrary/alert_aggregation/__init__.py +++ b/src/detectmatelibrary/alert_aggregation/__init__.py @@ -1,9 +1,15 @@ from .basic_concat import BasicConcatAggregation, BasicConcatAggregationConfig +from .density_clustering import ( + DensityClusteringAggregation, + DensityClusteringAggregationConfig, +) from .logid_count import LogIDCountAggegation, LogIDCountAggegationConfig __all__ = [ "BasicConcatAggregation", "BasicConcatAggregationConfig", + "DensityClusteringAggregation", + "DensityClusteringAggregationConfig", "LogIDCountAggegation", "LogIDCountAggegationConfig", ] diff --git a/src/detectmatelibrary/alert_aggregation/density_clustering.py b/src/detectmatelibrary/alert_aggregation/density_clustering.py new file mode 100644 index 00000000..0ea29405 --- /dev/null +++ b/src/detectmatelibrary/alert_aggregation/density_clustering.py @@ -0,0 +1,403 @@ +from typing import Any + +from pydantic import Field, NonNegativeFloat + +from detectmatelibrary.common.alert_aggregator import ( + CoreAlertAggregator, + CoreAlertAggregatorConfig, +) +from detectmatelibrary.schemas import AggregateSchema, DetectorSchema +from detectmatelibrary.utils.data_buffer import BufferMode + + +class DensityClusteringAggregationConfig(CoreAlertAggregatorConfig): + """Configuration for the DensityClusteringAggregation class. + + Parameters: + buffer_size (int): The size of the buffer. + buffer_mode (BufferMode): The mode of the buffer. + delta (NonNegativeFloat): The threshold for the clustering. + """ + + method_type: str = Field( + default="density_clustering_aggregator", + description="The kind of aggregation method.", + ) + buffer_size: int = Field(default=3, description="The size of the buffer.") + buffer_mode: BufferMode = Field( + default=BufferMode.BATCH, description="The mode of the buffer." + ) + delta: NonNegativeFloat = Field( + default=0.5, description="The threshold for the clustering." + ) + + +class DensityClusteringAggregation(CoreAlertAggregator): + """This class implements alert aggregation by density clustering. + + Attributes: + config (DensityClusteringAggregationConfig): The configuration for the aggregator. + next_group_id (int): The next group ID to be assigned. + relevant_alerts (list[DetectorSchema]): The list of alerts currently relevant for aggregation. + alert_id_to_group_id (dict[str, int | None]): A mapping from alert IDs to group IDs. + active_groups (dict[int, set[int]]): A mapping from group IDs to sets of merged group IDs. + message_buffer (list[str]): A buffer for messages to be published in the output schema. + """ + + def __init__( + self, + config: DensityClusteringAggregationConfig | dict[str, Any], + name: str = "DensityClusteringAggregator", + ) -> None: + """Initialize the density clustering aggregator. + + Args: + config: The configuration for the aggregator. + name: The name of the aggregator. + """ + if isinstance(config, dict): + config = DensityClusteringAggregationConfig.from_dict(config, name) + + self.buffer_size = config.buffer_size + self.buffer_mode = config.buffer_mode + super().__init__( + name=name, + config=config, + buffer_size=config.buffer_size, + buffer_mode=config.buffer_mode, + ) + self.config: DensityClusteringAggregationConfig # ty: ignore[invalid-mutable-override] + self.next_group_id: int = 0 + self.relevant_alerts: list[DetectorSchema] = [] + self.alert_id_to_group_id: dict[str, int | None] = {} + self.active_groups: dict[int, set[int]] = {} + self.message_buffer: list[str] = [] + self.description = "Alert aggregation by density clustering." + + def single_metric( + self, alert1: DetectorSchema, alert2: DetectorSchema # noqa: ARG002 + ) -> NonNegativeFloat: + """Compute the distance between two alerts. This placeholder + implementation always returns 0.0, indicating that all alerts are + considered identical. + + Args: + alert1: The first alert. + alert2: The second alert. + + Returns: + The distance between the two alerts. + """ + return 0.0 + + def metric( + self, alerts: list[DetectorSchema], return_last_rows: int = 0 + ) -> list[list[NonNegativeFloat]]: + """Compute the distance matrix between a list of alerts. + + Subclasses can override this method to provide a more efficient + implementation. If both this method and ``single_metric`` are + implemented, this method is used. + + Args: + alerts: The alerts for which to compute distances. + return_last_rows: The number of final rows to compute. If zero, + compute all rows. + + Returns: + The distance matrix between the alerts. + """ + return [ + [self.single_metric(alert1, alert2) for alert2 in alerts] + for alert1 in alerts[-return_last_rows:] + ] + + def discard_alert_fn(self, alerts: list[DetectorSchema]) -> list[bool]: + """Determine which alerts should no longer be considered for + aggregation. This placeholder implementation always returns a list of + ``True`` values, indicating that all alerts should be discarded. + + Args: + alerts: The alerts currently relevant for aggregation. + + Returns: + A list of booleans indicating which alerts should be discarded. + ``True`` values indicate alerts to discard, and ``False`` values + indicate alerts to keep. + """ + return [True] * len(alerts) + + def aggregate_alerts( + self, input_: list[DetectorSchema] | DetectorSchema, output_: AggregateSchema + ) -> bool: + """Aggregate input alerts and write the results to the output. + + Args: + input_: The alert or alerts to aggregate. + output_: The schema to which aggregation results are written. + + Returns: + ``True`` after the aggregation results have been written. + ``False`` if no new alerts were aggregated and the output schema was not modified. + """ + + if not isinstance(input_, list): + input_ = [input_] + + # identify new alerts and add them to relevant alerts + new_alerts = self._add_new_alerts(input_) + + if len(new_alerts) == 0: + return False # no new alerts to aggregate + + # check distance of new alerts to relevant alerts + is_same_group = self._check_threshold( + self.metric( + self.relevant_alerts, + len(new_alerts), + ) + ) + + # assign groups to new alerts and merge groups if necessary + for new_alert, same_group in zip(new_alerts, is_same_group, strict=True): + self._assign_new_alert_to_group(new_alert, same_group) + + # write results to output + self._write_output(input_, output_) + return True + + def _write_output( + self, input_: list[DetectorSchema], output_: AggregateSchema + ) -> None: + """Write group assignments and discard completed alerts. + + Args: + input_: The alerts whose group assignments should be written. + output_: The schema to which aggregation results are written. + """ + output_["description"] = self.description + output_["alertDetails"].update( + { + alert["alertID"]: str(self.alert_id_to_group_id[alert["alertID"]]) + for alert in input_ + } + ) + + # discard alerts that are no longer relevant + _, affected_group_ids = self._discard_alerts() + self._check_and_discard_group_ids(affected_group_ids) + output_["alertDetails"]["messages"] = "\n".join(self.message_buffer) + self.message_buffer = [] + + def _add_new_alerts(self, input_: list[DetectorSchema]) -> list[DetectorSchema]: + """Add previously unseen alerts to the relevant-alert collection. + + Args: + input_: Alerts received by the aggregator. + + Returns: + The subset of input alerts that had not been seen before. + """ + new_input = [ + alert + for alert in input_ + if alert["alertID"] not in self.alert_id_to_group_id + ] + self.relevant_alerts.extend(new_input) + self.alert_id_to_group_id.update( + {alert["alertID"]: None for alert in new_input} + ) + return new_input + + def _check_threshold( + self, distance_matrix: list[list[NonNegativeFloat]] + ) -> list[list[bool]]: + """Convert distances into threshold comparisons. + + Args: + distance_matrix: The pairwise alert distance matrix. + + Returns: + A matrix indicating whether each distance is within the configured + clustering threshold. + """ + return [ + [distance <= self.config.delta for distance in row] + for row in distance_matrix + ] + + def _new_group(self, alert_id: str) -> int: + """Create a group containing a single alert. + + Args: + alert_id: The identifier of the alert to place in the new group. + + Returns: + The identifier assigned to the new group. + """ + group_id = self.next_group_id + self.alert_id_to_group_id[alert_id] = group_id + self.active_groups[group_id] = {group_id} + self.next_group_id += 1 + return group_id + + def _check_and_merge_groups(self, close_group_ids: set[int]) -> set[int] | None: + """Merge distinct groups associated with nearby alerts. + + Args: + close_group_ids: The group identifiers associated with nearby + alerts. + + Returns: + The merged group when multiple groups were combined, or ``None`` + when the groups were already merged. + + Raises: + Exception: If no active groups can be found for the supplied + identifiers. + """ + # check if groups are merged already + groups_to_merge = [] + for group_id in close_group_ids: + if (full_group := self.active_groups[group_id]) not in groups_to_merge: + groups_to_merge.append(full_group) + + if (num_groups := len(groups_to_merge)) == 0: + raise Exception("An error occurred while merging groups!") + elif num_groups == 1: + # all groups are merged already + return None + + # merge groups + new_group = set.union(*groups_to_merge) + for group_id in new_group: + self.active_groups[group_id] = new_group + + self.message_buffer.append( + f"Groups {groups_to_merge} are merged into {new_group}." + ) + return new_group + + def _assign_new_alert_to_group( + self, new_alert: DetectorSchema, close_relevant_alerts: list[bool] + ) -> None: + """Assign a new alert to a group based on nearby relevant alerts. + + Args: + new_alert: The alert that should receive a group assignment. + close_relevant_alerts: Flags indicating which relevant alerts are + within the clustering threshold. + + Raises: + Exception: If the new alert has no close alerts or if close alerts + have no group identifiers. + """ + close_alerts = [ + relevant_alert + for relevant_alert, is_close in zip( + self.relevant_alerts, close_relevant_alerts, strict=True + ) + if is_close + ] + + if len(close_alerts) == 0: + raise Exception("Alert is not close to itself!") + + elif len(close_alerts) == 1: + # only close alert is itself, create new group + self._new_group(new_alert["alertID"]) + return + + # check distinct groups of close alerts + close_group_ids = { + self.alert_id_to_group_id[alert["alertID"]] for alert in close_alerts + } + + if len(close_group_ids) == 0: + raise Exception("Could not find group_id's of close alerts!") + + elif len(close_group_ids) == 1: + if (group_id := close_group_ids.pop()) is None: + # all close alerts are not assigned to a group, create new group + self._new_group(new_alert["alertID"]) + return + + # all close alerts belong to the same group, assign new alert to that group + self.alert_id_to_group_id[new_alert["alertID"]] = group_id + return + + # unlabelled close alerts can be ignored now, they will be labelled in their own iteration + valid_group_ids: set[int] = { + group_id for group_id in close_group_ids if group_id is not None + } + + # close alerts have multiple group_ids, add alert to some group_id + for group_id in valid_group_ids: + self.alert_id_to_group_id[new_alert["alertID"]] = group_id + break + + # merge groups if necessary + self._check_and_merge_groups(valid_group_ids) + + def _discard_alerts(self, keep_buffer: bool = True) -> tuple[list[str], set[int]]: + """Discard alerts no longer needed for future aggregation. + + Args: + keep_buffer: Whether to retain the most recent window-buffer + alerts when using window buffering. + + Returns: + A tuple containing discarded alert identifiers and group identifiers + affected by the discard operation. + """ + alerts_to_discard = self.discard_alert_fn(self.relevant_alerts) + if keep_buffer and self.buffer_mode == BufferMode.WINDOW: + # set the last self.buffer_size elements of alerts_to_discard to False + for i in range(1, self.buffer_size + 1): + if i > len(alerts_to_discard): + break + alerts_to_discard[-i] = False + discarded_alert_ids: list[str] = [ + alert["alertID"] + for alert, discard in zip( + self.relevant_alerts, alerts_to_discard, strict=True + ) + if discard + ] + affected_group_ids = { + group_id + for alert_id in discarded_alert_ids + if (group_id := self.alert_id_to_group_id.pop(alert_id)) is not None + } + self.relevant_alerts = [ + alert + for alert, discard in zip( + self.relevant_alerts, alerts_to_discard, strict=True + ) + if not discard + ] + return discarded_alert_ids, affected_group_ids + + def _check_and_discard_group_ids(self, group_ids_to_check: set[int]) -> None: + """Remove groups that are no longer reachable from active alerts. + + Args: + group_ids_to_check: Group identifiers affected by discarded alerts. + """ + # remove groups that are unreachable anymore + group_ids_to_discard = { + group_id + for group_id in group_ids_to_check + if group_id not in self.alert_id_to_group_id.values() + } + for group_id in group_ids_to_discard: + self.message_buffer.append(f"Group id {group_id} is finished.") + + # check if group_id has been merged with still active groups + full_group_to_check = self.active_groups.pop(group_id) + for group_id_to_check in full_group_to_check: + if group_id_to_check in self.active_groups: + break + else: + # full_group_to_check is not reachable anymore + self.message_buffer.append(f"Group {full_group_to_check} is finished.") diff --git a/tests/test_alert_aggregators/test_density_clustering.py b/tests/test_alert_aggregators/test_density_clustering.py new file mode 100644 index 00000000..7907ad37 --- /dev/null +++ b/tests/test_alert_aggregators/test_density_clustering.py @@ -0,0 +1,127 @@ +from detectmatelibrary.alert_aggregation.density_clustering import ( + DensityClusteringAggregation, + DensityClusteringAggregationConfig, +) +from detectmatelibrary.schemas import AggregateSchema, DetectorSchema +from detectmatelibrary.utils.data_buffer import BufferMode + + +def make_alert(alert_id: str, score: float) -> DetectorSchema: + return DetectorSchema( + { + "detectorID": "detector", + "detectorType": "dummy", + "alertID": alert_id, + "detectedTimestamp": 0, + "logIDs": [f"log-{alert_id}"], + "score": score, + "logTimestamps": [0], + "description": "test alert", + "receivedTimestamp": 0, + "alertDetails": {}, + } + ) + + +class ConcreteDensityClusteringAggregation(DensityClusteringAggregation): + def single_metric(self, alert1: DetectorSchema, alert2: DetectorSchema) -> float: + return abs(alert1["score"] - alert2["score"]) + + def discard_alert_fn(self, alerts: list[DetectorSchema]) -> list[bool]: + return [alert["alertID"].startswith("discard") for alert in alerts] + + +def create_aggregator( + *, + buffer_mode: BufferMode = BufferMode.BATCH, + buffer_size: int = 3, + delta: float = 1.0, +) -> ConcreteDensityClusteringAggregation: + return ConcreteDensityClusteringAggregation( + config=DensityClusteringAggregationConfig( + buffer_mode=buffer_mode, + buffer_size=buffer_size, + delta=delta, + ) + ) + + +class TestDensityClusteringAggregation: + def test_metric_and_threshold(self) -> None: + aggregator = create_aggregator(delta=2.0) + alerts = [make_alert("a", 1.0), make_alert("b", 4.0)] + + assert aggregator.metric(alerts) == [[0.0, 3.0], [3.0, 0.0]] + assert aggregator.metric(alerts, return_last_rows=1) == [[3.0, 0.0]] + assert aggregator._check_threshold([[0.0, 2.0], [2.1, 0.0]]) == [ + [True, True], + [False, True], + ] + + def test_aggregate_assigns_and_merges_groups(self) -> None: + aggregator = create_aggregator(delta=1.0) + first = make_alert("a", 0.0) + second = make_alert("b", 10.0) + bridge = make_alert("c", 5.0) + + first_output = AggregateSchema() + assert aggregator.aggregate_alerts(first, first_output) + assert first_output["alertDetails"]["a"] == "0" + + second_output = AggregateSchema() + aggregator.aggregate_alerts(second, second_output) + assert second_output["alertDetails"]["b"] == "1" + + bridge_output = AggregateSchema() + aggregator.config.delta = 6.0 + aggregator.aggregate_alerts(bridge, bridge_output) + + assert bridge_output["alertDetails"]["c"] == "0" + assert aggregator.active_groups == {0: {0, 1}, 1: {0, 1}} + assert "Groups" in bridge_output["alertDetails"]["messages"] + + def test_duplicate_alert_is_not_reassigned(self) -> None: + aggregator = create_aggregator() + alert = make_alert("same", 0.0) + + first_output = AggregateSchema() + aggregator.aggregate_alerts(alert, first_output) + duplicate_output = AggregateSchema() + aggregator.aggregate_alerts(alert, duplicate_output) + + assert "same" not in duplicate_output["alertDetails"] + assert aggregator.next_group_id == 1 + assert len(aggregator.relevant_alerts) == 1 + + def test_window_buffer_keeps_latest_alerts(self) -> None: + aggregator = create_aggregator( + buffer_mode=BufferMode.WINDOW, + buffer_size=2, + ) + alerts = [ + make_alert("discard-1", 0.0), + make_alert("keep-1", 0.0), + make_alert("keep-2", 0.0), + ] + aggregator.relevant_alerts.extend(alerts) + aggregator.alert_id_to_group_id.update( + {alert["alertID"]: 0 for alert in alerts} + ) + + discarded, affected_groups = aggregator._discard_alerts() + + assert discarded == ["discard-1"] + assert affected_groups == {0} + assert [alert["alertID"] for alert in aggregator.relevant_alerts] == [ + "keep-1", + "keep-2", + ] + + def test_default_hooks_return_dummy_values(self) -> None: + aggregator = DensityClusteringAggregation( + config=DensityClusteringAggregationConfig(delta=1.0) + ) + alert = make_alert("a", 0.0) + + assert aggregator.single_metric(alert, alert) == 0.0 + assert aggregator.discard_alert_fn([alert]) == [True]