[
https://issues.apache.org/jira/browse/BEAM-4258?focusedWorklogId=103496&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-103496
]
ASF GitHub Bot logged work on BEAM-4258:
----------------------------------------
Author: ASF GitHub Bot
Created on: 18/May/18 17:37
Start Date: 18/May/18 17:37
Worklog Time Spent: 10m
Work Description: tgroh closed pull request #5386: [BEAM-4258] Add a new
DockerEnvironmentFactory Constructor
URL: https://github.com/apache/beam/pull/5386
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/environment/DockerCommand.java
b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/environment/DockerCommand.java
index 73959357593..60b341ef244 100644
---
a/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/environment/DockerCommand.java
+++
b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/environment/DockerCommand.java
@@ -36,10 +36,16 @@
/** A docker command wrapper. Simplifies communications with the Docker
daemon. */
class DockerCommand {
+
+ private static final String DEFAULT_DOCKER_COMMAND = "docker";
// TODO: Should we require 64-character container ids? Docker technically
allows abbreviated ids,
// but we _should_ always capture full ids.
private static final Pattern CONTAINER_ID_PATTERN =
Pattern.compile("\\p{XDigit}{64}");
+ public static DockerCommand getDefault() {
+ return forExecutable(DEFAULT_DOCKER_COMMAND, Duration.ofMinutes(2));
+ }
+
static DockerCommand forExecutable(String dockerExecutable, Duration
commandTimeout) {
return new DockerCommand(dockerExecutable, commandTimeout);
}
diff --git
a/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/environment/DockerEnvironmentFactory.java
b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/environment/DockerEnvironmentFactory.java
index c80d9d91478..0eb5d6fc54b 100644
---
a/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/environment/DockerEnvironmentFactory.java
+++
b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/environment/DockerEnvironmentFactory.java
@@ -23,7 +23,6 @@
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeoutException;
-import java.util.function.Supplier;
import org.apache.beam.model.pipeline.v1.RunnerApi.Environment;
import org.apache.beam.runners.fnexecution.GrpcFnServer;
import org.apache.beam.runners.fnexecution.artifact.ArtifactRetrievalService;
@@ -32,6 +31,8 @@
import org.apache.beam.runners.fnexecution.control.InstructionRequestHandler;
import org.apache.beam.runners.fnexecution.logging.GrpcLoggingService;
import
org.apache.beam.runners.fnexecution.provisioning.StaticGrpcProvisionService;
+import org.apache.beam.sdk.fn.IdGenerator;
+import org.apache.beam.sdk.fn.IdGenerators;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -44,15 +45,34 @@
private static final Logger LOG =
LoggerFactory.getLogger(DockerEnvironmentFactory.class);
+ /**
+ * Returns a {@link DockerEnvironmentFactory} for the provided {@link
GrpcFnServer servers} using
+ * the default {@link DockerCommand} and {@link IdGenerators}.
+ */
public static DockerEnvironmentFactory forServices(
+ GrpcFnServer<FnApiControlClientPoolService> controlServiceServer,
+ GrpcFnServer<GrpcLoggingService> loggingServiceServer,
+ GrpcFnServer<ArtifactRetrievalService> retrievalServiceServer,
+ GrpcFnServer<StaticGrpcProvisionService> provisioningServiceServer,
+ ControlClientPool.Source clientSource) {
+ return forServicesWithDocker(
+ DockerCommand.getDefault(),
+ controlServiceServer,
+ loggingServiceServer,
+ retrievalServiceServer,
+ provisioningServiceServer,
+ clientSource,
+ IdGenerators.incrementingLongs());
+ }
+
+ static DockerEnvironmentFactory forServicesWithDocker(
DockerCommand docker,
GrpcFnServer<FnApiControlClientPoolService> controlServiceServer,
GrpcFnServer<GrpcLoggingService> loggingServiceServer,
GrpcFnServer<ArtifactRetrievalService> retrievalServiceServer,
GrpcFnServer<StaticGrpcProvisionService> provisioningServiceServer,
ControlClientPool.Source clientSource,
- // TODO: Refine this to IdGenerator when we determine where that should
live.
- Supplier<String> idGenerator) {
+ IdGenerator idGenerator) {
return new DockerEnvironmentFactory(
docker,
controlServiceServer,
@@ -68,7 +88,7 @@ public static DockerEnvironmentFactory forServices(
private final GrpcFnServer<GrpcLoggingService> loggingServiceServer;
private final GrpcFnServer<ArtifactRetrievalService> retrievalServiceServer;
private final GrpcFnServer<StaticGrpcProvisionService>
provisioningServiceServer;
- private final Supplier<String> idGenerator;
+ private final IdGenerator idGenerator;
private final ControlClientPool.Source clientSource;
private DockerEnvironmentFactory(
@@ -77,7 +97,7 @@ private DockerEnvironmentFactory(
GrpcFnServer<GrpcLoggingService> loggingServiceServer,
GrpcFnServer<ArtifactRetrievalService> retrievalServiceServer,
GrpcFnServer<StaticGrpcProvisionService> provisioningServiceServer,
- Supplier<String> idGenerator,
+ IdGenerator idGenerator,
ControlClientPool.Source clientSource) {
this.docker = docker;
this.controlServiceServer = controlServiceServer;
@@ -91,7 +111,7 @@ private DockerEnvironmentFactory(
/** Creates a new, active {@link RemoteEnvironment} backed by a local Docker
container. */
@Override
public RemoteEnvironment createEnvironment(Environment environment) throws
Exception {
- String workerId = idGenerator.get();
+ String workerId = idGenerator.getId();
// Prepare docker invocation.
Path workerPersistentDirectory =
Files.createTempDirectory("worker_persistent_directory");
diff --git
a/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/environment/DockerCommandTest.java
b/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/environment/DockerCommandTest.java
index 64ae4c72489..d4f1b8e1814 100644
---
a/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/environment/DockerCommandTest.java
+++
b/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/environment/DockerCommandTest.java
@@ -26,7 +26,6 @@
import com.google.common.base.Stopwatch;
import com.google.common.collect.ImmutableList;
import java.io.IOException;
-import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.concurrent.TimeUnit;
@@ -48,14 +47,14 @@
@Test
public void helloWorld() throws Exception {
- DockerCommand docker = getWrapper();
+ DockerCommand docker = DockerCommand.getDefault();
String container = docker.runImage("hello-world", Collections.emptyList());
System.out.printf("Started container: %s%n", container);
}
@Test
public void killContainer() throws Exception {
- DockerCommand docker = getWrapper();
+ DockerCommand docker = DockerCommand.getDefault();
String container = docker.runImage("debian", Arrays.asList("/bin/bash",
"-c", "sleep 60"));
Stopwatch stopwatch = Stopwatch.createStarted();
docker.killContainer(container);
@@ -68,7 +67,7 @@ public void killContainer() throws Exception {
@Test
public void capturesErrorOutput() throws Exception {
- DockerCommand docker = getWrapper();
+ DockerCommand docker = DockerCommand.getDefault();
thrown.expect(instanceOf(IOException.class));
thrown.expectMessage(containsString("Error response from daemon"));
String badImageName = "this-image-should-hopefully-never-exist";
@@ -78,7 +77,4 @@ public void capturesErrorOutput() throws Exception {
Assert.fail(String.format("Container creation for %s should have failed",
badImageName));
}
- private static DockerCommand getWrapper() {
- return DockerCommand.forExecutable("docker", Duration.ofMillis(100_000));
- }
}
diff --git
a/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/environment/DockerEnvironmentFactoryTest.java
b/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/environment/DockerEnvironmentFactoryTest.java
index 5cfaeff7928..35d72eec1c2 100644
---
a/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/environment/DockerEnvironmentFactoryTest.java
+++
b/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/environment/DockerEnvironmentFactoryTest.java
@@ -23,8 +23,6 @@
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.function.Supplier;
import org.apache.beam.model.pipeline.v1.Endpoints.ApiServiceDescriptor;
import org.apache.beam.model.pipeline.v1.RunnerApi.Environment;
import org.apache.beam.runners.fnexecution.GrpcFnServer;
@@ -33,6 +31,8 @@
import org.apache.beam.runners.fnexecution.control.InstructionRequestHandler;
import org.apache.beam.runners.fnexecution.logging.GrpcLoggingService;
import
org.apache.beam.runners.fnexecution.provisioning.StaticGrpcProvisionService;
+import org.apache.beam.sdk.fn.IdGenerator;
+import org.apache.beam.sdk.fn.IdGenerators;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -53,9 +53,7 @@
private static final String CONTAINER_ID =
"e4485f0f2b813b63470feacba5fe9cb89699878c095df4124abd320fd5401385";
- private static final AtomicLong nextId = new AtomicLong(0);
- private static final Supplier<String> ID_GENERATOR =
- () -> Long.toString(nextId.getAndIncrement());
+ private static final IdGenerator ID_GENERATOR =
IdGenerators.incrementingLongs();
@Mock private DockerCommand docker;
@@ -110,7 +108,7 @@ public void createsMultipleEnvironments() throws Exception {
}
private DockerEnvironmentFactory getFactory() {
- return DockerEnvironmentFactory.forServices(
+ return DockerEnvironmentFactory.forServicesWithDocker(
docker,
controlServiceServer,
loggingServiceServer,
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 103496)
Time Spent: 1.5h (was: 1h 20m)
> Integrate Docker Environment Management in the ReferenceRunner
> --------------------------------------------------------------
>
> Key: BEAM-4258
> URL: https://issues.apache.org/jira/browse/BEAM-4258
> Project: Beam
> Issue Type: New Feature
> Components: runner-direct
> Reporter: Thomas Groh
> Assignee: Thomas Groh
> Priority: Major
> Labels: portability
> Time Spent: 1.5h
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
