Add optional backend dial timeout

Add a configurable backend dial timeout for frontend dial requests. The option defaults to 0 to preserve existing behavior, and operators can opt in to bound DIAL_REQ send and DIAL_RSP wait time.

Clean up pending dials on timeout and report backend_dial_timeout as a dial failure reason. Add an HTTP CONNECT regression test for a blocked backend DIAL_REQ send.

Signed-off-by: DH Kim <inerplat@gmail.com>

Author: inerplat

cmd/server/app/options/options.go

@@ -122,6 +122,8 @@ type ProxyRunOptions struct {
 	NeedsKubernetesClient bool
 	// Graceful shutdown timeout duration
 	GracefulShutdownTimeout time.Duration
+	// Backend dial timeout duration for frontend dial requests
+	BackendDialTimeout time.Duration
 }
 
 func (o *ProxyRunOptions) Flags() *pflag.FlagSet {
@@ -164,6 +166,7 @@ func (o *ProxyRunOptions) Flags() *pflag.FlagSet {
 	flags.StringVar(&o.LeaseNamespace, "lease-namespace", o.LeaseNamespace, "The namespace where lease objects are managed by the controller.")
 	flags.StringVar(&o.LeaseLabel, "lease-label", o.LeaseLabel, "The labels on which the lease objects are managed.")
 	flags.DurationVar(&o.GracefulShutdownTimeout, "graceful-shutdown-timeout", o.GracefulShutdownTimeout, "Timeout duration for graceful shutdown of the server. The server will wait for active connections to close before forcefully terminating. Set to 0 to disable graceful shutdown (default: 0).")
+	flags.DurationVar(&o.BackendDialTimeout, "backend-dial-timeout", o.BackendDialTimeout, "Timeout duration for sending DIAL_REQ packets to backend agent streams and waiting for DIAL_RSP. Set to 0 to disable timeout.")
 	flags.Bool("warn-on-channel-limit", true, "This behavior is now thread safe and always on. This flag will be removed in a future release.")
 	flags.MarkDeprecated("warn-on-channel-limit", "This behavior is now thread safe and always on. This flag will be removed in a future release.")
 
@@ -209,6 +212,7 @@ func (o *ProxyRunOptions) Print() {
 	klog.V(1).Infof("TLSMinVersion set to %q.\n", o.TLSMinVersion)
 	klog.V(1).Infof("XfrChannelSize set to %d.\n", o.XfrChannelSize)
 	klog.V(1).Infof("GracefulShutdownTimeout set to %v.\n", o.GracefulShutdownTimeout)
+	klog.V(1).Infof("BackendDialTimeout set to %v.\n", o.BackendDialTimeout)
 }
 
 func (o *ProxyRunOptions) Validate() error {
@@ -368,6 +372,9 @@ func (o *ProxyRunOptions) Validate() error {
 	if o.GracefulShutdownTimeout < 0 {
 		return fmt.Errorf("graceful-shutdown-timeout must be >= 0, got %v", o.GracefulShutdownTimeout)
 	}
+	if o.BackendDialTimeout < 0 {
+		return fmt.Errorf("backend-dial-timeout must be >= 0, got %v", o.BackendDialTimeout)
+	}
 
 	o.NeedsKubernetesClient = usingServiceAccountAuth || o.EnableLeaseController
 
@@ -414,6 +421,7 @@ func NewProxyRunOptions() *ProxyRunOptions {
 		LeaseNamespace:            "kube-system",
 		LeaseLabel:                "k8s-app=konnectivity-server",
 		GracefulShutdownTimeout:   0,
+		BackendDialTimeout:        0,
 	}
 	return &o
 }

cmd/server/app/options/options_test.go

@@ -66,6 +66,7 @@ func TestDefaultServerOptions(t *testing.T) {
 	assertDefaultValue(t, "XfrChannelSize", defaultServerOptions.XfrChannelSize, 10)
 	assertDefaultValue(t, "APIContentType", defaultServerOptions.APIContentType, "application/vnd.kubernetes.protobuf")
 	assertDefaultValue(t, "GracefulShutdownTimeout", defaultServerOptions.GracefulShutdownTimeout, 0*time.Second)
+	assertDefaultValue(t, "BackendDialTimeout", defaultServerOptions.BackendDialTimeout, 0*time.Second)
 
 }
 
@@ -220,6 +221,21 @@ func TestValidate(t *testing.T) {
 			value:    30 * time.Second,
 			expected: nil,
 		},
+		"NegativeBackendDialTimeout": {
+			field:    "BackendDialTimeout",
+			value:    -1 * time.Second,
+			expected: fmt.Errorf("backend-dial-timeout must be >= 0, got -1s"),
+		},
+		"ZeroBackendDialTimeout": {
+			field:    "BackendDialTimeout",
+			value:    0 * time.Second,
+			expected: nil,
+		},
+		"PositiveBackendDialTimeout": {
+			field:    "BackendDialTimeout",
+			value:    30 * time.Second,
+			expected: nil,
+		},
 	} {
 		t.Run(desc, func(t *testing.T) {
 			testServerOptions := NewProxyRunOptions()

cmd/server/app/server.go

@@ -140,6 +140,7 @@ func (p *Proxy) Run(o *options.ProxyRunOptions, stopCh <-chan struct{}) error {
 		return err
 	}
 	p.server = server.NewProxyServer(o.ServerID, ps, o.ServerCount, authOpt, o.XfrChannelSize)
+	p.server.SetBackendDialTimeout(o.BackendDialTimeout)
 
 	frontendStop, err := p.runFrontendServer(ctx, o, p.server)
 	if err != nil {

pkg/server/metrics/metrics.go

@@ -328,6 +328,7 @@ const (
 	DialFailureUnrecognizedResponse DialFailureReason = "unrecognized_response" // Dial repsonse received for unrecognozide dial ID.
 	DialFailureSendResponse         DialFailureReason = "send_rsp"              // Successful dial response from agent, but failed to send to frontend.
 	DialFailureBackendClose         DialFailureReason = "backend_close"         // Received a DIAL_CLS from the backend before the dial completed.
+	DialFailureBackendDialTimeout   DialFailureReason = "backend_dial_timeout"  // Timed out sending DIAL_REQ to or waiting for DIAL_RSP from the backend.
 	DialFailureFrontendClose        DialFailureReason = "frontend_close"        // Received a DIAL_CLS from the frontend before the dial completed.
 )
 

pkg/server/server.go

@@ -90,6 +90,10 @@ const (
 	ModeHTTPConnect = "http-connect"
 )
 
+const defaultBackendDialTimeout = 0
+
+var errBackendDialTimeout = errors.New("timed out waiting for backend dial")
+
 type ProxyClientConnection struct {
 	Mode        string
 	HTTP        io.ReadWriter
@@ -274,6 +278,8 @@ type ProxyServer struct {
 	// TODO: move strategies into BackendStorage
 	proxyStrategies []proxystrategies.ProxyStrategy
 	xfrChannelSize  int
+
+	backendDialTimeout time.Duration
 }
 
 // AgentTokenAuthenticationOptions contains list of parameters required for agent token based authentication
@@ -317,6 +323,64 @@ func (s *ProxyServer) getBackend(reqHost string) (*Backend, error) {
 	return nil, &ErrNotFound{}
 }
 
+func (s *ProxyServer) sendDialRequestToBackend(backend *Backend, pkt *client.Packet) error {
+	timeout := s.backendDialTimeout
+	if timeout <= 0 {
+		return backend.Send(pkt)
+	}
+
+	errCh := make(chan error, 1)
+	go func() {
+		errCh <- backend.Send(pkt)
+	}()
+
+	timer := time.NewTimer(timeout)
+	defer timer.Stop()
+
+	ctx := backend.Context()
+	select {
+	case err := <-errCh:
+		return err
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-timer.C:
+		return errBackendDialTimeout
+	}
+}
+
+func (s *ProxyServer) startPendingDialTimeout(random int64, backend *Backend, frontend *GrpcFrontend) {
+	timeout := s.backendDialTimeout
+	if timeout <= 0 {
+		return
+	}
+
+	go func() {
+		timer := time.NewTimer(timeout)
+		defer timer.Stop()
+
+		select {
+		case <-timer.C:
+			if s.PendingDial.Remove(random) == nil {
+				return
+			}
+			metrics.Metrics.ObserveDialFailure(metrics.DialFailureBackendDialTimeout)
+			resp := &client.Packet{
+				Type: client.PacketType_DIAL_RSP,
+				Payload: &client.Packet_DialResponse{
+					DialResponse: &client.DialResponse{
+						Random: random,
+						Error:  errBackendDialTimeout.Error(),
+					},
+				},
+			}
+			if err := frontend.Send(resp); err != nil {
+				klog.V(5).InfoS("Failed to send DIAL_RSP for backend dial timeout", "error", err, "dialID", random)
+			}
+		case <-backend.Context().Done():
+		}
+	}()
+}
+
 func (s *ProxyServer) addBackend(backend *Backend) {
 	// TODO: refactor BackendStorage to acquire lock once, not up to 3 times.
 	for _, bm := range s.BackendManagers {
@@ -455,12 +519,17 @@ func NewProxyServer(serverID string, proxyStrategies []proxystrategies.ProxyStra
 		BackendManagers:            bms,
 		AgentAuthenticationOptions: agentAuthenticationOptions,
 		// use the first backend-manager as the Readiness Manager
-		Readiness:       bms[0],
-		proxyStrategies: proxyStrategies,
-		xfrChannelSize:  channelSize,
+		Readiness:          bms[0],
+		proxyStrategies:    proxyStrategies,
+		xfrChannelSize:     channelSize,
+		backendDialTimeout: defaultBackendDialTimeout,
 	}
 }
 
+func (s *ProxyServer) SetBackendDialTimeout(timeout time.Duration) {
+	s.backendDialTimeout = timeout
+}
+
 // Proxy handles incoming streams from gRPC frontend.
 func (s *ProxyServer) Proxy(stream client.ProxyService_ProxyServer) error {
 	metrics.Metrics.ConnectionInc(metrics.Proxy)
@@ -613,11 +682,31 @@ func (s *ProxyServer) serveRecvFrontend(frontend *GrpcFrontend, recvCh <-chan *c
 					backend:     backend,
 					dialAddress: address,
 				})
-			if err := backend.Send(pkt); err != nil {
+			if err := s.sendDialRequestToBackend(backend, pkt); err != nil {
 				klog.ErrorS(err, "DIAL_REQ to Backend failed", "dialID", random)
-			} else {
-				klog.V(5).InfoS("DIAL_REQ sent to backend", "dialID", random)
+				if s.PendingDial.Remove(random) != nil {
+					reason := metrics.DialFailureBackendClose
+					if errors.Is(err, errBackendDialTimeout) {
+						reason = metrics.DialFailureBackendDialTimeout
+					}
+					metrics.Metrics.ObserveDialFailure(reason)
+				}
+				resp := &client.Packet{
+					Type: client.PacketType_DIAL_RSP,
+					Payload: &client.Packet_DialResponse{
+						DialResponse: &client.DialResponse{
+							Random: random,
+							Error:  err.Error(),
+						},
+					},
+				}
+				if err := frontend.Send(resp); err != nil {
+					klog.V(5).InfoS("Failed to send DIAL_RSP for backend send failure", "error", err, "dialID", random)
+				}
+				return
 			}
+			klog.V(5).InfoS("DIAL_REQ sent to backend", "dialID", random)
+			s.startPendingDialTimeout(random, backend, frontend)
 
 		case client.PacketType_CLOSE_REQ:
 			connID := pkt.GetCloseRequest().ConnectID

pkg/server/server_test.go

@@ -17,9 +17,14 @@ limitations under the License.
 package server
 
 import (
+	"bufio"
 	"context"
 	"fmt"
 	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
 	"reflect"
 	"sync"
 	"testing"
@@ -219,6 +224,95 @@ func TestRemovePendingDialForStream(t *testing.T) {
 	}
 }
 
+func TestHTTPConnectTunnelBlockedBackendDialSendTimesOutAndCleansPendingDial(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	metrics.Metrics.Reset()
+
+	proxyServer := NewProxyServer(uuid.New().String(), []proxystrategies.ProxyStrategy{proxystrategies.ProxyStrategyDefault}, 1, &AgentTokenAuthenticationOptions{}, xfrChannelSize)
+	proxyServer.SetBackendDialTimeout(500 * time.Millisecond)
+
+	agentConn, _ := prepareAgentConnMD(t, ctrl, proxyServer, nil)
+	sendStarted := make(chan *client.Packet, 1)
+	releaseSend := make(chan struct{})
+	sendReleased := make(chan struct{})
+	agentConn.EXPECT().Send(gomock.AssignableToTypeOf(&client.Packet{})).DoAndReturn(func(pkt *client.Packet) error {
+		sendStarted <- pkt
+		<-releaseSend
+		close(sendReleased)
+		return nil
+	}).Times(1)
+
+	front := httptest.NewServer(&Tunnel{Server: proxyServer})
+	defer front.Close()
+
+	frontURL, err := url.Parse(front.URL)
+	if err != nil {
+		t.Fatalf("failed to parse front URL: %v", err)
+	}
+	conn, err := net.Dial("tcp", frontURL.Host)
+	if err != nil {
+		t.Fatalf("failed to connect to HTTP CONNECT front: %v", err)
+	}
+	defer conn.Close()
+
+	if _, err := fmt.Fprintf(conn, "CONNECT 127.0.0.1:443 HTTP/1.1\r\nHost: 127.0.0.1:443\r\n\r\n"); err != nil {
+		t.Fatalf("failed to write CONNECT request: %v", err)
+	}
+
+	var sent *client.Packet
+	select {
+	case sent = <-sendStarted:
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for backend DIAL_REQ send to start")
+	}
+	if sent.Type != client.PacketType_DIAL_REQ {
+		t.Fatalf("expected DIAL_REQ to backend, got %v", sent.Type)
+	}
+	if got := pendingDialCount(proxyServer); got != 1 {
+		t.Fatalf("expected one pending dial while backend Send is blocked, got %d", got)
+	}
+
+	respCh := make(chan struct {
+		resp *http.Response
+		err  error
+	}, 1)
+	go func() {
+		resp, err := http.ReadResponse(bufio.NewReader(conn), nil)
+		respCh <- struct {
+			resp *http.Response
+			err  error
+		}{resp: resp, err: err}
+	}()
+
+	var resp *http.Response
+	select {
+	case got := <-respCh:
+		if got.err != nil {
+			t.Fatalf("failed to read CONNECT response: %v", got.err)
+		}
+		resp = got.resp
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for backend dial timeout response")
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusGatewayTimeout {
+		t.Fatalf("expected %d for blocked backend dial send, got %s", http.StatusGatewayTimeout, resp.Status)
+	}
+	if got := pendingDialCount(proxyServer); got != 0 {
+		t.Fatalf("expected pending dial to be cleaned after backend dial timeout, got %d", got)
+	}
+
+	close(releaseSend)
+	select {
+	case <-sendReleased:
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for blocked backend Send goroutine to exit")
+	}
+}
+
 func TestAddRemoveFrontends(t *testing.T) {
 	agent1ConnID1 := new(ProxyClientConnection)
 	agent1ConnID2 := new(ProxyClientConnection)
@@ -950,6 +1044,12 @@ func assertEstablishedConnsMetric(t testing.TB, expect int) {
 	}
 }
 
+func pendingDialCount(p *ProxyServer) int {
+	p.PendingDial.mu.RLock()
+	defer p.PendingDial.mu.RUnlock()
+	return len(p.PendingDial.pendingDial)
+}
+
 func assertReadyBackendsMetric(t testing.TB, expect int) {
 	t.Helper()
 	if err := metricstest.DefaultTester.ExpectServerReadyBackends(expect); err != nil {