Calculate rejection increase and emit the delta increase of rejection as metric

src/main/java/com/amazon/opendistro/elasticsearch/performanceanalyzer/collectors/ThreadPoolMetricsCollector.java

@@ -27,17 +27,21 @@
 import java.lang.reflect.Method;
 import java.security.AccessController;
 import java.security.PrivilegedAction;
+import java.util.HashMap;
 import java.util.Iterator;
+import java.util.Map;
 import org.elasticsearch.threadpool.ThreadPoolStats.Stats;
 
 public class ThreadPoolMetricsCollector extends PerformanceAnalyzerMetricsCollector implements MetricsProcessor {
     public static final int SAMPLING_TIME_INTERVAL = MetricsConfiguration.CONFIG_MAP.get(ThreadPoolMetricsCollector.class).samplingInterval;
     private static final int KEYS_PATH_LENGTH = 0;
     private StringBuilder value;
+    private final Map<String, ThreadPoolStatsRecord> statsRecordMap;
 
     public ThreadPoolMetricsCollector() {
         super(SAMPLING_TIME_INTERVAL, "ThreadPoolMetrics");
         value = new StringBuilder();
+        statsRecordMap = new HashMap<>();
     }
 
     @Override
@@ -52,6 +56,24 @@ public void collectMetrics(long startTime) {
 
         while (statsIterator.hasNext()) {
             Stats stats = statsIterator.next();
+            long rejectionDelta = 0;
+            String name = stats.getName();
+            if (statsRecordMap.containsKey(name)) {
+                ThreadPoolStatsRecord lastRecord = statsRecordMap.get(name);
+                // if the timestamp in previous record is greater than 15s (3 * intervals),
+                // then the scheduler might hang or freeze due to long GC etc. We simply drop
+                // previous record here and set rejectionDelta to 0.
+                if (startTime - lastRecord.getTimestamp() <= SAMPLING_TIME_INTERVAL * 3) {
+                    rejectionDelta = stats.getRejected() - lastRecord.getRejected();
+                    // we might not run into this as rejection is a LongAdder which never decrement its count.
+                    // regardless, let's set it to 0 to be safe.
+                    if (rejectionDelta < 0) {
+                        rejectionDelta = 0;
+                    }
+                }
+            }
+            statsRecordMap.put(name, new ThreadPoolStatsRecord(startTime, stats.getRejected()));
+            final long finalRejectionDelta = rejectionDelta;
             ThreadPoolStatus threadPoolStatus = AccessController.doPrivileged((PrivilegedAction<ThreadPoolStatus>) () -> {
                 try {
                     //This is for backward compatibility. core ES may or may not emit latency metric
@@ -64,13 +86,13 @@ public void collectMetrics(long startTime) {
                     Method getCapacityMethod = Stats.class.getMethod("getCapacity");
                     int capacity = (Integer) getCapacityMethod.invoke(stats);
                     return new ThreadPoolStatus(stats.getName(),
-                        stats.getQueue(), stats.getRejected(),
+                        stats.getQueue(), finalRejectionDelta,
                         stats.getThreads(), stats.getActive(),
                         latency, capacity);
                 } catch (Exception e) {
                     //core ES does not have the latency patch. send the threadpool metrics without adding latency.
                     return new ThreadPoolStatus(stats.getName(),
-                        stats.getQueue(), stats.getRejected(),
+                        stats.getQueue(), finalRejectionDelta,
                         stats.getThreads(), stats.getActive());
                 }
             });
@@ -90,6 +112,24 @@ public String getMetricsPath(long startTime, String... keysPath) {
         return PerformanceAnalyzerMetrics.generatePath(startTime, PerformanceAnalyzerMetrics.sThreadPoolPath);
     }
 
+    private static class ThreadPoolStatsRecord {
+        private final long timestamp;
+        private final long rejected;
+
+        ThreadPoolStatsRecord(long timestamp, long rejected) {
+            this.timestamp = timestamp;
+            this.rejected = rejected;
+        }
+
+        public long getTimestamp() {
+            return timestamp;
+        }
+
+        public long getRejected() {
+            return rejected;
+        }
+    }
+
     public static class ThreadPoolStatus extends MetricStatus {
         public final String type;
         public final int queueSize;