[TE] Add support for data filter in detection pipeline emulation layer
Add support for legacy data filter in the detection pipeline emulation layer.
@@ -6,6 +6,8 @@ import com.google.common.base.Preconditions;
import com.google.common.collect.Collections2;
import com.linkedin.pinot.pql.parsers.utils.Pair;
import com.linkedin.thirdeye.anomalydetection.context.AnomalyResult;
+import com.linkedin.thirdeye.anomalydetection.datafilter.DataFilter;
+import com.linkedin.thirdeye.anomalydetection.datafilter.DataFilterFactory;
import com.linkedin.thirdeye.api.DimensionMap;
import com.linkedin.thirdeye.api.MetricSchema;
import com.linkedin.thirdeye.api.MetricSpec;
@@ -50,6 +52,7 @@ public class LegacyAnomalyFunctionAlgorithm extends DetectionPipeline {
private BaseAnomalyFunction anomalyFunction;
private String metricUrn;
private MetricEntity metricEntity;
+ private DataFilter dataFilter;
/**
* Instantiates a new Legacy anomaly function algorithm.
@@ -69,6 +72,7 @@ public class LegacyAnomalyFunctionAlgorithm extends DetectionPipeline {
anomalyFunction = (BaseAnomalyFunction) Class.forName(anomalyFunctionClassName).newInstance();
String specs = OBJECT_MAPPER.writeValueAsString(MapUtils.getMap(config.getProperties(), PROP_SPEC));
anomalyFunction.init(OBJECT_MAPPER.readValue(specs, AnomalyFunctionDTO.class));
+ dataFilter = DataFilterFactory.fromSpec(anomalyFunction.getSpec().getDataFilter());
this.metricUrn = MapUtils.getString(config.getProperties(), PROP_METRIC_URN);
metricEntity = MetricEntity.fromURN(metricUrn, 1.0);
}
@@ -110,6 +114,10 @@ public class LegacyAnomalyFunctionAlgorithm extends DetectionPipeline {
metricTimeSeries.set(timestamps.get(i), metricConfig.getName(), df.getDoubles(COL_VALUE).get(i));
}
+ if (!dataFilter.isQualified(metricTimeSeries, dimension, this.startTime, this.endTime)) {
+ return new DetectionPipelineResult(Collections.<MergedAnomalyResultDTO>emptyList());
+ }
+
List<AnomalyResult> result =
anomalyFunction.analyze(dimension, metricTimeSeries, new DateTime(this.startTime), new DateTime(this.endTime),
new ArrayList<>(historyMergedAnomalies));
