@@ -33,10 +33,12 @@ public class DockerService implements DisposableBean {
33
33
private final DockerClient client ;
34
34
private final Config config ;
35
35
private final ExecutorService executor = Executors .newSingleThreadExecutor ();
36
- private final ConcurrentHashMap <StartupScriptId , String > cachedContainers = new ConcurrentHashMap <>();
36
+ private final ConcurrentHashMap <StartupScriptId , String > cachedContainers =
37
+ new ConcurrentHashMap <>();
37
38
private final StartupScriptsService startupScriptsService ;
38
39
39
- public DockerService (Config config , StartupScriptsService startupScriptsService ) throws InterruptedException , IOException {
40
+ public DockerService (Config config , StartupScriptsService startupScriptsService )
41
+ throws InterruptedException {
40
42
this .startupScriptsService = startupScriptsService ;
41
43
DefaultDockerClientConfig clientConfig =
42
44
DefaultDockerClientConfig .createDefaultConfigBuilder ().build ();
@@ -77,11 +79,11 @@ private void cleanupLeftovers(UUID currentId) {
77
79
*/
78
80
private boolean isImagePresentLocally () {
79
81
return client .listImagesCmd ()
80
- .withFilter ("reference" , List .of (IMAGE_NAME ))
81
- .exec ()
82
- .stream ()
83
- .flatMap (it -> Arrays .stream (it .getRepoTags ()))
84
- .anyMatch (it -> it .endsWith (":" + IMAGE_TAG ));
82
+ .withFilter ("reference" , List .of (IMAGE_NAME ))
83
+ .exec ()
84
+ .stream ()
85
+ .flatMap (it -> Arrays .stream (it .getRepoTags ()))
86
+ .anyMatch (it -> it .endsWith (":" + IMAGE_TAG ));
85
87
}
86
88
87
89
/**
@@ -90,9 +92,9 @@ private boolean isImagePresentLocally() {
90
92
private void pullImage () throws InterruptedException {
91
93
if (!isImagePresentLocally ()) {
92
94
client .pullImageCmd (IMAGE_NAME )
93
- .withTag (IMAGE_TAG )
94
- .exec (new PullImageResultCallback ())
95
- .awaitCompletion (5 , TimeUnit .MINUTES );
95
+ .withTag (IMAGE_TAG )
96
+ .exec (new PullImageResultCallback ())
97
+ .awaitCompletion (5 , TimeUnit .MINUTES );
96
98
}
97
99
}
98
100
@@ -104,28 +106,28 @@ private void pullImage() throws InterruptedException {
104
106
*/
105
107
public String createContainer (String name ) {
106
108
HostConfig hostConfig = HostConfig .newHostConfig ()
107
- .withAutoRemove (true )
108
- .withInit (true )
109
- .withCapDrop (Capability .ALL )
110
- .withNetworkMode ("none" )
111
- .withPidsLimit (2000L )
112
- .withReadonlyRootfs (true )
113
- .withMemory ((long ) config .dockerMaxRamMegaBytes () * 1024 * 1024 )
114
- .withCpuCount ((long ) Math .ceil (config .dockerCPUsUsage ()))
115
- .withCpusetCpus (config .dockerCPUSetCPUs ());
109
+ .withAutoRemove (true )
110
+ .withInit (true )
111
+ .withCapDrop (Capability .ALL )
112
+ .withNetworkMode ("none" )
113
+ .withPidsLimit (2000L )
114
+ .withReadonlyRootfs (true )
115
+ .withMemory ((long ) config .dockerMaxRamMegaBytes () * 1024 * 1024 )
116
+ .withCpuCount ((long ) Math .ceil (config .dockerCPUsUsage ()))
117
+ .withCpusetCpus (config .dockerCPUSetCPUs ());
116
118
117
119
return client .createContainerCmd (IMAGE_NAME + ":" + IMAGE_TAG )
118
- .withHostConfig (hostConfig )
119
- .withStdinOpen (true )
120
- .withAttachStdin (true )
121
- .withAttachStderr (true )
122
- .withAttachStdout (true )
123
- .withEnv ("evalTimeoutSeconds=" + config .evalTimeoutSeconds (),
124
- "sysOutCharLimit=" + config .sysOutCharLimit ())
125
- .withLabels (Map .of (WORKER_LABEL , WORKER_UNIQUE_ID .toString ()))
126
- .withName (name )
127
- .exec ()
128
- .getId ();
120
+ .withHostConfig (hostConfig )
121
+ .withStdinOpen (true )
122
+ .withAttachStdin (true )
123
+ .withAttachStderr (true )
124
+ .withAttachStdout (true )
125
+ .withEnv ("evalTimeoutSeconds=" + config .evalTimeoutSeconds (),
126
+ "sysOutCharLimit=" + config .sysOutCharLimit ())
127
+ .withLabels (Map .of (WORKER_LABEL , WORKER_UNIQUE_ID .toString ()))
128
+ .withName (name )
129
+ .exec ()
130
+ .getId ();
129
131
}
130
132
131
133
/**
@@ -135,14 +137,15 @@ public String createContainer(String name) {
135
137
* @param startupScriptId Script to initialize the container with.
136
138
* @return The ContainerState of the newly created container.
137
139
*/
138
- public ContainerState initializeContainer (String name , StartupScriptId startupScriptId ) throws IOException {
140
+ public ContainerState initializeContainer (String name , StartupScriptId startupScriptId )
141
+ throws IOException {
139
142
if (cachedContainers .isEmpty () || !cachedContainers .containsKey (startupScriptId )) {
140
143
String containerId = createContainer (name );
141
144
return setupContainerWithScript (containerId , true , startupScriptId );
142
145
}
143
146
String containerId = cachedContainers .get (startupScriptId );
144
147
executor .submit (() -> initializeCachedContainer (startupScriptId ));
145
- // Rename container with new name.
148
+
146
149
client .renameContainerCmd (containerId ).withName (name ).exec ();
147
150
return setupContainerWithScript (containerId , false , startupScriptId );
148
151
}
@@ -158,7 +161,8 @@ private void initializeCachedContainer(StartupScriptId startupScriptId) {
158
161
startContainer (id );
159
162
160
163
try (PipedInputStream containerInput = new PipedInputStream ();
161
- BufferedWriter writer = new BufferedWriter (new OutputStreamWriter (new PipedOutputStream (containerInput )))) {
164
+ BufferedWriter writer = new BufferedWriter (
165
+ new OutputStreamWriter (new PipedOutputStream (containerInput )))) {
162
166
attachToContainer (id , containerInput );
163
167
164
168
writer .write (Utils .sanitizeStartupScript (startupScriptsService .get (startupScriptId )));
@@ -180,12 +184,14 @@ private void initializeCachedContainer(StartupScriptId startupScriptId) {
180
184
* @return ContainerState of the spawned container.
181
185
* @throws IOException if an I/O error occurs
182
186
*/
183
- private ContainerState setupContainerWithScript (String containerId , boolean isCached , StartupScriptId startupScriptId ) throws IOException {
187
+ private ContainerState setupContainerWithScript (String containerId , boolean isCached ,
188
+ StartupScriptId startupScriptId ) throws IOException {
184
189
if (!isCached ) {
185
190
startContainer (containerId );
186
191
}
187
192
PipedInputStream containerInput = new PipedInputStream ();
188
- BufferedWriter writer = new BufferedWriter (new OutputStreamWriter (new PipedOutputStream (containerInput )));
193
+ BufferedWriter writer =
194
+ new BufferedWriter (new OutputStreamWriter (new PipedOutputStream (containerInput )));
189
195
190
196
InputStream containerOutput = attachToContainer (containerId , containerInput );
191
197
BufferedReader reader = new BufferedReader (new InputStreamReader (containerOutput ));
@@ -201,6 +207,7 @@ private ContainerState setupContainerWithScript(String containerId, boolean isCa
201
207
202
208
/**
203
209
* Creates a new container
210
+ *
204
211
* @param containerId the ID of the container to start
205
212
*/
206
213
public void startContainer (String containerId ) {
@@ -214,35 +221,38 @@ public void startContainer(String containerId) {
214
221
* Logs any output from stderr and returns an InputStream to read stdout.
215
222
*
216
223
* @param containerId the ID of the running container to attach to
217
- * @param containerInput the input stream (containerInput) to send to the container
224
+ * @param containerInput the input stream (containerInput) to send to the container
218
225
* @return InputStream to read the container's stdout
219
226
* @throws IOException if an I/O error occurs
220
227
*/
221
- public InputStream attachToContainer (String containerId , InputStream containerInput ) throws IOException {
228
+ public InputStream attachToContainer (String containerId , InputStream containerInput )
229
+ throws IOException {
222
230
PipedInputStream pipeIn = new PipedInputStream ();
223
231
PipedOutputStream pipeOut = new PipedOutputStream (pipeIn );
224
232
225
233
client .attachContainerCmd (containerId )
226
- .withLogs (true )
227
- .withFollowStream (true )
228
- .withStdOut (true )
229
- .withStdErr (true )
230
- .withStdIn (containerInput )
231
- .exec (new ResultCallback .Adapter <>() {
232
- @ Override
233
- public void onNext (Frame object ) {
234
- try {
235
- String payloadString = new String (object .getPayload (), StandardCharsets .UTF_8 );
236
- if (object .getStreamType () == StreamType .STDOUT ) {
237
- pipeOut .write (object .getPayload ()); // Write stdout data to pipeOut
238
- } else {
239
- LOGGER .warn ("Received STDERR from container {}: {}" , containerId , payloadString );
240
- }
241
- } catch (IOException e ) {
242
- throw new UncheckedIOException (e );
234
+ .withLogs (true )
235
+ .withFollowStream (true )
236
+ .withStdOut (true )
237
+ .withStdErr (true )
238
+ .withStdIn (containerInput )
239
+ .exec (new ResultCallback .Adapter <>() {
240
+ @ Override
241
+ public void onNext (Frame object ) {
242
+ try {
243
+ String payloadString =
244
+ new String (object .getPayload (), StandardCharsets .UTF_8 );
245
+ if (object .getStreamType () == StreamType .STDOUT ) {
246
+ pipeOut .write (object .getPayload ()); // Write stdout data to pipeOut
247
+ } else {
248
+ LOGGER .warn ("Received STDERR from container {}: {}" , containerId ,
249
+ payloadString );
243
250
}
251
+ } catch (IOException e ) {
252
+ throw new UncheckedIOException (e );
244
253
}
245
- });
254
+ }
255
+ });
246
256
247
257
return pipeIn ;
248
258
}
0 commit comments