@Test public void testNoFalsePositiveDetectionDueToTrafficFluctuation() { SlowBrokerFinder slowBrokerFinder = createSlowBrokerFinder(); Map<BrokerEntity, ValuesAndExtrapolations> history = createHistory(populateMetricValues(NORMAL_BYTES_IN_RATE, NORMAL_BYTES_IN_RATE, NORMAL_LOG_FLUSH_TIME_MS), populateMetricValues(NORMAL_BYTES_IN_RATE * BYTE_IN_RATE_VARIANCE_RATIO, NORMAL_BYTES_IN_RATE * BYTE_IN_RATE_VARIANCE_RATIO, NORMAL_LOG_FLUSH_TIME_MS * LOG_FLUSH_TIME_MS_VARIANCE_RATIO), METRIC_HISTORY_WINDOW_SIZE, BROKER_ENTITIES.get(0)); Map<BrokerEntity, ValuesAndExtrapolations> currentMetrics = createCurrentMetrics(populateMetricValues(NORMAL_BYTES_IN_RATE, NORMAL_BYTES_IN_RATE, NORMAL_LOG_FLUSH_TIME_MS), CURRENT_METRIC_WINDOW, BROKER_ENTITIES.get(0)); Collection<MetricAnomaly<BrokerEntity>> anomalies = slowBrokerFinder.metricAnomalies(history, currentMetrics); assertTrue(anomalies.isEmpty()); }
@Override public Collection<MetricAnomaly<BrokerEntity>> metricAnomalies(Map<BrokerEntity, ValuesAndExtrapolations> metricsHistoryByBroker, Map<BrokerEntity, ValuesAndExtrapolations> currentMetricsByBroker) { LOG.info("Slow broker detection started."); try { Set<BrokerEntity> detectedMetricAnomalies = detectMetricAnomalies(metricsHistoryByBroker, currentMetricsByBroker); updateBrokerSlownessScore(detectedMetricAnomalies); return createSlowBrokerAnomalies(detectedMetricAnomalies, metricsHistoryByBroker.size()); } catch (Exception e) { LOG.warn("Slow broker detector encountered exception: ", e); } finally { LOG.info("Slow broker detection finished."); } return Collections.emptySet(); }
SlowBrokerFinder implements MetricAnomalyFinder<BrokerEntity> { @Override public Collection<MetricAnomaly<BrokerEntity>> metricAnomalies(Map<BrokerEntity, ValuesAndExtrapolations> metricsHistoryByBroker, Map<BrokerEntity, ValuesAndExtrapolations> currentMetricsByBroker) { LOG.info("Slow broker detection started."); try { Set<BrokerEntity> detectedMetricAnomalies = detectMetricAnomalies(metricsHistoryByBroker, currentMetricsByBroker); updateBrokerSlownessScore(detectedMetricAnomalies); return createSlowBrokerAnomalies(detectedMetricAnomalies, metricsHistoryByBroker.size()); } catch (Exception e) { LOG.warn("Slow broker detector encountered exception: ", e); } finally { LOG.info("Slow broker detection finished."); } return Collections.emptySet(); } }
SlowBrokerFinder implements MetricAnomalyFinder<BrokerEntity> { @Override public Collection<MetricAnomaly<BrokerEntity>> metricAnomalies(Map<BrokerEntity, ValuesAndExtrapolations> metricsHistoryByBroker, Map<BrokerEntity, ValuesAndExtrapolations> currentMetricsByBroker) { LOG.info("Slow broker detection started."); try { Set<BrokerEntity> detectedMetricAnomalies = detectMetricAnomalies(metricsHistoryByBroker, currentMetricsByBroker); updateBrokerSlownessScore(detectedMetricAnomalies); return createSlowBrokerAnomalies(detectedMetricAnomalies, metricsHistoryByBroker.size()); } catch (Exception e) { LOG.warn("Slow broker detector encountered exception: ", e); } finally { LOG.info("Slow broker detection finished."); } return Collections.emptySet(); } SlowBrokerFinder(); }
SlowBrokerFinder implements MetricAnomalyFinder<BrokerEntity> { @Override public Collection<MetricAnomaly<BrokerEntity>> metricAnomalies(Map<BrokerEntity, ValuesAndExtrapolations> metricsHistoryByBroker, Map<BrokerEntity, ValuesAndExtrapolations> currentMetricsByBroker) { LOG.info("Slow broker detection started."); try { Set<BrokerEntity> detectedMetricAnomalies = detectMetricAnomalies(metricsHistoryByBroker, currentMetricsByBroker); updateBrokerSlownessScore(detectedMetricAnomalies); return createSlowBrokerAnomalies(detectedMetricAnomalies, metricsHistoryByBroker.size()); } catch (Exception e) { LOG.warn("Slow broker detector encountered exception: ", e); } finally { LOG.info("Slow broker detection finished."); } return Collections.emptySet(); } SlowBrokerFinder(); @Override Collection<MetricAnomaly<BrokerEntity>> metricAnomalies(Map<BrokerEntity, ValuesAndExtrapolations> metricsHistoryByBroker,
                                                                 Map<BrokerEntity, ValuesAndExtrapolations> currentMetricsByBroker); @Override void configure(Map<String, ?> configs); }
SlowBrokerFinder implements MetricAnomalyFinder<BrokerEntity> { @Override public Collection<MetricAnomaly<BrokerEntity>> metricAnomalies(Map<BrokerEntity, ValuesAndExtrapolations> metricsHistoryByBroker, Map<BrokerEntity, ValuesAndExtrapolations> currentMetricsByBroker) { LOG.info("Slow broker detection started."); try { Set<BrokerEntity> detectedMetricAnomalies = detectMetricAnomalies(metricsHistoryByBroker, currentMetricsByBroker); updateBrokerSlownessScore(detectedMetricAnomalies); return createSlowBrokerAnomalies(detectedMetricAnomalies, metricsHistoryByBroker.size()); } catch (Exception e) { LOG.warn("Slow broker detector encountered exception: ", e); } finally { LOG.info("Slow broker detection finished."); } return Collections.emptySet(); } SlowBrokerFinder(); @Override Collection<MetricAnomaly<BrokerEntity>> metricAnomalies(Map<BrokerEntity, ValuesAndExtrapolations> metricsHistoryByBroker,
                                                                 Map<BrokerEntity, ValuesAndExtrapolations> currentMetricsByBroker); @Override void configure(Map<String, ?> configs); static final String SELF_HEALING_SLOW_BROKER_REMOVAL_ENABLED_CONFIG; static final String REMOVE_SLOW_BROKER_CONFIG; static final String SLOW_BROKER_BYTES_IN_RATE_DETECTION_THRESHOLD_CONFIG; static final double DEFAULT_SLOW_BROKER_BYTES_IN_RATE_DETECTION_THRESHOLD; static final String SLOW_BROKER_LOG_FLUSH_TIME_THRESHOLD_MS_CONFIG; static final double DEFAULT_SLOW_BROKER_LOG_FLUSH_TIME_THRESHOLD_MS_CONFIG; static final String SLOW_BROKER_METRIC_HISTORY_PERCENTILE_THRESHOLD_CONFIG; static final double DEFAULT_SLOW_BROKER_METRIC_HISTORY_PERCENTILE_THRESHOLD; static final String SLOW_BROKER_METRIC_HISTORY_MARGIN_CONFIG; static final double DEFAULT_SLOW_BROKER_METRIC_HISTORY_MARGIN; static final String SLOW_BROKER_PEER_METRIC_PERCENTILE_THRESHOLD_CONFIG; static final double DEFAULT_SLOW_BROKER_PEER_METRIC_PERCENTILE_THRESHOLD; static final String SLOW_BROKER_PEER_METRIC_MARGIN_CONFIG; static final double DEFAULT_SLOW_BROKER_PEER_METRIC_MARGIN; static final String SLOW_BROKER_DEMOTION_SCORE_CONFIG; static final int DEFAULT_SLOW_BROKER_DEMOTION_SCORE; static final String SLOW_BROKER_DECOMMISSION_SCORE_CONFIG; static final int DEFAULT_SLOW_BROKER_DECOMMISSION_SCORE; static final String SLOW_BROKER_SELF_HEALING_UNFIXABLE_RATIO_CONFIG; }