Only use v4 streaming protocol and parse exec status channel response as V1Status

Author: rjeberhard

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

@@ -12,13 +12,19 @@
 */
 package io.kubernetes.client;
 
+import com.google.gson.reflect.TypeToken;
 import io.kubernetes.client.models.V1Pod;
+import io.kubernetes.client.models.V1Status;
+import io.kubernetes.client.models.V1StatusCause;
+import io.kubernetes.client.models.V1StatusDetails;
 import io.kubernetes.client.util.WebSocketStreamHandler;
 import io.kubernetes.client.util.WebSockets;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.lang.reflect.Type;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import org.apache.commons.lang3.StringUtils;
@@ -184,7 +190,7 @@ public Process exec(
     return exec;
   }
 
-  static int parseExitCode(InputStream inputStream) {
+  static int parseExitCode(ApiClient client, InputStream inputStream) {
     try {
       int available = inputStream.available();
 
@@ -193,24 +199,38 @@ static int parseExitCode(InputStream inputStream) {
 
       byte[] b = new byte[available];
       inputStream.read(b);
-      String result = new String(b, "UTF-8");
-      int idx = result.lastIndexOf(':');
-      if (idx > 0) {
-        try {
-          return Integer.parseInt(result.substring(idx + 1).trim());
-        } catch (NumberFormatException nfe) {
-          log.error("Error parsing exit code from status channel response", nfe);
+      String body = new String(b, "UTF-8");
+
+      Type returnType = new TypeToken<V1Status>() {}.getType();
+      V1Status status = client.getJSON().deserialize(body, returnType);
+      if ("Success".equals(status.getStatus())) return 0;
+
+      if ("NonZeroExitCode".equals(status.getReason())) {
+        V1StatusDetails details = status.getDetails();
+        if (details != null) {
+          List<V1StatusCause> causes = details.getCauses();
+          if (causes != null) {
+            for (V1StatusCause cause : causes) {
+              if ("ExitCode".equals(cause.getReason())) {
+                try {
+                  return Integer.parseInt(cause.getMessage());
+                } catch (NumberFormatException nfe) {
+                  log.error("Error parsing exit code from status channel response", nfe);
+                }
+              }
+            }
+          }
         }
       }
-    } catch (IOException io) {
-      log.error("Error parsing exit code from status channel response", io);
+    } catch (Throwable t) {
+      log.error("Error parsing exit code from status channel response", t);
     }
 
     // Unable to parse the exit code from the content
     return -1;
   }
 
-  private static class ExecProcess extends Process {
+  private class ExecProcess extends Process {
     private final WebSocketStreamHandler streamHandler;
     private volatile int statusCode;
     private final Map<Integer, InputStream> input = new HashMap<>();
@@ -222,7 +242,7 @@ public ExecProcess() throws IOException {
             @Override
             public void close() {
               if (statusCode == -1) {
-                int exitCode = parseExitCode(ExecProcess.this.getInputStream(3));
+                int exitCode = parseExitCode(apiClient, ExecProcess.this.getInputStream(3));
 
                 // notify of process completion
                 synchronized (ExecProcess.this) {

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

@@ -38,11 +38,9 @@
 public class WebSockets {
   private static final Logger log = LoggerFactory.getLogger(WebSockets.class);
 
+  // Only support v4 stream protocol as it was available since k8s 1.4
   public static final String V4_STREAM_PROTOCOL = "v4.channel.k8s.io";
-  public static final String V3_STREAM_PROTOCOL = "v3.channel.k8s.io";
-  public static final String V2_STREAM_PROTOCOL = "v2.channel.k8s.io";
-  public static final String V1_STREAM_PROTOCOL = "channel.k8s.io";
-  public static final String STREAM_PROTOCOL_HEADER = "X-Stream-Protocol-Version";
+  public static final String STREAM_PROTOCOL_HEADER = "Sec-WebSocket-Protocol";
   public static final String SPDY_3_1 = "SPDY/3.1";
 
   /** A simple interface for a listener on a web socket */
@@ -86,11 +84,7 @@ public static void stream(
       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(STREAM_PROTOCOL_HEADER, V4_STREAM_PROTOCOL);
     headers.put(HttpHeaders.CONNECTION, HttpHeaders.UPGRADE);
     headers.put(HttpHeaders.UPGRADE, SPDY_3_1);
     String[] localVarAuthNames = new String[] {"BearerToken"};

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

@@ -14,57 +14,68 @@
 
 import static org.junit.Assert.*;
 
+import io.kubernetes.client.util.ClientBuilder;
 import java.io.ByteArrayInputStream;
+import java.io.IOException;
 import java.io.InputStream;
 import java.nio.charset.StandardCharsets;
+import org.junit.Before;
 import org.junit.Test;
 
 /** Tests for the Exec helper class */
 public class ExecTest {
+
+  private static final String OUTPUT_EXIT0 = "{\"metadata\":{},\"status\":\"Success\"}";
   private static final String OUTPUT_EXIT1 =
-      "command terminated with non-zero exit code: Error executing in Docker Container: 1";
+      "{\"metadata\":{},\"status\":\"Failure\",\"message\":\"command terminated with non-zero exit code: Error executing in Docker Container: 1\",\"reason\":\"NonZeroExitCode\",\"details\":{\"causes\":[{\"reason\":\"ExitCode\",\"message\":\"1\"}]}}";
   private static final String OUTPUT_EXIT126 =
-      "command terminated with non-zero exit code: Error executing in Docker Container: 126";
+      "{\"metadata\":{},\"status\":\"Failure\",\"message\":\"command terminated with non-zero exit code: Error executing in Docker Container: 126\",\"reason\":\"NonZeroExitCode\",\"details\":{\"causes\":[{\"reason\":\"ExitCode\",\"message\":\"126\"}]}}";
   private static final String BAD_OUTPUT_INCOMPLETE_MSG1 =
-      "command terminated with non-zero exit code: Error ";
-  private static final String BAD_OUTPUT_INCOMPLETE_MSG2 = "command terminated with non-zero";
+      "{\"metadata\":{},\"status\":\"Failure\",\"message\":\"command terminated with non-zero exit code: Error executing in Docker Container: 1\",\"reas";
+
+  private ApiClient client;
+
+  @Before
+  public void setup() throws IOException {
+    client = ClientBuilder.defaultClient();
+  }
 
   @Test
-  public void testExit0() {
+  public void testExit0NoMessage() {
     InputStream inputStream = new ByteArrayInputStream(new byte[0]);
-    int exitCode = Exec.parseExitCode(inputStream);
+    int exitCode = Exec.parseExitCode(client, inputStream);
+    assertEquals(0, exitCode);
+  }
+
+  @Test
+  public void testExit0() {
+    InputStream inputStream =
+        new ByteArrayInputStream(OUTPUT_EXIT0.getBytes(StandardCharsets.UTF_8));
+    int exitCode = Exec.parseExitCode(client, inputStream);
     assertEquals(0, exitCode);
   }
 
   @Test
   public void testExit1() {
     InputStream inputStream =
         new ByteArrayInputStream(OUTPUT_EXIT1.getBytes(StandardCharsets.UTF_8));
-    int exitCode = Exec.parseExitCode(inputStream);
+    int exitCode = Exec.parseExitCode(client, inputStream);
     assertEquals(1, exitCode);
   }
 
   @Test
   public void testExit126() {
     InputStream inputStream =
         new ByteArrayInputStream(OUTPUT_EXIT126.getBytes(StandardCharsets.UTF_8));
-    int exitCode = Exec.parseExitCode(inputStream);
+    int exitCode = Exec.parseExitCode(client, inputStream);
     assertEquals(126, exitCode);
   }
 
   @Test
   public void testIncompleteData1() {
     InputStream inputStream =
         new ByteArrayInputStream(BAD_OUTPUT_INCOMPLETE_MSG1.getBytes(StandardCharsets.UTF_8));
-    int exitCode = Exec.parseExitCode(inputStream);
-    assertEquals(-1, exitCode);
-  }
-
-  @Test
-  public void testIncompleteData2() {
-    InputStream inputStream =
-        new ByteArrayInputStream(BAD_OUTPUT_INCOMPLETE_MSG2.getBytes(StandardCharsets.UTF_8));
-    int exitCode = Exec.parseExitCode(inputStream);
+    int exitCode = Exec.parseExitCode(client, inputStream);
     assertEquals(-1, exitCode);
   }
 }