Detailed tasks information as metrics (#371)

* Add compaction stats, add metrics to the JobExecutor (but not exposed yet) and modify some methods to be more reusable

* Add parameter to allow disabling gather of extended metrics

* Add StreamInfo metrics

* Fix NPE

* Improve backwards compatibility

* Fix formatting

* Fix labelValues adder (defaultLabelValuse were overwritten), add tests to verify the methods do not error and have correct labelValues and labelNames sizes in the end results also.

* Add missing license header

* Fix naming of _total metric

* Add more defensives

Author: burmanm

management-api-agent-common/pom.xml

@@ -60,6 +60,12 @@
           <version>${junit.version}</version>
           <scope>test</scope>
         </dependency>
+        <dependency>
+          <groupId>org.mockito</groupId>
+          <artifactId>mockito-core</artifactId>
+          <version>5.5.0</version>
+          <scope>test</scope>
+        </dependency>
         <dependency>
           <groupId>io.netty</groupId>
           <artifactId>netty-all</artifactId>

management-api-agent-common/src/main/java/com/datastax/mgmtapi/util/JobExecutor.java

@@ -10,11 +10,12 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadPoolExecutor;
 import org.apache.cassandra.utils.Pair;
 
 public class JobExecutor {
   ExecutorService executorService = Executors.newFixedThreadPool(1);
-  Cache<String, Job> jobCache = CacheBuilder.newBuilder().maximumSize(1000).build();
+  Cache<String, Job> jobCache = CacheBuilder.newBuilder().recordStats().maximumSize(1000).build();
 
   public Pair<String, CompletableFuture<Void>> submit(String jobType, Runnable runnable) {
     // Where do I create the job details? Here? Add it to the Cache first?
@@ -45,4 +46,12 @@ public Pair<String, CompletableFuture<Void>> submit(String jobType, Runnable run
   public Job getJobWithId(String jobId) {
     return jobCache.getIfPresent(jobId);
   }
+
+  public int runningTasks() {
+    return ((ThreadPoolExecutor) executorService).getActiveCount();
+  }
+
+  public int queuedTasks() {
+    return ((ThreadPoolExecutor) executorService).getQueue().size();
+  }
 }

management-api-agent-common/src/main/java/io/k8ssandra/metrics/builder/CassandraMetricNameParser.java

@@ -42,6 +42,13 @@ public CassandraMetricNameParser(
     }
   }
 
+  public static CassandraMetricNameParser getDefaultParser(Configuration config) {
+    return new CassandraMetricNameParser(
+        CassandraMetricsTools.DEFAULT_LABEL_NAMES,
+        CassandraMetricsTools.DEFAULT_LABEL_VALUES,
+        config);
+  }
+
   private void parseEnvVariablesAsLabels(Map<String, String> envSettings) {
     for (Map.Entry<String, String> entry : envSettings.entrySet()) {
       String envValue = System.getenv(entry.getValue());

management-api-agent-common/src/main/java/io/k8ssandra/metrics/builder/CassandraMetricRegistryListener.java

@@ -77,11 +77,7 @@ public class CassandraMetricRegistryListener implements MetricRegistryListener {
   public CassandraMetricRegistryListener(
       ConcurrentHashMap<String, RefreshableMetricFamilySamples> familyCache, Configuration config)
       throws NoSuchMethodException {
-    parser =
-        new CassandraMetricNameParser(
-            CassandraMetricsTools.DEFAULT_LABEL_NAMES,
-            CassandraMetricsTools.DEFAULT_LABEL_VALUES,
-            config);
+    parser = CassandraMetricNameParser.getDefaultParser(config);
     cache = new ConcurrentHashMap<>();
 
     this.familyCache = familyCache;

management-api-agent-common/src/main/java/io/k8ssandra/metrics/config/Configuration.java

@@ -21,6 +21,9 @@ public class Configuration {
   @JsonProperty("labels")
   private LabelConfiguration labels;
 
+  @JsonProperty("extended_metrics_disabled")
+  private boolean extendedDisabled;
+
   public Configuration() {
     relabels = new ArrayList<>();
   }
@@ -44,4 +47,12 @@ public void setRelabels(List<RelabelSpec> relabels) {
   public void setLabels(LabelConfiguration labels) {
     this.labels = labels;
   }
+
+  public boolean isExtendedDisabled() {
+    return extendedDisabled;
+  }
+
+  public void setExtendedDisabled(boolean extendedDisabled) {
+    this.extendedDisabled = extendedDisabled;
+  }
 }

management-api-agent-common/src/main/java/io/k8ssandra/metrics/interceptors/MetricsInterceptor.java

@@ -9,6 +9,7 @@
 import io.k8ssandra.metrics.config.Configuration;
 import io.k8ssandra.metrics.http.NettyMetricsHttpServer;
 import io.k8ssandra.metrics.prometheus.CassandraDropwizardExports;
+import io.k8ssandra.metrics.prometheus.CassandraTasksExports;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.epoll.EpollEventLoopGroup;
 import io.prometheus.client.hotspot.DefaultExports;
@@ -72,6 +73,11 @@ public static void intercept(@SuperCall Callable<Void> zuper) throws Exception {
       // Add JVM metrics
       DefaultExports.initialize();
 
+      // Add task metrics
+      if (!config.isExtendedDisabled()) {
+        new CassandraTasksExports(CassandraMetricsRegistry.Metrics, config).register();
+      }
+
       // Create /metrics handler. Note, this doesn't support larger than nThreads=1
       final EventLoopGroup httpGroup = new EpollEventLoopGroup(1);
 
@@ -81,12 +87,7 @@ public static void intercept(@SuperCall Callable<Void> zuper) throws Exception {
 
       logger.info("Metrics collector started");
 
-      Runtime.getRuntime()
-          .addShutdownHook(
-              new Thread(
-                  () -> {
-                    httpGroup.shutdownGracefully();
-                  }));
+      Runtime.getRuntime().addShutdownHook(new Thread(httpGroup::shutdownGracefully));
     } catch (Throwable t) {
       logger.error("Unable to start metrics endpoint", t);
     }