Support exec over the SPDY protocol

Author: admilazz

src/KubernetesClient/ByteBuffer.cs

@@ -5,9 +5,11 @@
 
 namespace k8s
 {
-    // There may be already an async implementation that we can use:
-    // https://github.com/StephenCleary/AsyncEx/wiki/AsyncProducerConsumerQueue
-    // However, they focus on individual objects and may not be a good choice for use with fixed-with byte buffers
+    // Pipe could be used instead for an async implementation. However, since this class is public there's no telling whether somebody
+    // has referenced it, so removing this class would be a breaking change. Neither could we easily replace the implementation of
+    // this class with Pipe's implementation, since this class exposes various public members (like WriteWaterMark and the OnResize event)
+    // that don't make sense in Pipe's context. (It seems they are used for unit testing, but unfortunately they were made public rather
+    // than internal, so removing them is a breaking change.)
 
     /// <summary>
     /// Represents a bounded buffer. A dedicated thread can send bytes to this buffer (the producer); while another thread can

src/KubernetesClient/Kubernetes.ConfigInit.cs

@@ -56,8 +56,8 @@ public Kubernetes(KubernetesClientConfiguration config, HttpClient httpClient, b
         ///     Optional. The delegating handlers to add to the http client pipeline.
         /// </param>
         public Kubernetes(KubernetesClientConfiguration config, params DelegatingHandler[] handlers)
-            : this(handlers)
         {
+            CreateHttpClient(handlers);
             ValidateConfig(config);
             this.config = config;
             InitializeFromConfig();
@@ -77,6 +77,18 @@ public KubernetesScheme Scheme
             }
         }
 
+        // TODO: uncomment if we update to a later version of Microsoft.Rest.ClientRuntime
+        /*/// <inheritdoc/>
+        protected override DelegatingHandler CreateHttpHandlerPipeline(HttpClientHandler httpClientHandler, params DelegatingHandler[] handlers)
+        {
+            HttpMessageHandler rootHandler = RemoveRetryHandler(base.CreateHttpHandlerPipeline(httpClientHandler, handlers));
+            return rootHandler as DelegatingHandler ?? new ForwardingHandler(rootHandler);
+        }
+
+        private sealed class ForwardingHandler : DelegatingHandler
+        {
+            public ForwardingHandler(HttpMessageHandler handler) : base(handler) { }
+        }*/
 
         private void ValidateConfig(KubernetesClientConfiguration config)
         {
@@ -171,15 +183,21 @@ partial void CustomInitialize()
 #if NET452 
             ServicePointManager.SecurityProtocol |= SecurityProtocolType.Tls12;
 #endif
+            FirstMessageHandler = RemoveRetryHandler(FirstMessageHandler);
             AppendDelegatingHandler<WatcherDelegatingHandler>();
             DeserializationSettings.Converters.Add(new V1Status.V1StatusObjectViewConverter());
         }
 
         private void AppendDelegatingHandler<T>() where T : DelegatingHandler, new()
         {
             var cur = FirstMessageHandler as DelegatingHandler;
+            if (cur == null)
+            {
+                FirstMessageHandler = new T() { InnerHandler = FirstMessageHandler };
+                return;
+            }
 
-            while (cur != null)
+            do
             {
                 var next = cur.InnerHandler as DelegatingHandler;
 
@@ -195,7 +213,52 @@ partial void CustomInitialize()
                 }
 
                 cur = next;
+            } while (cur != null);
+        }
+
+        // NOTE: this method replicates the logic that the base ServiceClient uses except that it doesn't insert the RetryDelegatingHandler.
+        // (see RemoveRetryHandler below for why we don't want it.) it seems that depending on your framework version and/or
+        // Microsoft.Rest.ClientRuntime version and/or which constructor gets called, there are at least two ways that the retry handler
+        // gets added: 1) when the HTTP client is constructed (handled here), 2) in CreateHttpHandlerPipeline (handled in an override), and
+        // possibly 3) based on the handlers set in CustomInitialize
+        private void CreateHttpClient(DelegatingHandler[] handlers)
+        {
+            FirstMessageHandler = HttpClientHandler = CreateRootHandler();
+            if(handlers != null)
+            {
+                for(int i = handlers.Length - 1; i >= 0; i--)
+                {
+                    DelegatingHandler handler = handlers[i];
+                    while(handler.InnerHandler is DelegatingHandler d) handler = d;
+                    handler.InnerHandler = FirstMessageHandler;
+                    FirstMessageHandler = handlers[i];
+                }
+            }
+            HttpClient = new HttpClient(FirstMessageHandler, false);
+        }
+
+        /// <summary>Removes the retry handler added by the Microsoft.Rest.ClientRuntime.</summary>
+        // NOTE: we remove the RetryDelegatingHandler for two reasons. first, it has a very broad definition of what's considered a failed
+        // request. it considers everything outside 2xx to be failed, including 1xx (e.g. 101 Switching Protocols) and 3xx. and, it wants to
+        // retry 4xx errors, almost none of which are really retriable except 429 Too Many Requests and /maybe/ 423 Locked (but we're not
+        // dealing with WebDAV here). really, i don't think we want retry in a Kubernetes client, and as for 429 Too Many Requests, there's
+        // already a separate RetryAfterDelegatingHandler installed to handle 429 requests. a further problem with the RetryDelegatingHandler
+        // is that upon seeing a non-2xx status code it tries to read the entire response body as a string. this doesn't work well with
+        // streaming responses like watches and upgraded (SPDY / web socket) connections.
+        private HttpMessageHandler RemoveRetryHandler(HttpMessageHandler rootHandler)
+        {
+            for (DelegatingHandler prev = null, handler = rootHandler as DelegatingHandler; handler != null;)
+            {
+                if (handler is RetryDelegatingHandler)
+                {
+                    if (prev == null) rootHandler = handler.InnerHandler; // if 'handler' is at the head of the chain, just return the next item
+                    else prev.InnerHandler = handler.InnerHandler; // otherwise, unlink 'handler' from the chain
+                    break;
+                }
+                prev = handler;
+                handler = handler.InnerHandler as DelegatingHandler;
             }
+            return rootHandler;
         }
 
         /// <summary>

src/KubernetesClient/Kubernetes.Exec.cs

@@ -20,31 +20,17 @@ public async Task<int> NamespacedPodExecAsync(string name, string @namespace, st
                 throw new ArgumentNullException(nameof(action));
             }
 
-            try
+            Stream stdIn = new PipeStream(), stdOut = new PipeStream(), stdErr = new PipeStream();
+            Task<V1Status> execTask = Request<V1Pod>(@namespace, name).Body(stdIn)
+                .ExecCommandAsync(command.First(), command.Skip(1).ToArray(), container, stdOut, stdErr, tty, false, cancellationToken);
+            await action(stdIn, stdOut, stdErr).ConfigureAwait(false);
+            stdIn.Dispose(); // close STDIN just in case the action failed to do so and the remote process is waiting for it
+            var status = await execTask.ConfigureAwait(false);
+            if (status.Code.Value < 0)
             {
-                using (var muxedStream = await this.MuxedStreamNamespacedPodExecAsync(name: name, @namespace: @namespace, command: command, container: container, tty: tty, cancellationToken: cancellationToken).ConfigureAwait(false))
-                using (Stream stdIn = muxedStream.GetStream(null, ChannelIndex.StdIn))
-                using (Stream stdOut= muxedStream.GetStream(ChannelIndex.StdOut, null))
-                using (Stream stdErr = muxedStream.GetStream(ChannelIndex.StdErr, null))
-                using (Stream error = muxedStream.GetStream(ChannelIndex.Error, null))
-                using (StreamReader errorReader = new StreamReader(error))
-                {
-                    muxedStream.Start();
-
-                    await action(stdIn, stdOut, stdErr).ConfigureAwait(false);
-
-                    var errors = await errorReader.ReadToEndAsync().ConfigureAwait(false);
-
-                    // StatusError is defined here:
-                    // https://github.com/kubernetes/kubernetes/blob/068e1642f63a1a8c48c16c18510e8854a4f4e7c5/staging/src/k8s.io/apimachinery/pkg/api/errors/errors.go#L37
-                    var returnMessage = SafeJsonConvert.DeserializeObject<V1Status>(errors);
-                    return GetExitCodeOrThrow(returnMessage);
-                }
-            }
-            catch (HttpOperationException httpEx) when (httpEx.Body is V1Status)
-            {
-                throw new KubernetesException((V1Status)httpEx.Body);
+                throw new KubernetesException(status);
             }
+            return status.Code.Value;
         }
 
         /// <summary>

src/KubernetesClient/KubernetesClient.csproj

@@ -37,6 +37,7 @@
     <PackageReference Include="Microsoft.Rest.ClientRuntime" Version="2.3.10" />
     <PackageReference Include="Newtonsoft.Json" Version="9.0.1" Condition="'$(TargetFramework)' != 'netstandard2.0' and '$(TargetFramework)' != 'netcoreapp2.1'" />
     <PackageReference Include="Newtonsoft.Json" Version="12.0.2" Condition="'$(TargetFramework)' == 'netstandard2.0' or '$(TargetFramework)' == 'netcoreapp2.1'" />
+    <PackageReference Include="SPDY.net" Version="1.0.0" />
     <PackageReference Include="YamlDotNet" Version="6.0.0" />
     <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All" />
   </ItemGroup>