K8sRemoteInterpreterProcess extends RemoteInterpreterProcess { @VisibleForTesting Properties getTemplateBindings(String userName) { Properties k8sProperties = new Properties(); k8sProperties.put("zeppelin.k8s.namespace", getNamespace()); k8sProperties.put("zeppelin.k8s.interpreter.pod.name", getPodName()); k8sProperties.put("zeppelin.k8s.interpreter.container.name", interpreterGroupName.toLowerCase()); k8sProperties.put("zeppelin.k8s.interpreter.container.image", containerImage); k8sProperties.put("zeppelin.k8s.interpreter.group.id", interpreterGroupId); k8sProperties.put("zeppelin.k8s.interpreter.group.name", interpreterGroupName); k8sProperties.put("zeppelin.k8s.interpreter.setting.name", interpreterSettingName); k8sProperties.put("zeppelin.k8s.interpreter.localRepo", "/tmp/local-repo"); k8sProperties.put("zeppelin.k8s.interpreter.rpc.portRange", String.format("%d:%d", getPort(), getPort())); k8sProperties.put("zeppelin.k8s.server.rpc.service", intpEventServerHost); k8sProperties.put("zeppelin.k8s.server.rpc.portRange", intpEventServerPort); if (ownerUID() != null && ownerName() != null) { k8sProperties.put("zeppelin.k8s.server.uid", ownerUID()); k8sProperties.put("zeppelin.k8s.server.pod.name", ownerName()); } envs.put(ENV_SERVICE_DOMAIN, envs.getOrDefault(ENV_SERVICE_DOMAIN, System.getenv(ENV_SERVICE_DOMAIN))); envs.put(ENV_ZEPPELIN_HOME, envs.getOrDefault(ENV_ZEPPELIN_HOME, System.getenv(ENV_ZEPPELIN_HOME))); if (isSpark()) { int webUiPort = 4040; k8sProperties.put("zeppelin.k8s.spark.container.image", sparkImage); if (isSparkOnKubernetes(properties)) { envs.put("SPARK_SUBMIT_OPTIONS", envs.getOrDefault("SPARK_SUBMIT_OPTIONS", "") + buildSparkSubmitOptions(userName)); } envs.put("SPARK_HOME", envs.getOrDefault("SPARK_HOME", "/spark")); String webUrl = (String) properties.get("zeppelin.spark.uiWebUrl"); if (StringUtils.isBlank(webUrl)) { webUrl = " } properties.put("zeppelin.spark.uiWebUrl", sparkUiWebUrlFromTemplate( webUrl, webUiPort, getPodName(), envs.get(ENV_SERVICE_DOMAIN) )); if (properties.containsKey(SPARK_DRIVER_MEMORY)) { String memory; if (properties.containsKey(SPARK_DRIVER_MEMORY_OVERHEAD)) { memory = K8sUtils.calculateSparkMemory(properties.getProperty(SPARK_DRIVER_MEMORY), properties.getProperty(SPARK_DRIVER_MEMORY_OVERHEAD)); } else { memory = K8sUtils.calculateMemoryWithDefaultOverhead(properties.getProperty(SPARK_DRIVER_MEMORY)); } k8sProperties.put("zeppelin.k8s.interpreter.memory", memory); } if (properties.containsKey(SPARK_DRIVER_CORES)) { k8sProperties.put("zeppelin.k8s.interpreter.cores", properties.getProperty(SPARK_DRIVER_CORES)); } } k8sProperties.put("zeppelin.k8s.envs", envs); k8sProperties.putAll(Maps.fromProperties(properties)); return k8sProperties; } K8sRemoteInterpreterProcess(
          KubernetesClient client,
          String namespace,
          File specTemplates,
          String containerImage,
          String interpreterGroupId,
          String interpreterGroupName,
          String interpreterSettingName,
          Properties properties,
          Map<String, String> envs,
          String intpEventServerHost,
          int intpEventServerPort,
          boolean portForward,
          String sparkImage,
          int connectTimeout,
          int connectionPoolSize,
          boolean isUserImpersonatedForSpark
  ); String getPodName(); String getNamespace(); @Override String getInterpreterGroupId(); @Override String getInterpreterSettingName(); @Override void start(String userName); @Override void stop(); @Override String getHost(); @Override int getPort(); @Override boolean isRunning(); @Override void processStarted(int port, String host); @Override String getErrorMessage();  }
@Test public void testGetTemplateBindings() throws IOException { Properties properties = new Properties(); properties.put("my.key1", "v1"); HashMap<String, String> envs = new HashMap<String, String>(); envs.put("MY_ENV1", "V1"); K8sRemoteInterpreterProcess intp = new K8sRemoteInterpreterProcess( server.getClient(), "default", new File(".skip"), "interpreter-container:1.0", "shared_process", "sh", "shell", properties, envs, "zeppelin.server.service", 12320, false, "spark-container:1.0", 10, 10, false); Properties p = intp.getTemplateBindings(null); assertEquals("default", p.get("zeppelin.k8s.namespace")); assertEquals(intp.getPodName(), p.get("zeppelin.k8s.interpreter.pod.name")); assertEquals("sh", p.get("zeppelin.k8s.interpreter.container.name")); assertEquals("interpreter-container:1.0", p.get("zeppelin.k8s.interpreter.container.image")); assertEquals("shared_process", p.get("zeppelin.k8s.interpreter.group.id")); assertEquals("sh", p.get("zeppelin.k8s.interpreter.group.name")); assertEquals("shell", p.get("zeppelin.k8s.interpreter.setting.name")); assertEquals(true , p.containsKey("zeppelin.k8s.interpreter.localRepo")); assertEquals("12321:12321" , p.get("zeppelin.k8s.interpreter.rpc.portRange")); assertEquals("zeppelin.server.service" , p.get("zeppelin.k8s.server.rpc.service")); assertEquals(12320 , p.get("zeppelin.k8s.server.rpc.portRange")); assertEquals("v1", p.get("my.key1")); assertEquals("V1", envs.get("MY_ENV1")); envs = (HashMap<String, String>) p.get("zeppelin.k8s.envs"); assertTrue(envs.containsKey("SERVICE_DOMAIN")); assertTrue(envs.containsKey("ZEPPELIN_HOME")); }
@Test public void testSparkPodResources() { Properties properties = new Properties(); properties.put("spark.driver.memory", "1g"); properties.put("spark.driver.cores", "1"); HashMap<String, String> envs = new HashMap<String, String>(); envs.put("SERVICE_DOMAIN", "mydomain"); K8sRemoteInterpreterProcess intp = new K8sRemoteInterpreterProcess( server.getClient(), "default", new File(".skip"), "interpreter-container:1.0", "shared_process", "spark", "myspark", properties, envs, "zeppelin.server.service", 12320, false, "spark-container:1.0", 10, 10, false); Properties p = intp.getTemplateBindings(null); assertEquals("1", p.get("zeppelin.k8s.interpreter.cores")); assertEquals("1408Mi", p.get("zeppelin.k8s.interpreter.memory")); }
@Test public void testSparkPodResourcesMemoryOverhead() { Properties properties = new Properties(); properties.put("spark.driver.memory", "1g"); properties.put("spark.driver.memoryOverhead", "256m"); properties.put("spark.driver.cores", "5"); HashMap<String, String> envs = new HashMap<String, String>(); envs.put("SERVICE_DOMAIN", "mydomain"); K8sRemoteInterpreterProcess intp = new K8sRemoteInterpreterProcess( server.getClient(), "default", new File(".skip"), "interpreter-container:1.0", "shared_process", "spark", "myspark", properties, envs, "zeppelin.server.service", 12320, false, "spark-container:1.0", 10, 10, false); Properties p = intp.getTemplateBindings(null); assertEquals("5", p.get("zeppelin.k8s.interpreter.cores")); assertEquals("1280Mi", p.get("zeppelin.k8s.interpreter.memory")); }