Add dual-stack service support

pkg/agent/proxy/proxier.go

@@ -514,14 +514,8 @@ func (p *proxier) OnServiceUpdate(oldService, service *corev1.Service) {
 	} else {
 		metrics.ServicesUpdatesTotal.Inc()
 	}
-	var isIPv6 bool
-	if oldService != nil {
-		isIPv6 = utilnet.IsIPv6String(oldService.Spec.ClusterIP)
-	} else {
-		isIPv6 = utilnet.IsIPv6String(service.Spec.ClusterIP)
-	}
-	if isIPv6 == p.isIPv6 {
-		if p.serviceChanges.OnServiceUpdate(oldService, service) && p.isInitialized() {
+	if p.serviceChanges.OnServiceUpdate(oldService, service) {
+		if p.isInitialized() {
 			p.runner.Run()
 		}
 	}
@@ -632,12 +626,17 @@ func NewProxier(
 
 	enableEndpointSlice := features.DefaultFeatureGate.Enabled(features.EndpointSlice)
 
+	ipFamily := corev1.IPv4Protocol
+	if isIPv6 {
+		ipFamily = corev1.IPv6Protocol
+	}
+
 	p := &proxier{
 		enableEndpointSlice:      enableEndpointSlice,
 		endpointsConfig:          config.NewEndpointsConfig(informerFactory.Core().V1().Endpoints(), resyncPeriod),
 		serviceConfig:            config.NewServiceConfig(informerFactory.Core().V1().Services(), resyncPeriod),
 		endpointsChanges:         newEndpointsChangesTracker(hostname, enableEndpointSlice, isIPv6),
-		serviceChanges:           newServiceChangesTracker(recorder, isIPv6),
+		serviceChanges:           newServiceChangesTracker(recorder, ipFamily),
 		serviceMap:               k8sproxy.ServiceMap{},
 		serviceInstalledMap:      k8sproxy.ServiceMap{},
 		endpointsInstalledMap:    types.EndpointsMap{},
@@ -659,7 +658,6 @@ func NewProxier(
 		p.endpointsConfig = config.NewEndpointsConfig(informerFactory.Core().V1().Endpoints(), resyncPeriod)
 		p.endpointsConfig.RegisterEventHandler(p)
 	}
-	p.runner = k8sproxy.NewBoundedFrequencyRunner(componentName, p.syncProxyRules, 0, 30*time.Second, -1)
 	return p
 }
 

pkg/agent/proxy/proxier_test.go

@@ -18,6 +18,7 @@ import (
 	"fmt"
 	"net"
 	"testing"
+	"time"
 
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
@@ -87,9 +88,15 @@ func NewFakeProxier(ofClient openflow.Client, isIPv6 bool) *proxier {
 		runtime.NewScheme(),
 		corev1.EventSource{Component: componentName, Host: hostname},
 	)
+
+	ipFamily := corev1.IPv4Protocol
+	if isIPv6 {
+		ipFamily = corev1.IPv6Protocol
+	}
+
 	p := &proxier{
 		endpointsChanges:         newEndpointsChangesTracker(hostname, false, isIPv6),
-		serviceChanges:           newServiceChangesTracker(recorder, isIPv6),
+		serviceChanges:           newServiceChangesTracker(recorder, ipFamily),
 		serviceMap:               k8sproxy.ServiceMap{},
 		serviceInstalledMap:      k8sproxy.ServiceMap{},
 		endpointsInstalledMap:    types.EndpointsMap{},
@@ -100,6 +107,7 @@ func NewFakeProxier(ofClient openflow.Client, isIPv6 bool) *proxier {
 		serviceStringMap:         map[string]k8sproxy.ServicePortName{},
 		isIPv6:                   isIPv6,
 	}
+	p.runner = k8sproxy.NewBoundedFrequencyRunner(componentName, p.syncProxyRules, time.Second, 30*time.Second, 2)
 	return p
 }
 
@@ -216,6 +224,81 @@ func TestClusterIPv6(t *testing.T) {
 	testClusterIP(t, net.ParseIP("10:20::41"), net.ParseIP("10:180::1"), true)
 }
 
+func TestDualStackService(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	mockOFClient := ofmock.NewMockClient(ctrl)
+	fpv4 := NewFakeProxier(mockOFClient, false)
+	fpv6 := NewFakeProxier(mockOFClient, true)
+	metaProxier := k8sproxy.NewMetaProxier(fpv4, fpv6)
+
+	svcPort := 80
+	svcPortName := k8sproxy.ServicePortName{
+		NamespacedName: makeNamespaceName("ns1", "svc1"),
+		Port:           "80",
+		Protocol:       corev1.ProtocolTCP,
+	}
+	svcIPv4 := net.ParseIP("10.20.30.41")
+	svcIPv6 := net.ParseIP("10:20::41")
+
+	s := makeTestService(svcPortName.Namespace, svcPortName.Name, func(svc *corev1.Service) {
+		svc.Spec.ClusterIP = svcIPv4.String()
+		svc.Spec.ClusterIPs = []string{svcIPv4.String(), svcIPv6.String()}
+		svc.Spec.IPFamilies = []corev1.IPFamily{corev1.IPv4Protocol, corev1.IPv6Protocol}
+		svc.Spec.Ports = []corev1.ServicePort{{
+			Name:     svcPortName.Port,
+			Port:     int32(svcPort),
+			Protocol: corev1.ProtocolTCP,
+		}}
+	})
+
+	epv4 := makeTestEndpoints(svcPortName.Namespace, svcPortName.Name, func(ept *corev1.Endpoints) {
+		ept.Subsets = []corev1.EndpointSubset{{
+			Addresses: []corev1.EndpointAddress{{
+				IP: "10.180.30.41",
+			}},
+			Ports: []corev1.EndpointPort{{
+				Name:     svcPortName.Port,
+				Port:     int32(svcPort),
+				Protocol: corev1.ProtocolTCP,
+			}},
+		}}
+	})
+
+	epv6 := makeTestEndpoints(svcPortName.Namespace, svcPortName.Name, func(ept *corev1.Endpoints) {
+		ept.Subsets = []corev1.EndpointSubset{{
+			Addresses: []corev1.EndpointAddress{{
+				IP: "10:180::1",
+			}},
+			Ports: []corev1.EndpointPort{{
+				Name:     svcPortName.Port,
+				Port:     int32(svcPort),
+				Protocol: corev1.ProtocolTCP,
+			}},
+		}}
+	})
+
+	metaProxier.OnServiceUpdate(nil, s)
+	metaProxier.OnServiceSynced()
+	metaProxier.OnEndpointsUpdate(nil, epv4)
+	metaProxier.OnEndpointsUpdate(nil, epv6)
+	metaProxier.OnEndpointsSynced()
+
+	groupIDv4, _ := fpv4.groupCounter.Get(svcPortName)
+	groupIDv6, _ := fpv6.groupCounter.Get(svcPortName)
+
+	mockOFClient.EXPECT().InstallServiceGroup(groupIDv4, false, gomock.Any()).Times(1)
+	mockOFClient.EXPECT().InstallEndpointFlows(binding.ProtocolTCP, gomock.Any()).Times(1)
+	mockOFClient.EXPECT().InstallServiceFlows(groupIDv4, svcIPv4, uint16(svcPort), binding.ProtocolTCP, uint16(0)).Times(1)
+
+	mockOFClient.EXPECT().InstallServiceGroup(groupIDv6, false, gomock.Any()).Times(1)
+	mockOFClient.EXPECT().InstallEndpointFlows(binding.ProtocolTCPv6, gomock.Any()).Times(1)
+	mockOFClient.EXPECT().InstallServiceFlows(groupIDv6, svcIPv6, uint16(svcPort), binding.ProtocolTCPv6, uint16(0)).Times(1)
+
+	fpv4.syncProxyRules()
+	fpv6.syncProxyRules()
+}
+
 func testClusterIPRemoval(t *testing.T, svcIP net.IP, epIP net.IP, isIPv6 bool) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()

pkg/agent/proxy/service.go

@@ -31,8 +31,8 @@ type serviceChangesTracker struct {
 	initialized bool
 }
 
-func newServiceChangesTracker(recorder record.EventRecorder, isIPv6 bool) *serviceChangesTracker {
-	return &serviceChangesTracker{tracker: k8sproxy.NewServiceChangeTracker(types.NewServiceInfo, &isIPv6, recorder)}
+func newServiceChangesTracker(recorder record.EventRecorder, ipFamily v1.IPFamily) *serviceChangesTracker {
+	return &serviceChangesTracker{tracker: k8sproxy.NewServiceChangeTracker(types.NewServiceInfo, ipFamily, recorder, nil)}
 }
 
 func (sh *serviceChangesTracker) OnServiceSynced() {
@@ -53,5 +53,5 @@ func (sh *serviceChangesTracker) Synced() bool {
 }
 
 func (sh *serviceChangesTracker) Update(serviceMap k8sproxy.ServiceMap) k8sproxy.UpdateServiceMapResult {
-	return k8sproxy.UpdateServiceMap(serviceMap, sh.tracker)
+	return serviceMap.Update(sh.tracker)
 }

pkg/agent/proxy/types/types.go

@@ -32,7 +32,7 @@ type ServiceInfo struct {
 // NewServiceInfo returns a new k8sproxy.ServicePort which abstracts a serviceInfo.
 func NewServiceInfo(port *corev1.ServicePort, service *corev1.Service, baseInfo *k8sproxy.BaseServiceInfo) k8sproxy.ServicePort {
 	info := &ServiceInfo{BaseServiceInfo: baseInfo}
-	if utilnet.IsIPv6String(service.Spec.ClusterIP) {
+	if utilnet.IsIPv6(baseInfo.ClusterIP()) {
 		info.OFProtocol = openflow.ProtocolTCPv6
 		if port.Protocol == corev1.ProtocolUDP {
 			info.OFProtocol = openflow.ProtocolUDPv6

third_party/proxy/doc.go

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 // Package proxy is copied from
-// k8s.io/kubernetes@/v1.17.6(https://github.com/kubernetes/kubernetes/tree/v1.17.6)
+// k8s.io/kubernetes@/v1.21.0(https://github.com/kubernetes/kubernetes/tree/v1.21.0)
 // to avoid importing the whole kubernetes repo. Some unneeded functions are removed.
 
 package proxy

third_party/proxy/meta_proxier.go

@@ -69,35 +69,21 @@ func (proxier *metaProxier) SyncLoop() {
 
 // OnServiceAdd is called whenever creation of new service object is observed.
 func (proxier *metaProxier) OnServiceAdd(service *v1.Service) {
-	ipfamily := service.Spec.IPFamilies[0]
-	if ipfamily == v1.IPv4Protocol {
-		proxier.ipv4Proxier.OnServiceAdd(service)
-		return
-	}
+	proxier.ipv4Proxier.OnServiceAdd(service)
 	proxier.ipv6Proxier.OnServiceAdd(service)
 }
 
 // OnServiceUpdate is called whenever modification of an existing
 // service object is observed.
 func (proxier *metaProxier) OnServiceUpdate(oldService, service *v1.Service) {
-	// IPFamily is immutable, hence we only need to check on the new service
-	ipfamily := service.Spec.IPFamilies[0]
-	if ipfamily == v1.IPv4Protocol {
-		proxier.ipv4Proxier.OnServiceUpdate(oldService, service)
-		return
-	}
-
+	proxier.ipv4Proxier.OnServiceUpdate(oldService, service)
 	proxier.ipv6Proxier.OnServiceUpdate(oldService, service)
 }
 
 // OnServiceDelete is called whenever deletion of an existing service
 // object is observed.
 func (proxier *metaProxier) OnServiceDelete(service *v1.Service) {
-	ipfamily := service.Spec.IPFamilies[0]
-	if ipfamily == v1.IPv4Protocol {
-		proxier.ipv4Proxier.OnServiceDelete(service)
-		return
-	}
+	proxier.ipv4Proxier.OnServiceDelete(service)
 	proxier.ipv6Proxier.OnServiceDelete(service)
 }