From 63293a9b87daf08d9e0a4cf44870ccd8fa040bf1 Mon Sep 17 00:00:00 2001 From: Thomas Hammerl Date: Tue, 21 Aug 2018 13:32:37 +0200 Subject: [PATCH 1/5] Fix deadlock when ryuk does not acknowledge filters --- CHANGELOG.md | 1 + .../utility/ResourceReaper.java | 20 +++++++++++++++---- 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ceccc9319f6..815846afec6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ All notable changes to this project will be documented in this file. ## UNRELEASED ### Fixed +- Fixed deadlock when registering filters with ryuk container ([\#843](https://github.com/testcontainers/testcontainers-java/pull/843)) ### Changed diff --git a/core/src/main/java/org/testcontainers/utility/ResourceReaper.java b/core/src/main/java/org/testcontainers/utility/ResourceReaper.java index 054b216dd19..342cd95cc73 100644 --- a/core/src/main/java/org/testcontainers/utility/ResourceReaper.java +++ b/core/src/main/java/org/testcontainers/utility/ResourceReaper.java @@ -48,6 +48,7 @@ public final class ResourceReaper { private static final Logger LOGGER = LoggerFactory.getLogger(ResourceReaper.class); private static final List>> DEATH_NOTE = new ArrayList<>(); + private static final String ACKNOWLEDGMENT = "ACK"; private static ResourceReaper instance; private final DockerClient dockerClient; @@ -144,11 +145,14 @@ public static String start(String hostIpAddress, DockerClient client, boolean wi out.write('\n'); out.flush(); - while (!"ACK".equalsIgnoreCase(in.readLine())) { + boolean isAcknowledged = waitForAcknowledgment(in); + if (isAcknowledged) { + log.debug("Received 'ACK' from Ryuk"); + ryukScheduledLatch.countDown(); + index++; + } else { + log.debug("Didn't receive 'ACK' from Ryuk. Will retry to send filters."); } - - ryukScheduledLatch.countDown(); - index++; } } } catch (IOException e) { @@ -169,6 +173,14 @@ public static String start(String hostIpAddress, DockerClient client, boolean wi return ryukContainerId; } + private static boolean waitForAcknowledgment(BufferedReader in) throws IOException { + String line = in.readLine(); + while (line != null && !ACKNOWLEDGMENT.equalsIgnoreCase(line)) { + line = in.readLine(); + } + return ACKNOWLEDGMENT.equalsIgnoreCase(line); + } + public synchronized static ResourceReaper instance() { if (instance == null) { instance = new ResourceReaper(); From 8e71ac6c49f1d54815b937a224adf80dcc542e46 Mon Sep 17 00:00:00 2001 From: Thomas Hammerl Date: Tue, 4 Sep 2018 13:22:06 +0200 Subject: [PATCH 2/5] Extract FilterRegistry in ResourceReaper for unit-testability --- .../utility/ResourceReaper.java | 76 ++++++++++++------- .../utility/FilterRegistryTest.java | 66 ++++++++++++++++ 2 files changed, 116 insertions(+), 26 deletions(-) create mode 100644 core/src/test/java/org/testcontainers/utility/FilterRegistryTest.java diff --git a/core/src/main/java/org/testcontainers/utility/ResourceReaper.java b/core/src/main/java/org/testcontainers/utility/ResourceReaper.java index 342cd95cc73..e3f8b3af3fa 100644 --- a/core/src/main/java/org/testcontainers/utility/ResourceReaper.java +++ b/core/src/main/java/org/testcontainers/utility/ResourceReaper.java @@ -13,6 +13,7 @@ import com.github.dockerjava.api.model.Network; import com.github.dockerjava.api.model.Ports; import com.github.dockerjava.api.model.Volume; +import com.google.common.annotations.VisibleForTesting; import lombok.SneakyThrows; import lombok.extern.slf4j.Slf4j; import org.apache.http.client.utils.URLEncodedUtils; @@ -23,6 +24,7 @@ import java.io.BufferedReader; import java.io.IOException; +import java.io.InputStream; import java.io.InputStreamReader; import java.io.OutputStream; import java.net.Socket; @@ -48,7 +50,6 @@ public final class ResourceReaper { private static final Logger LOGGER = LoggerFactory.getLogger(ResourceReaper.class); private static final List>> DEATH_NOTE = new ArrayList<>(); - private static final String ACKNOWLEDGMENT = "ACK"; private static ResourceReaper instance; private final DockerClient dockerClient; @@ -118,8 +119,7 @@ public static String start(String hostIpAddress, DockerClient client, boolean wi while (true) { int index = 0; try(Socket clientSocket = new Socket(hostIpAddress, ryukPort)) { - OutputStream out = clientSocket.getOutputStream(); - BufferedReader in = new BufferedReader(new InputStreamReader(clientSocket.getInputStream())); + FilterRegistry registry = new FilterRegistry(clientSocket.getInputStream(), clientSocket.getOutputStream()); synchronized (DEATH_NOTE) { while (true) { @@ -132,20 +132,7 @@ public static String start(String hostIpAddress, DockerClient client, boolean wi } } List> filters = DEATH_NOTE.get(index); - - String query = URLEncodedUtils.format( - filters.stream() - .map(it -> new BasicNameValuePair(it.getKey(), it.getValue())) - .collect(Collectors.toList()), - (String) null - ); - - log.debug("Sending '{}' to Ryuk", query); - out.write(query.getBytes()); - out.write('\n'); - out.flush(); - - boolean isAcknowledged = waitForAcknowledgment(in); + boolean isAcknowledged = registry.register(filters); if (isAcknowledged) { log.debug("Received 'ACK' from Ryuk"); ryukScheduledLatch.countDown(); @@ -173,14 +160,6 @@ public static String start(String hostIpAddress, DockerClient client, boolean wi return ryukContainerId; } - private static boolean waitForAcknowledgment(BufferedReader in) throws IOException { - String line = in.readLine(); - while (line != null && !ACKNOWLEDGMENT.equalsIgnoreCase(line)) { - line = in.readLine(); - } - return ACKNOWLEDGMENT.equalsIgnoreCase(line); - } - public synchronized static ResourceReaper instance() { if (instance == null) { instance = new ResourceReaper(); @@ -191,7 +170,6 @@ public synchronized static ResourceReaper instance() { /** * Perform a cleanup. - * */ public synchronized void performCleanup() { registeredContainers.forEach(this::stopContainer); @@ -374,4 +352,50 @@ private void setHook() { Runtime.getRuntime().addShutdownHook(new Thread(DockerClientFactory.TESTCONTAINERS_THREAD_GROUP, this::performCleanup)); } } + + static class FilterRegistry { + + @VisibleForTesting + static final String ACKNOWLEDGMENT = "ACK"; + + private final BufferedReader in; + private final OutputStream out; + + FilterRegistry(InputStream ryukInputStream, OutputStream ryukOutputStream) { + this.in = new BufferedReader(new InputStreamReader(ryukInputStream)); + this.out = ryukOutputStream; + } + + /** + * Registers the given filters with Ryuk + * + * @param filters the filter to register + * @return true if the filters have been registered successfuly, false otherwise + * @throws IOException if communication with Ryuk fails + */ + boolean register(List> filters) throws IOException { + String query = URLEncodedUtils.format( + filters.stream() + .map(it -> new BasicNameValuePair(it.getKey(), it.getValue())) + .collect(Collectors.toList()), + (String) null + ); + + log.debug("Sending '{}' to Ryuk", query); + out.write(query.getBytes()); + out.write('\n'); + out.flush(); + + return waitForAcknowledgment(in); + } + + private static boolean waitForAcknowledgment(BufferedReader in) throws IOException { + String line = in.readLine(); + while (line != null && !ACKNOWLEDGMENT.equalsIgnoreCase(line)) { + line = in.readLine(); + } + return ACKNOWLEDGMENT.equalsIgnoreCase(line); + } + + } } diff --git a/core/src/test/java/org/testcontainers/utility/FilterRegistryTest.java b/core/src/test/java/org/testcontainers/utility/FilterRegistryTest.java new file mode 100644 index 00000000000..98e79a7022d --- /dev/null +++ b/core/src/test/java/org/testcontainers/utility/FilterRegistryTest.java @@ -0,0 +1,66 @@ +package org.testcontainers.utility; + +import org.junit.Test; +import org.testcontainers.utility.ResourceReaper.FilterRegistry; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.util.AbstractMap.SimpleEntry; +import java.util.List; +import java.util.Map.Entry; + +import static java.util.Arrays.asList; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +public class FilterRegistryTest { + + private static final List> FILTERS = asList( + new SimpleEntry<>("key1!", "value2?"), new SimpleEntry<>("key2#", "value2%") + ); + private static final String URL_ENCODED_FILTERS = "key1%21=value2%3F&key2%23=value2%25"; + private static final byte[] ACKNOWLEDGEMENT = FilterRegistry.ACKNOWLEDGMENT.getBytes(); + private static final byte[] NO_ACKNOWLEDGEMENT = "".getBytes(); + private static final String NEW_LINE = "\n"; + + @Test + public void registerReturnsTrueIfAcknowledgementIsReadFromInputStream() throws IOException { + FilterRegistry registry = new FilterRegistry(inputStream(ACKNOWLEDGEMENT), anyOutputStream()); + + assertTrue(registry.register(FILTERS)); + } + + @Test + public void registerReturnsFalseIfNoAcknowledgementIsReadFromInputStream() throws IOException { + FilterRegistry registry = new FilterRegistry(inputStream(NO_ACKNOWLEDGEMENT), anyOutputStream()); + + assertFalse(registry.register(FILTERS)); + } + + @Test + public void registerWritesUrlEncodedFiltersAndNewlineToOutputStream() throws IOException { + ByteArrayOutputStream outputStream = new ByteArrayOutputStream(); + FilterRegistry registry = new FilterRegistry(anyInputStream(), outputStream); + + registry.register(FILTERS); + + assertEquals(URL_ENCODED_FILTERS + NEW_LINE, new String(outputStream.toByteArray())); + } + + private static InputStream inputStream(byte[] bytes) { + return new ByteArrayInputStream(bytes); + } + + private static InputStream anyInputStream() { + return inputStream(ACKNOWLEDGEMENT); + } + + private static OutputStream anyOutputStream() { + return new ByteArrayOutputStream(); + } + +} From d72d44332284b136a5eee2cc8246f13038e98a57 Mon Sep 17 00:00:00 2001 From: Thomas Hammerl Date: Tue, 4 Sep 2018 13:31:04 +0200 Subject: [PATCH 3/5] Use explicit scoping for FilterRegistry#register --- .../main/java/org/testcontainers/utility/ResourceReaper.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/src/main/java/org/testcontainers/utility/ResourceReaper.java b/core/src/main/java/org/testcontainers/utility/ResourceReaper.java index e3f8b3af3fa..8d791064643 100644 --- a/core/src/main/java/org/testcontainers/utility/ResourceReaper.java +++ b/core/src/main/java/org/testcontainers/utility/ResourceReaper.java @@ -373,7 +373,7 @@ static class FilterRegistry { * @return true if the filters have been registered successfuly, false otherwise * @throws IOException if communication with Ryuk fails */ - boolean register(List> filters) throws IOException { + protected boolean register(List> filters) throws IOException { String query = URLEncodedUtils.format( filters.stream() .map(it -> new BasicNameValuePair(it.getKey(), it.getValue())) From 7881d05331e270840aababafb6fee051318133cc Mon Sep 17 00:00:00 2001 From: Thomas Hammerl Date: Tue, 4 Sep 2018 14:03:24 +0200 Subject: [PATCH 4/5] Restructure tests --- .../org/testcontainers/utility/FilterRegistryTest.java | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/core/src/test/java/org/testcontainers/utility/FilterRegistryTest.java b/core/src/test/java/org/testcontainers/utility/FilterRegistryTest.java index 98e79a7022d..49bc870e8d0 100644 --- a/core/src/test/java/org/testcontainers/utility/FilterRegistryTest.java +++ b/core/src/test/java/org/testcontainers/utility/FilterRegistryTest.java @@ -31,14 +31,18 @@ public class FilterRegistryTest { public void registerReturnsTrueIfAcknowledgementIsReadFromInputStream() throws IOException { FilterRegistry registry = new FilterRegistry(inputStream(ACKNOWLEDGEMENT), anyOutputStream()); - assertTrue(registry.register(FILTERS)); + boolean successful = registry.register(FILTERS); + + assertTrue(successful); } @Test public void registerReturnsFalseIfNoAcknowledgementIsReadFromInputStream() throws IOException { FilterRegistry registry = new FilterRegistry(inputStream(NO_ACKNOWLEDGEMENT), anyOutputStream()); - assertFalse(registry.register(FILTERS)); + boolean successful = registry.register(FILTERS); + + assertFalse(successful); } @Test From a1b5cbd4b59868a8534be5dd71ffe67d5c69c166 Mon Sep 17 00:00:00 2001 From: Sergei Egorov Date: Wed, 5 Sep 2018 13:44:08 +0200 Subject: [PATCH 5/5] Update CHANGELOG.md --- CHANGELOG.md | 1 - 1 file changed, 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 815846afec6..ceccc9319f6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,7 +4,6 @@ All notable changes to this project will be documented in this file. ## UNRELEASED ### Fixed -- Fixed deadlock when registering filters with ryuk container ([\#843](https://github.com/testcontainers/testcontainers-java/pull/843)) ### Changed