Skip to content
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

Jump to bottom

feat(controller): add docker dispatched container to a network #3041

Merged
merged 7 commits into from
Nov 29, 2023
Merged

feat(controller): add docker dispatched container to a network #3041

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
900de2c
docker dispatched container add to a network
anda-ren Nov 28, 2023
99c76a6
keep support for host network
anda-ren Nov 28, 2023
525e311
lint fix
anda-ren Nov 28, 2023
6609565
move network creation from code to ut
anda-ren Nov 29, 2023
7514a96
move network creation from code to ut
anda-ren Nov 29, 2023
a6bc7b9
move network creation from code to ut
anda-ren Nov 29, 2023
0a45f61
use withNetworkMode method instead of link to a network
anda-ren Nov 29, 2023
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
36 changes: 25 additions & 11 deletions ...ntroller/src/main/java/ai/starwhale/mlops/schedule/impl/docker/RunExecutorDockerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,8 @@

import ai.starwhale.mlops.domain.run.bo.Run;
import ai.starwhale.mlops.domain.run.bo.RunStatus;
import ai.starwhale.mlops.exception.SwValidationException;
import ai.starwhale.mlops.exception.SwValidationException.ValidSubject;
import ai.starwhale.mlops.schedule.executor.RunExecutor;
import ai.starwhale.mlops.schedule.impl.container.ContainerCommand;
import ai.starwhale.mlops.schedule.reporting.ReportedRun;
Expand All @@ -32,6 +34,7 @@
import com.github.dockerjava.api.exception.DockerException;
import com.github.dockerjava.api.model.Container;
import com.github.dockerjava.api.model.Frame;
import com.github.dockerjava.api.model.HostConfig;
import com.github.dockerjava.api.model.PullResponseItem;
import java.io.Closeable;
import java.io.IOException;
Expand All @@ -55,27 +58,34 @@ public class RunExecutorDockerImpl implements RunExecutor {
final ExecutorService cmdExecThreadPool;
final ContainerRunMapper containerRunMapper;
final String nodeIp;
final String network;
final String networkName;

static final String NETWORK_HOST = "host";

public RunExecutorDockerImpl(
DockerClientFinder dockerClientFinder,
HostResourceConfigBuilder hostResourceConfigBuilder,
ExecutorService cmdExecThreadPool,
ContainerRunMapper containerRunMapper,
String nodeIp,
String network
String networkName
) {
this.dockerClientFinder = dockerClientFinder;
this.hostResourceConfigBuilder = hostResourceConfigBuilder;
this.cmdExecThreadPool = cmdExecThreadPool;
this.containerRunMapper = containerRunMapper;
this.nodeIp = nodeIp;
this.network = network;
if (networkName.equals("bridge")) {
throw new SwValidationException(
ValidSubject.SETTING,
"default bridge is not supported, use host or user-defined bridge please"
);
}
this.networkName = networkName;
}

@Override
public void run(Run run, RunReportReceiver runReportReceiver) {

DockerClient dockerClient = dockerClientFinder.findProperDockerClient(run.getRunSpec().getResourcePool());
var runSpec = run.getRunSpec();
String image = runSpec.getImage();
Expand All @@ -85,7 +95,6 @@ public void onStart(Closeable closeable) {
ReportedRun reportedRun = ReportedRun.builder()
.id(run.getId())
.status(RunStatus.PENDING)
.ip(nodeIp)
.build();
runReportReceiver.receive(reportedRun);
}
Expand All @@ -102,7 +111,6 @@ public void onError(Throwable throwable) {
.id(run.getId())
.status(RunStatus.FAILED)
.startTimeMillis(System.currentTimeMillis())
.ip(nodeIp)
.stopTimeMillis(System.currentTimeMillis())
.failedReason(throwable.getMessage())
.build();
Expand All @@ -116,13 +124,19 @@ public void onComplete() {
labels.put(ContainerRunMapper.CONTAINER_LABEL_RUN_ID, run.getId().toString());
labels.putAll(CONTAINER_LABELS);

String containerName = containerRunMapper.containerName(run);
boolean hostNetwork = NETWORK_HOST.equals(networkName);
HostConfig hostConfig = hostResourceConfigBuilder
.build(run.getRunSpec().getRequestedResources())
.withNetworkMode(networkName);
CreateContainerCmd createContainerCmd = dockerClient.createContainerCmd(image)
.withEnv(buildEnvs(runSpec.getEnvs()))
.withName(containerRunMapper.containerName(run))
.withHostConfig(hostResourceConfigBuilder
.build(run.getRunSpec().getRequestedResources())
.withNetworkMode(network))
.withName(containerName)
.withHostConfig(hostConfig)
.withLabels(labels);
if (!hostNetwork) {
createContainerCmd.withAliases(containerName);
}
ContainerCommand containerCommand = runSpec.getCommand();
if (null != containerCommand.getEntrypoint()) {
createContainerCmd.withEntrypoint(containerCommand.getEntrypoint());
Expand All @@ -134,7 +148,7 @@ public void onComplete() {
dockerClient.startContainerCmd(createContainerResponse.getId()).exec();
ReportedRun reportedRun = ReportedRun.builder()
.id(run.getId())
.ip(nodeIp)
.ip(hostNetwork ? nodeIp : containerName)
.status(RunStatus.RUNNING)
.startTimeMillis(System.currentTimeMillis())
.build();
Expand Down
46 changes: 33 additions & 13 deletions ...ntroller/src/test/java/ai/starwhale/mlops/schedule/impl/docker/RunExecutorDockerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,9 +41,11 @@
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;

Expand All @@ -53,7 +55,7 @@ public class RunExecutorDockerTest {
DockerClientFinder dockerClientFinder;
ContainerRunMapper containerRunMapper;
ExecutorService cmdExecThreadPool;
String network;
String bridgeNetworkName;
String nodeIp;
RunExecutorDockerImpl runExecutorDocker;
DockerClient dockerClient;
Expand Down Expand Up @@ -87,16 +89,9 @@ public void setup() {
when(containerRunMapper.containerOfRun(any())).thenReturn(container);
when(containerRunMapper.containerName(any())).thenReturn(containerName);
cmdExecThreadPool = Executors.newCachedThreadPool();
network = "host";
bridgeNetworkName = "sw-ut-temp-network";
dockerClient.createNetworkCmd().withName(this.bridgeNetworkName).exec();
nodeIp = "127.1.0.2";
runExecutorDocker = new RunExecutorDockerImpl(
dockerClientFinder,
new HostResourceConfigBuilder(),
cmdExecThreadPool,
containerRunMapper,
nodeIp,
network
);
try {
dockerClient.removeContainerCmd(containerName).withForce(true).exec();
} catch (Exception e) {
Expand All @@ -106,14 +101,39 @@ public void setup() {

}

@Test
public void testExec() throws ExecutionException, InterruptedException {
@AfterEach
public void destroy() {
try {
dockerClient.removeNetworkCmd(bridgeNetworkName).exec();
} catch (Exception e) {
System.out.println(bridgeNetworkName + " delete error " + e.getMessage());
}
}

@ParameterizedTest
@ValueSource(strings = {"sw-ut-temp-network"})
public void testExec(String network) throws ExecutionException, InterruptedException {
runExecutorDocker = new RunExecutorDockerImpl(
dockerClientFinder,
new HostResourceConfigBuilder(),
cmdExecThreadPool,
containerRunMapper,
nodeIp,
network
);
testSchedule(run);
Object lock = new Object();
doAnswer(new Answer() {
@Override
public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
ReportedRun reportedRun = (ReportedRun) invocationOnMock.getArguments()[0];
if (null != reportedRun.getIp()) {
if (network.equals("host")) {
Assertions.assertEquals(nodeIp, reportedRun.getIp());
} else {
Assertions.assertNotEquals(nodeIp, reportedRun.getIp());
}
}
if (reportedRun.getStatus() == RunStatus.RUNNING
|| reportedRun.getStatus() == RunStatus.FAILED) {
synchronized (lock) {
Expand Down
Loading