SearchBackPressure Service Node/Cluster RCA

src/main/java/org/opensearch/performanceanalyzer/model/MetricsModel.java

@@ -464,6 +464,11 @@ public class MetricsModel {
                         MetricUnits.MILLISECOND.toString(),
                         AllMetrics.ShardIndexingPressureDimension.values()));
 
+        // Search Back Pressure Metrics
+        allMetricsInitializer.put(
+                AllMetrics.SearchBackPressureStatsValue.SEARCHBP_SHARD_STATS_CANCELLATIONCOUNT
+                        .toString(),
+                new MetricAttributes(MetricUnits.COUNT.toString(), EmptyDimension.values()));
         ALL_METRICS = Collections.unmodifiableMap(allMetricsInitializer);
     }
 }

src/main/java/org/opensearch/performanceanalyzer/rca/configs/SearchBackPressureRcaConfig.java

@@ -0,0 +1,148 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.performanceanalyzer.rca.configs;
+
+
+import org.opensearch.performanceanalyzer.rca.framework.core.RcaConf;
+
+public class SearchBackPressureRcaConfig {
+    public static final String CONFIG_NAME = "search-back-pressure-rca-policy";
+
+    /* Metadata fields for thresholds */
+    public static final String INCREASE_THRESHOLD_BY_JVM_STR = "increase_jvm";
+    public static final String DECREASE_THRESHOLD_BY_JVM_STR = "decrease_jvm";
+
+    public static final int SLIDING_WINDOW_SIZE_IN_MINS = 1;
+
+    // Interval period in seconds
+    public static final long DEFAULT_EVALUATION_INTERVAL_IN_S = 60;
+
+    /* interval period to call operate() */
+    public static final long EVAL_INTERVAL_IN_S = 5;
+
+    /* Increase Threshold */
+    // node max heap usage in last 60 secs is less than 70%
+    public static final int DEFAULT_MAX_HEAP_INCREASE_THRESHOLD = 70;
+    private Integer maxHeapIncreasePercentageThreshold;
+
+    // cancellationCount due to heap is more than 50% of all task cancellations in shard level
+    public static final int DEFAULT_SHARD_MAX_HEAP_CANCELLATION_THRESHOLD = 50;
+    private Integer maxShardHeapCancellationPercentageThreshold;
+
+    // cancellationCount due to heap is more than 50% of all task cancellations in task level
+    public static final int DEFAULT_TASK_MAX_HEAP_CANCELLATION_THRESHOLD = 50;
+    private Integer maxTaskHeapCancellationPercentageThreshold;
+
+    /* Decrease Threshold */
+    // node min heap usage in last 60 secs is more than 80%
+    public static final int DEFAULT_MIN_HEAP_DECREASE_THRESHOLD = 80;
+    private Integer minHeapDecreasePercentageThreshold;
+
+    // cancellationCount due to heap is less than 30% of all task cancellations in shard level
+    public static final int DEFAULT_SHARD_MIN_HEAP_CANCELLATION_THRESHOLD = 30;
+    private Integer minShardHeapCancellationPercentageThreshold;
+
+    // cancellationCount due to heap is less than 30% of all task cancellations in task level
+    public static final int DEFAULT_TASK_MIN_HEAP_CANCELLATION_THRESHOLD = 30;
+    private Integer minTaskHeapCancellationPercentageThreshold;
+
+    public SearchBackPressureRcaConfig(final RcaConf conf) {
+        // (s) -> s > 0 is the validator, if validated, fields from conf file will be returned,
+        // else, default value gets returned
+        maxHeapIncreasePercentageThreshold =
+                conf.readRcaConfig(
+                        CONFIG_NAME,
+                        SearchBackPressureRcaConfigKeys.MAX_HEAP_USAGE_INCREASE_FIELD.toString(),
+                        DEFAULT_MAX_HEAP_INCREASE_THRESHOLD,
+                        (s) -> s >= 0 && s <= 100,
+                        Integer.class);
+        maxShardHeapCancellationPercentageThreshold =
+                conf.readRcaConfig(
+                        CONFIG_NAME,
+                        SearchBackPressureRcaConfigKeys.MAX_SHARD_HEAP_CANCELLATION_PERCENTAGE_FIELD
+                                .toString(),
+                        DEFAULT_SHARD_MAX_HEAP_CANCELLATION_THRESHOLD,
+                        (s) -> s >= 0 && s <= 100,
+                        Integer.class);
+        maxTaskHeapCancellationPercentageThreshold =
+                conf.readRcaConfig(
+                        CONFIG_NAME,
+                        SearchBackPressureRcaConfigKeys.MAX_TASK_HEAP_CANCELLATION_PERCENTAGE_FIELD
+                                .toString(),
+                        DEFAULT_TASK_MAX_HEAP_CANCELLATION_THRESHOLD,
+                        (s) -> s >= 0 && s <= 100,
+                        Integer.class);
+        minHeapDecreasePercentageThreshold =
+                conf.readRcaConfig(
+                        CONFIG_NAME,
+                        SearchBackPressureRcaConfigKeys.MAX_HEAP_USAGE_DECREASE_FIELD.toString(),
+                        DEFAULT_MIN_HEAP_DECREASE_THRESHOLD,
+                        (s) -> s >= 0 && s <= 100,
+                        Integer.class);
+        minShardHeapCancellationPercentageThreshold =
+                conf.readRcaConfig(
+                        CONFIG_NAME,
+                        SearchBackPressureRcaConfigKeys.MIN_SHARD_HEAP_CANCELLATION_PERCENTAGE_FIELD
+                                .toString(),
+                        DEFAULT_SHARD_MIN_HEAP_CANCELLATION_THRESHOLD,
+                        (s) -> s >= 0 && s <= 100,
+                        Integer.class);
+        minTaskHeapCancellationPercentageThreshold =
+                conf.readRcaConfig(
+                        CONFIG_NAME,
+                        SearchBackPressureRcaConfigKeys.MIN_TASK_HEAP_CANCELLATION_PERCENTAGE_FIELD
+                                .toString(),
+                        DEFAULT_TASK_MIN_HEAP_CANCELLATION_THRESHOLD,
+                        (s) -> s >= 0 && s <= 100,
+                        Integer.class);
+    }
+
+    // Getters for private field
+    public Integer getMaxHeapIncreasePercentageThreshold() {
+        return maxHeapIncreasePercentageThreshold;
+    }
+
+    public Integer getMaxShardHeapCancellationPercentageThreshold() {
+        return maxShardHeapCancellationPercentageThreshold;
+    }
+
+    public Integer getMaxTaskHeapCancellationPercentageThreshold() {
+        return maxTaskHeapCancellationPercentageThreshold;
+    }
+
+    public Integer getMinHeapDecreasePercentageThreshold() {
+        return minHeapDecreasePercentageThreshold;
+    }
+
+    public Integer getMinShardHeapCancellationPercentageThreshold() {
+        return minShardHeapCancellationPercentageThreshold;
+    }
+
+    public Integer getMinTaskHeapCancellationPercentageThreshold() {
+        return minTaskHeapCancellationPercentageThreshold;
+    }
+
+    // name for the configuration field
+    public enum SearchBackPressureRcaConfigKeys {
+        MAX_HEAP_USAGE_INCREASE_FIELD("max-heap-usage-increase"),
+        MAX_SHARD_HEAP_CANCELLATION_PERCENTAGE_FIELD("max-shard-heap-cancellation-percentage"),
+        MAX_TASK_HEAP_CANCELLATION_PERCENTAGE_FIELD("max-task-heap-cancellation-percentage"),
+        MAX_HEAP_USAGE_DECREASE_FIELD("max-heap-usage-decrease"),
+        MIN_SHARD_HEAP_CANCELLATION_PERCENTAGE_FIELD("min-shard-heap-cancellation-percentage"),
+        MIN_TASK_HEAP_CANCELLATION_PERCENTAGE_FIELD("min-task-heap-cancellation-percentage");
+
+        private final String value;
+
+        SearchBackPressureRcaConfigKeys(final String value) {
+            this.value = value;
+        }
+
+        @Override
+        public String toString() {
+            return this.value;
+        }
+    }
+}

src/main/java/org/opensearch/performanceanalyzer/rca/framework/api/metrics/Searchbp_Stats.java

@@ -0,0 +1,18 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.performanceanalyzer.rca.framework.api.metrics;
+
+
+import org.opensearch.performanceanalyzer.commons.metrics.AllMetrics;
+import org.opensearch.performanceanalyzer.rca.framework.api.Metric;
+
+public class Searchbp_Stats extends Metric {
+    public Searchbp_Stats(long evaluationIntervalSeconds) {
+        super(
+                AllMetrics.SearchBackPressureStatsValue.SEARCHBP_TABLE_NAME.toString(),
+                evaluationIntervalSeconds);
+    }
+}

src/main/java/org/opensearch/performanceanalyzer/rca/framework/api/summaries/ResourceUtil.java

@@ -135,6 +135,24 @@ public class ResourceUtil {
                     .setResourceEnum(ResourceEnum.SHARD_REQUEST_CACHE)
                     .setMetricEnum(MetricEnum.CACHE_MAX_SIZE)
                     .build();
+    /*
+     * searchbackpressure related resource
+     * SEARCHBACKPRESSURE_SHARD resource indicate a searchbackpressure unhealthy resource unit is caused by shard level cancellation
+     */
+    public static final Resource SEARCHBACKPRESSURE_SHARD =
+            Resource.newBuilder()
+                    .setResourceEnum(ResourceEnum.SEARCHBP)
+                    .setMetricEnum(MetricEnum.SEARCHBP_SHARD)
+                    .build();
+
+    /*
+     *  SEARCHBACKPRESSURE_TASK resource indicate a searchbackpressure unhealthy resource unit is caused by task level cancellation
+     */
+    public static final Resource SEARCHBACKPRESSURE_TASK =
+            Resource.newBuilder()
+                    .setResourceEnum(ResourceEnum.SEARCHBP)
+                    .setMetricEnum(MetricEnum.SEARCHBP_TASK)
+                    .build();
 
     /**
      * Read the resourceType name from the ResourceType object

src/main/java/org/opensearch/performanceanalyzer/rca/framework/core/RcaConf.java

@@ -51,6 +51,7 @@
 import org.opensearch.performanceanalyzer.rca.configs.HotShardRcaConfig;
 import org.opensearch.performanceanalyzer.rca.configs.OldGenContendedRcaConfig;
 import org.opensearch.performanceanalyzer.rca.configs.QueueRejectionRcaConfig;
+import org.opensearch.performanceanalyzer.rca.configs.SearchBackPressureRcaConfig;
 import org.opensearch.performanceanalyzer.rca.configs.ShardRequestCacheRcaConfig;
 import org.opensearch.performanceanalyzer.rca.framework.api.summaries.bucket.BasicBucketCalculator;
 import org.opensearch.performanceanalyzer.rca.framework.api.summaries.bucket.BucketCalculator;
@@ -232,6 +233,10 @@ public OldGenContendedRcaConfig getOldGenContendedRcaConfig() {
         return new OldGenContendedRcaConfig(this);
     }
 
+    public SearchBackPressureRcaConfig getSearchBackPressureRcaConfig() {
+        return new SearchBackPressureRcaConfig(this);
+    }
+
     public <T> T readRcaConfig(
             String rcaName, String key, T defaultValue, Class<? extends T> clazz) {
         return readRcaConfig(rcaName, key, defaultValue, (s) -> true, clazz);

src/main/java/org/opensearch/performanceanalyzer/rca/store/OpenSearchAnalysisGraph.java

@@ -41,6 +41,7 @@
 import org.opensearch.performanceanalyzer.rca.framework.api.metrics.Heap_Max;
 import org.opensearch.performanceanalyzer.rca.framework.api.metrics.Heap_Used;
 import org.opensearch.performanceanalyzer.rca.framework.api.metrics.IndexWriter_Memory;
+import org.opensearch.performanceanalyzer.rca.framework.api.metrics.Searchbp_Stats;
 import org.opensearch.performanceanalyzer.rca.framework.api.metrics.ThreadPool_QueueCapacity;
 import org.opensearch.performanceanalyzer.rca.framework.api.metrics.ThreadPool_RejectedReqs;
 import org.opensearch.performanceanalyzer.rca.framework.api.metrics.Thread_Blocked_Time;
@@ -85,6 +86,8 @@
 import org.opensearch.performanceanalyzer.rca.store.rca.jvmsizing.LargeHeapClusterRca;
 import org.opensearch.performanceanalyzer.rca.store.rca.jvmsizing.OldGenContendedRca;
 import org.opensearch.performanceanalyzer.rca.store.rca.jvmsizing.OldGenReclamationRca;
+import org.opensearch.performanceanalyzer.rca.store.rca.searchbackpressure.SearchBackPressureClusterRCA;
+import org.opensearch.performanceanalyzer.rca.store.rca.searchbackpressure.SearchBackPressureRCA;
 import org.opensearch.performanceanalyzer.rca.store.rca.temperature.NodeTemperatureRca;
 import org.opensearch.performanceanalyzer.rca.store.rca.temperature.dimension.CpuUtilDimensionTemperatureRca;
 import org.opensearch.performanceanalyzer.rca.store.rca.temperature.dimension.HeapAllocRateTemperatureRca;
@@ -117,6 +120,9 @@ public void construct() {
                         MetricsDB.AVG,
                         AllMetrics.CommonDimension.OPERATION.toString());
 
+        // SearchBackpressure Metric
+        Metric searchbp_Stats = new Searchbp_Stats(EVALUATION_INTERVAL_SECONDS);
+
         heapUsed.addTag(
                 RcaConsts.RcaTagConstants.TAG_LOCUS,
                 RcaConsts.RcaTagConstants.LOCUS_DATA_CLUSTER_MANAGER_NODE);
@@ -141,6 +147,9 @@ public void construct() {
         threadWaitedTime.addTag(
                 RcaConsts.RcaTagConstants.TAG_LOCUS,
                 RcaConsts.RcaTagConstants.LOCUS_DATA_CLUSTER_MANAGER_NODE);
+        searchbp_Stats.addTag(
+                RcaConsts.RcaTagConstants.TAG_LOCUS,
+                RcaConsts.RcaTagConstants.LOCUS_DATA_CLUSTER_MANAGER_NODE);
 
         addLeaf(heapUsed);
         addLeaf(gcEvent);
@@ -150,6 +159,7 @@ public void construct() {
         addLeaf(cpuUtilizationGroupByOperation);
         addLeaf(threadBlockedTime);
         addLeaf(threadWaitedTime);
+        addLeaf(searchbp_Stats);
 
         // add node stats metrics
         List<Metric> nodeStatsMetrics = constructNodeStatsMetrics();
@@ -433,6 +443,28 @@ public void construct() {
                         shardRequestCacheClusterRca,
                         highHeapUsageClusterRca));
 
+        // Search Back Pressure Service RCA enabled
+        SearchBackPressureRCA searchBackPressureRCA =
+                new SearchBackPressureRCA(RCA_PERIOD, heapMax, heapUsed, searchbp_Stats);
+        searchBackPressureRCA.addTag(
+                RcaConsts.RcaTagConstants.TAG_LOCUS,
+                RcaConsts.RcaTagConstants.LOCUS_DATA_CLUSTER_MANAGER_NODE);
+        searchBackPressureRCA.addAllUpstreams(Arrays.asList(heapMax, heapUsed, searchbp_Stats));
+
+        // Search Back Pressure Service Cluster RCA enabled
+        SearchBackPressureClusterRCA searchBackPressureClusterRCA =
+                new SearchBackPressureClusterRCA(RCA_PERIOD, searchBackPressureRCA);
+        searchBackPressureClusterRCA.addTag(
+                RcaConsts.RcaTagConstants.TAG_LOCUS,
+                RcaConsts.RcaTagConstants.LOCUS_CLUSTER_MANAGER_NODE);
+        searchBackPressureClusterRCA.addAllUpstreams(
+                Collections.singletonList(searchBackPressureRCA));
+        searchBackPressureClusterRCA.addTag(
+                RcaConsts.RcaTagConstants.TAG_AGGREGATE_UPSTREAM,
+                RcaConsts.RcaTagConstants.LOCUS_DATA_NODE);
+
+        // TODO: Add SearchBackPressure Decider
+
         AdmissionControlDecider admissionControlDecider =
                 buildAdmissionControlDecider(heapUsed, heapMax);
 

src/main/java/org/opensearch/performanceanalyzer/rca/store/rca/OldGenRca.java

@@ -222,7 +222,7 @@ protected boolean isOldGenCollectorCMS() {
         return true;
     }
 
-    /** Sliding window to check the minimal olg gen usage within a given time frame */
+    /** Sliding window to check the minimal old gen usage within a given time frame */
     public static class MinOldGenSlidingWindow extends SlidingWindow<SlidingWindowData> {
 
         public MinOldGenSlidingWindow(int SLIDING_WINDOW_SIZE_IN_TIMESTAMP, TimeUnit timeUnit) {
@@ -250,4 +250,60 @@ public double readMin() {
             return Double.NaN;
         }
     }
+
+    /**
+     * Sliding window to check the max/min old gen usage within a given time frame
+     *
+     * @param isMinSlidingWindow true if the sliding window is for min usage, false for max usage
+     *     Provides a more general framework than MinOldGenSlidingWindow as this sliding window can
+     *     be implemented as minSlidingWindow or maxSlidingWindow depending on the need.
+     */
+    public static class MinMaxSlidingWindow extends SlidingWindow<SlidingWindowData> {
+        boolean isMinSlidingWindow;
+
+        public MinMaxSlidingWindow(
+                int SLIDING_WINDOW_SIZE_IN_TIMESTAMP,
+                TimeUnit timeUnit,
+                boolean isMinSlidingWindow) {
+            super(SLIDING_WINDOW_SIZE_IN_TIMESTAMP, timeUnit);
+            this.isMinSlidingWindow = isMinSlidingWindow;
+        }
+
+        @Override
+        public void next(SlidingWindowData e) {
+            if (isMinSlidingWindow) {
+                // monotonically decreasing sliding window
+                while (!windowDeque.isEmpty()
+                        && windowDeque.peekFirst().getValue() >= e.getValue()) {
+                    windowDeque.pollFirst();
+                }
+            } else {
+                // monotonically increasing sliding window
+                while (!windowDeque.isEmpty()
+                        && windowDeque.peekFirst().getValue() < e.getValue()) {
+                    windowDeque.pollFirst();
+                }
+            }
+
+            windowDeque.addFirst(e);
+            while (!windowDeque.isEmpty()
+                    && TimeUnit.MILLISECONDS.toSeconds(
+                                    e.getTimeStamp() - windowDeque.peekLast().getTimeStamp())
+                            > SLIDING_WINDOW_SIZE) {
+                windowDeque.pollLast();
+            }
+        }
+
+        /*
+         * read last element in the window
+         * if the sliding window is MinSlidingWindow then returns the min element
+         * else return the max element in the deque
+         */
+        public double readLastElementInWindow() {
+            if (!windowDeque.isEmpty()) {
+                return windowDeque.peekLast().getValue();
+            }
+            return Double.NaN;
+        }
+    }
 }

src/main/java/org/opensearch/performanceanalyzer/rca/store/rca/searchbackpressure/SearchBackPressureClusterRCA.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.performanceanalyzer.rca.store.rca.searchbackpressure;
+
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.opensearch.performanceanalyzer.rca.framework.api.Rca;
+import org.opensearch.performanceanalyzer.rca.framework.api.flow_units.ResourceFlowUnit;
+import org.opensearch.performanceanalyzer.rca.framework.api.summaries.HotNodeSummary;
+import org.opensearch.performanceanalyzer.rca.store.rca.cluster.BaseClusterRca;
+
+public class SearchBackPressureClusterRCA extends BaseClusterRca {
+
+    public static final String RCA_TABLE_NAME = SearchBackPressureClusterRCA.class.getSimpleName();
+    private static final Logger LOG = LogManager.getLogger(SearchBackPressureClusterRCA.class);
+
+    public <R extends Rca<ResourceFlowUnit<HotNodeSummary>>> SearchBackPressureClusterRCA(
+            final int rcaPeriod, final R SearchBackPressureRCA) {
+        super(rcaPeriod, SearchBackPressureRCA);
+    }
+}