fix #5036: addressing the handling of non-connection errors

Author: shawkins

CHANGELOG.md

@@ -15,6 +15,7 @@
 * Fix #5015: executing resync as a locking operation to ensure resync event ordering
 * Fix #5020: updating the resourceVersion on a delete with finalizers
 * Fix #5033: port forwarding for clients other than okhttp needs to specify the subprotocol
+* fix #5036: Better websocket error handling for protocol / client enforced errors, also update frame/message limits
 * Fix #5044: disable Vert.x instance file caching
 * Fix #5059: Vert.x InputStreamReader uses an empty Buffer sentinel to avoid NPE
 

httpclient-jdk/src/main/java/io/fabric8/kubernetes/client/jdkhttp/JdkWebSocketImpl.java

@@ -82,7 +82,7 @@ public CompletionStage<?> onClose(java.net.http.WebSocket webSocket, int statusC
 
     @Override
     public void onError(java.net.http.WebSocket webSocket, Throwable error) {
-      listener.onError(new JdkWebSocketImpl(queueSize, webSocket), error);
+      listener.onError(new JdkWebSocketImpl(queueSize, webSocket), error, false);
     }
 
     @Override

httpclient-jetty/src/main/java/io/fabric8/kubernetes/client/jetty/JettyHttpClientBuilder.java

@@ -39,6 +39,8 @@ public class JettyHttpClientBuilder
     extends StandardHttpClientBuilder<JettyHttpClient, JettyHttpClientFactory, JettyHttpClientBuilder> {
 
   private static final int MAX_CONNECTIONS = Integer.MAX_VALUE;
+  // the default for etcd seems to be 3 MB, but we'll default to unlimited to have the same behavior across clients
+  private static final int MAX_WS_MESSAGE_SIZE = Integer.MAX_VALUE;
 
   public JettyHttpClientBuilder(JettyHttpClientFactory clientFactory) {
     super(clientFactory);
@@ -62,6 +64,10 @@ public JettyHttpClient build() {
     }
     HttpClient sharedHttpClient = new HttpClient(newTransport(sslContextFactory, preferHttp11));
     WebSocketClient sharedWebSocketClient = new WebSocketClient(new HttpClient(newTransport(sslContextFactory, preferHttp11)));
+    sharedWebSocketClient.setMaxBinaryMessageSize(MAX_WS_MESSAGE_SIZE);
+    // the api-server does not seem to fragment messages, so the frames can be very large
+    sharedWebSocketClient.setMaxFrameSize(MAX_WS_MESSAGE_SIZE);
+    sharedWebSocketClient.setMaxTextMessageSize(MAX_WS_MESSAGE_SIZE);
     sharedWebSocketClient.setIdleTimeout(Duration.ZERO);
     if (connectTimeout != null) {
       sharedHttpClient.setConnectTimeout(connectTimeout.toMillis());

httpclient-jetty/src/main/java/io/fabric8/kubernetes/client/jetty/JettyWebSocket.java

@@ -26,6 +26,7 @@
 import org.eclipse.jetty.websocket.api.WriteCallback;
 import org.eclipse.jetty.websocket.api.exceptions.UpgradeException;
 
+import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.ClosedChannelException;
 import java.util.Collections;
@@ -147,7 +148,7 @@ public void onWebSocketError(Throwable cause) {
       // - Jetty throws a ClosedChannelException
       return;
     }
-    listener.onError(this, cause);
+    listener.onError(this, cause, cause instanceof IOException);
   }
 
   private void backPressure() {

httpclient-jetty/src/test/java/io/fabric8/kubernetes/client/jetty/JettyWebSocketTest.java

@@ -258,8 +258,8 @@ public void onClose(WebSocket webSocket, int code, String reason) {
     }
 
     @Override
-    public void onError(WebSocket webSocket, Throwable error) {
-      events.put("onError", new Object[] { error });
+    public void onError(WebSocket webSocket, Throwable error, boolean connectionError) {
+      events.put("onError", new Object[] { error, connectionError });
     }
   }
 }

httpclient-okhttp/src/main/java/io/fabric8/kubernetes/client/okhttp/OkHttpWebSocketImpl.java

@@ -90,7 +90,7 @@ public void onFailure(okhttp3.WebSocket webSocket, Throwable t, Response respons
             future.completeExceptionally(t);
           }
         } else {
-          listener.onError(new OkHttpWebSocketImpl(webSocket, this::request), t);
+          listener.onError(new OkHttpWebSocketImpl(webSocket, this::request), t, true);
         }
       }
 

httpclient-vertx/src/main/java/io/fabric8/kubernetes/client/vertx/VertxHttpClientBuilder.java

@@ -36,6 +36,8 @@ public class VertxHttpClientBuilder<F extends HttpClient.Factory>
     extends StandardHttpClientBuilder<VertxHttpClient<F>, F, VertxHttpClientBuilder<F>> {
 
   private static final int MAX_CONNECTIONS = 8192;
+  // the default for etcd seems to be 3 MB, but we'll default to unlimited to have the same behavior across clients
+  private static final int MAX_WS_MESSAGE_SIZE = Integer.MAX_VALUE;
 
   final Vertx vertx;
 
@@ -51,6 +53,9 @@ public VertxHttpClient<F> build() {
     options.setMaxPoolSize(MAX_CONNECTIONS);
     options.setMaxWebSockets(MAX_CONNECTIONS);
     options.setIdleTimeoutUnit(TimeUnit.SECONDS);
+    // the api-server does not seem to fragment messages, so the frames can be very large
+    options.setMaxWebSocketFrameSize(MAX_WS_MESSAGE_SIZE);
+    options.setMaxWebSocketMessageSize(MAX_WS_MESSAGE_SIZE);
 
     if (this.connectTimeout != null) {
       options.setConnectTimeout((int) this.connectTimeout.toMillis());

httpclient-vertx/src/main/java/io/fabric8/kubernetes/client/vertx/VertxWebSocket.java

@@ -20,6 +20,7 @@
 import io.netty.buffer.Unpooled;
 import io.vertx.core.Future;
 import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.HttpClosedException;
 
 import java.nio.ByteBuffer;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -52,7 +53,16 @@ void init() {
     ws.pongHandler(b -> ws.fetch(1));
     // use end, not close, because close is processed immediately vs. end is in frame order
     ws.endHandler(v -> listener.onClose(this, ws.closeStatusCode(), ws.closeReason()));
-    ws.exceptionHandler(err -> listener.onError(this, err));
+    ws.exceptionHandler(err -> {
+      try {
+        listener.onError(this, err, err instanceof HttpClosedException);
+      } finally {
+        // onError should be terminal
+        if (!ws.isClosed()) {
+          ws.close();
+        }
+      }
+    });
     listener.onOpen(this);
   }
 

kubernetes-client-api/src/main/java/io/fabric8/kubernetes/client/http/WebSocket.java

@@ -24,7 +24,7 @@ public interface WebSocket {
 
   /**
    * Callback methods for websocket events. The methods are
-   * guaranteed to be called serially - except for {@link #onError(WebSocket, Throwable)}
+   * guaranteed to be called serially - except for {@link Listener#onError(WebSocket, Throwable, boolean)}
    */
   interface Listener {
 
@@ -59,7 +59,7 @@ default void onClose(WebSocket webSocket, int code, String reason) {
      * Called when an error has occurred. It's a terminal event, calls to {@link WebSocket#request()}
      * do nothing after this.
      */
-    default void onError(WebSocket webSocket, Throwable error) {
+    default void onError(WebSocket webSocket, Throwable error, boolean connectionError) {
     }
 
   }

kubernetes-client/src/main/java/io/fabric8/kubernetes/client/dsl/internal/ExecWebSocketListener.java

@@ -261,7 +261,7 @@ public void onOpen(WebSocket webSocket) {
   }
 
   @Override
-  public void onError(WebSocket webSocket, Throwable t) {
+  public void onError(WebSocket webSocket, Throwable t, boolean connectionError) {
     closed.set(true);
     HttpResponse<?> response = null;