AnomaliesProcessor { public List<AnomalyModel> processMetrics(ConsumerRecords<String, MappedMetricData> metricRecords, ExecutorService executorService) { List<AnomalyModel> anomalyModels = new ArrayList(); for (ConsumerRecord<String, MappedMetricData> consumerRecord : metricRecords) { MappedMetricData mappedMetricData = consumerRecord.value(); if (mappedMetricData != null) { OutlierDetectorResult outlierDetectorResult = (OutlierDetectorResult) mappedMetricData.getAnomalyResult(); if (outlierDetectorResult.getAnomalyLevel() != AnomalyLevel.NORMAL) { AnomalyModel.Builder anomalyModel = AnomalyModel.newBuilder(); if (outlierDetectorResult != null) { anomalyModel.level(outlierDetectorResult.getAnomalyLevel().toString()); anomalyModel.anomalyThresholds(outlierDetectorResult.getThresholds()); } if (mappedMetricData.getDetectorUuid() != null) { anomalyModel.uuid(mappedMetricData.getDetectorUuid().toString()); } MetricData metricData = mappedMetricData.getMetricData(); if (metricData != null) { anomalyModel.timestamp(VisualizerUtility.convertToDate(metricData.getTimestamp())); anomalyModel.value(metricData.getValue()); if (metricData.getMetricDefinition() != null) { anomalyModel.key(metricData.getMetricDefinition().getKey()); anomalyModel.tags(metricData.getMetricDefinition().getTags()); } } anomalyModels.add(anomalyModel.build()); } } } if (anomalyModels.size() > 0) { ElasticSearchBulkService elasticSearchBulkService = new ElasticSearchBulkService(anomalyModels); executorService.submit(elasticSearchBulkService); log.info("sending anomaly records to elasticsearch: {}", anomalyModels.size()); } return anomalyModels; } AnomaliesProcessor(); List<AnomalyModel> processMetrics(ConsumerRecords<String, MappedMetricData> metricRecords, ExecutorService executorService);  }
@Test public void testProcessMetrics() { ConsumerRecords<String, MappedMetricData> metricDataConsumerRecords = buildMetricRecords(2, AnomalyLevel.STRONG); List<AnomalyModel> anomalyModels = anomaliesProcessor.processMetrics(metricDataConsumerRecords, executorService); assertNotNull(anomalyModels); assertTrue(anomalyModels.size() == 2); }
@Test public void testProcessZeroMetrics() { ConsumerRecords<String, MappedMetricData> metricDataConsumerRecords = buildMetricRecords(0, AnomalyLevel.WEAK); List<AnomalyModel> anomalyModels = anomaliesProcessor.processMetrics(metricDataConsumerRecords, executorService); assertNotNull(anomalyModels); assertTrue(anomalyModels.size() == 0); }
@Test public void testProcessZeroMetricsForNormalAnomalies() { ConsumerRecords<String, MappedMetricData> metricDataConsumerRecords = buildMetricRecords(4, AnomalyLevel.NORMAL); List<AnomalyModel> anomalyModels = anomaliesProcessor.processMetrics(metricDataConsumerRecords, executorService); assertNotNull(anomalyModels); assertTrue(anomalyModels.size() == 0); }