Allow callers to specify which WebSocket subprotocol to use (#154)

* Allow callers to specify which WebSocket subprotocol to use.

* Update unit tests

* Update unit test

* Fix ordering

* Trace CancellationToken

Author: qmfrederik

src/KubernetesClient/IKubernetes.WebSocket.cs

@@ -40,6 +40,10 @@ public partial interface IKubernetes
         /// TTY if true indicates that a tty will be allocated for the exec call.
         /// Defaults to <see langword="true"/>.
         /// </param>
+        /// <param name="webSocketSubProtocol">
+        /// The Kubernetes-specific WebSocket sub protocol to use. See <see cref="WebSocketProtocol"/> for a list of available
+        /// protocols.
+        /// </param>
         /// <param name='customHeaders'>
         /// Headers that will be added to request.
         /// </param>
@@ -52,7 +56,7 @@ public partial interface IKubernetes
         /// <return>
         /// A <see cref="ClientWebSocket"/> which can be used to communicate with the process running in the pod.
         /// </return>
-        Task<WebSocket> WebSocketNamespacedPodExecAsync(string name, string @namespace = "default", string command = null, string container = null, bool stderr = true, bool stdin = true, bool stdout = true, bool tty = true, Dictionary<string, List<string>> customHeaders = null, CancellationToken cancellationToken = default(CancellationToken));
+        Task<WebSocket> WebSocketNamespacedPodExecAsync(string name, string @namespace = "default", string command = null, string container = null, bool stderr = true, bool stdin = true, bool stdout = true, bool tty = true, string webSocketSubProtol = null, Dictionary<string, List<string>> customHeaders = null, CancellationToken cancellationToken = default(CancellationToken));
 
         /// <summary>
         /// Executes a command in a pod.
@@ -87,6 +91,10 @@ public partial interface IKubernetes
         /// TTY if true indicates that a tty will be allocated for the exec call.
         /// Defaults to <see langword="true"/>.
         /// </param>
+        /// <param name="webSocketSubProtocol">
+        /// The Kubernetes-specific WebSocket sub protocol to use. See <see cref="WebSocketProtocol"/> for a list of available
+        /// protocols.
+        /// </param>
         /// <param name='customHeaders'>
         /// Headers that will be added to request.
         /// </param>
@@ -99,7 +107,7 @@ public partial interface IKubernetes
         /// <return>
         /// A <see cref="ClientWebSocket"/> which can be used to communicate with the process running in the pod.
         /// </return>
-        Task<WebSocket> WebSocketNamespacedPodExecAsync(string name, string @namespace = "default", IEnumerable<string> command = null, string container = null, bool stderr = true, bool stdin = true, bool stdout = true, bool tty = true, Dictionary<string, List<string>> customHeaders = null, CancellationToken cancellationToken = default(CancellationToken));
+        Task<WebSocket> WebSocketNamespacedPodExecAsync(string name, string @namespace = "default", IEnumerable<string> command = null, string container = null, bool stderr = true, bool stdin = true, bool stdout = true, bool tty = true, string webSocketSubProtol = null, Dictionary<string, List<string>> customHeaders = null, CancellationToken cancellationToken = default(CancellationToken));
 
         /// <summary>
         /// Start port forwarding one or more ports of a pod.
@@ -113,13 +121,17 @@ public partial interface IKubernetes
         /// <param name='ports'>
         /// List of ports to forward.
         /// </param>
+        /// <param name="webSocketSubProtocol">
+        /// The Kubernetes-specific WebSocket sub protocol to use. See <see cref="WebSocketProtocol"/> for a list of available
+        /// protocols.
+        /// </param>
         /// <param name='customHeaders'>
         /// The headers that will be added to request.
         /// </param>
         /// <param name='cancellationToken'>
         /// The cancellation token.
         /// </param>
-        Task<WebSocket> WebSocketNamespacedPodPortForwardAsync(string name, string @namespace, IEnumerable<int> ports, Dictionary<string, List<string>> customHeaders = null, CancellationToken cancellationToken = default(CancellationToken));
+        Task<WebSocket> WebSocketNamespacedPodPortForwardAsync(string name, string @namespace, IEnumerable<int> ports, string webSocketSubProtocol = null, Dictionary<string, List<string>> customHeaders = null, CancellationToken cancellationToken = default(CancellationToken));
 
         /// <summary>
         /// connect GET requests to attach of Pod
@@ -151,6 +163,10 @@ public partial interface IKubernetes
         /// This is passed through the container runtime so the tty is allocated on the
         /// worker node by the container runtime. Defaults to false.
         /// </param>
+        /// <param name="webSocketSubProtocol">
+        /// The Kubernetes-specific WebSocket sub protocol to use. See <see cref="WebSocketProtocol"/> for a list of available
+        /// protocols.
+        /// </param>
         /// <param name='customHeaders'>
         /// Headers that will be added to request.
         /// </param>
@@ -172,6 +188,6 @@ public partial interface IKubernetes
         /// <return>
         /// A response object containing the response body and response headers.
         /// </return>
-        Task<WebSocket> WebSocketNamespacedPodAttachAsync(string name, string @namespace, string container = default(string), bool stderr = true, bool stdin = false, bool stdout = true, bool tty = false, Dictionary<string, List<string>> customHeaders = null, CancellationToken cancellationToken = default(CancellationToken));
+        Task<WebSocket> WebSocketNamespacedPodAttachAsync(string name, string @namespace, string container = default(string), bool stderr = true, bool stdin = false, bool stdout = true, bool tty = false, string webSocketSubProtol = null, Dictionary<string, List<string>> customHeaders = null, CancellationToken cancellationToken = default(CancellationToken));
     }
 }

src/KubernetesClient/K8sProtocol.cs

@@ -23,13 +23,13 @@ public partial class Kubernetes
         public Func<WebSocketBuilder> CreateWebSocketBuilder { get; set; } = () => new WebSocketBuilder();
 
         /// <inheritdoc/>
-        public Task<WebSocket> WebSocketNamespacedPodExecAsync(string name, string @namespace = "default", string command = null, string container = null, bool stderr = true, bool stdin = true, bool stdout = true, bool tty = true, Dictionary<string, List<string>> customHeaders = null, CancellationToken cancellationToken = default(CancellationToken))
+        public Task<WebSocket> WebSocketNamespacedPodExecAsync(string name, string @namespace = "default", string command = null, string container = null, bool stderr = true, bool stdin = true, bool stdout = true, bool tty = true, string webSocketSubProtol = null, Dictionary<string, List<string>> customHeaders = null, CancellationToken cancellationToken = default(CancellationToken))
         {
-            return WebSocketNamespacedPodExecAsync(name, @namespace, new string[] { command }, container, stderr, stdin, stdout, tty, customHeaders, cancellationToken);
+            return WebSocketNamespacedPodExecAsync(name, @namespace, new string[] { command }, container, stderr, stdin, stdout, tty, webSocketSubProtol, customHeaders, cancellationToken);
         }
 
         /// <inheritdoc/>
-        public Task<WebSocket> WebSocketNamespacedPodExecAsync(string name, string @namespace = "default", IEnumerable<string> command = null, string container = null, bool stderr = true, bool stdin = true, bool stdout = true, bool tty = true, Dictionary<string, List<string>> customHeaders = null, CancellationToken cancellationToken = default(CancellationToken))
+        public Task<WebSocket> WebSocketNamespacedPodExecAsync(string name, string @namespace = "default", IEnumerable<string> command = null, string container = null, bool stderr = true, bool stdin = true, bool stdout = true, bool tty = true, string webSocketSubProtol = null, Dictionary<string, List<string>> customHeaders = null, CancellationToken cancellationToken = default(CancellationToken))
         {
             if (name == null)
             {
@@ -66,6 +66,7 @@ public partial class Kubernetes
                 tracingParameters.Add("stdin", stdin);
                 tracingParameters.Add("stdout", stdout);
                 tracingParameters.Add("tty", tty);
+                tracingParameters.Add("webSocketSubProtol", webSocketSubProtol);
                 tracingParameters.Add("cancellationToken", cancellationToken);
                 ServiceClientTracing.Enter(_invocationId, this, nameof(WebSocketNamespacedPodExecAsync), tracingParameters);
             }
@@ -103,11 +104,11 @@ public partial class Kubernetes
 
             uriBuilder.Query = query;
 
-            return this.StreamConnectAsync(uriBuilder.Uri, _invocationId, customHeaders, cancellationToken);
+            return this.StreamConnectAsync(uriBuilder.Uri, _invocationId, webSocketSubProtol, customHeaders, cancellationToken);
         }
 
         /// <inheritdoc/>
-        public Task<WebSocket> WebSocketNamespacedPodPortForwardAsync(string name, string @namespace, IEnumerable<int> ports, Dictionary<string, List<string>> customHeaders = null, CancellationToken cancellationToken = default(CancellationToken))
+        public Task<WebSocket> WebSocketNamespacedPodPortForwardAsync(string name, string @namespace, IEnumerable<int> ports, string webSocketSubProtocol = null, Dictionary<string, List<string>> customHeaders = null, CancellationToken cancellationToken = default(CancellationToken))
         {
             if (name == null)
             {
@@ -134,6 +135,7 @@ public partial class Kubernetes
                 tracingParameters.Add("name", name);
                 tracingParameters.Add("@namespace", @namespace);
                 tracingParameters.Add("ports", ports);
+                tracingParameters.Add("webSocketSubProtocol", webSocketSubProtocol);
                 tracingParameters.Add("cancellationToken", cancellationToken);
                 ServiceClientTracing.Enter(_invocationId, this, nameof(WebSocketNamespacedPodPortForwardAsync), tracingParameters);
             }
@@ -158,11 +160,11 @@ public partial class Kubernetes
 
 
 
-            return StreamConnectAsync(uriBuilder.Uri, _invocationId, customHeaders, cancellationToken);
+            return StreamConnectAsync(uriBuilder.Uri, _invocationId, webSocketSubProtocol, customHeaders, cancellationToken);
         }
 
         /// <inheritdoc/>
-        public Task<WebSocket> WebSocketNamespacedPodAttachAsync(string name, string @namespace, string container = default(string), bool stderr = true, bool stdin = false, bool stdout = true, bool tty = false, Dictionary<string, List<string>> customHeaders = null, CancellationToken cancellationToken = default(CancellationToken))
+        public Task<WebSocket> WebSocketNamespacedPodAttachAsync(string name, string @namespace, string container = default(string), bool stderr = true, bool stdin = false, bool stdout = true, bool tty = false, string webSocketSubProtol = null, Dictionary<string, List<string>> customHeaders = null, CancellationToken cancellationToken = default(CancellationToken))
         {
             if (name == null)
             {
@@ -188,6 +190,7 @@ public partial class Kubernetes
                 tracingParameters.Add("stdin", stdin);
                 tracingParameters.Add("stdout", stdout);
                 tracingParameters.Add("tty", tty);
+                tracingParameters.Add("webSocketSubProtol", webSocketSubProtol);
                 tracingParameters.Add("cancellationToken", cancellationToken);
                 ServiceClientTracing.Enter(_invocationId, this, nameof(WebSocketNamespacedPodAttachAsync), tracingParameters);
             }
@@ -212,10 +215,10 @@ public partial class Kubernetes
                 { "tty", tty ? "1": "0"}
             }).TrimStart('?');
 
-            return StreamConnectAsync(uriBuilder.Uri, _invocationId, customHeaders, cancellationToken);
+            return StreamConnectAsync(uriBuilder.Uri, _invocationId, webSocketSubProtol, customHeaders, cancellationToken);
         }
 
-        protected async Task<WebSocket> StreamConnectAsync(Uri uri, string invocationId = null, Dictionary<string, List<string>> customHeaders = null, CancellationToken cancellationToken = default(CancellationToken))
+        protected async Task<WebSocket> StreamConnectAsync(Uri uri, string invocationId = null, string webSocketSubProtocol = null, Dictionary<string, List<string>> customHeaders = null, CancellationToken cancellationToken = default(CancellationToken))
         {
             bool _shouldTrace = ServiceClientTracing.IsEnabled;
 
@@ -258,11 +261,16 @@ public partial class Kubernetes
             {
                 webSocketBuilder.ExpectServerCertificate(this.CaCert);
             }
+
             if (this.SkipTlsVerify)
             {
                 webSocketBuilder.SkipServerCertificateValidation();
             }
-            webSocketBuilder.Options.RequestedSubProtocols.Add(K8sProtocol.ChannelV1);
+
+            if (webSocketSubProtocol != null)
+            {
+                webSocketBuilder.Options.RequestedSubProtocols.Add(webSocketSubProtocol);
+            }
 #endif // NETCOREAPP2_1
 
             // Send Request

src/KubernetesClient/Kubernetes.WebSocket.cs

@@ -0,0 +1,74 @@
+namespace k8s
+{
+    /// <summary>
+    ///     Well-known WebSocket sub-protocols used by the Kubernetes API.
+    /// </summary>
+    public static class WebSocketProtocol
+    {
+        // The protocols are defined here:
+        // https://github.com/kubernetes/kubernetes/blob/master/pkg/kubelet/server/remotecommand/websocket.go#L36
+        // and here:
+        // https://github.com/kubernetes/kubernetes/blob/714f97d7baf4975ad3aa47735a868a81a984d1f0/staging/src/k8s.io/apiserver/pkg/util/wsstream/conn.go
+        //
+        // For a description of what's different in v4:
+        // https://github.com/kubernetes/kubernetes/blob/317853c90c674920bfbbdac54fe66092ddc9f15f/pkg/kubelet/server/remotecommand/httpstream.go#L203
+
+        /// <summary>
+        /// <para>
+        /// The Websocket subprotocol "channel.k8s.io" prepends each binary message with a byte indicating
+        /// the channel number (zero indexed) the message was sent on. Messages in both directions should
+        /// prefix their messages with this channel byte. When used for remote execution, the channel numbers
+        /// are by convention defined to match the POSIX file-descriptors assigned to STDIN, STDOUT, and STDERR
+        /// (0, 1, and 2). No other conversion is performed on the raw subprotocol - writes are sent as they
+        /// are received by the server.
+        /// </para>
+        ///
+        /// <para>
+        /// Example client session:
+        /// <code>
+        ///    CONNECT http://server.com with subprotocol "channel.k8s.io"
+        ///    WRITE []byte{0, 102, 111, 111, 10} # send "foo\n" on channel 0 (STDIN)
+        ///    READ  []byte{1, 10}                # receive "\n" on channel 1 (STDOUT)
+        ///    CLOSE
+        /// </code>
+        /// </para>
+        /// </summary>
+        public const string ChannelWebSocketProtocol = "channel.k8s.io";
+
+        /// <summary>
+        /// <para>
+        /// The Websocket subprotocol "base64.channel.k8s.io" base64 encodes each message with a character
+        /// indicating the channel number (zero indexed) the message was sent on. Messages in both directions
+        /// should prefix their messages with this channel char. When used for remote execution, the channel
+        /// numbers are by convention defined to match the POSIX file-descriptors assigned to STDIN, STDOUT,
+        /// and STDERR ('0', '1', and '2'). The data received on the server is base64 decoded (and must be
+        /// be valid) and data written by the server to the client is base64 encoded.
+        /// </para>
+        ///
+        /// <para>
+        /// Example client session:
+        /// <code>
+        ///    CONNECT http://server.com with subprotocol "base64.channel.k8s.io"
+        ///    WRITE []byte{48, 90, 109, 57, 118, 67, 103, 111, 61} # send "foo\n" (base64: "Zm9vCgo=") on channel '0' (STDIN)
+        ///    READ  []byte{49, 67, 103, 61, 61} # receive "\n" (base64: "Cg==") on channel '1' (STDOUT)
+        ///    CLOSE
+        /// </code>
+        /// </para>
+        /// </summary>
+        public const string Base64ChannelWebSocketProtocol = "base64.channel.k8s.io";
+
+        /// <summary>
+        /// v4ProtocolHandler implements the V4 protocol version for streaming command execution. It only differs
+        /// in from v3 in the error stream format using an json-marshaled metav1.Status which carries
+        /// the process' exit code.
+        /// </summary>
+        public const string V4BinaryWebsocketProtocol = "v4." + ChannelWebSocketProtocol;
+
+        /// <summary>
+        /// v4ProtocolHandler implements the V4 protocol version for streaming command execution. It only differs
+        /// in from v3 in the error stream format using an json-marshaled metav1.Status which carries
+        /// the process' exit code.
+        /// </summary>
+        public const string V4Base64WebsocketProtocol = "v4." + Base64ChannelWebSocketProtocol;
+    }
+}