const serviceSyncPeriod … const nodeSyncPeriod … const minRetryDelay … const maxRetryDelay … const ToBeDeletedTaint … type cachedService … type serviceCache … type Controller … // New returns a new service controller to keep cloud provider service resources // (like load balancers) in sync with the registry. func New( cloud cloudprovider.Interface, kubeClient clientset.Interface, serviceInformer coreinformers.ServiceInformer, nodeInformer coreinformers.NodeInformer, clusterName string, featureGate featuregate.FeatureGate, ) (*Controller, error) { … } // obj could be an *v1.Service, or a DeletionFinalStateUnknown marker item. func (c *Controller) enqueueService(obj interface{ … } // obj could be an *v1.Service, or a DeletionFinalStateUnknown marker item. func (c *Controller) enqueueNode(obj interface{ … } // Run starts a background goroutine that watches for changes to services that // have (or had) LoadBalancers=true and ensures that they have // load balancers created and deleted appropriately. // serviceSyncPeriod controls how often we check the cluster's services to // ensure that the correct load balancers exist. // nodeSyncPeriod controls how often we check the cluster's nodes to determine // if load balancers need to be updated to point to a new set. // // It's an error to call Run() more than once for a given ServiceController // object. func (c *Controller) Run(ctx context.Context, workers int, controllerManagerMetrics *controllersmetrics.ControllerManagerMetrics) { … } // worker runs a worker thread that just dequeues items, processes them, and marks them done. // It enforces that the syncHandler is never invoked concurrently with the same key. func (c *Controller) serviceWorker(ctx context.Context) { … } // worker runs a worker thread that just dequeues items, processes them, and marks them done. // It enforces that the syncHandler is never invoked concurrently with the same key. func (c *Controller) nodeWorker(ctx context.Context, workers int) { … } func (c *Controller) processNextNodeItem(ctx context.Context, workers int) bool { … } func (c *Controller) processNextServiceItem(ctx context.Context) bool { … } func (c *Controller) init() error { … } // processServiceCreateOrUpdate operates loadbalancers for the incoming service accordingly. // Returns an error if processing the service update failed. func (c *Controller) processServiceCreateOrUpdate(ctx context.Context, service *v1.Service, key string) error { … } type loadBalancerOperation … const deleteLoadBalancer … const ensureLoadBalancer … const maxNodeNamesToLog … // syncLoadBalancerIfNeeded ensures that service's status is synced up with loadbalancer // i.e. creates loadbalancer for service if requested and deletes loadbalancer if the service // doesn't want a loadbalancer no more. Returns whatever error occurred. func (c *Controller) syncLoadBalancerIfNeeded(ctx context.Context, service *v1.Service, key string) (loadBalancerOperation, error) { … } func (c *Controller) ensureLoadBalancer(ctx context.Context, service *v1.Service) (*v1.LoadBalancerStatus, error) { … } func (c *Controller) storeLastSyncedNodes(svc *v1.Service, nodes []*v1.Node) { … } func (c *Controller) getLastSyncedNodes(svc *v1.Service) []*v1.Node { … } // ListKeys implements the interface required by DeltaFIFO to list the keys we // already know about. func (s *serviceCache) ListKeys() []string { … } // GetByKey returns the value stored in the serviceMap under the given key func (s *serviceCache) GetByKey(key string) (interface{ … } // ListKeys implements the interface required by DeltaFIFO to list the keys we // already know about. func (s *serviceCache) allServices() []*v1.Service { … } func (s *serviceCache) get(serviceName string) (*cachedService, bool) { … } func (s *serviceCache) getOrCreate(serviceName string) *cachedService { … } func (s *serviceCache) set(serviceName string, service *cachedService) { … } func (s *serviceCache) delete(serviceName string) { … } // needsCleanup checks if load balancer needs to be cleaned up as indicated by finalizer. func needsCleanup(service *v1.Service) bool { … } // needsUpdate checks if load balancer needs to be updated due to change in attributes. func needsUpdate(oldService *v1.Service, newService *v1.Service) bool { … } func getPortsForLB(service *v1.Service) []*v1.ServicePort { … } func portsEqualForLB(x, y *v1.Service) bool { … } func portSlicesEqualForLB(x, y []*v1.ServicePort) bool { … } func portEqualForLB(x, y *v1.ServicePort) bool { … } func nodeNames(nodes []*v1.Node) sets.String { … } func loggableNodeNames(nodes []*v1.Node) []string { … } func shouldSyncUpdatedNode(oldNode, newNode *v1.Node) bool { … } // syncNodes handles updating the hosts pointed to by all load // balancers whenever the set of nodes in the cluster changes. func (c *Controller) syncNodes(ctx context.Context, workers int) sets.String { … } // nodeSyncService syncs the nodes for one load balancer type service. The return value // indicates if we should retry. Hence, this functions returns false if we've updated // load balancers and finished doing it successfully, or didn't try to at all because // there's no need. This function returns true if we tried to update load balancers and // failed, indicating to the caller that we should try again. func (c *Controller) nodeSyncService(svc *v1.Service) bool { … } func nodesSufficientlyEqual(oldNodes, newNodes []*v1.Node) bool { … } // updateLoadBalancerHosts updates all existing load balancers so that // they will match the latest list of nodes with input number of workers. // Returns the list of services that couldn't be updated. func (c *Controller) updateLoadBalancerHosts(ctx context.Context, services []*v1.Service, workers int) (servicesToRetry sets.String) { … } // Updates the load balancer of a service, assuming we hold the mutex // associated with the service. func (c *Controller) lockedUpdateLoadBalancerHosts(service *v1.Service, hosts []*v1.Node) error { … } func wantsLoadBalancer(service *v1.Service) bool { … } func loadBalancerIPsAreEqual(oldService, newService *v1.Service) bool { … } // syncService will sync the Service with the given key if it has had its expectations fulfilled, // meaning it did not expect to see any more of its pods created or deleted. This function is not meant to be // invoked concurrently with the same key. func (c *Controller) syncService(ctx context.Context, key string) error { … } func (c *Controller) processServiceDeletion(ctx context.Context, key string) error { … } func (c *Controller) processLoadBalancerDelete(ctx context.Context, service *v1.Service, key string) error { … } // addFinalizer patches the service to add finalizer. func (c *Controller) addFinalizer(service *v1.Service) error { … } // removeFinalizer patches the service to remove finalizer. func (c *Controller) removeFinalizer(service *v1.Service) error { … } // removeString returns a newly created []string that contains all items from slice that // are not equal to s. func removeString(slice []string, s string) []string { … } // patchStatus patches the service with the given LoadBalancerStatus. func (c *Controller) patchStatus(service *v1.Service, previousStatus, newStatus *v1.LoadBalancerStatus) error { … } type NodeConditionPredicate … var allNodePredicates … var stableNodeSetPredicates … // We consider the node for load balancing only when the node is not labelled for exclusion. func nodeIncludedPredicate(node *v1.Node) bool { … } // We consider the node for load balancing only when its not tainted for deletion by the cluster autoscaler. func nodeUnTaintedPredicate(node *v1.Node) bool { … } // We consider the node for load balancing only when its NodeReady condition status is ConditionTrue func nodeReadyPredicate(node *v1.Node) bool { … } func nodeNotDeletedPredicate(node *v1.Node) bool { … } // listWithPredicate gets nodes that matches all predicate functions. func listWithPredicates(nodeLister corelisters.NodeLister, predicates ...NodeConditionPredicate) ([]*v1.Node, error) { … } func filterWithPredicates(nodes []*v1.Node, predicates ...NodeConditionPredicate) []*v1.Node { … } func respectsPredicates(node *v1.Node, predicates ...NodeConditionPredicate) bool { … }
Codebase Browser
kubernetes