Wrap http transport with retry in kube proxy to handle GOAWAY.

[creack]

Fixes #57766

Adding test is quite tricky, need a Kube server with the --goaway-chance flag but it is limited to 2% max.

Go script to reproduce the issue / confirm the fix

package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	v1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/clientcmd"
)

func main() {
	ctx := context.Background()

	kubeconfig := filepath.Join(os.Getenv("HOME"), ".kube", "config")
	config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
	if err != nil {
		log.Fatalf("Failed to build config: %s.", err)
	}

	// Increase limits to avoid rate limiting.
	config.QPS = 100
	config.Burst = 200

	clientset, err := kubernetes.NewForConfig(config)
	if err != nil {
		log.Fatalf("Failed to create clientset: %s.", err)
	}

	const numWorkers = 20
	const opsPerWorker = 50

	var wg sync.WaitGroup
	for workerID := range numWorkers {
		wg.Add(1)
		go func() {
			defer wg.Done()

			for i := range opsPerWorker {
				cm := &v1.ConfigMap{
					ObjectMeta: metav1.ObjectMeta{
						Name:      fmt.Sprintf("test-cm-%d-%d-%d", workerID, i, time.Now().Unix()),
						Namespace: "default",
					},
					Data: map[string]string{
						"key":    fmt.Sprintf("value-%d", i),
						"worker": fmt.Sprintf("%d", workerID),
						"data":   strings.Repeat("x", 1024), // Add some data to make the body larger.
					},
				}

				created, err := clientset.CoreV1().ConfigMaps("default").Create(ctx, cm, metav1.CreateOptions{})
				if err != nil {
					if isGoawayError(err) {
						fmt.Printf("[Worker %d] GOAWAY on CREATE: %s.\n", workerID, err)
					}
					continue
				}

				created.Data["updated"] = "true"
				created.Data["timestamp"] = time.Now().String()
				if _, err := clientset.CoreV1().ConfigMaps("default").Update(ctx, created, metav1.UpdateOptions{}); err != nil {
					if isGoawayError(err) {
						fmt.Printf("[Worker %d] GOAWAY on UPDATE: %s.\n", workerID, err)
					}
				}

				if err := clientset.CoreV1().ConfigMaps("default").Delete(ctx, cm.Name, metav1.DeleteOptions{}); err != nil {
					if isGoawayError(err) {
						fmt.Printf("[Worker %d] GOAWAY on DELETE: %s.\n", workerID, err)
					}
				}

				// No delay - stress the connection.
			}
		}()
	}

	wg.Wait()
}

func isGoawayError(err error) bool {
	if err == nil {
		return false
	}

	errStr := err.Error()
	goawayPatterns := []string{
		"cannot retry err",
		"GOAWAY",
		"http2: Transport received Server's graceful shutdown",
		"after Request.Body was written",
		"graceful shutdown",
	}

	for _, pattern := range goawayPatterns {
		if strings.Contains(errStr, pattern) {
			fmt.Printf("<<< % #v -- %T\n", err, err)
			return true
		}
	}

	return false
}

Kind config with goaway-chance

kind: Cluster
apiVersion: kind.x-k8s.io/v1alpha4
name: teleport-goaway-test
nodes:
- role: control-plane
  kubeadmConfigPatches:
  - |
    kind: ClusterConfiguration
    apiServer:
      extraArgs:
        goaway-chance: "0.02"
        v: "2"

kind create cluster --config goaway.yaml

changelog: Add retry logic in kube proxy to handle EKS goaway chance

[tigrato]

we should move this under the forward library so if App access needs to use it's available there and you can enabled. I didn't enabled because I was afraid of breaking anything without running the test plan

[creack]

I think while there may be a case to make it available widely under the forward lib, here, we discriminate on which endpoints get the retryable handling which is quite specific to kube proxy. To make it somewhere else we'd need to add an option to pass a function that checks whether or not to apply the logic, which is a lot of overhead.

[rosstimothy]

Let's start simple and iterate here and keep this limited to Kubernetes for now.

[rosstimothy]

Let's start simple and iterate here and keep this limited to Kubernetes for now.

[rosstimothy]

Suggestion: Rename this to something less generic and more descriptive.

[rosstimothy]

Suggestion: Remove the embedding if possible.

[rosstimothy]

As discussed offline, is this accurate? Aren't we storing the request body, and not the response body? Why would this be any less of a problem for requests which are not for /log or are for a watch?

[tigrato]

all endpoints - watch requests included - suffer from this issue. only HTTP/1 requests (such as WebSocket or SPDY) do not and a simple protocol check avoids the problem.

In the Kubernetes codebase, every other type of request is required to retry the request body - for them it's not a problem because requests are bytes.Buffer and http client already populates the getbody after a type assertion.

[tigrato]

Basically kubernetes added this feature for watch requests

[rosstimothy]

I was asking why the comment indicated that log and watches needed special handling. If we are only caching the request and not the responses, then does it really matter that the response for a long lived request could be very large? Shouldn't this approach work for all requests?

[tigrato]

Let me explain what I meant.

The reason why goway-chance in Kubernetes was introduced was to close long-lived connections to the same kubernetes api server and balance the incoming connections through all replicas - this was designed specially for watch streams.

Looking at Kuberentes implementation, you can see that only HTTP2 requests might receive the go away - connection close. This means that any HTTP1 request won't receive the connection close header.

This HTTP2 filter, excludes long-lived bidirectional connections such as kubectl exec, kubectl port-forward, kubectl cp where the user can send very large payloads since all of these use HTTP1 + WebSocket/SPDY.

Long lived read only - such as watch or log streams - aren't affected as well because the GO HTTP2 client only retries the request body - means we only need to record the request body and not the response. This makes our life simpler given that Teleport only needs to be able to reset the request body and never the response body. Request bodies in Kubernetes are way smaller than response for any endpoint apart for the bidi-directional streams mentioned above.

In theory, we can record the body into memory and retry - the go HTTP2 client already does that automatically if you provide a GetBody function. In reality we can make a threshold of 1MB that goes into memory and if the body is bigger we write to a temp file.

I still think the approach that this PR tries to implement isn't the best - load the body in advance, trim the body... and we can definitively improve it. This PR also includes HTTP1 requests which will cause some problems specially with connection upgrade mechanisms.