pkg/proxy: add flag to configure udp conntrack timeouts

Signed-off-by: Daman Arora <aroradaman@gmail.com>

cmd/kube-proxy/app/conntrack.go

@@ -37,8 +37,12 @@ type Conntracker interface {
 	SetMax(max int) error
 	// SetTCPEstablishedTimeout adjusts nf_conntrack_tcp_timeout_established.
 	SetTCPEstablishedTimeout(seconds int) error
-	// SetTCPCloseWaitTimeout nf_conntrack_tcp_timeout_close_wait.
+	// SetTCPCloseWaitTimeout adjusts nf_conntrack_tcp_timeout_close_wait.
 	SetTCPCloseWaitTimeout(seconds int) error
+	// SetUDPTimeout adjusts nf_conntrack_udp_timeout.
+	SetUDPTimeout(seconds int) error
+	// SetUDPStreamTimeout adjusts nf_conntrack_udp_timeout_stream.
+	SetUDPStreamTimeout(seconds int) error
 }
 
 type realConntracker struct{}
@@ -92,6 +96,14 @@ func (rct realConntracker) SetTCPCloseWaitTimeout(seconds int) error {
 	return rct.setIntSysCtl("nf_conntrack_tcp_timeout_close_wait", seconds)
 }
 
+func (rct realConntracker) SetUDPTimeout(seconds int) error {
+	return rct.setIntSysCtl("nf_conntrack_udp_timeout", seconds)
+}
+
+func (rct realConntracker) SetUDPStreamTimeout(seconds int) error {
+	return rct.setIntSysCtl("nf_conntrack_udp_timeout_stream", seconds)
+}
+
 func (realConntracker) setIntSysCtl(name string, value int) error {
 	entry := "net/netfilter/" + name
 

cmd/kube-proxy/app/server.go

@@ -199,6 +199,9 @@ func (o *Options) AddFlags(fs *pflag.FlagSet) {
 		&o.config.Conntrack.TCPCloseWaitTimeout.Duration, "conntrack-tcp-timeout-close-wait",
 		o.config.Conntrack.TCPCloseWaitTimeout.Duration,
 		"NAT timeout for TCP connections in the CLOSE_WAIT state")
+	fs.DurationVar(&o.config.Conntrack.UDPTimeout.Duration, "conntrack-udp-timeout", o.config.Conntrack.UDPTimeout.Duration, "Idle timeout for UNREPLIED UDP connections (0 to leave as-is)")
+	fs.DurationVar(&o.config.Conntrack.UDPStreamTimeout.Duration, "conntrack-udp-timeout-stream", o.config.Conntrack.UDPStreamTimeout.Duration, "Idle timeout for ASSURED UDP connections (0 to leave as-is)")
+
 	fs.DurationVar(&o.config.ConfigSyncPeriod.Duration, "config-sync-period", o.config.ConfigSyncPeriod.Duration, "How often configuration from the apiserver is refreshed.  Must be greater than 0.")
 
 	fs.BoolVar(&o.config.IPVS.StrictARP, "ipvs-strict-arp", o.config.IPVS.StrictARP, "Enable strict ARP by setting arp_ignore to 1 and arp_announce to 2")

cmd/kube-proxy/app/server_others.go

@@ -329,6 +329,20 @@ func (s *ProxyServer) setupConntrack() error {
 		}
 	}
 
+	if s.Config.Conntrack.UDPTimeout.Duration > 0 {
+		timeout := int(s.Config.Conntrack.UDPTimeout.Duration / time.Second)
+		if err := ct.SetUDPTimeout(timeout); err != nil {
+			return err
+		}
+	}
+
+	if s.Config.Conntrack.UDPStreamTimeout.Duration > 0 {
+		timeout := int(s.Config.Conntrack.UDPStreamTimeout.Duration / time.Second)
+		if err := ct.SetUDPStreamTimeout(timeout); err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 

pkg/generated/openapi/zz_generated.openapi.go

@@ -53539,8 +53539,22 @@ func schema_k8sio_kube_proxy_config_v1alpha1_KubeProxyConntrackConfiguration(ref
 							Ref:         ref("k8s.io/apimachinery/pkg/apis/meta/v1.Duration"),
 						},
 					},
+					"udpTimeout": {
+						SchemaProps: spec.SchemaProps{
+							Description: "udpTimeout is how long an idle UDP conntrack entry in UNREPLIED state will remain in the conntrack table (e.g. '30s'). Must be greater than 0 to set.",
+							Default:     0,
+							Ref:         ref("k8s.io/apimachinery/pkg/apis/meta/v1.Duration"),
+						},
+					},
+					"udpStreamTimeout": {
+						SchemaProps: spec.SchemaProps{
+							Description: "udpStreamTimeout is how long an idle UDP conntrack entry in ASSURED state will remain in the conntrack table (e.g. '300s'). Must be greater than 0 to set.",
+							Default:     0,
+							Ref:         ref("k8s.io/apimachinery/pkg/apis/meta/v1.Duration"),
+						},
+					},
 				},
-				Required: []string{"maxPerCore", "min", "tcpEstablishedTimeout", "tcpCloseWaitTimeout"},
+				Required: []string{"maxPerCore", "min", "tcpEstablishedTimeout", "tcpCloseWaitTimeout", "udpTimeout", "udpStreamTimeout"},
 			},
 		},
 		Dependencies: []string{

pkg/proxy/apis/config/scheme/testdata/KubeProxyConfiguration/after/v1alpha1.yaml

@@ -14,6 +14,8 @@ conntrack:
   min: 131072
   tcpCloseWaitTimeout: 1h0m0s
   tcpEstablishedTimeout: 24h0m0s
+  udpStreamTimeout: 0s
+  udpTimeout: 0s
 detectLocal:
   bridgeInterface: ""
   interfaceNamePrefix: ""

pkg/proxy/apis/config/scheme/testdata/KubeProxyConfiguration/roundtrip/default/v1alpha1.yaml

@@ -14,6 +14,8 @@ conntrack:
   min: 131072
   tcpCloseWaitTimeout: 1h0m0s
   tcpEstablishedTimeout: 24h0m0s
+  udpStreamTimeout: 0s
+  udpTimeout: 0s
 detectLocal:
   bridgeInterface: ""
   interfaceNamePrefix: ""

pkg/proxy/apis/config/types.go

@@ -89,6 +89,14 @@ type KubeProxyConntrackConfiguration struct {
 	// in CLOSE_WAIT state will remain in the conntrack
 	// table. (e.g. '60s'). Must be greater than 0 to set.
 	TCPCloseWaitTimeout *metav1.Duration
+	// udpTimeout is how long an idle UDP conntrack entry in
+	// UNREPLIED state will remain in the conntrack table
+	// (e.g. '30s'). Must be greater than 0 to set.
+	UDPTimeout metav1.Duration
+	// udpStreamTimeout is how long an idle UDP conntrack entry in
+	// ASSURED state will remain in the conntrack table
+	// (e.g. '300s'). Must be greater than 0 to set.
+	UDPStreamTimeout metav1.Duration
 }
 
 // KubeProxyWinkernelConfiguration contains Windows/HNS settings for

pkg/proxy/apis/config/v1alpha1/zz_generated.conversion.go

@@ -215,6 +215,8 @@ func autoConvert_v1alpha1_KubeProxyConntrackConfiguration_To_config_KubeProxyCon
 	out.Min = (*int32)(unsafe.Pointer(in.Min))
 	out.TCPEstablishedTimeout = (*v1.Duration)(unsafe.Pointer(in.TCPEstablishedTimeout))
 	out.TCPCloseWaitTimeout = (*v1.Duration)(unsafe.Pointer(in.TCPCloseWaitTimeout))
+	out.UDPTimeout = in.UDPTimeout
+	out.UDPStreamTimeout = in.UDPStreamTimeout
 	return nil
 }
 
@@ -228,6 +230,8 @@ func autoConvert_config_KubeProxyConntrackConfiguration_To_v1alpha1_KubeProxyCon
 	out.Min = (*int32)(unsafe.Pointer(in.Min))
 	out.TCPEstablishedTimeout = (*v1.Duration)(unsafe.Pointer(in.TCPEstablishedTimeout))
 	out.TCPCloseWaitTimeout = (*v1.Duration)(unsafe.Pointer(in.TCPCloseWaitTimeout))
+	out.UDPTimeout = in.UDPTimeout
+	out.UDPStreamTimeout = in.UDPStreamTimeout
 	return nil
 }
 

pkg/proxy/apis/config/validation/validation.go

@@ -163,14 +163,24 @@ func validateKubeProxyConntrackConfiguration(config kubeproxyconfig.KubeProxyCon
 		allErrs = append(allErrs, field.Invalid(fldPath.Child("Min"), config.Min, "must be greater than or equal to 0"))
 	}
 
+	// config.TCPEstablishedTimeout has a default value, so can't be nil.
 	if config.TCPEstablishedTimeout.Duration < 0 {
 		allErrs = append(allErrs, field.Invalid(fldPath.Child("TCPEstablishedTimeout"), config.TCPEstablishedTimeout, "must be greater than or equal to 0"))
 	}
 
+	// config.TCPCloseWaitTimeout has a default value, so can't be nil.
 	if config.TCPCloseWaitTimeout.Duration < 0 {
 		allErrs = append(allErrs, field.Invalid(fldPath.Child("TCPCloseWaitTimeout"), config.TCPCloseWaitTimeout, "must be greater than or equal to 0"))
 	}
 
+	if config.UDPTimeout.Duration < 0 {
+		allErrs = append(allErrs, field.Invalid(fldPath.Child("UDPTimeout"), config.UDPTimeout, "must be greater than or equal to 0"))
+	}
+
+	if config.UDPStreamTimeout.Duration < 0 {
+		allErrs = append(allErrs, field.Invalid(fldPath.Child("UDPStreamTimeout"), config.UDPStreamTimeout, "must be greater than or equal to 0"))
+	}
+
 	return allErrs
 }
 

pkg/proxy/apis/config/validation/validation_test.go

@@ -727,6 +727,8 @@ func TestValidateKubeProxyConntrackConfiguration(t *testing.T) {
 				Min:                   pointer.Int32(1),
 				TCPEstablishedTimeout: &metav1.Duration{Duration: 5 * time.Second},
 				TCPCloseWaitTimeout:   &metav1.Duration{Duration: 5 * time.Second},
+				UDPTimeout:            metav1.Duration{Duration: 5 * time.Second},
+				UDPStreamTimeout:      metav1.Duration{Duration: 5 * time.Second},
 			},
 			expectedErrs: field.ErrorList{},
 		},
@@ -736,6 +738,8 @@ func TestValidateKubeProxyConntrackConfiguration(t *testing.T) {
 				Min:                   pointer.Int32(1),
 				TCPEstablishedTimeout: &metav1.Duration{Duration: 0 * time.Second},
 				TCPCloseWaitTimeout:   &metav1.Duration{Duration: 0 * time.Second},
+				UDPTimeout:            metav1.Duration{Duration: 0 * time.Second},
+				UDPStreamTimeout:      metav1.Duration{Duration: 0 * time.Second},
 			},
 			expectedErrs: field.ErrorList{},
 		},
@@ -745,6 +749,8 @@ func TestValidateKubeProxyConntrackConfiguration(t *testing.T) {
 				Min:                   pointer.Int32(1),
 				TCPEstablishedTimeout: &metav1.Duration{Duration: 5 * time.Second},
 				TCPCloseWaitTimeout:   &metav1.Duration{Duration: 5 * time.Second},
+				UDPTimeout:            metav1.Duration{Duration: 5 * time.Second},
+				UDPStreamTimeout:      metav1.Duration{Duration: 5 * time.Second},
 			},
 			expectedErrs: field.ErrorList{field.Invalid(newPath.Child("KubeConntrackConfiguration.MaxPerCore"), -1, "must be greater than or equal to 0")},
 		},
@@ -754,27 +760,55 @@ func TestValidateKubeProxyConntrackConfiguration(t *testing.T) {
 				Min:                   pointer.Int32(-1),
 				TCPEstablishedTimeout: &metav1.Duration{Duration: 5 * time.Second},
 				TCPCloseWaitTimeout:   &metav1.Duration{Duration: 5 * time.Second},
+				UDPTimeout:            metav1.Duration{Duration: 5 * time.Second},
+				UDPStreamTimeout:      metav1.Duration{Duration: 5 * time.Second},
 			},
 			expectedErrs: field.ErrorList{field.Invalid(newPath.Child("KubeConntrackConfiguration.Min"), -1, "must be greater than or equal to 0")},
 		},
-		"invalid EstablishedTimeout < 0": {
+		"invalid TCPEstablishedTimeout < 0": {
 			config: kubeproxyconfig.KubeProxyConntrackConfiguration{
 				MaxPerCore:            pointer.Int32(1),
 				Min:                   pointer.Int32(1),
 				TCPEstablishedTimeout: &metav1.Duration{Duration: -5 * time.Second},
 				TCPCloseWaitTimeout:   &metav1.Duration{Duration: 5 * time.Second},
+				UDPTimeout:            metav1.Duration{Duration: 5 * time.Second},
+				UDPStreamTimeout:      metav1.Duration{Duration: 5 * time.Second},
 			},
 			expectedErrs: field.ErrorList{field.Invalid(newPath.Child("KubeConntrackConfiguration.TCPEstablishedTimeout"), metav1.Duration{Duration: -5 * time.Second}, "must be greater than or equal to 0")},
 		},
-		"invalid CloseWaitTimeout < 0": {
+		"invalid TCPCloseWaitTimeout < 0": {
 			config: kubeproxyconfig.KubeProxyConntrackConfiguration{
 				MaxPerCore:            pointer.Int32(1),
 				Min:                   pointer.Int32(1),
 				TCPEstablishedTimeout: &metav1.Duration{Duration: 5 * time.Second},
 				TCPCloseWaitTimeout:   &metav1.Duration{Duration: -5 * time.Second},
+				UDPTimeout:            metav1.Duration{Duration: 5 * time.Second},
+				UDPStreamTimeout:      metav1.Duration{Duration: 5 * time.Second},
 			},
 			expectedErrs: field.ErrorList{field.Invalid(newPath.Child("KubeConntrackConfiguration.TCPCloseWaitTimeout"), metav1.Duration{Duration: -5 * time.Second}, "must be greater than or equal to 0")},
 		},
+		"invalid UDPTimeout < 0": {
+			config: kubeproxyconfig.KubeProxyConntrackConfiguration{
+				MaxPerCore:            pointer.Int32(1),
+				Min:                   pointer.Int32(1),
+				TCPEstablishedTimeout: &metav1.Duration{Duration: 5 * time.Second},
+				TCPCloseWaitTimeout:   &metav1.Duration{Duration: 5 * time.Second},
+				UDPTimeout:            metav1.Duration{Duration: -5 * time.Second},
+				UDPStreamTimeout:      metav1.Duration{Duration: 5 * time.Second},
+			},
+			expectedErrs: field.ErrorList{field.Invalid(newPath.Child("KubeConntrackConfiguration.UDPTimeout"), metav1.Duration{Duration: -5 * time.Second}, "must be greater than or equal to 0")},
+		},
+		"invalid UDPStreamTimeout < 0": {
+			config: kubeproxyconfig.KubeProxyConntrackConfiguration{
+				MaxPerCore:            pointer.Int32(1),
+				Min:                   pointer.Int32(1),
+				TCPEstablishedTimeout: &metav1.Duration{Duration: 5 * time.Second},
+				TCPCloseWaitTimeout:   &metav1.Duration{Duration: 5 * time.Second},
+				UDPTimeout:            metav1.Duration{Duration: 5 * time.Second},
+				UDPStreamTimeout:      metav1.Duration{Duration: -5 * time.Second},
+			},
+			expectedErrs: field.ErrorList{field.Invalid(newPath.Child("KubeConntrackConfiguration.UDPStreamTimeout"), metav1.Duration{Duration: -5 * time.Second}, "must be greater than or equal to 0")},
+		},
 	}
 
 	for _, testCase := range testCases {

pkg/proxy/apis/config/zz_generated.deepcopy.go

@@ -137,6 +137,8 @@ func (in *KubeProxyConntrackConfiguration) DeepCopyInto(out *KubeProxyConntrackC
 		*out = new(v1.Duration)
 		**out = **in
 	}
+	out.UDPTimeout = in.UDPTimeout
+	out.UDPStreamTimeout = in.UDPStreamTimeout
 	return
 }
 

staging/src/k8s.io/kube-proxy/config/v1alpha1/types.go

@@ -85,6 +85,14 @@ type KubeProxyConntrackConfiguration struct {
 	// in CLOSE_WAIT state will remain in the conntrack
 	// table. (e.g. '60s'). Must be greater than 0 to set.
 	TCPCloseWaitTimeout *metav1.Duration `json:"tcpCloseWaitTimeout"`
+	// udpTimeout is how long an idle UDP conntrack entry in
+	// UNREPLIED state will remain in the conntrack table
+	// (e.g. '30s'). Must be greater than 0 to set.
+	UDPTimeout metav1.Duration `json:"udpTimeout"`
+	// udpStreamTimeout is how long an idle UDP conntrack entry in
+	// ASSURED state will remain in the conntrack table
+	// (e.g. '300s'). Must be greater than 0 to set.
+	UDPStreamTimeout metav1.Duration `json:"udpStreamTimeout"`
 }
 
 // KubeProxyWinkernelConfiguration contains Windows/HNS settings for

staging/src/k8s.io/kube-proxy/config/v1alpha1/zz_generated.deepcopy.go

@@ -115,6 +115,8 @@ func (in *KubeProxyConntrackConfiguration) DeepCopyInto(out *KubeProxyConntrackC
 		*out = new(v1.Duration)
 		**out = **in
 	}
+	out.UDPTimeout = in.UDPTimeout
+	out.UDPStreamTimeout = in.UDPStreamTimeout
 	return
 }