Changes to WinKernel to support EndpointSlices

cmd/kube-proxy/app/server_windows.go

@@ -174,7 +174,7 @@ func newProxyServer(config *proxyconfigapi.KubeProxyConfiguration, cleanupAndExi
 		OOMScoreAdj:         config.OOMScoreAdj,
 		ConfigSyncPeriod:    config.ConfigSyncPeriod.Duration,
 		HealthzServer:       healthzServer,
-		UseEndpointSlices:   false,
+		UseEndpointSlices:   utilfeature.DefaultFeatureGate.Enabled(features.EndpointSliceProxying),
 	}, nil
 }
 

pkg/proxy/winkernel/BUILD

@@ -15,19 +15,20 @@ go_library(
         "//staging/src/k8s.io/component-base/metrics/legacyregistry:go_default_library",
     ] + select({
         "@io_bazel_rules_go//go/platform:windows": [
-            "//pkg/api/v1/service:go_default_library",
             "//pkg/apis/core/v1/helper:go_default_library",
+            "//pkg/features:go_default_library",
             "//pkg/proxy:go_default_library",
             "//pkg/proxy/apis/config:go_default_library",
             "//pkg/proxy/config:go_default_library",
             "//pkg/proxy/healthcheck:go_default_library",
             "//pkg/proxy/metaproxier:go_default_library",
             "//pkg/proxy/metrics:go_default_library",
+            "//pkg/proxy/util:go_default_library",
             "//pkg/util/async:go_default_library",
             "//staging/src/k8s.io/api/core/v1:go_default_library",
+            "//staging/src/k8s.io/api/discovery/v1beta1:go_default_library",
             "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
             "//staging/src/k8s.io/apimachinery/pkg/util/intstr:go_default_library",
-            "//staging/src/k8s.io/apimachinery/pkg/util/sets:go_default_library",
             "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
             "//staging/src/k8s.io/apiserver/pkg/features:go_default_library",
             "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
@@ -67,11 +68,14 @@ go_test(
         "@io_bazel_rules_go//go/platform:windows": [
             "//pkg/proxy:go_default_library",
             "//pkg/proxy/healthcheck:go_default_library",
+            "//pkg/proxy/util:go_default_library",
             "//staging/src/k8s.io/api/core/v1:go_default_library",
             "//staging/src/k8s.io/api/discovery/v1beta1:go_default_library",
             "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
             "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
+            "//staging/src/k8s.io/apimachinery/pkg/util/intstr:go_default_library",
             "//vendor/github.com/Microsoft/hcsshim/hcn:go_default_library",
+            "//vendor/k8s.io/utils/pointer:go_default_library",
         ],
         "//conditions:default": [],
     }),

pkg/proxy/winkernel/proxier_test.go

@@ -19,18 +19,20 @@ limitations under the License.
 package winkernel
 
 import (
+	"fmt"
 	"k8s.io/api/core/v1"
 	discovery "k8s.io/api/discovery/v1beta1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/kubernetes/pkg/proxy"
 	"k8s.io/kubernetes/pkg/proxy/healthcheck"
-
+	utilproxy "k8s.io/kubernetes/pkg/proxy/util"
+	utilpointer "k8s.io/utils/pointer"
 	"net"
 	"strings"
 	"testing"
 	"time"
-
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 const testHostName = "test-hostname"
@@ -99,7 +101,7 @@ func (hns fakeHNS) getLoadBalancer(endpoints []endpointsInfo, flags loadBalancer
 func (hns fakeHNS) deleteLoadBalancer(hnsID string) error {
 	return nil
 }
-func NewFakeProxier(syncPeriod time.Duration, minSyncPeriod time.Duration, clusterCIDR string, hostname string, nodeIP net.IP, networkType string) *Proxier {
+func NewFakeProxier(syncPeriod time.Duration, minSyncPeriod time.Duration, clusterCIDR string, hostname string, nodeIP net.IP, networkType string, endpointSliceEnabled bool) *Proxier {
 	sourceVip := "192.168.1.2"
 	hnsNetworkInfo := &hnsNetworkInfo{
 		id:          strings.ToUpper(guid),
@@ -107,11 +109,9 @@ func NewFakeProxier(syncPeriod time.Duration, minSyncPeriod time.Duration, clust
 		networkType: networkType,
 	}
 	proxier := &Proxier{
-		portsMap:            make(map[localPort]closeable),
+		portsMap:            make(map[utilproxy.LocalPort]utilproxy.Closeable),
-		serviceMap:          make(proxyServiceMap),
+		serviceMap:          make(proxy.ServiceMap),
-		serviceChanges:      newServiceChangeMap(),
+		endpointsMap:        make(proxy.EndpointsMap),
-		endpointsMap:        make(proxyEndpointsMap),
-		endpointsChanges:    newEndpointsChangeMap(hostname),
 		clusterCIDR:         clusterCIDR,
 		hostname:            testHostName,
 		nodeIP:              nodeIP,
@@ -123,12 +123,19 @@ func NewFakeProxier(syncPeriod time.Duration, minSyncPeriod time.Duration, clust
 		hns:                 newFakeHNS(),
 		endPointsRefCount:   make(endPointsReferenceCountMap),
 	}
+
+	isIPv6 := false
+	serviceChanges := proxy.NewServiceChangeTracker(proxier.newServiceInfo, &isIPv6, nil, proxier.serviceMapChange)
+	endpointChangeTracker := proxy.NewEndpointChangeTracker(hostname, proxier.newEndpointInfo, &isIPv6, nil, endpointSliceEnabled, proxier.endpointsMapChange)
+	proxier.endpointsChanges = endpointChangeTracker
+	proxier.serviceChanges = serviceChanges
+
 	return proxier
 }
 
 func TestCreateServiceVip(t *testing.T) {
 	syncPeriod := 30 * time.Second
-	proxier := NewFakeProxier(syncPeriod, syncPeriod, clusterCIDR, "testhost", net.ParseIP("10.0.0.1"), "Overlay")
+	proxier := NewFakeProxier(syncPeriod, syncPeriod, clusterCIDR, "testhost", net.ParseIP("10.0.0.1"), "Overlay", false)
 	if proxier == nil {
 		t.Error()
 	}
@@ -140,6 +147,7 @@ func TestCreateServiceVip(t *testing.T) {
 	svcPortName := proxy.ServicePortName{
 		NamespacedName: makeNSN("ns1", "svc1"),
 		Port:           "p80",
+		Protocol:       v1.ProtocolTCP,
 	}
 	timeoutSeconds := v1.DefaultClientIPServiceAffinitySeconds
 
@@ -163,19 +171,26 @@ func TestCreateServiceVip(t *testing.T) {
 		}),
 	)
 	makeEndpointsMap(proxier)
-
+	proxier.setInitialized(true)
 	proxier.syncProxyRules()
 
-	if proxier.serviceMap[svcPortName].remoteEndpoint == nil {
+	svc := proxier.serviceMap[svcPortName]
+	svcInfo, ok := svc.(*serviceInfo)
+	if !ok {
+		t.Errorf("Failed to cast serviceInfo %q", svcPortName.String())
+
+	} else {
+		if svcInfo.remoteEndpoint == nil {
 			t.Error()
 		}
-	if proxier.serviceMap[svcPortName].remoteEndpoint.ip != svcIP {
+		if svcInfo.remoteEndpoint.ip != svcIP {
 			t.Error()
 		}
+	}
 }
 func TestCreateRemoteEndpointOverlay(t *testing.T) {
 	syncPeriod := 30 * time.Second
-	proxier := NewFakeProxier(syncPeriod, syncPeriod, clusterCIDR, "testhost", net.ParseIP("10.0.0.1"), "Overlay")
+	proxier := NewFakeProxier(syncPeriod, syncPeriod, clusterCIDR, "testhost", net.ParseIP("10.0.0.1"), "Overlay", false)
 	if proxier == nil {
 		t.Error()
 	}
@@ -186,6 +201,7 @@ func TestCreateRemoteEndpointOverlay(t *testing.T) {
 	svcPortName := proxy.ServicePortName{
 		NamespacedName: makeNSN("ns1", "svc1"),
 		Port:           "p80",
+		Protocol:       v1.ProtocolTCP,
 	}
 
 	makeServiceMap(proxier,
@@ -209,28 +225,36 @@ func TestCreateRemoteEndpointOverlay(t *testing.T) {
 				Ports: []v1.EndpointPort{{
 					Name:     svcPortName.Port,
 					Port:     int32(svcPort),
+					Protocol: v1.ProtocolTCP,
 				}},
 			}}
 		}),
 	)
-
+	proxier.setInitialized(true)
 	proxier.syncProxyRules()
 
-	if proxier.endpointsMap[svcPortName][0].hnsID != guid {
+	ep := proxier.endpointsMap[svcPortName][0]
-		t.Errorf("%v does not match %v", proxier.endpointsMap[svcPortName][0].hnsID, guid)
+	epInfo, ok := ep.(*endpointsInfo)
+	if !ok {
+		t.Errorf("Failed to cast endpointsInfo %q", svcPortName.String())
+
+	} else {
+		if epInfo.hnsID != guid {
+			t.Errorf("%v does not match %v", epInfo.hnsID, guid)
+		}
 	}
 
 	if *proxier.endPointsRefCount[guid] <= 0 {
 		t.Errorf("RefCount not incremented. Current value: %v", *proxier.endPointsRefCount[guid])
 	}
 
-	if *proxier.endPointsRefCount[guid] != *proxier.endpointsMap[svcPortName][0].refCount {
+	if *proxier.endPointsRefCount[guid] != *epInfo.refCount {
-		t.Errorf("Global refCount: %v does not match endpoint refCount: %v", *proxier.endPointsRefCount[guid], *proxier.endpointsMap[svcPortName][0].refCount)
+		t.Errorf("Global refCount: %v does not match endpoint refCount: %v", *proxier.endPointsRefCount[guid], *epInfo.refCount)
 	}
 }
 func TestCreateRemoteEndpointL2Bridge(t *testing.T) {
 	syncPeriod := 30 * time.Second
-	proxier := NewFakeProxier(syncPeriod, syncPeriod, clusterCIDR, "testhost", net.ParseIP("10.0.0.1"), "L2Bridge")
+	proxier := NewFakeProxier(syncPeriod, syncPeriod, clusterCIDR, "testhost", net.ParseIP("10.0.0.1"), "L2Bridge", false)
 	if proxier == nil {
 		t.Error()
 	}
@@ -241,6 +265,7 @@ func TestCreateRemoteEndpointL2Bridge(t *testing.T) {
 	svcPortName := proxy.ServicePortName{
 		NamespacedName: makeNSN("ns1", "svc1"),
 		Port:           "p80",
+		Protocol:       v1.ProtocolTCP,
 	}
 
 	makeServiceMap(proxier,
@@ -264,29 +289,36 @@ func TestCreateRemoteEndpointL2Bridge(t *testing.T) {
 				Ports: []v1.EndpointPort{{
 					Name:     svcPortName.Port,
 					Port:     int32(svcPort),
+					Protocol: v1.ProtocolTCP,
 				}},
 			}}
 		}),
 	)
-
+	proxier.setInitialized(true)
 	proxier.syncProxyRules()
+	ep := proxier.endpointsMap[svcPortName][0]
+	epInfo, ok := ep.(*endpointsInfo)
+	if !ok {
+		t.Errorf("Failed to cast endpointsInfo %q", svcPortName.String())
 
-	if proxier.endpointsMap[svcPortName][0].hnsID != guid {
+	} else {
-		t.Errorf("%v does not match %v", proxier.endpointsMap[svcPortName][0].hnsID, guid)
+		if epInfo.hnsID != guid {
+			t.Errorf("%v does not match %v", epInfo.hnsID, guid)
+		}
 	}
 
 	if *proxier.endPointsRefCount[guid] <= 0 {
 		t.Errorf("RefCount not incremented. Current value: %v", *proxier.endPointsRefCount[guid])
 	}
 
-	if *proxier.endPointsRefCount[guid] != *proxier.endpointsMap[svcPortName][0].refCount {
+	if *proxier.endPointsRefCount[guid] != *epInfo.refCount {
-		t.Errorf("Global refCount: %v does not match endpoint refCount: %v", *proxier.endPointsRefCount[guid], *proxier.endpointsMap[svcPortName][0].refCount)
+		t.Errorf("Global refCount: %v does not match endpoint refCount: %v", *proxier.endPointsRefCount[guid], *epInfo.refCount)
 	}
 }
 
 func TestCreateLoadBalancer(t *testing.T) {
 	syncPeriod := 30 * time.Second
-	proxier := NewFakeProxier(syncPeriod, syncPeriod, clusterCIDR, "testhost", net.ParseIP("10.0.0.1"), "Overlay")
+	proxier := NewFakeProxier(syncPeriod, syncPeriod, clusterCIDR, "testhost", net.ParseIP("10.0.0.1"), "Overlay", false)
 	if proxier == nil {
 		t.Error()
 	}
@@ -297,6 +329,7 @@ func TestCreateLoadBalancer(t *testing.T) {
 	svcPortName := proxy.ServicePortName{
 		NamespacedName: makeNSN("ns1", "svc1"),
 		Port:           "p80",
+		Protocol:       v1.ProtocolTCP,
 	}
 
 	makeServiceMap(proxier,
@@ -320,18 +353,99 @@ func TestCreateLoadBalancer(t *testing.T) {
 				Ports: []v1.EndpointPort{{
 					Name:     svcPortName.Port,
 					Port:     int32(svcPort),
+					Protocol: v1.ProtocolTCP,
 				}},
 			}}
 		}),
 	)
 
+	proxier.setInitialized(true)
 	proxier.syncProxyRules()
 
-	if proxier.serviceMap[svcPortName].hnsID != guid {
+	svc := proxier.serviceMap[svcPortName]
-		t.Errorf("%v does not match %v", proxier.serviceMap[svcPortName].hnsID, guid)
+	svcInfo, ok := svc.(*serviceInfo)
+	if !ok {
+		t.Errorf("Failed to cast serviceInfo %q", svcPortName.String())
+
+	} else {
+		if svcInfo.hnsID != guid {
+			t.Errorf("%v does not match %v", svcInfo.hnsID, guid)
+		}
+	}
+
+}
+func TestEndpointSlice(t *testing.T) {
+	syncPeriod := 30 * time.Second
+	proxier := NewFakeProxier(syncPeriod, syncPeriod, clusterCIDR, "testhost", net.ParseIP("10.0.0.1"), "Overlay", true)
+	if proxier == nil {
+		t.Error()
+	}
+
+	proxier.servicesSynced = true
+	proxier.endpointSlicesSynced = true
+
+	svcPortName := proxy.ServicePortName{
+		NamespacedName: makeNSN("ns1", "svc1"),
+		Port:           "p80",
+		Protocol:       v1.ProtocolTCP,
+	}
+
+	proxier.OnServiceAdd(&v1.Service{
+		ObjectMeta: metav1.ObjectMeta{Name: svcPortName.Name, Namespace: svcPortName.Namespace},
+		Spec: v1.ServiceSpec{
+			ClusterIP: "172.20.1.1",
+			Selector:  map[string]string{"foo": "bar"},
+			Ports:     []v1.ServicePort{{Name: svcPortName.Port, TargetPort: intstr.FromInt(80), Protocol: v1.ProtocolTCP}},
+		},
+	})
+
+	// Add initial endpoint slice
+	tcpProtocol := v1.ProtocolTCP
+	endpointSlice := &discovery.EndpointSlice{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      fmt.Sprintf("%s-1", svcPortName.Name),
+			Namespace: svcPortName.Namespace,
+			Labels:    map[string]string{discovery.LabelServiceName: svcPortName.Name},
+		},
+		Ports: []discovery.EndpointPort{{
+			Name:     &svcPortName.Port,
+			Port:     utilpointer.Int32Ptr(80),
+			Protocol: &tcpProtocol,
+		}},
+		AddressType: discovery.AddressTypeIPv4,
+		Endpoints: []discovery.Endpoint{{
+			Addresses:  []string{"192.168.2.3"},
+			Conditions: discovery.EndpointConditions{Ready: utilpointer.BoolPtr(true)},
+			Topology:   map[string]string{"kubernetes.io/hostname": "testhost2"},
+		}},
+	}
+
+	proxier.OnEndpointSliceAdd(endpointSlice)
+	proxier.setInitialized(true)
+	proxier.syncProxyRules()
+
+	svc := proxier.serviceMap[svcPortName]
+	svcInfo, ok := svc.(*serviceInfo)
+	if !ok {
+		t.Errorf("Failed to cast serviceInfo %q", svcPortName.String())
+
+	} else {
+		if svcInfo.hnsID != guid {
+			t.Errorf("The Hns Loadbalancer Id %v does not match %v. ServicePortName %q", svcInfo.hnsID, guid, svcPortName.String())
+		}
+	}
+
+	ep := proxier.endpointsMap[svcPortName][0]
+	epInfo, ok := ep.(*endpointsInfo)
+	if !ok {
+		t.Errorf("Failed to cast endpointsInfo %q", svcPortName.String())
+
+	} else {
+		if epInfo.hnsID != guid {
+			t.Errorf("Hns EndpointId %v does not match %v. ServicePortName %q", epInfo.hnsID, guid, svcPortName.String())
+		}
 	}
 }
-
 func TestNoopEndpointSlice(t *testing.T) {
 	p := Proxier{}
 	p.OnEndpointSliceAdd(&discovery.EndpointSlice{})