Skip to content

[ZEPPELIN-5876] Implement DockerInterpreterProcess.isAlive() using container state #5316

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
hyunw9 wants to merge 1 commit into
base: master
Choose a base branch
from
+91 −22
Open

[ZEPPELIN-5876] Implement DockerInterpreterProcess.isAlive() using container state #5316

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
47 changes: 45 additions & 2 deletions ...cker/src/main/java/org/apache/zeppelin/interpreter/launcher/DockerInterpreterProcess.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@
import com.spotify.docker.client.messages.Container;
import com.spotify.docker.client.messages.ContainerConfig;
import com.spotify.docker.client.messages.ContainerCreation;
import com.spotify.docker.client.messages.ContainerState;
import com.spotify.docker.client.messages.ExecCreation;
import com.spotify.docker.client.messages.HostConfig;
import com.spotify.docker.client.messages.PortBinding;
Expand Down Expand Up @@ -466,8 +467,20 @@ public int getPort() {

@Override
public boolean isAlive() {
//TODO(ZEPPELIN-5876): Implement it more accurately
return isRunning();
DockerClient client = docker;
if (client == null) {
return false;
}
try {
ContainerState state = client.inspectContainer(containerName).state();
return Boolean.TRUE.equals(state.running()) || Boolean.TRUE.equals(state.paused());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return false;
} catch (Exception e) {
LOGGER.warn("Failed to inspect container {} for liveness", containerName, e);
return false;
}
}

@Override
Expand All @@ -480,6 +493,36 @@ public boolean isRunning() {

@Override
public String getErrorMessage() {
DockerClient client = docker;
if (client == null) {
return null;
}
try {
ContainerState state = client.inspectContainer(containerName).state();
return describeContainerFailure(state);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return null;
} catch (Exception e) {
LOGGER.warn("Failed to inspect container {} for error message", containerName, e);
return null;
}
}

// Returns null when the container is still running or exited cleanly,
// otherwise a human-readable reason for the failure.
private String describeContainerFailure(ContainerState state) {
if (Boolean.TRUE.equals(state.running())) {
return null;
}
if (Boolean.TRUE.equals(state.oomKilled())) {
return "Interpreter container " + containerName
+ " was OOMKilled (exitCode=" + state.exitCode() + ")";
}
Long exitCode = state.exitCode();
if (exitCode != null && exitCode != 0) {
return "Interpreter container " + containerName + " exited with code " + exitCode;
}
return null;
}

Expand Down
66 changes: 46 additions & 20 deletions .../src/test/java/org/apache/zeppelin/interpreter/launcher/DockerInterpreterProcessTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,8 @@
import com.spotify.docker.client.exceptions.DockerException;
import com.spotify.docker.client.messages.ContainerConfig;
import com.spotify.docker.client.messages.ContainerCreation;
import com.spotify.docker.client.messages.ContainerInfo;
import com.spotify.docker.client.messages.ContainerState;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars;
import org.apache.zeppelin.interpreter.InterpreterOption;
Expand Down Expand Up @@ -69,7 +71,17 @@ private DockerInterpreterProcess newProcess() {
5000, 10);
}

// stop() must always close the DockerClient, even when killing the container
// Stubs docker.inspectContainer(...).state() and returns the ContainerState mock
// so each test can set running/paused/oomKilled/exitCode as needed.
private ContainerState stubContainerState(DockerClient mockDocker) throws Exception {
ContainerInfo info = mock(ContainerInfo.class);
ContainerState state = mock(ContainerState.class);
when(mockDocker.inspectContainer(anyString())).thenReturn(info);
when(info.state()).thenReturn(state);
return state;
}

// #1: stop() must always close the DockerClient, even when killing the container
// fails, so the underlying HTTP socket / file descriptors are never leaked.
@Test
void stop_alwaysClosesDockerClient_evenWhenKillContainerFails() throws Exception {
Expand All @@ -83,49 +95,63 @@ void stop_alwaysClosesDockerClient_evenWhenKillContainerFails() throws Exception
verify(mockDocker, times(1)).close();
}

// When start() fails after the container has been started (e.g. file copy / exec
// fails), the container must be cleaned up instead of being left orphaned.
// #2: when start() fails after the container has been started, it must be rolled
// back (kill + remove) instead of being left orphaned.
@Test
void start_removesContainer_whenContainerPreparationFails() throws Exception {
DockerInterpreterProcess intp = spy(newProcess());
DockerClient mockDocker = mock(DockerClient.class);
doReturn(mockDocker).when(intp).createDockerClient(anyString());

// No pre-existing container to remove.
when(mockDocker.listContainers(any())).thenReturn(Collections.emptyList());
// Container is created and started successfully...
when(mockDocker.createContainer(any(ContainerConfig.class), anyString()))
.thenReturn(ContainerCreation.builder().id("test-container-id").build());
// ...but preparing it (the first exec inside the container) fails.
// Container is created and started, then preparation (the first exec) fails.
doThrow(new DockerException("exec failed"))
.when(mockDocker).execCreate(anyString(), any(String[].class), any());

assertThrows(IOException.class, () -> intp.start("user1"));

// The container was started, so start() must roll it back before returning.
verify(mockDocker).startContainer("test-container-id");
verify(mockDocker).killContainer(anyString());
verify(mockDocker).removeContainer(anyString());
}

// #3: isAlive() reflects the container's actual state from the Docker daemon,
// not just whether the Thrift port is reachable.
@Test
void start_removesContainer_evenWhenKillFailsDuringCleanup() throws Exception {
DockerInterpreterProcess intp = spy(newProcess());
void isAlive_trueWhenContainerRunning() throws Exception {
DockerInterpreterProcess intp = newProcess();
DockerClient mockDocker = mock(DockerClient.class);
doReturn(mockDocker).when(intp).createDockerClient(anyString());
intp.docker = mockDocker;
ContainerState state = stubContainerState(mockDocker);
when(state.running()).thenReturn(true);

when(mockDocker.listContainers(any())).thenReturn(Collections.emptyList());
// Container is created...
when(mockDocker.createContainer(any(ContainerConfig.class), anyString()))
.thenReturn(ContainerCreation.builder().id("test-container-id").build());
// ...but fails to start, so it is created-but-not-running.
doThrow(new DockerException("start failed")).when(mockDocker).startContainer(anyString());
// Killing a non-running container fails, but removeContainer must still fire.
doThrow(new DockerException("not running")).when(mockDocker).killContainer(anyString());
assertTrue(intp.isAlive());
}

assertThrows(IOException.class, () -> intp.start("user1"));
@Test
void isAlive_falseWhenContainerExitedOrOomKilled() throws Exception {
DockerInterpreterProcess intp = newProcess();
DockerClient mockDocker = mock(DockerClient.class);
intp.docker = mockDocker;
ContainerState state = stubContainerState(mockDocker);
when(state.running()).thenReturn(false); // exited / OOMKilled -> running == false

verify(mockDocker).removeContainer(anyString());
assertFalse(intp.isAlive());
}

// #3: getErrorMessage() surfaces the failure reason from the container state.
@Test
void getErrorMessage_reportsOomKilled() throws Exception {
DockerInterpreterProcess intp = newProcess();
DockerClient mockDocker = mock(DockerClient.class);
intp.docker = mockDocker;
ContainerState state = stubContainerState(mockDocker);
when(state.oomKilled()).thenReturn(true);
when(state.exitCode()).thenReturn(137L);

assertTrue(intp.getErrorMessage().contains("OOMKilled"));
}

@Test
Expand Down
Loading