fix(proxy): fix data race in DialContext

Author: 3mbe

Makefile

@@ -943,7 +943,7 @@ test-infrastructure: $(SETUP_ENVTEST) ## Run unit and integration tests with rac
 	# Note: Fuzz tests are not executed with race detector because they would just time out.
 	# To achieve that, all files with fuzz tests have the "!race" build tag, to still run fuzz tests
 	# we have an additional `go test` run that focuses on "TestFuzzyConversion".
-	cd test/infrastructure; KUBEBUILDER_ASSETS="$(KUBEBUILDER_ASSETS)" go test ./... $(TEST_ARGS)
+	cd test/infrastructure; KUBEBUILDER_ASSETS="$(KUBEBUILDER_ASSETS)" go test -race $(TEST_ARGS) ./...
 	$(MAKE) test-infrastructure-conversions TEST_ARGS="$(TEST_ARGS)"
 
 .PHONY: test-infrastructure-conversions
@@ -956,10 +956,11 @@ test-infrastructure-verbose: ## Run unit and integration tests with race detecto
 
 .PHONY: test-infrastructure-junit
 test-infrastructure-junit: $(SETUP_ENVTEST) $(GOTESTSUM) ## Run unit and integration tests with race detector and generate a junit report for docker infrastructure provider
-	cd test/infrastructure; set +o errexit; (KUBEBUILDER_ASSETS="$(KUBEBUILDER_ASSETS)" go test -json ./... $(TEST_ARGS); echo $$? > $(ARTIFACTS)/junit.infra_docker.exitcode) | tee $(ARTIFACTS)/junit.infra_docker.stdout
+	@mkdir -p "$(ARTIFACTS)"
+	cd test/infrastructure; set +o errexit; (KUBEBUILDER_ASSETS="$(KUBEBUILDER_ASSETS)" go test -race $(TEST_ARGS) -json ./...; echo $$? > $(ARTIFACTS)/junit.infra_docker.exitcode) | tee $(ARTIFACTS)/junit.infra_docker.stdout
 	$(GOTESTSUM) --junitfile $(ARTIFACTS)/junit.infra_docker.xml --raw-command cat $(ARTIFACTS)/junit.infra_docker.stdout
 	exit $$(cat $(ARTIFACTS)/junit.infra_docker.exitcode)
-	cd test/infrastructure; set +o errexit; (KUBEBUILDER_ASSETS="$(KUBEBUILDER_ASSETS)" go test -run "^TestFuzzyConversion$$" -json ./... $(TEST_ARGS); echo $$? > $(ARTIFACTS)/junit-fuzz.infra_docker.exitcode) | tee $(ARTIFACTS)/junit-fuzz.infra_docker.stdout
+	cd test/infrastructure; set +o errexit; (KUBEBUILDER_ASSETS="$(KUBEBUILDER_ASSETS)" go test -run "^TestFuzzyConversion$$" $(TEST_ARGS) -json ./...; echo $$? > $(ARTIFACTS)/junit-fuzz.infra_docker.exitcode) | tee $(ARTIFACTS)/junit-fuzz.infra_docker.stdout
 	$(GOTESTSUM) --junitfile $(ARTIFACTS)/junit-fuzz.infra_docker.xml --raw-command cat $(ARTIFACTS)/junit-fuzz.infra_docker.stdout
 	exit $$(cat $(ARTIFACTS)/junit-fuzz.infra_docker.exitcode)
 

test/infrastructure/inmemory/pkg/server/proxy/dial.go

@@ -33,85 +33,80 @@ import (
 
 const defaultTimeout = 10 * time.Second
 
-// Dialer creates connections using Kubernetes API Server port-forwarding.
+// Dialer opens connections via Kubernetes API server port-forward.
+//
+// Safe for concurrent use: each dial builds its own SPDY transport and http.Client
+// to avoid races.
 type Dialer struct {
-	proxy          Proxy
-	clientset      *kubernetes.Clientset
-	proxyTransport http.RoundTripper
-	upgrader       spdy.Upgrader
-	timeout        time.Duration
+	proxy     Proxy
+	clientset *kubernetes.Clientset
+	timeout   time.Duration
 }
 
-// NewDialer creates a new dialer for a given API server scope.
+// NewDialer returns a Dialer for proxy p.
+// Options may set timeouts or other fields. The timeout also updates p.KubeConfig.
 func NewDialer(p Proxy, options ...func(*Dialer) error) (*Dialer, error) {
 	if p.Port == 0 {
 		return nil, errors.New("port required")
 	}
 
-	dialer := &Dialer{
-		proxy: p,
-	}
+	d := &Dialer{proxy: p}
 
 	for _, option := range options {
-		err := option(dialer)
-		if err != nil {
+		if err := option(d); err != nil {
 			return nil, err
 		}
 	}
 
-	if dialer.timeout == 0 {
-		dialer.timeout = defaultTimeout
-	}
-	p.KubeConfig.Timeout = dialer.timeout
-	clientset, err := kubernetes.NewForConfig(p.KubeConfig)
-	if err != nil {
-		return nil, err
+	if d.timeout == 0 {
+		d.timeout = defaultTimeout
 	}
-	proxyTransport, upgrader, err := spdy.RoundTripperFor(p.KubeConfig)
+	p.KubeConfig.Timeout = d.timeout
+
+	cs, err := kubernetes.NewForConfig(p.KubeConfig)
 	if err != nil {
 		return nil, err
 	}
-	dialer.proxyTransport = proxyTransport
-	dialer.upgrader = upgrader
-	dialer.clientset = clientset
-	return dialer, nil
+	d.clientset = cs
+	return d, nil
 }
 
-// DialContextWithAddr is a GO grpc compliant dialer construct.
+// DialContextWithAddr matches gRPC's dialer signature and calls DialContext.
 func (d *Dialer) DialContextWithAddr(ctx context.Context, addr string) (net.Conn, error) {
 	return d.DialContext(ctx, scheme, addr)
 }
 
-// DialContext creates proxied port-forwarded connections.
-// ctx is currently unused, but fulfils the type signature used by GRPC.
+// DialContext opens a port-forwarded connection to addr.
+//
+// Builds a fresh SPDY transport each time to avoid shared-state races.
+// ctx is kept for gRPC compatibility but ignored here.
 func (d *Dialer) DialContext(_ context.Context, _ string, addr string) (net.Conn, error) {
+	proxyTransport, upgrader, err := spdy.RoundTripperFor(d.proxy.KubeConfig)
+	if err != nil {
+		return nil, err
+	}
+	httpClient := &http.Client{Transport: proxyTransport}
+
 	req := d.clientset.CoreV1().RESTClient().
 		Post().
 		Resource(d.proxy.Kind).
 		Namespace(d.proxy.Namespace).
 		Name(addr).
 		SubResource("portforward")
 
-	dialer := spdy.NewDialer(d.upgrader, &http.Client{Transport: d.proxyTransport}, "POST", req.URL())
+	dialer := spdy.NewDialer(upgrader, httpClient, "POST", req.URL())
 
-	// Create a new connection from the dialer.
-	//
-	// Warning: Any early return should close this connection, otherwise we're going to leak them.
 	connection, _, err := dialer.Dial(portforward.PortForwardProtocolV1Name)
 	if err != nil {
 		return nil, errors.Wrap(err, "error upgrading connection")
 	}
 
-	// Create the headers.
+	// Port must match proxy.Port; request ID is always "0".
 	headers := http.Header{}
-
-	// Set the header port number to match the proxy one.
 	headers.Set(corev1.PortHeader, fmt.Sprintf("%d", d.proxy.Port))
-
-	// We only create a single stream over the connection
 	headers.Set(corev1.PortForwardRequestIDHeader, "0")
 
-	// Create the error stream.
+	// Error stream: required by protocol, closed immediately.
 	headers.Set(corev1.StreamType, corev1.StreamTypeError)
 	errorStream, err := connection.CreateStream(headers)
 	if err != nil {
@@ -120,18 +115,14 @@ func (d *Dialer) DialContext(_ context.Context, _ string, addr string) (net.Conn
 			connection.Close(),
 		})
 	}
-	// Close the error stream right away, we're not writing to it.
 	if err := errorStream.Close(); err != nil {
 		return nil, kerrors.NewAggregate([]error{
 			err,
 			connection.Close(),
 		})
 	}
 
-	// Create the data stream.
-	//
-	// NOTE: Given that we're reusing the headers,
-	// we need to overwrite the stream type before creating it.
+	// Data stream: switch type and create the I/O stream.
 	headers.Set(corev1.StreamType, corev1.StreamTypeData)
 	dataStream, err := connection.CreateStream(headers)
 	if err != nil {
@@ -141,17 +132,17 @@ func (d *Dialer) DialContext(_ context.Context, _ string, addr string) (net.Conn
 		})
 	}
 
-	// Create the net.Conn and return.
 	return NewConn(connection, dataStream), nil
 }
 
-// DialTimeout sets the timeout.
+// DialTimeout sets the per-dial timeout (also applied to KubeConfig).
 func DialTimeout(duration time.Duration) func(*Dialer) error {
 	return func(d *Dialer) error {
 		return d.setTimeout(duration)
 	}
 }
 
+// setTimeout updates timeout; used by DialTimeout.
 func (d *Dialer) setTimeout(duration time.Duration) error {
 	d.timeout = duration
 	return nil