kubectl with exec

zeppelin-interpreter/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java

@@ -665,6 +665,10 @@ public class ZeppelinConfiguration extends XMLConfiguration {
     return getInt(ConfVars.ZEPPELIN_CLUSTER_HEARTBEAT_TIMEOUT);
   }
 
+  public String getK8sKubectlCmd() {
+    return getString(ConfVars.ZEPPELIN_K8S_KUBECTL);
+  }
+
   public String getK8sNamespace() {
     return getString(ConfVars.ZEPPELIN_K8S_NAMESPACE);
   }
@@ -824,6 +828,7 @@ public class ZeppelinConfiguration extends XMLConfiguration {
     ZEPPELIN_CLUSTER_HEARTBEAT_INTERVAL("zeppelin.cluster.heartbeat.interval", 3000),
     ZEPPELIN_CLUSTER_HEARTBEAT_TIMEOUT("zeppelin.cluster.heartbeat.timeout", 9000),
 
+    ZEPPELIN_K8S_KUBECTL("zeppelin.k8s.kubectl", "kubectl"),  // kubectl command
     ZEPPELIN_K8S_NAMESPACE("zeppelin.k8s.namespace", "default"),
     ZEPPELIN_K8S_TEMPLATE_DIR("zeppelin.k8s.template.dir", "k8s"),
 

zeppelin-plugins/launcher/k8s-standard/src/main/java/org/apache/zeppelin/interpreter/launcher/K8sSpecTemplate.java

@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.zeppelin.interpreter.launcher;
+
+import com.hubspot.jinjava.Jinjava;
+import org.apache.commons.io.FileUtils;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.util.HashMap;
+
+public class K8sSpecTemplate extends HashMap<String, Object> {
+  private final Jinjava jinja = new Jinjava();
+
+  public String render(File templateFile) throws IOException {
+    String template = FileUtils.readFileToString(templateFile, Charset.defaultCharset());
+    return render(template);
+  }
+
+  public String render(String template) {
+    return jinja.render(template, this);
+  }
+}

zeppelin-plugins/launcher/k8s-standard/src/main/java/org/apache/zeppelin/interpreter/launcher/K8sStandardInterpreterLauncher.java

@@ -18,7 +18,6 @@
 
 package org.apache.zeppelin.interpreter.launcher;
 
-import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
 import com.google.gson.JsonSyntaxException;
 import com.hubspot.jinjava.Jinjava;
@@ -29,7 +28,6 @@ import io.kubernetes.client.Configuration;
 import io.kubernetes.client.apis.CoreV1Api;
 import io.kubernetes.client.models.V1ConfigMap;
 import io.kubernetes.client.models.V1Pod;
-import io.kubernetes.client.models.V1PodStatus;
 import io.kubernetes.client.models.V1Service;
 import io.kubernetes.client.util.Config;
 import io.kubernetes.client.util.Watch;

zeppelin-plugins/launcher/k8s-standard/src/main/java/org/apache/zeppelin/interpreter/launcher/Kubectl.java

@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.zeppelin.interpreter.launcher;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.gson.Gson;
+import com.google.gson.reflect.TypeToken;
+import org.apache.commons.exec.CommandLine;
+import org.apache.commons.exec.DefaultExecutor;
+import org.apache.commons.exec.ExecuteWatchdog;
+import org.apache.commons.exec.PumpStreamHandler;
+import org.apache.commons.io.IOUtils;
+
+import java.io.*;
+import java.util.Map;
+
+public class Kubectl {
+  private final String kubectlCmd;
+  private final Gson gson = new Gson();
+
+  public Kubectl(String kubectlCmd) {
+    this.kubectlCmd = kubectlCmd;
+  }
+
+  public Map<String, Object> apply(String spec) throws IOException {
+    return execAndGetJson(new String[]{"apply", "-o", "json", "-f", "-"}, spec);
+  }
+
+  public Map<String, Object> delete(String spec) throws IOException {
+    return execAndGetJson(new String[]{"delete", "-o", "json", "-f", "-"}, spec);
+  }
+
+  Map<String, Object> execAndGetJson(String [] args) throws IOException {
+    return execAndGetJson(args, "");
+  }
+
+  @VisibleForTesting
+  Map<String, Object> execAndGetJson(String [] args, String stdin) throws IOException {
+    InputStream ins = IOUtils.toInputStream(stdin);
+    ByteArrayOutputStream stdout = new ByteArrayOutputStream();
+    ByteArrayOutputStream stderr = new ByteArrayOutputStream();
+
+    int exitCode = execute(
+        args,
+        ins,
+        stdout,
+        stderr
+    );
+
+    if (exitCode == 0) {
+      String output = new String(stdout.toByteArray());
+      return gson.fromJson(output, new TypeToken<Map<String, Object>>() {}.getType());
+    } else {
+      throw new IOException(String.format("non zero return code (%d)", exitCode));
+    }
+  }
+
+  public int execute(String [] args, InputStream stdin, OutputStream stdout, OutputStream stderr) throws IOException {
+    DefaultExecutor executor = new DefaultExecutor();
+    CommandLine cmd = new CommandLine(kubectlCmd);
+    cmd.addArguments(args);
+
+    ExecuteWatchdog watchdog = new ExecuteWatchdog(60 * 1000);
+    executor.setWatchdog(watchdog);
+
+    PumpStreamHandler streamHandler = new PumpStreamHandler(stdout, stderr, stdin);
+    executor.setStreamHandler(streamHandler);
+    return executor.execute(cmd);
+  }
+}

zeppelin-plugins/launcher/k8s-standard/src/main/java/org/apache/zeppelin/interpreter/launcher/KubectlPollingWait.java

@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.zeppelin.interpreter.launcher;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+public abstract class KubectlPollingWait implements Runnable {
+  private final Logger logger = LoggerFactory.getLogger(KubectlPollingWait.class);
+  private final Kubectl kubectl;
+  private final int timeoutSec;
+  private final int intervalSec;
+  private final String[] args;
+  private final ScheduledFuture<?> t;
+  private Exception exception;
+  private boolean stopConditionMatched = false;
+  private Map<String, Object> lastResult;
+
+  @VisibleForTesting
+  KubectlPollingWait(Kubectl kubectl, String [] args, int intervalSec, int timeoutSec) {
+    this.kubectl = kubectl;
+    this.args = args;
+    this.timeoutSec = timeoutSec;
+    this.intervalSec = intervalSec;
+    ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
+    t = executor.schedule(this, 0, TimeUnit.SECONDS);
+  }
+
+  @Override
+  public void run() {
+    long start = System.currentTimeMillis();
+    while (System.currentTimeMillis() - start < timeoutSec) {
+      try {
+        Map<String, Object> ret = kubectl.execAndGetJson(args);
+        lastResult = ret;
+        if (shouldStop(ret)) {
+          stopConditionMatched = true;
+          break;
+        } else {
+          Thread.sleep(intervalSec * 1000);
+        }
+      } catch (Exception e) {
+        logger.error("Error", e);
+        exception = e;
+        break;
+      }
+    }
+
+    synchronized (this) {
+      this.notify();
+    }
+  }
+
+  protected abstract boolean shouldStop(Map<String, Object> ret);
+
+  public boolean isDone() {
+    return t != null && t.isDone() && stopConditionMatched;
+  }
+
+  public boolean isRunning() {
+    return !isDone();
+  }
+
+  public Exception getException() {
+    return exception;
+  }
+
+  public Map<String, Object> getLastResult() {
+    return lastResult;
+  }
+
+  public void await() {
+    while (isRunning()) {
+      synchronized (this) {
+        try {
+          this.wait(1000);
+        } catch (InterruptedException e) {
+          // ignore
+        }
+      }
+    }
+  }
+}

zeppelin-plugins/launcher/k8s-standard/src/test/java/org/apache/zeppelin/interpreter/launcher/K8sSpecTemplateTest.java

@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.zeppelin.interpreter.launcher;
+
+import org.junit.Test;
+
+import static org.fusesource.jansi.AnsiRenderer.render;
+import static org.junit.Assert.assertEquals;
+
+public class K8sSpecTemplateTest {
+  @Test
+  public void testRender() {
+    // given template variables
+    K8sSpecTemplate template = new K8sSpecTemplate();
+    template.put("name", "world");
+
+    // when
+    String spec = template.render("Hello {{name}}");
+
+    // then
+    assertEquals("Hello world", spec);
+  }
+}

zeppelin-plugins/launcher/k8s-standard/src/test/java/org/apache/zeppelin/interpreter/launcher/KubectlTest.java

@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.zeppelin.interpreter.launcher;
+
+import org.apache.commons.io.IOUtils;
+import org.junit.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class KubectlTest {
+
+  @Test(expected = IOException.class)
+  public void testKubeclCommandNotExists() throws IOException {
+    // given
+    Kubectl kubectl = new Kubectl("invalidcommand");
+    ByteArrayOutputStream stdout = new ByteArrayOutputStream();
+    ByteArrayOutputStream stderr = new ByteArrayOutputStream();
+
+    // when
+    kubectl.execute(new String[] {}, null, stdout, stderr);
+
+    // then throw IOException
+  }
+
+  @Test
+  public void testStdout() throws IOException {
+    // given
+    Kubectl kubectl = new Kubectl("echo");
+    ByteArrayOutputStream stdout = new ByteArrayOutputStream();
+    ByteArrayOutputStream stderr = new ByteArrayOutputStream();
+
+    // when
+    kubectl.execute(new String[] {"hello"}, null, stdout, stderr);
+
+    // then
+    assertEquals("hello\n", stdout.toString());
+    assertEquals("", stderr.toString());
+  }
+
+  @Test
+  public void testStderr() throws IOException {
+    // given
+    Kubectl kubectl = new Kubectl("sh");
+    ByteArrayOutputStream stdout = new ByteArrayOutputStream();
+    ByteArrayOutputStream stderr = new ByteArrayOutputStream();
+
+    // when
+    try {
+      kubectl.execute(new String[]{"-c", "yoyo"}, null, stdout, stderr);
+    } catch (IOException e) {
+    }
+
+    // then
+    assertEquals("", stdout.toString());
+    assertTrue(0 < stderr.toString().length());
+  }
+
+  @Test
+  public void testStdin() throws IOException {
+    // given
+    Kubectl kubectl = new Kubectl("wc");
+    ByteArrayOutputStream stdout = new ByteArrayOutputStream();
+    ByteArrayOutputStream stderr = new ByteArrayOutputStream();
+    InputStream stdin = IOUtils.toInputStream("Hello");
+
+    // when
+    kubectl.execute(new String[]{"-c"}, stdin, stdout, stderr);
+
+    // then
+    assertEquals("5", stdout.toString().trim());
+    assertEquals("", stderr.toString());
+  }
+
+  @Test
+  public void testExecSpecAndGetJson() throws IOException {
+    // given
+    Kubectl kubectl = new Kubectl("cat");
+    String spec = "{'k1': 'v1', 'k2': 2}";
+
+    // when
+    Map<String, Object> result = kubectl.execAndGetJson(new String[]{}, spec);
+
+    // then
+    assertEquals("v1", result.get("k1"));
+    assertEquals(2.0, result.get("k2"));
+  }
+
+  @Test(expected = com.google.gson.JsonSyntaxException.class)
+  public void testExecSpecAndGetJsonInvalidOutput() throws IOException {
+    // given
+    Kubectl kubectl = new Kubectl("cat");
+    String spec = "Not a json format";
+
+    // when
+    Map<String, Object> result = kubectl.execAndGetJson(new String[]{}, spec);
+
+    // then throw JsonSyntaxException
+  }
+}