Skip to content

K
k3s
Commit f6eed81f authored by Zihong Zheng's avatar Zihong Zheng
Browse files
Options

[kube-proxy] Mass service/endpoint info functions rename and comments

parent 06064498
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 110 additions and 115 deletions
pkg/proxy/endpoints.go
View file @ f6eed81f
......@@ -32,48 +32,51 @@ import (
utilnet "k8s.io/kubernetes/pkg/util/net"
)
// EndpointInfoCommon contains common endpoint information.
type EndpointInfoCommon struct {
// BaseEndpointInfo contains base information that defines an endpoint.
// This could be used directly by proxier while processing endpoints,
// or can be used for constructing a more specific EndpointInfo struct
// defined by the proxier if needed.
type BaseEndpointInfo struct {
Endpoint string // TODO: should be an endpointString type
// IsLocal indicates whether the endpoint is running in same host as kube-proxy.
IsLocal bool
}
var _ Endpoint = &EndpointInfoCommon{}
var _ Endpoint = &BaseEndpointInfo{}
// String is part of proxy.Endpoint interface.
func (info *EndpointInfoCommon) String() string {
func (info *BaseEndpointInfo) String() string {
return info.Endpoint
}
// IsLocal is part of proxy.Endpoint interface.
func (info *EndpointInfoCommon) GetIsLocal() bool {
// GetIsLocal is part of proxy.Endpoint interface.
func (info *BaseEndpointInfo) GetIsLocal() bool {
return info.IsLocal
}
// IP returns just the IP part of the endpoint, it's a part of proxy.Endpoint interface.
func (info *EndpointInfoCommon) IP() string {
func (info *BaseEndpointInfo) IP() string {
return utilproxy.IPPart(info.Endpoint)
}
// Port returns just the Port part of the endpoint.
func (info *EndpointInfoCommon) Port() (int, error) {
func (info *BaseEndpointInfo) Port() (int, error) {
return utilproxy.PortPart(info.Endpoint)
}
// Equal is part of proxy.Endpoint interface.
func (info *EndpointInfoCommon) Equal(other Endpoint) bool {
func (info *BaseEndpointInfo) Equal(other Endpoint) bool {
return info.String() == other.String() && info.GetIsLocal() == other.GetIsLocal()
}
func newEndpointInfoCommon(IP string, port int, isLocal bool) *EndpointInfoCommon {
return &EndpointInfoCommon{
func newBaseEndpointInfo(IP string, port int, isLocal bool) *BaseEndpointInfo {
return &BaseEndpointInfo{
Endpoint: net.JoinHostPort(IP, strconv.Itoa(port)),
IsLocal: isLocal,
}
}
type customizeEndpointInfoFunc func(IP string, port int, isLocal bool, info *EndpointInfoCommon) Endpoint
type makeEndpointFunc func(info *BaseEndpointInfo) Endpoint
// EndpointChangeTracker carries state about uncommitted changes to an arbitrary number of
// Endpoints, keyed by their namespace and name.
......@@ -84,21 +87,21 @@ type EndpointChangeTracker struct {
hostname string
// items maps a service to is endpointsChange.
items map[types.NamespacedName]*endpointsChange
// customizeEndpointInfo allows proxier to inject customized infomation when processing endpoint.
customizeEndpointInfo customizeEndpointInfoFunc
// makeEndpointInfo allows proxier to inject customized information when processing endpoint.
makeEndpointInfo makeEndpointFunc
// isIPv6Mode indicates if change tracker is under IPv6/IPv4 mode. Nil means not applicable.
isIPv6Mode *bool
recorder record.EventRecorder
}
// NewEndpointChangeTracker initializes an EndpointsChangeMap
func NewEndpointChangeTracker(hostname string, customizeEndpointInfo customizeEndpointInfoFunc, isIPv6Mode *bool, recorder record.EventRecorder) *EndpointChangeTracker {
func NewEndpointChangeTracker(hostname string, makeEndpointInfo makeEndpointFunc, isIPv6Mode *bool, recorder record.EventRecorder) *EndpointChangeTracker {
return &EndpointChangeTracker{
hostname: hostname,
items: make(map[types.NamespacedName]*endpointsChange),
customizeEndpointInfo: customizeEndpointInfo,
isIPv6Mode: isIPv6Mode,
recorder: recorder,
hostname: hostname,
items: make(map[types.NamespacedName]*endpointsChange),
makeEndpointInfo: makeEndpointInfo,
isIPv6Mode: isIPv6Mode,
recorder: recorder,
}
}
......@@ -174,7 +177,7 @@ func UpdateEndpointsMap(endpointsMap EndpointsMap, changes *EndpointChangeTracke
return result
}
// EndpointsMap maps a service to one of its Endpoint.
// EndpointsMap maps a service name to a list of all its Endpoints.
type EndpointsMap map[ServicePortName][]Endpoint
// endpointsToEndpointsMap translates single Endpoints object to EndpointsMap.
......@@ -208,6 +211,7 @@ func (ect *EndpointChangeTracker) endpointsToEndpointsMap(endpoints *api.Endpoin
continue
}
// Filter out the incorrect IP version case.
// Any endpoint port that contains incorrect IP version will be ignored.
if ect.isIPv6Mode != nil && utilnet.IsIPv6String(addr.IP) != *ect.isIPv6Mode {
// Emit event on the corresponding service which had a different
// IP version than the endpoint.
......@@ -215,11 +219,11 @@ func (ect *EndpointChangeTracker) endpointsToEndpointsMap(endpoints *api.Endpoin
continue
}
isLocal := addr.NodeName != nil && *addr.NodeName == ect.hostname
epInfoCommon := newEndpointInfoCommon(addr.IP, int(port.Port), isLocal)
if ect.customizeEndpointInfo != nil {
endpointsMap[svcPortName] = append(endpointsMap[svcPortName], ect.customizeEndpointInfo(addr.IP, int(port.Port), isLocal, epInfoCommon))
baseEndpointInfo := newBaseEndpointInfo(addr.IP, int(port.Port), isLocal)
if ect.makeEndpointInfo != nil {
endpointsMap[svcPortName] = append(endpointsMap[svcPortName], ect.makeEndpointInfo(baseEndpointInfo))
} else {
endpointsMap[svcPortName] = append(endpointsMap[svcPortName], epInfoCommon)
endpointsMap[svcPortName] = append(endpointsMap[svcPortName], baseEndpointInfo)
}
}
if glog.V(3) {
......
pkg/proxy/iptables/proxier.go
View file @ f6eed81f
......@@ -140,7 +140,7 @@ const sysctlBridgeCallIPTables = "net/bridge/bridge-nf-call-iptables"
// internal struct for string service information
type serviceInfo struct {
*proxy.ServiceInfoCommon
*proxy.BaseServiceInfo
// The following fields are computed and stored for performance reasons.
serviceNameString string
servicePortChainName utiliptables.Chain
......@@ -149,8 +149,8 @@ type serviceInfo struct {
}
// returns a new proxy.ServicePort which abstracts a serviceInfo
func customizeServiceInfo(port *api.ServicePort, service *api.Service, infoCommon *proxy.ServiceInfoCommon) proxy.ServicePort {
info := &serviceInfo{ServiceInfoCommon: infoCommon}
func newServiceInfo(port *api.ServicePort, service *api.Service, baseInfo *proxy.BaseServiceInfo) proxy.ServicePort {
info := &serviceInfo{BaseServiceInfo: baseInfo}
// Store the following for performance reasons.
svcName := types.NamespacedName{Namespace: service.Namespace, Name: service.Name}
......@@ -166,7 +166,7 @@ func customizeServiceInfo(port *api.ServicePort, service *api.Service, infoCommo
// internal struct for endpoints information
type endpointsInfo struct {
*proxy.EndpointInfoCommon
*proxy.BaseEndpointInfo
// The following fields we lazily compute and store here for performance
// reasons. If the protocol is the same as you expect it to be, then the
// chainName can be reused, otherwise it should be recomputed.
......@@ -175,11 +175,11 @@ type endpointsInfo struct {
}
// returns a new proxy.Endpoint which abstracts a endpointsInfo
func customizeEndpointInfo(IP string, port int, isLocal bool, infoCommon *proxy.EndpointInfoCommon) proxy.Endpoint {
return &endpointsInfo{EndpointInfoCommon: infoCommon}
func newEndpointInfo(baseInfo *proxy.BaseEndpointInfo) proxy.Endpoint {
return &endpointsInfo{BaseEndpointInfo: baseInfo}
}
// Equal overrides the Equal() function imlemented by proxy.EndpointInfoCommon.
// Equal overrides the Equal() function imlemented by proxy.BaseEndpointInfo.
func (e *endpointsInfo) Equal(other proxy.Endpoint) bool {
o, ok := other.(*endpointsInfo)
if !ok {
......@@ -319,9 +319,9 @@ func NewProxier(ipt utiliptables.Interface,
proxier := &Proxier{
portsMap: make(map[utilproxy.LocalPort]utilproxy.Closeable),
serviceMap: make(proxy.ServiceMap),
serviceChanges: proxy.NewServiceChangeTracker(customizeServiceInfo, &isIPv6, recorder),
serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, &isIPv6, recorder),
endpointsMap: make(proxy.EndpointsMap),
endpointsChanges: proxy.NewEndpointChangeTracker(hostname, customizeEndpointInfo, &isIPv6, recorder),
endpointsChanges: proxy.NewEndpointChangeTracker(hostname, newEndpointInfo, &isIPv6, recorder),
iptables: ipt,
masqueradeAll: masqueradeAll,
masqueradeMark: masqueradeMark,
......@@ -502,9 +502,7 @@ func (proxier *Proxier) isInitialized() bool {
}
func (proxier *Proxier) OnServiceAdd(service *api.Service) {
if proxier.serviceChanges.Update(nil, service) && proxier.isInitialized() {
proxier.syncRunner.Run()
}
proxier.OnServiceUpdate(nil, service)
}
func (proxier *Proxier) OnServiceUpdate(oldService, service *api.Service) {
......@@ -514,9 +512,8 @@ func (proxier *Proxier) OnServiceUpdate(oldService, service *api.Service) {
}
func (proxier *Proxier) OnServiceDelete(service *api.Service) {
if proxier.serviceChanges.Update(service, nil) && proxier.isInitialized() {
proxier.syncRunner.Run()
}
proxier.OnServiceUpdate(service, nil)
}
func (proxier *Proxier) OnServiceSynced() {
......@@ -530,9 +527,7 @@ func (proxier *Proxier) OnServiceSynced() {
}
func (proxier *Proxier) OnEndpointsAdd(endpoints *api.Endpoints) {
if proxier.endpointsChanges.Update(nil, endpoints) && proxier.isInitialized() {
proxier.syncRunner.Run()
}
proxier.OnEndpointsUpdate(nil, endpoints)
}
func (proxier *Proxier) OnEndpointsUpdate(oldEndpoints, endpoints *api.Endpoints) {
......@@ -542,9 +537,7 @@ func (proxier *Proxier) OnEndpointsUpdate(oldEndpoints, endpoints *api.Endpoints
}
func (proxier *Proxier) OnEndpointsDelete(endpoints *api.Endpoints) {
if proxier.endpointsChanges.Update(endpoints, nil) && proxier.isInitialized() {
proxier.syncRunner.Run()
}
proxier.OnEndpointsUpdate(endpoints, nil)
}
func (proxier *Proxier) OnEndpointsSynced() {
......@@ -605,7 +598,7 @@ func (proxier *Proxier) deleteEndpointConnections(connectionMap []proxy.ServiceE
for _, epSvcPair := range connectionMap {
if svcInfo, ok := proxier.serviceMap[epSvcPair.ServicePortName]; ok && svcInfo.GetProtocol() == api.ProtocolUDP {
endpointIP := utilproxy.IPPart(epSvcPair.Endpoint)
err := conntrack.ClearEntriesForNAT(proxier.exec, svcInfo.GetClusterIP(), endpointIP, v1.ProtocolUDP)
err := conntrack.ClearEntriesForNAT(proxier.exec, svcInfo.ClusterIPString(), endpointIP, v1.ProtocolUDP)
if err != nil {
glog.Errorf("Failed to delete %s endpoint connections, error: %v", epSvcPair.ServicePortName.String(), err)
}
......@@ -641,8 +634,8 @@ func (proxier *Proxier) syncProxyRules() {
// merge stale services gathered from updateEndpointsMap
for _, svcPortName := range endpointUpdateResult.StaleServiceNames {
if svcInfo, ok := proxier.serviceMap[svcPortName]; ok && svcInfo != nil && svcInfo.GetProtocol() == api.ProtocolUDP {
glog.V(2).Infof("Stale udp service %v -> %s", svcPortName, svcInfo.GetClusterIP())
staleServices.Insert(svcInfo.GetClusterIP())
glog.V(2).Infof("Stale udp service %v -> %s", svcPortName, svcInfo.ClusterIPString())
staleServices.Insert(svcInfo.ClusterIPString())
}
}
......
pkg/proxy/ipvs/proxier.go
View file @ f6eed81f
......@@ -310,7 +310,7 @@ func NewProxier(ipt utiliptables.Interface,
proxier := &Proxier{
portsMap: make(map[utilproxy.LocalPort]utilproxy.Closeable),
serviceMap: make(proxy.ServiceMap),
serviceChanges: proxy.NewServiceChangeTracker(customizeServiceInfo, &isIPv6, recorder),
serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, &isIPv6, recorder),
endpointsMap: make(proxy.EndpointsMap),
endpointsChanges: proxy.NewEndpointChangeTracker(hostname, nil, &isIPv6, recorder),
syncPeriod: syncPeriod,
......@@ -354,14 +354,14 @@ func NewProxier(ipt utiliptables.Interface,
// internal struct for string service information
type serviceInfo struct {
*proxy.ServiceInfoCommon
*proxy.BaseServiceInfo
// The following fields are computed and stored for performance reasons.
serviceNameString string
}
// returns a new proxy.ServicePort which abstracts a serviceInfo
func customizeServiceInfo(port *api.ServicePort, service *api.Service, infoCommon *proxy.ServiceInfoCommon) proxy.ServicePort {
info := &serviceInfo{ServiceInfoCommon: infoCommon}
func newServiceInfo(port *api.ServicePort, service *api.Service, baseInfo *proxy.BaseServiceInfo) proxy.ServicePort {
info := &serviceInfo{BaseServiceInfo: baseInfo}
// Store the following for performance reasons.
svcName := types.NamespacedName{Namespace: service.Namespace, Name: service.Name}
......@@ -552,9 +552,7 @@ func (proxier *Proxier) isInitialized() bool {
// OnServiceAdd is called whenever creation of new service object is observed.
func (proxier *Proxier) OnServiceAdd(service *api.Service) {
if proxier.serviceChanges.Update(nil, service) && proxier.isInitialized() {
proxier.syncRunner.Run()
}
proxier.OnServiceUpdate(nil, service)
}
// OnServiceUpdate is called whenever modification of an existing service object is observed.
......@@ -566,9 +564,7 @@ func (proxier *Proxier) OnServiceUpdate(oldService, service *api.Service) {
// OnServiceDelete is called whenever deletion of an existing service object is observed.
func (proxier *Proxier) OnServiceDelete(service *api.Service) {
if proxier.serviceChanges.Update(service, nil) && proxier.isInitialized() {
proxier.syncRunner.Run()
}
proxier.OnServiceUpdate(service, nil)
}
// OnServiceSynced is called once all the initial even handlers were called and the state is fully propagated to local cache.
......@@ -584,9 +580,7 @@ func (proxier *Proxier) OnServiceSynced() {
// OnEndpointsAdd is called whenever creation of new endpoints object is observed.
func (proxier *Proxier) OnEndpointsAdd(endpoints *api.Endpoints) {
if proxier.endpointsChanges.Update(nil, endpoints) && proxier.isInitialized() {
proxier.syncRunner.Run()
}
proxier.OnEndpointsUpdate(nil, endpoints)
}
// OnEndpointsUpdate is called whenever modification of an existing endpoints object is observed.
......@@ -598,9 +592,7 @@ func (proxier *Proxier) OnEndpointsUpdate(oldEndpoints, endpoints *api.Endpoints
// OnEndpointsDelete is called whenever deletion of an existing endpoints object is observed.
func (proxier *Proxier) OnEndpointsDelete(endpoints *api.Endpoints) {
if proxier.endpointsChanges.Update(endpoints, nil) && proxier.isInitialized() {
proxier.syncRunner.Run()
}
proxier.OnEndpointsUpdate(endpoints, nil)
}
// OnEndpointsSynced is called once all the initial event handlers were called and the state is fully propagated to local cache.
......@@ -642,8 +634,8 @@ func (proxier *Proxier) syncProxyRules() {
// merge stale services gathered from updateEndpointsMap
for _, svcPortName := range endpointUpdateResult.StaleServiceNames {
if svcInfo, ok := proxier.serviceMap[svcPortName]; ok && svcInfo != nil && svcInfo.GetProtocol() == api.ProtocolUDP {
glog.V(2).Infof("Stale udp service %v -> %s", svcPortName, svcInfo.GetClusterIP())
staleServices.Insert(svcInfo.GetClusterIP())
glog.V(2).Infof("Stale udp service %v -> %s", svcPortName, svcInfo.ClusterIPString())
staleServices.Insert(svcInfo.ClusterIPString())
}
}
......@@ -759,9 +751,9 @@ func (proxier *Proxier) syncProxyRules() {
// Handle traffic that loops back to the originator with SNAT.
for _, e := range proxier.endpointsMap[svcName] {
ep, ok := e.(*proxy.EndpointInfoCommon)
ep, ok := e.(*proxy.BaseEndpointInfo)
if !ok {
glog.Errorf("Failed to cast EndpointInfoCommon %q", e.String())
glog.Errorf("Failed to cast BaseEndpointInfo %q", e.String())
continue
}
epIP := ep.IP()
......@@ -1269,7 +1261,7 @@ func (proxier *Proxier) deleteEndpointConnections(connectionMap []proxy.ServiceE
for _, epSvcPair := range connectionMap {
if svcInfo, ok := proxier.serviceMap[epSvcPair.ServicePortName]; ok && svcInfo.GetProtocol() == api.ProtocolUDP {
endpointIP := utilproxy.IPPart(epSvcPair.Endpoint)
err := conntrack.ClearEntriesForNAT(proxier.exec, svcInfo.GetClusterIP(), endpointIP, clientv1.ProtocolUDP)
err := conntrack.ClearEntriesForNAT(proxier.exec, svcInfo.ClusterIPString(), endpointIP, clientv1.ProtocolUDP)
if err != nil {
glog.Errorf("Failed to delete %s endpoint connections, error: %v", epSvcPair.ServicePortName.String(), err)
}
......
pkg/proxy/service.go
View file @ f6eed81f
......@@ -35,8 +35,11 @@ import (
utilnet "k8s.io/kubernetes/pkg/util/net"
)
// ServiceInfoCommon contains common service information.
type ServiceInfoCommon struct {
// BaseServiceInfo contains base information that defines a service.
// This could be used directly by proxier while processing services,
// or can be used for constructing a more specific ServiceInfo struct
// defined by the proxier if needed.
type BaseServiceInfo struct {
ClusterIP net.IP
Port int
Protocol api.Protocol
......@@ -50,29 +53,29 @@ type ServiceInfoCommon struct {
OnlyNodeLocalEndpoints bool
}
var _ ServicePort = &ServiceInfoCommon{}
var _ ServicePort = &BaseServiceInfo{}
// String is part of ServicePort interface.
func (info *ServiceInfoCommon) String() string {
func (info *BaseServiceInfo) String() string {
return fmt.Sprintf("%s:%d/%s", info.ClusterIP, info.Port, info.Protocol)
}
// GetClusterIP is part of ServicePort interface.
func (info *ServiceInfoCommon) GetClusterIP() string {
// ClusterIPString is part of ServicePort interface.
func (info *BaseServiceInfo) ClusterIPString() string {
return info.ClusterIP.String()
}
// GetProtocol is part of ServicePort interface.
func (info *ServiceInfoCommon) GetProtocol() api.Protocol {
func (info *BaseServiceInfo) GetProtocol() api.Protocol {
return info.Protocol
}
// GetHealthCheckNodePort is part of ServicePort interface.
func (info *ServiceInfoCommon) GetHealthCheckNodePort() int {
func (info *BaseServiceInfo) GetHealthCheckNodePort() int {
return info.HealthCheckNodePort
}
func (sct *ServiceChangeTracker) newServiceInfoCommon(port *api.ServicePort, service *api.Service) *ServiceInfoCommon {
func (sct *ServiceChangeTracker) newBaseServiceInfo(port *api.ServicePort, service *api.Service) *BaseServiceInfo {
onlyNodeLocalEndpoints := false
if apiservice.RequestsOnlyLocalTraffic(service) {
onlyNodeLocalEndpoints = true
......@@ -82,7 +85,7 @@ func (sct *ServiceChangeTracker) newServiceInfoCommon(port *api.ServicePort, ser
// Kube-apiserver side guarantees SessionAffinityConfig won't be nil when session affinity type is ClientIP
stickyMaxAgeSeconds = int(*service.Spec.SessionAffinityConfig.ClientIP.TimeoutSeconds)
}
info := &ServiceInfoCommon{
info := &BaseServiceInfo{
ClusterIP: net.ParseIP(service.Spec.ClusterIP),
Port: int(port.Port),
Protocol: port.Protocol,
......@@ -101,6 +104,8 @@ func (sct *ServiceChangeTracker) newServiceInfoCommon(port *api.ServicePort, ser
copy(info.ExternalIPs, service.Spec.ExternalIPs)
} else {
// Filter out the incorrect IP version case.
// If ExternalIPs and LoadBalancerSourceRanges on service contains incorrect IP versions,
// only filter out the incorrect ones.
var incorrectIPs []string
info.ExternalIPs, incorrectIPs = utilnet.FilterIncorrectIPVersion(service.Spec.ExternalIPs, *sct.isIPv6Mode)
if len(incorrectIPs) > 0 {
......@@ -124,7 +129,7 @@ func (sct *ServiceChangeTracker) newServiceInfoCommon(port *api.ServicePort, ser
return info
}
type customizeServiceInfoFunc func(*api.ServicePort, *api.Service, *ServiceInfoCommon) ServicePort
type makeServicePortFunc func(*api.ServicePort, *api.Service, *BaseServiceInfo) ServicePort
// serviceChange contains all changes to services that happened since proxy rules were synced. For a single object,
// changes are accumulated, i.e. previous is state from before applying the changes,
......@@ -141,20 +146,20 @@ type ServiceChangeTracker struct {
lock sync.Mutex
// items maps a service to its serviceChange.
items map[types.NamespacedName]*serviceChange
// customizeServiceInfo allows proxier to inject customized infomation when processing service.
customizeServiceInfo customizeServiceInfoFunc
// makeServiceInfo allows proxier to inject customized information when processing service.
makeServiceInfo makeServicePortFunc
// isIPv6Mode indicates if change tracker is under IPv6/IPv4 mode. Nil means not applicable.
isIPv6Mode *bool
recorder record.EventRecorder
}
// NewServiceChangeTracker initializes a ServiceChangeTracker
func NewServiceChangeTracker(customizeServiceInfo customizeServiceInfoFunc, isIPv6Mode *bool, recorder record.EventRecorder) *ServiceChangeTracker {
func NewServiceChangeTracker(makeServiceInfo makeServicePortFunc, isIPv6Mode *bool, recorder record.EventRecorder) *ServiceChangeTracker {
return &ServiceChangeTracker{
items: make(map[types.NamespacedName]*serviceChange),
customizeServiceInfo: customizeServiceInfo,
isIPv6Mode: isIPv6Mode,
recorder: recorder,
items: make(map[types.NamespacedName]*serviceChange),
makeServiceInfo: makeServiceInfo,
isIPv6Mode: isIPv6Mode,
recorder: recorder,
}
}
......@@ -238,6 +243,7 @@ func (sct *ServiceChangeTracker) serviceToServiceMap(service *api.Service) Servi
if len(service.Spec.ClusterIP) != 0 {
// Filter out the incorrect IP version case.
// If ClusterIP on service has incorrect IP version, service itself will be ignored.
if sct.isIPv6Mode != nil && utilnet.IsIPv6String(service.Spec.ClusterIP) != *sct.isIPv6Mode {
utilproxy.LogAndEmitIncorrectIPVersionEvent(sct.recorder, "clusterIP", service.Spec.ClusterIP, service.Namespace, service.Name, service.UID)
return nil
......@@ -248,11 +254,11 @@ func (sct *ServiceChangeTracker) serviceToServiceMap(service *api.Service) Servi
for i := range service.Spec.Ports {
servicePort := &service.Spec.Ports[i]
svcPortName := ServicePortName{NamespacedName: svcName, Port: servicePort.Name}
svcInfoCommon := sct.newServiceInfoCommon(servicePort, service)
if sct.customizeServiceInfo != nil {
serviceMap[svcPortName] = sct.customizeServiceInfo(servicePort, service, svcInfoCommon)
baseSvcInfo := sct.newBaseServiceInfo(servicePort, service)
if sct.makeServiceInfo != nil {
serviceMap[svcPortName] = sct.makeServiceInfo(servicePort, service, baseSvcInfo)
} else {
serviceMap[svcPortName] = svcInfoCommon
serviceMap[svcPortName] = baseSvcInfo
}
}
return serviceMap
......@@ -328,7 +334,7 @@ func (sm *ServiceMap) unmerge(other ServiceMap, UDPStaleClusterIP sets.String) {
if exists {
glog.V(1).Infof("Removing service port %q", svcPortName)
if info.GetProtocol() == api.ProtocolUDP {
UDPStaleClusterIP.Insert(info.GetClusterIP())
UDPStaleClusterIP.Insert(info.ClusterIPString())
}
delete(*sm, svcPortName)
} else {
......
pkg/proxy/service_test.go
View file @ f6eed81f
......@@ -31,8 +31,8 @@ import (
const testHostname = "test-hostname"
func makeTestServiceInfo(clusterIP string, port int, protocol string, healthcheckNodePort int, svcInfoFuncs ...func(*ServiceInfoCommon)) *ServiceInfoCommon {
info := &ServiceInfoCommon{
func makeTestServiceInfo(clusterIP string, port int, protocol string, healthcheckNodePort int, svcInfoFuncs ...func(*BaseServiceInfo)) *BaseServiceInfo {
info := &BaseServiceInfo{
ClusterIP: net.ParseIP(clusterIP),
Port: port,
Protocol: api.Protocol(protocol),
......@@ -97,13 +97,13 @@ func TestServiceToServiceMap(t *testing.T) {
testCases := []struct {
desc string
service *api.Service
expected map[ServicePortName]*ServiceInfoCommon
expected map[ServicePortName]*BaseServiceInfo
isIPv6Mode *bool
}{
{
desc: "nothing",
service: nil,
expected: map[ServicePortName]*ServiceInfoCommon{},
expected: map[ServicePortName]*BaseServiceInfo{},
},
{
desc: "headless service",
......@@ -112,7 +112,7 @@ func TestServiceToServiceMap(t *testing.T) {
svc.Spec.ClusterIP = api.ClusterIPNone
svc.Spec.Ports = addTestPort(svc.Spec.Ports, "rpc", "UDP", 1234, 0, 0)
}),
expected: map[ServicePortName]*ServiceInfoCommon{},
expected: map[ServicePortName]*BaseServiceInfo{},
},
{
desc: "headless service without port",
......@@ -120,7 +120,7 @@ func TestServiceToServiceMap(t *testing.T) {
svc.Spec.Type = api.ServiceTypeClusterIP
svc.Spec.ClusterIP = api.ClusterIPNone
}),
expected: map[ServicePortName]*ServiceInfoCommon{},
expected: map[ServicePortName]*BaseServiceInfo{},
},
{
desc: "cluster ip service",
......@@ -130,7 +130,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, "p2", "UDP", 1235, 5321, 0)
}),
expected: map[ServicePortName]*ServiceInfoCommon{
expected: map[ServicePortName]*BaseServiceInfo{
makeServicePortName("ns2", "cluster-ip", "p1"): makeTestServiceInfo("172.16.55.4", 1234, "UDP", 0),
makeServicePortName("ns2", "cluster-ip", "p2"): makeTestServiceInfo("172.16.55.4", 1235, "UDP", 0),
},
......@@ -143,7 +143,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, "port2", "TCP", 344, 677, 0)
}),
expected: map[ServicePortName]*ServiceInfoCommon{
expected: map[ServicePortName]*BaseServiceInfo{
makeServicePortName("ns2", "node-port", "port1"): makeTestServiceInfo("172.16.55.10", 345, "UDP", 0),
makeServicePortName("ns2", "node-port", "port2"): makeTestServiceInfo("172.16.55.10", 344, "TCP", 0),
},
......@@ -162,7 +162,7 @@ func TestServiceToServiceMap(t *testing.T) {
},
}
}),
expected: map[ServicePortName]*ServiceInfoCommon{
expected: map[ServicePortName]*BaseServiceInfo{
makeServicePortName("ns1", "load-balancer", "port3"): makeTestServiceInfo("172.16.55.11", 8675, "UDP", 0),
makeServicePortName("ns1", "load-balancer", "port4"): makeTestServiceInfo("172.16.55.11", 8676, "UDP", 0),
},
......@@ -183,7 +183,7 @@ func TestServiceToServiceMap(t *testing.T) {
svc.Spec.ExternalTrafficPolicy = api.ServiceExternalTrafficPolicyTypeLocal
svc.Spec.HealthCheckNodePort = 345
}),
expected: map[ServicePortName]*ServiceInfoCommon{
expected: map[ServicePortName]*BaseServiceInfo{
makeServicePortName("ns1", "only-local-load-balancer", "portx"): makeTestServiceInfo("172.16.55.12", 8677, "UDP", 345),
makeServicePortName("ns1", "only-local-load-balancer", "porty"): makeTestServiceInfo("172.16.55.12", 8678, "UDP", 345),
},
......@@ -196,7 +196,7 @@ func TestServiceToServiceMap(t *testing.T) {
svc.Spec.ExternalName = "foo2.bar.com"
svc.Spec.Ports = addTestPort(svc.Spec.Ports, "portz", "UDP", 1235, 5321, 0)
}),
expected: map[ServicePortName]*ServiceInfoCommon{},
expected: map[ServicePortName]*BaseServiceInfo{},
},
{
desc: "service with ipv6 clusterIP under ipv4 mode, service should be filtered",
......@@ -258,8 +258,8 @@ func TestServiceToServiceMap(t *testing.T) {
},
},
},
expected: map[ServicePortName]*ServiceInfoCommon{
makeServicePortName("test", "validIPv4", "testPort"): makeTestServiceInfo(testClusterIPv4, 12345, "TCP", 0, func(info *ServiceInfoCommon) {
expected: map[ServicePortName]*BaseServiceInfo{
makeServicePortName("test", "validIPv4", "testPort"): makeTestServiceInfo(testClusterIPv4, 12345, "TCP", 0, func(info *BaseServiceInfo) {
info.ExternalIPs = []string{testExternalIPv4}
info.LoadBalancerSourceRanges = []string{testSourceRangeIPv4}
}),
......@@ -286,8 +286,8 @@ func TestServiceToServiceMap(t *testing.T) {
},
},
},
expected: map[ServicePortName]*ServiceInfoCommon{
makeServicePortName("test", "validIPv6", "testPort"): makeTestServiceInfo(testClusterIPv6, 12345, "TCP", 0, func(info *ServiceInfoCommon) {
expected: map[ServicePortName]*BaseServiceInfo{
makeServicePortName("test", "validIPv6", "testPort"): makeTestServiceInfo(testClusterIPv6, 12345, "TCP", 0, func(info *BaseServiceInfo) {
info.ExternalIPs = []string{testExternalIPv6}
info.LoadBalancerSourceRanges = []string{testSourceRangeIPv6}
}),
......@@ -314,8 +314,8 @@ func TestServiceToServiceMap(t *testing.T) {
},
},
},
expected: map[ServicePortName]*ServiceInfoCommon{
makeServicePortName("test", "filterIPv6InIPV4Mode", "testPort"): makeTestServiceInfo(testClusterIPv4, 12345, "TCP", 0, func(info *ServiceInfoCommon) {
expected: map[ServicePortName]*BaseServiceInfo{
makeServicePortName("test", "filterIPv6InIPV4Mode", "testPort"): makeTestServiceInfo(testClusterIPv4, 12345, "TCP", 0, func(info *BaseServiceInfo) {
info.ExternalIPs = []string{testExternalIPv4}
info.LoadBalancerSourceRanges = []string{testSourceRangeIPv4}
}),
......@@ -342,8 +342,8 @@ func TestServiceToServiceMap(t *testing.T) {
},
},
},
expected: map[ServicePortName]*ServiceInfoCommon{
makeServicePortName("test", "filterIPv4InIPV6Mode", "testPort"): makeTestServiceInfo(testClusterIPv6, 12345, "TCP", 0, func(info *ServiceInfoCommon) {
expected: map[ServicePortName]*BaseServiceInfo{
makeServicePortName("test", "filterIPv4InIPV6Mode", "testPort"): makeTestServiceInfo(testClusterIPv6, 12345, "TCP", 0, func(info *BaseServiceInfo) {
info.ExternalIPs = []string{testExternalIPv6}
info.LoadBalancerSourceRanges = []string{testSourceRangeIPv6}
}),
......@@ -361,7 +361,7 @@ func TestServiceToServiceMap(t *testing.T) {
t.Errorf("[%s] expected %d new, got %d: %v", tc.desc, len(tc.expected), len(newServices), spew.Sdump(newServices))
}
for svcKey, expectedInfo := range tc.expected {
svcInfo := newServices[svcKey].(*ServiceInfoCommon)
svcInfo := newServices[svcKey].(*BaseServiceInfo)
if !svcInfo.ClusterIP.Equal(expectedInfo.ClusterIP) ||
svcInfo.Port != expectedInfo.Port ||
svcInfo.Protocol != expectedInfo.Protocol ||
......
pkg/proxy/types.go
View file @ f6eed81f
......@@ -48,8 +48,8 @@ func (spn ServicePortName) String() string {
type ServicePort interface {
// String returns service string. An example format can be: `IP:Port/Protocol`.
String() string
// GetClusterIP returns service cluster IP.
GetClusterIP() string
// ClusterIPString returns service cluster IP in string format.
ClusterIPString() string
// GetProtocol returns service protocol.
GetProtocol() api.Protocol
// GetHealthCheckNodePort returns service health check node port if present. If return 0, it means not present.
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment