github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2026-07-17 11:28:58 +00:00
Code Issues Projects Releases Wiki Activity

Move setupConntrack from server_linux.go to conntrack.go

Browse Source
This commit is contained in:
Dan Winship
2025-11-26 13:39:14 -05:00
parent 905492223b
commit d8a481a696
4 changed files with 339 additions and 311 deletions
Download Patch File Download Diff File Expand all files Collapse all files

107
cmd/kube-proxy/app/conntrack.go
View File

@@ -1,3 +1,6 @@
//go:build linux
// +build linux
/*
Copyright 2015 The Kubernetes Authors.
@@ -19,18 +22,24 @@ package app
import (
"context"
"os"
"runtime"
"strconv"
"strings"
"time"
"github.com/google/cadvisor/machine"
"github.com/google/cadvisor/utils/sysfs"
"k8s.io/component-helpers/node/util/sysctl"
"k8s.io/klog/v2"
proxyconfigapi "k8s.io/kubernetes/pkg/proxy/apis/config"
)
// Conntracker is an interface to the global sysctl. Descriptions of the various
// sysctl fields can be found here:
// conntrackConfigurer is a mockable interface for setting conntrack sysctls.
//
// Descriptions of the various sysctl fields can be found here:
// https://www.kernel.org/doc/Documentation/networking/nf_conntrack-sysctl.txt
type Conntracker interface {
type conntrackConfigurer interface {
// SetMax adjusts nf_conntrack_max.
SetMax(ctx context.Context, max int) error
// SetTCPEstablishedTimeout adjusts nf_conntrack_tcp_timeout_established.
@@ -45,10 +54,86 @@ type Conntracker interface {
SetUDPStreamTimeout(ctx context.Context, seconds int) error
}
type realConntracker struct {
func setupConntrack(ctx context.Context, ct conntrackConfigurer, config *proxyconfigapi.KubeProxyConntrackConfiguration) error {
max, err := getConntrackMax(ctx, config)
if err != nil {
return err
}
if max > 0 {
err := ct.SetMax(ctx, max)
if err != nil {
return err
}
}
if config.TCPEstablishedTimeout != nil && config.TCPEstablishedTimeout.Duration > 0 {
timeout := int(config.TCPEstablishedTimeout.Duration / time.Second)
if err := ct.SetTCPEstablishedTimeout(ctx, timeout); err != nil {
return err
}
}
if config.TCPCloseWaitTimeout != nil && config.TCPCloseWaitTimeout.Duration > 0 {
timeout := int(config.TCPCloseWaitTimeout.Duration / time.Second)
if err := ct.SetTCPCloseWaitTimeout(ctx, timeout); err != nil {
return err
}
}
if config.TCPBeLiberal {
if err := ct.SetTCPBeLiberal(ctx, 1); err != nil {
return err
}
}
if config.UDPTimeout.Duration > 0 {
timeout := int(config.UDPTimeout.Duration / time.Second)
if err := ct.SetUDPTimeout(ctx, timeout); err != nil {
return err
}
}
if config.UDPStreamTimeout.Duration > 0 {
timeout := int(config.UDPStreamTimeout.Duration / time.Second)
if err := ct.SetUDPStreamTimeout(ctx, timeout); err != nil {
return err
}
}
return nil
}
func (rct realConntracker) SetMax(ctx context.Context, max int) error {
func getConntrackMax(ctx context.Context, config *proxyconfigapi.KubeProxyConntrackConfiguration) (int, error) {
logger := klog.FromContext(ctx)
if config.MaxPerCore != nil && *config.MaxPerCore > 0 {
floor := 0
if config.Min != nil {
floor = int(*config.Min)
}
scaled := int(*config.MaxPerCore) * detectNumCPU()
if scaled > floor {
logger.V(3).Info("GetConntrackMax: using scaled conntrack-max-per-core")
return scaled, nil
}
logger.V(3).Info("GetConntrackMax: using conntrack-min")
return floor, nil
}
return 0, nil
}
func detectNumCPU() int {
// try get numCPU from /sys firstly due to a known issue (https://github.com/kubernetes/kubernetes/issues/99225)
_, numCPU, err := machine.GetTopology(sysfs.NewRealSysFs())
if err != nil || numCPU < 1 {
return runtime.NumCPU()
}
return numCPU
}
type realConntrackConfigurer struct {
}
func (rct realConntrackConfigurer) SetMax(ctx context.Context, max int) error {
logger := klog.FromContext(ctx)
logger.Info("Setting nf_conntrack_max", "nfConntrackMax", max)
if err := rct.setIntSysCtl(ctx, "nf_conntrack_max", max); err != nil {
@@ -68,27 +153,27 @@ func (rct realConntracker) SetMax(ctx context.Context, max int) error {
return writeIntStringFile("/sys/module/nf_conntrack/parameters/hashsize", max/4)
}
func (rct realConntracker) SetTCPEstablishedTimeout(ctx context.Context, seconds int) error {
func (rct realConntrackConfigurer) SetTCPEstablishedTimeout(ctx context.Context, seconds int) error {
return rct.setIntSysCtl(ctx, "nf_conntrack_tcp_timeout_established", seconds)
}
func (rct realConntracker) SetTCPCloseWaitTimeout(ctx context.Context, seconds int) error {
func (rct realConntrackConfigurer) SetTCPCloseWaitTimeout(ctx context.Context, seconds int) error {
return rct.setIntSysCtl(ctx, "nf_conntrack_tcp_timeout_close_wait", seconds)
}
func (rct realConntracker) SetTCPBeLiberal(ctx context.Context, value int) error {
func (rct realConntrackConfigurer) SetTCPBeLiberal(ctx context.Context, value int) error {
return rct.setIntSysCtl(ctx, "nf_conntrack_tcp_be_liberal", value)
}
func (rct realConntracker) SetUDPTimeout(ctx context.Context, seconds int) error {
func (rct realConntrackConfigurer) SetUDPTimeout(ctx context.Context, seconds int) error {
return rct.setIntSysCtl(ctx, "nf_conntrack_udp_timeout", seconds)
}
func (rct realConntracker) SetUDPStreamTimeout(ctx context.Context, seconds int) error {
func (rct realConntrackConfigurer) SetUDPStreamTimeout(ctx context.Context, seconds int) error {
return rct.setIntSysCtl(ctx, "nf_conntrack_udp_timeout_stream", seconds)
}
func (rct realConntracker) setIntSysCtl(ctx context.Context, name string, value int) error {
func (rct realConntrackConfigurer) setIntSysCtl(ctx context.Context, name string, value int) error {
logger := klog.FromContext(ctx)
entry := "net/netfilter/" + name

241
cmd/kube-proxy/app/conntrack_test.go Normal file
View File

@@ -0,0 +1,241 @@
//go:build linux
// +build linux
/*
Copyright 2025 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package app
import (
"context"
"errors"
"fmt"
"runtime"
"strings"
"testing"
"time"
"github.com/google/go-cmp/cmp"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
proxyconfigapi "k8s.io/kubernetes/pkg/proxy/apis/config"
"k8s.io/kubernetes/test/utils/ktesting"
"k8s.io/utils/ptr"
)
func TestGetConntrackMax(t *testing.T) {
ncores := runtime.NumCPU()
testCases := []struct {
min int32
maxPerCore int32
expected int
err string
}{
{
expected: 0,
},
{
maxPerCore: 67890, // use this if Max is 0
min: 1, // avoid 0 default
expected: 67890 * ncores,
},
{
maxPerCore: 1, // ensure that Min is considered
min: 123456,
expected: 123456,
},
{
maxPerCore: 0, // leave system setting
min: 123456,
expected: 0,
},
}
for i, tc := range testCases {
cfg := proxyconfigapi.KubeProxyConntrackConfiguration{
Min: ptr.To(tc.min),
MaxPerCore: ptr.To(tc.maxPerCore),
}
_, ctx := ktesting.NewTestContext(t)
x, e := getConntrackMax(ctx, &cfg)
if e != nil {
if tc.err == "" {
t.Errorf("[%d] unexpected error: %v", i, e)
} else if !strings.Contains(e.Error(), tc.err) {
t.Errorf("[%d] expected an error containing %q: %v", i, tc.err, e)
}
} else if x != tc.expected {
t.Errorf("[%d] expected %d, got %d", i, tc.expected, x)
}
}
}
type fakeConntracker struct {
called []string
err error
}
// SetMax value is calculated based on the number of CPUs by getConntrackMax()
func (fc *fakeConntracker) SetMax(ctx context.Context, max int) error {
fc.called = append(fc.called, "SetMax")
return fc.err
}
func (fc *fakeConntracker) SetTCPEstablishedTimeout(ctx context.Context, seconds int) error {
fc.called = append(fc.called, fmt.Sprintf("SetTCPEstablishedTimeout(%d)", seconds))
return fc.err
}
func (fc *fakeConntracker) SetTCPCloseWaitTimeout(ctx context.Context, seconds int) error {
fc.called = append(fc.called, fmt.Sprintf("SetTCPCloseWaitTimeout(%d)", seconds))
return fc.err
}
func (fc *fakeConntracker) SetTCPBeLiberal(ctx context.Context, value int) error {
fc.called = append(fc.called, fmt.Sprintf("SetTCPBeLiberal(%d)", value))
return fc.err
}
func (fc *fakeConntracker) SetUDPTimeout(ctx context.Context, seconds int) error {
fc.called = append(fc.called, fmt.Sprintf("SetUDPTimeout(%d)", seconds))
return fc.err
}
func (fc *fakeConntracker) SetUDPStreamTimeout(ctx context.Context, seconds int) error {
fc.called = append(fc.called, fmt.Sprintf("SetUDPStreamTimeout(%d)", seconds))
return fc.err
}
func TestSetupConntrack(t *testing.T) {
_, ctx := ktesting.NewTestContext(t)
tests := []struct {
name string
config proxyconfigapi.KubeProxyConntrackConfiguration
expect []string
conntrackErr error
wantErr bool
}{
{
name: "do nothing if conntrack config is empty",
config: proxyconfigapi.KubeProxyConntrackConfiguration{},
expect: nil,
},
{
name: "SetMax is called if conntrack.maxPerCore is specified",
config: proxyconfigapi.KubeProxyConntrackConfiguration{
MaxPerCore: ptr.To(int32(12)),
},
expect: []string{"SetMax"},
},
{
name: "SetMax is not called if conntrack.maxPerCore is 0",
config: proxyconfigapi.KubeProxyConntrackConfiguration{
MaxPerCore: ptr.To(int32(0)),
},
expect: nil,
},
{
name: "SetTCPEstablishedTimeout is called if conntrack.tcpEstablishedTimeout is specified",
config: proxyconfigapi.KubeProxyConntrackConfiguration{
TCPEstablishedTimeout: &metav1.Duration{Duration: 5 * time.Second},
},
expect: []string{"SetTCPEstablishedTimeout(5)"},
},
{
name: "SetTCPEstablishedTimeout is not called if conntrack.tcpEstablishedTimeout is 0",
config: proxyconfigapi.KubeProxyConntrackConfiguration{
TCPEstablishedTimeout: &metav1.Duration{Duration: 0 * time.Second},
},
expect: nil,
},
{
name: "SetTCPCloseWaitTimeout is called if conntrack.tcpCloseWaitTimeout is specified",
config: proxyconfigapi.KubeProxyConntrackConfiguration{
TCPCloseWaitTimeout: &metav1.Duration{Duration: 5 * time.Second},
},
expect: []string{"SetTCPCloseWaitTimeout(5)"},
},
{
name: "SetTCPCloseWaitTimeout is not called if conntrack.tcpCloseWaitTimeout is 0",
config: proxyconfigapi.KubeProxyConntrackConfiguration{
TCPCloseWaitTimeout: &metav1.Duration{Duration: 0 * time.Second},
},
expect: nil,
},
{
name: "SetTCPBeLiberal is called if conntrack.tcpBeLiberal is true",
config: proxyconfigapi.KubeProxyConntrackConfiguration{
TCPBeLiberal: true,
},
expect: []string{"SetTCPBeLiberal(1)"},
},
{
name: "SetTCPBeLiberal is not called if conntrack.tcpBeLiberal is false",
config: proxyconfigapi.KubeProxyConntrackConfiguration{
TCPBeLiberal: false,
},
expect: nil,
},
{
name: "SetUDPTimeout is called if conntrack.udpTimeout is specified",
config: proxyconfigapi.KubeProxyConntrackConfiguration{
UDPTimeout: metav1.Duration{Duration: 5 * time.Second},
},
expect: []string{"SetUDPTimeout(5)"},
},
{
name: "SetUDPTimeout is called if conntrack.udpTimeout is zero",
config: proxyconfigapi.KubeProxyConntrackConfiguration{
UDPTimeout: metav1.Duration{Duration: 0 * time.Second},
},
expect: nil,
},
{
name: "SetUDPStreamTimeout is called if conntrack.udpStreamTimeout is specified",
config: proxyconfigapi.KubeProxyConntrackConfiguration{
UDPStreamTimeout: metav1.Duration{Duration: 5 * time.Second},
},
expect: []string{"SetUDPStreamTimeout(5)"},
},
{
name: "SetUDPStreamTimeout is called if conntrack.udpStreamTimeout is zero",
config: proxyconfigapi.KubeProxyConntrackConfiguration{
UDPStreamTimeout: metav1.Duration{Duration: 0 * time.Second},
},
expect: nil,
},
{
name: "an error is returned if conntrack.SetTCPEstablishedTimeout fails",
config: proxyconfigapi.KubeProxyConntrackConfiguration{
TCPEstablishedTimeout: &metav1.Duration{Duration: 5 * time.Second},
},
expect: []string{"SetTCPEstablishedTimeout(5)"},
conntrackErr: errors.New("random error"),
wantErr: true,
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
fc := &fakeConntracker{err: test.conntrackErr}
err := setupConntrack(ctx, fc, &test.config)
if test.wantErr && err == nil {
t.Errorf("Test %q: Expected error, got nil", test.name)
}
if !test.wantErr && err != nil {
t.Errorf("Test %q: Expected no error, got %v", test.name, err)
}
if !cmp.Equal(fc.called, test.expect) {
t.Errorf("Test %q: Expected conntrack calls: %v, got: %v", test.name, test.expect, fc.called)
}
})
}
}

85
cmd/kube-proxy/app/server_linux.go
View File

@@ -26,11 +26,6 @@ import (
"errors"
"fmt"
"os"
goruntime "runtime"
"time"
"github.com/google/cadvisor/machine"
"github.com/google/cadvisor/utils/sysfs"
v1 "k8s.io/api/core/v1"
utilsysctl "k8s.io/component-helpers/node/util/sysctl"
@@ -69,8 +64,8 @@ func (o *Options) platformApplyDefaults(config *proxyconfigapi.KubeProxyConfigur
// Proxier. It should fill in any platform-specific fields and perform other
// platform-specific setup.
func (s *ProxyServer) platformSetup(ctx context.Context) error {
ct := &realConntracker{}
err := s.setupConntrack(ctx, ct)
ct := &realConntrackConfigurer{}
err := setupConntrack(ctx, ct, &s.Config.Linux.Conntrack)
if err != nil {
return err
}
@@ -296,82 +291,6 @@ func (s *ProxyServer) createProxier(ctx context.Context, config *proxyconfigapi.
return proxier, nil
}
func (s *ProxyServer) setupConntrack(ctx context.Context, ct Conntracker) error {
max, err := getConntrackMax(ctx, s.Config.Linux.Conntrack)
if err != nil {
return err
}
if max > 0 {
err := ct.SetMax(ctx, max)
if err != nil {
return err
}
}
if s.Config.Linux.Conntrack.TCPEstablishedTimeout != nil && s.Config.Linux.Conntrack.TCPEstablishedTimeout.Duration > 0 {
timeout := int(s.Config.Linux.Conntrack.TCPEstablishedTimeout.Duration / time.Second)
if err := ct.SetTCPEstablishedTimeout(ctx, timeout); err != nil {
return err
}
}
if s.Config.Linux.Conntrack.TCPCloseWaitTimeout != nil && s.Config.Linux.Conntrack.TCPCloseWaitTimeout.Duration > 0 {
timeout := int(s.Config.Linux.Conntrack.TCPCloseWaitTimeout.Duration / time.Second)
if err := ct.SetTCPCloseWaitTimeout(ctx, timeout); err != nil {
return err
}
}
if s.Config.Linux.Conntrack.TCPBeLiberal {
if err := ct.SetTCPBeLiberal(ctx, 1); err != nil {
return err
}
}
if s.Config.Linux.Conntrack.UDPTimeout.Duration > 0 {
timeout := int(s.Config.Linux.Conntrack.UDPTimeout.Duration / time.Second)
if err := ct.SetUDPTimeout(ctx, timeout); err != nil {
return err
}
}
if s.Config.Linux.Conntrack.UDPStreamTimeout.Duration > 0 {
timeout := int(s.Config.Linux.Conntrack.UDPStreamTimeout.Duration / time.Second)
if err := ct.SetUDPStreamTimeout(ctx, timeout); err != nil {
return err
}
}
return nil
}
func getConntrackMax(ctx context.Context, config proxyconfigapi.KubeProxyConntrackConfiguration) (int, error) {
logger := klog.FromContext(ctx)
if config.MaxPerCore != nil && *config.MaxPerCore > 0 {
floor := 0
if config.Min != nil {
floor = int(*config.Min)
}
scaled := int(*config.MaxPerCore) * detectNumCPU()
if scaled > floor {
logger.V(3).Info("GetConntrackMax: using scaled conntrack-max-per-core")
return scaled, nil
}
logger.V(3).Info("GetConntrackMax: using conntrack-min")
return floor, nil
}
return 0, nil
}
func detectNumCPU() int {
// try get numCPU from /sys firstly due to a known issue (https://github.com/kubernetes/kubernetes/issues/99225)
_, numCPU, err := machine.GetTopology(sysfs.NewRealSysFs())
if err != nil || numCPU < 1 {
return goruntime.NumCPU()
}
return numCPU
}
func getLocalDetectors(logger klog.Logger, primaryIPFamily v1.IPFamily, config *proxyconfigapi.KubeProxyConfiguration, nodePodCIDRs []string) map[v1.IPFamily]proxyutil.LocalTrafficDetector {
localDetectors := map[v1.IPFamily]proxyutil.LocalTrafficDetector{
v1.IPv4Protocol: proxyutil.NewNoOpLocalDetector(),

217
cmd/kube-proxy/app/server_linux_test.go
View File

@@ -20,25 +20,19 @@ limitations under the License.
package app
import (
"context"
"errors"
"fmt"
"os"
"path/filepath"
"reflect"
goruntime "runtime"
"strings"
"testing"
"time"
"github.com/google/go-cmp/cmp"
"github.com/spf13/pflag"
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
proxyconfigapi "k8s.io/kubernetes/pkg/proxy/apis/config"
proxyutil "k8s.io/kubernetes/pkg/proxy/util"
"k8s.io/kubernetes/test/utils/ktesting"
"k8s.io/utils/ptr"
)
func Test_platformApplyDefaults(t *testing.T) {
@@ -555,214 +549,3 @@ detectLocalMode: "BridgeInterface"`)
tearDown(file, tempDir)
}
}
func TestGetConntrackMax(t *testing.T) {
ncores := goruntime.NumCPU()
testCases := []struct {
min int32
maxPerCore int32
expected int
err string
}{
{
expected: 0,
},
{
maxPerCore: 67890, // use this if Max is 0
min: 1, // avoid 0 default
expected: 67890 * ncores,
},
{
maxPerCore: 1, // ensure that Min is considered
min: 123456,
expected: 123456,
},
{
maxPerCore: 0, // leave system setting
min: 123456,
expected: 0,
},
}
for i, tc := range testCases {
cfg := proxyconfigapi.KubeProxyConntrackConfiguration{
Min: ptr.To(tc.min),
MaxPerCore: ptr.To(tc.maxPerCore),
}
_, ctx := ktesting.NewTestContext(t)
x, e := getConntrackMax(ctx, cfg)
if e != nil {
if tc.err == "" {
t.Errorf("[%d] unexpected error: %v", i, e)
} else if !strings.Contains(e.Error(), tc.err) {
t.Errorf("[%d] expected an error containing %q: %v", i, tc.err, e)
}
} else if x != tc.expected {
t.Errorf("[%d] expected %d, got %d", i, tc.expected, x)
}
}
}
type fakeConntracker struct {
called []string
err error
}
// SetMax value is calculated based on the number of CPUs by getConntrackMax()
func (fc *fakeConntracker) SetMax(ctx context.Context, max int) error {
fc.called = append(fc.called, "SetMax")
return fc.err
}
func (fc *fakeConntracker) SetTCPEstablishedTimeout(ctx context.Context, seconds int) error {
fc.called = append(fc.called, fmt.Sprintf("SetTCPEstablishedTimeout(%d)", seconds))
return fc.err
}
func (fc *fakeConntracker) SetTCPCloseWaitTimeout(ctx context.Context, seconds int) error {
fc.called = append(fc.called, fmt.Sprintf("SetTCPCloseWaitTimeout(%d)", seconds))
return fc.err
}
func (fc *fakeConntracker) SetTCPBeLiberal(ctx context.Context, value int) error {
fc.called = append(fc.called, fmt.Sprintf("SetTCPBeLiberal(%d)", value))
return fc.err
}
func (fc *fakeConntracker) SetUDPTimeout(ctx context.Context, seconds int) error {
fc.called = append(fc.called, fmt.Sprintf("SetUDPTimeout(%d)", seconds))
return fc.err
}
func (fc *fakeConntracker) SetUDPStreamTimeout(ctx context.Context, seconds int) error {
fc.called = append(fc.called, fmt.Sprintf("SetUDPStreamTimeout(%d)", seconds))
return fc.err
}
func TestSetupConntrack(t *testing.T) {
_, ctx := ktesting.NewTestContext(t)
tests := []struct {
name string
config proxyconfigapi.KubeProxyConntrackConfiguration
expect []string
conntrackErr error
wantErr bool
}{
{
name: "do nothing if conntrack config is empty",
config: proxyconfigapi.KubeProxyConntrackConfiguration{},
expect: nil,
},
{
name: "SetMax is called if conntrack.maxPerCore is specified",
config: proxyconfigapi.KubeProxyConntrackConfiguration{
MaxPerCore: ptr.To(int32(12)),
},
expect: []string{"SetMax"},
},
{
name: "SetMax is not called if conntrack.maxPerCore is 0",
config: proxyconfigapi.KubeProxyConntrackConfiguration{
MaxPerCore: ptr.To(int32(0)),
},
expect: nil,
},
{
name: "SetTCPEstablishedTimeout is called if conntrack.tcpEstablishedTimeout is specified",
config: proxyconfigapi.KubeProxyConntrackConfiguration{
TCPEstablishedTimeout: &metav1.Duration{Duration: 5 * time.Second},
},
expect: []string{"SetTCPEstablishedTimeout(5)"},
},
{
name: "SetTCPEstablishedTimeout is not called if conntrack.tcpEstablishedTimeout is 0",
config: proxyconfigapi.KubeProxyConntrackConfiguration{
TCPEstablishedTimeout: &metav1.Duration{Duration: 0 * time.Second},
},
expect: nil,
},
{
name: "SetTCPCloseWaitTimeout is called if conntrack.tcpCloseWaitTimeout is specified",
config: proxyconfigapi.KubeProxyConntrackConfiguration{
TCPCloseWaitTimeout: &metav1.Duration{Duration: 5 * time.Second},
},
expect: []string{"SetTCPCloseWaitTimeout(5)"},
},
{
name: "SetTCPCloseWaitTimeout is not called if conntrack.tcpCloseWaitTimeout is 0",
config: proxyconfigapi.KubeProxyConntrackConfiguration{
TCPCloseWaitTimeout: &metav1.Duration{Duration: 0 * time.Second},
},
expect: nil,
},
{
name: "SetTCPBeLiberal is called if conntrack.tcpBeLiberal is true",
config: proxyconfigapi.KubeProxyConntrackConfiguration{
TCPBeLiberal: true,
},
expect: []string{"SetTCPBeLiberal(1)"},
},
{
name: "SetTCPBeLiberal is not called if conntrack.tcpBeLiberal is false",
config: proxyconfigapi.KubeProxyConntrackConfiguration{
TCPBeLiberal: false,
},
expect: nil,
},
{
name: "SetUDPTimeout is called if conntrack.udpTimeout is specified",
config: proxyconfigapi.KubeProxyConntrackConfiguration{
UDPTimeout: metav1.Duration{Duration: 5 * time.Second},
},
expect: []string{"SetUDPTimeout(5)"},
},
{
name: "SetUDPTimeout is called if conntrack.udpTimeout is zero",
config: proxyconfigapi.KubeProxyConntrackConfiguration{
UDPTimeout: metav1.Duration{Duration: 0 * time.Second},
},
expect: nil,
},
{
name: "SetUDPStreamTimeout is called if conntrack.udpStreamTimeout is specified",
config: proxyconfigapi.KubeProxyConntrackConfiguration{
UDPStreamTimeout: metav1.Duration{Duration: 5 * time.Second},
},
expect: []string{"SetUDPStreamTimeout(5)"},
},
{
name: "SetUDPStreamTimeout is called if conntrack.udpStreamTimeout is zero",
config: proxyconfigapi.KubeProxyConntrackConfiguration{
UDPStreamTimeout: metav1.Duration{Duration: 0 * time.Second},
},
expect: nil,
},
{
name: "an error is returned if conntrack.SetTCPEstablishedTimeout fails",
config: proxyconfigapi.KubeProxyConntrackConfiguration{
TCPEstablishedTimeout: &metav1.Duration{Duration: 5 * time.Second},
},
expect: []string{"SetTCPEstablishedTimeout(5)"},
conntrackErr: errors.New("random error"),
wantErr: true,
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
fc := &fakeConntracker{err: test.conntrackErr}
s := &ProxyServer{
Config: &proxyconfigapi.KubeProxyConfiguration{
Linux: proxyconfigapi.KubeProxyLinuxConfiguration{
Conntrack: test.config,
},
},
}
err := s.setupConntrack(ctx, fc)
if test.wantErr && err == nil {
t.Errorf("Test %q: Expected error, got nil", test.name)
}
if !test.wantErr && err != nil {
t.Errorf("Test %q: Expected no error, got %v", test.name, err)
}
if !cmp.Equal(fc.called, test.expect) {
t.Errorf("Test %q: Expected conntrack calls: %v, got: %v", test.name, test.expect, fc.called)
}
})
}
}