[v14] Partially revert #30248

lib/kube/proxy/responsewriters/watcher.go

@@ -207,7 +207,7 @@ func (w *WatcherResponseWriter) watchDecoder(contentType string, writer io.Write
 	// wait for events received from upstream until the connection is terminated.
 	for {
 		eventType, obj, err := w.decodeStreamingMessage(streamingDecoder, objectDecoder)
-		if errors.Is(err, io.EOF) {
+		if errors.Is(err, io.EOF) || errors.Is(err, io.ErrClosedPipe) {
 			return nil
 		} else if err != nil {
 			return trace.Wrap(err)

lib/kube/proxy/server.go

@@ -239,11 +239,14 @@ func NewTLSServer(cfg TLSServerConfig) (*TLSServer, error) {
 		Server: &http.Server{
 			Handler:           httplib.MakeTracingHandler(limiter, teleport.ComponentKube),
 			ReadHeaderTimeout: apidefaults.DefaultIOTimeout * 2,
-			ReadTimeout:       apidefaults.DefaultIOTimeout,
-			WriteTimeout:      apidefaults.DefaultIOTimeout,
-			IdleTimeout:       apidefaults.DefaultIdleTimeout,
-			TLSConfig:         cfg.TLS,
-			ConnState:         ingress.HTTPConnStateReporter(ingress.Kube, cfg.IngressReporter),
+			// Setting ReadTimeout and WriteTimeout will cause the server to
+			// terminate long running requests. This will cause issues with
+			// long running watch streams. The server will close the connection
+			// and the client will receive incomplete data and will fail to
+			// parse it.
+			IdleTimeout: apidefaults.DefaultIdleTimeout,
+			TLSConfig:   cfg.TLS,
+			ConnState:   ingress.HTTPConnStateReporter(ingress.Kube, cfg.IngressReporter),
 			ConnContext: func(ctx context.Context, c net.Conn) context.Context {
 				return utils.ClientAddrContext(ctx, c.RemoteAddr(), c.LocalAddr())
 			},

lib/kube/proxy/utils_testing.go

@@ -336,7 +336,8 @@ func SetupTestContext(ctx context.Context, t *testing.T, cfg TestConfig) *TestCo
 		Log: log,
 	})
 	require.NoError(t, err)
-
+	require.Equal(t, testCtx.KubeServer.Server.ReadTimeout, time.Duration(0), "kube server write timeout must be 0")
+	require.Equal(t, testCtx.KubeServer.Server.WriteTimeout, time.Duration(0), "kube server write timeout must be 0")
 	// Waits for len(clusters) heartbeats to start
 	waitForHeartbeats := len(cfg.Clusters)