kube-vip/pkg/services/watch_services.go

package services

import (
	"context"
	"fmt"
	"sync"

	log "log/slog"

	"github.com/kube-vip/kube-vip/pkg/debouncer"
	"github.com/kube-vip/kube-vip/pkg/kubevip"
	"github.com/kube-vip/kube-vip/pkg/metrics"
	"github.com/kube-vip/kube-vip/pkg/trafficmirror"
	"github.com/kube-vip/kube-vip/pkg/utils"
	"github.com/prometheus/client_golang/prometheus"
	v1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/watch"
	"k8s.io/client-go/tools/cache"
	watchtools "k8s.io/client-go/tools/watch"
)

// This function handles the watching of a services endpoints and updates a load balancers endpoint configurations accordingly
func (p *Processor) ServicesWatcher(ctx context.Context, serviceFunc *Callback, forcedOnly bool) error {
	// first start port mirroring if enabled
	if err := p.startTrafficMirroringIfEnabled(); err != nil {
		return err
	}
	defer func() {
		// clean up traffic mirror related config
		err := p.stopTrafficMirroringIfEnabled()
		if err != nil {
			log.Error("Stopping traffic mirroring", "err", err)
		}
	}()

	if p.config.ServiceNamespace == "" {
		// v1.NamespaceAll is actually "", but we'll stay with the const in case things change upstream
		p.config.ServiceNamespace = v1.NamespaceAll
		log.Info("(svcs) starting services watcher for all namespaces")
	} else {
		log.Info("(svcs) starting services watcher", "namespace", p.config.ServiceNamespace)
	}

	// Use a restartable watcher, as this should help in the event of etcd or timeout issues
	rw, err := watchtools.NewRetryWatcherWithContext(ctx, "1", &cache.ListWatch{
		WatchFunc: func(_ metav1.ListOptions) (watch.Interface, error) {
			return utils.WatchWithAuthRetry(ctx, func(ctx context.Context) (watch.Interface, error) {
				return p.rwClientSet.CoreV1().Services(p.config.ServiceNamespace).Watch(ctx, metav1.ListOptions{})
			})
		},
	})
	if err != nil {
		return fmt.Errorf("error creating services watcher: %s", err.Error())
	}

	d, err := debouncer.New(rw.ResultChan(), p.config.DebounceTime)
	if err != nil {
		return fmt.Errorf("failed to create debouncer for endpoints event: %w", err)
	}

	var wg sync.WaitGroup
	defer func() {
		if d != nil {
			d.Stop()
		}
		rw.Stop()
		wg.Wait()
	}()

	watcherCtx, cancelWatcher := context.WithCancelCause(ctx)
	defer cancelWatcher(nil)

	wg.Go(func() {
		if d != nil {
			if err := d.Start(watcherCtx); err != nil {
				log.Error("(svcs) debouncer, cancelling context", "error", err.Error())
				cancelWatcher(utils.WrapPanicError(err, "service debouncer failed"))
			}
		}
		<-watcherCtx.Done()
		log.Debug("(svcs) watcher context cancelled")
		if d != nil {
			d.Stop()
		}
		rw.Stop()
		p.Stop()
	})

	ch := rw.ResultChan()
	if d != nil {
		ch = d.Output()
	}

	// Used for tracking an active endpoint / pod
EventLoop:
	for event := range ch {
		metrics.CountServiceWatchEvent.With(prometheus.Labels{"type": string(event.Type)}).Add(1)
		select {
		case <-ctx.Done():
			log.Info("global context done")
		case <-watcherCtx.Done():
			log.Info("WatcheConotext done")
			break EventLoop
		default:
			// We need to inspect the event and get ResourceVersion out of it
			switch event.Type {
			case watch.Added, watch.Modified:
				if err := p.AddOrModify(watcherCtx, event, serviceFunc, forcedOnly, &wg, cancelWatcher); err != nil {
					if utils.IsPanicError(err) {
						return fmt.Errorf("add/modify service error: %w", err)
					}
					log.Error("service watcher event failed", "type", event.Type, "error", err)
				}
			case watch.Deleted:
				if err := p.Delete(event, forcedOnly); err != nil {
					if utils.IsPanicError(err) {
						return fmt.Errorf("delete service error: %w", err)
					}
					log.Error("service watcher event failed", "type", event.Type, "error", err)
				}
			case watch.Bookmark:
				// Un-used
			case watch.Error:
				log.Error("Error attempting to watch Kubernetes services")
				watchErr := utils.WatchError(event.Object)
				log.Error("services", "err", watchErr)
				return utils.WrapPanicError(watchErr, "service watch failed")
			default:
			}
		}
	}

	if ctx.Err() != nil {
		return nil
	}
	if watcherErr := context.Cause(watcherCtx); watcherErr != nil {
		return watcherErr
	}
	log.Warn("Stopping watching services for type: LoadBalancer in all namespaces")
	return utils.NewPanicError("service watch channel closed unexpectedly")
}

func lbClassFilterLegacy(svc *v1.Service, config *kubevip.Config) bool {
	if svc == nil {
		log.Info("(svcs) service is nil, ignoring")
		return true
	}
	if svc.Spec.LoadBalancerClass != nil {
		// if this isn't nil then it has been configured, check if it the kube-vip loadBalancer class
		if *svc.Spec.LoadBalancerClass != config.LoadBalancerClassName {
			log.Info("(svcs) specified the wrong loadBalancer class", "service name", svc.Name, "lbClass", *svc.Spec.LoadBalancerClass)
			return true
		}
	} else if config.LoadBalancerClassOnly {
		// if kube-vip is configured to only recognize services with kube-vip's lb class, then ignore the services without any lb class
		log.Info("(svcs) kube-vip configured to only recognize services with kube-vip's lb class but the service didn't specify any loadBalancer class, ignoring", "service name", svc.Name)
		return true
	}
	return false
}

func lbClassFilter(svc *v1.Service, config *kubevip.Config) bool {
	if svc == nil {
		log.Info("(svcs) service is nil, ignoring")
		return true
	}
	if svc.Spec.LoadBalancerClass == nil && config.LoadBalancerClassName != "" {
		log.Info("(svcs) no loadBalancer class, ignoring", "service name", svc.Name, "expected lbClass", config.LoadBalancerClassName)
		return true
	}
	if svc.Spec.LoadBalancerClass == nil && config.LoadBalancerClassName == "" {
		return false
	}
	if *svc.Spec.LoadBalancerClass != config.LoadBalancerClassName {
		log.Info("(svcs) specified wrong loadBalancer class, ignoring", "service name", svc.Name, "wrong lbClass", *svc.Spec.LoadBalancerClass, "expected lbClass", config.LoadBalancerClassName)
		return true
	}
	return false
}

func (p *Processor) serviceInterface() string {
	svcIf := p.config.Interface
	if p.config.ServicesInterface != "" {
		svcIf = p.config.ServicesInterface
	}
	return svcIf
}

func (p *Processor) startTrafficMirroringIfEnabled() error {
	if p.config.MirrorDestInterface != "" {
		svcIf := p.serviceInterface()
		log.Info("mirroring traffic", "src", svcIf, "dest", p.config.MirrorDestInterface)
		if err := trafficmirror.MirrorTrafficFromNIC(svcIf, p.config.MirrorDestInterface); err != nil {
			return err
		}
	} else {
		log.Debug("skip starting traffic mirroring since it's not enabled.")
	}
	return nil
}

func (p *Processor) stopTrafficMirroringIfEnabled() error {
	if p.config.MirrorDestInterface != "" {
		svcIf := p.serviceInterface()
		log.Info("clean up qdisc config", "interface", svcIf)
		if err := trafficmirror.CleanupQDSICFromNIC(svcIf); err != nil {
			return err
		}
	} else {
		log.Debug("skip stopping traffic mirroring since it's not enabled.")
	}
	return nil
}