github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-29 06:27:05 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #113277 from manav014/master

Browse Source 
Cleanup: kube-proxy internal naming
This commit is contained in:
Kubernetes Prow Robot 2022-11-02 18:21:05 -07:00 committed by GitHub
commit 86a6ace994
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 274 additions and 274 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
pkg/proxy/iptables/proxier.go
View File

@ -92,7 +92,7 @@ const sysctlBridgeCallIPTables = "net/bridge/bridge-nf-call-iptables"
// internal struct for string service information // internal struct for string service information
type servicePortInfo struct { type servicePortInfo struct {
*proxy.BaseServiceInfo *proxy.BaseServicePortInfo
// The following fields are computed and stored for performance reasons. // The following fields are computed and stored for performance reasons.
nameString string nameString string
clusterPolicyChainName utiliptables.Chain clusterPolicyChainName utiliptables.Chain
@ -102,8 +102,8 @@ type servicePortInfo struct {
} }
// returns a new proxy.ServicePort which abstracts a serviceInfo // returns a new proxy.ServicePort which abstracts a serviceInfo
func newServiceInfo(port *v1.ServicePort, service *v1.Service, baseInfo *proxy.BaseServiceInfo) proxy.ServicePort { func newServiceInfo(port *v1.ServicePort, service *v1.Service, bsvcPortInfo *proxy.BaseServicePortInfo) proxy.ServicePort {
svcPort := &servicePortInfo{BaseServiceInfo: baseInfo} svcPort := &servicePortInfo{BaseServicePortInfo: bsvcPortInfo}
// Store the following for performance reasons. // Store the following for performance reasons.
svcName := types.NamespacedName{Namespace: service.Namespace, Name: service.Name} svcName := types.NamespacedName{Namespace: service.Namespace, Name: service.Name}
@ -157,7 +157,7 @@ type Proxier struct {
serviceChanges *proxy.ServiceChangeTracker serviceChanges *proxy.ServiceChangeTracker
mu sync.Mutex // protects the following fields mu sync.Mutex // protects the following fields
serviceMap proxy.ServiceMap svcPortMap proxy.ServicePortMap
endpointsMap proxy.EndpointsMap endpointsMap proxy.EndpointsMap
nodeLabels map[string]string nodeLabels map[string]string
// endpointSlicesSynced, and servicesSynced are set to true // endpointSlicesSynced, and servicesSynced are set to true
@ -267,7 +267,7 @@ func NewProxier(ipt utiliptables.Interface,
} }
proxier := &Proxier{ proxier := &Proxier{
serviceMap: make(proxy.ServiceMap), svcPortMap: make(proxy.ServicePortMap),
serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, ipFamily, recorder, nil), serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, ipFamily, recorder, nil),
endpointsMap: make(proxy.EndpointsMap), endpointsMap: make(proxy.EndpointsMap),
endpointsChanges: proxy.NewEndpointChangeTracker(hostname, newEndpointInfo, ipFamily, recorder, nil), endpointsChanges: proxy.NewEndpointChangeTracker(hostname, newEndpointInfo, ipFamily, recorder, nil),
@ -733,7 +733,7 @@ func isServiceChainName(chainString string) bool {
// TODO: move it to util // TODO: move it to util
func (proxier *Proxier) deleteEndpointConnections(connectionMap []proxy.ServiceEndpoint) { func (proxier *Proxier) deleteEndpointConnections(connectionMap []proxy.ServiceEndpoint) {
for _, epSvcPair := range connectionMap { for _, epSvcPair := range connectionMap {
if svcInfo, ok := proxier.serviceMap[epSvcPair.ServicePortName]; ok && conntrack.IsClearConntrackNeeded(svcInfo.Protocol()) { if svcInfo, ok := proxier.svcPortMap[epSvcPair.ServicePortName]; ok && conntrack.IsClearConntrackNeeded(svcInfo.Protocol()) {
endpointIP := utilproxy.IPPart(epSvcPair.Endpoint) endpointIP := utilproxy.IPPart(epSvcPair.Endpoint)
nodePort := svcInfo.NodePort() nodePort := svcInfo.NodePort()
svcProto := svcInfo.Protocol() svcProto := svcInfo.Protocol()
@ -812,7 +812,7 @@ func (proxier *Proxier) syncProxyRules() {
serviceChanged = proxier.serviceChanges.PendingChanges() serviceChanged = proxier.serviceChanges.PendingChanges()
endpointsChanged = proxier.endpointsChanges.PendingChanges() endpointsChanged = proxier.endpointsChanges.PendingChanges()
} }
serviceUpdateResult := proxier.serviceMap.Update(proxier.serviceChanges) serviceUpdateResult := proxier.svcPortMap.Update(proxier.serviceChanges)
endpointUpdateResult := proxier.endpointsMap.Update(proxier.endpointsChanges) endpointUpdateResult := proxier.endpointsMap.Update(proxier.endpointsChanges)
// We need to detect stale connections to UDP Services so we // We need to detect stale connections to UDP Services so we
@ -822,7 +822,7 @@ func (proxier *Proxier) syncProxyRules() {
// merge stale services gathered from updateEndpointsMap // merge stale services gathered from updateEndpointsMap
// an UDP service that changes from 0 to non-0 endpoints is considered stale. // an UDP service that changes from 0 to non-0 endpoints is considered stale.
for _, svcPortName := range endpointUpdateResult.StaleServiceNames { for _, svcPortName := range endpointUpdateResult.StaleServiceNames {
if svcInfo, ok := proxier.serviceMap[svcPortName]; ok && svcInfo != nil && conntrack.IsClearConntrackNeeded(svcInfo.Protocol()) { if svcInfo, ok := proxier.svcPortMap[svcPortName]; ok && svcInfo != nil && conntrack.IsClearConntrackNeeded(svcInfo.Protocol()) {
klog.V(2).InfoS("Stale service", "protocol", strings.ToLower(string(svcInfo.Protocol())), "servicePortName", svcPortName, "clusterIP", svcInfo.ClusterIP()) klog.V(2).InfoS("Stale service", "protocol", strings.ToLower(string(svcInfo.Protocol())), "servicePortName", svcPortName, "clusterIP", svcInfo.ClusterIP())
conntrackCleanupServiceIPs.Insert(svcInfo.ClusterIP().String()) conntrackCleanupServiceIPs.Insert(svcInfo.ClusterIP().String())
for _, extIP := range svcInfo.ExternalIPStrings() { for _, extIP := range svcInfo.ExternalIPStrings() {
@ -935,7 +935,7 @@ func (proxier *Proxier) syncProxyRules() {
// Compute total number of endpoint chains across all services to get // Compute total number of endpoint chains across all services to get
// a sense of how big the cluster is. // a sense of how big the cluster is.
totalEndpoints := 0 totalEndpoints := 0
for svcName := range proxier.serviceMap { for svcName := range proxier.svcPortMap {
totalEndpoints += len(proxier.endpointsMap[svcName]) totalEndpoints += len(proxier.endpointsMap[svcName])
} }
proxier.largeClusterMode = (totalEndpoints > largeClusterEndpointsThreshold) proxier.largeClusterMode = (totalEndpoints > largeClusterEndpointsThreshold)
@ -961,7 +961,7 @@ func (proxier *Proxier) syncProxyRules() {
serviceNoLocalEndpointsTotalExternal := 0 serviceNoLocalEndpointsTotalExternal := 0
// Build rules for each service-port. // Build rules for each service-port.
for svcName, svc := range proxier.serviceMap { for svcName, svc := range proxier.svcPortMap {
svcInfo, ok := svc.(*servicePortInfo) svcInfo, ok := svc.(*servicePortInfo)
if !ok { if !ok {
klog.ErrorS(nil, "Failed to cast serviceInfo", "serviceName", svcName) klog.ErrorS(nil, "Failed to cast serviceInfo", "serviceName", svcName)
@ -1491,7 +1491,7 @@ func (proxier *Proxier) syncProxyRules() {
proxier.iptablesData.WriteString("COMMIT\n") proxier.iptablesData.WriteString("COMMIT\n")
klog.V(2).InfoS("Reloading service iptables data", klog.V(2).InfoS("Reloading service iptables data",
"numServices", len(proxier.serviceMap), "numServices", len(proxier.svcPortMap),
"numEndpoints", totalEndpoints, "numEndpoints", totalEndpoints,
"numFilterChains", proxier.filterChains.Lines(), "numFilterChains", proxier.filterChains.Lines(),
"numFilterRules", proxier.filterRules.Lines(), "numFilterRules", proxier.filterRules.Lines(),

54
pkg/proxy/iptables/proxier_test.go
View File

@ -162,7 +162,7 @@ func TestDeleteEndpointConnectionsIPv4(t *testing.T) {
}), }),
) )
fp.serviceMap.Update(fp.serviceChanges) fp.svcPortMap.Update(fp.serviceChanges)
} }
// Run the test cases // Run the test cases
@ -305,7 +305,7 @@ func TestDeleteEndpointConnectionsIPv6(t *testing.T) {
}), }),
) )
fp.serviceMap.Update(fp.serviceChanges) fp.svcPortMap.Update(fp.serviceChanges)
} }
// Run the test cases // Run the test cases
@ -404,7 +404,7 @@ func NewFakeProxier(ipt utiliptables.Interface) *Proxier {
p := &Proxier{ p := &Proxier{
exec: &fakeexec.FakeExec{}, exec: &fakeexec.FakeExec{},
serviceMap: make(proxy.ServiceMap), svcPortMap: make(proxy.ServicePortMap),
serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, ipfamily, nil, nil), serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, ipfamily, nil, nil),
endpointsMap: make(proxy.EndpointsMap), endpointsMap: make(proxy.EndpointsMap),
endpointsChanges: proxy.NewEndpointChangeTracker(testHostname, newEndpointInfo, ipfamily, nil, nil), endpointsChanges: proxy.NewEndpointChangeTracker(testHostname, newEndpointInfo, ipfamily, nil, nil),
@ -3562,9 +3562,9 @@ func TestBuildServiceMapAddRemove(t *testing.T) {
for i := range services { for i := range services {
fp.OnServiceAdd(services[i]) fp.OnServiceAdd(services[i])
} }
result := fp.serviceMap.Update(fp.serviceChanges) result := fp.svcPortMap.Update(fp.serviceChanges)
if len(fp.serviceMap) != 10 { if len(fp.svcPortMap) != 10 {
t.Errorf("expected service map length 10, got %v", fp.serviceMap) t.Errorf("expected service map length 10, got %v", fp.svcPortMap)
} }
// The only-local-loadbalancer ones get added // The only-local-loadbalancer ones get added
@ -3595,9 +3595,9 @@ func TestBuildServiceMapAddRemove(t *testing.T) {
fp.OnServiceDelete(services[2]) fp.OnServiceDelete(services[2])
fp.OnServiceDelete(services[3]) fp.OnServiceDelete(services[3])
result = fp.serviceMap.Update(fp.serviceChanges) result = fp.svcPortMap.Update(fp.serviceChanges)
if len(fp.serviceMap) != 1 { if len(fp.svcPortMap) != 1 {
t.Errorf("expected service map length 1, got %v", fp.serviceMap) t.Errorf("expected service map length 1, got %v", fp.svcPortMap)
} }
if len(result.HCServiceNodePorts) != 0 { if len(result.HCServiceNodePorts) != 0 {
@ -3635,9 +3635,9 @@ func TestBuildServiceMapServiceHeadless(t *testing.T) {
) )
// Headless service should be ignored // Headless service should be ignored
result := fp.serviceMap.Update(fp.serviceChanges) result := fp.svcPortMap.Update(fp.serviceChanges)
if len(fp.serviceMap) != 0 { if len(fp.svcPortMap) != 0 {
t.Errorf("expected service map length 0, got %d", len(fp.serviceMap)) t.Errorf("expected service map length 0, got %d", len(fp.svcPortMap))
} }
// No proxied services, so no healthchecks // No proxied services, so no healthchecks
@ -3663,9 +3663,9 @@ func TestBuildServiceMapServiceTypeExternalName(t *testing.T) {
}), }),
) )
result := fp.serviceMap.Update(fp.serviceChanges) result := fp.svcPortMap.Update(fp.serviceChanges)
if len(fp.serviceMap) != 0 { if len(fp.svcPortMap) != 0 {
t.Errorf("expected service map length 0, got %v", fp.serviceMap) t.Errorf("expected service map length 0, got %v", fp.svcPortMap)
} }
// No proxied services, so no healthchecks // No proxied services, so no healthchecks
if len(result.HCServiceNodePorts) != 0 { if len(result.HCServiceNodePorts) != 0 {
@ -3703,9 +3703,9 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
fp.OnServiceAdd(servicev1) fp.OnServiceAdd(servicev1)
result := fp.serviceMap.Update(fp.serviceChanges) result := fp.svcPortMap.Update(fp.serviceChanges)
if len(fp.serviceMap) != 2 { if len(fp.svcPortMap) != 2 {
t.Errorf("expected service map length 2, got %v", fp.serviceMap) t.Errorf("expected service map length 2, got %v", fp.svcPortMap)
} }
if len(result.HCServiceNodePorts) != 0 { if len(result.HCServiceNodePorts) != 0 {
t.Errorf("expected healthcheck ports length 0, got %v", result.HCServiceNodePorts) t.Errorf("expected healthcheck ports length 0, got %v", result.HCServiceNodePorts)
@ -3717,9 +3717,9 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
// Change service to load-balancer // Change service to load-balancer
fp.OnServiceUpdate(servicev1, servicev2) fp.OnServiceUpdate(servicev1, servicev2)
result = fp.serviceMap.Update(fp.serviceChanges) result = fp.svcPortMap.Update(fp.serviceChanges)
if len(fp.serviceMap) != 2 { if len(fp.svcPortMap) != 2 {
t.Errorf("expected service map length 2, got %v", fp.serviceMap) t.Errorf("expected service map length 2, got %v", fp.svcPortMap)
} }
if len(result.HCServiceNodePorts) != 1 { if len(result.HCServiceNodePorts) != 1 {
t.Errorf("expected healthcheck ports length 1, got %v", result.HCServiceNodePorts) t.Errorf("expected healthcheck ports length 1, got %v", result.HCServiceNodePorts)
@ -3731,9 +3731,9 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
// No change; make sure the service map stays the same and there are // No change; make sure the service map stays the same and there are
// no health-check changes // no health-check changes
fp.OnServiceUpdate(servicev2, servicev2) fp.OnServiceUpdate(servicev2, servicev2)
result = fp.serviceMap.Update(fp.serviceChanges) result = fp.svcPortMap.Update(fp.serviceChanges)
if len(fp.serviceMap) != 2 { if len(fp.svcPortMap) != 2 {
t.Errorf("expected service map length 2, got %v", fp.serviceMap) t.Errorf("expected service map length 2, got %v", fp.svcPortMap)
} }
if len(result.HCServiceNodePorts) != 1 { if len(result.HCServiceNodePorts) != 1 {
t.Errorf("expected healthcheck ports length 1, got %v", result.HCServiceNodePorts) t.Errorf("expected healthcheck ports length 1, got %v", result.HCServiceNodePorts)
@ -3744,9 +3744,9 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
// And back to ClusterIP // And back to ClusterIP
fp.OnServiceUpdate(servicev2, servicev1) fp.OnServiceUpdate(servicev2, servicev1)
result = fp.serviceMap.Update(fp.serviceChanges) result = fp.svcPortMap.Update(fp.serviceChanges)
if len(fp.serviceMap) != 2 { if len(fp.svcPortMap) != 2 {
t.Errorf("expected service map length 2, got %v", fp.serviceMap) t.Errorf("expected service map length 2, got %v", fp.svcPortMap)
} }
if len(result.HCServiceNodePorts) != 0 { if len(result.HCServiceNodePorts) != 0 {
t.Errorf("expected healthcheck ports length 0, got %v", result.HCServiceNodePorts) t.Errorf("expected healthcheck ports length 0, got %v", result.HCServiceNodePorts)

22
pkg/proxy/ipvs/proxier.go
View File

@ -228,7 +228,7 @@ type Proxier struct {
serviceChanges *proxy.ServiceChangeTracker serviceChanges *proxy.ServiceChangeTracker
mu sync.Mutex // protects the following fields mu sync.Mutex // protects the following fields
serviceMap proxy.ServiceMap svcPortMap proxy.ServicePortMap
endpointsMap proxy.EndpointsMap endpointsMap proxy.EndpointsMap
nodeLabels map[string]string nodeLabels map[string]string
// initialSync is a bool indicating if the proxier is syncing for the first time. // initialSync is a bool indicating if the proxier is syncing for the first time.
@ -479,7 +479,7 @@ func NewProxier(ipt utiliptables.Interface,
proxier := &Proxier{ proxier := &Proxier{
ipFamily: ipFamily, ipFamily: ipFamily,
serviceMap: make(proxy.ServiceMap), svcPortMap: make(proxy.ServicePortMap),
serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, ipFamily, recorder, nil), serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, ipFamily, recorder, nil),
endpointsMap: make(proxy.EndpointsMap), endpointsMap: make(proxy.EndpointsMap),
endpointsChanges: proxy.NewEndpointChangeTracker(hostname, nil, ipFamily, recorder, nil), endpointsChanges: proxy.NewEndpointChangeTracker(hostname, nil, ipFamily, recorder, nil),
@ -590,14 +590,14 @@ func filterCIDRs(wantIPv6 bool, cidrs []string) []string {
// internal struct for string service information // internal struct for string service information
type servicePortInfo struct { type servicePortInfo struct {
*proxy.BaseServiceInfo *proxy.BaseServicePortInfo
// The following fields are computed and stored for performance reasons. // The following fields are computed and stored for performance reasons.
nameString string nameString string
} }
// returns a new proxy.ServicePort which abstracts a serviceInfo // returns a new proxy.ServicePort which abstracts a serviceInfo
func newServiceInfo(port *v1.ServicePort, service *v1.Service, baseInfo *proxy.BaseServiceInfo) proxy.ServicePort { func newServiceInfo(port *v1.ServicePort, service *v1.Service, bsvcPortInfo *proxy.BaseServicePortInfo) proxy.ServicePort {
svcPort := &servicePortInfo{BaseServiceInfo: baseInfo} svcPort := &servicePortInfo{BaseServicePortInfo: bsvcPortInfo}
// Store the following for performance reasons. // Store the following for performance reasons.
svcName := types.NamespacedName{Namespace: service.Namespace, Name: service.Name} svcName := types.NamespacedName{Namespace: service.Namespace, Name: service.Name}
@ -1042,13 +1042,13 @@ func (proxier *Proxier) syncProxyRules() {
// We assume that if this was called, we really want to sync them, // We assume that if this was called, we really want to sync them,
// even if nothing changed in the meantime. In other words, callers are // even if nothing changed in the meantime. In other words, callers are
// responsible for detecting no-op changes and not calling this function. // responsible for detecting no-op changes and not calling this function.
serviceUpdateResult := proxier.serviceMap.Update(proxier.serviceChanges) serviceUpdateResult := proxier.svcPortMap.Update(proxier.serviceChanges)
endpointUpdateResult := proxier.endpointsMap.Update(proxier.endpointsChanges) endpointUpdateResult := proxier.endpointsMap.Update(proxier.endpointsChanges)
staleServices := serviceUpdateResult.UDPStaleClusterIP staleServices := serviceUpdateResult.UDPStaleClusterIP
// merge stale services gathered from updateEndpointsMap // merge stale services gathered from updateEndpointsMap
for _, svcPortName := range endpointUpdateResult.StaleServiceNames { for _, svcPortName := range endpointUpdateResult.StaleServiceNames {
if svcInfo, ok := proxier.serviceMap[svcPortName]; ok && svcInfo != nil && conntrack.IsClearConntrackNeeded(svcInfo.Protocol()) { if svcInfo, ok := proxier.svcPortMap[svcPortName]; ok && svcInfo != nil && conntrack.IsClearConntrackNeeded(svcInfo.Protocol()) {
klog.V(2).InfoS("Stale service", "protocol", strings.ToLower(string(svcInfo.Protocol())), "servicePortName", svcPortName, "clusterIP", svcInfo.ClusterIP()) klog.V(2).InfoS("Stale service", "protocol", strings.ToLower(string(svcInfo.Protocol())), "servicePortName", svcPortName, "clusterIP", svcInfo.ClusterIP())
staleServices.Insert(svcInfo.ClusterIP().String()) staleServices.Insert(svcInfo.ClusterIP().String())
for _, extIP := range svcInfo.ExternalIPStrings() { for _, extIP := range svcInfo.ExternalIPStrings() {
@ -1107,7 +1107,7 @@ func (proxier *Proxier) syncProxyRules() {
} }
hasNodePort := false hasNodePort := false
for _, svc := range proxier.serviceMap { for _, svc := range proxier.svcPortMap {
svcInfo, ok := svc.(*servicePortInfo) svcInfo, ok := svc.(*servicePortInfo)
if ok && svcInfo.NodePort() != 0 { if ok && svcInfo.NodePort() != 0 {
hasNodePort = true hasNodePort = true
@ -1159,7 +1159,7 @@ func (proxier *Proxier) syncProxyRules() {
nodeIPs = nodeIPs[:idx] nodeIPs = nodeIPs[:idx]
// Build IPVS rules for each service. // Build IPVS rules for each service.
for svcPortName, svcPort := range proxier.serviceMap { for svcPortName, svcPort := range proxier.svcPortMap {
svcInfo, ok := svcPort.(*servicePortInfo) svcInfo, ok := svcPort.(*servicePortInfo)
if !ok { if !ok {
klog.ErrorS(nil, "Failed to cast serviceInfo", "servicePortName", svcPortName) klog.ErrorS(nil, "Failed to cast serviceInfo", "servicePortName", svcPortName)
@ -1913,7 +1913,7 @@ func (proxier *Proxier) createAndLinkKubeChain() {
// This assumes the proxier mutex is held // This assumes the proxier mutex is held
func (proxier *Proxier) deleteEndpointConnections(connectionMap []proxy.ServiceEndpoint) { func (proxier *Proxier) deleteEndpointConnections(connectionMap []proxy.ServiceEndpoint) {
for _, epSvcPair := range connectionMap { for _, epSvcPair := range connectionMap {
if svcInfo, ok := proxier.serviceMap[epSvcPair.ServicePortName]; ok && conntrack.IsClearConntrackNeeded(svcInfo.Protocol()) { if svcInfo, ok := proxier.svcPortMap[epSvcPair.ServicePortName]; ok && conntrack.IsClearConntrackNeeded(svcInfo.Protocol()) {
endpointIP := utilproxy.IPPart(epSvcPair.Endpoint) endpointIP := utilproxy.IPPart(epSvcPair.Endpoint)
svcProto := svcInfo.Protocol() svcProto := svcInfo.Protocol()
err := conntrack.ClearEntriesForNAT(proxier.exec, svcInfo.ClusterIP().String(), endpointIP, svcProto) err := conntrack.ClearEntriesForNAT(proxier.exec, svcInfo.ClusterIP().String(), endpointIP, svcProto)
@ -2003,7 +2003,7 @@ func (proxier *Proxier) syncEndpoint(svcPortName proxy.ServicePortName, onlyNode
// filter endpoints if appropriate feature gates are enabled and the // filter endpoints if appropriate feature gates are enabled and the
// Service does not have conflicting configuration such as // Service does not have conflicting configuration such as
// externalTrafficPolicy=Local. // externalTrafficPolicy=Local.
svcInfo, ok := proxier.serviceMap[svcPortName] svcInfo, ok := proxier.svcPortMap[svcPortName]
if !ok { if !ok {
klog.InfoS("Unable to filter endpoints due to missing service info", "servicePortName", svcPortName) klog.InfoS("Unable to filter endpoints due to missing service info", "servicePortName", svcPortName)
} else { } else {

50
pkg/proxy/ipvs/proxier_test.go
View File

@ -135,7 +135,7 @@ func NewFakeProxier(ipt utiliptables.Interface, ipvs utilipvs.Interface, ipset u
} }
p := &Proxier{ p := &Proxier{
exec: fexec, exec: fexec,
serviceMap: make(proxy.ServiceMap), svcPortMap: make(proxy.ServicePortMap),
serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, ipFamily, nil, nil), serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, ipFamily, nil, nil),
endpointsMap: make(proxy.EndpointsMap), endpointsMap: make(proxy.EndpointsMap),
endpointsChanges: proxy.NewEndpointChangeTracker(testHostname, nil, ipFamily, nil, nil), endpointsChanges: proxy.NewEndpointChangeTracker(testHostname, nil, ipFamily, nil, nil),
@ -2599,9 +2599,9 @@ func TestBuildServiceMapAddRemove(t *testing.T) {
for i := range services { for i := range services {
fp.OnServiceAdd(services[i]) fp.OnServiceAdd(services[i])
} }
result := fp.serviceMap.Update(fp.serviceChanges) result := fp.svcPortMap.Update(fp.serviceChanges)
if len(fp.serviceMap) != 12 { if len(fp.svcPortMap) != 12 {
t.Errorf("expected service map length 12, got %v", fp.serviceMap) t.Errorf("expected service map length 12, got %v", fp.svcPortMap)
} }
// The only-local-loadbalancer ones get added // The only-local-loadbalancer ones get added
@ -2632,9 +2632,9 @@ func TestBuildServiceMapAddRemove(t *testing.T) {
fp.OnServiceDelete(services[2]) fp.OnServiceDelete(services[2])
fp.OnServiceDelete(services[3]) fp.OnServiceDelete(services[3])
result = fp.serviceMap.Update(fp.serviceChanges) result = fp.svcPortMap.Update(fp.serviceChanges)
if len(fp.serviceMap) != 1 { if len(fp.svcPortMap) != 1 {
t.Errorf("expected service map length 1, got %v", fp.serviceMap) t.Errorf("expected service map length 1, got %v", fp.svcPortMap)
} }
if len(result.HCServiceNodePorts) != 0 { if len(result.HCServiceNodePorts) != 0 {
@ -2679,9 +2679,9 @@ func TestBuildServiceMapServiceHeadless(t *testing.T) {
) )
// Headless service should be ignored // Headless service should be ignored
result := fp.serviceMap.Update(fp.serviceChanges) result := fp.svcPortMap.Update(fp.serviceChanges)
if len(fp.serviceMap) != 0 { if len(fp.svcPortMap) != 0 {
t.Errorf("expected service map length 0, got %d", len(fp.serviceMap)) t.Errorf("expected service map length 0, got %d", len(fp.svcPortMap))
} }
// No proxied services, so no healthchecks // No proxied services, so no healthchecks
@ -2709,9 +2709,9 @@ func TestBuildServiceMapServiceTypeExternalName(t *testing.T) {
}), }),
) )
result := fp.serviceMap.Update(fp.serviceChanges) result := fp.svcPortMap.Update(fp.serviceChanges)
if len(fp.serviceMap) != 0 { if len(fp.svcPortMap) != 0 {
t.Errorf("expected service map length 0, got %v", fp.serviceMap) t.Errorf("expected service map length 0, got %v", fp.svcPortMap)
} }
// No proxied services, so no healthchecks // No proxied services, so no healthchecks
if len(result.HCServiceNodePorts) != 0 { if len(result.HCServiceNodePorts) != 0 {
@ -2751,9 +2751,9 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
fp.OnServiceAdd(servicev1) fp.OnServiceAdd(servicev1)
result := fp.serviceMap.Update(fp.serviceChanges) result := fp.svcPortMap.Update(fp.serviceChanges)
if len(fp.serviceMap) != 2 { if len(fp.svcPortMap) != 2 {
t.Errorf("expected service map length 2, got %v", fp.serviceMap) t.Errorf("expected service map length 2, got %v", fp.svcPortMap)
} }
if len(result.HCServiceNodePorts) != 0 { if len(result.HCServiceNodePorts) != 0 {
t.Errorf("expected healthcheck ports length 0, got %v", result.HCServiceNodePorts) t.Errorf("expected healthcheck ports length 0, got %v", result.HCServiceNodePorts)
@ -2765,9 +2765,9 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
// Change service to load-balancer // Change service to load-balancer
fp.OnServiceUpdate(servicev1, servicev2) fp.OnServiceUpdate(servicev1, servicev2)
result = fp.serviceMap.Update(fp.serviceChanges) result = fp.svcPortMap.Update(fp.serviceChanges)
if len(fp.serviceMap) != 2 { if len(fp.svcPortMap) != 2 {
t.Errorf("expected service map length 2, got %v", fp.serviceMap) t.Errorf("expected service map length 2, got %v", fp.svcPortMap)
} }
if len(result.HCServiceNodePorts) != 1 { if len(result.HCServiceNodePorts) != 1 {
t.Errorf("expected healthcheck ports length 1, got %v", result.HCServiceNodePorts) t.Errorf("expected healthcheck ports length 1, got %v", result.HCServiceNodePorts)
@ -2779,9 +2779,9 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
// No change; make sure the service map stays the same and there are // No change; make sure the service map stays the same and there are
// no health-check changes // no health-check changes
fp.OnServiceUpdate(servicev2, servicev2) fp.OnServiceUpdate(servicev2, servicev2)
result = fp.serviceMap.Update(fp.serviceChanges) result = fp.svcPortMap.Update(fp.serviceChanges)
if len(fp.serviceMap) != 2 { if len(fp.svcPortMap) != 2 {
t.Errorf("expected service map length 2, got %v", fp.serviceMap) t.Errorf("expected service map length 2, got %v", fp.svcPortMap)
} }
if len(result.HCServiceNodePorts) != 1 { if len(result.HCServiceNodePorts) != 1 {
t.Errorf("expected healthcheck ports length 1, got %v", result.HCServiceNodePorts) t.Errorf("expected healthcheck ports length 1, got %v", result.HCServiceNodePorts)
@ -2792,9 +2792,9 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
// And back to ClusterIP // And back to ClusterIP
fp.OnServiceUpdate(servicev2, servicev1) fp.OnServiceUpdate(servicev2, servicev1)
result = fp.serviceMap.Update(fp.serviceChanges) result = fp.svcPortMap.Update(fp.serviceChanges)
if len(fp.serviceMap) != 2 { if len(fp.svcPortMap) != 2 {
t.Errorf("expected service map length 2, got %v", fp.serviceMap) t.Errorf("expected service map length 2, got %v", fp.svcPortMap)
} }
if len(result.HCServiceNodePorts) != 0 { if len(result.HCServiceNodePorts) != 0 {
t.Errorf("expected healthcheck ports length 0, got %v", result.HCServiceNodePorts) t.Errorf("expected healthcheck ports length 0, got %v", result.HCServiceNodePorts)

144
pkg/proxy/service.go
View File

@ -37,11 +37,11 @@ import (
utilproxy "k8s.io/kubernetes/pkg/proxy/util" utilproxy "k8s.io/kubernetes/pkg/proxy/util"
) )
// BaseServiceInfo contains base information that defines a service. // BaseServicePortInfo contains base information that defines a service.
// This could be used directly by proxier while processing services, // This could be used directly by proxier while processing services,
// or can be used for constructing a more specific ServiceInfo struct // or can be used for constructing a more specific ServiceInfo struct
// defined by the proxier if needed. // defined by the proxier if needed.
type BaseServiceInfo struct { type BaseServicePortInfo struct {
clusterIP net.IP clusterIP net.IP
port int port int
protocol v1.Protocol protocol v1.Protocol
@ -58,106 +58,106 @@ type BaseServiceInfo struct {
hintsAnnotation string hintsAnnotation string
} }
var _ ServicePort = &BaseServiceInfo{} var _ ServicePort = &BaseServicePortInfo{}
// String is part of ServicePort interface. // String is part of ServicePort interface.
func (info *BaseServiceInfo) String() string { func (bsvcPortInfo *BaseServicePortInfo) String() string {
return fmt.Sprintf("%s:%d/%s", info.clusterIP, info.port, info.protocol) return fmt.Sprintf("%s:%d/%s", bsvcPortInfo.clusterIP, bsvcPortInfo.port, bsvcPortInfo.protocol)
} }
// ClusterIP is part of ServicePort interface. // ClusterIP is part of ServicePort interface.
func (info *BaseServiceInfo) ClusterIP() net.IP { func (bsvcPortInfo *BaseServicePortInfo) ClusterIP() net.IP {
return info.clusterIP return bsvcPortInfo.clusterIP
} }
// Port is part of ServicePort interface. // Port is part of ServicePort interface.
func (info *BaseServiceInfo) Port() int { func (bsvcPortInfo *BaseServicePortInfo) Port() int {
return info.port return bsvcPortInfo.port
} }
// SessionAffinityType is part of the ServicePort interface. // SessionAffinityType is part of the ServicePort interface.
func (info *BaseServiceInfo) SessionAffinityType() v1.ServiceAffinity { func (bsvcPortInfo *BaseServicePortInfo) SessionAffinityType() v1.ServiceAffinity {
return info.sessionAffinityType return bsvcPortInfo.sessionAffinityType
} }
// StickyMaxAgeSeconds is part of the ServicePort interface // StickyMaxAgeSeconds is part of the ServicePort interface
func (info *BaseServiceInfo) StickyMaxAgeSeconds() int { func (bsvcPortInfo *BaseServicePortInfo) StickyMaxAgeSeconds() int {
return info.stickyMaxAgeSeconds return bsvcPortInfo.stickyMaxAgeSeconds
} }
// Protocol is part of ServicePort interface. // Protocol is part of ServicePort interface.
func (info *BaseServiceInfo) Protocol() v1.Protocol { func (bsvcPortInfo *BaseServicePortInfo) Protocol() v1.Protocol {
return info.protocol return bsvcPortInfo.protocol
} }
// LoadBalancerSourceRanges is part of ServicePort interface // LoadBalancerSourceRanges is part of ServicePort interface
func (info *BaseServiceInfo) LoadBalancerSourceRanges() []string { func (bsvcPortInfo *BaseServicePortInfo) LoadBalancerSourceRanges() []string {
return info.loadBalancerSourceRanges return bsvcPortInfo.loadBalancerSourceRanges
} }
// HealthCheckNodePort is part of ServicePort interface. // HealthCheckNodePort is part of ServicePort interface.
func (info *BaseServiceInfo) HealthCheckNodePort() int { func (bsvcPortInfo *BaseServicePortInfo) HealthCheckNodePort() int {
return info.healthCheckNodePort return bsvcPortInfo.healthCheckNodePort
} }
// NodePort is part of the ServicePort interface. // NodePort is part of the ServicePort interface.
func (info *BaseServiceInfo) NodePort() int { func (bsvcPortInfo *BaseServicePortInfo) NodePort() int {
return info.nodePort return bsvcPortInfo.nodePort
} }
// ExternalIPStrings is part of ServicePort interface. // ExternalIPStrings is part of ServicePort interface.
func (info *BaseServiceInfo) ExternalIPStrings() []string { func (bsvcPortInfo *BaseServicePortInfo) ExternalIPStrings() []string {
return info.externalIPs return bsvcPortInfo.externalIPs
} }
// LoadBalancerIPStrings is part of ServicePort interface. // LoadBalancerIPStrings is part of ServicePort interface.
func (info *BaseServiceInfo) LoadBalancerIPStrings() []string { func (bsvcPortInfo *BaseServicePortInfo) LoadBalancerIPStrings() []string {
var ips []string var ips []string
for _, ing := range info.loadBalancerStatus.Ingress { for _, ing := range bsvcPortInfo.loadBalancerStatus.Ingress {
ips = append(ips, ing.IP) ips = append(ips, ing.IP)
} }
return ips return ips
} }
// ExternalPolicyLocal is part of ServicePort interface. // ExternalPolicyLocal is part of ServicePort interface.
func (info *BaseServiceInfo) ExternalPolicyLocal() bool { func (bsvcPortInfo *BaseServicePortInfo) ExternalPolicyLocal() bool {
return info.externalPolicyLocal return bsvcPortInfo.externalPolicyLocal
} }
// InternalPolicyLocal is part of ServicePort interface // InternalPolicyLocal is part of ServicePort interface
func (info *BaseServiceInfo) InternalPolicyLocal() bool { func (bsvcPortInfo *BaseServicePortInfo) InternalPolicyLocal() bool {
return info.internalPolicyLocal return bsvcPortInfo.internalPolicyLocal
} }
// InternalTrafficPolicy is part of ServicePort interface // InternalTrafficPolicy is part of ServicePort interface
func (info *BaseServiceInfo) InternalTrafficPolicy() *v1.ServiceInternalTrafficPolicyType { func (bsvcPortInfo *BaseServicePortInfo) InternalTrafficPolicy() *v1.ServiceInternalTrafficPolicyType {
return info.internalTrafficPolicy return bsvcPortInfo.internalTrafficPolicy
} }
// HintsAnnotation is part of ServicePort interface. // HintsAnnotation is part of ServicePort interface.
func (info *BaseServiceInfo) HintsAnnotation() string { func (bsvcPortInfo *BaseServicePortInfo) HintsAnnotation() string {
return info.hintsAnnotation return bsvcPortInfo.hintsAnnotation
} }
// ExternallyAccessible is part of ServicePort interface. // ExternallyAccessible is part of ServicePort interface.
func (info *BaseServiceInfo) ExternallyAccessible() bool { func (bsvcPortInfo *BaseServicePortInfo) ExternallyAccessible() bool {
return info.nodePort != 0 || len(info.loadBalancerStatus.Ingress) != 0 || len(info.externalIPs) != 0 return bsvcPortInfo.nodePort != 0 || len(bsvcPortInfo.loadBalancerStatus.Ingress) != 0 || len(bsvcPortInfo.externalIPs) != 0
} }
// UsesClusterEndpoints is part of ServicePort interface. // UsesClusterEndpoints is part of ServicePort interface.
func (info *BaseServiceInfo) UsesClusterEndpoints() bool { func (bsvcPortInfo *BaseServicePortInfo) UsesClusterEndpoints() bool {
// The service port uses Cluster endpoints if the internal traffic policy is "Cluster", // The service port uses Cluster endpoints if the internal traffic policy is "Cluster",
// or if it accepts external traffic at all. (Even if the external traffic policy is // or if it accepts external traffic at all. (Even if the external traffic policy is
// "Local", we need Cluster endpoints to implement short circuiting.) // "Local", we need Cluster endpoints to implement short circuiting.)
return !info.internalPolicyLocal || info.ExternallyAccessible() return !bsvcPortInfo.internalPolicyLocal || bsvcPortInfo.ExternallyAccessible()
} }
// UsesLocalEndpoints is part of ServicePort interface. // UsesLocalEndpoints is part of ServicePort interface.
func (info *BaseServiceInfo) UsesLocalEndpoints() bool { func (bsvcPortInfo *BaseServicePortInfo) UsesLocalEndpoints() bool {
return info.internalPolicyLocal || (info.externalPolicyLocal && info.ExternallyAccessible()) return bsvcPortInfo.internalPolicyLocal || (bsvcPortInfo.externalPolicyLocal && bsvcPortInfo.ExternallyAccessible())
} }
func (sct *ServiceChangeTracker) newBaseServiceInfo(port *v1.ServicePort, service *v1.Service) *BaseServiceInfo { func (sct *ServiceChangeTracker) newBaseServiceInfo(port *v1.ServicePort, service *v1.Service) *BaseServicePortInfo {
externalPolicyLocal := false externalPolicyLocal := false
if apiservice.ExternalPolicyLocal(service) { if apiservice.ExternalPolicyLocal(service) {
externalPolicyLocal = true externalPolicyLocal = true
@ -173,7 +173,7 @@ func (sct *ServiceChangeTracker) newBaseServiceInfo(port *v1.ServicePort, servic
} }
clusterIP := utilproxy.GetClusterIPByFamily(sct.ipFamily, service) clusterIP := utilproxy.GetClusterIPByFamily(sct.ipFamily, service)
info := &BaseServiceInfo{ info := &BaseServicePortInfo{
clusterIP: netutils.ParseIPSloppy(clusterIP), clusterIP: netutils.ParseIPSloppy(clusterIP),
port: int(port.Port), port: int(port.Port),
protocol: port.Protocol, protocol: port.Protocol,
@ -245,18 +245,18 @@ func (sct *ServiceChangeTracker) newBaseServiceInfo(port *v1.ServicePort, servic
return info return info
} }
type makeServicePortFunc func(*v1.ServicePort, *v1.Service, *BaseServiceInfo) ServicePort type makeServicePortFunc func(*v1.ServicePort, *v1.Service, *BaseServicePortInfo) ServicePort
// This handler is invoked by the apply function on every change. This function should not modify the // This handler is invoked by the apply function on every change. This function should not modify the
// ServiceMap's but just use the changes for any Proxier specific cleanup. // ServicePortMap's but just use the changes for any Proxier specific cleanup.
type processServiceMapChangeFunc func(previous, current ServiceMap) type processServiceMapChangeFunc func(previous, current ServicePortMap)
// serviceChange contains all changes to services that happened since proxy rules were synced. For a single object, // 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, // changes are accumulated, i.e. previous is state from before applying the changes,
// current is state after applying all of the changes. // current is state after applying all of the changes.
type serviceChange struct { type serviceChange struct {
previous ServiceMap previous ServicePortMap
current ServiceMap current ServicePortMap
} }
// ServiceChangeTracker carries state about uncommitted changes to an arbitrary number of // ServiceChangeTracker carries state about uncommitted changes to an arbitrary number of
@ -352,13 +352,13 @@ type UpdateServiceMapResult struct {
UDPStaleClusterIP sets.String UDPStaleClusterIP sets.String
} }
// Update updates ServiceMap base on the given changes. // Update updates ServicePortMap base on the given changes.
func (sm ServiceMap) Update(changes *ServiceChangeTracker) (result UpdateServiceMapResult) { func (sm ServicePortMap) Update(changes *ServiceChangeTracker) (result UpdateServiceMapResult) {
result.UDPStaleClusterIP = sets.NewString() result.UDPStaleClusterIP = sets.NewString()
sm.apply(changes, result.UDPStaleClusterIP) sm.apply(changes, result.UDPStaleClusterIP)
// TODO: If this will appear to be computationally expensive, consider // TODO: If this will appear to be computationally expensive, consider
// computing this incrementally similarly to serviceMap. // computing this incrementally similarly to svcPortMap.
result.HCServiceNodePorts = make(map[types.NamespacedName]uint16) result.HCServiceNodePorts = make(map[types.NamespacedName]uint16)
for svcPortName, info := range sm { for svcPortName, info := range sm {
if info.HealthCheckNodePort() != 0 { if info.HealthCheckNodePort() != 0 {
@ -369,13 +369,13 @@ func (sm ServiceMap) Update(changes *ServiceChangeTracker) (result UpdateService
return result return result
} }
// ServiceMap maps a service to its ServicePort. // ServicePortMap maps a service to its ServicePort.
type ServiceMap map[ServicePortName]ServicePort type ServicePortMap map[ServicePortName]ServicePort
// serviceToServiceMap translates a single Service object to a ServiceMap. // serviceToServiceMap translates a single Service object to a ServicePortMap.
// //
// NOTE: service object should NOT be modified. // NOTE: service object should NOT be modified.
func (sct *ServiceChangeTracker) serviceToServiceMap(service *v1.Service) ServiceMap { func (sct *ServiceChangeTracker) serviceToServiceMap(service *v1.Service) ServicePortMap {
if service == nil { if service == nil {
return nil return nil
} }
@ -389,25 +389,25 @@ func (sct *ServiceChangeTracker) serviceToServiceMap(service *v1.Service) Servic
return nil return nil
} }
serviceMap := make(ServiceMap) svcPortMap := make(ServicePortMap)
svcName := types.NamespacedName{Namespace: service.Namespace, Name: service.Name} svcName := types.NamespacedName{Namespace: service.Namespace, Name: service.Name}
for i := range service.Spec.Ports { for i := range service.Spec.Ports {
servicePort := &service.Spec.Ports[i] servicePort := &service.Spec.Ports[i]
svcPortName := ServicePortName{NamespacedName: svcName, Port: servicePort.Name, Protocol: servicePort.Protocol} svcPortName := ServicePortName{NamespacedName: svcName, Port: servicePort.Name, Protocol: servicePort.Protocol}
baseSvcInfo := sct.newBaseServiceInfo(servicePort, service) baseSvcInfo := sct.newBaseServiceInfo(servicePort, service)
if sct.makeServiceInfo != nil { if sct.makeServiceInfo != nil {
serviceMap[svcPortName] = sct.makeServiceInfo(servicePort, service, baseSvcInfo) svcPortMap[svcPortName] = sct.makeServiceInfo(servicePort, service, baseSvcInfo)
} else { } else {
serviceMap[svcPortName] = baseSvcInfo svcPortMap[svcPortName] = baseSvcInfo
} }
} }
return serviceMap return svcPortMap
} }
// apply the changes to ServiceMap and update the stale udp cluster IP set. The UDPStaleClusterIP argument is passed in to store the // apply the changes to ServicePortMap and update the stale udp cluster IP set. The UDPStaleClusterIP argument is passed in to store the
// udp protocol service cluster ip when service is deleted from the ServiceMap. // udp protocol service cluster ip when service is deleted from the ServicePortMap.
// apply triggers processServiceMapChange on every change. // apply triggers processServiceMapChange on every change.
func (sm *ServiceMap) apply(changes *ServiceChangeTracker, UDPStaleClusterIP sets.String) { func (sm *ServicePortMap) apply(changes *ServiceChangeTracker, UDPStaleClusterIP sets.String) {
changes.lock.Lock() changes.lock.Lock()
defer changes.lock.Unlock() defer changes.lock.Unlock()
for _, change := range changes.items { for _, change := range changes.items {
@ -420,20 +420,20 @@ func (sm *ServiceMap) apply(changes *ServiceChangeTracker, UDPStaleClusterIP set
change.previous.filter(change.current) change.previous.filter(change.current)
sm.unmerge(change.previous, UDPStaleClusterIP) sm.unmerge(change.previous, UDPStaleClusterIP)
} }
// clear changes after applying them to ServiceMap. // clear changes after applying them to ServicePortMap.
changes.items = make(map[types.NamespacedName]*serviceChange) changes.items = make(map[types.NamespacedName]*serviceChange)
metrics.ServiceChangesPending.Set(0) metrics.ServiceChangesPending.Set(0)
} }
// merge adds other ServiceMap's elements to current ServiceMap. // merge adds other ServicePortMap's elements to current ServicePortMap.
// If collision, other ALWAYS win. Otherwise add the other to current. // If collision, other ALWAYS win. Otherwise add the other to current.
// In other words, if some elements in current collisions with other, update the current by other. // In other words, if some elements in current collisions with other, update the current by other.
// It returns a string type set which stores all the newly merged services' identifier, ServicePortName.String(), to help users // It returns a string type set which stores all the newly merged services' identifier, ServicePortName.String(), to help users
// tell if a service is deleted or updated. // tell if a service is deleted or updated.
// The returned value is one of the arguments of ServiceMap.unmerge(). // The returned value is one of the arguments of ServicePortMap.unmerge().
// ServiceMap A Merge ServiceMap B will do following 2 things: // ServicePortMap A Merge ServicePortMap B will do following 2 things:
// - update ServiceMap A. // - update ServicePortMap A.
// - produce a string set which stores all other ServiceMap's ServicePortName.String(). // - produce a string set which stores all other ServicePortMap's ServicePortName.String().
// //
// For example, // For example,
// - A{} // - A{}
@ -444,8 +444,8 @@ func (sm *ServiceMap) apply(changes *ServiceChangeTracker, UDPStaleClusterIP set
// - B{{"ns", "cluster-ip", "http"}: {"172.16.55.10", 1234, "TCP"}} // - B{{"ns", "cluster-ip", "http"}: {"172.16.55.10", 1234, "TCP"}}
// - A updated to be {{"ns", "cluster-ip", "http"}: {"172.16.55.10", 1234, "TCP"}} // - A updated to be {{"ns", "cluster-ip", "http"}: {"172.16.55.10", 1234, "TCP"}}
// - produce string set {"ns/cluster-ip:http"} // - produce string set {"ns/cluster-ip:http"}
func (sm *ServiceMap) merge(other ServiceMap) sets.String { func (sm *ServicePortMap) merge(other ServicePortMap) sets.String {
// existingPorts is going to store all identifiers of all services in `other` ServiceMap. // existingPorts is going to store all identifiers of all services in `other` ServicePortMap.
existingPorts := sets.NewString() existingPorts := sets.NewString()
for svcPortName, info := range other { for svcPortName, info := range other {
// Take ServicePortName.String() as the newly merged service's identifier and put it into existingPorts. // Take ServicePortName.String() as the newly merged service's identifier and put it into existingPorts.
@ -461,8 +461,8 @@ func (sm *ServiceMap) merge(other ServiceMap) sets.String {
return existingPorts return existingPorts
} }
// filter filters out elements from ServiceMap base on given ports string sets. // filter filters out elements from ServicePortMap base on given ports string sets.
func (sm *ServiceMap) filter(other ServiceMap) { func (sm *ServicePortMap) filter(other ServicePortMap) {
for svcPortName := range *sm { for svcPortName := range *sm {
// skip the delete for Update event. // skip the delete for Update event.
if _, ok := other[svcPortName]; ok { if _, ok := other[svcPortName]; ok {
@ -471,9 +471,9 @@ func (sm *ServiceMap) filter(other ServiceMap) {
} }
} }
// unmerge deletes all other ServiceMap's elements from current ServiceMap. We pass in the UDPStaleClusterIP strings sets // unmerge deletes all other ServicePortMap's elements from current ServicePortMap. We pass in the UDPStaleClusterIP strings sets
// for storing the stale udp service cluster IPs. We will clear stale udp connection base on UDPStaleClusterIP later // for storing the stale udp service cluster IPs. We will clear stale udp connection base on UDPStaleClusterIP later
func (sm *ServiceMap) unmerge(other ServiceMap, UDPStaleClusterIP sets.String) { func (sm *ServicePortMap) unmerge(other ServicePortMap, UDPStaleClusterIP sets.String) {
for svcPortName := range other { for svcPortName := range other {
info, exists := (*sm)[svcPortName] info, exists := (*sm)[svcPortName]
if exists { if exists {

148
pkg/proxy/service_test.go
View File

@ -33,19 +33,19 @@ import (
const testHostname = "test-hostname" const testHostname = "test-hostname"
func makeTestServiceInfo(clusterIP string, port int, protocol string, healthcheckNodePort int, svcInfoFuncs ...func(*BaseServiceInfo)) *BaseServiceInfo { func makeTestServiceInfo(clusterIP string, port int, protocol string, healthcheckNodePort int, svcInfoFuncs ...func(*BaseServicePortInfo)) *BaseServicePortInfo {
info := &BaseServiceInfo{ bsvcPortInfo := &BaseServicePortInfo{
clusterIP: netutils.ParseIPSloppy(clusterIP), clusterIP: netutils.ParseIPSloppy(clusterIP),
port: port, port: port,
protocol: v1.Protocol(protocol), protocol: v1.Protocol(protocol),
} }
if healthcheckNodePort != 0 { if healthcheckNodePort != 0 {
info.healthCheckNodePort = healthcheckNodePort bsvcPortInfo.healthCheckNodePort = healthcheckNodePort
} }
for _, svcInfoFunc := range svcInfoFuncs { for _, svcInfoFunc := range svcInfoFuncs {
svcInfoFunc(info) svcInfoFunc(bsvcPortInfo)
} }
return info return bsvcPortInfo
} }
func makeTestService(namespace, name string, svcFunc func(*v1.Service)) *v1.Service { func makeTestService(namespace, name string, svcFunc func(*v1.Service)) *v1.Service {
@ -96,7 +96,7 @@ func TestServiceToServiceMap(t *testing.T) {
testCases := []struct { testCases := []struct {
desc string desc string
service *v1.Service service *v1.Service
expected map[ServicePortName]*BaseServiceInfo expected map[ServicePortName]*BaseServicePortInfo
ipFamily v1.IPFamily ipFamily v1.IPFamily
}{ }{
{ {
@ -104,7 +104,7 @@ func TestServiceToServiceMap(t *testing.T) {
ipFamily: v1.IPv4Protocol, ipFamily: v1.IPv4Protocol,
service: nil, service: nil,
expected: map[ServicePortName]*BaseServiceInfo{}, expected: map[ServicePortName]*BaseServicePortInfo{},
}, },
{ {
desc: "headless service", desc: "headless service",
@ -115,7 +115,7 @@ func TestServiceToServiceMap(t *testing.T) {
svc.Spec.ClusterIP = v1.ClusterIPNone svc.Spec.ClusterIP = v1.ClusterIPNone
svc.Spec.Ports = addTestPort(svc.Spec.Ports, "rpc", "UDP", 1234, 0, 0) svc.Spec.Ports = addTestPort(svc.Spec.Ports, "rpc", "UDP", 1234, 0, 0)
}), }),
expected: map[ServicePortName]*BaseServiceInfo{}, expected: map[ServicePortName]*BaseServicePortInfo{},
}, },
{ {
desc: "headless sctp service", desc: "headless sctp service",
@ -126,7 +126,7 @@ func TestServiceToServiceMap(t *testing.T) {
svc.Spec.ClusterIP = v1.ClusterIPNone svc.Spec.ClusterIP = v1.ClusterIPNone
svc.Spec.Ports = addTestPort(svc.Spec.Ports, "sip", "SCTP", 7777, 0, 0) svc.Spec.Ports = addTestPort(svc.Spec.Ports, "sip", "SCTP", 7777, 0, 0)
}), }),
expected: map[ServicePortName]*BaseServiceInfo{}, expected: map[ServicePortName]*BaseServicePortInfo{},
}, },
{ {
desc: "headless service without port", desc: "headless service without port",
@ -136,7 +136,7 @@ func TestServiceToServiceMap(t *testing.T) {
svc.Spec.Type = v1.ServiceTypeClusterIP svc.Spec.Type = v1.ServiceTypeClusterIP
svc.Spec.ClusterIP = v1.ClusterIPNone svc.Spec.ClusterIP = v1.ClusterIPNone
}), }),
expected: map[ServicePortName]*BaseServiceInfo{}, expected: map[ServicePortName]*BaseServicePortInfo{},
}, },
{ {
desc: "cluster ip service", desc: "cluster ip service",
@ -148,7 +148,7 @@ func TestServiceToServiceMap(t *testing.T) {
svc.Spec.Ports = addTestPort(svc.Spec.Ports, "p1", "UDP", 1234, 4321, 0) svc.Spec.Ports = addTestPort(svc.Spec.Ports, "p1", "UDP", 1234, 4321, 0)
svc.Spec.Ports = addTestPort(svc.Spec.Ports, "p2", "UDP", 1235, 5321, 0) svc.Spec.Ports = addTestPort(svc.Spec.Ports, "p2", "UDP", 1235, 5321, 0)
}), }),
expected: map[ServicePortName]*BaseServiceInfo{ expected: map[ServicePortName]*BaseServicePortInfo{
makeServicePortName("ns2", "cluster-ip", "p1", v1.ProtocolUDP): makeTestServiceInfo("172.16.55.4", 1234, "UDP", 0), makeServicePortName("ns2", "cluster-ip", "p1", v1.ProtocolUDP): makeTestServiceInfo("172.16.55.4", 1234, "UDP", 0),
makeServicePortName("ns2", "cluster-ip", "p2", v1.ProtocolUDP): makeTestServiceInfo("172.16.55.4", 1235, "UDP", 0), makeServicePortName("ns2", "cluster-ip", "p2", v1.ProtocolUDP): makeTestServiceInfo("172.16.55.4", 1235, "UDP", 0),
}, },
@ -163,7 +163,7 @@ func TestServiceToServiceMap(t *testing.T) {
svc.Spec.Ports = addTestPort(svc.Spec.Ports, "port1", "UDP", 345, 678, 0) svc.Spec.Ports = addTestPort(svc.Spec.Ports, "port1", "UDP", 345, 678, 0)
svc.Spec.Ports = addTestPort(svc.Spec.Ports, "port2", "TCP", 344, 677, 0) svc.Spec.Ports = addTestPort(svc.Spec.Ports, "port2", "TCP", 344, 677, 0)
}), }),
expected: map[ServicePortName]*BaseServiceInfo{ expected: map[ServicePortName]*BaseServicePortInfo{
makeServicePortName("ns2", "node-port", "port1", v1.ProtocolUDP): makeTestServiceInfo("172.16.55.10", 345, "UDP", 0), makeServicePortName("ns2", "node-port", "port1", v1.ProtocolUDP): makeTestServiceInfo("172.16.55.10", 345, "UDP", 0),
makeServicePortName("ns2", "node-port", "port2", v1.ProtocolTCP): makeTestServiceInfo("172.16.55.10", 344, "TCP", 0), makeServicePortName("ns2", "node-port", "port2", v1.ProtocolTCP): makeTestServiceInfo("172.16.55.10", 344, "TCP", 0),
}, },
@ -180,12 +180,12 @@ func TestServiceToServiceMap(t *testing.T) {
svc.Spec.Ports = addTestPort(svc.Spec.Ports, "port4", "UDP", 8676, 30062, 7001) svc.Spec.Ports = addTestPort(svc.Spec.Ports, "port4", "UDP", 8676, 30062, 7001)
svc.Status.LoadBalancer.Ingress = []v1.LoadBalancerIngress{{IP: "10.1.2.4"}} svc.Status.LoadBalancer.Ingress = []v1.LoadBalancerIngress{{IP: "10.1.2.4"}}
}), }),
expected: map[ServicePortName]*BaseServiceInfo{ expected: map[ServicePortName]*BaseServicePortInfo{
makeServicePortName("ns1", "load-balancer", "port3", v1.ProtocolUDP): makeTestServiceInfo("172.16.55.11", 8675, "UDP", 0, func(info *BaseServiceInfo) { makeServicePortName("ns1", "load-balancer", "port3", v1.ProtocolUDP): makeTestServiceInfo("172.16.55.11", 8675, "UDP", 0, func(bsvcPortInfo *BaseServicePortInfo) {
info.loadBalancerStatus.Ingress = []v1.LoadBalancerIngress{{IP: "10.1.2.4"}} bsvcPortInfo.loadBalancerStatus.Ingress = []v1.LoadBalancerIngress{{IP: "10.1.2.4"}}
}), }),
makeServicePortName("ns1", "load-balancer", "port4", v1.ProtocolUDP): makeTestServiceInfo("172.16.55.11", 8676, "UDP", 0, func(info *BaseServiceInfo) { makeServicePortName("ns1", "load-balancer", "port4", v1.ProtocolUDP): makeTestServiceInfo("172.16.55.11", 8676, "UDP", 0, func(bsvcPortInfo *BaseServicePortInfo) {
info.loadBalancerStatus.Ingress = []v1.LoadBalancerIngress{{IP: "10.1.2.4"}} bsvcPortInfo.loadBalancerStatus.Ingress = []v1.LoadBalancerIngress{{IP: "10.1.2.4"}}
}), }),
}, },
}, },
@ -203,12 +203,12 @@ func TestServiceToServiceMap(t *testing.T) {
svc.Spec.ExternalTrafficPolicy = v1.ServiceExternalTrafficPolicyTypeLocal svc.Spec.ExternalTrafficPolicy = v1.ServiceExternalTrafficPolicyTypeLocal
svc.Spec.HealthCheckNodePort = 345 svc.Spec.HealthCheckNodePort = 345
}), }),
expected: map[ServicePortName]*BaseServiceInfo{ expected: map[ServicePortName]*BaseServicePortInfo{
makeServicePortName("ns1", "only-local-load-balancer", "portx", v1.ProtocolUDP): makeTestServiceInfo("172.16.55.12", 8677, "UDP", 345, func(info *BaseServiceInfo) { makeServicePortName("ns1", "only-local-load-balancer", "portx", v1.ProtocolUDP): makeTestServiceInfo("172.16.55.12", 8677, "UDP", 345, func(bsvcPortInfo *BaseServicePortInfo) {
info.loadBalancerStatus.Ingress = []v1.LoadBalancerIngress{{IP: "10.1.2.3"}} bsvcPortInfo.loadBalancerStatus.Ingress = []v1.LoadBalancerIngress{{IP: "10.1.2.3"}}
}), }),
makeServicePortName("ns1", "only-local-load-balancer", "porty", v1.ProtocolUDP): makeTestServiceInfo("172.16.55.12", 8678, "UDP", 345, func(info *BaseServiceInfo) { makeServicePortName("ns1", "only-local-load-balancer", "porty", v1.ProtocolUDP): makeTestServiceInfo("172.16.55.12", 8678, "UDP", 345, func(bsvcPortInfo *BaseServicePortInfo) {
info.loadBalancerStatus.Ingress = []v1.LoadBalancerIngress{{IP: "10.1.2.3"}} bsvcPortInfo.loadBalancerStatus.Ingress = []v1.LoadBalancerIngress{{IP: "10.1.2.3"}}
}), }),
}, },
}, },
@ -222,7 +222,7 @@ func TestServiceToServiceMap(t *testing.T) {
svc.Spec.ExternalName = "foo2.bar.com" svc.Spec.ExternalName = "foo2.bar.com"
svc.Spec.Ports = addTestPort(svc.Spec.Ports, "portz", "UDP", 1235, 5321, 0) svc.Spec.Ports = addTestPort(svc.Spec.Ports, "portz", "UDP", 1235, 5321, 0)
}), }),
expected: map[ServicePortName]*BaseServiceInfo{}, expected: map[ServicePortName]*BaseServicePortInfo{},
}, },
{ {
desc: "service with ipv6 clusterIP under ipv4 mode, service should be filtered", desc: "service with ipv6 clusterIP under ipv4 mode, service should be filtered",
@ -312,11 +312,11 @@ func TestServiceToServiceMap(t *testing.T) {
}, },
}, },
}, },
expected: map[ServicePortName]*BaseServiceInfo{ expected: map[ServicePortName]*BaseServicePortInfo{
makeServicePortName("test", "validIPv4", "testPort", v1.ProtocolTCP): makeTestServiceInfo(testClusterIPv4, 12345, "TCP", 0, func(info *BaseServiceInfo) { makeServicePortName("test", "validIPv4", "testPort", v1.ProtocolTCP): makeTestServiceInfo(testClusterIPv4, 12345, "TCP", 0, func(bsvcPortInfo *BaseServicePortInfo) {
info.externalIPs = []string{testExternalIPv4} bsvcPortInfo.externalIPs = []string{testExternalIPv4}
info.loadBalancerSourceRanges = []string{testSourceRangeIPv4} bsvcPortInfo.loadBalancerSourceRanges = []string{testSourceRangeIPv4}
info.loadBalancerStatus.Ingress = []v1.LoadBalancerIngress{{IP: testExternalIPv4}} bsvcPortInfo.loadBalancerStatus.Ingress = []v1.LoadBalancerIngress{{IP: testExternalIPv4}}
}), }),
}, },
}, },
@ -350,11 +350,11 @@ func TestServiceToServiceMap(t *testing.T) {
}, },
}, },
}, },
expected: map[ServicePortName]*BaseServiceInfo{ expected: map[ServicePortName]*BaseServicePortInfo{
makeServicePortName("test", "validIPv6", "testPort", v1.ProtocolTCP): makeTestServiceInfo(testClusterIPv6, 12345, "TCP", 0, func(info *BaseServiceInfo) { makeServicePortName("test", "validIPv6", "testPort", v1.ProtocolTCP): makeTestServiceInfo(testClusterIPv6, 12345, "TCP", 0, func(bsvcPortInfo *BaseServicePortInfo) {
info.externalIPs = []string{testExternalIPv6} bsvcPortInfo.externalIPs = []string{testExternalIPv6}
info.loadBalancerSourceRanges = []string{testSourceRangeIPv6} bsvcPortInfo.loadBalancerSourceRanges = []string{testSourceRangeIPv6}
info.loadBalancerStatus.Ingress = []v1.LoadBalancerIngress{{IP: testExternalIPv6}} bsvcPortInfo.loadBalancerStatus.Ingress = []v1.LoadBalancerIngress{{IP: testExternalIPv6}}
}), }),
}, },
}, },
@ -388,11 +388,11 @@ func TestServiceToServiceMap(t *testing.T) {
}, },
}, },
}, },
expected: map[ServicePortName]*BaseServiceInfo{ expected: map[ServicePortName]*BaseServicePortInfo{
makeServicePortName("test", "filterIPv6InIPV4Mode", "testPort", v1.ProtocolTCP): makeTestServiceInfo(testClusterIPv4, 12345, "TCP", 0, func(info *BaseServiceInfo) { makeServicePortName("test", "filterIPv6InIPV4Mode", "testPort", v1.ProtocolTCP): makeTestServiceInfo(testClusterIPv4, 12345, "TCP", 0, func(bsvcPortInfo *BaseServicePortInfo) {
info.externalIPs = []string{testExternalIPv4} bsvcPortInfo.externalIPs = []string{testExternalIPv4}
info.loadBalancerSourceRanges = []string{testSourceRangeIPv4} bsvcPortInfo.loadBalancerSourceRanges = []string{testSourceRangeIPv4}
info.loadBalancerStatus.Ingress = []v1.LoadBalancerIngress{{IP: testExternalIPv4}} bsvcPortInfo.loadBalancerStatus.Ingress = []v1.LoadBalancerIngress{{IP: testExternalIPv4}}
}), }),
}, },
}, },
@ -426,11 +426,11 @@ func TestServiceToServiceMap(t *testing.T) {
}, },
}, },
}, },
expected: map[ServicePortName]*BaseServiceInfo{ expected: map[ServicePortName]*BaseServicePortInfo{
makeServicePortName("test", "filterIPv4InIPV6Mode", "testPort", v1.ProtocolTCP): makeTestServiceInfo(testClusterIPv6, 12345, "TCP", 0, func(info *BaseServiceInfo) { makeServicePortName("test", "filterIPv4InIPV6Mode", "testPort", v1.ProtocolTCP): makeTestServiceInfo(testClusterIPv6, 12345, "TCP", 0, func(bsvcPortInfo *BaseServicePortInfo) {
info.externalIPs = []string{testExternalIPv6} bsvcPortInfo.externalIPs = []string{testExternalIPv6}
info.loadBalancerSourceRanges = []string{testSourceRangeIPv6} bsvcPortInfo.loadBalancerSourceRanges = []string{testSourceRangeIPv6}
info.loadBalancerStatus.Ingress = []v1.LoadBalancerIngress{{IP: testExternalIPv6}} bsvcPortInfo.loadBalancerStatus.Ingress = []v1.LoadBalancerIngress{{IP: testExternalIPv6}}
}), }),
}, },
}, },
@ -455,9 +455,9 @@ func TestServiceToServiceMap(t *testing.T) {
}, },
}, },
}, },
expected: map[ServicePortName]*BaseServiceInfo{ expected: map[ServicePortName]*BaseServicePortInfo{
makeServicePortName("test", "extra-space", "testPort", v1.ProtocolTCP): makeTestServiceInfo(testClusterIPv4, 12345, "TCP", 0, func(info *BaseServiceInfo) { makeServicePortName("test", "extra-space", "testPort", v1.ProtocolTCP): makeTestServiceInfo(testClusterIPv4, 12345, "TCP", 0, func(bsvcPortInfo *BaseServicePortInfo) {
info.loadBalancerSourceRanges = []string{"10.1.2.0/28"} bsvcPortInfo.loadBalancerSourceRanges = []string{"10.1.2.0/28"}
}), }),
}, },
}, },
@ -473,7 +473,7 @@ func TestServiceToServiceMap(t *testing.T) {
t.Fatalf("expected %d new, got %d: %v", len(tc.expected), len(newServices), spew.Sdump(newServices)) t.Fatalf("expected %d new, got %d: %v", len(tc.expected), len(newServices), spew.Sdump(newServices))
} }
for svcKey, expectedInfo := range tc.expected { for svcKey, expectedInfo := range tc.expected {
svcInfo, exists := newServices[svcKey].(*BaseServiceInfo) svcInfo, exists := newServices[svcKey].(*BaseServicePortInfo)
if !exists { if !exists {
t.Fatalf("[%s] expected to find key %s", tc.desc, svcKey) t.Fatalf("[%s] expected to find key %s", tc.desc, svcKey)
} }
@ -488,7 +488,7 @@ func TestServiceToServiceMap(t *testing.T) {
t.Errorf("[%s] expected new[%v]to be %v, got %v", tc.desc, svcKey, expectedInfo, *svcInfo) t.Errorf("[%s] expected new[%v]to be %v, got %v", tc.desc, svcKey, expectedInfo, *svcInfo)
} }
for svcKey, expectedInfo := range tc.expected { for svcKey, expectedInfo := range tc.expected {
svcInfo, _ := newServices[svcKey].(*BaseServiceInfo) svcInfo, _ := newServices[svcKey].(*BaseServicePortInfo)
if !svcInfo.clusterIP.Equal(expectedInfo.clusterIP) || if !svcInfo.clusterIP.Equal(expectedInfo.clusterIP) ||
svcInfo.port != expectedInfo.port || svcInfo.port != expectedInfo.port ||
svcInfo.protocol != expectedInfo.protocol || svcInfo.protocol != expectedInfo.protocol ||
@ -507,14 +507,14 @@ func TestServiceToServiceMap(t *testing.T) {
type FakeProxier struct { type FakeProxier struct {
endpointsChanges *EndpointChangeTracker endpointsChanges *EndpointChangeTracker
serviceChanges *ServiceChangeTracker serviceChanges *ServiceChangeTracker
serviceMap ServiceMap svcPortMap ServicePortMap
endpointsMap EndpointsMap endpointsMap EndpointsMap
hostname string hostname string
} }
func newFakeProxier(ipFamily v1.IPFamily, t time.Time) *FakeProxier { func newFakeProxier(ipFamily v1.IPFamily, t time.Time) *FakeProxier {
return &FakeProxier{ return &FakeProxier{
serviceMap: make(ServiceMap), svcPortMap: make(ServicePortMap),
serviceChanges: NewServiceChangeTracker(nil, ipFamily, nil, nil), serviceChanges: NewServiceChangeTracker(nil, ipFamily, nil, nil),
endpointsMap: make(EndpointsMap), endpointsMap: make(EndpointsMap),
endpointsChanges: &EndpointChangeTracker{ endpointsChanges: &EndpointChangeTracker{
@ -568,9 +568,9 @@ func TestServiceMapUpdateHeadless(t *testing.T) {
if pending.Len() != 0 { if pending.Len() != 0 {
t.Errorf("expected 0 pending service changes, got %d", pending.Len()) t.Errorf("expected 0 pending service changes, got %d", pending.Len())
} }
result := fp.serviceMap.Update(fp.serviceChanges) result := fp.svcPortMap.Update(fp.serviceChanges)
if len(fp.serviceMap) != 0 { if len(fp.svcPortMap) != 0 {
t.Errorf("expected service map length 0, got %d", len(fp.serviceMap)) t.Errorf("expected service map length 0, got %d", len(fp.svcPortMap))
} }
// No proxied services, so no healthchecks // No proxied services, so no healthchecks
@ -599,9 +599,9 @@ func TestUpdateServiceTypeExternalName(t *testing.T) {
if pending.Len() != 0 { if pending.Len() != 0 {
t.Errorf("expected 0 pending service changes, got %d", pending.Len()) t.Errorf("expected 0 pending service changes, got %d", pending.Len())
} }
result := fp.serviceMap.Update(fp.serviceChanges) result := fp.svcPortMap.Update(fp.serviceChanges)
if len(fp.serviceMap) != 0 { if len(fp.svcPortMap) != 0 {
t.Errorf("expected service map length 0, got %v", fp.serviceMap) t.Errorf("expected service map length 0, got %v", fp.svcPortMap)
} }
// No proxied services, so no healthchecks // No proxied services, so no healthchecks
if len(result.HCServiceNodePorts) != 0 { if len(result.HCServiceNodePorts) != 0 {
@ -671,9 +671,9 @@ func TestBuildServiceMapAddRemove(t *testing.T) {
t.Errorf("expected %d pending service changes, got %d", len(services), pending.Len()) t.Errorf("expected %d pending service changes, got %d", len(services), pending.Len())
} }
result := fp.serviceMap.Update(fp.serviceChanges) result := fp.svcPortMap.Update(fp.serviceChanges)
if len(fp.serviceMap) != 8 { if len(fp.svcPortMap) != 8 {
t.Errorf("expected service map length 2, got %v", fp.serviceMap) t.Errorf("expected service map length 2, got %v", fp.svcPortMap)
} }
// The only-local-loadbalancer ones get added // The only-local-loadbalancer ones get added
@ -708,9 +708,9 @@ func TestBuildServiceMapAddRemove(t *testing.T) {
if pending.Len() != 4 { if pending.Len() != 4 {
t.Errorf("expected 4 pending service changes, got %d", pending.Len()) t.Errorf("expected 4 pending service changes, got %d", pending.Len())
} }
result = fp.serviceMap.Update(fp.serviceChanges) result = fp.svcPortMap.Update(fp.serviceChanges)
if len(fp.serviceMap) != 1 { if len(fp.svcPortMap) != 1 {
t.Errorf("expected service map length 1, got %v", fp.serviceMap) t.Errorf("expected service map length 1, got %v", fp.svcPortMap)
} }
if len(result.HCServiceNodePorts) != 0 { if len(result.HCServiceNodePorts) != 0 {
@ -761,9 +761,9 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
if pending.Len() != 1 { if pending.Len() != 1 {
t.Errorf("expected 1 pending service change, got %d", pending.Len()) t.Errorf("expected 1 pending service change, got %d", pending.Len())
} }
result := fp.serviceMap.Update(fp.serviceChanges) result := fp.svcPortMap.Update(fp.serviceChanges)
if len(fp.serviceMap) != 2 { if len(fp.svcPortMap) != 2 {
t.Errorf("expected service map length 2, got %v", fp.serviceMap) t.Errorf("expected service map length 2, got %v", fp.svcPortMap)
} }
if len(result.HCServiceNodePorts) != 0 { if len(result.HCServiceNodePorts) != 0 {
t.Errorf("expected healthcheck ports length 0, got %v", result.HCServiceNodePorts) t.Errorf("expected healthcheck ports length 0, got %v", result.HCServiceNodePorts)
@ -779,9 +779,9 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
if pending.Len() != 1 { if pending.Len() != 1 {
t.Errorf("expected 1 pending service change, got %d", pending.Len()) t.Errorf("expected 1 pending service change, got %d", pending.Len())
} }
result = fp.serviceMap.Update(fp.serviceChanges) result = fp.svcPortMap.Update(fp.serviceChanges)
if len(fp.serviceMap) != 2 { if len(fp.svcPortMap) != 2 {
t.Errorf("expected service map length 2, got %v", fp.serviceMap) t.Errorf("expected service map length 2, got %v", fp.svcPortMap)
} }
if len(result.HCServiceNodePorts) != 1 { if len(result.HCServiceNodePorts) != 1 {
t.Errorf("expected healthcheck ports length 1, got %v", result.HCServiceNodePorts) t.Errorf("expected healthcheck ports length 1, got %v", result.HCServiceNodePorts)
@ -797,9 +797,9 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
if pending.Len() != 0 { if pending.Len() != 0 {
t.Errorf("expected 0 pending service changes, got %d", pending.Len()) t.Errorf("expected 0 pending service changes, got %d", pending.Len())
} }
result = fp.serviceMap.Update(fp.serviceChanges) result = fp.svcPortMap.Update(fp.serviceChanges)
if len(fp.serviceMap) != 2 { if len(fp.svcPortMap) != 2 {
t.Errorf("expected service map length 2, got %v", fp.serviceMap) t.Errorf("expected service map length 2, got %v", fp.svcPortMap)
} }
if len(result.HCServiceNodePorts) != 1 { if len(result.HCServiceNodePorts) != 1 {
t.Errorf("expected healthcheck ports length 1, got %v", result.HCServiceNodePorts) t.Errorf("expected healthcheck ports length 1, got %v", result.HCServiceNodePorts)
@ -814,9 +814,9 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
if pending.Len() != 1 { if pending.Len() != 1 {
t.Errorf("expected 1 pending service change, got %d", pending.Len()) t.Errorf("expected 1 pending service change, got %d", pending.Len())
} }
result = fp.serviceMap.Update(fp.serviceChanges) result = fp.svcPortMap.Update(fp.serviceChanges)
if len(fp.serviceMap) != 2 { if len(fp.svcPortMap) != 2 {
t.Errorf("expected service map length 2, got %v", fp.serviceMap) t.Errorf("expected service map length 2, got %v", fp.svcPortMap)
} }
if len(result.HCServiceNodePorts) != 0 { if len(result.HCServiceNodePorts) != 0 {
t.Errorf("expected healthcheck ports length 0, got %v", result.HCServiceNodePorts) t.Errorf("expected healthcheck ports length 0, got %v", result.HCServiceNodePorts)

74
pkg/proxy/topology_test.go
View File

@ -68,7 +68,7 @@ func TestCategorizeEndpoints(t *testing.T) {
name: "hints enabled, hints annotation == auto", name: "hints enabled, hints annotation == auto",
hintsEnabled: true, hintsEnabled: true,
nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"}, nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"},
serviceInfo: &BaseServiceInfo{hintsAnnotation: "auto"}, serviceInfo: &BaseServicePortInfo{hintsAnnotation: "auto"},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true},
&BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true},
@ -81,7 +81,7 @@ func TestCategorizeEndpoints(t *testing.T) {
name: "hints, hints annotation == disabled, hints ignored", name: "hints, hints annotation == disabled, hints ignored",
hintsEnabled: true, hintsEnabled: true,
nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"}, nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"},
serviceInfo: &BaseServiceInfo{hintsAnnotation: "disabled"}, serviceInfo: &BaseServicePortInfo{hintsAnnotation: "disabled"},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true},
&BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true},
@ -94,7 +94,7 @@ func TestCategorizeEndpoints(t *testing.T) {
name: "hints disabled, hints annotation == auto", name: "hints disabled, hints annotation == auto",
hintsEnabled: false, hintsEnabled: false,
nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"}, nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"},
serviceInfo: &BaseServiceInfo{hintsAnnotation: "auto"}, serviceInfo: &BaseServicePortInfo{hintsAnnotation: "auto"},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true},
&BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true},
@ -108,7 +108,7 @@ func TestCategorizeEndpoints(t *testing.T) {
name: "hints, hints annotation == aUto (wrong capitalization), hints ignored", name: "hints, hints annotation == aUto (wrong capitalization), hints ignored",
hintsEnabled: true, hintsEnabled: true,
nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"}, nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"},
serviceInfo: &BaseServiceInfo{hintsAnnotation: "aUto"}, serviceInfo: &BaseServicePortInfo{hintsAnnotation: "aUto"},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true},
&BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true},
@ -121,7 +121,7 @@ func TestCategorizeEndpoints(t *testing.T) {
name: "hints, hints annotation empty, hints ignored", name: "hints, hints annotation empty, hints ignored",
hintsEnabled: true, hintsEnabled: true,
nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"}, nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"},
serviceInfo: &BaseServiceInfo{}, serviceInfo: &BaseServicePortInfo{},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true},
&BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true},
@ -134,7 +134,7 @@ func TestCategorizeEndpoints(t *testing.T) {
name: "externalTrafficPolicy: Local, topology ignored for Local endpoints", name: "externalTrafficPolicy: Local, topology ignored for Local endpoints",
hintsEnabled: true, hintsEnabled: true,
nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"}, nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"},
serviceInfo: &BaseServiceInfo{externalPolicyLocal: true, nodePort: 8080, hintsAnnotation: "auto"}, serviceInfo: &BaseServicePortInfo{externalPolicyLocal: true, nodePort: 8080, hintsAnnotation: "auto"},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true, IsLocal: true}, &BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true, IsLocal: true},
&BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true, IsLocal: true}, &BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true, IsLocal: true},
@ -148,7 +148,7 @@ func TestCategorizeEndpoints(t *testing.T) {
name: "internalTrafficPolicy: Local, topology ignored for Local endpoints", name: "internalTrafficPolicy: Local, topology ignored for Local endpoints",
hintsEnabled: true, hintsEnabled: true,
nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"}, nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"},
serviceInfo: &BaseServiceInfo{internalPolicyLocal: true, hintsAnnotation: "auto", externalPolicyLocal: false, nodePort: 8080}, serviceInfo: &BaseServicePortInfo{internalPolicyLocal: true, hintsAnnotation: "auto", externalPolicyLocal: false, nodePort: 8080},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true, IsLocal: true}, &BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true, IsLocal: true},
&BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true, IsLocal: true}, &BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true, IsLocal: true},
@ -162,7 +162,7 @@ func TestCategorizeEndpoints(t *testing.T) {
name: "empty node labels", name: "empty node labels",
hintsEnabled: true, hintsEnabled: true,
nodeLabels: map[string]string{}, nodeLabels: map[string]string{},
serviceInfo: &BaseServiceInfo{hintsAnnotation: "auto"}, serviceInfo: &BaseServicePortInfo{hintsAnnotation: "auto"},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true},
}, },
@ -172,7 +172,7 @@ func TestCategorizeEndpoints(t *testing.T) {
name: "empty zone label", name: "empty zone label",
hintsEnabled: true, hintsEnabled: true,
nodeLabels: map[string]string{v1.LabelTopologyZone: ""}, nodeLabels: map[string]string{v1.LabelTopologyZone: ""},
serviceInfo: &BaseServiceInfo{hintsAnnotation: "auto"}, serviceInfo: &BaseServicePortInfo{hintsAnnotation: "auto"},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true},
}, },
@ -182,7 +182,7 @@ func TestCategorizeEndpoints(t *testing.T) {
name: "node in different zone, no endpoint filtering", name: "node in different zone, no endpoint filtering",
hintsEnabled: true, hintsEnabled: true,
nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-b"}, nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-b"},
serviceInfo: &BaseServiceInfo{hintsAnnotation: "auto"}, serviceInfo: &BaseServicePortInfo{hintsAnnotation: "auto"},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true},
}, },
@ -192,7 +192,7 @@ func TestCategorizeEndpoints(t *testing.T) {
name: "normal endpoint filtering, auto annotation", name: "normal endpoint filtering, auto annotation",
hintsEnabled: true, hintsEnabled: true,
nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"}, nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"},
serviceInfo: &BaseServiceInfo{hintsAnnotation: "auto"}, serviceInfo: &BaseServicePortInfo{hintsAnnotation: "auto"},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true},
&BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true},
@ -205,7 +205,7 @@ func TestCategorizeEndpoints(t *testing.T) {
name: "unready endpoint", name: "unready endpoint",
hintsEnabled: true, hintsEnabled: true,
nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"}, nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"},
serviceInfo: &BaseServiceInfo{hintsAnnotation: "auto"}, serviceInfo: &BaseServicePortInfo{hintsAnnotation: "auto"},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true},
&BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true},
@ -218,7 +218,7 @@ func TestCategorizeEndpoints(t *testing.T) {
name: "only unready endpoints in same zone (should not filter)", name: "only unready endpoints in same zone (should not filter)",
hintsEnabled: true, hintsEnabled: true,
nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"}, nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"},
serviceInfo: &BaseServiceInfo{hintsAnnotation: "auto"}, serviceInfo: &BaseServicePortInfo{hintsAnnotation: "auto"},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: false}, &BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: false},
&BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true},
@ -231,7 +231,7 @@ func TestCategorizeEndpoints(t *testing.T) {
name: "normal endpoint filtering, Auto annotation", name: "normal endpoint filtering, Auto annotation",
hintsEnabled: true, hintsEnabled: true,
nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"}, nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"},
serviceInfo: &BaseServiceInfo{hintsAnnotation: "Auto"}, serviceInfo: &BaseServicePortInfo{hintsAnnotation: "Auto"},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true},
&BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true},
@ -244,7 +244,7 @@ func TestCategorizeEndpoints(t *testing.T) {
name: "hintsAnnotation empty, no filtering applied", name: "hintsAnnotation empty, no filtering applied",
hintsEnabled: true, hintsEnabled: true,
nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"}, nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"},
serviceInfo: &BaseServiceInfo{hintsAnnotation: ""}, serviceInfo: &BaseServicePortInfo{hintsAnnotation: ""},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true},
&BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true},
@ -257,7 +257,7 @@ func TestCategorizeEndpoints(t *testing.T) {
name: "hintsAnnotation disabled, no filtering applied", name: "hintsAnnotation disabled, no filtering applied",
hintsEnabled: true, hintsEnabled: true,
nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"}, nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"},
serviceInfo: &BaseServiceInfo{hintsAnnotation: "disabled"}, serviceInfo: &BaseServicePortInfo{hintsAnnotation: "disabled"},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true},
&BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true},
@ -270,7 +270,7 @@ func TestCategorizeEndpoints(t *testing.T) {
name: "missing hints, no filtering applied", name: "missing hints, no filtering applied",
hintsEnabled: true, hintsEnabled: true,
nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"}, nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"},
serviceInfo: &BaseServiceInfo{hintsAnnotation: "auto"}, serviceInfo: &BaseServicePortInfo{hintsAnnotation: "auto"},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a"), Ready: true},
&BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b"), Ready: true},
@ -283,7 +283,7 @@ func TestCategorizeEndpoints(t *testing.T) {
name: "multiple hints per endpoint, filtering includes any endpoint with zone included", name: "multiple hints per endpoint, filtering includes any endpoint with zone included",
hintsEnabled: true, hintsEnabled: true,
nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-c"}, nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-c"},
serviceInfo: &BaseServiceInfo{hintsAnnotation: "auto"}, serviceInfo: &BaseServicePortInfo{hintsAnnotation: "auto"},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a", "zone-b", "zone-c"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.3:80", ZoneHints: sets.NewString("zone-a", "zone-b", "zone-c"), Ready: true},
&BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b", "zone-c"), Ready: true}, &BaseEndpointInfo{Endpoint: "10.1.2.4:80", ZoneHints: sets.NewString("zone-b", "zone-c"), Ready: true},
@ -296,7 +296,7 @@ func TestCategorizeEndpoints(t *testing.T) {
name: "conflicting topology and localness require merging allEndpoints", name: "conflicting topology and localness require merging allEndpoints",
hintsEnabled: true, hintsEnabled: true,
nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"}, nodeLabels: map[string]string{v1.LabelTopologyZone: "zone-a"},
serviceInfo: &BaseServiceInfo{internalPolicyLocal: false, externalPolicyLocal: true, nodePort: 8080, hintsAnnotation: "auto"}, serviceInfo: &BaseServicePortInfo{internalPolicyLocal: false, externalPolicyLocal: true, nodePort: 8080, hintsAnnotation: "auto"},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.0.0.0:80", ZoneHints: sets.NewString("zone-a"), Ready: true, IsLocal: true}, &BaseEndpointInfo{Endpoint: "10.0.0.0:80", ZoneHints: sets.NewString("zone-a"), Ready: true, IsLocal: true},
&BaseEndpointInfo{Endpoint: "10.0.0.1:80", ZoneHints: sets.NewString("zone-b"), Ready: true, IsLocal: true}, &BaseEndpointInfo{Endpoint: "10.0.0.1:80", ZoneHints: sets.NewString("zone-b"), Ready: true, IsLocal: true},
@ -308,13 +308,13 @@ func TestCategorizeEndpoints(t *testing.T) {
allEndpoints: sets.NewString("10.0.0.0:80", "10.0.0.1:80", "10.0.0.2:80"), allEndpoints: sets.NewString("10.0.0.0:80", "10.0.0.1:80", "10.0.0.2:80"),
}, { }, {
name: "internalTrafficPolicy: Local, with empty endpoints", name: "internalTrafficPolicy: Local, with empty endpoints",
serviceInfo: &BaseServiceInfo{internalPolicyLocal: true}, serviceInfo: &BaseServicePortInfo{internalPolicyLocal: true},
endpoints: []Endpoint{}, endpoints: []Endpoint{},
clusterEndpoints: nil, clusterEndpoints: nil,
localEndpoints: sets.NewString(), localEndpoints: sets.NewString(),
}, { }, {
name: "internalTrafficPolicy: Local, but all endpoints are remote", name: "internalTrafficPolicy: Local, but all endpoints are remote",
serviceInfo: &BaseServiceInfo{internalPolicyLocal: true}, serviceInfo: &BaseServicePortInfo{internalPolicyLocal: true},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: true, IsLocal: false}, &BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: true, IsLocal: false},
&BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: true, IsLocal: false}, &BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: true, IsLocal: false},
@ -324,7 +324,7 @@ func TestCategorizeEndpoints(t *testing.T) {
onlyRemoteEndpoints: true, onlyRemoteEndpoints: true,
}, { }, {
name: "internalTrafficPolicy: Local, all endpoints are local", name: "internalTrafficPolicy: Local, all endpoints are local",
serviceInfo: &BaseServiceInfo{internalPolicyLocal: true}, serviceInfo: &BaseServicePortInfo{internalPolicyLocal: true},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: true, IsLocal: true}, &BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: true, IsLocal: true},
&BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: true, IsLocal: true}, &BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: true, IsLocal: true},
@ -333,7 +333,7 @@ func TestCategorizeEndpoints(t *testing.T) {
localEndpoints: sets.NewString("10.0.0.0:80", "10.0.0.1:80"), localEndpoints: sets.NewString("10.0.0.0:80", "10.0.0.1:80"),
}, { }, {
name: "internalTrafficPolicy: Local, some endpoints are local", name: "internalTrafficPolicy: Local, some endpoints are local",
serviceInfo: &BaseServiceInfo{internalPolicyLocal: true}, serviceInfo: &BaseServicePortInfo{internalPolicyLocal: true},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: true, IsLocal: true}, &BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: true, IsLocal: true},
&BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: true, IsLocal: false}, &BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: true, IsLocal: false},
@ -342,7 +342,7 @@ func TestCategorizeEndpoints(t *testing.T) {
localEndpoints: sets.NewString("10.0.0.0:80"), localEndpoints: sets.NewString("10.0.0.0:80"),
}, { }, {
name: "Cluster traffic policy, endpoints not Ready", name: "Cluster traffic policy, endpoints not Ready",
serviceInfo: &BaseServiceInfo{}, serviceInfo: &BaseServicePortInfo{},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: false}, &BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: false},
&BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: false}, &BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: false},
@ -351,7 +351,7 @@ func TestCategorizeEndpoints(t *testing.T) {
localEndpoints: nil, localEndpoints: nil,
}, { }, {
name: "Cluster traffic policy, some endpoints are Ready", name: "Cluster traffic policy, some endpoints are Ready",
serviceInfo: &BaseServiceInfo{}, serviceInfo: &BaseServicePortInfo{},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: false}, &BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: false},
&BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: true}, &BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: true},
@ -361,7 +361,7 @@ func TestCategorizeEndpoints(t *testing.T) {
}, { }, {
name: "Cluster traffic policy, PTE enabled, all endpoints are terminating", name: "Cluster traffic policy, PTE enabled, all endpoints are terminating",
pteEnabled: true, pteEnabled: true,
serviceInfo: &BaseServiceInfo{}, serviceInfo: &BaseServicePortInfo{},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: false, Serving: true, Terminating: true, IsLocal: true}, &BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: false, Serving: true, Terminating: true, IsLocal: true},
&BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: false, Serving: true, Terminating: true, IsLocal: false}, &BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: false, Serving: true, Terminating: true, IsLocal: false},
@ -371,7 +371,7 @@ func TestCategorizeEndpoints(t *testing.T) {
}, { }, {
name: "Cluster traffic policy, PTE disabled, all endpoints are terminating", name: "Cluster traffic policy, PTE disabled, all endpoints are terminating",
pteEnabled: false, pteEnabled: false,
serviceInfo: &BaseServiceInfo{}, serviceInfo: &BaseServicePortInfo{},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: false, Serving: true, Terminating: true, IsLocal: true}, &BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: false, Serving: true, Terminating: true, IsLocal: true},
&BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: false, Serving: true, Terminating: true, IsLocal: false}, &BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: false, Serving: true, Terminating: true, IsLocal: false},
@ -380,7 +380,7 @@ func TestCategorizeEndpoints(t *testing.T) {
localEndpoints: nil, localEndpoints: nil,
}, { }, {
name: "iTP: Local, eTP: Cluster, some endpoints local", name: "iTP: Local, eTP: Cluster, some endpoints local",
serviceInfo: &BaseServiceInfo{internalPolicyLocal: true, externalPolicyLocal: false, nodePort: 8080}, serviceInfo: &BaseServicePortInfo{internalPolicyLocal: true, externalPolicyLocal: false, nodePort: 8080},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: true, IsLocal: true}, &BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: true, IsLocal: true},
&BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: true, IsLocal: false}, &BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: true, IsLocal: false},
@ -390,7 +390,7 @@ func TestCategorizeEndpoints(t *testing.T) {
allEndpoints: sets.NewString("10.0.0.0:80", "10.0.0.1:80"), allEndpoints: sets.NewString("10.0.0.0:80", "10.0.0.1:80"),
}, { }, {
name: "iTP: Cluster, eTP: Local, some endpoints local", name: "iTP: Cluster, eTP: Local, some endpoints local",
serviceInfo: &BaseServiceInfo{internalPolicyLocal: false, externalPolicyLocal: true, nodePort: 8080}, serviceInfo: &BaseServicePortInfo{internalPolicyLocal: false, externalPolicyLocal: true, nodePort: 8080},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: true, IsLocal: true}, &BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: true, IsLocal: true},
&BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: true, IsLocal: false}, &BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: true, IsLocal: false},
@ -400,7 +400,7 @@ func TestCategorizeEndpoints(t *testing.T) {
allEndpoints: sets.NewString("10.0.0.0:80", "10.0.0.1:80"), allEndpoints: sets.NewString("10.0.0.0:80", "10.0.0.1:80"),
}, { }, {
name: "iTP: Local, eTP: Local, some endpoints local", name: "iTP: Local, eTP: Local, some endpoints local",
serviceInfo: &BaseServiceInfo{internalPolicyLocal: true, externalPolicyLocal: true, nodePort: 8080}, serviceInfo: &BaseServicePortInfo{internalPolicyLocal: true, externalPolicyLocal: true, nodePort: 8080},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: true, IsLocal: true}, &BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: true, IsLocal: true},
&BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: true, IsLocal: false}, &BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: true, IsLocal: false},
@ -410,7 +410,7 @@ func TestCategorizeEndpoints(t *testing.T) {
allEndpoints: sets.NewString("10.0.0.0:80", "10.0.0.1:80"), allEndpoints: sets.NewString("10.0.0.0:80", "10.0.0.1:80"),
}, { }, {
name: "iTP: Local, eTP: Local, all endpoints remote", name: "iTP: Local, eTP: Local, all endpoints remote",
serviceInfo: &BaseServiceInfo{internalPolicyLocal: true, externalPolicyLocal: true, nodePort: 8080}, serviceInfo: &BaseServicePortInfo{internalPolicyLocal: true, externalPolicyLocal: true, nodePort: 8080},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: true, IsLocal: false}, &BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: true, IsLocal: false},
&BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: true, IsLocal: false}, &BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: true, IsLocal: false},
@ -420,7 +420,7 @@ func TestCategorizeEndpoints(t *testing.T) {
allEndpoints: sets.NewString("10.0.0.0:80", "10.0.0.1:80"), allEndpoints: sets.NewString("10.0.0.0:80", "10.0.0.1:80"),
}, { }, {
name: "iTP: Local, eTP: Local, PTE disabled, all endpoints remote and terminating", name: "iTP: Local, eTP: Local, PTE disabled, all endpoints remote and terminating",
serviceInfo: &BaseServiceInfo{internalPolicyLocal: true, externalPolicyLocal: true, nodePort: 8080}, serviceInfo: &BaseServicePortInfo{internalPolicyLocal: true, externalPolicyLocal: true, nodePort: 8080},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: false, Serving: true, Terminating: true, IsLocal: false}, &BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: false, Serving: true, Terminating: true, IsLocal: false},
&BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: false, Serving: true, Terminating: true, IsLocal: false}, &BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: false, Serving: true, Terminating: true, IsLocal: false},
@ -431,7 +431,7 @@ func TestCategorizeEndpoints(t *testing.T) {
}, { }, {
name: "iTP: Local, eTP: Local, PTE enabled, all endpoints remote and terminating", name: "iTP: Local, eTP: Local, PTE enabled, all endpoints remote and terminating",
pteEnabled: true, pteEnabled: true,
serviceInfo: &BaseServiceInfo{internalPolicyLocal: true, externalPolicyLocal: true, nodePort: 8080}, serviceInfo: &BaseServicePortInfo{internalPolicyLocal: true, externalPolicyLocal: true, nodePort: 8080},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: false, Serving: true, Terminating: true, IsLocal: false}, &BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: false, Serving: true, Terminating: true, IsLocal: false},
&BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: false, Serving: true, Terminating: true, IsLocal: false}, &BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: false, Serving: true, Terminating: true, IsLocal: false},
@ -442,7 +442,7 @@ func TestCategorizeEndpoints(t *testing.T) {
onlyRemoteEndpoints: true, onlyRemoteEndpoints: true,
}, { }, {
name: "iTP: Cluster, eTP: Local, PTE disabled, with terminating endpoints", name: "iTP: Cluster, eTP: Local, PTE disabled, with terminating endpoints",
serviceInfo: &BaseServiceInfo{internalPolicyLocal: false, externalPolicyLocal: true, nodePort: 8080}, serviceInfo: &BaseServicePortInfo{internalPolicyLocal: false, externalPolicyLocal: true, nodePort: 8080},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: true, IsLocal: false}, &BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: true, IsLocal: false},
&BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: false, Serving: false, IsLocal: true}, &BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: false, Serving: false, IsLocal: true},
@ -455,7 +455,7 @@ func TestCategorizeEndpoints(t *testing.T) {
}, { }, {
name: "iTP: Cluster, eTP: Local, PTE enabled, with terminating endpoints", name: "iTP: Cluster, eTP: Local, PTE enabled, with terminating endpoints",
pteEnabled: true, pteEnabled: true,
serviceInfo: &BaseServiceInfo{internalPolicyLocal: false, externalPolicyLocal: true, nodePort: 8080}, serviceInfo: &BaseServicePortInfo{internalPolicyLocal: false, externalPolicyLocal: true, nodePort: 8080},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: true, IsLocal: false}, &BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: true, IsLocal: false},
&BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: false, Serving: false, IsLocal: true}, &BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: false, Serving: false, IsLocal: true},
@ -467,7 +467,7 @@ func TestCategorizeEndpoints(t *testing.T) {
allEndpoints: sets.NewString("10.0.0.0:80", "10.0.0.2:80"), allEndpoints: sets.NewString("10.0.0.0:80", "10.0.0.2:80"),
}, { }, {
name: "externalTrafficPolicy ignored if not externally accessible", name: "externalTrafficPolicy ignored if not externally accessible",
serviceInfo: &BaseServiceInfo{externalPolicyLocal: true}, serviceInfo: &BaseServicePortInfo{externalPolicyLocal: true},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: true, IsLocal: false}, &BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: true, IsLocal: false},
&BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: true, IsLocal: true}, &BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: true, IsLocal: true},
@ -477,7 +477,7 @@ func TestCategorizeEndpoints(t *testing.T) {
allEndpoints: sets.NewString("10.0.0.0:80", "10.0.0.1:80"), allEndpoints: sets.NewString("10.0.0.0:80", "10.0.0.1:80"),
}, { }, {
name: "no cluster endpoints for iTP:Local internal-only service", name: "no cluster endpoints for iTP:Local internal-only service",
serviceInfo: &BaseServiceInfo{internalPolicyLocal: true}, serviceInfo: &BaseServicePortInfo{internalPolicyLocal: true},
endpoints: []Endpoint{ endpoints: []Endpoint{
&BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: true, IsLocal: false}, &BaseEndpointInfo{Endpoint: "10.0.0.0:80", Ready: true, IsLocal: false},
&BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: true, IsLocal: true}, &BaseEndpointInfo{Endpoint: "10.0.0.1:80", Ready: true, IsLocal: true},

24
pkg/proxy/winkernel/proxier.go
View File

@ -116,7 +116,7 @@ type loadBalancerFlags struct {
// internal struct for string service information // internal struct for string service information
type serviceInfo struct { type serviceInfo struct {
*proxy.BaseServiceInfo *proxy.BaseServicePortInfo
targetPort int targetPort int
externalIPs []*externalIPInfo externalIPs []*externalIPInfo
loadBalancerIngressIPs []*loadBalancerIngressInfo loadBalancerIngressIPs []*loadBalancerIngressInfo
@ -323,7 +323,7 @@ func (proxier *Proxier) endpointsMapChange(oldEndpointsMap, newEndpointsMap prox
func (proxier *Proxier) onEndpointsMapChange(svcPortName *proxy.ServicePortName) { func (proxier *Proxier) onEndpointsMapChange(svcPortName *proxy.ServicePortName) {
svc, exists := proxier.serviceMap[*svcPortName] svc, exists := proxier.svcPortMap[*svcPortName]
if exists { if exists {
svcInfo, ok := svc.(*serviceInfo) svcInfo, ok := svc.(*serviceInfo)
@ -355,7 +355,7 @@ func (proxier *Proxier) onEndpointsMapChange(svcPortName *proxy.ServicePortName)
} }
} }
func (proxier *Proxier) serviceMapChange(previous, current proxy.ServiceMap) { func (proxier *Proxier) serviceMapChange(previous, current proxy.ServicePortMap) {
for svcPortName := range current { for svcPortName := range current {
proxier.onServiceMapChange(&svcPortName) proxier.onServiceMapChange(&svcPortName)
} }
@ -370,7 +370,7 @@ func (proxier *Proxier) serviceMapChange(previous, current proxy.ServiceMap) {
func (proxier *Proxier) onServiceMapChange(svcPortName *proxy.ServicePortName) { func (proxier *Proxier) onServiceMapChange(svcPortName *proxy.ServicePortName) {
svc, exists := proxier.serviceMap[*svcPortName] svc, exists := proxier.svcPortMap[*svcPortName]
if exists { if exists {
svcInfo, ok := svc.(*serviceInfo) svcInfo, ok := svc.(*serviceInfo)
@ -458,8 +458,8 @@ func (refCountMap endPointsReferenceCountMap) getRefCount(hnsID string) *uint16
} }
// returns a new proxy.ServicePort which abstracts a serviceInfo // returns a new proxy.ServicePort which abstracts a serviceInfo
func (proxier *Proxier) newServiceInfo(port *v1.ServicePort, service *v1.Service, baseInfo *proxy.BaseServiceInfo) proxy.ServicePort { func (proxier *Proxier) newServiceInfo(port *v1.ServicePort, service *v1.Service, bsvcPortInfo *proxy.BaseServicePortInfo) proxy.ServicePort {
info := &serviceInfo{BaseServiceInfo: baseInfo} info := &serviceInfo{BaseServicePortInfo: bsvcPortInfo}
preserveDIP := service.Annotations["preserve-destination"] == "true" preserveDIP := service.Annotations["preserve-destination"] == "true"
localTrafficDSR := service.Spec.ExternalTrafficPolicy == v1.ServiceExternalTrafficPolicyTypeLocal localTrafficDSR := service.Spec.ExternalTrafficPolicy == v1.ServiceExternalTrafficPolicyTypeLocal
err := hcn.DSRSupported() err := hcn.DSRSupported()
@ -525,7 +525,7 @@ type Proxier struct {
serviceChanges *proxy.ServiceChangeTracker serviceChanges *proxy.ServiceChangeTracker
endPointsRefCount endPointsReferenceCountMap endPointsRefCount endPointsReferenceCountMap
mu sync.Mutex // protects the following fields mu sync.Mutex // protects the following fields
serviceMap proxy.ServiceMap svcPortMap proxy.ServicePortMap
endpointsMap proxy.EndpointsMap endpointsMap proxy.EndpointsMap
// endpointSlicesSynced and servicesSynced are set to true when corresponding // endpointSlicesSynced and servicesSynced are set to true when corresponding
// objects are synced after startup. This is used to avoid updating hns policies // objects are synced after startup. This is used to avoid updating hns policies
@ -699,7 +699,7 @@ func NewProxier(
isIPv6 := netutils.IsIPv6(nodeIP) isIPv6 := netutils.IsIPv6(nodeIP)
proxier := &Proxier{ proxier := &Proxier{
endPointsRefCount: make(endPointsReferenceCountMap), endPointsRefCount: make(endPointsReferenceCountMap),
serviceMap: make(proxy.ServiceMap), svcPortMap: make(proxy.ServicePortMap),
endpointsMap: make(proxy.EndpointsMap), endpointsMap: make(proxy.EndpointsMap),
masqueradeAll: masqueradeAll, masqueradeAll: masqueradeAll,
masqueradeMark: masqueradeMark, masqueradeMark: masqueradeMark,
@ -986,7 +986,7 @@ func (proxier *Proxier) OnEndpointSlicesSynced() {
} }
func (proxier *Proxier) cleanupAllPolicies() { func (proxier *Proxier) cleanupAllPolicies() {
for svcName, svc := range proxier.serviceMap { for svcName, svc := range proxier.svcPortMap {
svcInfo, ok := svc.(*serviceInfo) svcInfo, ok := svc.(*serviceInfo)
if !ok { if !ok {
klog.ErrorS(nil, "Failed to cast serviceInfo", "serviceName", svcName) klog.ErrorS(nil, "Failed to cast serviceInfo", "serviceName", svcName)
@ -1050,13 +1050,13 @@ func (proxier *Proxier) syncProxyRules() {
// We assume that if this was called, we really want to sync them, // We assume that if this was called, we really want to sync them,
// even if nothing changed in the meantime. In other words, callers are // even if nothing changed in the meantime. In other words, callers are
// responsible for detecting no-op changes and not calling this function. // responsible for detecting no-op changes and not calling this function.
serviceUpdateResult := proxier.serviceMap.Update(proxier.serviceChanges) serviceUpdateResult := proxier.svcPortMap.Update(proxier.serviceChanges)
endpointUpdateResult := proxier.endpointsMap.Update(proxier.endpointsChanges) endpointUpdateResult := proxier.endpointsMap.Update(proxier.endpointsChanges)
staleServices := serviceUpdateResult.UDPStaleClusterIP staleServices := serviceUpdateResult.UDPStaleClusterIP
// merge stale services gathered from updateEndpointsMap // merge stale services gathered from updateEndpointsMap
for _, svcPortName := range endpointUpdateResult.StaleServiceNames { for _, svcPortName := range endpointUpdateResult.StaleServiceNames {
if svcInfo, ok := proxier.serviceMap[svcPortName]; ok && svcInfo != nil && svcInfo.Protocol() == v1.ProtocolUDP { if svcInfo, ok := proxier.svcPortMap[svcPortName]; ok && svcInfo != nil && svcInfo.Protocol() == v1.ProtocolUDP {
klog.V(2).InfoS("Stale udp service", "servicePortName", svcPortName, "clusterIP", svcInfo.ClusterIP()) klog.V(2).InfoS("Stale udp service", "servicePortName", svcPortName, "clusterIP", svcInfo.ClusterIP())
staleServices.Insert(svcInfo.ClusterIP().String()) staleServices.Insert(svcInfo.ClusterIP().String())
} }
@ -1093,7 +1093,7 @@ func (proxier *Proxier) syncProxyRules() {
klog.V(3).InfoS("Syncing Policies") klog.V(3).InfoS("Syncing Policies")
// Program HNS by adding corresponding policies for each service. // Program HNS by adding corresponding policies for each service.
for svcName, svc := range proxier.serviceMap { for svcName, svc := range proxier.svcPortMap {
svcInfo, ok := svc.(*serviceInfo) svcInfo, ok := svc.(*serviceInfo)
if !ok { if !ok {
klog.ErrorS(nil, "Failed to cast serviceInfo", "serviceName", svcName) klog.ErrorS(nil, "Failed to cast serviceInfo", "serviceName", svcName)

10
pkg/proxy/winkernel/proxier_test.go
View File

@ -138,7 +138,7 @@ func NewFakeProxier(syncPeriod time.Duration, minSyncPeriod time.Duration, clust
networkType: networkType, networkType: networkType,
} }
proxier := &Proxier{ proxier := &Proxier{
serviceMap: make(proxy.ServiceMap), svcPortMap: make(proxy.ServicePortMap),
endpointsMap: make(proxy.EndpointsMap), endpointsMap: make(proxy.EndpointsMap),
clusterCIDR: clusterCIDR, clusterCIDR: clusterCIDR,
hostname: testHostName, hostname: testHostName,
@ -201,7 +201,7 @@ func TestCreateServiceVip(t *testing.T) {
proxier.setInitialized(true) proxier.setInitialized(true)
proxier.syncProxyRules() proxier.syncProxyRules()
svc := proxier.serviceMap[svcPortName] svc := proxier.svcPortMap[svcPortName]
svcInfo, ok := svc.(*serviceInfo) svcInfo, ok := svc.(*serviceInfo)
if !ok { if !ok {
t.Errorf("Failed to cast serviceInfo %q", svcPortName.String()) t.Errorf("Failed to cast serviceInfo %q", svcPortName.String())
@ -707,7 +707,7 @@ func TestCreateLoadBalancer(t *testing.T) {
proxier.setInitialized(true) proxier.setInitialized(true)
proxier.syncProxyRules() proxier.syncProxyRules()
svc := proxier.serviceMap[svcPortName] svc := proxier.svcPortMap[svcPortName]
svcInfo, ok := svc.(*serviceInfo) svcInfo, ok := svc.(*serviceInfo)
if !ok { if !ok {
t.Errorf("Failed to cast serviceInfo %q", svcPortName.String()) t.Errorf("Failed to cast serviceInfo %q", svcPortName.String())
@ -770,7 +770,7 @@ func TestCreateDsrLoadBalancer(t *testing.T) {
proxier.setInitialized(true) proxier.setInitialized(true)
proxier.syncProxyRules() proxier.syncProxyRules()
svc := proxier.serviceMap[svcPortName] svc := proxier.svcPortMap[svcPortName]
svcInfo, ok := svc.(*serviceInfo) svcInfo, ok := svc.(*serviceInfo)
if !ok { if !ok {
t.Errorf("Failed to cast serviceInfo %q", svcPortName.String()) t.Errorf("Failed to cast serviceInfo %q", svcPortName.String())
@ -840,7 +840,7 @@ func TestEndpointSlice(t *testing.T) {
proxier.setInitialized(true) proxier.setInitialized(true)
proxier.syncProxyRules() proxier.syncProxyRules()
svc := proxier.serviceMap[svcPortName] svc := proxier.svcPortMap[svcPortName]
svcInfo, ok := svc.(*serviceInfo) svcInfo, ok := svc.(*serviceInfo)
if !ok { if !ok {
t.Errorf("Failed to cast serviceInfo %q", svcPortName.String()) t.Errorf("Failed to cast serviceInfo %q", svcPortName.String())