Merge pull request #230 from lanwen/portutils

Waiter until ssh connection can be estabilished

docker-plugin/src/main/java/com/nirima/jenkins/plugins/docker/DockerComputerLauncher.java

@@ -20,6 +20,8 @@
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
+import static java.util.concurrent.TimeUnit.SECONDS;
+
 
 /**
  * {@link hudson.slaves.ComputerLauncher} for Docker that waits for the instance to really come up before proceeding to
@@ -62,7 +64,7 @@ private static SSHLauncher getSSHLauncher(InspectContainerResponse detail, Docke
 
             LOGGER.log(Level.INFO, "Creating slave SSH launcher for " + host + ":" + port);
 
-            PortUtils.waitForPort(host, port);
+            PortUtils.canConnect(host, port).withEveryRetryWaitFor(2, SECONDS);
 
             StandardUsernameCredentials credentials = SSHLauncher.lookupSystemCredentials(template.credentialsId);
 

docker-plugin/src/main/java/com/nirima/jenkins/plugins/docker/utils/PortUtils.java

@@ -1,70 +1,116 @@
 package com.nirima.jenkins.plugins.docker.utils;
 
+import com.trilead.ssh2.Connection;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.io.IOException;
 import java.net.Socket;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static shaded.com.google.common.base.Preconditions.checkState;
 
 public class PortUtils {
+    private static final Logger LOGGER = LoggerFactory.getLogger(PortUtils.class);
 
-    private static final int RETRIES = 10;
-    private static final int WAIT_TIME_MS = 2000;
+    private final String host;
+    private final int port;
 
-    public static boolean isPortAvailable(String host, int port) {
-        Socket socket = null;
-        boolean available = false;
-        try {
-            socket = new Socket(host, port);
-            available = true;
+    private int retries = 10;
+    private int sshTimeoutMillis = (int) SECONDS.toMillis(2);
+
+    private PortUtils(String host, int port) {
+        this.host = host;
+        this.port = port;
+    }
+
+    /**
+     * @param host hostname to connect to
+     * @param port port to open socket
+     *
+     * @return util class to check connection
+     */
+    public static PortUtils canConnect(String host, int port) {
+        return new PortUtils(host, port);
+    }
+
+    public PortUtils withRetries(int retries) {
+        this.retries = retries;
+        return this;
+    }
+
+    public PortUtils withSshTimeout(int time, TimeUnit units) {
+        this.sshTimeoutMillis = (int) units.toMillis(time);
+        return this;
+    }
+
+    /**
+     * @return true if socket opened successfully, false otherwise
+     */
+    public boolean now() {
+        try (Socket ignored = new Socket(host, port)) {
+            return true;
         } catch (IOException e) {
-            // no-op
-        } finally {
-            if (socket != null) {
-                try {
-                    socket.close();
-                } catch (IOException e) {
-                    // no-op
-                }
-            }
+            return false;
         }
-        return available;
     }
 
-    public static boolean waitForPort(String host, int port) {
-        for (int i = 0; i < RETRIES; i++) {
-            if(isPortAvailable(host, port))
+    /**
+     * Use {@link #now()} to check.
+     * Retries while attempts reached with delay
+     *
+     * @return true if socket opened successfully, false otherwise
+     */
+    public boolean withEveryRetryWaitFor(int time, TimeUnit units) {
+        for (int i = 1; i <= retries; i++) {
+            if (now()) {
                 return true;
-
-            try {
-                Thread.sleep(WAIT_TIME_MS);
-            } catch (InterruptedException e) {
-                // no-op
             }
+            sleepFor(time, units);
         }
         return false;
     }
 
-    public static Map<String, List<Integer>> parsePorts(String waitPorts) throws IllegalArgumentException,
-            NumberFormatException {
-        Map<String, List<Integer>> containers = new HashMap<String, List<Integer>>();
-        String[] containerPorts = waitPorts.split(System.getProperty("line.separator"));
-        for (String container : containerPorts) {
-            String[] idPorts = container.split(" ", 2);
-            if (idPorts.length < 2)
-                throw new IllegalArgumentException("Cannot parse " + Arrays.toString(idPorts) + " as '[conainerId] [port1],[port2],...'");
-            String containerId = idPorts[0].trim();
-            String portsStr = idPorts[1].trim();
-
-            List<Integer> ports = new ArrayList<Integer>();
-            for (String port : portsStr.split(",")) {
-                ports.add(Integer.valueOf(port));
+    /**
+     * Connects to sshd on host:port
+     * Retries while attempts reached with delay
+     * First with tcp port wait, then with ssh connection wait
+     *
+     * @throws IOException if no retries left
+     */
+    public void bySshWithEveryRetryWaitFor(int time, TimeUnit units) throws IOException {
+        checkState(withEveryRetryWaitFor(time, units), "Port %s is not opened to connect to", port);
+
+        for (int i = 1; i <= retries; i++) {
+            Connection connection = new Connection(host, port);
+            try {
+                connection.connect(null, 0, sshTimeoutMillis, sshTimeoutMillis);
+                return;
+            } catch (IOException e) {
+                LOGGER.info("Failed to connect to {}:{} (try {}/{}) - {}", host, port, i, retries, e.getMessage());
+                if (i == retries) {
+                    throw e;
+                }
+            } finally {
+                connection.close();
             }
-            containers.put(containerId, ports);
+            sleepFor(time, units);
+        }
+    }
+
+    /**
+     * Blocks current thread for {@code time} of {@code units}
+     *
+     * @param time  number of units
+     * @param units to convert to millis
+     */
+    public static void sleepFor(int time, TimeUnit units) {
+        try {
+            Thread.sleep(units.toMillis(time));
+        } catch (InterruptedException e) {
+            // no-op
         }
-        return containers;
     }
 
 }

docker-plugin/src/test/java/com/nirima/jenkins/plugins/docker/utils/PortUtilsTest.java

@@ -0,0 +1,143 @@
+package com.nirima.jenkins.plugins.docker.utils;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.rules.ExternalResource;
+
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.util.Date;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static com.nirima.jenkins.plugins.docker.utils.PortUtils.canConnect;
+import static java.lang.System.currentTimeMillis;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.both;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.lessThan;
+
+/**
+ * @author lanwen (Merkushev Kirill)
+ */
+public class PortUtilsTest {
+
+    public static final int RETRY_COUNT = 2;
+    public static final int DELAY = (int) SECONDS.toMillis(1);
+
+    @Rule
+    public SomeServerRule server = new SomeServerRule();
+
+    @Rule
+    public ExpectedException ex = ExpectedException.none();
+
+    @Test
+    public void shouldConnectToServerSuccessfully() throws Exception {
+        assertThat("Server is up and should connect", canConnect(server.host(), server.port()).now(), is(true));
+    }
+
+    @Test
+    public void shouldNotConnectToUnusedPort() throws Exception {
+        assertThat("Unused port should not be connectible", canConnect("localhost", 0).now(), is(false));
+    }
+
+    @Test
+    public void shouldWaitForPortAvailableUntilTimeout() throws Exception {
+        long before = currentTimeMillis();
+        assertThat("Unused port should not be connectible",
+                canConnect("localhost", 0).withRetries(RETRY_COUNT)
+                        .withEveryRetryWaitFor(DELAY, MILLISECONDS), is(false));
+        assertThat("Should wait for timeout", new Date(currentTimeMillis()),
+                greaterThanOrEqualTo(new Date(before + RETRY_COUNT * DELAY)));
+    }
+
+    @Test
+    public void shouldThrowIllegalStateExOnNotAvailPort() throws Exception {
+        ex.expect(IllegalStateException.class);
+        canConnect("localhost", 0).withRetries(RETRY_COUNT).bySshWithEveryRetryWaitFor(DELAY, MILLISECONDS);
+    }
+
+    @Test
+    public void shouldWaitIfPortAvailableButNotSshUntilTimeoutAndThrowEx() throws Exception {
+        ex.expect(IOException.class);
+        long before = currentTimeMillis();
+        try {
+            canConnect(server.host(), server.port()).withRetries(RETRY_COUNT)
+                    .bySshWithEveryRetryWaitFor(DELAY, MILLISECONDS);
+        } catch (IOException e) {
+            assertThat("Should wait for timeout", new Date(currentTimeMillis()),
+                    greaterThanOrEqualTo(new Date(before + RETRY_COUNT * DELAY)));
+            throw e;
+        }
+    }
+
+    @Test
+    public void shouldReturnWithoutWaitIfPortAvailable() throws Exception {
+        long before = currentTimeMillis();
+        assertThat("Used port should be connectible",
+                canConnect(server.host(), server.port()).withEveryRetryWaitFor(DELAY, MILLISECONDS), is(true));
+        assertThat("Should not wait", new Date(currentTimeMillis()), lessThan(new Date(before + DELAY)));
+    }
+
+    @Test
+    public void shouldRetryIfPortIsNotAvailableNow() throws Exception {
+        int retries = RETRY_COUNT * 2;
+
+        long before = currentTimeMillis();
+        server.stopAndRebindAfter(2 * DELAY, MILLISECONDS);
+
+        assertThat("Used port should be connectible",
+                canConnect(server.host(), server.port())
+                        .withRetries(retries).withEveryRetryWaitFor(DELAY, MILLISECONDS), is(true));
+
+        assertThat("Should wait then retry", new Date(currentTimeMillis()),
+                both(greaterThanOrEqualTo(new Date(before + 2 * DELAY)))
+                        .and(lessThan(new Date(before + retries * DELAY))));
+    }
+
+    private class SomeServerRule extends ExternalResource {
+        private ServerSocket socket;
+
+        public int port() {
+            return socket.getLocalPort();
+        }
+
+        public String host() {
+            return socket.getInetAddress().getHostAddress();
+        }
+
+        public void stopAndRebindAfter(long delay, TimeUnit unit) throws IOException {
+            final int port = port();
+            socket.close();
+            Executors.newSingleThreadScheduledExecutor().schedule(new Runnable() {
+                @Override
+                public void run() {
+                    try {
+                        socket = new ServerSocket(port);
+                    } catch (IOException e) {
+                        throw new RuntimeException("Can't rebind socket", e);
+                    }
+                }
+            }, delay, unit);
+        }
+
+        @Override
+        protected void before() throws Throwable {
+            socket = new ServerSocket(0);
+            socket.setReuseAddress(true);
+        }
+
+        @Override
+        protected void after() {
+            try {
+                socket.close();
+            } catch (IOException e) {
+                // ignore
+            }
+        }
+    }
+}