Support for v4 stream protocol

- Set Sec-WebSocket-Protocol header from cfg param
- Add ERROR and RESIZE websocket channels
  As defined in pkg/kubelet/server/remotecommand/websocket.go
- Adjust tests to show v4 behavior wtr to status message
- Tweak read_all() to return only data from stdout and stderr.

Author: jraby

kubernetes/client/ws_client.py

@@ -24,6 +24,8 @@
 STDIN_CHANNEL = 0
 STDOUT_CHANNEL = 1
 STDERR_CHANNEL = 2
+ERROR_CHANNEL = 3
+RESIZE_CHANNEL = 4
 
 
 class WSClient:
@@ -46,6 +48,10 @@ def __init__(self, configuration, url, headers):
         if headers and 'authorization' in headers:
             header.append("authorization: %s" % headers['authorization'])
 
+        if configuration.ws_streaming_protocol:
+            header.append("Sec-WebSocket-Protocol: %s" %
+                          configuration.ws_streaming_protocol)
+
         if url.startswith('wss://') and configuration.verify_ssl:
             ssl_opts = {
                 'cert_reqs': ssl.CERT_REQUIRED,
@@ -131,10 +137,10 @@ def readline_stderr(self, timeout=None):
         return self.readline_channel(STDERR_CHANNEL, timeout=timeout)
 
     def read_all(self):
-        """Read all of the inputs with the same order they recieved. The channel
-        information would be part of the string. This is useful for
-        non-interactive call where a set of command passed to the API call and
-        their result is needed after the call is concluded.
+        """Return buffered data received on stdout and stderr channels.
+        This is useful for non-interactive call where a set of command passed
+        to the API call and their result is needed after the call is concluded.
+        Should be called after run_forever() or update()
 
         TODO: Maybe we can process this and return a more meaningful map with
         channels mapped for each input.
@@ -174,9 +180,10 @@ def update(self, timeout=0):
                     channel = ord(data[0])
                     data = data[1:]
                     if data:
-                        # keeping all messages in the order they received for
-                        # non-blocking call.
-                        self._all += data
+                        if channel in [STDOUT_CHANNEL, STDERR_CHANNEL]:
+                            # keeping all messages in the order they received for
+                            # non-blocking call.
+                            self._all += data
                         if channel not in self._channels:
                             self._channels[channel] = data
                         else:

kubernetes/e2e_test/test_client.py

@@ -12,6 +12,7 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
+import json
 import time
 import unittest
 import uuid
@@ -103,6 +104,10 @@ def test_pod_apis(self):
         self.assertEqual("test string 2", line)
         resp.write_stdin("exit\n")
         resp.update(timeout=5)
+        line = resp.read_channel(api_client.ws_client.ERROR_CHANNEL)
+        status = json.loads(line)
+        self.assertEqual(status['status'], 'Success')
+        resp.update(timeout=5)
         self.assertFalse(resp.is_open())
 
         number_of_pods = len(api.list_pod_for_all_namespaces().items)
@@ -226,4 +231,4 @@ def test_node_apis(self):
         for item in api.list_node().items:
             node = api.read_node(name=item.metadata.name)
             self.assertTrue(len(node.metadata.labels) > 0)
-            self.assertTrue(isinstance(node.metadata.labels, dict))
+            self.assertTrue(isinstance(node.metadata.labels, dict))