chore(anomaly detection): store datapoint even if we do not have enough history for AD

Author: ram-senth

src/seer/anomaly_detection/anomaly_detection.py

@@ -1,7 +1,7 @@
 import datetime
 import logging
 import random
-from typing import List, Tuple
+from typing import List, Optional, Tuple
 
 import numpy as np
 import pandas as pd
@@ -15,6 +15,7 @@
 from seer.anomaly_detection.detectors.prophet_anomaly_detector import ProphetAnomalyDetector
 from seer.anomaly_detection.models import (
     AlgoConfig,
+    DynamicAlert,
     MPTimeSeriesAnomalies,
     MPTimeSeriesAnomaliesSingleWindow,
     TimeSeries,
@@ -141,6 +142,57 @@ def _batch_detect(
 
         return timeseries, batch_anomalies, prophet_df
 
+    @sentry_sdk.trace
+    def _save_and_fire_pruning_task(
+        self,
+        timepoint: TimeSeriesPoint,
+        streamed_anomalies: MPTimeSeriesAnomaliesSingleWindow,
+        algo_data: Optional[dict],
+        historic: DynamicAlert,
+        config: AnomalyDetectionConfig,
+        alert_data_accessor: AlertDataAccessor,
+        queue_cleanup_predict_task: bool,
+        force_cleanup: bool,
+    ) -> None:
+        # Save new data point
+        alert_data_accessor.save_timepoint(
+            external_alert_id=historic.external_alert_id,
+            timepoint=timepoint,
+            anomaly=streamed_anomalies,
+            anomaly_algo_data=algo_data,
+        )
+
+        if queue_cleanup_predict_task:
+            # Delayed import due to circular imports
+            from seer.anomaly_detection.tasks import cleanup_timeseries_and_predict
+
+            try:
+                # Set flag and create new task for cleanup if we have enough history and too many old points or not enough predictions remaining
+                cleanup_predict_config = historic.cleanup_predict_config
+                if alert_data_accessor.can_queue_cleanup_predict_task(
+                    historic.external_alert_id
+                ) and (
+                    force_cleanup
+                    or (
+                        cleanup_predict_config.num_old_points
+                        >= cleanup_predict_config.num_acceptable_points
+                        or cleanup_predict_config.num_predictions_remaining
+                        <= cleanup_predict_config.num_acceptable_predictions
+                    )
+                ):
+                    alert_data_accessor.queue_data_purge_flag(historic.external_alert_id)
+                    cleanup_timeseries_and_predict.apply_async(
+                        (historic.external_alert_id, cleanup_predict_config.timestamp_threshold),
+                        countdown=random.randint(
+                            0, config.time_period * 45
+                        ),  # Wait between 0 - time_period * 45 seconds before queuing so the tasks are not all queued at the same time and stll have a chance to run before nex detection call
+                    )
+            except Exception as e:
+                # Reset task and capture exception
+                alert_data_accessor.reset_cleanup_predict_task(historic.external_alert_id)
+                sentry_sdk.capture_exception(e)
+                logger.exception(e)
+
     @inject
     @sentry_sdk.trace
     def _online_detect(
@@ -205,6 +257,28 @@ def _online_detect(
         # Confirm that there is enough data (after purge)
         min_data = self._min_required_timesteps(historic.config.time_period)
         if len(historic.timeseries.timestamps) < min_data:
+            # If there is not enough data, we will not detect any anomalies. We still should save the timepoint
+            # so that we can detect anomalies in the future.
+            self._save_and_fire_pruning_task(
+                timepoint=ts_external[0],
+                streamed_anomalies=MPTimeSeriesAnomaliesSingleWindow(
+                    flags=["none"] * len(ts_external),
+                    scores=[0.0] * len(ts_external),
+                    thresholds=[],
+                    matrix_profile=np.array([]),
+                    window_size=3,
+                    original_flags=[],
+                    confidence_levels=[],
+                ),
+                algo_data=None,
+                historic=historic,
+                config=config,
+                alert_data_accessor=alert_data_accessor,
+                queue_cleanup_predict_task=(
+                    len(historic.timeseries.timestamps) == min_data - 1
+                ),  # we queue cleanup task if we have just enough data to detect anomalies in the next call
+                force_cleanup=True,
+            )
             logger.error(f"Not enough timeseries data. At least {min_data} data points required")
             raise ClientError(
                 f"Not enough timeseries data. At least {min_data} data points required"
@@ -223,7 +297,6 @@ def _online_detect(
             raise ServerError("Invalid state")
 
         # Run stream detection
-
         # SuSS Window
         stream_detector = MPStreamAnomalyDetector(
             history_timestamps=historic.timeseries.timestamps,
@@ -276,40 +349,17 @@ def _online_detect(
         streamed_anomalies_online = alert_data_accessor.combine_anomalies(
             streamed_anomalies, None, anomalies.use_suss[-num_anomlies:]
         )
-
-        # Save new data point
-        alert_data_accessor.save_timepoint(
-            external_alert_id=alert.id,
+        self._save_and_fire_pruning_task(
             timepoint=ts_external[0],
-            anomaly=streamed_anomalies,
-            anomaly_algo_data=streamed_anomalies_online.get_anomaly_algo_data(len(ts_external))[0],
+            streamed_anomalies=streamed_anomalies,
+            algo_data=streamed_anomalies_online.get_anomaly_algo_data(len(ts_external))[0],
+            historic=historic,
+            config=config,
+            alert_data_accessor=alert_data_accessor,
+            queue_cleanup_predict_task=True,
+            force_cleanup=False,
         )
 
-        # Delayed import due to circular imports
-        from seer.anomaly_detection.tasks import cleanup_timeseries_and_predict
-
-        try:
-            # Set flag and create new task for cleanup if too many old points or not enough predictions remaining
-            cleanup_predict_config = historic.cleanup_predict_config
-            if alert_data_accessor.can_queue_cleanup_predict_task(historic.external_alert_id) and (
-                cleanup_predict_config.num_old_points
-                >= cleanup_predict_config.num_acceptable_points
-                or cleanup_predict_config.num_predictions_remaining
-                <= cleanup_predict_config.num_acceptable_predictions
-            ):
-                alert_data_accessor.queue_data_purge_flag(historic.external_alert_id)
-                cleanup_timeseries_and_predict.apply_async(
-                    (historic.external_alert_id, cleanup_predict_config.timestamp_threshold),
-                    countdown=random.randint(
-                        0, config.time_period * 60
-                    ),  # Wait between 0 - time_period * 60 seconds before queuing so the tasks are not all queued at the same time
-                )
-        except Exception as e:
-            # Reset task and capture exception
-            alert_data_accessor.reset_cleanup_predict_task(historic.external_alert_id)
-            sentry_sdk.capture_exception(e)
-            logger.exception(e)
-
         return ts_external, streamed_anomalies_online
 
     def _min_required_timesteps(self, time_period, min_num_days=7):