mirror of
https://github.com/k3s-io/kubernetes.git
synced 2025-09-26 12:46:06 +00:00
dual stack services (#91824)
* api: structure change * api: defaulting, conversion, and validation * [FIX] validation: auto remove second ip/family when service changes to SingleStack * [FIX] api: defaulting, conversion, and validation * api-server: clusterIPs alloc, printers, storage and strategy * [FIX] clusterIPs default on read * alloc: auto remove second ip/family when service changes to SingleStack * api-server: repair loop handling for clusterIPs * api-server: force kubernetes default service into single stack * api-server: tie dualstack feature flag with endpoint feature flag * controller-manager: feature flag, endpoint, and endpointSlice controllers handling multi family service * [FIX] controller-manager: feature flag, endpoint, and endpointSlicecontrollers handling multi family service * kube-proxy: feature-flag, utils, proxier, and meta proxier * [FIX] kubeproxy: call both proxier at the same time * kubenet: remove forced pod IP sorting * kubectl: modify describe to include ClusterIPs, IPFamilies, and IPFamilyPolicy * e2e: fix tests that depends on IPFamily field AND add dual stack tests * e2e: fix expected error message for ClusterIP immutability * add integration tests for dualstack the third phase of dual stack is a very complex change in the API, basically it introduces Dual Stack services. Main changes are: - It pluralizes the Service IPFamily field to IPFamilies, and removes the singular field. - It introduces a new field IPFamilyPolicyType that can take 3 values to express the "dual-stack(mad)ness" of the cluster: SingleStack, PreferDualStack and RequireDualStack - It pluralizes ClusterIP to ClusterIPs. The goal is to add coverage to the services API operations, taking into account the 6 different modes a cluster can have: - single stack: IP4 or IPv6 (as of today) - dual stack: IPv4 only, IPv6 only, IPv4 - IPv6, IPv6 - IPv4 * [FIX] add integration tests for dualstack * generated data * generated files Co-authored-by: Antonio Ojea <aojea@redhat.com>
This commit is contained in:
committed by
GitHub
parent
d0e06cf3e0
commit
6675eba3ef
@@ -289,18 +289,23 @@ func NewProxier(ipt utiliptables.Interface,
|
||||
|
||||
serviceHealthServer := healthcheck.NewServiceHealthServer(hostname, recorder)
|
||||
|
||||
isIPv6 := ipt.IsIPv6()
|
||||
ipFamily := v1.IPv4Protocol
|
||||
if ipt.IsIPv6() {
|
||||
ipFamily = v1.IPv6Protocol
|
||||
}
|
||||
|
||||
var incorrectAddresses []string
|
||||
nodePortAddresses, incorrectAddresses = utilproxy.FilterIncorrectCIDRVersion(nodePortAddresses, isIPv6)
|
||||
nodePortAddresses, incorrectAddresses = utilproxy.FilterIncorrectCIDRVersion(nodePortAddresses, ipFamily)
|
||||
if len(incorrectAddresses) > 0 {
|
||||
klog.Warning("NodePortAddresses of wrong family; ", incorrectAddresses)
|
||||
}
|
||||
|
||||
proxier := &Proxier{
|
||||
portsMap: make(map[utilproxy.LocalPort]utilproxy.Closeable),
|
||||
serviceMap: make(proxy.ServiceMap),
|
||||
serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, &isIPv6, recorder, nil),
|
||||
serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, ipFamily, recorder, nil),
|
||||
endpointsMap: make(proxy.EndpointsMap),
|
||||
endpointsChanges: proxy.NewEndpointChangeTracker(hostname, newEndpointInfo, &isIPv6, recorder, endpointSlicesEnabled, nil),
|
||||
endpointsChanges: proxy.NewEndpointChangeTracker(hostname, newEndpointInfo, ipFamily, recorder, endpointSlicesEnabled, nil),
|
||||
syncPeriod: syncPeriod,
|
||||
iptables: ipt,
|
||||
masqueradeAll: masqueradeAll,
|
||||
@@ -362,7 +367,7 @@ func NewDualStackProxier(
|
||||
nodePortAddresses []string,
|
||||
) (proxy.Provider, error) {
|
||||
// Create an ipv4 instance of the single-stack proxier
|
||||
nodePortAddresses4, nodePortAddresses6 := utilproxy.FilterIncorrectCIDRVersion(nodePortAddresses, false)
|
||||
nodePortAddresses4, nodePortAddresses6 := utilproxy.FilterIncorrectCIDRVersion(nodePortAddresses, v1.IPv4Protocol)
|
||||
ipv4Proxier, err := NewProxier(ipt[0], sysctl,
|
||||
exec, syncPeriod, minSyncPeriod, masqueradeAll, masqueradeBit, localDetectors[0], hostname,
|
||||
nodeIP[0], recorder, healthzServer, nodePortAddresses4)
|
||||
@@ -376,8 +381,7 @@ func NewDualStackProxier(
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("unable to create ipv6 proxier: %v", err)
|
||||
}
|
||||
|
||||
return metaproxier.NewMetaProxier(ipv4Proxier, ipv6Proxier), nil // TODO move meta-proxier to mode-neutral package
|
||||
return metaproxier.NewMetaProxier(ipv4Proxier, ipv6Proxier), nil
|
||||
}
|
||||
|
||||
type iptablesJumpChain struct {
|
||||
|
@@ -150,8 +150,7 @@ func TestGetChainLinesMultipleTables(t *testing.T) {
|
||||
}
|
||||
checkAllLines(t, utiliptables.TableNAT, []byte(iptablesSave), expected)
|
||||
}
|
||||
|
||||
func TestDeleteEndpointConnections(t *testing.T) {
|
||||
func TestDeleteEndpointConnectionsIPv4(t *testing.T) {
|
||||
const (
|
||||
UDP = v1.ProtocolUDP
|
||||
TCP = v1.ProtocolTCP
|
||||
@@ -175,21 +174,24 @@ func TestDeleteEndpointConnections(t *testing.T) {
|
||||
svcPort: 80,
|
||||
protocol: UDP,
|
||||
endpoint: "10.240.0.3:80",
|
||||
}, {
|
||||
},
|
||||
{
|
||||
description: "V4 TCP",
|
||||
svcName: "v4-tcp",
|
||||
svcIP: "10.96.2.2",
|
||||
svcPort: 80,
|
||||
protocol: TCP,
|
||||
endpoint: "10.240.0.4:80",
|
||||
}, {
|
||||
},
|
||||
{
|
||||
description: "V4 SCTP",
|
||||
svcName: "v4-sctp",
|
||||
svcIP: "10.96.3.3",
|
||||
svcPort: 80,
|
||||
protocol: SCTP,
|
||||
endpoint: "10.240.0.5:80",
|
||||
}, {
|
||||
},
|
||||
{
|
||||
description: "V4 UDP, nothing to delete, benign error",
|
||||
svcName: "v4-udp-nothing-to-delete",
|
||||
svcIP: "10.96.1.1",
|
||||
@@ -197,7 +199,8 @@ func TestDeleteEndpointConnections(t *testing.T) {
|
||||
protocol: UDP,
|
||||
endpoint: "10.240.0.3:80",
|
||||
simulatedErr: conntrack.NoConnectionToDelete,
|
||||
}, {
|
||||
},
|
||||
{
|
||||
description: "V4 UDP, unexpected error, should be glogged",
|
||||
svcName: "v4-udp-simulated-error",
|
||||
svcIP: "10.96.1.1",
|
||||
@@ -205,27 +208,6 @@ func TestDeleteEndpointConnections(t *testing.T) {
|
||||
protocol: UDP,
|
||||
endpoint: "10.240.0.3:80",
|
||||
simulatedErr: "simulated error",
|
||||
}, {
|
||||
description: "V6 UDP",
|
||||
svcName: "v6-udp",
|
||||
svcIP: "fd00:1234::20",
|
||||
svcPort: 80,
|
||||
protocol: UDP,
|
||||
endpoint: "[2001:db8::2]:80",
|
||||
}, {
|
||||
description: "V6 TCP",
|
||||
svcName: "v6-tcp",
|
||||
svcIP: "fd00:1234::30",
|
||||
svcPort: 80,
|
||||
protocol: TCP,
|
||||
endpoint: "[2001:db8::3]:80",
|
||||
}, {
|
||||
description: "V6 SCTP",
|
||||
svcName: "v6-sctp",
|
||||
svcIP: "fd00:1234::40",
|
||||
svcPort: 80,
|
||||
protocol: SCTP,
|
||||
endpoint: "[2001:db8::4]:80",
|
||||
},
|
||||
}
|
||||
|
||||
@@ -329,6 +311,149 @@ func TestDeleteEndpointConnections(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeleteEndpointConnectionsIPv6(t *testing.T) {
|
||||
const (
|
||||
UDP = v1.ProtocolUDP
|
||||
TCP = v1.ProtocolTCP
|
||||
SCTP = v1.ProtocolSCTP
|
||||
)
|
||||
|
||||
testCases := []struct {
|
||||
description string
|
||||
svcName string
|
||||
svcIP string
|
||||
svcPort int32
|
||||
protocol v1.Protocol
|
||||
endpoint string // IP:port endpoint
|
||||
epSvcPair proxy.ServiceEndpoint // Will be generated by test
|
||||
simulatedErr string
|
||||
}{
|
||||
{
|
||||
description: "V6 UDP",
|
||||
svcName: "v6-udp",
|
||||
svcIP: "fd00:1234::20",
|
||||
svcPort: 80,
|
||||
protocol: UDP,
|
||||
endpoint: "[2001:db8::2]:80",
|
||||
},
|
||||
{
|
||||
description: "V6 TCP",
|
||||
svcName: "v6-tcp",
|
||||
svcIP: "fd00:1234::30",
|
||||
svcPort: 80,
|
||||
protocol: TCP,
|
||||
endpoint: "[2001:db8::3]:80",
|
||||
},
|
||||
{
|
||||
description: "V6 SCTP",
|
||||
svcName: "v6-sctp",
|
||||
svcIP: "fd00:1234::40",
|
||||
svcPort: 80,
|
||||
protocol: SCTP,
|
||||
endpoint: "[2001:db8::4]:80",
|
||||
},
|
||||
}
|
||||
|
||||
// Create a fake executor for the conntrack utility. This should only be
|
||||
// invoked for UDP and SCTP connections, since no conntrack cleanup is needed for TCP
|
||||
fcmd := fakeexec.FakeCmd{}
|
||||
fexec := fakeexec.FakeExec{
|
||||
LookPathFunc: func(cmd string) (string, error) { return cmd, nil },
|
||||
}
|
||||
execFunc := func(cmd string, args ...string) exec.Cmd {
|
||||
return fakeexec.InitFakeCmd(&fcmd, cmd, args...)
|
||||
}
|
||||
for _, tc := range testCases {
|
||||
if conntrack.IsClearConntrackNeeded(tc.protocol) {
|
||||
var cmdOutput string
|
||||
var simErr error
|
||||
if tc.simulatedErr == "" {
|
||||
cmdOutput = "1 flow entries have been deleted"
|
||||
} else {
|
||||
simErr = fmt.Errorf(tc.simulatedErr)
|
||||
}
|
||||
cmdFunc := func() ([]byte, []byte, error) { return []byte(cmdOutput), nil, simErr }
|
||||
fcmd.CombinedOutputScript = append(fcmd.CombinedOutputScript, cmdFunc)
|
||||
fexec.CommandScript = append(fexec.CommandScript, execFunc)
|
||||
}
|
||||
}
|
||||
|
||||
ipt := iptablestest.NewIPv6Fake()
|
||||
fp := NewFakeProxier(ipt, false)
|
||||
fp.exec = &fexec
|
||||
|
||||
for _, tc := range testCases {
|
||||
makeServiceMap(fp,
|
||||
makeTestService("ns1", tc.svcName, func(svc *v1.Service) {
|
||||
svc.Spec.ClusterIP = tc.svcIP
|
||||
svc.Spec.Ports = []v1.ServicePort{{
|
||||
Name: "p80",
|
||||
Port: tc.svcPort,
|
||||
Protocol: tc.protocol,
|
||||
}}
|
||||
svc.Spec.ExternalTrafficPolicy = v1.ServiceExternalTrafficPolicyTypeLocal
|
||||
}),
|
||||
)
|
||||
|
||||
proxy.UpdateServiceMap(fp.serviceMap, fp.serviceChanges)
|
||||
}
|
||||
|
||||
// Run the test cases
|
||||
for _, tc := range testCases {
|
||||
priorExecs := fexec.CommandCalls
|
||||
priorGlogErrs := klog.Stats.Error.Lines()
|
||||
|
||||
svc := proxy.ServicePortName{
|
||||
NamespacedName: types.NamespacedName{Namespace: "ns1", Name: tc.svcName},
|
||||
Port: "p80",
|
||||
Protocol: tc.protocol,
|
||||
}
|
||||
input := []proxy.ServiceEndpoint{
|
||||
{
|
||||
Endpoint: tc.endpoint,
|
||||
ServicePortName: svc,
|
||||
},
|
||||
}
|
||||
|
||||
fp.deleteEndpointConnections(input)
|
||||
|
||||
// For UDP and SCTP connections, check the executed conntrack command
|
||||
var expExecs int
|
||||
if conntrack.IsClearConntrackNeeded(tc.protocol) {
|
||||
isIPv6 := func(ip string) bool {
|
||||
netIP := net.ParseIP(ip)
|
||||
return netIP.To4() == nil
|
||||
}
|
||||
endpointIP := utilproxy.IPPart(tc.endpoint)
|
||||
expectCommand := fmt.Sprintf("conntrack -D --orig-dst %s --dst-nat %s -p %s", tc.svcIP, endpointIP, strings.ToLower(string((tc.protocol))))
|
||||
if isIPv6(endpointIP) {
|
||||
expectCommand += " -f ipv6"
|
||||
}
|
||||
actualCommand := strings.Join(fcmd.CombinedOutputLog[fexec.CommandCalls-1], " ")
|
||||
if actualCommand != expectCommand {
|
||||
t.Errorf("%s: Expected command: %s, but executed %s", tc.description, expectCommand, actualCommand)
|
||||
}
|
||||
expExecs = 1
|
||||
}
|
||||
|
||||
// Check the number of times conntrack was executed
|
||||
execs := fexec.CommandCalls - priorExecs
|
||||
if execs != expExecs {
|
||||
t.Errorf("%s: Expected conntrack to be executed %d times, but got %d", tc.description, expExecs, execs)
|
||||
}
|
||||
|
||||
// Check the number of new glog errors
|
||||
var expGlogErrs int64
|
||||
if tc.simulatedErr != "" && tc.simulatedErr != conntrack.NoConnectionToDelete {
|
||||
expGlogErrs = 1
|
||||
}
|
||||
glogErrs := klog.Stats.Error.Lines() - priorGlogErrs
|
||||
if glogErrs != expGlogErrs {
|
||||
t.Errorf("%s: Expected %d glogged errors, but got %d", tc.description, expGlogErrs, glogErrs)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// fakePortOpener implements portOpener.
|
||||
type fakePortOpener struct {
|
||||
openPorts []*utilproxy.LocalPort
|
||||
@@ -346,13 +471,17 @@ const testHostname = "test-hostname"
|
||||
func NewFakeProxier(ipt utiliptables.Interface, endpointSlicesEnabled bool) *Proxier {
|
||||
// TODO: Call NewProxier after refactoring out the goroutine
|
||||
// invocation into a Run() method.
|
||||
ipfamily := v1.IPv4Protocol
|
||||
if ipt.IsIPv6() {
|
||||
ipfamily = v1.IPv6Protocol
|
||||
}
|
||||
detectLocal, _ := proxyutiliptables.NewDetectLocalByCIDR("10.0.0.0/24", ipt)
|
||||
p := &Proxier{
|
||||
exec: &fakeexec.FakeExec{},
|
||||
serviceMap: make(proxy.ServiceMap),
|
||||
serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, nil, nil, nil),
|
||||
serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, ipfamily, nil, nil),
|
||||
endpointsMap: make(proxy.EndpointsMap),
|
||||
endpointsChanges: proxy.NewEndpointChangeTracker(testHostname, newEndpointInfo, nil, nil, endpointSlicesEnabled, nil),
|
||||
endpointsChanges: proxy.NewEndpointChangeTracker(testHostname, newEndpointInfo, ipfamily, nil, endpointSlicesEnabled, nil),
|
||||
iptables: ipt,
|
||||
masqueradeMark: "0x4000",
|
||||
localDetector: detectLocal,
|
||||
|
Reference in New Issue
Block a user