Merge pull request #1144 from brendandburns/portforward

Add port forward to kubectl

Author: k8s-ci-robot

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

@@ -15,12 +15,14 @@
 import static io.kubernetes.client.extended.kubectl.Kubectl.exec;
 import static io.kubernetes.client.extended.kubectl.Kubectl.label;
 import static io.kubernetes.client.extended.kubectl.Kubectl.log;
+import static io.kubernetes.client.extended.kubectl.Kubectl.portforward;
 import static io.kubernetes.client.extended.kubectl.Kubectl.scale;
 import static io.kubernetes.client.extended.kubectl.Kubectl.version;
 
 import com.google.common.io.ByteStreams;
 import io.kubernetes.client.common.KubernetesObject;
 import io.kubernetes.client.extended.kubectl.KubectlExec;
+import io.kubernetes.client.extended.kubectl.KubectlPortForward;
 import io.kubernetes.client.extended.kubectl.exception.KubectlException;
 import io.kubernetes.client.openapi.ApiClient;
 import io.kubernetes.client.openapi.models.V1Deployment;
@@ -84,10 +86,21 @@ public static void main(String[] args)
     String name = null;
 
     switch (verb) {
+      case "portforward":
+        name = args[1];
+        KubectlPortForward forward = portforward(client).name(name).namespace(ns);
+        for (int i = 2; i < args.length; i++) {
+          String port = args[i];
+          String[] ports = port.split(":");
+          System.out.println("Forwarding " + ns + "/" + name + " " + ports[0] + "->" + ports[1]);
+          forward.ports(Integer.parseInt(ports[0]), Integer.parseInt(ports[1]));
+        }
+        forward.execute();
+        System.exit(0);
       case "log":
         name = args[1];
         ByteStreams.copy(
-            log().name(name).namespace(ns).container(cli.getOptionValue("c", "")).execute(),
+            log(client).name(name).namespace(ns).container(cli.getOptionValue("c", "")).execute(),
             System.out);
         System.exit(0);
       case "scale":

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

@@ -41,12 +41,13 @@ public static void main(String[] args) throws IOException, ApiException, Interru
 
     PortForward forward = new PortForward();
     List<Integer> ports = new ArrayList<>();
-    ports.add(8080);
-    ports.add(80);
+    int localPort = 8080;
+    int targetPort = 8080;
+    ports.add(targetPort);
     final PortForward.PortForwardResult result =
-        forward.forward("default", "nginx-d5dc44cf7-x7475", ports);
-
-    ServerSocket ss = new ServerSocket(8080);
+        forward.forward("default", "camera-viz-7949dbf7c6-lpxkd", ports);
+    System.out.println("Forwarding!");
+    ServerSocket ss = new ServerSocket(localPort);
 
     final Socket s = ss.accept();
     System.out.println("Connected!");
@@ -55,7 +56,7 @@ public static void main(String[] args) throws IOException, ApiException, Interru
             new Runnable() {
               public void run() {
                 try {
-                  ByteStreams.copy(result.getInputStream(80), s.getOutputStream());
+                  ByteStreams.copy(result.getInputStream(targetPort), s.getOutputStream());
                 } catch (IOException ex) {
                   ex.printStackTrace();
                 } catch (Exception ex) {
@@ -69,7 +70,7 @@ public void run() {
             new Runnable() {
               public void run() {
                 try {
-                  ByteStreams.copy(s.getInputStream(), result.getOutboundStream(80));
+                  ByteStreams.copy(s.getInputStream(), result.getOutboundStream(targetPort));
                 } catch (IOException ex) {
                   ex.printStackTrace();
                 } catch (Exception ex) {

extended/src/main/java/io/kubernetes/client/extended/kubectl/Kubectl.java

@@ -155,6 +155,14 @@ public static KubectlLog log() {
     return log(Configuration.getDefaultApiClient());
   }
 
+  public static KubectlPortForward portforward() {
+    return portforward(Configuration.getDefaultApiClient());
+  }
+
+  public static KubectlPortForward portforward(ApiClient apiClient) {
+    return new KubectlPortForward(apiClient);
+  }
+
   /**
    * Executable executes a kubectl helper.
    *

extended/src/main/java/io/kubernetes/client/extended/kubectl/KubectlExec.java

@@ -66,8 +66,9 @@ public Integer execute() throws KubectlException {
     }
   }
 
-  private static void copyAsync(InputStream in, OutputStream out) {
-    new Thread(
+  protected static Thread copyAsync(InputStream in, OutputStream out) {
+    Thread t =
+        new Thread(
             new Runnable() {
               public void run() {
                 try {
@@ -76,7 +77,8 @@ public void run() {
                   ex.printStackTrace();
                 }
               }
-            })
-        .start();
+            });
+    t.start();
+    return t;
   }
 }

extended/src/main/java/io/kubernetes/client/extended/kubectl/KubectlPortForward.java

@@ -0,0 +1,116 @@
+/*
+Copyright 2020 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.extended.kubectl;
+
+import static io.kubernetes.client.extended.kubectl.KubectlExec.copyAsync;
+
+import io.kubernetes.client.PortForward;
+import io.kubernetes.client.extended.kubectl.exception.KubectlException;
+import io.kubernetes.client.openapi.ApiClient;
+import io.kubernetes.client.openapi.ApiException;
+import io.kubernetes.client.openapi.models.V1Pod;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.util.ArrayList;
+import java.util.List;
+
+public class KubectlPortForward
+    extends Kubectl.ResourceAndContainerBuilder<V1Pod, KubectlPortForward>
+    implements Kubectl.Executable<Boolean> {
+  List<Integer> localPorts;
+  List<Integer> targetPorts;
+  boolean running;
+
+  KubectlPortForward(ApiClient client) {
+    super(client, V1Pod.class);
+
+    localPorts = new ArrayList<>();
+    targetPorts = new ArrayList<>();
+  }
+
+  /**
+   * Add a port pair
+   *
+   * @param localPort The port to listen to on the local machine
+   * @param targetPort The port to connect to on the target Pod
+   */
+  public KubectlPortForward ports(int localPort, int targetPort) {
+    localPorts.add(localPort);
+    targetPorts.add(targetPort);
+    return this;
+  }
+
+  @Override
+  public Boolean execute() throws KubectlException {
+    running = true;
+    try {
+      executeInternal();
+      return true;
+    } catch (ApiException | IOException | InterruptedException ex) {
+      throw new KubectlException(ex);
+    }
+  }
+
+  public void shutdown() {
+    running = false;
+  }
+
+  private void executeInternal()
+      throws ApiException, KubectlException, IOException, InterruptedException {
+    PortForward pf = new PortForward(apiClient);
+    PortForward.PortForwardResult result = pf.forward(namespace, name, targetPorts);
+    if (result == null) {
+      throw new KubectlException("PortForward failed!");
+    }
+    // TODO: Convert this to NIO to reduce the number of threads?
+    List<Thread> threads = new ArrayList<>();
+    for (int i = 0; i < localPorts.size(); i++) {
+      int targetPort = targetPorts.get(i);
+      threads.add(
+          portForward(
+              new ServerSocket(localPorts.get(i)),
+              result.getInputStream(targetPort),
+              result.getOutboundStream(targetPort)));
+    }
+    for (Thread t : threads) {
+      t.join();
+    }
+  }
+
+  private Thread portForward(ServerSocket server, InputStream in, OutputStream out) {
+    Thread t =
+        new Thread(
+            new Runnable() {
+              @Override
+              public void run() {
+                while (running) {
+                  try {
+                    Socket sock = server.accept();
+                    Thread t1 = copyAsync(sock.getInputStream(), out);
+                    Thread t2 = copyAsync(in, sock.getOutputStream());
+
+                    t1.join();
+                    t2.join();
+                  } catch (InterruptedException | IOException ex) {
+                    ex.printStackTrace();
+                  }
+                }
+              }
+            });
+    t.start();
+    return t;
+  }
+}

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

@@ -114,7 +114,15 @@ public PortForwardResult forward(String namespace, String name, List<Integer> po
       queryParams.add(new Pair("ports", port.toString()));
     }
     WebSockets.stream(path, "GET", queryParams, apiClient, handler);
-
+    try {
+      handler.waitForInitialized();
+    } catch (InterruptedException ex) {
+      throw new ApiException(ex);
+    }
+    Throwable err = handler.getError();
+    if (err != null) {
+      throw new ApiException(err);
+    }
     // Wait for streams to start.
     result.init();
 

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

@@ -55,6 +55,13 @@ private static enum State {
     CLOSED
   };
 
+  public synchronized void waitForInitialized() throws InterruptedException {
+    if (state != State.UNINITIALIZED) {
+      return;
+    }
+    this.wait();
+  }
+
   @Override
   public synchronized void open(String protocol, WebSocket socket) {
     if (state != State.UNINITIALIZED) throw new IllegalStateException();
@@ -137,6 +144,7 @@ public synchronized void close() {
         }
       }
     }
+    notifyAll();
   }
 
   /**

util/src/test/java/io/kubernetes/client/PortForwardTest.java

@@ -22,6 +22,7 @@
 import static io.kubernetes.client.ExecTest.makeStream;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThrows;
 
 import com.github.tomakehurst.wiremock.junit.WireMockRule;
 import io.kubernetes.client.PortForward.PortForwardResult;
@@ -77,7 +78,11 @@ public void testUrl() throws IOException, ApiException, InterruptedException {
     int portNumber = 8080;
     List<Integer> ports = new ArrayList<>();
     ports.add(portNumber);
-    forward.forward(pod, ports);
+    assertThrows(
+        ApiException.class,
+        () -> {
+          forward.forward(pod, ports);
+        });
 
     // TODO: Kill this sleep, the trouble is that the test tries to validate before the
     // connection