restore int value

Signed-off-by: Karol Szwaj <[email protected]>

Author: cnvergence

cmd/agent/app/options/options.go

@@ -79,7 +79,7 @@ type GrpcProxyAgentOptions struct {
 	WarnOnChannelLimit bool
 
 	SyncForever    bool
-	XfrChannelSize uint
+	XfrChannelSize int
 }
 
 func (o *GrpcProxyAgentOptions) ClientSetConfig(dialOptions ...grpc.DialOption) *agent.ClientSetConfig {
@@ -121,7 +121,7 @@ func (o *GrpcProxyAgentOptions) Flags() *pflag.FlagSet {
 	flags.StringVar(&o.AgentIdentifiers, "agent-identifiers", o.AgentIdentifiers, "Identifiers of the agent that will be used by the server when choosing agent. N.B. the list of identifiers must be in URL encoded format. e.g.,host=localhost&host=node1.mydomain.com&cidr=127.0.0.1/16&ipv4=1.2.3.4&ipv4=5.6.7.8&ipv6=:::::&default-route=true")
 	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.")
 	flags.BoolVar(&o.SyncForever, "sync-forever", o.SyncForever, "If true, the agent continues syncing, in order to support server count changes.")
-	flags.UintVar(&o.XfrChannelSize, "xfr-channel-size", 150, "Set the size of the channel")
+	flags.IntVar(&o.XfrChannelSize, "xfr-channel-size", 150, "Set the size of the channel for transferring data between the agent and the proxy server.")
 	return flags
 }
 

cmd/agent/app/options/options_test.go

@@ -50,7 +50,7 @@ func TestDefaultServerOptions(t *testing.T) {
 	assertDefaultValue(t, "ServiceAccountTokenPath", defaultAgentOptions.ServiceAccountTokenPath, "")
 	assertDefaultValue(t, "WarnOnChannelLimit", defaultAgentOptions.WarnOnChannelLimit, false)
 	assertDefaultValue(t, "SyncForever", defaultAgentOptions.SyncForever, false)
-	assertDefaultValue(t, "XfrChannelSize", defaultAgentOptions.XfrChannelSize, uint(150))
+	assertDefaultValue(t, "XfrChannelSize", defaultAgentOptions.XfrChannelSize, 150)
 }
 
 func assertDefaultValue(t *testing.T, fieldName string, actual, expected interface{}) {
@@ -148,9 +148,13 @@ func TestValidate(t *testing.T) {
 			expected: fmt.Errorf("if --enable-contention-profiling is set, --enable-profiling must also be set"),
 		},
 		"ZeroXfrChannelSize": {
-			fieldMap: map[string]interface{}{"XfrChannelSize": uint(0)},
+			fieldMap: map[string]interface{}{"XfrChannelSize": 0},
 			expected: fmt.Errorf("channel size 0 must be greater than 0"),
 		},
+		"NegativeXfrChannelSize": {
+			fieldMap: map[string]interface{}{"XfrChannelSize": -10},
+			expected: fmt.Errorf("channel size -10 must be greater than 0"),
+		},
 	} {
 		t.Run(desc, func(t *testing.T) {
 			testAgentOptions := NewGrpcProxyAgentOptions()
@@ -168,9 +172,6 @@ func TestValidate(t *testing.T) {
 				case reflect.Bool:
 					bvalue := value.(bool)
 					fv.SetBool(bvalue)
-				case reflect.Uint:
-					uvalue := value.(uint)
-					fv.SetUint(uint64(uvalue))
 				}
 			}
 			actual := testAgentOptions.Validate()

cmd/server/app/options/options.go

@@ -101,7 +101,7 @@ type ProxyRunOptions struct {
 	// NOTE that cipher suites are not configurable for TLS1.3,
 	// see: https://pkg.go.dev/crypto/tls#Config, so in that case, this option won't have any effect.
 	CipherSuites   []string
-	XfrChannelSize uint
+	XfrChannelSize int
 }
 
 func (o *ProxyRunOptions) Flags() *pflag.FlagSet {
@@ -137,7 +137,7 @@ func (o *ProxyRunOptions) Flags() *pflag.FlagSet {
 	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 an agent/tunnel, available strategies are: default, destHost, defaultRoute.")
 	flags.StringSliceVar(&o.CipherSuites, "cipher-suites", o.CipherSuites, "The comma separated list of allowed cipher suites. Has no effect on TLS1.3. Empty means allow default list.")
-	flags.UintVar(&o.XfrChannelSize, "xfr-channel-size", o.XfrChannelSize, "The size of the channel for transferring data between the proxy server and the agent.")
+	flags.IntVar(&o.XfrChannelSize, "xfr-channel-size", o.XfrChannelSize, "The size of the channel for transferring data between the proxy server and the agent.")
 
 	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.")

cmd/server/app/options/options_test.go

@@ -61,7 +61,7 @@ func TestDefaultServerOptions(t *testing.T) {
 	assertDefaultValue(t, "AuthenticationAudience", defaultServerOptions.AuthenticationAudience, "")
 	assertDefaultValue(t, "ProxyStrategies", defaultServerOptions.ProxyStrategies, "default")
 	assertDefaultValue(t, "CipherSuites", defaultServerOptions.CipherSuites, make([]string, 0))
-	assertDefaultValue(t, "XfrChannelSize", defaultServerOptions.XfrChannelSize, uint(10))
+	assertDefaultValue(t, "XfrChannelSize", defaultServerOptions.XfrChannelSize, 10)
 
 }
 
@@ -159,9 +159,14 @@ func TestValidate(t *testing.T) {
 		},
 		"ZeroXfrChannelSize": {
 			field:    "XfrChannelSize",
-			value:    uint(0),
+			value:    0,
 			expected: fmt.Errorf("channel size 0 must be greater than 0"),
 		},
+		"NegativeXfrChannelSize": {
+			field:    "XfrChannelSize",
+			value:    -10,
+			expected: fmt.Errorf("channel size -10 must be greater than 0"),
+		},
 	} {
 		t.Run(desc, func(t *testing.T) {
 			testServerOptions := NewProxyRunOptions()
@@ -178,9 +183,6 @@ func TestValidate(t *testing.T) {
 				case reflect.Int:
 					ivalue := tc.value.(int)
 					fv.SetInt(int64(ivalue))
-				case reflect.Uint:
-					uvalue := tc.value.(uint)
-					fv.SetUint(uint64(uvalue))
 				}
 			}
 			actual := testServerOptions.Validate()

pkg/agent/clientset.go

@@ -61,7 +61,7 @@ type ClientSet struct {
 	// by the server when choosing agent
 
 	warnOnChannelLimit bool
-	xfrChannelSize     uint
+	xfrChannelSize     int
 
 	syncForever bool // Continue syncing (support dynamic server count).
 }
@@ -142,7 +142,7 @@ type ClientSetConfig struct {
 	ServiceAccountTokenPath string
 	WarnOnChannelLimit      bool
 	SyncForever             bool
-	XfrChannelSize          uint
+	XfrChannelSize          int
 }
 
 func (cc *ClientSetConfig) NewAgentClientSet(drainCh, stopCh <-chan struct{}) *ClientSet {

pkg/server/server.go

@@ -216,7 +216,7 @@ type ProxyServer struct {
 
 	// TODO: move strategies into BackendStorage
 	proxyStrategies []ProxyStrategy
-	xfrChannelSize  uint
+	xfrChannelSize  int
 }
 
 // AgentTokenAuthenticationOptions contains list of parameters required for agent token based authentication
@@ -375,7 +375,7 @@ func (s *ProxyServer) removeEstablishedForStream(streamUID string) []*ProxyClien
 }
 
 // NewProxyServer creates a new ProxyServer instance
-func NewProxyServer(serverID string, proxyStrategies []ProxyStrategy, serverCount int, agentAuthenticationOptions *AgentTokenAuthenticationOptions, channelSize uint) *ProxyServer {
+func NewProxyServer(serverID string, proxyStrategies []ProxyStrategy, serverCount int, agentAuthenticationOptions *AgentTokenAuthenticationOptions, channelSize int) *ProxyServer {
 	var bms []BackendManager
 	for _, ps := range proxyStrategies {
 		switch ps {
@@ -417,7 +417,7 @@ func (s *ProxyServer) Proxy(stream client.ProxyService_ProxyServer) error {
 	streamUID := uuid.New().String()
 	klog.V(5).InfoS("Proxy request from client", "userAgent", userAgent, "serverID", s.serverID, "streamUID", streamUID)
 
-	recvCh := make(chan *client.Packet, int(s.xfrChannelSize))
+	recvCh := make(chan *client.Packet, s.xfrChannelSize)
 	stopCh := make(chan error, 1)
 
 	frontend := GrpcFrontend{