feat: feat controller queue size, execution thread count (#1649)

Author: csviri

micrometer-support/src/main/java/io/javaoperatorsdk/operator/monitoring/micrometer/MicrometerMetrics.java

@@ -5,28 +5,56 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.OperatorException;
 import io.javaoperatorsdk.operator.api.monitoring.Metrics;
 import io.javaoperatorsdk.operator.api.reconciler.Constants;
 import io.javaoperatorsdk.operator.api.reconciler.RetryInfo;
+import io.javaoperatorsdk.operator.processing.Controller;
 import io.javaoperatorsdk.operator.processing.GroupVersionKind;
 import io.javaoperatorsdk.operator.processing.event.Event;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
 import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Tag;
 import io.micrometer.core.instrument.Timer;
 
+import static io.javaoperatorsdk.operator.api.reconciler.Constants.CONTROLLER_NAME;
+
 public class MicrometerMetrics implements Metrics {
 
   private static final String PREFIX = "operator.sdk.";
   private static final String RECONCILIATIONS = "reconciliations.";
+  private static final String RECONCILIATIONS_EXECUTIONS = PREFIX + RECONCILIATIONS + "executions.";
+  private static final String RECONCILIATIONS_QUEUE_SIZE = PREFIX + RECONCILIATIONS + "queue.size.";
   private final MeterRegistry registry;
+  private final Map<String, AtomicInteger> gauges = new ConcurrentHashMap<>();
 
   public MicrometerMetrics(MeterRegistry registry) {
     this.registry = registry;
   }
 
+  @Override
+  public void controllerRegistered(Controller<?> controller) {
+    String executingThreadsName =
+        RECONCILIATIONS_EXECUTIONS + controller.getConfiguration().getName();
+    AtomicInteger executingThreads =
+        registry.gauge(executingThreadsName,
+            gvkTags(controller.getConfiguration().getResourceClass()),
+            new AtomicInteger(0));
+    gauges.put(executingThreadsName, executingThreads);
+
+    String controllerQueueName =
+        RECONCILIATIONS_QUEUE_SIZE + controller.getConfiguration().getName();
+    AtomicInteger controllerQueueSize =
+        registry.gauge(controllerQueueName,
+            gvkTags(controller.getConfiguration().getResourceClass()),
+            new AtomicInteger(0));
+    gauges.put(controllerQueueName, controllerQueueSize);
+  }
+
   public <T> T timeControllerExecution(ControllerExecution<T> execution) {
     final var name = execution.controllerName();
     final var execName = PREFIX + "controllers.execution." + execution.name();
@@ -94,13 +122,35 @@ public void reconcileCustomResource(HasMetadata resource, RetryInfo retryInfoNul
         "" + retryInfo.map(RetryInfo::getAttemptCount).orElse(0),
         RECONCILIATIONS + "retries.last",
         "" + retryInfo.map(RetryInfo::isLastAttempt).orElse(true));
+
+    AtomicInteger controllerQueueSize =
+        gauges.get(RECONCILIATIONS_QUEUE_SIZE + metadata.get(CONTROLLER_NAME));
+    controllerQueueSize.incrementAndGet();
   }
 
   @Override
   public void finishedReconciliation(HasMetadata resource, Map<String, Object> metadata) {
     incrementCounter(ResourceID.fromResource(resource), RECONCILIATIONS + "success", metadata);
   }
 
+  @Override
+  public void reconciliationExecutionStarted(HasMetadata resource, Map<String, Object> metadata) {
+    AtomicInteger reconcilerExecutions =
+        gauges.get(RECONCILIATIONS_EXECUTIONS + metadata.get(CONTROLLER_NAME));
+    reconcilerExecutions.incrementAndGet();
+  }
+
+  @Override
+  public void reconciliationExecutionFinished(HasMetadata resource, Map<String, Object> metadata) {
+    AtomicInteger reconcilerExecutions =
+        gauges.get(RECONCILIATIONS_EXECUTIONS + metadata.get(CONTROLLER_NAME));
+    reconcilerExecutions.decrementAndGet();
+
+    AtomicInteger controllerQueueSize =
+        gauges.get(RECONCILIATIONS_QUEUE_SIZE + metadata.get(CONTROLLER_NAME));
+    controllerQueueSize.decrementAndGet();
+  }
+
   public void failedReconciliation(HasMetadata resource, Exception exception,
       Map<String, Object> metadata) {
     var cause = exception.getCause();
@@ -118,6 +168,12 @@ public void failedReconciliation(HasMetadata resource, Exception exception,
     return registry.gaugeMapSize(PREFIX + name + ".size", Collections.emptyList(), map);
   }
 
+  public static List<Tag> gvkTags(Class<? extends HasMetadata> resourceClass) {
+    final var gvk = GroupVersionKind.gvkFor(resourceClass);
+    return List.of(Tag.of("group", gvk.group), Tag.of("version", gvk.version),
+        Tag.of("kind", gvk.kind));
+  }
+
   private void incrementCounter(ResourceID id, String counterName, Map<String, Object> metadata,
       String... additionalTags) {
     final var additionalTagsNb =

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/monitoring/Metrics.java

@@ -6,6 +6,7 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.RetryInfo;
+import io.javaoperatorsdk.operator.processing.Controller;
 import io.javaoperatorsdk.operator.processing.event.Event;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
 
@@ -20,6 +21,11 @@ public interface Metrics {
    */
   Metrics NOOP = new Metrics() {};
 
+  /**
+   * Do initialization if necessary;
+   */
+  default void controllerRegistered(Controller<?> controller) {}
+
   /**
    * Called when an event has been accepted by the SDK from an event source, which would result in
    * potentially triggering the associated Reconciler.
@@ -63,6 +69,12 @@ default void failedReconciliation(HasMetadata resource, Exception exception,
     failedReconciliation(ResourceID.fromResource(resource), exception, metadata);
   }
 
+
+  default void reconciliationExecutionStarted(HasMetadata resource, Map<String, Object> metadata) {}
+
+  default void reconciliationExecutionFinished(HasMetadata resource,
+      Map<String, Object> metadata) {}
+
   /**
    *
    * @deprecated Use (and implement) {@link #cleanupDoneFor(ResourceID, Map)} instead

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/Constants.java

@@ -21,6 +21,7 @@ public final class Constants {
   public static final String SAME_AS_CONTROLLER = "JOSDK_SAME_AS_CONTROLLER";
 
   public static final String RESOURCE_GVK_KEY = "josdk.resource.gvk";
+  public static final String CONTROLLER_NAME = "controller.name";
 
   private Constants() {}
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/Controller.java

@@ -90,10 +90,10 @@ public Controller(Reconciler<P> reconciler,
     eventProcessor = new EventProcessor<>(eventSourceManager);
     eventSourceManager.postProcessDefaultEventSourcesAfterProcessorInitializer();
     controllerHealthInfo = new ControllerHealthInfo(eventSourceManager);
-
     final var context = new EventSourceContext<>(
         eventSourceManager.getControllerResourceEventSource(), configuration, kubernetesClient);
     initAndRegisterEventSources(context);
+    ConfigurationServiceProvider.instance().getMetrics().controllerRegistered(this);
   }
 
   @Override

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java

@@ -1,7 +1,7 @@
 package io.javaoperatorsdk.operator.processing.event;
 
 import java.time.Duration;
-import java.util.Collections;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.ExecutorService;
@@ -18,7 +18,6 @@
 import io.javaoperatorsdk.operator.api.config.ExecutorServiceManager;
 import io.javaoperatorsdk.operator.api.monitoring.Metrics;
 import io.javaoperatorsdk.operator.api.reconciler.Constants;
-import io.javaoperatorsdk.operator.processing.Controller;
 import io.javaoperatorsdk.operator.processing.LifecycleAware;
 import io.javaoperatorsdk.operator.processing.MDCUtils;
 import io.javaoperatorsdk.operator.processing.event.rate.RateLimiter;
@@ -98,9 +97,10 @@ private EventProcessor(
     this.rateLimiter = controllerConfiguration.getRateLimiter();
 
     metricsMetadata = Optional.ofNullable(eventSourceManager.getController())
-        .map(Controller::getAssociatedGroupVersionKind)
-        .map(gvk -> Map.of(Constants.RESOURCE_GVK_KEY, (Object) gvk))
-        .orElse(Collections.emptyMap());
+        .map(c -> Map.of(
+            Constants.RESOURCE_GVK_KEY, c.getAssociatedGroupVersionKind(),
+            Constants.CONTROLLER_NAME, controllerConfiguration.getName()))
+        .orElse(new HashMap<>());
   }
 
   @Override
@@ -409,11 +409,13 @@ public void run() {
         }
         actualResource.ifPresent(executionScope::setResource);
         MDCUtils.addResourceInfo(executionScope.getResource());
+        metrics.reconciliationExecutionStarted(executionScope.getResource(), metricsMetadata);
         thread.setName("ReconcilerExecutor-" + controllerName() + "-" + thread.getId());
         PostExecutionControl<P> postExecutionControl =
             reconciliationDispatcher.handleExecution(executionScope);
         eventProcessingFinished(executionScope, postExecutionControl);
       } finally {
+        metrics.reconciliationExecutionFinished(executionScope.getResource(), metricsMetadata);
         // restore original name
         thread.setName(name);
         MDCUtils.removeResourceInfo();

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/MockControllerConfiguration.java

@@ -14,6 +14,7 @@ public static <R extends HasMetadata> ControllerConfiguration<R> forResource(
     when(configuration.getResourceClass()).thenReturn(resourceType);
     when(configuration.getNamespaces()).thenReturn(DEFAULT_NAMESPACES_SET);
     when(configuration.getEffectiveNamespaces()).thenCallRealMethod();
+    when(configuration.getName()).thenReturn(resourceType.getSimpleName());
     return configuration;
   }
 }

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/ResourceEventFilterTest.java

@@ -135,7 +135,7 @@ public ControllerConfig(String finalizer, boolean generationAware,
         ResourceEventFilter<T> eventFilter, Class<T> customResourceClass) {
       super(
           null,
-          null,
+          "testController",
           null,
           finalizer,
           generationAware,

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerResourceEventSourceTest.java

@@ -178,7 +178,7 @@ public TestConfiguration(boolean generationAware, OnAddFilter<TestCustomResource
         GenericFilter<TestCustomResource> genericFilter) {
       super(
           null,
-          null,
+          "testController",
           null,
           FINALIZER,
           generationAware,