Merge pull request #56880 from MrHohn/kube-proxy-ipv6-fix

pkg/kubelet/network/hostport/BUILD

@@ -19,6 +19,7 @@ go_library(
         "//pkg/proxy/iptables:go_default_library",
         "//pkg/util/conntrack:go_default_library",
         "//pkg/util/iptables:go_default_library",
+        "//pkg/util/net:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/errors:go_default_library",

pkg/kubelet/network/hostport/hostport_manager.go

@@ -31,6 +31,7 @@ import (
 	iptablesproxy "k8s.io/kubernetes/pkg/proxy/iptables"
 	"k8s.io/kubernetes/pkg/util/conntrack"
 	utiliptables "k8s.io/kubernetes/pkg/util/iptables"
+	utilnet "k8s.io/kubernetes/pkg/util/net"
 	"k8s.io/utils/exec"
 )
 
@@ -165,7 +166,7 @@ func (hm *hostportManager) Add(id string, podPortMapping *PodPortMapping, natInt
 		// clean up opened host port if encounter any error
 		return utilerrors.NewAggregate([]error{err, hm.closeHostports(hostportMappings)})
 	}
-	isIpv6 := conntrack.IsIPv6(podPortMapping.IP)
+	isIpv6 := utilnet.IsIPv6(podPortMapping.IP)
 
 	// Remove conntrack entries just after adding the new iptables rules. If the conntrack entry is removed along with
 	// the IP tables rule, it can be the case that the packets received by the node after iptables rule removal will

pkg/proxy/BUILD

@@ -16,11 +16,15 @@ go_library(
     ],
     importpath = "k8s.io/kubernetes/pkg/proxy",
     deps = [
+        "//pkg/api/service:go_default_library",
         "//pkg/apis/core:go_default_library",
+        "//pkg/apis/core/helper:go_default_library",
         "//pkg/proxy/util:go_default_library",
+        "//pkg/util/net:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
+        "//vendor/k8s.io/client-go/tools/record:go_default_library",
     ],
 )
 
@@ -57,7 +61,6 @@ go_test(
     ],
     embed = [":go_default_library"],
     deps = [
-        "//pkg/api/service:go_default_library",
         "//pkg/apis/core:go_default_library",
         "//vendor/github.com/davecgh/go-spew/spew:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",

pkg/proxy/endpoints.go

@@ -17,16 +17,67 @@ limitations under the License.
 package proxy
 
 import (
+	"net"
 	"reflect"
+	"strconv"
 	"sync"
 
 	"github.com/golang/glog"
 
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/client-go/tools/record"
 	api "k8s.io/kubernetes/pkg/apis/core"
+	utilproxy "k8s.io/kubernetes/pkg/proxy/util"
+	utilnet "k8s.io/kubernetes/pkg/util/net"
 )
 
+// BaseEndpointInfo contains base information that defines an endpoint.
+// This could be used directly by proxier while processing endpoints,
+// or can be used for constructing a more specific EndpointInfo struct
+// defined by the proxier if needed.
+type BaseEndpointInfo struct {
+	Endpoint string // TODO: should be an endpointString type
+	// IsLocal indicates whether the endpoint is running in same host as kube-proxy.
+	IsLocal bool
+}
+
+var _ Endpoint = &BaseEndpointInfo{}
+
+// String is part of proxy.Endpoint interface.
+func (info *BaseEndpointInfo) String() string {
+	return info.Endpoint
+}
+
+// GetIsLocal is part of proxy.Endpoint interface.
+func (info *BaseEndpointInfo) GetIsLocal() bool {
+	return info.IsLocal
+}
+
+// IP returns just the IP part of the endpoint, it's a part of proxy.Endpoint interface.
+func (info *BaseEndpointInfo) IP() string {
+	return utilproxy.IPPart(info.Endpoint)
+}
+
+// Port returns just the Port part of the endpoint.
+func (info *BaseEndpointInfo) Port() (int, error) {
+	return utilproxy.PortPart(info.Endpoint)
+}
+
+// Equal is part of proxy.Endpoint interface.
+func (info *BaseEndpointInfo) Equal(other Endpoint) bool {
+	return info.String() == other.String() && info.GetIsLocal() == other.GetIsLocal()
+}
+
+func newBaseEndpointInfo(IP string, port int, isLocal bool) *BaseEndpointInfo {
+	return &BaseEndpointInfo{
+		Endpoint: net.JoinHostPort(IP, strconv.Itoa(port)),
+		IsLocal:  isLocal,
+	}
+}
+
+type makeEndpointFunc func(info *BaseEndpointInfo) Endpoint
+
 // EndpointChangeTracker carries state about uncommitted changes to an arbitrary number of
 // Endpoints, keyed by their namespace and name.
 type EndpointChangeTracker struct {
@@ -36,13 +87,21 @@ type EndpointChangeTracker struct {
 	hostname string
 	// items maps a service to is endpointsChange.
 	items map[types.NamespacedName]*endpointsChange
+	// makeEndpointInfo allows proxier to inject customized information when processing endpoint.
+	makeEndpointInfo makeEndpointFunc
+	// isIPv6Mode indicates if change tracker is under IPv6/IPv4 mode. Nil means not applicable.
+	isIPv6Mode *bool
+	recorder   record.EventRecorder
 }
 
 // NewEndpointChangeTracker initializes an EndpointsChangeMap
-func NewEndpointChangeTracker(hostname string) *EndpointChangeTracker {
+func NewEndpointChangeTracker(hostname string, makeEndpointInfo makeEndpointFunc, isIPv6Mode *bool, recorder record.EventRecorder) *EndpointChangeTracker {
 	return &EndpointChangeTracker{
 		hostname:         hostname,
 		items:            make(map[types.NamespacedName]*endpointsChange),
+		makeEndpointInfo: makeEndpointInfo,
+		isIPv6Mode:       isIPv6Mode,
+		recorder:         recorder,
 	}
 }
 
@@ -54,7 +113,7 @@ func NewEndpointChangeTracker(hostname string) *EndpointChangeTracker {
 //   - pass <oldEndpoints, endpoints> as the <previous, current> pair.
 // Delete item
 //   - pass <endpoints, nil> as the <previous, current> pair.
-func (ect *EndpointChangeTracker) Update(previous, current *api.Endpoints, makeEndpoints func(IP string, port int, isLocal bool) Endpoint) bool {
+func (ect *EndpointChangeTracker) Update(previous, current *api.Endpoints) bool {
 	endpoints := current
 	if endpoints == nil {
 		endpoints = previous
@@ -71,10 +130,10 @@ func (ect *EndpointChangeTracker) Update(previous, current *api.Endpoints, makeE
 	change, exists := ect.items[namespacedName]
 	if !exists {
 		change = &endpointsChange{}
-		change.previous = endpointsToEndpointsMap(previous, ect.hostname, makeEndpoints)
+		change.previous = ect.endpointsToEndpointsMap(previous)
 		ect.items[namespacedName] = change
 	}
-	change.current = endpointsToEndpointsMap(current, ect.hostname, makeEndpoints)
+	change.current = ect.endpointsToEndpointsMap(current)
 	// if change.previous equal to change.current, it means no change
 	if reflect.DeepEqual(change.previous, change.current) {
 		delete(ect.items, namespacedName)
@@ -118,14 +177,14 @@ func UpdateEndpointsMap(endpointsMap EndpointsMap, changes *EndpointChangeTracke
 	return result
 }
 
-// EndpointsMap maps a service to one of its endpoint.
+// EndpointsMap maps a service name to a list of all its Endpoints.
 type EndpointsMap map[ServicePortName][]Endpoint
 
 // endpointsToEndpointsMap translates single Endpoints object to EndpointsMap.
 // This function is used for incremental updated of endpointsMap.
 //
 // NOTE: endpoints object should NOT be modified.
-func endpointsToEndpointsMap(endpoints *api.Endpoints, hostname string, makeEndpoints func(IP string, port int, isLocal bool) Endpoint) EndpointsMap {
+func (ect *EndpointChangeTracker) endpointsToEndpointsMap(endpoints *api.Endpoints) EndpointsMap {
 	if endpoints == nil {
 		return nil
 	}
@@ -151,9 +210,21 @@ func endpointsToEndpointsMap(endpoints *api.Endpoints, hostname string, makeEndp
 					glog.Warningf("ignoring invalid endpoint port %s with empty host", port.Name)
 					continue
 				}
-				isLocal := addr.NodeName != nil && *addr.NodeName == hostname
-				epInfo := makeEndpoints(addr.IP, int(port.Port), isLocal)
-				endpointsMap[svcPortName] = append(endpointsMap[svcPortName], epInfo)
+				// Filter out the incorrect IP version case.
+				// Any endpoint port that contains incorrect IP version will be ignored.
+				if ect.isIPv6Mode != nil && utilnet.IsIPv6String(addr.IP) != *ect.isIPv6Mode {
+					// Emit event on the corresponding service which had a different
+					// IP version than the endpoint.
+					utilproxy.LogAndEmitIncorrectIPVersionEvent(ect.recorder, "endpoints", addr.IP, endpoints.Name, endpoints.Namespace, "")
+					continue
+				}
+				isLocal := addr.NodeName != nil && *addr.NodeName == ect.hostname
+				baseEndpointInfo := newBaseEndpointInfo(addr.IP, int(port.Port), isLocal)
+				if ect.makeEndpointInfo != nil {
+					endpointsMap[svcPortName] = append(endpointsMap[svcPortName], ect.makeEndpointInfo(baseEndpointInfo))
+				} else {
+					endpointsMap[svcPortName] = append(endpointsMap[svcPortName], baseEndpointInfo)
+				}
 			}
 			if glog.V(3) {
 				newEPList := []string{}
@@ -203,7 +274,7 @@ func GetLocalEndpointIPs(endpointsMap EndpointsMap) map[types.NamespacedName]set
 	localIPs := make(map[types.NamespacedName]sets.String)
 	for svcPortName, epList := range endpointsMap {
 		for _, ep := range epList {
-			if ep.IsLocal() {
+			if ep.GetIsLocal() {
 				nsn := svcPortName.NamespacedName
 				if localIPs[nsn] == nil {
 					localIPs[nsn] = sets.NewString()

pkg/proxy/iptables/BUILD

@@ -13,9 +13,7 @@ go_library(
     ],
     importpath = "k8s.io/kubernetes/pkg/proxy/iptables",
     deps = [
-        "//pkg/api/service:go_default_library",
         "//pkg/apis/core:go_default_library",
-        "//pkg/apis/core/helper:go_default_library",
         "//pkg/proxy:go_default_library",
         "//pkg/proxy/healthcheck:go_default_library",
         "//pkg/proxy/metrics:go_default_library",
@@ -23,6 +21,7 @@ go_library(
         "//pkg/util/async:go_default_library",
         "//pkg/util/conntrack:go_default_library",
         "//pkg/util/iptables:go_default_library",
+        "//pkg/util/net:go_default_library",
         "//pkg/util/sysctl:go_default_library",
         "//pkg/util/version:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",

pkg/proxy/ipvs/BUILD

@@ -78,9 +78,7 @@ go_library(
     }),
     importpath = "k8s.io/kubernetes/pkg/proxy/ipvs",
     deps = [
-        "//pkg/api/service:go_default_library",
         "//pkg/apis/core:go_default_library",
-        "//pkg/apis/core/helper:go_default_library",
         "//pkg/proxy:go_default_library",
         "//pkg/proxy/healthcheck:go_default_library",
         "//pkg/proxy/metrics:go_default_library",
@@ -90,6 +88,7 @@ go_library(
         "//pkg/util/ipset:go_default_library",
         "//pkg/util/iptables:go_default_library",
         "//pkg/util/ipvs:go_default_library",
+        "//pkg/util/net:go_default_library",
         "//pkg/util/sysctl:go_default_library",
         "//pkg/util/version:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",

pkg/proxy/service.go

@@ -17,17 +17,120 @@ limitations under the License.
 package proxy
 
 import (
+	"fmt"
+	"net"
 	"reflect"
+	"strings"
 	"sync"
 
 	"github.com/golang/glog"
 
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/client-go/tools/record"
+	apiservice "k8s.io/kubernetes/pkg/api/service"
 	api "k8s.io/kubernetes/pkg/apis/core"
-	proxyutil "k8s.io/kubernetes/pkg/proxy/util"
+	"k8s.io/kubernetes/pkg/apis/core/helper"
+	utilproxy "k8s.io/kubernetes/pkg/proxy/util"
+	utilnet "k8s.io/kubernetes/pkg/util/net"
 )
 
+// BaseServiceInfo contains base information that defines a service.
+// This could be used directly by proxier while processing services,
+// or can be used for constructing a more specific ServiceInfo struct
+// defined by the proxier if needed.
+type BaseServiceInfo struct {
+	ClusterIP                net.IP
+	Port                     int
+	Protocol                 api.Protocol
+	NodePort                 int
+	LoadBalancerStatus       api.LoadBalancerStatus
+	SessionAffinityType      api.ServiceAffinity
+	StickyMaxAgeSeconds      int
+	ExternalIPs              []string
+	LoadBalancerSourceRanges []string
+	HealthCheckNodePort      int
+	OnlyNodeLocalEndpoints   bool
+}
+
+var _ ServicePort = &BaseServiceInfo{}
+
+// String is part of ServicePort interface.
+func (info *BaseServiceInfo) String() string {
+	return fmt.Sprintf("%s:%d/%s", info.ClusterIP, info.Port, info.Protocol)
+}
+
+// ClusterIPString is part of ServicePort interface.
+func (info *BaseServiceInfo) ClusterIPString() string {
+	return info.ClusterIP.String()
+}
+
+// GetProtocol is part of ServicePort interface.
+func (info *BaseServiceInfo) GetProtocol() api.Protocol {
+	return info.Protocol
+}
+
+// GetHealthCheckNodePort is part of ServicePort interface.
+func (info *BaseServiceInfo) GetHealthCheckNodePort() int {
+	return info.HealthCheckNodePort
+}
+
+func (sct *ServiceChangeTracker) newBaseServiceInfo(port *api.ServicePort, service *api.Service) *BaseServiceInfo {
+	onlyNodeLocalEndpoints := false
+	if apiservice.RequestsOnlyLocalTraffic(service) {
+		onlyNodeLocalEndpoints = true
+	}
+	var stickyMaxAgeSeconds int
+	if service.Spec.SessionAffinity == api.ServiceAffinityClientIP {
+		// Kube-apiserver side guarantees SessionAffinityConfig won't be nil when session affinity type is ClientIP
+		stickyMaxAgeSeconds = int(*service.Spec.SessionAffinityConfig.ClientIP.TimeoutSeconds)
+	}
+	info := &BaseServiceInfo{
+		ClusterIP: net.ParseIP(service.Spec.ClusterIP),
+		Port:      int(port.Port),
+		Protocol:  port.Protocol,
+		NodePort:  int(port.NodePort),
+		// Deep-copy in case the service instance changes
+		LoadBalancerStatus:     *helper.LoadBalancerStatusDeepCopy(&service.Status.LoadBalancer),
+		SessionAffinityType:    service.Spec.SessionAffinity,
+		StickyMaxAgeSeconds:    stickyMaxAgeSeconds,
+		OnlyNodeLocalEndpoints: onlyNodeLocalEndpoints,
+	}
+
+	if sct.isIPv6Mode == nil {
+		info.ExternalIPs = make([]string, len(service.Spec.ExternalIPs))
+		info.LoadBalancerSourceRanges = make([]string, len(service.Spec.LoadBalancerSourceRanges))
+		copy(info.LoadBalancerSourceRanges, service.Spec.LoadBalancerSourceRanges)
+		copy(info.ExternalIPs, service.Spec.ExternalIPs)
+	} else {
+		// Filter out the incorrect IP version case.
+		// If ExternalIPs and LoadBalancerSourceRanges on service contains incorrect IP versions,
+		// only filter out the incorrect ones.
+		var incorrectIPs []string
+		info.ExternalIPs, incorrectIPs = utilnet.FilterIncorrectIPVersion(service.Spec.ExternalIPs, *sct.isIPv6Mode)
+		if len(incorrectIPs) > 0 {
+			utilproxy.LogAndEmitIncorrectIPVersionEvent(sct.recorder, "externalIPs", strings.Join(incorrectIPs, ","), service.Namespace, service.Name, service.UID)
+		}
+		info.LoadBalancerSourceRanges, incorrectIPs = utilnet.FilterIncorrectCIDRVersion(service.Spec.LoadBalancerSourceRanges, *sct.isIPv6Mode)
+		if len(incorrectIPs) > 0 {
+			utilproxy.LogAndEmitIncorrectIPVersionEvent(sct.recorder, "loadBalancerSourceRanges", strings.Join(incorrectIPs, ","), service.Namespace, service.Name, service.UID)
+		}
+	}
+
+	if apiservice.NeedsHealthCheck(service) {
+		p := service.Spec.HealthCheckNodePort
+		if p == 0 {
+			glog.Errorf("Service %s/%s has no healthcheck nodeport", service.Namespace, service.Name)
+		} else {
+			info.HealthCheckNodePort = int(p)
+		}
+	}
+
+	return info
+}
+
+type makeServicePortFunc func(*api.ServicePort, *api.Service, *BaseServiceInfo) ServicePort
+
 // serviceChange contains all changes to services that happened since proxy rules were synced.  For a single object,
 // changes are accumulated, i.e. previous is state from before applying the changes,
 // current is state after applying all of the changes.
@@ -43,12 +146,20 @@ type ServiceChangeTracker struct {
 	lock sync.Mutex
 	// items maps a service to its serviceChange.
 	items map[types.NamespacedName]*serviceChange
+	// makeServiceInfo allows proxier to inject customized information when processing service.
+	makeServiceInfo makeServicePortFunc
+	// isIPv6Mode indicates if change tracker is under IPv6/IPv4 mode. Nil means not applicable.
+	isIPv6Mode *bool
+	recorder   record.EventRecorder
 }
 
 // NewServiceChangeTracker initializes a ServiceChangeTracker
-func NewServiceChangeTracker() *ServiceChangeTracker {
+func NewServiceChangeTracker(makeServiceInfo makeServicePortFunc, isIPv6Mode *bool, recorder record.EventRecorder) *ServiceChangeTracker {
 	return &ServiceChangeTracker{
 		items:           make(map[types.NamespacedName]*serviceChange),
+		makeServiceInfo: makeServiceInfo,
+		isIPv6Mode:      isIPv6Mode,
+		recorder:        recorder,
 	}
 }
 
@@ -60,10 +171,7 @@ func NewServiceChangeTracker() *ServiceChangeTracker {
 //   - pass <oldService, service> as the <previous, current> pair.
 // Delete item
 //   - pass <service, nil> as the <previous, current> pair.
-//
-// makeServicePort() return a proxy.ServicePort based on the given Service and its ServicePort.  We inject makeServicePort()
-// so that giving caller side a chance to initialize proxy.ServicePort interface.
-func (sct *ServiceChangeTracker) Update(previous, current *api.Service, makeServicePort func(servicePort *api.ServicePort, service *api.Service) ServicePort) bool {
+func (sct *ServiceChangeTracker) Update(previous, current *api.Service) bool {
 	svc := current
 	if svc == nil {
 		svc = previous
@@ -80,10 +188,10 @@ func (sct *ServiceChangeTracker) Update(previous, current *api.Service, makeServ
 	change, exists := sct.items[namespacedName]
 	if !exists {
 		change = &serviceChange{}
-		change.previous = serviceToServiceMap(previous, makeServicePort)
+		change.previous = sct.serviceToServiceMap(previous)
 		sct.items[namespacedName] = change
 	}
-	change.current = serviceToServiceMap(current, makeServicePort)
+	change.current = sct.serviceToServiceMap(current)
 	// if change.previous equal to change.current, it means no change
 	if reflect.DeepEqual(change.previous, change.current) {
 		delete(sct.items, namespacedName)
@@ -110,36 +218,48 @@ func UpdateServiceMap(serviceMap ServiceMap, changes *ServiceChangeTracker) (res
 	// computing this incrementally similarly to serviceMap.
 	result.HCServiceNodePorts = make(map[types.NamespacedName]uint16)
 	for svcPortName, info := range serviceMap {
-		if info.HealthCheckNodePort() != 0 {
-			result.HCServiceNodePorts[svcPortName.NamespacedName] = uint16(info.HealthCheckNodePort())
+		if info.GetHealthCheckNodePort() != 0 {
+			result.HCServiceNodePorts[svcPortName.NamespacedName] = uint16(info.GetHealthCheckNodePort())
 		}
 	}
 
 	return result
 }
 
-// ServiceMap maps a service to its ServicePort information.
+// ServiceMap maps a service to its ServicePort.
 type ServiceMap map[ServicePortName]ServicePort
 
 // serviceToServiceMap translates a single Service object to a ServiceMap.
-// makeServicePort() return a proxy.ServicePort based on the given Service and its ServicePort.  We inject makeServicePort()
-// so that giving caller side a chance to initialize proxy.ServicePort interface.
 //
 // NOTE: service object should NOT be modified.
-func serviceToServiceMap(service *api.Service, makeServicePort func(servicePort *api.ServicePort, service *api.Service) ServicePort) ServiceMap {
+func (sct *ServiceChangeTracker) serviceToServiceMap(service *api.Service) ServiceMap {
 	if service == nil {
 		return nil
 	}
 	svcName := types.NamespacedName{Namespace: service.Namespace, Name: service.Name}
-	if proxyutil.ShouldSkipService(svcName, service) {
+	if utilproxy.ShouldSkipService(svcName, service) {
+		return nil
+	}
+
+	if len(service.Spec.ClusterIP) != 0 {
+		// Filter out the incorrect IP version case.
+		// If ClusterIP on service has incorrect IP version, service itself will be ignored.
+		if sct.isIPv6Mode != nil && utilnet.IsIPv6String(service.Spec.ClusterIP) != *sct.isIPv6Mode {
+			utilproxy.LogAndEmitIncorrectIPVersionEvent(sct.recorder, "clusterIP", service.Spec.ClusterIP, service.Namespace, service.Name, service.UID)
 			return nil
 		}
+	}
 
 	serviceMap := make(ServiceMap)
 	for i := range service.Spec.Ports {
 		servicePort := &service.Spec.Ports[i]
 		svcPortName := ServicePortName{NamespacedName: svcName, Port: servicePort.Name}
-		serviceMap[svcPortName] = makeServicePort(servicePort, service)
+		baseSvcInfo := sct.newBaseServiceInfo(servicePort, service)
+		if sct.makeServiceInfo != nil {
+			serviceMap[svcPortName] = sct.makeServiceInfo(servicePort, service, baseSvcInfo)
+		} else {
+			serviceMap[svcPortName] = baseSvcInfo
+		}
 	}
 	return serviceMap
 }
@@ -213,8 +333,8 @@ func (sm *ServiceMap) unmerge(other ServiceMap, UDPStaleClusterIP sets.String) {
 		info, exists := (*sm)[svcPortName]
 		if exists {
 			glog.V(1).Infof("Removing service port %q", svcPortName)
-			if info.Protocol() == api.ProtocolUDP {
-				UDPStaleClusterIP.Insert(info.ClusterIP())
+			if info.GetProtocol() == api.ProtocolUDP {
+				UDPStaleClusterIP.Insert(info.ClusterIPString())
 			}
 			delete(*sm, svcPortName)
 		} else {

pkg/proxy/types.go

@@ -48,23 +48,26 @@ func (spn ServicePortName) String() string {
 type ServicePort interface {
 	// String returns service string.  An example format can be: `IP:Port/Protocol`.
 	String() string
-	// ClusterIP returns service cluster IP.
-	ClusterIP() string
-	// Protocol returns service protocol.
-	Protocol() api.Protocol
-	// HealthCheckNodePort returns service health check node port if present.  If return 0, it means not present.
-	HealthCheckNodePort() int
+	// ClusterIPString returns service cluster IP in string format.
+	ClusterIPString() string
+	// GetProtocol returns service protocol.
+	GetProtocol() api.Protocol
+	// GetHealthCheckNodePort returns service health check node port if present.  If return 0, it means not present.
+	GetHealthCheckNodePort() int
 }
 
 // Endpoint in an interface which abstracts information about an endpoint.
+// TODO: Rename functions to be consistent with ServicePort.
 type Endpoint interface {
 	// String returns endpoint string.  An example format can be: `IP:Port`.
 	// We take the returned value as ServiceEndpoint.Endpoint.
 	String() string
-	// IsLocal returns true if the endpoint is running in same host as kube-proxy, otherwise returns false.
-	IsLocal() bool
-	// IP returns IP part of endpoints.
+	// GetIsLocal returns true if the endpoint is running in same host as kube-proxy, otherwise returns false.
+	GetIsLocal() bool
+	// IP returns IP part of the endpoint.
 	IP() string
+	// Port returns the Port part of the endpoint.
+	Port() (int, error)
 	// Equal checks if two endpoints are equal.
 	Equal(Endpoint) bool
 }

pkg/proxy/userspace/proxier.go

@@ -36,7 +36,7 @@ import (
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/sets"
-	proxyutil "k8s.io/kubernetes/pkg/proxy/util"
+	utilproxy "k8s.io/kubernetes/pkg/proxy/util"
 	"k8s.io/kubernetes/pkg/util/conntrack"
 	"k8s.io/kubernetes/pkg/util/iptables"
 	utilexec "k8s.io/utils/exec"
@@ -588,7 +588,7 @@ func (proxier *Proxier) openPortal(service proxy.ServicePortName, info *ServiceI
 }
 
 func (proxier *Proxier) openOnePortal(portal portal, protocol api.Protocol, proxyIP net.IP, proxyPort int, name proxy.ServicePortName) error {
-	if local, err := proxyutil.IsLocalIP(portal.ip.String()); err != nil {
+	if local, err := utilproxy.IsLocalIP(portal.ip.String()); err != nil {
 		return fmt.Errorf("can't determine if IP %s is local, assuming not: %v", portal.ip, err)
 	} else if local {
 		err := proxier.claimNodePort(portal.ip, portal.port, protocol, name)
@@ -767,7 +767,7 @@ func (proxier *Proxier) closePortal(service proxy.ServicePortName, info *Service
 func (proxier *Proxier) closeOnePortal(portal portal, protocol api.Protocol, proxyIP net.IP, proxyPort int, name proxy.ServicePortName) []error {
 	el := []error{}
 
-	if local, err := proxyutil.IsLocalIP(portal.ip.String()); err != nil {
+	if local, err := utilproxy.IsLocalIP(portal.ip.String()); err != nil {
 		el = append(el, fmt.Errorf("can't determine if IP %s is local, assuming not: %v", portal.ip, err))
 	} else if local {
 		if err := proxier.releaseNodePort(portal.ip, portal.port, protocol, name); err != nil {
@@ -967,7 +967,7 @@ func iptablesCommonPortalArgs(destIP net.IP, addPhysicalInterfaceMatch bool, add
 	}
 
 	if destIP != nil {
-		args = append(args, "-d", proxyutil.ToCIDR(destIP))
+		args = append(args, "-d", utilproxy.ToCIDR(destIP))
 	}
 
 	if addPhysicalInterfaceMatch {

pkg/proxy/util/BUILD

@@ -13,10 +13,12 @@ go_library(
     deps = [
         "//pkg/apis/core:go_default_library",
         "//pkg/apis/core/helper:go_default_library",
-        "//pkg/util/conntrack:go_default_library",
+        "//pkg/util/net:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
+        "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
+        "//vendor/k8s.io/client-go/tools/record:go_default_library",
     ],
 )
 

pkg/proxy/util/utils.go

@@ -20,11 +20,13 @@ import (
 	"fmt"
 	"net"
 
+	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/client-go/tools/record"
 	api "k8s.io/kubernetes/pkg/apis/core"
 	"k8s.io/kubernetes/pkg/apis/core/helper"
-	"k8s.io/kubernetes/pkg/util/conntrack"
+	utilnet "k8s.io/kubernetes/pkg/util/net"
 
 	"github.com/golang/glog"
 )
@@ -117,10 +119,10 @@ func GetNodeAddresses(cidrs []string, nw NetworkInterfacer) (sets.String, error)
 					return nil, fmt.Errorf("error parsing CIDR for interface %s, error: %v", itf.Name, err)
 				}
 				if ipNet.Contains(ip) {
-					if conntrack.IsIPv6(ip) && !uniqueAddressList.Has(IPv6ZeroCIDR) {
+					if utilnet.IsIPv6(ip) && !uniqueAddressList.Has(IPv6ZeroCIDR) {
 						uniqueAddressList.Insert(ip.String())
 					}
-					if !conntrack.IsIPv6(ip) && !uniqueAddressList.Has(IPv4ZeroCIDR) {
+					if !utilnet.IsIPv6(ip) && !uniqueAddressList.Has(IPv4ZeroCIDR) {
 						uniqueAddressList.Insert(ip.String())
 					}
 				}
@@ -129,3 +131,18 @@ func GetNodeAddresses(cidrs []string, nw NetworkInterfacer) (sets.String, error)
 	}
 	return uniqueAddressList, nil
 }
+
+// LogAndEmitIncorrectIPVersionEvent logs and emits incorrect IP version event.
+func LogAndEmitIncorrectIPVersionEvent(recorder record.EventRecorder, fieldName, fieldValue, svcNamespace, svcName string, svcUID types.UID) {
+	errMsg := fmt.Sprintf("%s in %s has incorrect IP version", fieldValue, fieldName)
+	glog.Errorf("%s (service %s/%s).", errMsg, svcNamespace, svcName)
+	if recorder != nil {
+		recorder.Eventf(
+			&v1.ObjectReference{
+				Kind:      "Service",
+				Name:      svcName,
+				Namespace: svcNamespace,
+				UID:       svcUID,
+			}, v1.EventTypeWarning, "KubeProxyIncorrectIPVersion", errMsg)
+	}
+}

pkg/util/conntrack/BUILD

@@ -8,6 +8,7 @@ go_library(
     importpath = "k8s.io/kubernetes/pkg/util/conntrack",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/util/net:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/utils/exec:go_default_library",
     ],
@@ -20,6 +21,7 @@ go_test(
     ],
     embed = [":go_default_library"],
     deps = [
+        "//pkg/util/net:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/utils/exec:go_default_library",
         "//vendor/k8s.io/utils/exec/testing:go_default_library",

pkg/util/conntrack/conntrack.go

@@ -18,11 +18,11 @@ package conntrack
 
 import (
 	"fmt"
-	"net"
 	"strconv"
 	"strings"
 
 	"k8s.io/api/core/v1"
+	utilnet "k8s.io/kubernetes/pkg/util/net"
 	"k8s.io/utils/exec"
 )
 
@@ -31,17 +31,6 @@ import (
 // NoConnectionToDelete is the error string returned by conntrack when no matching connections are found
 const NoConnectionToDelete = "0 flow entries have been deleted"
 
-// IsIPv6 returns true if the given ip address is a valid ipv6 address
-func IsIPv6(netIP net.IP) bool {
-	return netIP != nil && netIP.To4() == nil
-}
-
-// IsIPv6String returns true if the given string is a valid ipv6 address
-func IsIPv6String(ip string) bool {
-	netIP := net.ParseIP(ip)
-	return IsIPv6(netIP)
-}
-
 func protoStr(proto v1.Protocol) string {
 	return strings.ToLower(string(proto))
 }
@@ -56,7 +45,7 @@ func parametersWithFamily(isIPv6 bool, parameters ...string) []string {
 // ClearEntriesForIP uses the conntrack tool to delete the conntrack entries
 // for the UDP connections specified by the given service IP
 func ClearEntriesForIP(execer exec.Interface, ip string, protocol v1.Protocol) error {
-	parameters := parametersWithFamily(IsIPv6String(ip), "-D", "--orig-dst", ip, "-p", protoStr(protocol))
+	parameters := parametersWithFamily(utilnet.IsIPv6String(ip), "-D", "--orig-dst", ip, "-p", protoStr(protocol))
 	err := Exec(execer, parameters...)
 	if err != nil && !strings.Contains(err.Error(), NoConnectionToDelete) {
 		// TODO: Better handling for deletion failure. When failure occur, stale udp connection may not get flushed.
@@ -107,7 +96,7 @@ func ClearEntriesForPort(execer exec.Interface, port int, isIPv6 bool, protocol
 // ClearEntriesForNAT uses the conntrack tool to delete the conntrack entries
 // for connections specified by the {origin, dest} IP pair.
 func ClearEntriesForNAT(execer exec.Interface, origin, dest string, protocol v1.Protocol) error {
-	parameters := parametersWithFamily(IsIPv6String(origin), "-D", "--orig-dst", origin, "--dst-nat", dest,
+	parameters := parametersWithFamily(utilnet.IsIPv6String(origin), "-D", "--orig-dst", origin, "--dst-nat", dest,
 		"-p", protoStr(protocol))
 	err := Exec(execer, parameters...)
 	if err != nil && !strings.Contains(err.Error(), NoConnectionToDelete) {

pkg/util/conntrack/conntrack_test.go

@@ -18,11 +18,11 @@ package conntrack
 
 import (
 	"fmt"
-	"net"
 	"strings"
 	"testing"
 
 	"k8s.io/api/core/v1"
+	utilnet "k8s.io/kubernetes/pkg/util/net"
 	"k8s.io/utils/exec"
 	fakeexec "k8s.io/utils/exec/testing"
 )
@@ -119,7 +119,7 @@ func TestClearUDPConntrackForIP(t *testing.T) {
 		if err := ClearEntriesForIP(&fexec, tc.ip, v1.ProtocolUDP); err != nil {
 			t.Errorf("%s test case:, Unexpected error: %v", tc.name, err)
 		}
-		expectCommand := fmt.Sprintf("conntrack -D --orig-dst %s -p udp", tc.ip) + familyParamStr(IsIPv6String(tc.ip))
+		expectCommand := fmt.Sprintf("conntrack -D --orig-dst %s -p udp", tc.ip) + familyParamStr(utilnet.IsIPv6String(tc.ip))
 		execCommand := strings.Join(fcmd.CombinedOutputLog[svcCount], " ")
 		if expectCommand != execCommand {
 			t.Errorf("%s test case: Expect command: %s, but executed %s", tc.name, expectCommand, execCommand)
@@ -223,7 +223,7 @@ func TestDeleteUDPConnections(t *testing.T) {
 		if err != nil {
 			t.Errorf("%s test case: unexpected error: %v", tc.name, err)
 		}
-		expectCommand := fmt.Sprintf("conntrack -D --orig-dst %s --dst-nat %s -p udp", tc.origin, tc.dest) + familyParamStr(IsIPv6String(tc.origin))
+		expectCommand := fmt.Sprintf("conntrack -D --orig-dst %s --dst-nat %s -p udp", tc.origin, tc.dest) + familyParamStr(utilnet.IsIPv6String(tc.origin))
 		execCommand := strings.Join(fcmd.CombinedOutputLog[i], " ")
 		if expectCommand != execCommand {
 			t.Errorf("%s test case: Expect command: %s, but executed %s", tc.name, expectCommand, execCommand)
@@ -234,99 +234,3 @@ func TestDeleteUDPConnections(t *testing.T) {
 		t.Errorf("Expect command executed %d times, but got %d", svcCount, fexec.CommandCalls)
 	}
 }
-
-func TestIsIPv6String(t *testing.T) {
-	testCases := []struct {
-		ip         string
-		expectIPv6 bool
-	}{
-		{
-			ip:         "127.0.0.1",
-			expectIPv6: false,
-		},
-		{
-			ip:         "192.168.0.0",
-			expectIPv6: false,
-		},
-		{
-			ip:         "1.2.3.4",
-			expectIPv6: false,
-		},
-		{
-			ip:         "bad ip",
-			expectIPv6: false,
-		},
-		{
-			ip:         "::1",
-			expectIPv6: true,
-		},
-		{
-			ip:         "fd00::600d:f00d",
-			expectIPv6: true,
-		},
-		{
-			ip:         "2001:db8::5",
-			expectIPv6: true,
-		},
-	}
-	for i := range testCases {
-		isIPv6 := IsIPv6String(testCases[i].ip)
-		if isIPv6 != testCases[i].expectIPv6 {
-			t.Errorf("[%d] Expect ipv6 %v, got %v", i+1, testCases[i].expectIPv6, isIPv6)
-		}
-	}
-}
-
-func TestIsIPv6(t *testing.T) {
-	testCases := []struct {
-		ip         net.IP
-		expectIPv6 bool
-	}{
-		{
-			ip:         net.IPv4zero,
-			expectIPv6: false,
-		},
-		{
-			ip:         net.IPv4bcast,
-			expectIPv6: false,
-		},
-		{
-			ip:         net.ParseIP("127.0.0.1"),
-			expectIPv6: false,
-		},
-		{
-			ip:         net.ParseIP("10.20.40.40"),
-			expectIPv6: false,
-		},
-		{
-			ip:         net.ParseIP("172.17.3.0"),
-			expectIPv6: false,
-		},
-		{
-			ip:         nil,
-			expectIPv6: false,
-		},
-		{
-			ip:         net.IPv6loopback,
-			expectIPv6: true,
-		},
-		{
-			ip:         net.IPv6zero,
-			expectIPv6: true,
-		},
-		{
-			ip:         net.ParseIP("fd00::600d:f00d"),
-			expectIPv6: true,
-		},
-		{
-			ip:         net.ParseIP("2001:db8::5"),
-			expectIPv6: true,
-		},
-	}
-	for i := range testCases {
-		isIPv6 := IsIPv6(testCases[i].ip)
-		if isIPv6 != testCases[i].expectIPv6 {
-			t.Errorf("[%d] Expect ipv6 %v, got %v", i+1, testCases[i].expectIPv6, isIPv6)
-		}
-	}
-}

pkg/util/net/BUILD

+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
 package(default_visibility = ["//visibility:public"])
 
 filegroup(
@@ -15,3 +17,15 @@ filegroup(
     ],
     tags = ["automanaged"],
 )
+
+go_library(
+    name = "go_default_library",
+    srcs = ["net.go"],
+    importpath = "k8s.io/kubernetes/pkg/util/net",
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["net_test.go"],
+    embed = [":go_default_library"],
+)

pkg/util/net/OWNERS

+reviewers:
+  - sig-network-reviewers
+approvers:
+  - sig-network-approvers

pkg/util/net/net.go

+/*
+Copyright 2018 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 net
+
+import (
+	"net"
+)
+
+// IsIPv6 returns if netIP is IPv6.
+func IsIPv6(netIP net.IP) bool {
+	return netIP != nil && netIP.To4() == nil
+}
+
+// IsIPv6String returns if ip is IPv6.
+func IsIPv6String(ip string) bool {
+	netIP := net.ParseIP(ip)
+	return IsIPv6(netIP)
+}
+
+// IsIPv6CIDR returns if cidr is IPv6.
+// This assumes cidr is a valid CIDR.
+func IsIPv6CIDR(cidr string) bool {
+	ip, _, _ := net.ParseCIDR(cidr)
+	return IsIPv6(ip)
+}
+
+// FilterIncorrectIPVersion filters out the incorrect IP version case from a slice of IP strings.
+func FilterIncorrectIPVersion(ipStrings []string, isIPv6Mode bool) ([]string, []string) {
+	return filterWithCondition(ipStrings, isIPv6Mode, IsIPv6String)
+}
+
+// FilterIncorrectCIDRVersion filters out the incorrect IP version case from a slice of CIDR strings.
+func FilterIncorrectCIDRVersion(ipStrings []string, isIPv6Mode bool) ([]string, []string) {
+	return filterWithCondition(ipStrings, isIPv6Mode, IsIPv6CIDR)
+}
+
+func filterWithCondition(strs []string, expectedCondition bool, conditionFunc func(string) bool) ([]string, []string) {
+	var corrects, incorrects []string
+	for _, str := range strs {
+		if conditionFunc(str) != expectedCondition {
+			incorrects = append(incorrects, str)
+		} else {
+			corrects = append(corrects, str)
+		}
+	}
+	return corrects, incorrects
+}

pkg/util/net/net_test.go

+/*
+Copyright 2018 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 net
+
+import (
+	"net"
+	"reflect"
+	"testing"
+)
+
+func TestIsIPv6String(t *testing.T) {
+	testCases := []struct {
+		ip         string
+		expectIPv6 bool
+	}{
+		{
+			ip:         "127.0.0.1",
+			expectIPv6: false,
+		},
+		{
+			ip:         "192.168.0.0",
+			expectIPv6: false,
+		},
+		{
+			ip:         "1.2.3.4",
+			expectIPv6: false,
+		},
+		{
+			ip:         "bad ip",
+			expectIPv6: false,
+		},
+		{
+			ip:         "::1",
+			expectIPv6: true,
+		},
+		{
+			ip:         "fd00::600d:f00d",
+			expectIPv6: true,
+		},
+		{
+			ip:         "2001:db8::5",
+			expectIPv6: true,
+		},
+	}
+	for i := range testCases {
+		isIPv6 := IsIPv6String(testCases[i].ip)
+		if isIPv6 != testCases[i].expectIPv6 {
+			t.Errorf("[%d] Expect ipv6 %v, got %v", i+1, testCases[i].expectIPv6, isIPv6)
+		}
+	}
+}
+
+func TestIsIPv6(t *testing.T) {
+	testCases := []struct {
+		ip         net.IP
+		expectIPv6 bool
+	}{
+		{
+			ip:         net.IPv4zero,
+			expectIPv6: false,
+		},
+		{
+			ip:         net.IPv4bcast,
+			expectIPv6: false,
+		},
+		{
+			ip:         net.ParseIP("127.0.0.1"),
+			expectIPv6: false,
+		},
+		{
+			ip:         net.ParseIP("10.20.40.40"),
+			expectIPv6: false,
+		},
+		{
+			ip:         net.ParseIP("172.17.3.0"),
+			expectIPv6: false,
+		},
+		{
+			ip:         nil,
+			expectIPv6: false,
+		},
+		{
+			ip:         net.IPv6loopback,
+			expectIPv6: true,
+		},
+		{
+			ip:         net.IPv6zero,
+			expectIPv6: true,
+		},
+		{
+			ip:         net.ParseIP("fd00::600d:f00d"),
+			expectIPv6: true,
+		},
+		{
+			ip:         net.ParseIP("2001:db8::5"),
+			expectIPv6: true,
+		},
+	}
+	for i := range testCases {
+		isIPv6 := IsIPv6(testCases[i].ip)
+		if isIPv6 != testCases[i].expectIPv6 {
+			t.Errorf("[%d] Expect ipv6 %v, got %v", i+1, testCases[i].expectIPv6, isIPv6)
+		}
+	}
+}
+
+func TestIsIPv6CIDR(t *testing.T) {
+	testCases := []struct {
+		desc         string
+		cidr         string
+		expectResult bool
+	}{
+		{
+			desc:         "ipv4 CIDR 1",
+			cidr:         "10.0.0.0/8",
+			expectResult: false,
+		},
+		{
+			desc:         "ipv4 CIDR 2",
+			cidr:         "192.168.0.0/16",
+			expectResult: false,
+		},
+		{
+			desc:         "ipv6 CIDR 1",
+			cidr:         "::/1",
+			expectResult: true,
+		},
+		{
+			desc:         "ipv6 CIDR 2",
+			cidr:         "2000::/10",
+			expectResult: true,
+		},
+		{
+			desc:         "ipv6 CIDR 3",
+			cidr:         "2001:db8::/32",
+			expectResult: true,
+		},
+	}
+
+	for _, tc := range testCases {
+		res := IsIPv6CIDR(tc.cidr)
+		if res != tc.expectResult {
+			t.Errorf("%v: want IsIPv6CIDR=%v, got %v", tc.desc, tc.expectResult, res)
+		}
+	}
+}
+
+func TestFilterIncorrectIPVersion(t *testing.T) {
+	testCases := []struct {
+		desc             string
+		isIPv6           bool
+		ipStrings        []string
+		expectCorrects   []string
+		expectIncorrects []string
+	}{
+		{
+			desc:             "all ipv4 strings in ipv4 mode",
+			isIPv6:           false,
+			ipStrings:        []string{"10.0.0.1", "192.168.0.1", "127.0.0.1"},
+			expectCorrects:   []string{"10.0.0.1", "192.168.0.1", "127.0.0.1"},
+			expectIncorrects: nil,
+		},
+		{
+			desc:             "all ipv6 strings in ipv4 mode",
+			isIPv6:           false,
+			ipStrings:        []string{"::1", "fd00::600d:f00d", "2001:db8::5"},
+			expectCorrects:   nil,
+			expectIncorrects: []string{"::1", "fd00::600d:f00d", "2001:db8::5"},
+		},
+		{
+			desc:             "mixed versions in ipv4 mode",
+			isIPv6:           false,
+			ipStrings:        []string{"10.0.0.1", "192.168.0.1", "127.0.0.1", "::1", "fd00::600d:f00d", "2001:db8::5"},
+			expectCorrects:   []string{"10.0.0.1", "192.168.0.1", "127.0.0.1"},
+			expectIncorrects: []string{"::1", "fd00::600d:f00d", "2001:db8::5"},
+		},
+		{
+			desc:             "all ipv4 strings in ipv6 mode",
+			isIPv6:           true,
+			ipStrings:        []string{"10.0.0.1", "192.168.0.1", "127.0.0.1"},
+			expectCorrects:   nil,
+			expectIncorrects: []string{"10.0.0.1", "192.168.0.1", "127.0.0.1"},
+		},
+		{
+			desc:             "all ipv6 strings in ipv6 mode",
+			isIPv6:           true,
+			ipStrings:        []string{"::1", "fd00::600d:f00d", "2001:db8::5"},
+			expectCorrects:   []string{"::1", "fd00::600d:f00d", "2001:db8::5"},
+			expectIncorrects: nil,
+		},
+		{
+			desc:             "mixed versions in ipv6 mode",
+			isIPv6:           true,
+			ipStrings:        []string{"10.0.0.1", "192.168.0.1", "127.0.0.1", "::1", "fd00::600d:f00d", "2001:db8::5"},
+			expectCorrects:   []string{"::1", "fd00::600d:f00d", "2001:db8::5"},
+			expectIncorrects: []string{"10.0.0.1", "192.168.0.1", "127.0.0.1"},
+		},
+	}
+
+	for _, tc := range testCases {
+		corrects, incorrects := FilterIncorrectIPVersion(tc.ipStrings, tc.isIPv6)
+		if !reflect.DeepEqual(tc.expectCorrects, corrects) {
+			t.Errorf("%v: want corrects=%v, got %v", tc.desc, tc.expectCorrects, corrects)
+		}
+		if !reflect.DeepEqual(tc.expectIncorrects, incorrects) {
+			t.Errorf("%v: want incorrects=%v, got %v", tc.desc, tc.expectIncorrects, incorrects)
+		}
+	}
+}
+
+func TestFilterIncorrectCIDRVersion(t *testing.T) {
+	testCases := []struct {
+		desc             string
+		isIPv6           bool
+		cidrStrings      []string
+		expectCorrects   []string
+		expectIncorrects []string
+	}{
+		{
+			desc:             "all ipv4 strings in ipv4 mode",
+			isIPv6:           false,
+			cidrStrings:      []string{"0.0.0.0/1", "1.0.0.0/1"},
+			expectCorrects:   []string{"0.0.0.0/1", "1.0.0.0/1"},
+			expectIncorrects: nil,
+		},
+		{
+			desc:             "all ipv6 strings in ipv4 mode",
+			isIPv6:           false,
+			cidrStrings:      []string{"2001:db8::/32", "2001:0db8:0123:4567::/64"},
+			expectCorrects:   nil,
+			expectIncorrects: []string{"2001:db8::/32", "2001:0db8:0123:4567::/64"},
+		},
+		{
+			desc:             "mixed versions in ipv4 mode",
+			isIPv6:           false,
+			cidrStrings:      []string{"0.0.0.0/1", "1.0.0.0/1", "2001:db8::/32", "2001:0db8:0123:4567::/64"},
+			expectCorrects:   []string{"0.0.0.0/1", "1.0.0.0/1"},
+			expectIncorrects: []string{"2001:db8::/32", "2001:0db8:0123:4567::/64"},
+		},
+		{
+			desc:             "all ipv4 strings in ipv6 mode",
+			isIPv6:           true,
+			cidrStrings:      []string{"0.0.0.0/1", "1.0.0.0/1"},
+			expectCorrects:   nil,
+			expectIncorrects: []string{"0.0.0.0/1", "1.0.0.0/1"},
+		},
+		{
+			desc:             "all ipv6 strings in ipv6 mode",
+			isIPv6:           true,
+			cidrStrings:      []string{"2001:db8::/32", "2001:0db8:0123:4567::/64"},
+			expectCorrects:   []string{"2001:db8::/32", "2001:0db8:0123:4567::/64"},
+			expectIncorrects: nil,
+		},
+		{
+			desc:             "mixed versions in ipv6 mode",
+			isIPv6:           true,
+			cidrStrings:      []string{"0.0.0.0/1", "1.0.0.0/1", "2001:db8::/32", "2001:0db8:0123:4567::/64"},
+			expectCorrects:   []string{"2001:db8::/32", "2001:0db8:0123:4567::/64"},
+			expectIncorrects: []string{"0.0.0.0/1", "1.0.0.0/1"},
+		},
+	}
+
+	for _, tc := range testCases {
+		corrects, incorrects := FilterIncorrectCIDRVersion(tc.cidrStrings, tc.isIPv6)
+		if !reflect.DeepEqual(tc.expectCorrects, corrects) {
+			t.Errorf("%v: want corrects=%v, got %v", tc.desc, tc.expectCorrects, corrects)
+		}
+		if !reflect.DeepEqual(tc.expectIncorrects, incorrects) {
+			t.Errorf("%v: want incorrects=%v, got %v", tc.desc, tc.expectIncorrects, incorrects)
+		}
+	}
+}