Add a flag to "hide" the channel full check race condition.

The check now has to be explicitly turned on.

Fixed tests to use warnOnChannelLimit false so they don't trigger the
race violation.

Author: cheftako

cmd/server/app/options/options.go

@@ -70,6 +70,13 @@ type ProxyRunOptions struct {
 	// backend within the destCIDR. if it still can't find any backend,
 	// it will use the default backend manager to choose a random backend.
 	ProxyStrategies string
+
+	// This controls if we attempt to push onto a "full" transfer channel.
+	// However checking that the transfer channel is full is not safe.
+	// It violates our race condition checking. Adding locks around a potentially
+	// blocking call has its own problems, so it cannot easily be made race condition safe.
+	// The check is an "unlocked" read but is still use at your own peril.
+	WarnOnChannelLimit bool
 }
 
 func (o *ProxyRunOptions) Flags() *pflag.FlagSet {
@@ -100,6 +107,7 @@ func (o *ProxyRunOptions) Flags() *pflag.FlagSet {
 	flags.IntVar(&o.KubeconfigBurst, "kubeconfig-burst", o.KubeconfigBurst, "Maximum client burst (proxy server uses this client to authenticate agent tokens).")
 	flags.StringVar(&o.AuthenticationAudience, "authentication-audience", o.AuthenticationAudience, "Expected agent's token authentication audience (used with agent-namespace, agent-service-account, kubeconfig).")
 	flags.StringVar(&o.ProxyStrategies, "proxy-strategies", o.ProxyStrategies, "The list of proxy strategies used by the server to pick a backend/tunnel, available strategies are: default, destHost.")
+	flags.BoolVar(&o.WarnOnChannelLimit, "warn-on-channel-limit", o.WarnOnChannelLimit, "Turns on a warning if the system is going to push to a full channel. The check involves an unsafe read.")
 	return flags
 }
 
@@ -130,6 +138,7 @@ func (o *ProxyRunOptions) Print() {
 	klog.V(1).Infof("KubeconfigQPS set to %f.\n", o.KubeconfigQPS)
 	klog.V(1).Infof("KubeconfigBurst set to %d.\n", o.KubeconfigBurst)
 	klog.V(1).Infof("ProxyStrategies set to %q.\n", o.ProxyStrategies)
+	klog.V(1).Infof("WarnOnChannelLimit set to %t.\n", o.WarnOnChannelLimit)
 }
 
 func (o *ProxyRunOptions) Validate() error {
@@ -290,6 +299,7 @@ func NewProxyRunOptions() *ProxyRunOptions {
 		KubeconfigBurst:           0,
 		AuthenticationAudience:    "",
 		ProxyStrategies:           "default",
+		WarnOnChannelLimit:        false,
 	}
 	return &o
 }

cmd/server/app/server.go

@@ -92,7 +92,7 @@ func (p *Proxy) run(o *options.ProxyRunOptions) error {
 	if err != nil {
 		return err
 	}
-	server := server.NewProxyServer(o.ServerID, ps, int(o.ServerCount), authOpt)
+	server := server.NewProxyServer(o.ServerID, ps, int(o.ServerCount), authOpt, o.WarnOnChannelLimit)
 
 	frontendStop, err := p.runFrontendServer(ctx, o, server)
 	if err != nil {

pkg/server/server.go

@@ -40,6 +40,8 @@ import (
 	"sigs.k8s.io/apiserver-network-proxy/proto/header"
 )
 
+const xfrChannelSize = 10
+
 type key int
 
 type ProxyClientConnection struct {
@@ -136,6 +138,9 @@ type ProxyServer struct {
 	serverID    string // unique ID of this server
 	serverCount int    // Number of proxy server instances, should be 1 unless it is a HA server.
 
+	// Allows a special debug flag which warns if we write to a full transfer channel
+	warnOnChannelLimit bool
+
 	// agent authentication
 	AgentAuthenticationOptions *AgentTokenAuthenticationOptions
 
@@ -321,7 +326,7 @@ func (s *ProxyServer) getFrontendsForBackendConn(agentID string, backend Backend
 }
 
 // NewProxyServer creates a new ProxyServer instance
-func NewProxyServer(serverID string, proxyStrategies []ProxyStrategy, serverCount int, agentAuthenticationOptions *AgentTokenAuthenticationOptions) *ProxyServer {
+func NewProxyServer(serverID string, proxyStrategies []ProxyStrategy, serverCount int, agentAuthenticationOptions *AgentTokenAuthenticationOptions, warnOnChannelLimit bool) *ProxyServer {
 	var bms []BackendManager
 	for _, ps := range proxyStrategies {
 		switch ps {
@@ -343,9 +348,10 @@ func NewProxyServer(serverID string, proxyStrategies []ProxyStrategy, serverCoun
 		serverCount:                serverCount,
 		BackendManagers:            bms,
 		AgentAuthenticationOptions: agentAuthenticationOptions,
-		// use the first backendmanager as the Readiness Manager
-		Readiness:       bms[0],
-		proxyStrategies: proxyStrategies,
+		// use the first backend-manager as the Readiness Manager
+		Readiness:          bms[0],
+		proxyStrategies:    proxyStrategies,
+		warnOnChannelLimit: warnOnChannelLimit,
 	}
 }
 
@@ -361,7 +367,7 @@ func (s *ProxyServer) Proxy(stream client.ProxyService_ProxyServer) error {
 	userAgent := md.Get(header.UserAgent)
 	klog.V(2).InfoS("proxy request from client", "userAgent", userAgent)
 
-	recvCh := make(chan *client.Packet, 10)
+	recvCh := make(chan *client.Packet, xfrChannelSize)
 	stopCh := make(chan error)
 
 	go s.serveRecvFrontend(stream, recvCh)
@@ -386,7 +392,7 @@ func (s *ProxyServer) Proxy(stream client.ProxyService_ProxyServer) error {
 				return
 			}
 
-			if len(recvCh) > 9 {
+			if s.warnOnChannelLimit && len(recvCh) >= xfrChannelSize {
 				klog.V(2).InfoS("Receive channel on Proxy is full", "userAgent", userAgent, "serverID", s.serverID)
 			}
 			recvCh <- in
@@ -640,7 +646,7 @@ func (s *ProxyServer) Connect(stream agent.AgentService_ConnectServer) error {
 	backend := s.addBackend(agentID, stream)
 	defer s.removeBackend(agentID, stream)
 
-	recvCh := make(chan *client.Packet, 10)
+	recvCh := make(chan *client.Packet, xfrChannelSize)
 
 	go s.serveRecvBackend(backend, stream, agentID, recvCh)
 
@@ -664,7 +670,7 @@ func (s *ProxyServer) Connect(stream agent.AgentService_ConnectServer) error {
 				return
 			}
 
-			if len(recvCh) > 9 {
+			if s.warnOnChannelLimit && len(recvCh) >= xfrChannelSize {
 				klog.V(2).InfoS("Receive channel on Connect is full", "agentID", agentID, "serverID", s.serverID)
 			}
 			recvCh <- in

pkg/server/server_test.go

@@ -164,7 +164,9 @@ func TestAgentTokenAuthenticationErrorsToken(t *testing.T) {
 				KubernetesClient:    kcs,
 				AgentNamespace:      tc.wantNamespace,
 				AgentServiceAccount: tc.wantServiceAccount,
-			})
+			},
+				false,
+			)
 
 			err := p.Connect(conn)
 			if tc.wantError {
@@ -187,15 +189,15 @@ func TestAddRemoveFrontends(t *testing.T) {
 	agent2ConnID2 := new(ProxyClientConnection)
 	agent3ConnID1 := new(ProxyClientConnection)
 
-	p := NewProxyServer("", []ProxyStrategy{ProxyStrategyDefault}, 1, nil)
+	p := NewProxyServer("", []ProxyStrategy{ProxyStrategyDefault}, 1, nil, false)
 	p.addFrontend("agent1", int64(1), agent1ConnID1)
 	p.removeFrontend("agent1", int64(1))
 	expectedFrontends := make(map[string]map[int64]*ProxyClientConnection)
 	if e, a := expectedFrontends, p.frontends; !reflect.DeepEqual(e, a) {
 		t.Errorf("expected %v, got %v", e, a)
 	}
 
-	p = NewProxyServer("", []ProxyStrategy{ProxyStrategyDefault}, 1, nil)
+	p = NewProxyServer("", []ProxyStrategy{ProxyStrategyDefault}, 1, nil, false)
 	p.addFrontend("agent1", int64(1), agent1ConnID1)
 	p.addFrontend("agent1", int64(2), agent1ConnID2)
 	p.addFrontend("agent2", int64(1), agent2ConnID1)

tests/proxy_test.go

@@ -356,7 +356,7 @@ func runGRPCProxyServerWithServerCount(serverCount int) (proxy, *server.ProxySer
 	var err error
 	var lis, lis2 net.Listener
 
-	server := server.NewProxyServer(uuid.New().String(), []server.ProxyStrategy{server.ProxyStrategyDefault}, serverCount, &server.AgentTokenAuthenticationOptions{})
+	server := server.NewProxyServer(uuid.New().String(), []server.ProxyStrategy{server.ProxyStrategyDefault}, serverCount, &server.AgentTokenAuthenticationOptions{}, false)
 	grpcServer := grpc.NewServer()
 	agentServer := grpc.NewServer()
 	cleanup := func() {
@@ -395,7 +395,7 @@ func runGRPCProxyServerWithServerCount(serverCount int) (proxy, *server.ProxySer
 func runHTTPConnProxyServer() (proxy, func(), error) {
 	ctx := context.Background()
 	var proxy proxy
-	s := server.NewProxyServer(uuid.New().String(), []server.ProxyStrategy{server.ProxyStrategyDefault}, 0, &server.AgentTokenAuthenticationOptions{})
+	s := server.NewProxyServer(uuid.New().String(), []server.ProxyStrategy{server.ProxyStrategyDefault}, 0, &server.AgentTokenAuthenticationOptions{}, false)
 	agentServer := grpc.NewServer()
 
 	agentproto.RegisterAgentServiceServer(agentServer, s)