Add support for PortForward.

examples/src/main/java/io/kubernetes/client/examples/PortForwardExample.java

@@ -0,0 +1,92 @@
+/*
+Copyright 2017 The Kubernetes Authors.
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+package io.kubernetes.client.examples;
+
+import io.kubernetes.client.ApiClient;
+import io.kubernetes.client.ApiException;
+import io.kubernetes.client.Configuration;
+import io.kubernetes.client.PortForward;
+import io.kubernetes.client.apis.CoreV1Api;
+import io.kubernetes.client.models.V1Pod;
+import io.kubernetes.client.models.V1PodList;
+import io.kubernetes.client.util.Config;
+
+import com.google.common.io.ByteStreams;
+
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.util.ArrayList;
+import java.util.List;
+
+
+/**
+ * A simple example of how to use the Java API
+ * 
+ * Easiest way to run this:
+ *   mvn exec:java -Dexec.mainClass="io.kubernetes.client.examples.PortForwardExample"
+ * from inside $REPO_DIR/examples
+ * 
+ * Then:
+ *   curl localhost:8080
+ * from a different terminal (but be quick about it, the socket times out pretty fast...)
+ * 
+ */
+public class PortForwardExample {
+    public static void main(String[] args) throws IOException, ApiException, InterruptedException {
+        ApiClient client = Config.defaultClient();
+        Configuration.setDefaultApiClient(client);
+
+        PortForward forward = new PortForward();
+        List<Integer> ports = new ArrayList<>();
+        ports.add(80);
+        final PortForward.PortForwardResult result =
+            forward.forward("default", "nginx-4217019353-fg6zx", ports);
+
+        ServerSocket ss = new ServerSocket(8080);
+
+        final Socket s = ss.accept();
+        System.out.println("Connected!");
+
+        new Thread(new Runnable() {
+            public void run() {
+                try {
+                    ByteStreams.copy(result.getInputStream(80), s.getOutputStream());
+                } catch (IOException ex) {
+                    ex.printStackTrace();
+                } catch (Exception ex) {
+                    ex.printStackTrace();
+                }
+            }
+        }).start();
+
+        new Thread(new Runnable() {
+            public void run() {
+                try {
+                    ByteStreams.copy(s.getInputStream(), result.getOutboundStream(80));
+                } catch (IOException ex) {
+                    ex.printStackTrace();
+                } catch (Exception ex) {
+                    ex.printStackTrace();
+                }
+            }
+        }).start();
+
+        Thread.sleep(10 * 1000);
+
+        System.exit(0);
+    }
+}

util/src/main/java/io/kubernetes/client/PortForward.java

@@ -0,0 +1,193 @@
+package io.kubernetes.client;
+
+import io.kubernetes.client.Configuration;
+import io.kubernetes.client.models.V1Pod;
+import io.kubernetes.client.util.WebSockets;
+import io.kubernetes.client.util.WebSocketStreamHandler;
+
+import java.io.InputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+
+/**
+ * Utility class for setting up port-forwarding connections.
+ * Uses the WebSockets API, not the SPDY API (which the Go client uses)
+ * 
+ * The protocol is undocumented as far as I can tell, but the PR that added
+ * it is here:
+ * https://github.com/kubernetes/kubernetes/pull/33684
+ * 
+ * And the protocol is:
+ * 
+ * ws://server/api/v1/namespaces/<namespace>/pods/<pod>/portforward?ports=80&ports=8080
+ * 
+ * I/O for first port (80) is on Channel 0
+ * Err for first port (80) is on Channel 1
+ * I/O for second port (8080) is on Channel 2
+ * Err for second port (8080) is on Channel 3
+ * <and so on for remaining ports>
+ * 
+ * The first two bytes of each output stream is the port that is being forwarded
+ * in little-endian format.
+ */
+public class PortForward {
+    private ApiClient apiClient;
+
+    /**
+     * Simple PortForward API constructor, uses default configuration
+     */
+    public PortForward() {
+        this(Configuration.getDefaultApiClient());
+    }
+
+    /**
+     * PortForward API Constructor
+     * @param apiClient The api client to use.
+     */
+    public PortForward(ApiClient apiClient) {
+        this.apiClient = apiClient;
+    }
+
+    /**
+     * Get the API client for these PortForward operations.
+     * @return The API client that will be used.
+     */
+    public ApiClient getApiClient() {
+        return apiClient;
+    }
+
+    /**
+     * Set the API client for subsequent PortForward operations.
+     * @param apiClient The new API client to use.
+     */
+    public void setApiClient(ApiClient apiClient) {
+        this.apiClient = apiClient;
+    }
+
+    private String makePath(String namespace, String name) {
+          return "/api/v1/namespaces/" +
+            namespace +
+            "/pods/" +
+            name +
+            "/portforward";
+    }
+
+    /**
+     * PortForward to a container
+     * 
+     * @param pod The pod where the port forward is run.
+     * @param ports The ports to forward
+     * @return The result of the Port Forward request.
+     */
+    public PortForwardResult forward(V1Pod pod, List<Integer> ports) throws ApiException, IOException {
+        return forward(pod.getMetadata().getNamespace(), pod.getMetadata().getNamespace(), ports);
+    }
+
+    /**
+     * PortForward to a container.
+     * 
+     * @param namespace The namespace of the Pod
+     * @param name The name of the Pod
+     * @param ports The ports to forward
+     * @return The result of the Port Forward request.
+     */
+    public PortForwardResult forward(String namespace, String name, List<Integer> ports) throws ApiException, IOException {
+        String path = makePath(namespace, name);
+        WebSocketStreamHandler handler = new WebSocketStreamHandler();
+        PortForwardResult result = new PortForwardResult(handler, ports);
+        List<Pair> queryParams = new ArrayList<>();
+        for (Integer port : ports) {
+            queryParams.add(new Pair("ports", port.toString()));
+        }
+        WebSockets.stream(path, "GET", queryParams, apiClient, handler);
+
+        // Wait for streams to start.
+        result.init();
+
+        return result;
+    }
+
+    /**
+     * PortForwardResult contains the result of an Attach call, it includes streams for stdout
+     * stderr and stdin. 
+     */
+    public static class PortForwardResult { 
+        private WebSocketStreamHandler handler;
+        private HashMap<Integer, Integer> streams;
+        private List<Integer> ports;
+
+        /**
+         * Constructor
+         * @param handler The web socket handler
+         * @param ports The list of ports that are being forwarded.
+         */
+        public PortForwardResult(WebSocketStreamHandler handler, List<Integer> ports) throws IOException {
+            this.handler = handler;
+            this.streams = new HashMap<>();
+            this.ports = ports;
+        }
+
+        /**
+         * Initialize the connection. Must be called after the web socket has been opened.
+         */
+        public void init() throws IOException {
+            for (int i = 0; i < ports.size(); i++) {
+                InputStream is = handler.getInputStream(i);
+                byte[] data = new byte[2];
+                is.read(data);
+                int port = data[0] + data[1] * 256;
+                streams.put(port, i);
+            }
+        }
+
+        private int findPortIndex(int portNumber) {
+            Integer ix = streams.get(portNumber);
+            if (ix == null) {
+                return -1;
+            }
+            return ix.intValue();
+        }
+
+        /**
+         * Get the output stream for the specified port number (e.g. 80)
+         * @param port The port number to get the stream for.
+         * @return The OutputStream for the specified port, null if there is no such port.
+         */
+        public OutputStream getOutboundStream(int port) {
+            int portIndex = findPortIndex(port);
+            if (portIndex == -1) {
+                return null;
+            }
+            return handler.getOutputStream(portIndex * 2);
+        }
+
+        /**
+         * Get the error stream for a port number (e.g. 80)
+         * @param port The port number to get the stream for.
+         * @return The error stream, or null if there is no such port.
+         */
+        public OutputStream getErrorStream(int port) {
+            int portIndex = findPortIndex(port);
+            if (portIndex == -1) {
+                return null;
+            }
+            return handler.getOutputStream(portIndex * 2 + 1);
+        }
+
+        /**
+         * Get the input stream for a port number (e.g. 80)
+         * @param port The port number to get the stream for.
+         * @return The input stream, or null if no such port exists.
+         */
+        public InputStream getInputStream(int port) throws IOException {
+            int portIndex = findPortIndex(port);
+            if (portIndex == -1) {
+                return null;
+            }
+            return handler.getInputStream(portIndex * 2);
+        }
+    }
+}

util/src/main/java/io/kubernetes/client/util/WebSockets.java

@@ -13,6 +13,7 @@
 package io.kubernetes.client.util;
 
 import com.google.common.net.HttpHeaders;
+import com.squareup.okhttp.Call;
 import com.squareup.okhttp.Request;
 import com.squareup.okhttp.Response;
 import com.squareup.okhttp.ResponseBody;
@@ -30,6 +31,7 @@
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.io.Reader;
+import java.util.List;
 import java.util.ArrayList;
 import java.util.HashMap;
 
@@ -81,13 +83,29 @@ public interface SocketListener {
      * @param listener The socket listener to handle socket events
      */
     public static void stream(String path, String method, ApiClient client, SocketListener listener) throws ApiException, IOException {
+        stream(path, method, new ArrayList<Pair>(), client, listener);
+    }
+
+    public static void stream(String path, String method, List<Pair> queryParams, ApiClient client, SocketListener listener) throws ApiException, IOException {
+
         HashMap<String, String> headers = new HashMap<String, String>();
         String allProtocols = String.format("%s,%s,%s,%s", V4_STREAM_PROTOCOL, V3_STREAM_PROTOCOL, V2_STREAM_PROTOCOL, V1_STREAM_PROTOCOL);
         headers.put(STREAM_PROTOCOL_HEADER, allProtocols);
         headers.put(HttpHeaders.CONNECTION, HttpHeaders.UPGRADE);
         headers.put(HttpHeaders.UPGRADE, SPDY_3_1);
 
-        Request request = client.buildRequest(path, method, new ArrayList<Pair>(), new ArrayList<Pair>(), null, headers, new HashMap<String, Object>(), new String[0], null);
+        Request request = client.buildRequest(path, method, queryParams, new ArrayList<Pair>(), null, headers, new HashMap<String, Object>(), new String[0], null);
+        streamRequest(request, client, listener);
+    }
+
+    /* 
+    If we ever upgrade to okhttp 3...
+    public static void stream(Call call, ApiClient client, SocketListener listener) {
+        streamRequest(call.request(), client, listener);
+    }
+    */
+
+    private static void streamRequest(Request request, ApiClient client, SocketListener listener) {
         WebSocketCall.create(client.getHttpClient(), request).enqueue(new Listener(listener));
     }