kube-proxy: some code cleanup

This commit is contained in:
SataQiu 2020-03-17 21:46:54 +08:00
parent f8ff8f4420
commit 64a496e645
6 changed files with 53 additions and 5 deletions

View File

@ -20,7 +20,7 @@ import (
"fmt" "fmt"
"time" "time"
"k8s.io/api/core/v1" v1 "k8s.io/api/core/v1"
discovery "k8s.io/api/discovery/v1beta1" discovery "k8s.io/api/discovery/v1beta1"
utilruntime "k8s.io/apimachinery/pkg/util/runtime" utilruntime "k8s.io/apimachinery/pkg/util/runtime"
coreinformers "k8s.io/client-go/informers/core/v1" coreinformers "k8s.io/client-go/informers/core/v1"
@ -381,7 +381,7 @@ type NodeHandler interface {
// OnNodeUpdate is called whenever modification of an existing // OnNodeUpdate is called whenever modification of an existing
// node object is observed. // node object is observed.
OnNodeUpdate(oldNode, node *v1.Node) OnNodeUpdate(oldNode, node *v1.Node)
// OnNodeDelete is called whever deletion of an existing node // OnNodeDelete is called whenever deletion of an existing node
// object is observed. // object is observed.
OnNodeDelete(node *v1.Node) OnNodeDelete(node *v1.Node)
// OnNodeSynced is called once all the initial event handlers were // OnNodeSynced is called once all the initial event handlers were

View File

@ -541,7 +541,7 @@ func (proxier *Proxier) OnServiceDelete(service *v1.Service) {
} }
// OnServiceSynced is called once all the initial even handlers were // OnServiceSynced is called once all the initial event handlers were
// called and the state is fully propagated to local cache. // called and the state is fully propagated to local cache.
func (proxier *Proxier) OnServiceSynced() { func (proxier *Proxier) OnServiceSynced() {
proxier.mu.Lock() proxier.mu.Lock()
@ -663,7 +663,7 @@ func (proxier *Proxier) OnNodeUpdate(oldNode, node *v1.Node) {
proxier.syncProxyRules() proxier.syncProxyRules()
} }
// OnNodeDelete is called whever deletion of an existing node // OnNodeDelete is called whenever deletion of an existing node
// object is observed. // object is observed.
func (proxier *Proxier) OnNodeDelete(node *v1.Node) { func (proxier *Proxier) OnNodeDelete(node *v1.Node) {
if node.Name != proxier.hostname { if node.Name != proxier.hostname {

View File

@ -963,7 +963,7 @@ func (proxier *Proxier) OnNodeUpdate(oldNode, node *v1.Node) {
proxier.syncProxyRules() proxier.syncProxyRules()
} }
// OnNodeDelete is called whever deletion of an existing node // OnNodeDelete is called whenever deletion of an existing node
// object is observed. // object is observed.
func (proxier *Proxier) OnNodeDelete(node *v1.Node) { func (proxier *Proxier) OnNodeDelete(node *v1.Node) {
if node.Name != proxier.hostname { if node.Name != proxier.hostname {

View File

@ -621,18 +621,26 @@ func (proxier *Proxier) serviceChange(previous, current *v1.Service, detail stri
} }
} }
// OnServiceAdd is called whenever creation of new service object
// is observed.
func (proxier *Proxier) OnServiceAdd(service *v1.Service) { func (proxier *Proxier) OnServiceAdd(service *v1.Service) {
proxier.serviceChange(nil, service, "OnServiceAdd") proxier.serviceChange(nil, service, "OnServiceAdd")
} }
// OnServiceUpdate is called whenever modification of an existing
// service object is observed.
func (proxier *Proxier) OnServiceUpdate(oldService, service *v1.Service) { func (proxier *Proxier) OnServiceUpdate(oldService, service *v1.Service) {
proxier.serviceChange(oldService, service, "OnServiceUpdate") proxier.serviceChange(oldService, service, "OnServiceUpdate")
} }
// OnServiceDelete is called whenever deletion of an existing service
// object is observed.
func (proxier *Proxier) OnServiceDelete(service *v1.Service) { func (proxier *Proxier) OnServiceDelete(service *v1.Service) {
proxier.serviceChange(service, nil, "OnServiceDelete") proxier.serviceChange(service, nil, "OnServiceDelete")
} }
// OnServiceSynced is called once all the initial event handlers were
// called and the state is fully propagated to local cache.
func (proxier *Proxier) OnServiceSynced() { func (proxier *Proxier) OnServiceSynced() {
klog.V(2).Infof("userspace OnServiceSynced") klog.V(2).Infof("userspace OnServiceSynced")
@ -649,18 +657,26 @@ func (proxier *Proxier) OnServiceSynced() {
go proxier.syncProxyRules() go proxier.syncProxyRules()
} }
// OnEndpointsAdd is called whenever creation of new endpoints object
// is observed.
func (proxier *Proxier) OnEndpointsAdd(endpoints *v1.Endpoints) { func (proxier *Proxier) OnEndpointsAdd(endpoints *v1.Endpoints) {
proxier.loadBalancer.OnEndpointsAdd(endpoints) proxier.loadBalancer.OnEndpointsAdd(endpoints)
} }
// OnEndpointsUpdate is called whenever modification of an existing
// endpoints object is observed.
func (proxier *Proxier) OnEndpointsUpdate(oldEndpoints, endpoints *v1.Endpoints) { func (proxier *Proxier) OnEndpointsUpdate(oldEndpoints, endpoints *v1.Endpoints) {
proxier.loadBalancer.OnEndpointsUpdate(oldEndpoints, endpoints) proxier.loadBalancer.OnEndpointsUpdate(oldEndpoints, endpoints)
} }
// OnEndpointsDelete is called whenever deletion of an existing endpoints
// object is observed.
func (proxier *Proxier) OnEndpointsDelete(endpoints *v1.Endpoints) { func (proxier *Proxier) OnEndpointsDelete(endpoints *v1.Endpoints) {
proxier.loadBalancer.OnEndpointsDelete(endpoints) proxier.loadBalancer.OnEndpointsDelete(endpoints)
} }
// OnEndpointsSynced is called once all the initial event handlers were
// called and the state is fully propagated to local cache.
func (proxier *Proxier) OnEndpointsSynced() { func (proxier *Proxier) OnEndpointsSynced() {
klog.V(2).Infof("userspace OnEndpointsSynced") klog.V(2).Infof("userspace OnEndpointsSynced")
proxier.loadBalancer.OnEndpointsSynced() proxier.loadBalancer.OnEndpointsSynced()

View File

@ -761,6 +761,8 @@ func (proxier *Proxier) isInitialized() bool {
return atomic.LoadInt32(&proxier.initialized) > 0 return atomic.LoadInt32(&proxier.initialized) > 0
} }
// OnServiceAdd is called whenever creation of new service object
// is observed.
func (proxier *Proxier) OnServiceAdd(service *v1.Service) { func (proxier *Proxier) OnServiceAdd(service *v1.Service) {
namespacedName := types.NamespacedName{Namespace: service.Namespace, Name: service.Name} namespacedName := types.NamespacedName{Namespace: service.Namespace, Name: service.Name}
if proxier.serviceChanges.update(&namespacedName, nil, service, proxier.hns) && proxier.isInitialized() { if proxier.serviceChanges.update(&namespacedName, nil, service, proxier.hns) && proxier.isInitialized() {
@ -768,6 +770,8 @@ func (proxier *Proxier) OnServiceAdd(service *v1.Service) {
} }
} }
// OnServiceUpdate is called whenever modification of an existing
// service object is observed.
func (proxier *Proxier) OnServiceUpdate(oldService, service *v1.Service) { func (proxier *Proxier) OnServiceUpdate(oldService, service *v1.Service) {
namespacedName := types.NamespacedName{Namespace: service.Namespace, Name: service.Name} namespacedName := types.NamespacedName{Namespace: service.Namespace, Name: service.Name}
if proxier.serviceChanges.update(&namespacedName, oldService, service, proxier.hns) && proxier.isInitialized() { if proxier.serviceChanges.update(&namespacedName, oldService, service, proxier.hns) && proxier.isInitialized() {
@ -775,6 +779,8 @@ func (proxier *Proxier) OnServiceUpdate(oldService, service *v1.Service) {
} }
} }
// OnServiceDelete is called whenever deletion of an existing service
// object is observed.
func (proxier *Proxier) OnServiceDelete(service *v1.Service) { func (proxier *Proxier) OnServiceDelete(service *v1.Service) {
namespacedName := types.NamespacedName{Namespace: service.Namespace, Name: service.Name} namespacedName := types.NamespacedName{Namespace: service.Namespace, Name: service.Name}
if proxier.serviceChanges.update(&namespacedName, service, nil, proxier.hns) && proxier.isInitialized() { if proxier.serviceChanges.update(&namespacedName, service, nil, proxier.hns) && proxier.isInitialized() {
@ -782,6 +788,8 @@ func (proxier *Proxier) OnServiceDelete(service *v1.Service) {
} }
} }
// OnServiceSynced is called once all the initial event handlers were
// called and the state is fully propagated to local cache.
func (proxier *Proxier) OnServiceSynced() { func (proxier *Proxier) OnServiceSynced() {
proxier.mu.Lock() proxier.mu.Lock()
proxier.servicesSynced = true proxier.servicesSynced = true
@ -836,6 +844,8 @@ func (proxier *Proxier) updateServiceMap() (result updateServiceMapResult) {
return result return result
} }
// OnEndpointsAdd is called whenever creation of new endpoints object
// is observed.
func (proxier *Proxier) OnEndpointsAdd(endpoints *v1.Endpoints) { func (proxier *Proxier) OnEndpointsAdd(endpoints *v1.Endpoints) {
namespacedName := types.NamespacedName{Namespace: endpoints.Namespace, Name: endpoints.Name} namespacedName := types.NamespacedName{Namespace: endpoints.Namespace, Name: endpoints.Name}
if proxier.endpointsChanges.update(&namespacedName, nil, endpoints, proxier.hns) && proxier.isInitialized() { if proxier.endpointsChanges.update(&namespacedName, nil, endpoints, proxier.hns) && proxier.isInitialized() {
@ -843,6 +853,8 @@ func (proxier *Proxier) OnEndpointsAdd(endpoints *v1.Endpoints) {
} }
} }
// OnEndpointsUpdate is called whenever modification of an existing
// endpoints object is observed.
func (proxier *Proxier) OnEndpointsUpdate(oldEndpoints, endpoints *v1.Endpoints) { func (proxier *Proxier) OnEndpointsUpdate(oldEndpoints, endpoints *v1.Endpoints) {
namespacedName := types.NamespacedName{Namespace: endpoints.Namespace, Name: endpoints.Name} namespacedName := types.NamespacedName{Namespace: endpoints.Namespace, Name: endpoints.Name}
if proxier.endpointsChanges.update(&namespacedName, oldEndpoints, endpoints, proxier.hns) && proxier.isInitialized() { if proxier.endpointsChanges.update(&namespacedName, oldEndpoints, endpoints, proxier.hns) && proxier.isInitialized() {
@ -850,6 +862,8 @@ func (proxier *Proxier) OnEndpointsUpdate(oldEndpoints, endpoints *v1.Endpoints)
} }
} }
// OnEndpointsDelete is called whenever deletion of an existing endpoints
// object is observed.
func (proxier *Proxier) OnEndpointsDelete(endpoints *v1.Endpoints) { func (proxier *Proxier) OnEndpointsDelete(endpoints *v1.Endpoints) {
namespacedName := types.NamespacedName{Namespace: endpoints.Namespace, Name: endpoints.Name} namespacedName := types.NamespacedName{Namespace: endpoints.Namespace, Name: endpoints.Name}
if proxier.endpointsChanges.update(&namespacedName, endpoints, nil, proxier.hns) && proxier.isInitialized() { if proxier.endpointsChanges.update(&namespacedName, endpoints, nil, proxier.hns) && proxier.isInitialized() {
@ -857,6 +871,8 @@ func (proxier *Proxier) OnEndpointsDelete(endpoints *v1.Endpoints) {
} }
} }
// OnEndpointsSynced is called once all the initial event handlers were
// called and the state is fully propagated to local cache.
func (proxier *Proxier) OnEndpointsSynced() { func (proxier *Proxier) OnEndpointsSynced() {
proxier.mu.Lock() proxier.mu.Lock()
proxier.endpointsSynced = true proxier.endpointsSynced = true

View File

@ -411,34 +411,50 @@ func (proxier *Proxier) unmergeService(service *v1.Service, existingPortPortals
} }
} }
// OnServiceAdd is called whenever creation of new service object
// is observed.
func (proxier *Proxier) OnServiceAdd(service *v1.Service) { func (proxier *Proxier) OnServiceAdd(service *v1.Service) {
_ = proxier.mergeService(service) _ = proxier.mergeService(service)
} }
// OnServiceUpdate is called whenever modification of an existing
// service object is observed.
func (proxier *Proxier) OnServiceUpdate(oldService, service *v1.Service) { func (proxier *Proxier) OnServiceUpdate(oldService, service *v1.Service) {
existingPortPortals := proxier.mergeService(service) existingPortPortals := proxier.mergeService(service)
proxier.unmergeService(oldService, existingPortPortals) proxier.unmergeService(oldService, existingPortPortals)
} }
// OnServiceDelete is called whenever deletion of an existing service
// object is observed.
func (proxier *Proxier) OnServiceDelete(service *v1.Service) { func (proxier *Proxier) OnServiceDelete(service *v1.Service) {
proxier.unmergeService(service, map[ServicePortPortalName]bool{}) proxier.unmergeService(service, map[ServicePortPortalName]bool{})
} }
// OnServiceSynced is called once all the initial event handlers were
// called and the state is fully propagated to local cache.
func (proxier *Proxier) OnServiceSynced() { func (proxier *Proxier) OnServiceSynced() {
} }
// OnEndpointsAdd is called whenever creation of new endpoints object
// is observed.
func (proxier *Proxier) OnEndpointsAdd(endpoints *v1.Endpoints) { func (proxier *Proxier) OnEndpointsAdd(endpoints *v1.Endpoints) {
proxier.loadBalancer.OnEndpointsAdd(endpoints) proxier.loadBalancer.OnEndpointsAdd(endpoints)
} }
// OnEndpointsUpdate is called whenever modification of an existing
// endpoints object is observed.
func (proxier *Proxier) OnEndpointsUpdate(oldEndpoints, endpoints *v1.Endpoints) { func (proxier *Proxier) OnEndpointsUpdate(oldEndpoints, endpoints *v1.Endpoints) {
proxier.loadBalancer.OnEndpointsUpdate(oldEndpoints, endpoints) proxier.loadBalancer.OnEndpointsUpdate(oldEndpoints, endpoints)
} }
// OnEndpointsDelete is called whenever deletion of an existing endpoints
// object is observed.
func (proxier *Proxier) OnEndpointsDelete(endpoints *v1.Endpoints) { func (proxier *Proxier) OnEndpointsDelete(endpoints *v1.Endpoints) {
proxier.loadBalancer.OnEndpointsDelete(endpoints) proxier.loadBalancer.OnEndpointsDelete(endpoints)
} }
// OnEndpointsSynced is called once all the initial event handlers were
// called and the state is fully propagated to local cache.
func (proxier *Proxier) OnEndpointsSynced() { func (proxier *Proxier) OnEndpointsSynced() {
proxier.loadBalancer.OnEndpointsSynced() proxier.loadBalancer.OnEndpointsSynced()
} }