revert: restore original code with minimal diff

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 -race $(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,11 +956,10 @@ 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
-	@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
+	cd test/infrastructure; set +o errexit; (KUBEBUILDER_ASSETS="$(KUBEBUILDER_ASSETS)" go test -race -json ./... $(TEST_ARGS); 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$$" $(TEST_ARGS) -json ./...; 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$$" -json ./... $(TEST_ARGS); 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,53 +33,49 @@ import (
 
 const defaultTimeout = 10 * time.Second
 
-// 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.
+// Dialer creates connections using Kubernetes API Server port-forwarding.
 type Dialer struct {
 	proxy     Proxy
 	clientset *kubernetes.Clientset
 	timeout   time.Duration
 }
 
-// NewDialer returns a Dialer for proxy p.
-// Options may set timeouts or other fields. The timeout also updates p.KubeConfig.
+// NewDialer creates a new dialer for a given API server scope.
 func NewDialer(p Proxy, options ...func(*Dialer) error) (*Dialer, error) {
 	if p.Port == 0 {
 		return nil, errors.New("port required")
 	}
 
-	d := &Dialer{proxy: p}
+	dialer := &Dialer{
+		proxy: p,
+	}
 
 	for _, option := range options {
-		if err := option(d); err != nil {
+		err := option(dialer)
+		if err != nil {
 			return nil, err
 		}
 	}
 
-	if d.timeout == 0 {
-		d.timeout = defaultTimeout
+	if dialer.timeout == 0 {
+		dialer.timeout = defaultTimeout
 	}
-	p.KubeConfig.Timeout = d.timeout
-
-	cs, err := kubernetes.NewForConfig(p.KubeConfig)
+	p.KubeConfig.Timeout = dialer.timeout
+	clientset, err := kubernetes.NewForConfig(p.KubeConfig)
 	if err != nil {
 		return nil, err
 	}
-	d.clientset = cs
-	return d, nil
+	dialer.clientset = clientset
+	return dialer, nil
 }
 
-// DialContextWithAddr matches gRPC's dialer signature and calls DialContext.
+// DialContextWithAddr is a GO grpc compliant dialer construct.
 func (d *Dialer) DialContextWithAddr(ctx context.Context, addr string) (net.Conn, error) {
 	return d.DialContext(ctx, scheme, addr)
 }
 
-// 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.
+// DialContext creates proxied port-forwarded connections.
+// ctx is currently unused, but fulfils the type signature used by GRPC.
 func (d *Dialer) DialContext(_ context.Context, _ string, addr string) (net.Conn, error) {
 	proxyTransport, upgrader, err := spdy.RoundTripperFor(d.proxy.KubeConfig)
 	if err != nil {
@@ -96,17 +92,24 @@ func (d *Dialer) DialContext(_ context.Context, _ string, addr string) (net.Conn
 
 	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")
 	}
 
-	// Port must match proxy.Port; request ID is always "0".
+	// Create the headers.
 	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")
 
-	// Error stream: required by protocol, closed immediately.
+	// Create the error stream.
 	headers.Set(corev1.StreamType, corev1.StreamTypeError)
 	errorStream, err := connection.CreateStream(headers)
 	if err != nil {
@@ -115,14 +118,18 @@ 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(),
 		})
 	}
 
-	// Data stream: switch type and create the I/O stream.
+	// Create the data stream.
+	//
+	// NOTE: Given that we're reusing the headers,
+	// we need to overwrite the stream type before creating it.
 	headers.Set(corev1.StreamType, corev1.StreamTypeData)
 	dataStream, err := connection.CreateStream(headers)
 	if err != nil {
@@ -132,17 +139,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 per-dial timeout (also applied to KubeConfig).
+// DialTimeout sets the timeout.
 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