Merge pull request #461 from yue9944882/feature/informer-impl

Java informer portal implementation

Author: k8s-ci-robot

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

@@ -0,0 +1,86 @@
+package io.kubernetes.client.examples;
+
+import io.kubernetes.client.ApiClient;
+import io.kubernetes.client.ApiException;
+import io.kubernetes.client.Configuration;
+import io.kubernetes.client.apis.CoreV1Api;
+import io.kubernetes.client.informer.*;
+import io.kubernetes.client.informer.cache.Lister;
+import io.kubernetes.client.models.V1Node;
+import io.kubernetes.client.models.V1NodeList;
+import io.kubernetes.client.models.V1ObjectMeta;
+import io.kubernetes.client.util.CallGeneratorParams;
+import io.kubernetes.client.util.Config;
+
+public class InformerExample {
+  public static void main(String[] args) throws Exception {
+
+    ApiClient client = Config.defaultClient();
+    Configuration.setDefaultApiClient(client);
+
+    SharedInformerFactory factory = new SharedInformerFactory();
+
+    CoreV1Api coreV1Api = new CoreV1Api();
+
+    // Node informer
+    SharedIndexInformer<V1Node> nodeInformer =
+        factory.sharedIndexInformerFor(
+            (CallGeneratorParams params) -> {
+              try {
+                return coreV1Api.listNodeCall(
+                    null,
+                    null,
+                    null,
+                    null,
+                    null,
+                    null,
+                    params.resourceVersion,
+                    params.timeoutSeconds,
+                    params.watch,
+                    null,
+                    null);
+              } catch (ApiException e) {
+                throw new RuntimeException(e);
+              }
+            },
+            V1Node.class,
+            V1NodeList.class);
+
+    nodeInformer.addEventHandler(
+        new ResourceEventHandler<V1Node>() {
+          @Override
+          public void onAdd(V1Node node) {
+            System.out.printf("%s node added!\n", node.getMetadata().getName());
+          }
+
+          @Override
+          public void onUpdate(V1Node oldNode, V1Node newNode) {
+            System.out.printf(
+                "%s => %s node updated!\n",
+                oldNode.getMetadata().getName(), newNode.getMetadata().getName());
+          }
+
+          @Override
+          public void onDelete(V1Node node, boolean deletedFinalStateUnknown) {
+            System.out.printf("%s node deleted!\n", node.getMetadata().getName());
+          }
+        });
+
+    factory.startAllRegisteredInformers();
+
+    V1Node nodeToCreate = new V1Node();
+    V1ObjectMeta metadata = new V1ObjectMeta();
+    metadata.setName("noxu");
+    nodeToCreate.setMetadata(metadata);
+    String s = coreV1Api.getApiClient().getJSON().serialize(nodeToCreate);
+    V1Node createdNode = coreV1Api.createNode(nodeToCreate, null, null, null);
+    Thread.sleep(3000);
+
+    Lister<V1Node> nodeLister = new Lister<V1Node>(nodeInformer.getIndexer());
+    V1Node node = nodeLister.get("noxu");
+    System.out.printf("noxu created! %s\n", node.getMetadata().getCreationTimestamp());
+    factory.stopAllRegisteredInformers();
+    Thread.sleep(3000);
+    System.out.println("informer stopped..");
+  }
+}

util/pom.xml

@@ -140,8 +140,8 @@
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-compiler-plugin</artifactId>
                 <configuration>
-                    <source>1.7</source>
-                    <target>1.7</target>
+                    <source>8</source>
+                    <target>8</target>
                 </configuration>
             </plugin>
             <plugin>

util/src/main/java/io/kubernetes/client/informer/EventType.java

@@ -0,0 +1,28 @@
+package io.kubernetes.client.informer;
+
+public enum EventType {
+  ADDED,
+
+  MODIFIED,
+
+  DELETED,
+
+  ERROR;
+
+  /**
+   * getByType returns the corresponding EventType by type.
+   *
+   * @param type specific code
+   * @return corresponding EventType
+   */
+  public static EventType getByType(String type) {
+    if (type != null && type.length() > 0) {
+      for (EventType eventType : EventType.values()) {
+        if (eventType.name().equalsIgnoreCase(type)) {
+          return eventType;
+        }
+      }
+    }
+    return null;
+  }
+}

util/src/main/java/io/kubernetes/client/informer/ListerWatcher.java

@@ -0,0 +1,12 @@
+package io.kubernetes.client.informer;
+
+import io.kubernetes.client.ApiException;
+import io.kubernetes.client.util.CallGeneratorParams;
+import io.kubernetes.client.util.Watchable;
+
+public interface ListerWatcher<ApiType, ApiListType> {
+
+  ApiListType list(CallGeneratorParams params) throws ApiException;
+
+  Watchable<ApiType> watch(CallGeneratorParams params) throws ApiException;
+}

util/src/main/java/io/kubernetes/client/informer/ResourceEventHandler.java

@@ -0,0 +1,10 @@
+package io.kubernetes.client.informer;
+
+public interface ResourceEventHandler<ApiType> {
+
+  void onAdd(ApiType obj);
+
+  void onUpdate(ApiType oldObj, ApiType newObj);
+
+  void onDelete(ApiType obj, boolean deletedFinalStateUnknown);
+}

util/src/main/java/io/kubernetes/client/informer/ResyncRunnable.java

@@ -0,0 +1,36 @@
+package io.kubernetes.client.informer;
+
+import io.kubernetes.client.informer.cache.Store;
+import java.util.function.Supplier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * ResyncRunnable class implements Runnable interface. It calls the resync function of Store
+ * interface which is actually always implemented by DeltaFIFO.
+ */
+public class ResyncRunnable<ApiType> implements Runnable {
+
+  private static final Logger log = LoggerFactory.getLogger(ResyncRunnable.class);
+
+  private Store<ApiType> store;
+  private Supplier<Boolean> shouldResyncFunc;
+
+  public ResyncRunnable(Store<ApiType> store, Supplier<Boolean> shouldResyncFunc) {
+    this.store = store;
+    this.shouldResyncFunc = shouldResyncFunc;
+  }
+
+  public void run() {
+    if (log.isDebugEnabled()) {
+      log.debug("ResyncRunnable#resync ticker tick");
+    }
+
+    if (shouldResyncFunc == null || shouldResyncFunc.get()) {
+      if (log.isDebugEnabled()) {
+        log.debug("ResyncRunnable#force resync");
+      }
+      this.store.resync();
+    }
+  }
+}

util/src/main/java/io/kubernetes/client/informer/SharedIndexInformer.java

@@ -0,0 +1,26 @@
+package io.kubernetes.client.informer;
+
+import io.kubernetes.client.informer.cache.Indexer;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+
+/*
+ * SharedIndexInformer extends SharedInformer and provides indexer operability additionally.
+ */
+public interface SharedIndexInformer<ApiType> extends SharedInformer<ApiType> {
+
+  /**
+   * Add indexers.
+   *
+   * @param indexers the indexers
+   */
+  void addIndexers(Map<String, Function<ApiType, List<String>>> indexers);
+
+  /**
+   * getIndexer returns the internal indexer store.
+   *
+   * @return the internal indexer store
+   */
+  Indexer<ApiType> getIndexer();
+}

util/src/main/java/io/kubernetes/client/informer/SharedInformer.java

@@ -0,0 +1,43 @@
+package io.kubernetes.client.informer;
+
+/*
+ * SharedInformer defines basic methods of a informer.
+ */
+public interface SharedInformer<ApiType> {
+
+  /**
+   * Add event handler.
+   *
+   * @param handler the handler
+   */
+  void addEventHandler(ResourceEventHandler<ApiType> handler);
+
+  /**
+   * addEventHandlerWithResyncPeriod adds an event handler to the shared informer using the
+   * specified resync period. Events to a single handler are delivered sequentially, but there is no
+   * coordination between different handlers.
+   *
+   * @param handler the event handler
+   * @param resyncPeriod the specific resync period
+   */
+  void addEventHandlerWithResyncPeriod(ResourceEventHandler<ApiType> handler, long resyncPeriod);
+
+  /** run starts the shared informer, which will be stopped until stop() is called. */
+  void run();
+
+  /** stop stops the shared informer. */
+  void stop();
+
+  /** hasSynced returns true if the shared informer's store has synced. */
+  boolean hasSynced();
+
+  /**
+   * Last sync resource version string.
+   *
+   * @return the string
+   */
+  // LastSyncResourceVersion is the resource version observed when last synced with the underlying
+  // store. The value returned is not synchronized with access to the underlying store and is not
+  // thread-safe.
+  String lastSyncResourceVersion();
+}

util/src/main/java/io/kubernetes/client/informer/SharedInformerFactory.java

@@ -0,0 +1,136 @@
+package io.kubernetes.client.informer;
+
+import com.google.gson.reflect.TypeToken;
+import com.squareup.okhttp.Call;
+import io.kubernetes.client.*;
+import io.kubernetes.client.informer.impl.DefaultSharedIndexInformer;
+import io.kubernetes.client.util.CallGeneratorParams;
+import io.kubernetes.client.util.Watch;
+import io.kubernetes.client.util.common.Collections;
+import java.lang.reflect.Type;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.function.Function;
+
+/** SharedInformerFactory class constructs and caches informers for api types. */
+public class SharedInformerFactory {
+
+  private Map<Type, SharedIndexInformer> informers;
+
+  private Map<Type, Future> startedInformers;
+
+  private ExecutorService informerExecutor;
+
+  /** Constructor w/ default thread pool. */
+  public SharedInformerFactory() {
+    this(Executors.newCachedThreadPool());
+  }
+
+  /**
+   * Constructor w/ thread pool specified.
+   *
+   * @param threadPool specified thread pool
+   */
+  public SharedInformerFactory(ExecutorService threadPool) {
+    informerExecutor = threadPool;
+    informers = new HashMap<>();
+    startedInformers = new HashMap<>();
+  }
+
+  /**
+   * Shared index informer for shared index informer.
+   *
+   * @param <ApiType> the type parameter
+   * @param <ApiListType> the type parameter
+   * @param callGenerator the call generator
+   * @param apiTypeClass the api type class
+   * @param apiListTypeClass the api list type class
+   * @return the shared index informer
+   */
+  public synchronized <ApiType, ApiListType> SharedIndexInformer<ApiType> sharedIndexInformerFor(
+      Function<CallGeneratorParams, Call> callGenerator,
+      Class<ApiType> apiTypeClass,
+      Class<ApiListType> apiListTypeClass) {
+    return sharedIndexInformerFor(callGenerator, apiTypeClass, apiListTypeClass, 0);
+  }
+
+  /**
+   * Constructs and returns a shared index informer w/ resync period specified. And the informer
+   * cache will be overwritten.
+   *
+   * @param <ApiType> the type parameter
+   * @param <ApiListType> the type parameter
+   * @param callGenerator the call generator
+   * @param apiTypeClass the api type class
+   * @param apiListTypeClass the api list type class
+   * @param resyncPeriodInMillis the resync period in millis
+   * @return the shared index informer
+   */
+  public synchronized <ApiType, ApiListType> SharedIndexInformer<ApiType> sharedIndexInformerFor(
+      Function<CallGeneratorParams, Call> callGenerator,
+      Class<ApiType> apiTypeClass,
+      Class<ApiListType> apiListTypeClass,
+      long resyncPeriodInMillis) {
+    ListerWatcher<ApiType, ApiListType> listerWatcher =
+        listerWatcherFor(callGenerator, apiTypeClass, apiListTypeClass);
+    SharedIndexInformer<ApiType> informer =
+        new DefaultSharedIndexInformer<ApiType, ApiListType>(
+            apiTypeClass, listerWatcher, resyncPeriodInMillis);
+    this.informers.put(TypeToken.get(apiTypeClass).getType(), informer);
+    return informer;
+  }
+
+  private <ApiType, ApiListType> ListerWatcher<ApiType, ApiListType> listerWatcherFor(
+      Function<CallGeneratorParams, Call> callGenerator,
+      Class<ApiType> apiTypeClass,
+      Class<ApiListType> apiListTypeClass) {
+    ApiClient apiClient = Configuration.getDefaultApiClient();
+    return new ListerWatcher<ApiType, ApiListType>() {
+      @Override
+      public ApiListType list(CallGeneratorParams params) throws ApiException {
+        Call call = callGenerator.apply(params);
+        return (ApiListType) apiClient.execute(call, apiListTypeClass).getData();
+      }
+
+      @Override
+      public Watch<ApiType> watch(CallGeneratorParams params) throws ApiException {
+        Call call = callGenerator.apply(params);
+        return Watch.createWatch(
+            apiClient,
+            call,
+            TypeToken.getParameterized(Watch.Response.class, apiTypeClass).getType());
+      }
+    };
+  }
+
+  /** Start all registered informers. */
+  public synchronized void startAllRegisteredInformers() {
+    if (Collections.isEmptyMap(informers)) {
+      return;
+    }
+    informers.forEach(
+        (informerType, informer) -> {
+          if (!startedInformers.containsKey(informerType)) {
+            startedInformers.put(informerType, informerExecutor.submit(informer::run));
+          }
+        });
+  }
+
+  /** Stop all registered informers. */
+  public synchronized void stopAllRegisteredInformers() {
+    if (Collections.isEmptyMap(informers)) {
+      return;
+    }
+    informers.forEach(
+        (informerType, informer) -> {
+          if (startedInformers.containsKey(informerType)) {
+            startedInformers.remove(informerType);
+            informer.stop();
+          }
+        });
+    informerExecutor.shutdown();
+  }
+}