Expose queue rejection time period setting in rca.conf

pa_config/rca.conf

@@ -49,6 +49,9 @@
       "promotion-rate-mb-per-second" : 500,
       "young-gen-gc-time-ms-per-second" : 400
     },
+    "queue-rejection-rca": {
+      "rejection-time-period-in-seconds" : 300
+    },
     //hot shard rca
     "hot-shard-rca": {
       "cpu-utilization" : 0.01,

pa_config/rca_idle_master.conf

@@ -49,6 +49,9 @@
       "promotion-rate-mb-per-second" : 500,
       "young-gen-gc-time-ms-per-second" : 400
     },
+    "queue-rejection-rca": {
+      "rejection-time-period-in-seconds" : 300
+    },
     //unbalanced node rca
     "hot-node-cluster-rca": {
       "unbalanced-resource-percentage" : 0.3,

pa_config/rca_master.conf

@@ -49,6 +49,9 @@
       "promotion-rate-mb-per-second" : 500,
       "young-gen-gc-time-ms-per-second" : 400
     },
+    "queue-rejection-rca": {
+      "rejection-time-period-in-seconds" : 300
+    },
     //unbalanced node rca
     "hot-node-cluster-rca": {
       "unbalanced-resource-percentage" : 0.3,

src/main/java/com/amazon/opendistro/elasticsearch/performanceanalyzer/rca/configs/QueueRejectionRcaConfig.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.configs;
+
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.core.RcaConf;
+
+public class QueueRejectionRcaConfig {
+  private Integer rejectionTimePeriodInSeconds;
+  public static final int DEFAULT_REJECTION_TIME_PERIOD_IN_SECONDS = 300;
+  public static final String CONFIG_NAME = "queue-rejection-rca";
+
+  public QueueRejectionRcaConfig(final RcaConf rcaConf) {
+    rejectionTimePeriodInSeconds = rcaConf.readRcaConfig(
+        CONFIG_NAME, RCA_CONF_KEY_CONSTANTS.REJECTION_TIME_PERIOD_IN_SECONDS, Integer.class);
+    if (rejectionTimePeriodInSeconds == null) {
+      rejectionTimePeriodInSeconds = DEFAULT_REJECTION_TIME_PERIOD_IN_SECONDS;
+    }
+  }
+
+  public int getRejectionTimePeriodInSeconds() {
+    return rejectionTimePeriodInSeconds;
+  }
+
+  public static class RCA_CONF_KEY_CONSTANTS {
+    public static final String REJECTION_TIME_PERIOD_IN_SECONDS = "rejection-time-period-in-seconds";
+  }
+}

src/main/java/com/amazon/opendistro/elasticsearch/performanceanalyzer/rca/framework/core/RcaConf.java

@@ -25,6 +25,7 @@
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.configs.HotNodeClusterRcaConfig;
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.configs.HotShardClusterRcaConfig;
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.configs.HotShardRcaConfig;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.configs.QueueRejectionRcaConfig;
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.util.RcaConsts;
 import com.fasterxml.jackson.core.JsonFactory;
 import com.fasterxml.jackson.core.JsonParser;
@@ -145,6 +146,10 @@ public HighHeapUsageYoungGenRcaConfig getHighHeapUsageYoungGenRcaConfig() {
     return new HighHeapUsageYoungGenRcaConfig(this);
   }
 
+  public QueueRejectionRcaConfig getQueueRejectionRcaConfig() {
+    return new QueueRejectionRcaConfig(this);
+  }
+
   public HotNodeClusterRcaConfig getHotNodeClusterRcaConfig() {
     return new HotNodeClusterRcaConfig(this);
   }

src/main/java/com/amazon/opendistro/elasticsearch/performanceanalyzer/rca/store/rca/threadpool/QueueRejectionRca.java

@@ -21,6 +21,8 @@
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.grpc.Resource;
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.metrics.AllMetrics.ThreadPoolType;
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.metricsdb.MetricsDB;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.configs.HighHeapUsageOldGenRcaConfig;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.configs.QueueRejectionRcaConfig;
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.api.Metric;
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.api.Rca;
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.api.Resources;
@@ -31,6 +33,7 @@
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.api.summaries.HotNodeSummary;
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.api.summaries.HotResourceSummary;
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.api.summaries.ResourceUtil;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.core.RcaConf;
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.util.InstanceDetails;
 import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.scheduler.FlowUnitOperationArgWrapper;
 import com.google.common.annotations.VisibleForTesting;
@@ -51,7 +54,6 @@
  */
 public class QueueRejectionRca extends Rca<ResourceFlowUnit<HotNodeSummary>> {
   private static final Logger LOG = LogManager.getLogger(QueueRejectionRca.class);
-  private static final long REJECTION_TIME_PERIOD_IN_MILLISECOND = TimeUnit.SECONDS.toMillis(300);
   private final int rcaPeriod;
   private final List<QueueRejectionCollector> queueRejectionCollectors;
   private int counter;
@@ -64,9 +66,9 @@ public <M extends Metric> QueueRejectionRca(final int rcaPeriod, M threadPool_Re
     clock = Clock.systemUTC();
     queueRejectionCollectors = Collections.unmodifiableList(Arrays.asList(
         new QueueRejectionCollector(ResourceUtil.WRITE_QUEUE_REJECTION, ThreadPoolType.WRITE,
-            threadPool_RejectedReqs, REJECTION_TIME_PERIOD_IN_MILLISECOND),
+            threadPool_RejectedReqs),
         new QueueRejectionCollector(ResourceUtil.SEARCH_QUEUE_REJECTION, ThreadPoolType.SEARCH,
-            threadPool_RejectedReqs, REJECTION_TIME_PERIOD_IN_MILLISECOND)
+            threadPool_RejectedReqs)
     ));
   }
 
@@ -123,6 +125,18 @@ public void generateFlowUnitListFromWire(FlowUnitOperationArgWrapper args) {
     setFlowUnits(flowUnitList);
   }
 
+  /**
+   * read rejection-time-period-in-seconds from rca.conf
+   * @param conf RcaConf object
+   */
+  @Override
+  public void readRcaConf(RcaConf conf) {
+    QueueRejectionRcaConfig configObj = conf.getQueueRejectionRcaConfig();
+    long rejectedTimePeriod = TimeUnit.SECONDS.toMillis(configObj.getRejectionTimePeriodInSeconds());
+    queueRejectionCollectors.forEach(
+        collector -> collector.setRejectionTimePeriod(rejectedTimePeriod));
+  }
+
   /**
    * A collector class to collect rejection from each queue type
    */
@@ -132,16 +146,21 @@ private static class QueueRejectionCollector {
     private final Metric threadPool_RejectedReqs;
     private boolean hasRejection;
     private long rejectionTimestamp;
-    private long rejectionTimePeriodThreshold;
+    private long rejectionTimePeriodInMillis;
 
     public QueueRejectionCollector(final Resource threadPool, final ThreadPoolType threadPoolMetric,
-        final Metric threadPool_RejectedReqs, final long threshold) {
+        final Metric threadPool_RejectedReqs) {
       this.threadPool = threadPool;
       this.threadPoolMetric = threadPoolMetric;
       this.threadPool_RejectedReqs = threadPool_RejectedReqs;
       this.hasRejection = false;
       this.rejectionTimestamp = 0;
-      this.rejectionTimePeriodThreshold = threshold;
+      this.rejectionTimePeriodInMillis =
+          TimeUnit.SECONDS.toMillis(QueueRejectionRcaConfig.DEFAULT_REJECTION_TIME_PERIOD_IN_SECONDS);
+    }
+
+    public void setRejectionTimePeriod(long rejectionTimePeriodInMillis) {
+      this.rejectionTimePeriodInMillis = rejectionTimePeriodInMillis;
     }
 
     public void collect(final long currTimestamp) {
@@ -169,14 +188,14 @@ public void collect(final long currTimestamp) {
     }
 
     public boolean isUnhealthy(final long currTimestamp) {
-      return hasRejection && (currTimestamp - rejectionTimestamp) >= rejectionTimePeriodThreshold;
+      return hasRejection && (currTimestamp - rejectionTimestamp) >= rejectionTimePeriodInMillis;
     }
 
     public HotResourceSummary generateSummary(final long currTimestamp) {
       HotResourceSummary resourceSummary = null;
       if (isUnhealthy(currTimestamp)) {
         resourceSummary = new HotResourceSummary(threadPool,
-            TimeUnit.MILLISECONDS.toSeconds(rejectionTimePeriodThreshold),
+            TimeUnit.MILLISECONDS.toSeconds(rejectionTimePeriodInMillis),
             TimeUnit.MILLISECONDS.toSeconds(currTimestamp - rejectionTimestamp),
             0);
       }

src/test/java/com/amazon/opendistro/elasticsearch/performanceanalyzer/rca/configs/QueueRejectionRcaConfigTest.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ *  permissions and limitations under the License.
+ */
+
+package com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.configs;
+
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.core.RcaConf;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.rca.framework.util.RcaConsts;
+import java.nio.file.Paths;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class QueueRejectionRcaConfigTest {
+
+  private RcaConf rcaConf;
+  private RcaConf rcaOldConf;
+
+  @Before
+  public void init() {
+    String rcaConfPath = Paths.get(RcaConsts.TEST_CONFIG_PATH, "rca_elected_master.conf").toString();
+    String rcaOldConfPath = Paths.get(RcaConsts.TEST_CONFIG_PATH, "rca_master.conf").toString();
+    rcaConf = new RcaConf(rcaConfPath);
+    rcaOldConf = new RcaConf(rcaOldConfPath);
+  }
+
+  @Test
+  public void testReadConfig() {
+    QueueRejectionRcaConfig config = new QueueRejectionRcaConfig(rcaConf);
+    Assert.assertEquals(400, config.getRejectionTimePeriodInSeconds());
+
+    QueueRejectionRcaConfig oldConfig = new QueueRejectionRcaConfig(rcaOldConf);
+    Assert.assertEquals(QueueRejectionRcaConfig.DEFAULT_REJECTION_TIME_PERIOD_IN_SECONDS,
+        oldConfig.getRejectionTimePeriodInSeconds());
+  }
+}

src/test/resources/rca/rca.conf

@@ -54,6 +54,9 @@
       "promotion-rate-mb-per-second" : 500,
       "young-gen-gc-time-ms-per-second" : 400
     },
+    "queue-rejection-rca": {
+      "rejection-time-period-in-seconds" : 300
+    },
     //hot shard rca
     "hot-shard-rca": {
       "cpu-utilization" : 0.01,

src/test/resources/rca/rca_elected_master.conf

@@ -53,6 +53,9 @@
       "promotion-rate-mb-per-second" : 500,
       "young-gen-gc-time-ms-per-second" : 400
     },
+    "queue-rejection-rca": {
+      "rejection-time-period-in-seconds" : 400
+    },
     //unbalanced node rca
     "hot-node-cluster-rca": {
       "unbalanced-resource-percentage" : 0.3,