PUBDEV-6852 - Kubernetes support (#4370)

* PUBDEV-6852 - Kubernetes support

* Lookup constraints

* Creating host-extracting pattern only once per lookup.

* Improved cluster startup logging - environment variable values crucial for H2O cloud forming are logged.

* Introduced the concept of EmbeddedConfigProvider as proposed by Michal Kurka

* H2O-K8S module uses H2O logging.

* EmbeddedConfigProvider uses full Class name with package included in getName method

* Reverted accidental changes in H2O.java

* Renamed H2OKubernetesEmbeddedConfigProvider to KubernetesEmbeddedConfigProvider

* Removed SLF4J dependencies. Removed JAR task from h2o-k8s.

* Test for KubernetesEmbeddedConfigProvider.

* Documentation and class names changes.

* K8S Readme.md

* Fixed documentation typos

* Documentation enhancements

* Environment variable keys renaming.

* H2O.getEmbeddedH2OConfig now uses map/orElse chain.

* H2O K8S inside main assembly. Removed from H2OApp.

* Container resource requests 4 Gigabytes of memory. Image name placeholder.

* Warn on DNS Lookup naming exception.

* Remove duplicate inclusion of h2o-k8s module in settings.gradle

* KubernetesEmbeddedConfigProvider is active only on 'H2O_KUBERNETES_SERVICE_DNS' env var presence.

* Fix 'Routess' typo

* Do not rely on H2O.exit() to do System.exit in KubernetesEmbeddedConfig.

build.gradle

@@ -94,7 +94,8 @@ ext {
       project(':h2o-hive'),
       project(':h2o-security'),
       project(':h2o-logger'),
-      project(':h2o-genmodel-extensions:jgrapht')
+      project(':h2o-genmodel-extensions:jgrapht'),
+      project(':h2o-k8s')
     ]
 
     javaProjects = [
@@ -128,7 +129,8 @@ ext {
       project(':h2o-hive'),
       project(':h2o-security'),
       project(':h2o-logger'),
-      project(':h2o-genmodel-extensions:jgrapht')
+      project(':h2o-genmodel-extensions:jgrapht'),
+      project(':h2o-k8s')
     ]
 
     scalaProjects = [

h2o-assemblies/main/build.gradle

@@ -28,6 +28,7 @@ dependencies {
         compile project(":h2o-orc-parser")
     }
     compile project(":h2o-parquet-parser")
+    compile project(":h2o-k8s")
     compile "org.slf4j:slf4j-log4j12:1.7.10"
 }
 

h2o-core/src/main/java/water/H2O.java

@@ -43,6 +43,7 @@
 */
 final public class H2O {
   public static final String DEFAULT_JKS_PASS = "h2oh2o";
+  public static final int H2O_DEFAULT_PORT = 54321;
 
   //-------------------------------------------------------------------------------------------------------------------
   // Command-line argument parsing and help
@@ -268,7 +269,7 @@ public static void printHelp() {
     public int port;
 
     /** -baseport=####; Port to start upward searching from. */
-    public int baseport = 54321;
+    public int baseport = H2O_DEFAULT_PORT;
 
     /** -port_offset=####; Offset between the API(=web) port and the internal communication port; api_port + port_offset = h2o_port */
     public int port_offset = 1;
@@ -793,8 +794,60 @@ private static void validateArguments() {
   /**
    * Register embedded H2O configuration object with H2O instance.
    */
-  public static void setEmbeddedH2OConfig(AbstractEmbeddedH2OConfig c) { embeddedH2OConfig = c; }
-  public static AbstractEmbeddedH2OConfig getEmbeddedH2OConfig() { return embeddedH2OConfig; }
+  public static void setEmbeddedH2OConfig(AbstractEmbeddedH2OConfig c) {
+    embeddedH2OConfig = c;
+  }
+
+  /**
+   * Returns an instance of {@link AbstractEmbeddedH2OConfig}. The origin of the embedded config might be either
+   * from directly setting the embeddedH2OConfig field via setEmbeddedH2OConfig setter, or dynamically provided via
+   * service loader. Directly set {@link AbstractEmbeddedH2OConfig} is always prioritized. ServiceLoader lookup is only
+   * performed if no config is previously set.
+   * <p>
+   * Result of first ServiceLoader lookup is also considered final - once a service is found, dynamic lookup is not
+   * performed any further.
+   *
+   * @return An instance of {@link AbstractEmbeddedH2OConfig}, if set or dynamically provided. Otherwise null
+   * @author Michal Kurka
+   */
+  public static AbstractEmbeddedH2OConfig getEmbeddedH2OConfig() {
+    if (embeddedH2OConfig != null) {
+      return embeddedH2OConfig;
+    }
+
+    embeddedH2OConfig = discoverEmbeddedConfigProvider()
+            .map(embeddedConfigProvider -> {
+              Log.info(String.format("Dynamically loaded '%s' as AbstractEmbeddedH2OConfigProvider.", embeddedConfigProvider.getName()));
+              return embeddedConfigProvider.getConfig();
+            }).orElse(null);
+
+    return embeddedH2OConfig;
+  }
+
+  /**
+   * Uses {@link ServiceLoader} to discover active instances of {@link EmbeddedConfigProvider}. Only one provider
+   * may be active at a time. If more providers are detected, {@link IllegalStateException} is thrown.
+   *
+   * @return An {@link Optional} of {@link EmbeddedConfigProvider}, if a single active provider is found. Otherwise
+   * an empty optional.
+   * @throws IllegalStateException When there are multiple active instances {@link EmbeddedConfigProvider} discovered.
+   */
+  private static Optional<EmbeddedConfigProvider> discoverEmbeddedConfigProvider() throws IllegalStateException {
+    final ServiceLoader<EmbeddedConfigProvider> configProviders = ServiceLoader.load(EmbeddedConfigProvider.class);
+    EmbeddedConfigProvider provider = null;
+    for (final EmbeddedConfigProvider candidateProvider : configProviders) {
+      candidateProvider.init();
+      if (!candidateProvider.isActive())
+        continue;
+      if (provider != null) {
+        throw new IllegalStateException("Multiple active EmbeddedH2OConfig providers: " + provider.getName() +
+                " and " + candidateProvider.getName() + " (possibly other as well).");
+      }
+      provider = candidateProvider;
+    }
+
+    return Optional.ofNullable(provider);
+  }
 
   /**
    * Tell the embedding software that this H2O instance belongs to

h2o-core/src/main/java/water/init/EmbeddedConfigProvider.java

@@ -0,0 +1,29 @@
+package water.init;
+
+public interface EmbeddedConfigProvider {
+
+    default String getName() {
+        return getClass().getName();
+    }
+
+    /**
+     * Provider initialization. Guaranteed to be called before any other method is called, including the`isActive`
+     * method.
+     */
+    void init();
+
+    /**
+     * Whether the provider is active and should be used by H2O.
+     *
+     * @return True if H2O should use this {@link EmbeddedConfigProvider}, otherwise false.
+     */
+    default boolean isActive() {
+        return false;
+    }
+
+    /**
+     * @return An instance of {@link AbstractEmbeddedH2OConfig} configuration. Never null.
+     */
+    AbstractEmbeddedH2OConfig getConfig();
+
+}

h2o-k8s/README.md

@@ -0,0 +1,106 @@
+# H2O Kubernetes integration
+
+The integration of Kubernetes and H2O is possible via `water.k8s.KubernetesEmbeddedConfigProvider` - to be found
+in this module. This implementation of `EmbeddedConfigProvider` is dynamically loaded on H2O start and remains inactive
+unless H2O is running in a Docker container managed by Kubernetes is detected.  
+
+## Running H2O in K8s - user's guide
+
+H2O Pods deployed on Kubernetes cluster require a 
+[headless service](https://kubernetes.io/docs/concepts/services-networking/service/#headless-services) 
+for H2O Node discovery. The headless service, instead of load-balancing incoming requests to the underlying
+H2O pods, returns a set of adresses of all the underlying pods. It is therefore the responsibility of the K8S
+cluster administrator to set-up the service correctly to cover H2O nodes only.
+
+### Creating the headless service
+First, a headless service must be created on Kubernetes.  
+
+```yaml
+apiVersion: v1
+kind: Service
+metadata:
+  name: h2o-service
+spec:
+  type: ClusterIP
+  clusterIP: None
+  selector:
+    app: h2o-k8s
+  ports:
+  - protocol: TCP
+    port: 54321
+```
+
+The `clusterIP: None` defines the service as headless. The `port: 54321` is the default H2O port. Users and client libraries
+use this port to talk to the H2O cluster.
+
+The `app: h2o-k8s` setting is of **great importance**, as this is the name of the application with H2O pods inside. 
+Please make sure this setting corresponds to the name of H2O deployment name chosen.
+
+### Creating the H2O deployment
+
+It is **strongly recommended** to run H2O as a [Stateful set](https://kubernetes.io/docs/concepts/workloads/controllers/statefulset/)
+on a Kubernetes cluster. Kubernetes assumes all the pods inside the cluster are stateful and does not attempt to restart
+the individual pods on failure. Once a job is triggered on an H2O cluster, the cluster is locked and no additional nodes
+can be added. Therefore, the cluster has to be restarted as a whole if required - which is a perfect fit for a StatefulSet.
+
+
+```yaml
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: h2o-stateful-set
+  namespace: h2o-statefulset
+spec:
+  serviceName: h2o-service
+  replicas: 3
+  selector:
+    matchLabels:
+      app: h2o-k8s
+  template:
+    metadata:
+      labels:
+        app: h2o-k8s
+    spec:
+      terminationGracePeriodSeconds: 10
+      containers:
+        - name: h2o-k8s
+          image: '<someDockerImageWithH2OInside>'
+          resources:
+            requests:
+              memory: "4Gi"
+          ports:
+            - containerPort: 54321
+              protocol: TCP
+          env:
+          - name: H2O_KUBERNETES_SERVICE_DNS
+            value: h2o-service.h2o-statefulset.svc.cluster.local
+          - name: H2O_NODE_LOOKUP_TIMEOUT
+            value: '180'
+          - name: H2O_NODE_EXPECTED_COUNT
+            value: '3'
+```
+Besides standardized Kubernetes settings, like `replicas: 3` defining the number of pods with H2O instantiated, there are
+several settings to pay attention to.
+
+The name of the application `app: h2o-k8s` must correspond to the name expected by the above-defined headless service in order
+for the H2O node discovery to work. H2O communicates on port 54321, therefore `containerPort: 54321`must be exposed to
+make it possible for the clients to connect.
+
+Environment variables:
+
+1. `H2O_KUBERNETES_SERVICE_DNS` - **[MANDATORY]** Crucial for the clustering to work. The format usually follows the
+ `<service-name>.<project-name>.svc.cluster.local` pattern. This setting enables H2O node discovery via DNS.
+  It must be modified to match the name of the headless service created. Also, pay attention to the rest of the address
+  to match the specifics of your Kubernetes implementation.
+1. `H2O_NODE_LOOKUP_TIMEOUT` - **[OPTIONAL]** Node lookup constraint. Time before the node lookup is ended. 
+1. `H2O_NODE_EXPECTED_COUNT` - **[OPTIONAL]** Node lookup constraint. Expected number of H2O pods to be discovered.
+
+If none of the optional lookup constraints is specified, a sensible default node lookup timeout will be set - currently
+defaults to 3 minutes. If any of the lookup constraints are defined, the H2O node lookup is terminated on whichever 
+condition is met first.
+
+### Exposing H2O cluster
+
+Exposing the H2O cluster is a responsibility of the Kubernetes administrator. By default, an
+ [Ingress](https://kubernetes.io/docs/concepts/services-networking/ingress/) can be created. Different platforms offer
+ different capabilities, e.g. OpenShift offers [Routes](https://docs.openshift.com/container-platform/4.3/networking/routes/route-configuration.html).

h2o-k8s/build.gradle

@@ -0,0 +1,13 @@
+apply plugin: 'java'
+
+sourceCompatibility = 1.8
+
+repositories {
+    mavenCentral()
+}
+
+dependencies {
+    compile project(":h2o-core")
+    testCompile group: 'junit', name: 'junit', version: '4.12'
+    testCompile 'com.github.stefanbirkner:system-rules:1.19.0'
+}

h2o-k8s/src/main/java/water/k8s/KubernetesEmbeddedConfig.java

@@ -0,0 +1,58 @@
+package water.k8s;
+
+import water.H2O;
+import water.init.AbstractEmbeddedH2OConfig;
+import water.util.Log;
+
+import java.net.InetAddress;
+import java.util.Collection;
+
+public class KubernetesEmbeddedConfig extends AbstractEmbeddedH2OConfig {
+
+    private final String flatfile;
+
+    public KubernetesEmbeddedConfig(final Collection<String> nodeIPs) {
+        this.flatfile = writeFlatFile(nodeIPs);
+    }
+
+    private String writeFlatFile(final Collection<String> nodeIPs) {
+        final StringBuilder flatFileBuilder = new StringBuilder();
+
+        nodeIPs.forEach(nodeIP -> {
+            flatFileBuilder.append(nodeIP);
+            flatFileBuilder.append(":");
+            flatFileBuilder.append(H2O.H2O_DEFAULT_PORT); // All pods are expected to utilize the default H2O port
+            flatFileBuilder.append("\n");
+        });
+
+        return flatFileBuilder.toString();
+    }
+
+    @Override
+    public void notifyAboutEmbeddedWebServerIpPort(InetAddress ip, int port) {
+    }
+
+    @Override
+    public void notifyAboutCloudSize(InetAddress ip, int port, InetAddress leaderIp, int leaderPort, int size) {
+        Log.info(String.format("Created cluster of size %d, leader node IP is '%s'", size, leaderIp.toString()));
+    }
+
+    @Override
+    public boolean providesFlatfile() {
+        return true;
+    }
+
+    @Override
+    public String fetchFlatfile() {
+        return flatfile;
+    }
+
+    @Override
+    public void exit(int status) {
+        System.exit(status);
+    }
+
+    @Override
+    public void print() {
+    }
+}