Repository: k3s-io/k3s
File: pkg/cloudprovider/servicelb.go (Go)

Code before:
	if err := k.updateDaemonSets(); err != nil {
		return node, err
	}

	return node, nil
}

// onChangeEndpointSlice handles changes to EndpointSlices. This is used to ensure that LoadBalancer
// addresses only list Nodes with ready Pods, when their ExternalTrafficPolicy is set to Local.
func (k *k3s) onChangeEndpointSlice(key string, eps *discovery.EndpointSlice) (*discovery.EndpointSlice, error) {
	if eps == nil {
		return nil, nil
	}

	serviceName, ok := eps.Labels[discovery.LabelServiceName]
	if !ok {
		return eps, nil
	}

	k.workqueue.Add(eps.Namespace + "/" + serviceName)
	return eps, nil
}

// onChangeService handles changes to Services. This is used to ensure that changes in the Service are
// reflected in the LoadBalancer status (e.g. changing to externalTrafficPolicy=Local which should remove externalIPs)
func (k *k3s) onChangeService(key string, svc *core.Service) (*core.Service, error) {
	k.workqueue.Add(svc.Namespace + "/" + svc.Name)
	return svc, nil
}

// runWorker dequeues Service changes from the work queue
// We run a lightweight work queue to handle service updates. We don't need the full overhead
// of a wrangler service controller and shared informer cache, but we do want to run changes
// through a keyed queue to reduce thrashing when pods are updated. Much of this is cribbed from
// https://github.com/rancher/lasso/blob/release/v2.5/pkg/controller/controller.go#L173-L215
func (k *k3s) runWorker() {
	for k.processNextWorkItem() {
	}
}

// processNextWorkItem does work for a single item in the queue,
// returning a boolean that indicates if the queue should continue
// to be serviced.
func (k *k3s) processNextWorkItem() bool {
	obj, shutdown := k.workqueue.Get()

	if shutdown {
		return false
	}

	if err := k.processSingleItem(obj); err != nil && !apierrors.IsConflict(err) {

Review comment (refactor):
I'm curious why this is necessary. This would appear to duplicate the OnChange handler in the core Kubernetes cloud-controller code base, which calls EnsureLoadBalancer or UpdateLoadBalancer when the Service object changes. Is it not calling those handlers on the right events to see these changes?

Code after:
		return node, err
	}

	return node, nil
}

// onChangeEndpointSlice handles changes to EndpointSlices. This is used to ensure that LoadBalancer
// addresses only list Nodes with ready Pods, when their ExternalTrafficPolicy is set to Local.
func (k *k3s) onChangeEndpointSlice(key string, eps *discovery.EndpointSlice) (*discovery.EndpointSlice, error) {
	if eps == nil {
		return nil, nil
	}

	serviceName, ok := eps.Labels[discovery.LabelServiceName]
	if !ok {
		return eps, nil
	}

	k.workqueue.Add(eps.Namespace + "/" + serviceName)
	return eps, nil
}

// runWorker dequeues Service changes from the work queue
// We run a lightweight work queue to handle service updates. We don't need the full overhead
// of a wrangler service controller and shared informer cache, but we do want to run changes
// through a keyed queue to reduce thrashing when pods are updated. Much of this is cribbed from
// https://github.com/rancher/lasso/blob/release/v2.5/pkg/controller/controller.go#L173-L215
func (k *k3s) runWorker() {
	for k.processNextWorkItem() {
	}
}

// processNextWorkItem does work for a single item in the queue,
// returning a boolean that indicates if the queue should continue
// to be serviced.
func (k *k3s) processNextWorkItem() bool {
	obj, shutdown := k.workqueue.Get()

	if shutdown {
		return false
	}

	if err := k.processSingleItem(obj); err != nil && !apierrors.IsConflict(err) {
		logrus.Errorf("%s: %v", controllerName, err)
	}
	return true
}

// processSingleItem processes a single item from the work queue,
// requeueing it if the handler fails.
func (k *k3s) processSingleItem(obj interface{}) error {