Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,7 @@ Thumbs.db
**/.settings/
.project
.settings/
.factorypath

# intelliJ IDEA project files
.idea/
Expand Down
8 changes: 7 additions & 1 deletion docs/setup/operation/configuration.md
Original file line number Diff line number Diff line change
Expand Up @@ -418,7 +418,13 @@ If both are defined, then the **environment variables** will take priority.
<td><h6 class="properties">zeppelin.k8s.template.dir</h6></td>
<td>k8s</td>
<td>Kubernetes yaml spec files</td>
</tr>
</tr>
<tr>
<td><h6 class="properties">ZEPPELIN_K8S_SERVICE_NAME</h6></td>
<td><h6 class="properties">zeppelin.k8s.service.name</h6></td>
<td>zeppelin-server</td>
<td>Name of the Zeppelin server service resources</td>
</tr>
</table>


Expand Down
2 changes: 1 addition & 1 deletion k8s/interpreter/100-interpreter-spec.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ spec:
containers:
- name: {{zeppelin.k8s.interpreter.container.name}}
image: {{zeppelin.k8s.interpreter.container.image}}
command: ["sh", "-c", "$(ZEPPELIN_HOME)/bin/interpreter.sh -d $(ZEPPELIN_HOME)/interpreter/{{zeppelin.k8s.interpreter.group.name}} -r {{zeppelin.k8s.interpreter.rpc.portRange}} -c {{zeppelin.k8s.server.rpc.host}} -p {{zeppelin.k8s.server.rpc.portRange}} -i {{zeppelin.k8s.interpreter.group.id}} -l {{zeppelin.k8s.interpreter.localRepo}} -g {{zeppelin.k8s.interpreter.setting.name}}"]
command: ["sh", "-c", "$(ZEPPELIN_HOME)/bin/interpreter.sh -d $(ZEPPELIN_HOME)/interpreter/{{zeppelin.k8s.interpreter.group.name}} -r {{zeppelin.k8s.interpreter.rpc.portRange}} -c {{zeppelin.k8s.server.rpc.service}} -p {{zeppelin.k8s.server.rpc.portRange}} -i {{zeppelin.k8s.interpreter.group.id}} -l {{zeppelin.k8s.interpreter.localRepo}} -g {{zeppelin.k8s.interpreter.setting.name}}"]
lifecycle:
preStop:
exec:
Expand Down
2 changes: 1 addition & 1 deletion k8s/zeppelin-server.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -186,7 +186,7 @@ spec:
kind: Service
apiVersion: v1
metadata:
name: zeppelin-server # keep Service name the same to Pod name.
name: zeppelin-server
spec:
ports:
- name: http
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -836,6 +836,10 @@ public String getK8sTemplatesDir() {
return getRelativeDir(ConfVars.ZEPPELIN_K8S_TEMPLATE_DIR);
}

public String getK8sServiceName() {
return getString(ConfVars.ZEPPELIN_K8S_SERVICE_NAME);
}

public String getDockerContainerImage() {
return getString(ConfVars.ZEPPELIN_DOCKER_CONTAINER_IMAGE);
}
Expand Down Expand Up @@ -1019,6 +1023,7 @@ public enum ConfVars {
ZEPPELIN_K8S_CONTAINER_IMAGE("zeppelin.k8s.container.image", "apache/zeppelin:" + Util.getVersion()),
ZEPPELIN_K8S_SPARK_CONTAINER_IMAGE("zeppelin.k8s.spark.container.image", "apache/spark:latest"),
ZEPPELIN_K8S_TEMPLATE_DIR("zeppelin.k8s.template.dir", "k8s"),
ZEPPELIN_K8S_SERVICE_NAME("zeppelin.k8s.service.name", "zeppelin-server"),

ZEPPELIN_DOCKER_CONTAINER_IMAGE("zeppelin.docker.container.image", "apache/zeppelin:" + Util.getVersion()),

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,6 @@
import org.apache.commons.exec.ExecuteWatchdog;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcess;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterUtils;
import org.slf4j.Logger;
Expand All @@ -31,7 +30,7 @@ public class K8sRemoteInterpreterProcess extends RemoteInterpreterProcess {
private final String containerImage;
private final Properties properties;
private final Map<String, String> envs;
private final String zeppelinServiceHost;
private final String zeppelinService;
private final String zeppelinServiceRpcPort;

private final Gson gson = new Gson();
Expand All @@ -55,7 +54,7 @@ public K8sRemoteInterpreterProcess(
String interpreterSettingName,
Properties properties,
Map<String, String> envs,
String zeppelinServiceHost,
String zeppelinService,
String zeppelinServiceRpcPort,
boolean portForward,
String sparkImage,
Expand All @@ -71,7 +70,7 @@ public K8sRemoteInterpreterProcess(
this.interpreterSettingName = interpreterSettingName;
this.properties = properties;
this.envs = new HashMap<>(envs);
this.zeppelinServiceHost = zeppelinServiceHost;
this.zeppelinService = zeppelinService;
this.zeppelinServiceRpcPort = zeppelinServiceRpcPort;
this.portForward = portForward;
this.sparkImage = sparkImage;
Expand Down Expand Up @@ -266,7 +265,7 @@ Properties getTemplateBindings() {
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.host", zeppelinServiceHost);
k8sProperties.put("zeppelin.k8s.server.rpc.service", zeppelinService);
k8sProperties.put("zeppelin.k8s.server.rpc.portRange", zeppelinServiceRpcPort);
if (ownerUID() != null && ownerName() != null) {
k8sProperties.put("zeppelin.k8s.server.uid", ownerUID());
Expand All @@ -287,7 +286,7 @@ Properties getTemplateBindings() {

// configure interpreter property "zeppelin.spark.uiWebUrl" if not defined, to enable spark ui through reverse proxy
String webUrl = (String) properties.get("zeppelin.spark.uiWebUrl");
if (webUrl == null || webUrl.trim().isEmpty()) {
if (StringUtils.isBlank(webUrl)) {
webUrl = "//{{PORT}}-{{SERVICE_NAME}}.{{SERVICE_DOMAIN}}";
}
properties.put("zeppelin.spark.uiWebUrl",
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -98,14 +98,14 @@ String getHostname() {
}

/**
* get Zeppelin server host dns.
* return <hostname>.<namespace>.svc
* get Zeppelin service.
* return <service-name>.<namespace>.svc
* @throws IOException
*/
private String getZeppelinServiceHost() throws IOException {
private String getZeppelinService() throws IOException {
if (isRunningOnKubernetes()) {
return String.format("%s.%s.svc",
getHostname(), // service name and pod name should be the same
zConf.getK8sServiceName(),
getNamespace());
} else {
return context.getZeppelinServerHost();
Expand Down Expand Up @@ -154,7 +154,7 @@ public InterpreterClient launch(InterpreterLaunchContext context) throws IOExcep
context.getInterpreterSettingName(),
properties,
buildEnvFromProperties(context),
getZeppelinServiceHost(),
getZeppelinService(),
getZeppelinServiceRpcPort(),
zConf.getK8sPortForward(),
zConf.getK8sSparkContainerImage(),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import org.apache.commons.io.IOUtils;

import java.io.*;
import java.nio.charset.StandardCharsets;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down Expand Up @@ -106,7 +107,7 @@ String execAndGet(String [] args) throws IOException {

@VisibleForTesting
String execAndGet(String [] args, String stdin) throws IOException {
InputStream ins = IOUtils.toInputStream(stdin);
InputStream ins = IOUtils.toInputStream(stdin, StandardCharsets.UTF_8);
ByteArrayOutputStream stdout = new ByteArrayOutputStream();
ByteArrayOutputStream stderr = new ByteArrayOutputStream();
ArrayList<String> argsToOverride = new ArrayList<>(Arrays.asList(args));
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -115,7 +115,7 @@ public void testGetTemplateBindings() throws IOException {
"shell",
properties,
envs,
"zeppelin.server.hostname",
"zeppelin.server.service",
"12320",
false,
"spark-container:1.0",
Expand All @@ -135,7 +135,7 @@ public void testGetTemplateBindings() throws IOException {
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.hostname" , p.get("zeppelin.k8s.server.rpc.host"));
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"));
Expand Down Expand Up @@ -168,7 +168,7 @@ public void testGetTemplateBindingsForSpark() throws IOException {
"myspark",
properties,
envs,
"zeppelin.server.hostname",
"zeppelin.server.service",
"12320",
false,
"spark-container:1.0",
Expand Down Expand Up @@ -221,7 +221,7 @@ public void testGetTemplateBindingsForSparkWithProxyUser() throws IOException {
"myspark",
properties,
envs,
"zeppelin.server.hostname",
"zeppelin.server.service",
"12320",
false,
"spark-container:1.0",
Expand Down Expand Up @@ -273,7 +273,7 @@ public void testGetTemplateBindingsForSparkWithProxyUserAnonymous() throws IOExc
"myspark",
properties,
envs,
"zeppelin.server.hostname",
"zeppelin.server.service",
"12320",
false,
"spark-container:1.0",
Expand Down Expand Up @@ -314,7 +314,7 @@ public void testSparkUiWebUrlTemplate() {
"myspark",
properties,
envs,
"zeppelin.server.hostname",
"zeppelin.server.service",
"12320",
false,
"spark-container:1.0",
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -54,8 +54,6 @@ public void testK8sLauncher() throws IOException {
Properties properties = new Properties();
properties.setProperty("ENV_1", "VALUE_1");
properties.setProperty("property_1", "value_1");
properties.setProperty("CALLBACK_HOST", "zeppelin-server.default.svc");
properties.setProperty("CALLBACK_PORT", "12320");
InterpreterOption option = new InterpreterOption();
option.setUserImpersonate(true);
InterpreterLaunchContext context = new InterpreterLaunchContext(
Expand Down Expand Up @@ -88,8 +86,6 @@ public void testK8sLauncherWithSparkAndUserImpersonate() throws IOException {
Properties properties = new Properties();
properties.setProperty("ENV_1", "VALUE_1");
properties.setProperty("property_1", "value_1");
properties.setProperty("CALLBACK_HOST", "zeppelin-server.default.svc");
properties.setProperty("CALLBACK_PORT", "12320");
properties.setProperty("SERVICE_DOMAIN", "example.com");
properties.setProperty("zeppelin.interpreter.connect.timeout", "60");
InterpreterOption option = new InterpreterOption();
Expand Down Expand Up @@ -131,8 +127,6 @@ public void testK8sLauncherWithSparkAndWithoutUserImpersonate() throws IOExcepti
Properties properties = new Properties();
properties.setProperty("ENV_1", "VALUE_1");
properties.setProperty("property_1", "value_1");
properties.setProperty("CALLBACK_HOST", "zeppelin-server.default.svc");
properties.setProperty("CALLBACK_PORT", "12320");
properties.setProperty("SERVICE_DOMAIN", "example.com");
properties.setProperty("zeppelin.interpreter.connect.timeout", "60");
InterpreterOption option = new InterpreterOption();
Expand Down