diff --git a/docs/detectors.md b/docs/detectors.md index e3e744a3..c3615e5d 100644 --- a/docs/detectors.md +++ b/docs/detectors.md @@ -222,6 +222,79 @@ def set_configuration(self): When `auto_config` is `False`, steps 1 and 2 are skipped entirely. +### Stability segmentation (optional) + +Stability classification splits a variable's change history into four segments and +compares each segment's rate of change against a threshold. By default the segments +are **equal-count**: each holds the same number of observations, regardless of how much +time they cover. For bursty log sources that is misleading — a variable that changed +constantly during a quiet night and then went silent under a flood of daytime traffic +looks stable, because the flood supplies enough samples to dominate the later segments. + +Setting `stability_segmentation: time` switches the segmentation to **equal-duration** cuts +of the observed time span, so each segment covers the same amount of wall-clock time. The +detector then needs an event time per record, which it reads from the log's named +variables (`logFormatVariables`, i.e. the fields declared in the parser's `log_format`) +under the name given by `timestamp_variable`. + +These three parameters live on every `VariableDetector` subclass (`NewValueDetector`, +`NewValueComboDetector`, `ValueRangeDetector`, `CharsetDetector`, `BigramDetector`, …) +and go in the detector's top-level `params` block: + +```yaml +detectors: + NewValueDetector: + method_type: new_value_detector + auto_config: True + params: + stability_segmentation: time + timestamp_variable: Time # a field name from the parser's log_format + timestamp_format: "%y%m%d %H%M%S" # optional; omit to auto-detect +``` + +Setting `stability_segmentation: both` runs *both* segmentations and calls the variable +stable only when each one does. Neither segmentation subsumes the other — a variable that +churns in a burst and then settles is unstable by count but stable by time, and one whose +late churn is buried under a dense tail of repeats is the reverse — so `both` is strictly +stricter than either. Use it when a false "stable" is more costly than a missed one; use +`time` when the point is specifically to forgive early churn on a bursty source. + +#### Fields + +| Field | Type | Default | Description | +|---|---|---|---| +| `stability_segmentation` | `"count" \| "time" \| "both"` | `"count"` | How to cut the change history into segments. `count` uses equal sample counts; `time` uses equal time spans; `both` requires the variable to be stable under each. With `count` the other two fields are ignored and no timestamps are recorded. | +| `timestamp_variable` | `str \| null` | `null` | Name of the field in `logFormatVariables` holding the record's event time. Required for `time` and `both` to have any effect. Only named log-format fields are consulted — never the positional `variables` list. | +| `timestamp_format` | `str \| null` | `null` | Explicit [`strftime`](https://docs.python.org/3/library/datetime.html#strftime-and-strptime-format-codes) pattern for parsing that field. When unset, `TimeFormatHandler` auto-detects the format (ISO 8601, Apache, syslog, numeric epoch seconds/milliseconds, and other common layouts). | + +Set `timestamp_format` when the source uses a layout the auto-detection does not +know. The HDFS loghub corpus, for example, stamps records as `081109 203615`, which +only parses with an explicit `"%y%m%d %H%M%S"`. + +#### Fallback behaviour + +Time-aware segmentation is best-effort and never fails a run: + +* If `stability_segmentation` is not `count` but `timestamp_variable` is unset, or the named field + is absent from a record, or its value cannot be parsed, the detector logs a + **single** warning (once per detector, so a bad config cannot flood the log) and + falls back to count-based segmentation. +* If timestamps stop lining up with the recorded observations, or the observed time + span is zero, or they arrive out of order, the classifier silently falls back to + count-based segmentation for that variable. +* Under `both`, any of the fallbacks above make the time pass reuse the count boundaries, + so the mode degrades to plain `count` rather than to an unconditional pass. + +In every fallback case classification still runs and produces a result — only the +segmentation rule changes back to the default. + +A segment with no observations in it is *not* a fallback: it scores a mean of 0.0, +because nothing observed means nothing changed. Equal-duration cuts of a bursty +variable leave such segments routinely, so `time` on its own is lenient towards a +burst of churn followed by silence. Use `both` when that leniency matters — the +count pass keeps every segment populated. + + ### Saving state (persist) Detectors can persist their training state to disk (or cloud storage) so it diff --git a/src/detectmatelibrary/common/variable_detector.py b/src/detectmatelibrary/common/variable_detector.py index 8b57395a..928f1c45 100644 --- a/src/detectmatelibrary/common/variable_detector.py +++ b/src/detectmatelibrary/common/variable_detector.py @@ -13,11 +13,12 @@ ) from detectmatelibrary.utils.persistency.event_persistency import EventPersistency from detectmatelibrary.utils.data_buffer import BufferMode +from detectmatelibrary.utils.time_format_handler import TimeFormatHandler from detectmatelibrary.schemas import ParserSchema, DetectorSchema from detectmatelibrary.constants import GLOBAL_EVENT_ID from detectmatelibrary.tools.logging import logger -from typing import Any, Dict, Optional, cast +from typing import Any, Dict, Literal, Optional, cast from typing_extensions import override @@ -25,6 +26,15 @@ class VariableDetectorConfig(CoreDetectorConfig): use_stable_vars: bool = True use_static_vars: bool = True + # Stability segmentation. "count" cuts the classifier's segments at equal + # sample counts (the historical behaviour). "time" cuts them at equal + # durations instead. "both" requires the variable to pass under *both* + # segmentations. The two time-aware modes need a per-record event time, + # named here and read from the record's logFormatVariables. + stability_segmentation: Literal["count", "time", "both"] = "count" + timestamp_variable: str | None = None + timestamp_format: str | None = None # None -> TimeFormatHandler auto-detect + class VariableDetector(CoreDetector): """Abstract base for detectors that learn a per-variable model from @@ -44,17 +54,32 @@ class VariableDetector(CoreDetector): def __init__(self, name: str, config: VariableDetectorConfig) -> None: super().__init__(name=name, buffer_mode=BufferMode.NO_BUF, config=config) self.config: VariableDetectorConfig # type narrowing for IDE + self._time_handler = TimeFormatHandler() + self._warned_bad_timestamp = False self.persistency = EventPersistency( event_data_class=self._event_data_class(), - event_data_kwargs=self._event_data_kwargs(), + event_data_kwargs=self._with_segmentation(self._event_data_kwargs()), ) # auto config checks individual-variable stability to select features self.auto_conf_persistency = EventPersistency( event_data_class=self._event_data_class(), - event_data_kwargs=self._auto_conf_kwargs(), + event_data_kwargs=self._with_segmentation(self._auto_conf_kwargs()), ) self._register_persistency(self.persistency) + def _with_segmentation(self, kwargs: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]: + """Add the segmentation mode to tracker kwargs when it is not the + default. + + Done here rather than in _stability_kwargs so every + VariableDetector subclass is covered -- NewValueDetector + overrides neither construction hook and NewValueComboDetector + returns only a converter_function. + """ + if self.config.stability_segmentation == "count": + return kwargs + return {**(kwargs or {}), "segmentation": self.config.stability_segmentation} + # ---- construction hooks ------------------------------------------------- def _event_data_class(self) -> type: @@ -75,6 +100,43 @@ def _stability_kwargs(self) -> Dict[str, Any]: "detector_config": self.config.to_dict(method_id=name), } + def _warn_time_fallback_once(self, reason: str) -> None: + """Log the first time-dependent misconfiguration, then stay quiet. + + A bad config would otherwise emit one warning per record, so the + flag latches after the first message. + """ + if self._warned_bad_timestamp: + return + self._warned_bad_timestamp = True + logger.warning( + "%s: %s; falling back to count-based stability segmentation.", + self.name, reason, + ) + + def _timestamp(self, input_: ParserSchema) -> float | None: + """Resolve the record's event time, or None to use count + segmentation.""" + if self.config.stability_segmentation == "count": + return None + if not self.config.timestamp_variable: + # Selecting a time-aware mode without naming the field is an operator + # error, not an opt-out -- say so rather than silently no-op. + self._warn_time_fallback_once( + f"stability_segmentation is {self.config.stability_segmentation!r} " + "but timestamp_variable is not set" + ) + return None + raw = input_["logFormatVariables"].get(self.config.timestamp_variable) + ts = self._time_handler.parse_timestamp(str(raw or ""), self.config.timestamp_format) + if ts == "0": + self._warn_time_fallback_once( + f"timestamp_variable {self.config.timestamp_variable!r} is missing or " + f"unparseable (got {raw!r})" + ) + return None + return float(ts) + # ---- per-detector hooks ------------------------------------------------- def _prepare_variables(self, variables: Dict[str, Any], stage: str) -> Dict[str, Any]: @@ -112,6 +174,7 @@ def _ingest(self, input_: ParserSchema, variables: Dict[str, Any], event_id: Any event_id=event_id, event_template=input_["template"], named_variables=variables, + timestamp=self._timestamp(input_), ) def detect(self, input_: ParserSchema, output_: DetectorSchema) -> bool: # type: ignore @@ -175,6 +238,7 @@ def configure(self, input_: ParserSchema) -> None: # type: ignore event_template=input_["template"], variables=input_["variables"], named_variables=input_["logFormatVariables"], + timestamp=self._timestamp(input_), ) @override @@ -200,6 +264,9 @@ def set_configuration(self) -> None: if selected: variables[event_id] = selected old_persist = self.config.persist + old_segmentation = self.config.stability_segmentation + old_timestamp_variable = self.config.timestamp_variable + old_timestamp_format = self.config.timestamp_format config_dict = generate_detector_config( variable_selection=variables, detector_name=self.name, @@ -207,6 +274,9 @@ def set_configuration(self) -> None: ) self.config = type(self.config).from_dict(config_dict, self.name) self.config.persist = old_persist + self.config.stability_segmentation = old_segmentation + self.config.timestamp_variable = old_timestamp_variable + self.config.timestamp_format = old_timestamp_format events = self.config.events if isinstance(events, EventsConfig) and not events.events: logger.warning( diff --git a/src/detectmatelibrary/detectors/new_value_combo_detector.py b/src/detectmatelibrary/detectors/new_value_combo_detector.py index e166250b..84dc5f26 100644 --- a/src/detectmatelibrary/detectors/new_value_combo_detector.py +++ b/src/detectmatelibrary/detectors/new_value_combo_detector.py @@ -59,7 +59,9 @@ def __init__( # second-pass persistency to learn stability of variable combinations self.auto_conf_persistency_combos = persistency.EventPersistency( event_data_class=persistency.EventStabilityTracker, - event_data_kwargs={"converter_function": get_all_possible_combos}, + event_data_kwargs=self._with_segmentation( + {"converter_function": get_all_possible_combos} + ), ) self.inputs: list[ParserSchema] = [] @@ -101,6 +103,24 @@ def set_configuration(self, max_combo_size: int | None = None) -> None: every possible combo up front would explode combinatorially). """ old_persist = self.config.persist + segmentation_fields = { + "stability_segmentation": self.config.stability_segmentation, + "timestamp_variable": self.config.timestamp_variable, + "timestamp_format": self.config.timestamp_format, + } + + def restore_segmentation_fields() -> None: + """Carry the segmentation settings across a config reassignment. + + generate_detector_config only emits method_type / auto_config / + params / events, so every ``from_dict`` below resets these to their + defaults. The re-ingest loop calls ``_timestamp()`` under the pass-1 + config, so restoring only at the end would leave the combo trackers + timestamp-less. + """ + for field, value in segmentation_fields.items(): + setattr(self.config, field, value) + # pass 1: stable individual variables -> combos variable_combos = {} for event_id, tracker in self.auto_conf_persistency.get_events_data().items(): @@ -114,6 +134,7 @@ def set_configuration(self, max_combo_size: int | None = None) -> None: max_combo_size=max_combo_size or self.config.max_combo_size, ) self.config = NewValueComboDetectorConfig.from_dict(config_dict, self.name) + restore_segmentation_fields() # re-ingest all inputs to learn combos under the new configuration for input_ in self.inputs: @@ -122,6 +143,7 @@ def set_configuration(self, max_combo_size: int | None = None) -> None: event_id=input_["EventID"], event_template=input_["template"], named_variables=configured_variables, + timestamp=self._timestamp(input_), ) # pass 2: stable/static combos -> final config @@ -148,6 +170,7 @@ def set_configuration(self, max_combo_size: int | None = None) -> None: ) self.config = NewValueComboDetectorConfig.from_dict(config_dict, self.name) self.config.persist = old_persist + restore_segmentation_fields() events = self.config.events if isinstance(events, EventsConfig) and not events.events: logger.warning( diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/base.py b/src/detectmatelibrary/utils/persistency/event_data_structures/base.py index 08e4909b..31eb768a 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/base.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/base.py @@ -11,7 +11,7 @@ class EventDataStructure(ABC): template: str = "" @abstractmethod - def add_data(self, data_object: Any) -> None: ... + def add_data(self, data_object: Any, timestamp: float | None = None) -> None: ... @abstractmethod def get_data(self) -> Any: ... diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/dataframes/chunked_event_dataframe.py b/src/detectmatelibrary/utils/persistency/event_data_structures/dataframes/chunked_event_dataframe.py index 53c32306..fac9c73c 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/dataframes/chunked_event_dataframe.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/dataframes/chunked_event_dataframe.py @@ -24,7 +24,7 @@ class ChunkedEventDataFrame(EventDataStructure): chunks: list[pl.DataFrame] = field(default_factory=list) _rows: int = 0 - def add_data(self, data: pl.DataFrame) -> None: + def add_data(self, data: pl.DataFrame, timestamp: float | None = None) -> None: if data.height == 0: return self.chunks.append(data) diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/dataframes/event_dataframe.py b/src/detectmatelibrary/utils/persistency/event_data_structures/dataframes/event_dataframe.py index 2a024883..5519393e 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/dataframes/event_dataframe.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/dataframes/event_dataframe.py @@ -17,7 +17,7 @@ class EventDataFrame(EventDataStructure): """ data: pd.DataFrame = field(default_factory=pd.DataFrame) - def add_data(self, data: pd.DataFrame) -> None: + def add_data(self, data: pd.DataFrame, timestamp: float | None = None) -> None: if len(self.data) > 0: self.data = pd.concat([self.data, data], ignore_index=True) else: diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/event_tracker.py b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/event_tracker.py index f43a22a7..7e12103a 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/event_tracker.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/event_tracker.py @@ -27,9 +27,9 @@ def __init__( self.converter_function = converter_function self.multi_tracker = self.multi_tracker_type(single_tracker_type=self.single_tracker_type) - def add_data(self, data_object: Any) -> None: + def add_data(self, data_object: Any, timestamp: float | None = None) -> None: """Add data to the variable trackers.""" - self.multi_tracker.add_data(data_object) + self.multi_tracker.add_data(data_object, timestamp=timestamp) def get_data(self) -> Dict[str, SingleTracker]: """Retrieve the tracker's stored data.""" @@ -72,7 +72,7 @@ def load(cls, data: bytes, **kwargs: Any) -> "EventTracker": ``multi_tracker_type`` recorded in the snapshot. For any subclass, ``cls(**kwargs)`` is called instead, which lets subclasses with closure-based factories (e.g. ``EventStabilityTracker``'s - ``expand_value``) rebuild their factory so it survives load. + ``segmentation``) rebuild their factory so it survives load. Contract for subclasses: ``__init__`` must accept the kwargs forwarded to ``load()`` and must not require additional positional arguments. diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/multi_tracker.py b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/multi_tracker.py index 08df8218..19aa2d28 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/multi_tracker.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/multi_tracker.py @@ -13,12 +13,12 @@ def __init__(self, single_tracker_type: Callable[[], SingleTracker] = SingleTrac self.single_trackers: Dict[str, SingleTracker] = {} self.single_tracker_type: Callable[[], SingleTracker] = single_tracker_type - def add_data(self, data_object: Dict[str, Any]) -> None: + def add_data(self, data_object: Dict[str, Any], timestamp: float | None = None) -> None: """Add data to the appropriate feature trackers.""" for name, value in data_object.items(): if name not in self.single_trackers: self.single_trackers[name] = self.single_tracker_type() - self.single_trackers[name].add_value(value) + self.single_trackers[name].add_value(value, timestamp=timestamp) def get_trackers(self) -> Dict[str, SingleTracker]: """Get the current feature trackers.""" diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/single_tracker.py b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/single_tracker.py index 01808d89..c5c12a43 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/single_tracker.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/base/single_tracker.py @@ -15,7 +15,7 @@ class SingleTracker(ABC): """Tracks whether a single variable is converging to a constant value.""" @abstractmethod - def add_value(self, value: Any) -> None: + def add_value(self, value: Any, timestamp: float | None = None) -> None: """Add a new value to the tracker.""" @abstractmethod diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_classifier.py b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_classifier.py index 25db8bc2..c629547c 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_classifier.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_classifier.py @@ -18,22 +18,74 @@ def __init__(self, segment_thresholds: List[float], min_samples: int = 10): # for lists self.segment_means: List[float] = [] - def is_stable(self, change_series: RLEList[bool] | List[bool]) -> bool: + def _segment_boundaries(self, total_len: int, timestamps: List[float] | None = None) -> List[int]: + """Index boundaries of n_segments segments over total_len items. + + Equal-count by default. When timestamps are given (one per item, + non-decreasing, non-zero span), boundaries are equal-DURATION + cuts of the observed time span, mapped back to indices. Falls + back to equal-count on missing / mismatched / non-finite / out- + of-order timestamps and on zero span. A duration cut that leaves + a segment empty is kept as-is: nothing observed in that window + means no changes in it, which ``is_stable`` scores as a mean of + 0.0. + """ + segment_size = total_len / self.n_segments + count_boundaries = [int(i * segment_size) for i in range(self.n_segments + 1)] + count_boundaries[-1] = total_len + try: + use_time = ( + timestamps is not None + and len(timestamps) == total_len + and total_len > 0 + and bool(np.all(np.isfinite(timestamps))) + # np.searchsorted below requires sorted input. Merged sources or + # concurrent writers can deliver stamps out of order, which would + # silently produce wrong boundaries rather than an error. O(N), + # same cost as the isfinite scan above. + and bool(np.all(np.diff(timestamps) >= 0)) + and timestamps[-1] > timestamps[0] + ) + except TypeError: + # e.g. a None entry: not comparable/convertible -> equal-count + use_time = False + if use_time and timestamps is not None: # 2nd clause narrows for mypy + t_first, t_last = timestamps[0], timestamps[-1] + cuts = [ + t_first + k * (t_last - t_first) / self.n_segments + for k in range(self.n_segments + 1) + ] + boundaries = [int(np.searchsorted(timestamps, t, side="left")) for t in cuts] + boundaries[0] = 0 + boundaries[-1] = total_len + return boundaries + return count_boundaries + + def is_stable( + self, + change_series: RLEList[bool] | List[bool], + timestamps: List[float] | None = None, + ) -> bool: """Determine if a list of segment means is stable. Works efficiently with RLEList without expanding to a full list. + When timestamps are given (one per occurrence, non-decreasing), + segments are equal-duration cuts of the time span instead of + equal-count cuts of the series. See ``_segment_boundaries`` for + the conditions under which time mode falls back to count mode. + + A segment with no observations in it scores a mean of 0.0 -- no + occurrences means no changes. Equal-duration cuts of a bursty + series leave such segments routinely; pair ``time`` with + ``count`` (segmentation ``both``) if that leniency matters. """ - # Handle both RLEList and regular list - if isinstance(change_series, RLEList): - total_len = len(change_series) - if total_len == 0: - return True + total_len = len(change_series) + if total_len == 0: + return True - # Calculate segment boundaries - segment_size = total_len / self.n_segments - segment_boundaries = [int(i * segment_size) for i in range(self.n_segments + 1)] - segment_boundaries[-1] = total_len + segment_boundaries = self._segment_boundaries(total_len, timestamps) + if isinstance(change_series, RLEList): # Compute segment means directly from RLE runs segment_sums = [0.0] * self.n_segments segment_counts = [0] * self.n_segments @@ -59,29 +111,28 @@ def is_stable(self, change_series: RLEList[bool] | List[bool]) -> bool: position = run_end - # Calculate means self.segment_means = [ - segment_sums[i] / segment_counts[i] if segment_counts[i] > 0 else np.nan + segment_sums[i] / segment_counts[i] if segment_counts[i] > 0 else 0.0 for i in range(self.n_segments) ] else: - # Original implementation for regular lists - self.segment_means = self._compute_segment_means(change_series) + self.segment_means = [ + float(np.mean(change_series[segment_boundaries[i]:segment_boundaries[i + 1]])) + if segment_boundaries[i + 1] > segment_boundaries[i] else 0.0 + for i in range(self.n_segments) + ] return all([not q >= thresh for q, thresh in zip(self.segment_means, self.segment_threshs)]) - def _compute_segment_means(self, change_series: List[bool]) -> List[float]: - """Get means of each segment for a normal list.""" - segments = np.array_split(change_series, self.n_segments) - return list(map(lambda x: np.mean(x) if len(x) > 0 else np.nan, segments)) - def get_last_segment_means(self) -> List[float]: return self.segment_means def get_segment_thresholds(self) -> List[float]: return self.segment_threshs - def __call__(self, change_series: RLEList[bool] | List[bool]) -> bool: - return self.is_stable(change_series) + def __call__( + self, change_series: RLEList[bool] | List[bool], timestamps: List[float] | None = None + ) -> bool: + return self.is_stable(change_series, timestamps=timestamps) def __repr__(self) -> str: return ( diff --git a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_tracker.py b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_tracker.py index 72e79d58..a66ea4b3 100644 --- a/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_tracker.py +++ b/src/detectmatelibrary/utils/persistency/event_data_structures/trackers/stability/stability_tracker.py @@ -43,33 +43,60 @@ def _strip_persist(detector_config: Any, method_id: str) -> Any: class SingleStabilityTracker(SingleTracker): """Tracks stability of a single feature.""" - def __init__(self, min_samples: int = 3, add_value_fn: str = "default", - detector_config: "CoreDetectorConfig | None" = None) -> None: + def __init__( + self, + min_samples: int = 3, + segmentation: Literal["count", "time", "both"] = "count", + add_value_fn: str = "default", + detector_config: "CoreDetectorConfig | None" = None, + ) -> None: self.min_samples = min_samples + self.segmentation = segmentation self.change_series: RLEList[bool] = RLEList() self.unique_set: Set[Any] = set() self.stability_classifier: StabilityClassifier = StabilityClassifier( segment_thresholds=[1.1, 0.3, 0.1, 0.01], ) + # ponytail: O(N) timestamps; switch to fixed-width time buckets if + # this ever runs unbounded/streaming. + self.timestamps: List[float] = [] # Opaque slot for detectors to stash per-variable model state that # must survive save/load. Schema-free; the tracker does not interpret it. self.extra_state: Dict[str, Any] = {} self.add_value_fn = add_value_fn self.detector_config = detector_config + # Transient: set by _is_stable() for classify()'s reason string. Not + # persisted -- it is derived from change_series on every classify(). + self._stability_note: str = "" + self._value_fn: Callable[[Any], None] = self._default_add_value if add_value_fn != "default": detector_cls = getattr(importlib.import_module("detectmatelibrary.detectors"), add_value_fn) if detector_config is not None: detector = detector_cls(config=_strip_persist(detector_config, add_value_fn)) else: detector = detector_cls() - self.add_value = partial(detector.add_value, self) # type: ignore[method-assign] + self._value_fn = partial(detector.add_value, self) - def add_value(self, value: Any) -> None: - """Add a new value to the tracker.""" + def _default_add_value(self, value: Any) -> None: + """Default value semantics: one set entry per whole value.""" before = len(self.unique_set) self.unique_set.add(value) self.change_series.append(len(self.unique_set) > before) + def add_value(self, value: Any, timestamp: float | None = None) -> None: + """Add a new value to the tracker. + + Value semantics belong to ``_value_fn`` -- either the default above or a + detector's ``add_value``. Timestamp bookkeeping stays here so + ``timestamps`` cannot drift from ``change_series``: a detector may record + nothing for a value (ValueRangeDetector on non-numeric input), and a + length mismatch silently demotes the variable to count segmentation. + """ + before = len(self.change_series) + self._value_fn(value) + if self.segmentation != "count" and timestamp is not None and len(self.change_series) > before: + self.timestamps.append(float(timestamp)) + def classify(self) -> Classification: """Classify the variable.""" if len(self.change_series) < self.min_samples: @@ -87,20 +114,65 @@ def classify(self) -> Classification: type="RANDOM", reason=f"Unique set size equals number of samples ({len(self.change_series)})" ) - elif self.stability_classifier.is_stable(self.change_series): + elif self._is_stable(): return Classification( type="STABLE", reason=( - f"Segment means of change series {self.stability_classifier.get_last_segment_means()} " - f"are below segment thresholds: {self.stability_classifier.get_segment_thresholds()}" + f"{self._stability_note} are below segment thresholds: " + f"{self.stability_classifier.get_segment_thresholds()}" ) ) else: return Classification( type="UNSTABLE", - reason="No classification matched; variable is unstable" + reason=( + f"{self._stability_note} exceed segment thresholds: " + f"{self.stability_classifier.get_segment_thresholds()}" + ) ) + def _is_stable(self) -> bool: + """Stability verdict under the configured segmentation. + + Sets ``_stability_note`` for ``classify()``'s reason string. + + ``both`` runs the count pass and the time pass over the same + change series and requires both. Neither segmentation subsumes + the other -- a variable that churns in a burst and then settles is + count-UNSTABLE but time-STABLE, and one whose late churn is buried + under a dense settled tail is the reverse -- so the conjunction is + strictly stricter than either input. + + Deliberately not short-circuited: both passes always run so the + note carries both mean vectors, which is what anyone debugging a + ``both`` verdict needs. Costs one extra O(runs x n_segments) scan. + """ + clf, ts = self.stability_classifier, self._aligned_timestamps() + if self.segmentation != "both": + stable = clf.is_stable(self.change_series, timestamps=ts) + self._stability_note = f"Segment means of change series {clf.get_last_segment_means()}" + return stable + count_stable = clf.is_stable(self.change_series) + # Snapshot now, not after the time pass: is_stable() rebinds + # clf.segment_means to a fresh list on every call, so calling + # get_last_segment_means() after the time pass below would return + # the time means for both halves of the note instead of the count + # means it is meant to capture here. + count_means = clf.get_last_segment_means() + time_stable = clf.is_stable(self.change_series, timestamps=ts) + self._stability_note = ( + f"Segment means of change series: count {count_means}, " + f"time {clf.get_last_segment_means()}" + ) + return count_stable and time_stable + + def _aligned_timestamps(self) -> List[float] | None: + """Timestamps to classify with, or None to fall back to count + segments.""" + if self.segmentation != "count" and len(self.timestamps) == len(self.change_series): + return self.timestamps + return None + def to_state(self) -> Dict[str, Any]: """Serialize tracker state to a plain dict (must be msgpack- compatible).""" @@ -108,6 +180,8 @@ def to_state(self) -> Dict[str, Any]: "type": self.__class__.__name__, "module": self.__class__.__module__, "min_samples": self.min_samples, + "segmentation": self.segmentation, + "timestamps": self.timestamps, "add_value_fn": self.add_value_fn, "detector_config": self.detector_config, "runs": self.change_series.runs(), @@ -119,10 +193,14 @@ def to_state(self) -> Dict[str, Any]: @classmethod def from_state(cls, state: Dict[str, Any]) -> "SingleStabilityTracker": """Restore tracker from a state dict produced by to_state().""" + # Every optional key is read with .get(): a snapshot old enough to still + # carry the removed `expand_value` predates `add_value_fn` too, so + # indexing here would KeyError on exactly the states this tolerance is for. tracker = cls( min_samples=state["min_samples"], - add_value_fn=state["add_value_fn"], - detector_config=state["detector_config"] + segmentation=state.get("segmentation", "count"), + add_value_fn=state.get("add_value_fn", "default"), + detector_config=state.get("detector_config"), ) runs = [(bool(r[0]), int(r[1])) for r in state["runs"]] tracker.change_series._runs = runs @@ -133,6 +211,7 @@ def from_state(cls, state: Dict[str, Any]) -> "SingleStabilityTracker": tracker.stability_classifier = StabilityClassifier( segment_thresholds=state["segment_thresholds"] ) + tracker.timestamps = [float(t) for t in state.get("timestamps", [])] tracker.extra_state = state.get("extra_state", {}) return tracker @@ -172,6 +251,7 @@ class EventStabilityTracker(EventTracker): def __init__( self, converter_function: Callable[[Any], Any] = lambda x: x, + segmentation: Literal["count", "time", "both"] = "count", add_value_fn: str = "default", detector_config: "CoreDetectorConfig | None" = None @@ -179,7 +259,11 @@ def __init__( self.multi_tracker: MultiStabilityTracker # for type hinting def make_tracker() -> SingleStabilityTracker: - return SingleStabilityTracker(add_value_fn=add_value_fn, detector_config=detector_config) + return SingleStabilityTracker( + segmentation=segmentation, + add_value_fn=add_value_fn, + detector_config=detector_config, + ) # Mirror class identity onto the closure so dump()/load() can resolve # the underlying SingleStabilityTracker via its module + qualname. diff --git a/src/detectmatelibrary/utils/persistency/event_persistency.py b/src/detectmatelibrary/utils/persistency/event_persistency.py index ba9ae18c..8dd22d39 100644 --- a/src/detectmatelibrary/utils/persistency/event_persistency.py +++ b/src/detectmatelibrary/utils/persistency/event_persistency.py @@ -44,7 +44,8 @@ def ingest_event( event_id: int | str, event_template: str, variables: list[Any] = [], - named_variables: Dict[str, Any] = {} + named_variables: Dict[str, Any] = {}, + timestamp: float | None = None, ) -> None: """Ingest event data into the appropriate EventData store.""" with self._lock: @@ -60,7 +61,7 @@ def ingest_event( self.events_data[event_id] = data_structure data = data_structure.to_data(all_variables) - data_structure.add_data(data) + data_structure.add_data(data, timestamp=timestamp) # ponytail: fire callbacks outside the lock so a count-triggered save # doesn't hold the ingest lock across serialize + file I/O. for _cb in self._on_ingest_callbacks: diff --git a/tests/test_detectors/test_new_value_combo_detector.py b/tests/test_detectors/test_new_value_combo_detector.py index 5d5fcd79..cc1f319e 100644 --- a/tests/test_detectors/test_new_value_combo_detector.py +++ b/tests/test_detectors/test_new_value_combo_detector.py @@ -1,5 +1,8 @@ -from detectmatelibrary.detectors.new_value_combo_detector import NewValueComboDetector +from detectmatelibrary.detectors.new_value_combo_detector import ( + NewValueComboDetector, + NewValueComboDetectorConfig, +) from detectmatelibrary.utils.data_buffer import BufferMode from detectmatelibrary.common._config import generate_detector_config from detectmatelibrary.parsers.template_matcher import MatcherParser @@ -577,3 +580,101 @@ def test_audit_log_anomalies(self): detected_ids.add(log["logID"]) assert detected_ids == {"1859", "1862", "1865", "1866"} + + +class TestNewValueComboDetectorSegmentationConfigPreservation: + """set_configuration() reassigns self.config twice (pass 1: combo + candidates, pass 2: final selection), each from a freshly generated config + dict whose params only ever carry max_combo_size. + + stability_segmentation, timestamp_variable and timestamp_format must + survive both reassignments, the same way persist already does. + """ + + def test_segmentation_fields_survive_set_configuration(self): + cfg = NewValueComboDetectorConfig( + stability_segmentation="time", + timestamp_variable="level", + timestamp_format="%y%m%d %H%M%S", + ) + detector = NewValueComboDetector(config=cfg, name="NewValueComboDetector") + assert detector.config.auto_config is True # the default (set_configuration-first) path + + for i in range(10): + parser_data = schemas.ParserSchema({ + "parserType": "test", + "EventID": 1, + "template": "Template 1", + "variables": ["constant", f"varying_{i}", "another_constant"], + "logID": str(i), + "parsedLogID": str(i), + "parserID": "test_parser", + "log": "test log", + "logFormatVariables": {"level": "081109 203615"}, + }) + detector.configure(parser_data) + + detector.set_configuration(max_combo_size=2) + + assert detector.config.stability_segmentation == "time" + assert detector.config.timestamp_variable == "level" + assert detector.config.timestamp_format == "%y%m%d %H%M%S" + + +class TestNewValueComboDetectorSegmentationCombos: + """The combo-stability pass must honour stability_segmentation too. + + auto_conf_persistency_combos is built directly in __init__ rather than from + the _event_data_kwargs hook, and its re-ingest loop in set_configuration + calls ingest_event itself -- so both halves of the flag (the tracker kwarg + and the per-record timestamp) have to be wired up explicitly. A flag that + reaches the first-pass trackers but not the combo trackers is worse than no + flag at all: the generated config would be selected on a different rule + than the one the operator asked for. + """ + + @staticmethod + def _records(segmentation="time"): + detector = NewValueComboDetector( + config=NewValueComboDetectorConfig( + stability_segmentation=segmentation, + timestamp_variable="ts", + ), + name="NewValueComboDetector", + ) + for i in range(12): + # var_0 and var_1 both cycle over 3 values -> both STABLE in pass 1, + # so pass 2 gets a ("var_0", "var_1") combo tracker to look at. + detector.configure(schemas.ParserSchema({ + "parserType": "test", + "EventID": 1, + "template": "Template 1", + "variables": [f"a{i % 3}", f"b{i % 3}", "constant"], + "logID": str(i), + "parsedLogID": str(i), + "parserID": "test_parser", + "log": "test log", + "logFormatVariables": {"ts": str(1700000000 + i * 60)}, + })) + return detector + + def test_combo_trackers_record_timestamps(self): + detector = self._records() + detector.set_configuration(max_combo_size=2) + + combo_trackers = detector.auto_conf_persistency_combos.get_events_data()[1].get_data() + assert ("var_0", "var_1") in combo_trackers + tracker = combo_trackers[("var_0", "var_1")] + assert tracker.segmentation == "time" + assert len(tracker.timestamps) == len(tracker.change_series) == 12 + assert tracker.timestamps[1] - tracker.timestamps[0] == 60.0 + + def test_combo_trackers_stay_count_based_when_flag_is_off(self): + detector = self._records(segmentation="count") + detector.set_configuration(max_combo_size=2) + + tracker = detector.auto_conf_persistency_combos.get_events_data()[1].get_data()[ + ("var_0", "var_1") + ] + assert tracker.segmentation == "count" + assert tracker.timestamps == [] diff --git a/tests/test_persistency/test_time_dependent_stability.py b/tests/test_persistency/test_time_dependent_stability.py new file mode 100644 index 00000000..182e784f --- /dev/null +++ b/tests/test_persistency/test_time_dependent_stability.py @@ -0,0 +1,633 @@ +"""Tests for the stability_segmentation option of the stability trackers.""" + +import logging + +import detectmatelibrary.schemas as schemas +from detectmatelibrary.detectors.charset_detector import CharsetDetector, CharsetDetectorConfig +from detectmatelibrary.utils.persistency.rle_list import RLEList +from detectmatelibrary.utils.persistency import EventPersistency +from detectmatelibrary.utils.persistency.event_data_structures.trackers import ( + StabilityClassifier, + SingleStabilityTracker, + EventStabilityTracker, +) + +THRESHOLDS = [1.1, 0.3, 0.1, 0.01] # same defaults SingleStabilityTracker uses + + +def make_classifier() -> StabilityClassifier: + return StabilityClassifier(segment_thresholds=THRESHOLDS) + + +# Divergence fixture: 3 changes up front, then a quiet tail of 37. +# In *time*, the changes span most of the observed window and the quiet +# tail is a burst compressed into ~0.04s at the end. +# +# The three change stamps are spaced so that each equal-duration quarter of the +# window holds exactly one occurrence (boundaries [0, 1, 2, 3, 40]), which is +# what puts a lone mean of 1.0 into a quarter whose threshold is 0.3. +DIVERGENT_SERIES = [True, True, True] + [False] * 37 +DIVERGENT_TIMES = [0.0, 30.0, 60.0] + [90.0 + 0.001 * i for i in range(37)] + +# Burst fixture: 20 fresh values, each immediately repeated once, inside 40 ms, +# then a single further repeat an hour later. Equal-duration quarters put all 40 +# early occurrences in quarter 0 and leave quarters 1 and 2 with nothing in them, +# which scores them 0.0 -- so time mode calls this churning variable STABLE and +# only count mode (or `both`) catches it. +BURSTY_VALUES = [f"v{i // 2}" for i in range(40)] + ["v19"] +BURSTY_SERIES = [True, False] * 20 + [False] +BURSTY_TIMES = [0.001 * i for i in range(40)] + [3600.0] + +# Opposite-direction divergence fixture. DIVERGENT_* above is count-STABLE and +# time-UNSTABLE; this one is count-UNSTABLE and time-STABLE. 30 fresh values one +# second apart, then the same value repeated 10 times spread over ~17 minutes. +# +# count quarters -> means [1.0, 1.0, 1.0, 0.0] -> UNSTABLE (segments 2 and 3) +# time quarters -> means [0.938, 0.0, 0.0, 0.0] -> STABLE (0.938 < 1.1) +# +# The pair of fixtures is what makes "both" testable in each direction: neither +# segmentation subsumes the other. +CHURN_VALUES = [f"v{i}" for i in range(30)] + ["v29"] * 10 +CHURN_TIMES = [float(i) for i in range(30)] + [100.0 * (i + 1) for i in range(10)] + +# Both segmentations agree on STABLE: two values, then one of them repeated, +# evenly spaced so the duration cuts coincide with the count cuts. +# Both give means [0.2, 0.0, 0.0, 0.0]. +AGREE_VALUES = ["a", "b"] + ["b"] * 38 +AGREE_TIMES = [float(i) for i in range(40)] + + +class TestClassifierTimeBoundaries: + def test_count_mode_is_stable_on_divergent_fixture(self): + clf = make_classifier() + # count segments of 10: means [0.3, 0, 0, 0] -> all below thresholds + assert clf.is_stable(RLEList(DIVERGENT_SERIES)) is True + + def test_time_mode_is_unstable_on_divergent_fixture(self): + clf = make_classifier() + # time quarters put a lone change (mean 1.0) into segment 2 (thresh 0.3) + assert clf.is_stable(RLEList(DIVERGENT_SERIES), timestamps=DIVERGENT_TIMES) is False + + def test_uniform_timestamps_match_count_mode(self): + # N divisible by n_segments -> boundaries coincide exactly + series = [True, False, False, True, False, False, False, False] + ts = [float(i) for i in range(8)] + clf_count = make_classifier() + clf_count.is_stable(RLEList(series)) + clf_time = make_classifier() + clf_time.is_stable(RLEList(series), timestamps=ts) + assert clf_time.get_last_segment_means() == clf_count.get_last_segment_means() + + def test_plain_list_path_supports_timestamps(self): + clf = make_classifier() + assert clf.is_stable(list(DIVERGENT_SERIES), timestamps=DIVERGENT_TIMES) is False + + def test_zero_span_falls_back_to_count_mode(self): + series = [True, False, False, False, False, False, False, False] + clf_count = make_classifier() + expected = clf_count.is_stable(RLEList(series)) + clf_time = make_classifier() + result = clf_time.is_stable(RLEList(series), timestamps=[5.0] * 8) + assert result == expected + assert clf_time.get_last_segment_means() == clf_count.get_last_segment_means() + + def test_length_mismatch_falls_back_to_count_mode(self): + series = [True, False, False, False, False, False, False, False] + clf_count = make_classifier() + expected = clf_count.is_stable(RLEList(series)) + clf_time = make_classifier() + assert clf_time.is_stable(RLEList(series), timestamps=[1.0, 2.0]) == expected + + def test_none_timestamp_entry_falls_back_to_count_mode(self): + series = [True, False, False, False, False, False, False, False] + clf_count = make_classifier() + expected = clf_count.is_stable(RLEList(series)) + clf_time = make_classifier() + ts = [0.0, 1.0, None, 3.0, 4.0, 5.0, 6.0, 7.0] + assert clf_time.is_stable(RLEList(series), timestamps=ts) == expected + assert clf_time.get_last_segment_means() == clf_count.get_last_segment_means() + + def test_nan_timestamp_entry_falls_back_to_count_mode(self): + series = [True, False, False, False, False, False, False, False] + clf_count = make_classifier() + expected = clf_count.is_stable(RLEList(series)) + clf_time = make_classifier() + ts = [0.0, 1.0, float("nan"), 3.0, 4.0, 5.0, 6.0, 7.0] + assert clf_time.is_stable(RLEList(series), timestamps=ts) == expected + assert clf_time.get_last_segment_means() == clf_count.get_last_segment_means() + + def test_empty_time_segment_scores_zero(self): + """An empty segment means nothing was observed in that window, so its + mean is 0.0 -- the boundaries are kept, not discarded. + + Equal-duration cuts of BURSTY_TIMES give boundaries [0, 40, 40, + 40, 41] and means [0.5, 0.0, 0.0, 0.0], which passes: time mode + alone is lenient on a burst followed by silence. Count mode + still sees the churn, so `both` catches it. + """ + clf_count = make_classifier() + assert clf_count.is_stable(RLEList(BURSTY_SERIES)) is False + + clf_time = make_classifier() + assert clf_time.is_stable(RLEList(BURSTY_SERIES), timestamps=BURSTY_TIMES) is True + assert clf_time.get_last_segment_means() == [0.5, 0.0, 0.0, 0.0] + + def test_empty_time_segment_scores_zero_on_plain_list_path(self): + clf_time = make_classifier() + assert clf_time.is_stable(list(BURSTY_SERIES), timestamps=BURSTY_TIMES) is True + assert clf_time.get_last_segment_means() == [0.5, 0.0, 0.0, 0.0] + + def test_out_of_order_timestamps_fall_back_to_count_mode(self): + """np.searchsorted requires sorted input. + + UNSORTED_TIMES is SORTED_TIMES with two entries transposed -- + what a multi-source merge or concurrent writers produce. + Unguarded, the cuts land at [0, 9, 20, 31, 40] instead of [0, + 10, 20, 30, 40], turning the exact means [0.5, 0.5, 0.5, 0.5] + into [0.556, 0.455, 0.545, 0.444] -- wrong, and silently so. + """ + series = [True, False] * 20 + sorted_times = [float(i) for i in range(40)] + unsorted_times = list(sorted_times) + unsorted_times[9], unsorted_times[30] = unsorted_times[30], unsorted_times[9] + + clf_count = make_classifier() + expected = clf_count.is_stable(RLEList(series)) + clf_time = make_classifier() + assert clf_time.is_stable(RLEList(series), timestamps=unsorted_times) == expected + assert clf_time.get_last_segment_means() == clf_count.get_last_segment_means() + + def test_sorted_timestamps_still_use_time_mode(self): + """The monotonicity guard must not disable time mode for valid + input.""" + clf = make_classifier() + assert clf.is_stable(RLEList(DIVERGENT_SERIES), timestamps=DIVERGENT_TIMES) is False + count = make_classifier() + count.is_stable(RLEList(DIVERGENT_SERIES)) + assert clf.get_last_segment_means() != count.get_last_segment_means() + + def test_equal_timestamps_are_not_treated_as_out_of_order(self): + """Duplicate stamps are non-decreasing, so they stay in time mode.""" + series = [True, False] * 20 + times = [float(i // 2) for i in range(40)] # each stamp used twice + clf = make_classifier() + assert clf.is_stable(RLEList(series), timestamps=times) is False + assert not any(mean != mean for mean in clf.get_last_segment_means()) # no nan + + def test_list_and_rle_agree_on_ragged_length(self): + """13 items over 4 segments: both paths must cut identically.""" + series = [True, False, True] + [False] * 10 + clf_list = make_classifier() + clf_list.is_stable(list(series)) + clf_rle = make_classifier() + clf_rle.is_stable(RLEList(series)) + assert clf_list.get_last_segment_means() == clf_rle.get_last_segment_means() + + +def feed_divergent(tracker: SingleStabilityTracker) -> None: + """3 new values spread over ~80s, then a repeated value bursting at + t~100.""" + values = ["a", "b", "c"] + ["c"] * 37 + for value, ts in zip(values, DIVERGENT_TIMES): + tracker.add_value(value, timestamp=ts) + + +def feed_churn(tracker: SingleStabilityTracker) -> None: + """30 new values one second apart, then a repeated value over ~17 min.""" + for value, ts in zip(CHURN_VALUES, CHURN_TIMES): + tracker.add_value(value, timestamp=ts) + + +def feed_agreeing(tracker: SingleStabilityTracker) -> None: + """One change up front, then a settled value, evenly spaced.""" + for value, ts in zip(AGREE_VALUES, AGREE_TIMES): + tracker.add_value(value, timestamp=ts) + + +class TestSingleStabilityTrackerSegmentation: + def test_timestamps_stored_only_when_enabled(self): + on = SingleStabilityTracker(segmentation="time") + on.add_value("a", timestamp=1.0) + on.add_value("b", timestamp=2.0) + assert on.timestamps == [1.0, 2.0] + + off = SingleStabilityTracker() # default segmentation="count" + off.add_value("a", timestamp=1.0) + assert off.timestamps == [] + + def test_classification_diverges_between_modes(self): + count_mode = SingleStabilityTracker() + feed_divergent(count_mode) + assert count_mode.classify().type == "STABLE" + + time_mode = SingleStabilityTracker(segmentation="time") + feed_divergent(time_mode) + assert time_mode.classify().type == "UNSTABLE" + + def test_missing_timestamps_fall_back_to_count_mode(self): + # time segmentation on, but values arrive without timestamps + tracker = SingleStabilityTracker(segmentation="time") + for value in ["a", "b", "c"] + ["c"] * 37: + tracker.add_value(value) + reference = SingleStabilityTracker() + for value in ["a", "b", "c"] + ["c"] * 37: + reference.add_value(value) + assert tracker.classify().type == reference.classify().type + + def test_round_trip_preserves_time_state(self): + tracker = SingleStabilityTracker(segmentation="time") + feed_divergent(tracker) + restored = SingleStabilityTracker.from_state(tracker.to_state()) + assert restored.segmentation == "time" + assert restored.timestamps == tracker.timestamps + assert restored.classify().type == "UNSTABLE" + + def test_legacy_state_without_time_keys_defaults_off(self): + tracker = SingleStabilityTracker() + tracker.add_value("hello") + state = tracker.to_state() + state.pop("segmentation", None) # simulate pre-flag snapshot + state.pop("timestamps", None) + restored = SingleStabilityTracker.from_state(state) + assert restored.segmentation == "count" + assert restored.timestamps == [] + + def test_legacy_state_without_add_value_keys_loads(self): + """A snapshot old enough to predate add_value_fn/detector_config must + still load; those keys were indexed, not .get()-ed, so it raised + KeyError.""" + tracker = SingleStabilityTracker() + tracker.add_value("hello") + state = tracker.to_state() + state.pop("add_value_fn", None) + state.pop("detector_config", None) + state["expand_value"] = True # the flag such a snapshot would carry + restored = SingleStabilityTracker.from_state(state) + assert restored.add_value_fn == "default" + assert restored.detector_config is None + assert restored.unique_set == {"hello"} + + def test_bursty_series_needs_the_count_pass(self): + """A burst followed by silence is time-STABLE (empty quarters score + 0.0) but count-UNSTABLE, so only `both` refuses to hand it to auto- + config variable selection as a monitoring candidate.""" + trackers = { + mode: SingleStabilityTracker(segmentation=mode) + for mode in ("count", "time", "both") + } + for value, ts in zip(BURSTY_VALUES, BURSTY_TIMES): + for tracker in trackers.values(): + tracker.add_value(value, timestamp=ts) + assert trackers["count"].classify().type == "UNSTABLE" + assert trackers["time"].classify().type == "STABLE" + assert trackers["both"].classify().type == "UNSTABLE" + + +class TestSegmentationPlumbing: + def test_event_tracker_propagates_flag_and_timestamp(self): + event_tracker = EventStabilityTracker(segmentation="time") + event_tracker.add_data({"var1": "a"}, timestamp=1.0) + event_tracker.add_data({"var1": "b"}, timestamp=2.0) + single = event_tracker.get_data()["var1"] + assert single.segmentation == "time" + assert single.timestamps == [1.0, 2.0] + + def test_ingest_event_forwards_timestamp(self): + storage = EventPersistency( + EventStabilityTracker, + event_data_kwargs={"segmentation": "time"}, + ) + storage.ingest_event(1, "tpl <*>", variables=["a"], timestamp=10.0) + storage.ingest_event(1, "tpl <*>", variables=["b"], timestamp=20.0) + single = storage.get_events_data()[1].get_data()["var_0"] + assert single.timestamps == [10.0, 20.0] + + def test_ingest_event_without_timestamp_still_works(self): + storage = EventPersistency(EventStabilityTracker) + storage.ingest_event(1, "tpl <*>", variables=["a"]) + single = storage.get_events_data()[1].get_data()["var_0"] + assert list(single.change_series) == [True] + assert single.timestamps == [] + + def test_event_tracker_dump_load_preserves_timestamps(self): + event_tracker = EventStabilityTracker(segmentation="time") + event_tracker.add_data({"var1": "a"}, timestamp=1.0) + event_tracker.add_data({"var1": "b"}, timestamp=2.0) + restored = EventStabilityTracker.load(event_tracker.dump(), segmentation="time") + single = restored.get_data()["var1"] + assert single.segmentation == "time" + assert single.timestamps == [1.0, 2.0] + + +class TestSegmentationWithDetectorAddValueFn: + """Segmentation="time" must work when a detector owns the value + semantics.""" + + def test_detector_backed_tracker_records_timestamps(self): + tracker = SingleStabilityTracker( + add_value_fn="CharsetDetector", segmentation="time" + ) + tracker.add_value("ab", timestamp=1.0) + tracker.add_value("cd", timestamp=2.0) + assert tracker.unique_set == {"a", "b", "c", "d"} + assert tracker.timestamps == [1.0, 2.0] + assert len(tracker.timestamps) == len(tracker.change_series) + + def test_value_range_skipped_value_keeps_alignment(self): + """ValueRangeDetector returns early on non-numeric input without + appending to change_series; timestamps must not drift.""" + tracker = SingleStabilityTracker( + add_value_fn="ValueRangeDetector", segmentation="time" + ) + tracker.add_value("1", timestamp=1.0) + tracker.add_value("not-a-number", timestamp=2.0) # detector records nothing + tracker.add_value("5", timestamp=3.0) + assert len(tracker.change_series) == 2 + assert tracker.timestamps == [1.0, 3.0] + + def test_event_tracker_detector_backed_round_trip(self): + event_tracker = EventStabilityTracker( + add_value_fn="CharsetDetector", segmentation="time" + ) + event_tracker.add_data({"var1": "ab"}, timestamp=1.0) + event_tracker.add_data({"var1": "cd"}, timestamp=2.0) + restored = EventStabilityTracker.load( + event_tracker.dump(), add_value_fn="CharsetDetector", segmentation="time" + ) + single = restored.get_data()["var1"] + assert single.unique_set == {"a", "b", "c", "d"} + assert single.timestamps == [1.0, 2.0] + + +def _parser_record(ts_value, event_id=1): + return schemas.ParserSchema({ + "parserType": "test", + "EventID": event_id, + "template": "test template", + "variables": ["abc"], + "logID": "1", + "parsedLogID": "1", + "parserID": "test_parser", + "log": "test log message", + "logFormatVariables": {"ts": ts_value}, + }) + + +class TestTimestampResolution: + # Each test below constructs CharsetDetector(config=CharsetDetectorConfig()) + # explicitly rather than bare CharsetDetector(). CharsetDetector.__init__'s + # `config` default argument is a single shared CharsetDetectorConfig() + # instance (pre-existing mutable-default-arg pitfall, see + # TestSegmentationConfigWiring.test_flag_reaches_per_variable_trackers), and + # several tests here mutate `detector.config.*` in place -- writing through + # to that shared instance and leaking state into any other bare-constructed + # CharsetDetector for the rest of the process. Passing a fresh config keeps + # every test isolated regardless of run order. + def test_returns_none_when_not_configured(self): + detector = CharsetDetector(config=CharsetDetectorConfig()) + assert detector._timestamp(_parser_record("2026-08-04 10:00:00")) is None + + def test_parses_iso_timestamp(self): + detector = CharsetDetector(config=CharsetDetectorConfig()) + detector.config.stability_segmentation = "time" + detector.config.timestamp_variable = "ts" + assert detector._timestamp(_parser_record("2026-08-04 10:00:00")) == 1785837600.0 + + def test_parses_explicit_format(self): + """HDFS loghub style, absent from COMMON_TIME_FORMATS.""" + detector = CharsetDetector(config=CharsetDetectorConfig()) + detector.config.stability_segmentation = "time" + detector.config.timestamp_variable = "ts" + detector.config.timestamp_format = "%y%m%d %H%M%S" + first = detector._timestamp(_parser_record("081109 203615")) + second = detector._timestamp(_parser_record("081109 203645")) + assert second - first == 30.0 + + def test_unparseable_warns_once_and_falls_back(self, caplog): + detector = CharsetDetector(config=CharsetDetectorConfig()) + detector.config.stability_segmentation = "time" + detector.config.timestamp_variable = "ts" + with caplog.at_level(logging.WARNING): + assert detector._timestamp(_parser_record("not-a-time")) is None + assert detector._timestamp(_parser_record("also-not-a-time")) is None + warnings = [r for r in caplog.records if "timestamp_variable" in r.message] + assert len(warnings) == 1 + + def test_unset_timestamp_variable_warns_once_and_falls_back(self, caplog): + """stability_segmentation="time" without timestamp_variable is an + operator error, not an opt-out: it must be distinguishable from a + working time-dependent run, and must not flood the log.""" + detector = CharsetDetector(config=CharsetDetectorConfig()) + detector.config.stability_segmentation = "time" # timestamp_variable left unset + with caplog.at_level(logging.WARNING): + assert detector._timestamp(_parser_record("2026-08-04 10:00:00")) is None + assert detector._timestamp(_parser_record("2026-08-04 10:00:01")) is None + warnings = [r for r in caplog.records if "timestamp_variable" in r.message] + assert len(warnings) == 1 + assert "not set" in warnings[0].message + + def test_flag_off_stays_silent(self, caplog): + """No warning when the feature simply is not enabled.""" + detector = CharsetDetector(config=CharsetDetectorConfig()) + with caplog.at_level(logging.WARNING): + assert detector._timestamp(_parser_record("2026-08-04 10:00:00")) is None + assert not [r for r in caplog.records if "timestamp_variable" in r.message] + + def test_missing_variable_warns_and_falls_back(self, caplog): + detector = CharsetDetector(config=CharsetDetectorConfig()) + detector.config.stability_segmentation = "time" + detector.config.timestamp_variable = "absent" + with caplog.at_level(logging.WARNING): + assert detector._timestamp(_parser_record("2026-08-04 10:00:00")) is None + assert any("timestamp_variable" in r.message for r in caplog.records) + + +class TestSegmentationConfigWiring: + def test_flag_reaches_per_variable_trackers(self): + # CharsetDetector's `config` parameter default is a single shared + # CharsetDetectorConfig() instance (pre-existing mutable-default-arg + # pitfall, unrelated to stability_segmentation). Other tests in this + # module mutate `detector.config.*` in place on a bare + # CharsetDetector(), so we pass explicit fresh configs here to stay + # isolated from that. + detector = CharsetDetector(config=CharsetDetectorConfig()) + assert detector.persistency.event_data_kwargs.get("segmentation") is None + + configured = CharsetDetector(config=CharsetDetectorConfig()) + configured.config.stability_segmentation = "time" + rebuilt = CharsetDetector(config=configured.config.to_dict(method_id="CharsetDetector")) + assert rebuilt.persistency.event_data_kwargs["segmentation"] == "time" + + def test_config_fields_round_trip(self): + detector = CharsetDetector(config=CharsetDetectorConfig()) + detector.config.stability_segmentation = "time" + detector.config.timestamp_variable = "ts" + detector.config.timestamp_format = "%y%m%d %H%M%S" + restored = type(detector.config).from_dict( + detector.config.to_dict(method_id="CharsetDetector"), "CharsetDetector" + ) + assert restored.stability_segmentation == "time" + assert restored.timestamp_variable == "ts" + assert restored.timestamp_format == "%y%m%d %H%M%S" + + def test_train_populates_timestamps_end_to_end(self): + cfg = { + "detectors": { + "CharsetDetector": { + "method_type": "charset_detector", + "auto_config": False, + "params": { + "stability_segmentation": "time", + "timestamp_variable": "ts", + "timestamp_format": "%y%m%d %H%M%S", + }, + "events": { + 1: { + "inst": { + "params": {}, + "variables": [{"pos": 0, "name": "v", "params": {}}], + } + } + }, + } + } + } + detector = CharsetDetector(config=cfg, name="CharsetDetector") + detector.train(_parser_record("081109 203615")) + detector.train(_parser_record("081109 203645")) + tracker = detector.persistency.get_events_data()[1].get_data()["v"] + assert tracker.segmentation == "time" + assert len(tracker.timestamps) == len(tracker.change_series) == 2 + assert tracker.timestamps[1] - tracker.timestamps[0] == 30.0 + + def test_segmentation_fields_survive_auto_config_set_configuration(self): + """set_configuration() reassigns self.config wholesale from a config + dict generated with empty params (generate_detector_config only emits + method_type/auto_config/params/events), so stability_segmentation, + timestamp_variable and timestamp_format must be carried across that + reassignment explicitly -- same as `persist` already is. + + auto_config defaults to True and core.py runs + set_configuration() before train(), so this is the path every + detector takes unless auto_config is explicitly disabled. + """ + cfg = CharsetDetectorConfig( + stability_segmentation="time", + timestamp_variable="ts", + timestamp_format="%y%m%d %H%M%S", + ) + detector = CharsetDetector(config=cfg, name="CharsetDetector") + assert detector.config.auto_config is True + + for _ in range(5): + detector.configure(_parser_record("081109 203615")) + detector.set_configuration() + + assert detector.config.stability_segmentation == "time" + assert detector.config.timestamp_variable == "ts" + assert detector.config.timestamp_format == "%y%m%d %H%M%S" + + +class TestBothSegmentation: + """`both` is STABLE only when count and time segmentation agree.""" + + def test_rejects_when_only_time_is_unstable(self): + count_mode = SingleStabilityTracker() + feed_divergent(count_mode) + assert count_mode.classify().type == "STABLE" + + time_mode = SingleStabilityTracker(segmentation="time") + feed_divergent(time_mode) + assert time_mode.classify().type == "UNSTABLE" + + both_mode = SingleStabilityTracker(segmentation="both") + feed_divergent(both_mode) + assert both_mode.classify().type == "UNSTABLE" + + def test_rejects_when_only_count_is_unstable(self): + """The opposite direction: proves neither pass is dead code.""" + count_mode = SingleStabilityTracker() + feed_churn(count_mode) + assert count_mode.classify().type == "UNSTABLE" + + time_mode = SingleStabilityTracker(segmentation="time") + feed_churn(time_mode) + assert time_mode.classify().type == "STABLE" + + both_mode = SingleStabilityTracker(segmentation="both") + feed_churn(both_mode) + assert both_mode.classify().type == "UNSTABLE" + + def test_accepts_when_both_agree(self): + """`both` must not be vacuously strict.""" + for mode in ("count", "time", "both"): + tracker = SingleStabilityTracker(segmentation=mode) + feed_agreeing(tracker) + assert tracker.classify().type == "STABLE", mode + + def test_without_timestamps_matches_count_mode(self): + """No usable timestamps -> the time pass runs on count boundaries, so + `both` degrades to plain `count` rather than to a free pass. + + Uses CHURN_VALUES (count-UNSTABLE) precisely because an + implementation that degrades to an unconditional free pass would + also call an all-STABLE fixture STABLE here; only a fixture that + is UNSTABLE when fed without timestamps can tell the two apart. + """ + both_mode = SingleStabilityTracker(segmentation="both") + reference = SingleStabilityTracker() + for value in CHURN_VALUES: + both_mode.add_value(value) # no timestamp argument + reference.add_value(value) + assert reference.classify().type == "UNSTABLE" + assert both_mode.classify().type == "UNSTABLE" + + def test_reason_reports_both_mean_vectors(self): + """The note must carry the *actual* count and time mean vectors, not + just the words "count"/"time" -- and they must be distinct, which + catches a snapshot-ordering bug where the time pass's overwrite of + StabilityClassifier.segment_means leaks into the count half of the note + (see the comment at the count_means snapshot in _is_stable()). + + feed_churn is used because its count and time means genuinely + differ ([1.0, 1.0, 1.0, 0.0] vs [0.9375, 0.0, 0.0, 0.0]); it + also yields UNSTABLE, so this exercises the note on the branch + finding 1 wires it into. + """ + tracker = SingleStabilityTracker(segmentation="both") + feed_churn(tracker) + classification = tracker.classify() + assert classification.type == "UNSTABLE" + reason = classification.reason + assert "count [1.0, 1.0, 1.0, 0.0]" in reason + assert "time [0.9375, 0.0, 0.0, 0.0]" in reason + + def test_round_trip_preserves_both_mode(self): + tracker = SingleStabilityTracker(segmentation="both") + feed_churn(tracker) + restored = SingleStabilityTracker.from_state(tracker.to_state()) + assert restored.segmentation == "both" + assert restored.timestamps == tracker.timestamps + assert restored.classify().type == "UNSTABLE" + + def test_stability_note_is_not_persisted(self): + tracker = SingleStabilityTracker(segmentation="both") + feed_agreeing(tracker) + tracker.classify() + assert "_stability_note" not in tracker.to_state() + + def test_config_accepts_both_and_reaches_trackers(self): + configured = CharsetDetector(config=CharsetDetectorConfig()) + configured.config.stability_segmentation = "both" + rebuilt = CharsetDetector( + config=configured.config.to_dict(method_id="CharsetDetector") + ) + assert rebuilt.persistency.event_data_kwargs["segmentation"] == "both" + + def test_event_tracker_propagates_both(self): + event_tracker = EventStabilityTracker(segmentation="both") + event_tracker.add_data({"var1": "a"}, timestamp=1.0) + event_tracker.add_data({"var1": "b"}, timestamp=2.0) + single = event_tracker.get_data()["var1"] + assert single.segmentation == "both" + assert single.timestamps == [1.0, 2.0]