Merge pull request #1352 from mingjie-li/master

unable to use autowired in the reconciler

Author: k8s-ci-robot

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

@@ -25,17 +25,20 @@
 import io.kubernetes.client.openapi.models.V1Pod;
 import io.kubernetes.client.openapi.models.V1PodList;
 import io.kubernetes.client.spring.extended.controller.KubernetesInformerConfigurer;
-import io.kubernetes.client.spring.extended.controller.KubernetesReconcilerConfigurer;
 import io.kubernetes.client.spring.extended.controller.annotation.*;
+import io.kubernetes.client.spring.extended.controller.factory.KubernetesControllerFactory;
 import io.kubernetes.client.util.ClientBuilder;
 import java.io.IOException;
 import java.time.Duration;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.CommandLineRunner;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.stereotype.Component;
 
 @SpringBootApplication
 public class SpringControllerExample {
@@ -68,11 +71,11 @@ public KubernetesInformerConfigurer kubernetesInformerConfigurer(ApiClient apiCl
     }
 
     // *REQUIRED*
-    // Configurer components that registers reconciler to the controller-manager in the context.
-    @Bean
-    public KubernetesReconcilerConfigurer kubernetesReconcilerConfigurer(
-        SharedInformerFactory sharedInformerFactory) {
-      return new KubernetesReconcilerConfigurer(sharedInformerFactory);
+    // factorybean to crete controller
+    @Bean("node-printing-controller")
+    public KubernetesControllerFactory kubernetesReconcilerConfigurer(
+        SharedInformerFactory sharedInformerFactory, Reconciler reconciler) {
+      return new KubernetesControllerFactory(sharedInformerFactory, reconciler);
     }
 
     // *OPTIONAL*
@@ -91,12 +94,6 @@ public ApiClient myApiClient() throws IOException {
     public SharedInformerFactory sharedInformerFactory() {
       return new MySharedInformerFactory();
     }
-
-    @Bean
-    public NodePrintingReconciler nodePrintingReconciler(
-        Lister<V1Pod> podLister, Lister<V1Node> nodeLister, SharedInformer<V1Node> nodeInformer) {
-      return new NodePrintingReconciler(podLister, nodeLister, nodeInformer);
-    }
   }
 
   @KubernetesInformers({ // Defining what resources is the informer-factory actually watching.
@@ -128,31 +125,33 @@ public static class MySharedInformerFactory extends SharedInformerFactory {}
                 resyncPeriodMillis = 60 * 1000L // fully resync every 1 minute
                 ),
           }))
+  @Component
   public static class NodePrintingReconciler implements Reconciler {
 
-    public NodePrintingReconciler(
-        Lister<V1Pod> podLister, Lister<V1Node> nodeLister, SharedInformer<V1Node> nodeInformer) {
-      this.nodeLister = nodeLister;
-      this.podLister = podLister;
-      this.nodeInformer = nodeInformer;
-    }
+    @Value("${namespace}")
+    private String namespace;
 
-    private SharedInformer<V1Node> nodeInformer;
-
-    private Lister<V1Node> nodeLister;
-
-    private Lister<V1Pod> podLister;
+    @Autowired private SharedInformer<V1Node> nodeInformer;
+    @Autowired private SharedInformer<V1Pod> podInformer;
+    @Autowired private Lister<V1Node> nodeLister;
+    @Autowired private Lister<V1Pod> podLister;
 
     // *OPTIONAL*
     // If you feed like hold the controller from running util some condition..
     @KubernetesReconcilerReadyFunc
-    boolean informerReady() {
-      return nodeInformer.hasSynced();
+    public boolean informerReady() {
+      return podInformer.hasSynced() && nodeInformer.hasSynced();
     }
 
     @Override
     public Result reconcile(Request request) {
       V1Node node = nodeLister.get(request.getName());
+
+      System.out.println("get all pods in namespace " + namespace);
+      podLister.namespace(namespace).list().stream()
+          .map(pod -> pod.getMetadata().getName())
+          .forEach(System.out::println);
+
       System.out.println("triggered reconciling " + node.getMetadata().getName());
       return new Result(false);
     }

examples/src/main/resources/application.properties

@@ -0,0 +1 @@
+namespace=airflow

spring/src/main/java/io/kubernetes/client/spring/extended/controller/KubernetesReconcilerProcessor.java

@@ -12,38 +12,17 @@
 */
 package io.kubernetes.client.spring.extended.controller;
 
-import io.kubernetes.client.common.KubernetesObject;
 import io.kubernetes.client.extended.controller.Controller;
 import io.kubernetes.client.extended.controller.ControllerManager;
-import io.kubernetes.client.extended.controller.builder.ControllerBuilder;
-import io.kubernetes.client.extended.controller.builder.DefaultControllerBuilder;
 import io.kubernetes.client.extended.controller.reconciler.Reconciler;
-import io.kubernetes.client.extended.controller.reconciler.Request;
-import io.kubernetes.client.extended.workqueue.DefaultRateLimitingQueue;
-import io.kubernetes.client.extended.workqueue.RateLimitingQueue;
-import io.kubernetes.client.extended.workqueue.WorkQueue;
 import io.kubernetes.client.informer.SharedInformerFactory;
 import io.kubernetes.client.spring.extended.controller.annotation.*;
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
-import java.time.Duration;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
+import io.kubernetes.client.spring.extended.controller.factory.KubernetesControllerFactory;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.function.BiPredicate;
-import java.util.function.Function;
-import java.util.function.Predicate;
-import java.util.function.Supplier;
-import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.BeansException;
-import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.core.Ordered;
@@ -88,232 +67,12 @@ public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory)
       KubernetesReconciler kubernetesReconciler =
           reconciler.getClass().getAnnotation(KubernetesReconciler.class);
       String reconcilerName = kubernetesReconciler.value();
-      Controller controller = buildController(this.sharedInformerFactory, reconciler);
-      beanFactory.registerSingleton(reconcilerName, controller);
-    }
-  }
-
-  private Controller buildController(SharedInformerFactory sharedInformerFactory, Reconciler r)
-      throws BeansException {
-    KubernetesReconciler kubernetesReconciler =
-        r.getClass().getAnnotation(KubernetesReconciler.class);
-    String reconcilerName = kubernetesReconciler.value();
-
-    KubernetesReconcilerWatches watches = kubernetesReconciler.watches();
-    DefaultControllerBuilder builder = ControllerBuilder.defaultBuilder(sharedInformerFactory);
-    RateLimitingQueue<Request> workQueue = new DefaultRateLimitingQueue<>();
-    builder = builder.withWorkQueue(workQueue);
-    Map<Class, AddFilterAdaptor> addFilters = getAddFilters(watches, r);
-    Map<Class, UpdateFilterAdaptor> updateFilters = getUpdateFilters(watches, r);
-    Map<Class, DeleteFilterAdaptor> deleteFilters = getDeleteFilters(watches, r);
-    List<ReadyFuncAdaptor> readyFuncs = getReadyFuncs(r);
-    for (KubernetesReconcilerWatch watch : watches.value()) {
-      try {
-
-        Function<? extends KubernetesObject, Request> workQueueKeyFunc;
-        try {
-          workQueueKeyFunc =
-              watch.workQueueKeyFunc().getConstructor(WorkQueue.class).newInstance(workQueue);
-        } catch (NoSuchMethodException e) {
-          workQueueKeyFunc = watch.workQueueKeyFunc().newInstance();
-        } catch (InvocationTargetException e) {
-          throw new BeanCreationException(
-              "Failed instantiating controller watch: " + e.getMessage());
-        }
-
-        final Function<? extends KubernetesObject, Request> finalWorkQueueKeyFunc =
-            workQueueKeyFunc;
-        builder =
-            builder.watch(
-                (q) -> {
-                  return ControllerBuilder.controllerWatchBuilder(watch.apiTypeClass(), q)
-                      .withOnAddFilter(addFilters.get(watch.apiTypeClass()))
-                      .withOnUpdateFilter(updateFilters.get(watch.apiTypeClass()))
-                      .withOnDeleteFilter(deleteFilters.get(watch.apiTypeClass()))
-                      .withWorkQueueKeyFunc(finalWorkQueueKeyFunc)
-                      .withResyncPeriod(Duration.ofMillis(watch.resyncPeriodMillis()))
-                      .build();
-                });
-        for (Supplier<Boolean> readyFunc : readyFuncs) {
-          builder = builder.withReadyFunc(readyFunc);
-        }
-      } catch (IllegalAccessException | InstantiationException e) {
-        throw new BeanCreationException("Failed instantiating controller: " + e.getMessage());
-      }
-    }
-
-    builder = builder.withWorkerCount(kubernetesReconciler.workerCount());
-
-    return builder.withReconciler(r).withName(reconcilerName).build();
-  }
-
-  private Map<Class, AddFilterAdaptor> getAddFilters(
-      KubernetesReconcilerWatches watches, Reconciler reconciler) {
-    Map<Class, AddFilterAdaptor> filters = new HashMap<>();
-    Set<Method> allAnnotatedMethods = new HashSet<>();
-    Set<Method> adoptedMethods = new HashSet<>();
-    for (KubernetesReconcilerWatch watch : watches.value()) {
-      for (Method method : reconciler.getClass().getMethods()) {
-        AddWatchEventFilter annotation = method.getAnnotation(AddWatchEventFilter.class);
-        if (annotation != null && watch.apiTypeClass().equals(annotation.apiTypeClass())) {
-          if (filters.containsKey(watch.apiTypeClass())) {
-            log.warn(
-                "Duplicated watch ADD event filter upon apiType {}", annotation.apiTypeClass());
-          }
-          filters.put(watch.apiTypeClass(), new AddFilterAdaptor(reconciler, method));
-          adoptedMethods.add(method);
-        }
-        allAnnotatedMethods.add(method);
-      }
-    }
-    allAnnotatedMethods.removeAll(adoptedMethods);
-    if (allAnnotatedMethods.size() > 0) {
-      log.warn("Dangling watch ADD event filters {}", StringUtils.join(allAnnotatedMethods, ","));
-    }
-    return filters;
-  }
-
-  private Map<Class, UpdateFilterAdaptor> getUpdateFilters(
-      KubernetesReconcilerWatches watches, Reconciler reconciler) {
-    Map<Class, UpdateFilterAdaptor> filters = new HashMap<>();
-    Set<Method> allAnnotatedMethods = new HashSet<>();
-    Set<Method> adoptedMethods = new HashSet<>();
-    for (KubernetesReconcilerWatch watch : watches.value()) {
-      for (Method method : reconciler.getClass().getMethods()) {
-        UpdateWatchEventFilter annotation = method.getAnnotation(UpdateWatchEventFilter.class);
-        if (annotation != null && watch.apiTypeClass().equals(annotation.apiTypeClass())) {
-          if (filters.containsKey(watch.apiTypeClass())) {
-            log.warn(
-                "Duplicated watch UPDATE event filter upon apiType {}", annotation.apiTypeClass());
-          }
-          filters.put(watch.apiTypeClass(), new UpdateFilterAdaptor(reconciler, method));
-          adoptedMethods.add(method);
-        }
-        allAnnotatedMethods.add(method);
-      }
-    }
-    allAnnotatedMethods.removeAll(adoptedMethods);
-    if (allAnnotatedMethods.size() > 0) {
-      log.warn(
-          "Dangling watch UPDATE event filters {}", StringUtils.join(allAnnotatedMethods, ","));
-    }
-    return filters;
-  }
-
-  private Map<Class, DeleteFilterAdaptor> getDeleteFilters(
-      KubernetesReconcilerWatches watches, Reconciler reconciler) {
-    Map<Class, DeleteFilterAdaptor> filters = new HashMap<>();
-    Set<Method> allAnnotatedMethods = new HashSet<>();
-    Set<Method> adoptedMethods = new HashSet<>();
-    for (KubernetesReconcilerWatch watch : watches.value()) {
-      for (Method method : reconciler.getClass().getMethods()) {
-        DeleteWatchEventFilter annotation = method.getAnnotation(DeleteWatchEventFilter.class);
-        if (annotation != null && watch.apiTypeClass().equals(annotation.apiTypeClass())) {
-          if (filters.containsKey(watch.apiTypeClass())) {
-            log.warn(
-                "Duplicated watch DELETE event filter upon apiType {}", annotation.apiTypeClass());
-          }
-          filters.put(watch.apiTypeClass(), new DeleteFilterAdaptor(reconciler, method));
-          adoptedMethods.add(method);
-        }
-        allAnnotatedMethods.add(method);
-      }
-    }
-    allAnnotatedMethods.removeAll(adoptedMethods);
-    if (allAnnotatedMethods.size() > 0) {
-      log.warn(
-          "Dangling watch DELETE event filters {}", StringUtils.join(allAnnotatedMethods, ","));
-    }
-    return filters;
-  }
 
-  private List<ReadyFuncAdaptor> getReadyFuncs(Reconciler reconciler) {
-    List<ReadyFuncAdaptor> readyFuncs = new ArrayList<>();
-    for (Method method : reconciler.getClass().getMethods()) {
-      if (method.isAnnotationPresent(KubernetesReconcilerReadyFunc.class)) {
-        readyFuncs.add(new ReadyFuncAdaptor(reconciler, method));
-      }
-    }
-    return readyFuncs;
-  }
-
-  private static class AddFilterAdaptor implements Predicate {
-    private Method method;
-    private Object target;
-
-    private AddFilterAdaptor(Object target, Method method) {
-      this.method = method;
-      this.target = target;
-    }
-
-    @Override
-    public boolean test(Object addedObj) {
-      try {
-        return (boolean) method.invoke(target, addedObj);
-      } catch (IllegalAccessException | InvocationTargetException e) {
-        log.error("invalid EventAddFilter method signature", e);
-        return true;
-      }
-    }
-  }
-
-  private static class UpdateFilterAdaptor implements BiPredicate {
-    private Method method;
-    private Object target;
-
-    private UpdateFilterAdaptor(Object target, Method method) {
-      this.method = method;
-      this.target = target;
-    }
-
-    @Override
-    public boolean test(Object oldObj, Object newObj) {
-      try {
-        return (boolean) method.invoke(target, oldObj, newObj);
-      } catch (IllegalAccessException | InvocationTargetException e) {
-        log.error("invalid EventUpdateFilter method signature", e);
-        return true;
-      }
-    }
-  }
-
-  private static class DeleteFilterAdaptor implements BiPredicate {
-    private Method method;
-    private Object target;
-
-    private DeleteFilterAdaptor(Object target, Method method) {
-      this.method = method;
-      this.target = target;
-    }
-
-    @Override
-    public boolean test(Object deleteObj, Object cacheStatusUnknown) {
-      try {
-        return (boolean) method.invoke(target, deleteObj, cacheStatusUnknown);
-      } catch (IllegalAccessException | InvocationTargetException e) {
-        log.error("invalid EventDeleteFilter method signature", e);
-        return true;
-      }
-    }
-  }
-
-  private static class ReadyFuncAdaptor implements Supplier<Boolean> {
-    private Method method;
-    private Object target;
+      KubernetesControllerFactory controllerFactory =
+          new KubernetesControllerFactory(sharedInformerFactory, reconciler);
 
-    private ReadyFuncAdaptor(Object target, Method method) {
-      this.method = method;
-      this.target = target;
-    }
-
-    @Override
-    public Boolean get() {
-      try {
-        return (boolean) method.invoke(target);
-      } catch (IllegalAccessException | InvocationTargetException e) {
-        log.error("invalid ReadyFunc method signature", e);
-        return false;
-      }
+      Controller controller = controllerFactory.getObject();
+      beanFactory.registerSingleton(reconcilerName, controller);
     }
   }
 }