Merge pull request #1360 from yue9944882/controller-metrics

Feat: Controller metrics

Author: k8s-ci-robot

examples/pom.xml

@@ -76,6 +76,18 @@
 			<artifactId>wiremock</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<!--for spring controller example-->
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-web</artifactId>
+			<version>${spring.boot.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-actuator</artifactId>
+			<version>${spring.boot.version}</version>
+		</dependency>
+
 	</dependencies>
 	<build>
 		<plugins>

examples/src/main/java/io/kubernetes/client/examples/SpringControllerExample.java

@@ -27,6 +27,7 @@
 import io.kubernetes.client.spring.extended.controller.KubernetesInformerConfigurer;
 import io.kubernetes.client.spring.extended.controller.annotation.*;
 import io.kubernetes.client.spring.extended.controller.factory.KubernetesControllerFactory;
+import io.kubernetes.client.spring.extended.controller.metrics.PrometheusScrapeEndpoint;
 import io.kubernetes.client.util.ClientBuilder;
 import java.io.IOException;
 import java.time.Duration;
@@ -94,6 +95,14 @@ public ApiClient myApiClient() throws IOException {
     public SharedInformerFactory sharedInformerFactory() {
       return new MySharedInformerFactory();
     }
+
+    // *OPTIONAL*
+    // Enabling prometheus scraping endpoint at `/actuator/prometheus`
+    // SHOULD set `management.endpoints.web.exposure.include=prometheus` property.
+    @Bean
+    public PrometheusScrapeEndpoint prometheusScrapeEndpoint() {
+      return new PrometheusScrapeEndpoint();
+    }
   }
 
   @KubernetesInformers({ // Defining what resources is the informer-factory actually watching.

examples/src/main/resources/application.properties

@@ -1 +1,2 @@
-namespace=airflow
+namespace=airflow
+management.endpoints.web.exposure.include=prometheus

extended/src/main/java/io/kubernetes/client/extended/controller/DefaultController.java

@@ -17,6 +17,9 @@
 import io.kubernetes.client.extended.controller.reconciler.Result;
 import io.kubernetes.client.extended.wait.Wait;
 import io.kubernetes.client.extended.workqueue.RateLimitingQueue;
+import io.prometheus.client.CollectorRegistry;
+import io.prometheus.client.Counter;
+import io.prometheus.client.Gauge;
 import java.time.Duration;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
@@ -36,28 +39,58 @@
 public class DefaultController implements Controller {
   private static final Logger log = LoggerFactory.getLogger(DefaultController.class);
 
-  private Reconciler reconciler;
-  private String name;
+  private static Gauge gaugeWorkQueueLength =
+      Gauge.build("controller_work_queue_length", "Current length of the controller's work-queue")
+          .labelNames("name")
+          .register();
+  private static Counter counterControllerReconcile =
+      Counter.build("controller_reconcile_count_total", "Total count of controller reconciliation")
+          .labelNames("name", "requeue")
+          .register();
+
+  private final Reconciler reconciler;
+  private final String name;
+  private final RateLimitingQueue<Request> workQueue;
+  private final Supplier<Boolean>[] readyFuncs;
+
   private int workerCount;
   private ScheduledExecutorService workerThreadPool;
 
-  private RateLimitingQueue<Request> workQueue;
-  private Supplier<Boolean>[] readyFuncs;
-
   private Duration readyTimeout;
   private Duration readyCheckInternal;
 
   /**
    * Instantiates a new Default controller.
    *
+   * @param name the name
    * @param reconciler the reconciler
    * @param workQueue the work queue
    * @param readyFuncs the ready funcs
    */
   public DefaultController(
+      String name,
       Reconciler reconciler,
       RateLimitingQueue<Request> workQueue,
       Supplier<Boolean>... readyFuncs) {
+    this(name, reconciler, workQueue, null, readyFuncs);
+  }
+
+  /**
+   * Instantiates a new Default controller.
+   *
+   * @param name the name
+   * @param reconciler the reconciler
+   * @param workQueue the work queue
+   * @param collectorRegistry the collector registry
+   * @param readyFuncs the ready funcs
+   */
+  public DefaultController(
+      String name,
+      Reconciler reconciler,
+      RateLimitingQueue<Request> workQueue,
+      CollectorRegistry collectorRegistry,
+      Supplier<Boolean>... readyFuncs) {
+    this.name = name;
     this.reconciler = reconciler;
     this.workQueue = workQueue;
     this.readyFuncs = readyFuncs;
@@ -149,6 +182,7 @@ public void shutdown() {
   private void worker() {
     // taking tasks from work-queue in a loop
     while (!workQueue.isShuttingDown()) {
+      gaugeWorkQueueLength.labels(name).set(workQueue.length());
       Request request = null;
       try {
         request = workQueue.get();
@@ -173,6 +207,8 @@ private void worker() {
       } catch (Throwable t) {
         log.error("Reconciler aborted unexpectedly", t);
         result = new Result(true);
+      } finally {
+        counterControllerReconcile.labels(this.name, Boolean.toString(result.isRequeue())).inc();
       }
 
       try {
@@ -194,6 +230,7 @@ private void worker() {
         }
       } finally {
         workQueue.done(request);
+        gaugeWorkQueueLength.labels(name).set(workQueue.length());
         log.debug("Controller {} finished reconciling {}..", this.name, request);
       }
     }
@@ -203,19 +240,10 @@ public RateLimitingQueue<Request> getWorkQueue() {
     return workQueue;
   }
 
-  public DefaultController setWorkQueue(RateLimitingQueue<Request> workQueue) {
-    this.workQueue = workQueue;
-    return this;
-  }
-
   public String getName() {
     return name;
   }
 
-  public void setName(String name) {
-    this.name = name;
-  }
-
   public int getWorkerCount() {
     return workerCount;
   }
@@ -236,10 +264,6 @@ public Reconciler getReconciler() {
     return reconciler;
   }
 
-  public void setReconciler(Reconciler reconciler) {
-    this.reconciler = reconciler;
-  }
-
   public Duration getReadyTimeout() {
     return readyTimeout;
   }

extended/src/main/java/io/kubernetes/client/extended/controller/builder/DefaultControllerBuilder.java

@@ -149,16 +149,16 @@ public Controller build() throws IllegalStateException {
 
     DefaultController controller =
         new DefaultController(
-            this.reconciler, this.workQueue, this.readyFuncs.stream().toArray(Supplier[]::new));
+            this.controllerName,
+            this.reconciler,
+            this.workQueue,
+            this.readyFuncs.stream().toArray(Supplier[]::new));
 
-    controller.setName(this.controllerName);
     controller.setWorkerCount(this.workerCount);
     controller.setWorkerThreadPool(
         Executors.newScheduledThreadPool(
             this.workerCount, Controllers.namedControllerThreadFactory(this.controllerName)));
 
-    controller.setReconciler(this.reconciler);
-
     return controller;
   }
 }

extended/src/test/java/io/kubernetes/client/extended/controller/DefaultControllerTest.java

@@ -61,7 +61,7 @@ public void tearDown() throws Exception {}
   @Test
   public void testStartingStoppingController() throws InterruptedException {
 
-    DefaultController testController = new DefaultController(mockReconciler, workQueue);
+    DefaultController testController = new DefaultController("", mockReconciler, workQueue);
 
     testController.setWorkerCount(1);
     testController.setWorkerThreadPool(Executors.newScheduledThreadPool(1));
@@ -96,7 +96,7 @@ public void testControllerWontStartBeforeReady() throws InterruptedException {
 
     AtomicBoolean ready = new AtomicBoolean(false);
     DefaultController testController =
-        new DefaultController(mockReconciler, workQueue, () -> ready.get());
+        new DefaultController("", mockReconciler, workQueue, () -> ready.get());
     testController.setWorkerCount(1);
     testController.setWorkerThreadPool(Executors.newScheduledThreadPool(1));
     testController.setReadyCheckInternal(Duration.ofMillis(100));
@@ -122,6 +122,7 @@ public void testControllerKeepsWorkingWhenReconcilerAbortsWithRuntimeException()
     List<Request> finishedRequests = new ArrayList<>();
     DefaultController testController =
         new DefaultController(
+            "",
             new Reconciler() {
               @Override
               public Result reconcile(Request request) {

pom.xml

@@ -168,6 +168,12 @@
         <artifactId>spring-boot-autoconfigure</artifactId>
         <version>${spring.boot.version}</version>
       </dependency>
+      <dependency>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-actuator</artifactId>
+        <version>${spring.boot.version}</version>
+        <optional>true</optional>
+      </dependency>
       <dependency>
         <groupId>io.prometheus</groupId>
         <artifactId>simpleclient</artifactId>

spring/pom.xml

@@ -35,6 +35,10 @@
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-autoconfigure</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-actuator</artifactId>
+    </dependency>
 
     <dependency>
       <groupId>junit</groupId>

spring/src/main/java/io/kubernetes/client/spring/extended/controller/metrics/PrometheusScrapeEndpoint.java

@@ -0,0 +1,47 @@
+/*
+Copyright 2020 The Kubernetes Authors.
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License 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 io.kubernetes.client.spring.extended.controller.metrics;
+
+import io.prometheus.client.CollectorRegistry;
+import io.prometheus.client.exporter.common.TextFormat;
+import java.io.IOException;
+import java.io.StringWriter;
+import java.io.Writer;
+import org.springframework.boot.actuate.endpoint.annotation.Endpoint;
+import org.springframework.boot.actuate.endpoint.annotation.ReadOperation;
+
+@Endpoint(id = "prometheus")
+public class PrometheusScrapeEndpoint {
+
+  private final CollectorRegistry collectorRegistry;
+
+  public PrometheusScrapeEndpoint() {
+    this(CollectorRegistry.defaultRegistry);
+  }
+
+  public PrometheusScrapeEndpoint(CollectorRegistry collectorRegistry) {
+    this.collectorRegistry = collectorRegistry;
+  }
+
+  @ReadOperation
+  public String scrape() {
+    try {
+      Writer writer = new StringWriter();
+      TextFormat.write004(writer, collectorRegistry.metricFamilySamples());
+      return writer.toString();
+    } catch (IOException e) {
+      // This actually never happens since StringWriter::write() doesn't throw any IOException
+      throw new RuntimeException("Writing metrics failed", e);
+    }
+  }
+}