Add some metric logs

Author: drempapis

server/src/main/java/org/elasticsearch/cluster/coordination/SearchIndexTimeTrackingCleanupService.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.cluster.coordination;
+
+import org.elasticsearch.cluster.ClusterChangedEvent;
+import org.elasticsearch.cluster.ClusterStateListener;
+import org.elasticsearch.common.util.concurrent.TaskExecutionTimeTrackingPerIndexEsThreadPoolExecutor;
+import org.elasticsearch.index.Index;
+
+public class SearchIndexTimeTrackingCleanupService implements ClusterStateListener {
+    private TaskExecutionTimeTrackingPerIndexEsThreadPoolExecutor executor;
+
+    public SearchIndexTimeTrackingCleanupService(TaskExecutionTimeTrackingPerIndexEsThreadPoolExecutor executor) {
+        this.executor = executor;
+    }
+
+    @Override
+    public void clusterChanged(ClusterChangedEvent event) {
+        for (Index index : event.indicesDeleted()) {
+            executor.stopTrackingIndex(index.getName());
+        }
+    }
+}

server/src/main/java/org/elasticsearch/common/util/concurrent/TaskExecutionTimeTrackingPerIndexEsThreadPoolExecutor.java

@@ -11,8 +11,11 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.cluster.ClusterChangedEvent;
+import org.elasticsearch.cluster.ClusterStateListener;
 import org.elasticsearch.common.ExponentiallyWeightedMovingAverage;
 import org.elasticsearch.core.Tuple;
+import org.elasticsearch.index.Index;
 
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
@@ -28,8 +31,8 @@
  */
 public class TaskExecutionTimeTrackingPerIndexEsThreadPoolExecutor extends TaskExecutionTimeTrackingEsThreadPoolExecutor {
     private static final Logger logger = LogManager.getLogger(TaskExecutionTimeTrackingPerIndexEsThreadPoolExecutor.class);
-    private final ConcurrentHashMap<String, Tuple<LongAdder, ExponentiallyWeightedMovingAverage>> indexExecutionTime;
-    private final ConcurrentHashMap<Runnable, String> runnableToIndexName;
+    public final ConcurrentHashMap<String, Tuple<LongAdder, ExponentiallyWeightedMovingAverage>> indexExecutionTime;
+    public final ConcurrentHashMap<Runnable, String> runnableToIndexName;
 
     TaskExecutionTimeTrackingPerIndexEsThreadPoolExecutor(
         String name,

server/src/main/java/org/elasticsearch/node/NodeConstruction.java

@@ -41,6 +41,7 @@
 import org.elasticsearch.cluster.coordination.CoordinationDiagnosticsService;
 import org.elasticsearch.cluster.coordination.Coordinator;
 import org.elasticsearch.cluster.coordination.MasterHistoryService;
+import org.elasticsearch.cluster.coordination.SearchIndexTimeTrackingCleanupService;
 import org.elasticsearch.cluster.coordination.StableMasterHealthIndicatorService;
 import org.elasticsearch.cluster.metadata.DataStreamFailureStoreSettings;
 import org.elasticsearch.cluster.metadata.DataStreamGlobalRetentionSettings;
@@ -79,6 +80,8 @@
 import org.elasticsearch.common.settings.SettingsModule;
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.util.PageCacheRecycler;
+import org.elasticsearch.common.util.concurrent.TaskExecutionTimeTrackingEsThreadPoolExecutor;
+import org.elasticsearch.common.util.concurrent.TaskExecutionTimeTrackingPerIndexEsThreadPoolExecutor;
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.core.IOUtils;
 import org.elasticsearch.core.SuppressForbidden;
@@ -235,6 +238,8 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
+import java.util.Timer;
+import java.util.TimerTask;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 import java.util.function.UnaryOperator;
@@ -700,6 +705,12 @@ private void construct(
         ClusterService clusterService = createClusterService(settingsModule, threadPool, taskManager);
         clusterService.addStateApplier(scriptService);
 
+        // TODO DR - this is a bit of a hack to get the cluster service into the plugins
+        var executor = (TaskExecutionTimeTrackingEsThreadPoolExecutor) threadPool.executor(ThreadPool.Names.SEARCH);
+        TaskExecutionTimeTrackingPerIndexEsThreadPoolExecutor perIndexEsThreadPoolExecutor = (TaskExecutionTimeTrackingPerIndexEsThreadPoolExecutor) executor;
+        searchLoadMetricsReporter(perIndexEsThreadPoolExecutor);
+        clusterService.addListener(new SearchIndexTimeTrackingCleanupService(perIndexEsThreadPoolExecutor));
+
         modules.bindToInstance(DocumentParsingProvider.class, documentParsingProvider);
 
         FailureStoreMetrics failureStoreMetrics = new FailureStoreMetrics(telemetryProvider.getMeterRegistry());
@@ -1253,6 +1264,26 @@ public Map<String, String> searchFields() {
         postInjection(clusterModule, actionModule, clusterService, transportService, featureService);
     }
 
+    private void searchLoadMetricsReporter(TaskExecutionTimeTrackingPerIndexEsThreadPoolExecutor executor) {
+        Timer timer = new Timer();
+        TimerTask task = new TimerTask() {
+            @Override
+            public void run() {
+                if(executor.indexExecutionTime.size() > 0) {
+                    logger.info("Number of reported indices: {}", executor.indexExecutionTime.size());
+                    logger.info("Number of runnables: {}", executor.runnableToIndexName.size());
+                    executor.indexExecutionTime.forEach((index, tuple) -> {
+                        logger.info("Index: {}, Total execution time: {}, EWMA: {}", index, tuple.v1().sum(), tuple.v2().getAverage());
+                    });
+                    logger.info("Total task execution time: {}", executor.getTotalTaskExecutionTime());
+                    logger.info("----------------------------------------------------------------------------------");
+                }
+            }
+        };
+
+        timer.scheduleAtFixedRate(task, 0, 4000);
+    }
+
     /**
      * For each "component" (getter) <em>c</em> of a {@link Record},
      * calls {@link org.elasticsearch.injection.Injector#addInstance(Object) Injector.addInstance}