[JEP-222] WebSocket-based agents

pom.xml

@@ -94,6 +94,12 @@ THE SOFTWARE.
       <artifactId>animal-sniffer-annotations</artifactId>
       <scope>provided</scope>
     </dependency>
+    <dependency>
+      <groupId>org.glassfish.tyrus.bundles</groupId>
+      <artifactId>tyrus-standalone-client-jdk</artifactId>
+      <version>1.12</version>
+      <scope>provided</scope>
+    </dependency>
     <!-- test dependencies -->
     <dependency>
       <groupId>junit</groupId>
@@ -342,6 +348,18 @@ THE SOFTWARE.
               <includeGroupIds>org.jenkins-ci</includeGroupIds>
             </configuration>
           </execution>
+          <execution>
+            <id>bundle-tyrus</id>
+            <phase>process-classes</phase>
+            <goals>
+              <goal>unpack-dependencies</goal>
+            </goals>
+            <configuration>
+              <outputDirectory>${project.build.outputDirectory}</outputDirectory>
+              <includeScope>provided</includeScope>
+              <includeArtifactIds>tyrus-standalone-client-jdk</includeArtifactIds>
+            </configuration>
+          </execution>
         </executions>
       </plugin>
       <plugin>

src/main/java/hudson/remoting/AbstractByteArrayCommandTransport.java

@@ -44,11 +44,12 @@ public static interface ByteArrayReceiver {
          *
          * As discussed in {@link AbstractByteArrayCommandTransport#writeBlock(Channel, byte[])},
          * the block boundary is significant.
+         * @see CommandReceiver#handle
          */
         void handle(byte[] payload);
 
         /**
-         * See {@link CommandReceiver#handle(Command)} for details.
+         * @see CommandReceiver#terminate
          */
         void terminate(IOException e);
     }

src/main/java/hudson/remoting/Capability.java

@@ -1,6 +1,8 @@
 package hudson.remoting;
 
 import hudson.remoting.Channel.Mode;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.ObjectInputStream;
@@ -24,6 +26,12 @@
  * @see Channel#remoteCapability
  */
 public final class Capability implements Serializable {
+
+    /**
+     * Key usable as a WebSocket HTTP header to negotiate capabilities.
+     */
+    public static final String KEY = "X-Remoting-Capability";
+
     /**
      * Bit mask of optional capabilities.
      */
@@ -120,10 +128,17 @@ public boolean supportsProxyExceptionFallback() {
 
     //TODO: ideally preamble handling needs to be reworked in order to avoid FB suppression
     /**
-     * Writes out the capacity preamble.
+     * Writes {@link #PREAMBLE} then uses {@link #write}.
      */
     void writePreamble(OutputStream os) throws IOException {
         os.write(PREAMBLE);
+        write(os);
+    }
+
+    /**
+     * Writes this capability to a stream.
+     */
+    private void write(OutputStream os) throws IOException {
         try (ObjectOutputStream oos = new ObjectOutputStream(Mode.TEXT.wrap(os)) {
             @Override
             public void close() throws IOException {
@@ -143,7 +158,7 @@ protected void annotateClass(Class<?> c) throws IOException {
     }
 
     /**
-     * The opposite operation of {@link #writePreamble(OutputStream)}.
+     * The opposite operation of {@link #write}.
      */
     public static Capability read(InputStream is) throws IOException {
         try (ObjectInputStream ois = new ObjectInputStream(Mode.TEXT.wrap(is)) {
@@ -171,6 +186,25 @@ public void close() throws IOException {
         }
     }
 
+    /**
+     * Uses {@link #write} to serialize this object to a Base64-encoded ASCII stream.
+     */
+    public String toASCII() throws IOException {
+        try (ByteArrayOutputStream baos = new ByteArrayOutputStream()) {
+            write(baos);
+            return baos.toString("US-ASCII");
+        }
+    }
+
+    /**
+     * The inverse of {@link #toASCII}.
+     */
+    public static Capability fromASCII(String ascii) throws IOException {
+        try (ByteArrayInputStream bais = new ByteArrayInputStream(ascii.getBytes(StandardCharsets.US_ASCII))) {
+            return Capability.read(bais);
+        }
+    }
+
     private static final long serialVersionUID = 1L;
 
     /**

src/main/java/hudson/remoting/Engine.java

@@ -23,13 +23,16 @@
  */
 package hudson.remoting;
 
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import hudson.remoting.Channel.Mode;
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.net.Socket;
+import java.net.URI;
 import java.net.URL;
+import java.nio.ByteBuffer;
 import java.nio.file.Path;
 import java.security.AccessController;
 import java.security.KeyManagementException;
@@ -44,13 +47,15 @@
 import java.security.cert.X509Certificate;
 import java.security.interfaces.RSAPublicKey;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import javax.annotation.CheckForNull;
@@ -61,6 +66,13 @@
 import javax.net.ssl.SSLSocketFactory;
 import javax.net.ssl.TrustManager;
 import javax.net.ssl.TrustManagerFactory;
+import javax.websocket.ClientEndpointConfig;
+import javax.websocket.CloseReason;
+import javax.websocket.ContainerProvider;
+import javax.websocket.Endpoint;
+import javax.websocket.EndpointConfig;
+import javax.websocket.HandshakeResponse;
+import javax.websocket.Session;
 
 import org.jenkinsci.remoting.engine.JnlpEndpointResolver;
 import org.jenkinsci.remoting.engine.Jnlp4ConnectionState;
@@ -78,6 +90,7 @@
 import org.jenkinsci.remoting.protocol.cert.PublicKeyMatchingX509ExtendedTrustManager;
 import org.jenkinsci.remoting.protocol.impl.ConnectionRefusalException;
 import org.jenkinsci.remoting.util.KeyUtils;
+import org.jenkinsci.remoting.util.VersionNumber;
 
 /**
  * Agent engine that proactively connects to Jenkins master.
@@ -134,6 +147,7 @@ public Thread newThread(final Runnable r) {
     private URL hudsonUrl;
     private final String secretKey;
     public final String slaveName;
+    private boolean webSocket;
     private String credentials;
     private String proxyCredentials = System.getProperty("proxyCredentials");
 
@@ -322,6 +336,10 @@ public URL getHudsonUrl() {
         return hudsonUrl;
     }
 
+    public void setWebSocket(boolean webSocket) {
+        this.webSocket = webSocket;
+    }
+
     /**
      * If set, connect to the specified host and port instead of connecting directly to Jenkins.
      * @param tunnel Value. {@code null} to disable tunneling
@@ -439,6 +457,10 @@ public void removeListener(EngineListener el) {
 
     @Override
     public void run() {
+        if (webSocket) {
+            runWebSocket();
+            return;
+        }
         // Create the engine
         try {
             IOHub hub = IOHub.create(executor);
@@ -494,6 +516,116 @@ public void run() {
         }
     }
 
+    @SuppressFBWarnings(value = "REC_CATCH_EXCEPTION", justification = "checked exceptions were a mistake to begin with")
+    private void runWebSocket() {
+        try {
+            AtomicReference<Channel> ch = new AtomicReference<>();
+            String localCap = new Capability().toASCII();
+            class HeaderHandler extends ClientEndpointConfig.Configurator {
+                Capability remoteCapability = new Capability();
+                @Override
+                public void beforeRequest(Map<String, List<String>> headers) {
+                    headers.put(JnlpConnectionState.CLIENT_NAME_KEY, Collections.singletonList(slaveName));
+                    headers.put(JnlpConnectionState.SECRET_KEY, Collections.singletonList(secretKey));
+                    headers.put(Capability.KEY, Collections.singletonList(localCap));
+                    // TODO use JnlpConnectionState.COOKIE_KEY somehow (see EngineJnlpConnectionStateListener.afterChannel)
+                    LOGGER.fine(() -> "Sending: " + headers);
+                }
+                @Override
+                public void afterResponse(HandshakeResponse hr) {
+                    LOGGER.fine(() -> "Receiving: " + hr.getHeaders());
+                    List<String> remotingMinimumVersion = hr.getHeaders().get("X-Remoting-Minimum-Version");
+                    if (remotingMinimumVersion != null && !remotingMinimumVersion.isEmpty()) {
+                        VersionNumber minimumSupportedVersion = new VersionNumber(remotingMinimumVersion.get(0));
+                        VersionNumber currentVersion = new VersionNumber(Launcher.VERSION);
+                        if (currentVersion.isOlderThan(minimumSupportedVersion)) {
+                            // TODO these errors should trigger a connection close
+                            events.error(new IOException("Agent version " + minimumSupportedVersion + " or newer is required."));
+                        }
+                    }
+                    try {
+                        remoteCapability = Capability.fromASCII(hr.getHeaders().get(Capability.KEY).get(0));
+                        LOGGER.fine(() -> "received " + remoteCapability);
+                    } catch (IOException x) {
+                        events.error(x);
+                    }
+                }
+            }
+            HeaderHandler headerHandler = new HeaderHandler();
+            class AgentEndpoint extends Endpoint {
+                @SuppressFBWarnings(value = "UWF_FIELD_NOT_INITIALIZED_IN_CONSTRUCTOR", justification = "just trust me here")
+                AbstractByteArrayCommandTransport.ByteArrayReceiver receiver;
+                @Override
+                public void onOpen(Session session, EndpointConfig config) {
+                    events.status("WebSocket connection open");
+                    session.addMessageHandler(byte[].class, this::onMessage);
+                    try {
+                        ch.set(new ChannelBuilder(slaveName, executor).
+                            withJarCacheOrDefault(jarCache). // unless EngineJnlpConnectionStateListener can be used for this purpose
+                            build(new Transport(session)));
+                    } catch (IOException x) {
+                        events.error(x);
+                    }
+                }
+                private void onMessage(byte[] message) {
+                    LOGGER.finest(() -> "received message of length " + message.length);
+                    receiver.handle(message);
+                }
+                @Override
+                public void onClose(Session session, CloseReason closeReason) {
+                    LOGGER.fine(() -> "onClose: " + closeReason);
+                    receiver.terminate(new ChannelClosedException(ch.get(), null));
+                }
+                @Override
+                public void onError(Session session, Throwable x) {
+                    LOGGER.log(Level.FINE, null, x);
+                    receiver.terminate(new ChannelClosedException(ch.get(), x));
+                }
+                class Transport extends AbstractByteArrayCommandTransport {
+                    final Session session;
+                    Transport(Session session) {
+                        this.session = session;
+                    }
+                    @Override
+                    public void setup(AbstractByteArrayCommandTransport.ByteArrayReceiver _receiver) {
+                        events.status("Setting up channel");
+                        receiver = _receiver;
+                    }
+                    @Override
+                    public void writeBlock(Channel channel, byte[] payload) throws IOException {
+                        LOGGER.finest(() -> "sending message of length " + payload.length);
+                        session.getBasicRemote().sendBinary(ByteBuffer.wrap(payload));
+                    }
+                    @Override
+                    public Capability getRemoteCapability() throws IOException {
+                        return headerHandler.remoteCapability;
+                    }
+                    @Override
+                    public void closeWrite() throws IOException {
+                        events.status("Write side closed");
+                        session.close();
+                    }
+                    @Override
+                    public void closeRead() throws IOException {
+                        events.status("Read side closed");
+                        session.close();
+                    }
+                }
+            }
+            ContainerProvider.getWebSocketContainer().connectToServer(new AgentEndpoint(),
+                ClientEndpointConfig.Builder.create().configurator(headerHandler).build(), URI.create(candidateUrls.get(0).toString().replaceFirst("^http", "ws") + "wsagents/"));
+            while (ch.get() == null) {
+                Thread.sleep(100);
+            }
+            LOGGER.info(() -> "Waiting for channel");
+            ch.get().join();
+            // TODO handle multiple candidate URLs
+            // TODO handle reconnection
+        } catch (Exception e) {
+            events.error(e);
+        }
+    }
+
     @SuppressWarnings({"ThrowableInstanceNeverThrown"})
     private void innerRun(IOHub hub, SSLContext context, ExecutorService service) {
         // Create the protocols that will be attempted to connect to the master.

src/main/java/hudson/remoting/jnlp/Main.java

@@ -83,6 +83,10 @@ public class Main {
             usage="Specify the Jenkins root URLs to connect to.")
     public List<URL> urls = new ArrayList<>();
 
+    @Option(name="-webSocket",
+            usage="Make a WebSocket connection to Jenkins rather than using the TCP port.")
+    public boolean webSocket;
+
     @Option(name="-credentials",metaVar="USER:PASSWORD",
             usage="HTTP BASIC AUTH header to pass in for making HTTP requests.")
     public String credentials;
@@ -267,6 +271,38 @@ public static void _main(String[] args) throws IOException, InterruptedException
         if(m.urls.isEmpty() && m.directConnection == null) {
             throw new CmdLineException(p, "At least one -url option is required.", null);
         }
+        if (m.webSocket) {
+            if (m.urls.isEmpty()) {
+                throw new CmdLineException(p, "-url is required in -webSocket mode", null);
+            }
+            if (m.urls.size() > 1) {
+                throw new CmdLineException(p, "multiple -url is not currently supported in -webSocket mode", null);
+            }
+            if (m.directConnection != null) {
+                throw new CmdLineException(p, "-webSocket and -direct are mutually exclusive", null);
+            }
+            if (m.tunnel != null) {
+                throw new CmdLineException(p, "-tunnel is not currently supported in -webSocket mode", null);
+            }
+            if (m.credentials != null) {
+                throw new CmdLineException(p, "-credentials is not currently supported in -webSocket mode", null);
+            }
+            if (m.proxyCredentials != null) {
+                throw new CmdLineException(p, "-proxyCredentials is not currently supported in -webSocket mode", null);
+            }
+            if (m.candidateCertificates != null) {
+                throw new CmdLineException(p, "-candidateCertificates is not currently supported in -webSocket mode", null);
+            }
+            if (m.disableHttpsCertValidation) {
+                throw new CmdLineException(p, "-disableHttpsCertValidation is not currently supported in -webSocket mode", null);
+            }
+            if (m.noKeepAlive) {
+                throw new CmdLineException(p, "-noKeepAlive is not currently supported in -webSocket mode", null);
+            }
+            if (m.noReconnect) {
+                throw new CmdLineException(p, "-noReconnect is not currently supported in -webSocket mode", null);
+            }
+        }
         m.main();
     }
 
@@ -290,6 +326,7 @@ public Engine createEngine() {
         Engine engine = new Engine(
                 headlessMode ? new CuiListener() : new GuiListener(),
                 urls, args.get(0), agentName, directConnection, instanceIdentity, new HashSet<>(protocols));
+        engine.setWebSocket(webSocket);
         if(tunnel!=null)
             engine.setTunnel(tunnel);
         if(credentials!=null)