mirror of
https://github.com/k3s-io/kubernetes.git
synced 2026-07-17 20:00:07 +00:00
Move conntrack sysctl setup from cmd/kube-proxy/ to pkg/proxy/conntrack/
Eventually this code will be called from the backends themselves.
This commit is contained in:
Dan Winship
@@ -1,200 +0,0 @@
|
||||
//go:build linux
|
||||
// +build linux
|
||||
|
||||
/*
|
||||
Copyright 2015 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"
|
||||
"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"
|
||||
)
|
||||
|
||||
// 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 conntrackConfigurer interface {
|
||||
// SetMax adjusts nf_conntrack_max.
|
||||
SetMax(ctx context.Context, max int) error
|
||||
// SetTCPEstablishedTimeout adjusts nf_conntrack_tcp_timeout_established.
|
||||
SetTCPEstablishedTimeout(ctx context.Context, seconds int) error
|
||||
// SetTCPCloseWaitTimeout adjusts nf_conntrack_tcp_timeout_close_wait.
|
||||
SetTCPCloseWaitTimeout(ctx context.Context, seconds int) error
|
||||
// SetTCPBeLiberal adjusts nf_conntrack_tcp_be_liberal.
|
||||
SetTCPBeLiberal(ctx context.Context, value int) error
|
||||
// SetUDPTimeout adjusts nf_conntrack_udp_timeout.
|
||||
SetUDPTimeout(ctx context.Context, seconds int) error
|
||||
// SetUDPStreamTimeout adjusts nf_conntrack_udp_timeout_stream.
|
||||
SetUDPStreamTimeout(ctx context.Context, seconds int) error
|
||||
}
|
||||
|
||||
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 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 {
|
||||
return err
|
||||
}
|
||||
|
||||
// Check if hashsize is large enough for the nf_conntrack_max value.
|
||||
hashsize, err := readIntStringFile("/sys/module/nf_conntrack/parameters/hashsize")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if hashsize >= (max / 4) {
|
||||
return nil
|
||||
}
|
||||
|
||||
logger.Info("Setting conntrack hashsize", "conntrackHashsize", max/4)
|
||||
return writeIntStringFile("/sys/module/nf_conntrack/parameters/hashsize", max/4)
|
||||
}
|
||||
|
||||
func (rct realConntrackConfigurer) SetTCPEstablishedTimeout(ctx context.Context, seconds int) error {
|
||||
return rct.setIntSysCtl(ctx, "nf_conntrack_tcp_timeout_established", seconds)
|
||||
}
|
||||
|
||||
func (rct realConntrackConfigurer) SetTCPCloseWaitTimeout(ctx context.Context, seconds int) error {
|
||||
return rct.setIntSysCtl(ctx, "nf_conntrack_tcp_timeout_close_wait", seconds)
|
||||
}
|
||||
|
||||
func (rct realConntrackConfigurer) SetTCPBeLiberal(ctx context.Context, value int) error {
|
||||
return rct.setIntSysCtl(ctx, "nf_conntrack_tcp_be_liberal", value)
|
||||
}
|
||||
|
||||
func (rct realConntrackConfigurer) SetUDPTimeout(ctx context.Context, seconds int) error {
|
||||
return rct.setIntSysCtl(ctx, "nf_conntrack_udp_timeout", seconds)
|
||||
}
|
||||
|
||||
func (rct realConntrackConfigurer) SetUDPStreamTimeout(ctx context.Context, seconds int) error {
|
||||
return rct.setIntSysCtl(ctx, "nf_conntrack_udp_timeout_stream", seconds)
|
||||
}
|
||||
|
||||
func (rct realConntrackConfigurer) setIntSysCtl(ctx context.Context, name string, value int) error {
|
||||
logger := klog.FromContext(ctx)
|
||||
entry := "net/netfilter/" + name
|
||||
|
||||
sys := sysctl.New()
|
||||
if val, _ := sys.GetSysctl(entry); val != value {
|
||||
logger.Info("Set sysctl", "entry", entry, "value", value)
|
||||
if err := sys.SetSysctl(entry, value); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func readIntStringFile(filename string) (int, error) {
|
||||
b, err := os.ReadFile(filename)
|
||||
if err != nil {
|
||||
return -1, err
|
||||
}
|
||||
return strconv.Atoi(strings.TrimSpace(string(b)))
|
||||
}
|
||||
|
||||
func writeIntStringFile(filename string, value int) error {
|
||||
return os.WriteFile(filename, []byte(strconv.Itoa(value)), 0640)
|
||||
}
|
||||
@@ -1,241 +0,0 @@
|
||||
//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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -32,6 +32,7 @@ import (
|
||||
"k8s.io/klog/v2"
|
||||
"k8s.io/kubernetes/pkg/proxy"
|
||||
proxyconfigapi "k8s.io/kubernetes/pkg/proxy/apis/config"
|
||||
"k8s.io/kubernetes/pkg/proxy/conntrack"
|
||||
"k8s.io/kubernetes/pkg/proxy/iptables"
|
||||
"k8s.io/kubernetes/pkg/proxy/ipvs"
|
||||
utilipset "k8s.io/kubernetes/pkg/proxy/ipvs/ipset"
|
||||
@@ -64,10 +65,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 := &realConntrackConfigurer{}
|
||||
err := setupConntrack(ctx, ct, &s.Config.Linux.Conntrack)
|
||||
if err != nil {
|
||||
return err
|
||||
if err := conntrack.SetSysctls(ctx, &s.Config.Linux.Conntrack); err != nil {
|
||||
return fmt.Errorf("could not set conntrack parameters from kube-proxy configuration: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
|
||||
Reference in New Issue
Block a user