kube-vip/pkg/services/processor.go

package services

import (
	"context"
	"errors"
	"fmt"
	log "log/slog"
	"reflect"
	"sync"

	"github.com/kube-vip/kube-vip/pkg/arp"
	"github.com/kube-vip/kube-vip/pkg/bgp"
	"github.com/kube-vip/kube-vip/pkg/election"
	"github.com/kube-vip/kube-vip/pkg/endpoints"
	"github.com/kube-vip/kube-vip/pkg/endpoints/providers"
	"github.com/kube-vip/kube-vip/pkg/instance"
	"github.com/kube-vip/kube-vip/pkg/kubevip"
	"github.com/kube-vip/kube-vip/pkg/lease"
	"github.com/kube-vip/kube-vip/pkg/networkinterface"
	"github.com/kube-vip/kube-vip/pkg/servicecontext"
	"github.com/kube-vip/kube-vip/pkg/utils"
	"github.com/kube-vip/kube-vip/pkg/vip"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/vishvananda/netlink"
	v1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/apimachinery/pkg/watch"
	"k8s.io/client-go/kubernetes"
)

type Processor struct {
	config        *kubevip.Config
	lbClassFilter func(svc *v1.Service, config *kubevip.Config) bool
	svcMap        sync.Map

	// Keeps track of all running instances
	ServiceInstances []*instance.Instance

	mutex     sync.Mutex
	bgpServer *bgp.Server

	clientSet   *kubernetes.Clientset
	rwClientSet *kubernetes.Clientset

	shutdownChan chan struct{}

	// This is a prometheus counter used to count the number of events received
	// from the service watcher
	CountServiceWatchEvent *prometheus.CounterVec

	intfMgr *networkinterface.Manager
	arpMgr  *arp.Manager

	leaseMgr *lease.Manager

	// nodeLabelManager is the manager for the node labels
	nodeLabelManager labelManager

	electionMgr *election.Manager
}

// labelManager is the interface for the node label manager to add/remove labels
type labelManager interface {
	AddLabel(ctx context.Context, svc *v1.Service) error
	RemoveLabel(ctx context.Context, svc *v1.Service) error
}

func NewServicesProcessor(config *kubevip.Config, bgpServer *bgp.Server,
	clientSet *kubernetes.Clientset, rwClientSet *kubernetes.Clientset, shutdownChan chan struct{},
	intfMgr *networkinterface.Manager, arpMgr *arp.Manager, nodeLabelManager labelManager,
	electionMgr *election.Manager, leaseMgr *lease.Manager) *Processor {
	lbClassFilterFunc := lbClassFilter
	if config.LoadBalancerClassLegacyHandling {
		lbClassFilterFunc = lbClassFilterLegacy
	}

	return &Processor{
		config:           config,
		lbClassFilter:    lbClassFilterFunc,
		ServiceInstances: []*instance.Instance{},
		bgpServer:        bgpServer,
		clientSet:        clientSet,
		rwClientSet:      rwClientSet,
		shutdownChan:     shutdownChan,
		CountServiceWatchEvent: prometheus.NewCounterVec(prometheus.CounterOpts{
			Namespace: "kube_vip",
			Subsystem: "manager",
			Name:      "all_services_events",
			Help:      "Count all events fired by the service watcher categorised by event type",
		}, []string{"type"}),

		intfMgr:          intfMgr,
		arpMgr:           arpMgr,
		leaseMgr:         leaseMgr,
		nodeLabelManager: nodeLabelManager,
		electionMgr:      electionMgr,
	}
}

func (p *Processor) AddOrModify(ctx context.Context, event watch.Event, serviceFunc func(*servicecontext.Context, *v1.Service) error) (bool, error) {
	// log.Debugf("Endpoints for service [%s] have been Created or modified", s.service.ServiceName)
	svc, ok := event.Object.(*v1.Service)
	if !ok {
		return false, fmt.Errorf("unable to parse Kubernetes services from API watcher")
	}

	// We only care about LoadBalancer services
	if svc.Spec.Type != v1.ServiceTypeLoadBalancer {
		return true, nil
	}

	// Check if we ignore this service
	if svc.Annotations[kubevip.LoadbalancerIgnore] == "true" {
		log.Info("ignore annotation for kube-vip", "service name", svc.Name)
		return true, nil
	}

	// Check the loadBalancer class
	if p.lbClassFilter(svc, p.config) {
		return true, nil
	}

	svcAddresses, svcHostnames := instance.FetchServiceAddresses(svc)

	// We only care about LoadBalancer services that have been allocated an address
	if len(svcAddresses) <= 0 && len(svcHostnames) <= 0 {
		return true, nil
	}

	_, usesCommonLease := svc.Annotations[kubevip.ServiceLease]
	if usesCommonLease && svc.Spec.ExternalTrafficPolicy != v1.ServiceExternalTrafficPolicyTypeCluster {
		return false, fmt.Errorf("annotation %q cannot be used with service traffic policy other than %q",
			kubevip.ServiceLease, v1.ServiceExternalTrafficPolicyTypeCluster)
	}

	svcCtx, err := p.getServiceContext(svc.UID)
	if err != nil {
		return false, fmt.Errorf("failed to get service context: %w", err)
	}

	// The modified event should only be triggered if the service has been modified (i.e. moved somewhere else)
	if event.Type == watch.Modified {
		i := instance.FindServiceInstance(svc, p.ServiceInstances)
		shouldGarbageCollect := false
		if i != nil {
			originalServiceAddresses, originalServiceHostnames := instance.FetchServiceAddresses(i.ServiceSnapshot)
			shouldGarbageCollect =
				// Service addresses changed
				!reflect.DeepEqual(originalServiceAddresses, svcAddresses) ||
					// Service hostnames changed
					!reflect.DeepEqual(originalServiceHostnames, svcHostnames) ||
					// ExternalTrafficPolicy changed
					svc.Spec.ExternalTrafficPolicy != i.ServiceSnapshot.Spec.ExternalTrafficPolicy ||
					// IP stack configuration changed
					!reflect.DeepEqual(svc.Spec.IPFamilies, i.ServiceSnapshot.Spec.IPFamilies) ||
					*svc.Spec.IPFamilyPolicy != *i.ServiceSnapshot.Spec.IPFamilyPolicy ||
					// DDNS was disabled/enabled
					svc.Annotations[kubevip.ServiceDDNS] != i.ServiceSnapshot.Annotations[kubevip.ServiceDDNS]

		}
		if shouldGarbageCollect {
			for _, addr := range svcAddresses {
				// log.Debugf("(svcs) Retrieving local addresses, to ensure that this modified address doesn't exist: %s", addr)
				f, err := vip.GarbageCollect(p.config.Interface, addr, p.intfMgr)
				if err != nil {
					log.Error("(svcs) cleaning existing address error", "err", err)
				}
				if f {
					log.Warn("(svcs) already found existing config", "address", addr, "adapter", p.config.Interface)
				}
			}
			// This service has been modified, but it was also active.
			if svcCtx != nil && svcCtx.IsActive {
				log.Warn("(svcs) The load balancer has changed, cancelling original load balancer")
				//Set it to inactive
				svcCtx.IsActive = false
				svcCtx.Cancel()
				log.Warn("(svcs) waiting for load balancer to finish")
				<-svcCtx.Ctx.Done()

				if err := p.deleteService(ctx, svc.UID); err != nil {
					log.Error("(svc) unable to remove", "service", svc.UID)
				}
				// in theory this should never fail
				p.svcMap.Delete(svc.UID)
				// Reset the the svcCtx when it was garbage collected
				// As the next function will create a new context when nil
				svcCtx = nil
			}
		}
	}

	// Architecture walkthrough: (Had to do this as this code path is making my head hurt)

	// Is the service active (bool), if not then process this new service
	// Does this service use an election per service?
	//

	if svcCtx == nil || svcCtx != nil && !svcCtx.IsActive {
		ips, hostnames := instance.FetchServiceAddresses(svc)
		log.Debug("(svcs) has been added/modified with addresses", "service name", svc.Name, "ips", ips, "hostnames", hostnames)

		if svcCtx == nil {
			ns, name := lease.ServiceName(svc)
			leaseID := lease.NewID(p.config.LeaderElectionType, ns, name)
			lease := p.leaseMgr.Add(ctx, leaseID)
			svcCtx = servicecontext.New(lease.Ctx)
			p.svcMap.Store(svc.UID, svcCtx)
		}

		if p.config.EnableServicesElection || // Service Election
			((p.config.EnableRoutingTable || p.config.EnableBGP) && // Routing table mode or BGP
				(!p.config.EnableLeaderElection && !p.config.EnableServicesElection)) { // No leaderelection or services election

			// If this load balancer Traffic Policy is "local"
			if svc.Spec.ExternalTrafficPolicy == v1.ServiceExternalTrafficPolicyTypeLocal {

				// Start an endpoint watcher if we're not watching it already
				if !svcCtx.IsWatched {
					// background the endpoint watcher
					if (p.config.EnableRoutingTable || p.config.EnableBGP) && (!p.config.EnableLeaderElection && !p.config.EnableServicesElection) {
						err = serviceFunc(svcCtx, svc)
						if err != nil {
							log.Error(err.Error())
							if errors.Is(err, &utils.PanicError{}) {
								return false, err
							}
						}
					}

					go func() {
						defer func() {
							if svcCtx != nil {
								svcCtx.IsWatched = false
							}
						}()

						if svc.Spec.ExternalTrafficPolicy == v1.ServiceExternalTrafficPolicyTypeLocal {
							// Add Endpoint or EndpointSlices watcher
							var provider providers.Provider
							if p.config.EnableEndpoints {
								provider = providers.NewEndpoints()
							} else {
								provider = providers.NewEndpointslices()
							}
							if err = p.watchEndpoint(svcCtx, p.config.NodeName, svc, provider); err != nil {
								log.Error(err.Error())
							}
						}
					}()

					// We're now watching this service
					svcCtx.IsWatched = true
				}
			} else if (p.config.EnableBGP || p.config.EnableRoutingTable) && (!p.config.EnableLeaderElection && !p.config.EnableServicesElection) {
				err = serviceFunc(svcCtx, svc)
				if err != nil {
					log.Error(err.Error())
					if errors.Is(err, &utils.PanicError{}) {
						return false, err
					}
				}

				go func() {
					defer func() {
						if svcCtx != nil {
							svcCtx.IsWatched = false
						}
					}()

					if svc.Spec.ExternalTrafficPolicy == v1.ServiceExternalTrafficPolicyTypeCluster {
						// Add Endpoint watcher
						var provider providers.Provider
						if p.config.EnableEndpoints {
							provider = providers.NewEndpoints()
						} else {
							provider = providers.NewEndpointslices()
						}
						if err = p.watchEndpoint(svcCtx, p.config.NodeName, svc, provider); err != nil {
							log.Error(err.Error())
						}
					}
				}()
				// We're now watching this service
				svcCtx.IsWatched = true
			} else {
				go func() {
					for {
						select {
						case <-svcCtx.Ctx.Done():
							log.Warn("(svcs) restartable service watcher ending", "uid", svc.UID)
							return
						default:
							if !svcCtx.IsActive {
								log.Info("(svcs) restartable service watcher starting", "uid", svc.UID)
								err = serviceFunc(svcCtx, svc)
								if err != nil {
									log.Error(err.Error())
									if errors.Is(err, &utils.PanicError{}) {
										svcCtx.Cancel()
									}
								}
								log.Info("(svcs) restartable service watcher done", "uid", svc.UID)
							}
						}
					}
				}()
			}
		} else {
			// Increment the waitGroup before the service Func is called (Done is completed in there)
			err = serviceFunc(svcCtx, svc)
			if err != nil {
				log.Error(err.Error())
				if errors.Is(err, &utils.PanicError{}) {
					return false, err
				}
			}
		}
		if !p.config.EnableServicesElection {
			log.Debug("Service now active", "name", svc.Name, "uid", svc.UID)
			svcCtx.IsActive = true
		}
	}

	return false, nil
}

func (p *Processor) Delete(event watch.Event) (bool, error) {
	svc, ok := event.Object.(*v1.Service)
	if !ok {
		return false, fmt.Errorf("(svcs) unable to parse Kubernetes services from API watcher")
	}
	svcCtx, err := p.getServiceContext(svc.UID)
	if err != nil {
		return false, fmt.Errorf("(svcs) unable to get context: %w", err)
	}

	if svcCtx != nil {
		// We only care about LoadBalancer services
		if svc.Spec.Type != v1.ServiceTypeLoadBalancer {
			return true, nil
		}

		// We can ignore this service
		if svc.Annotations[kubevip.LoadbalancerIgnore] == "true" {
			log.Info("(svcs) ignore annotation for kube-vip", "service name", svc.Name)
			return true, nil
		}

		// If no leader election is enabled, delete routes here
		if !p.config.EnableLeaderElection && !p.config.EnableServicesElection &&
			p.config.EnableRoutingTable && svcCtx.HasConfiguredNetworks() {
			if errs := endpoints.ClearRoutes(svc, &p.ServiceInstances); len(errs) == 0 {
				svcCtx.ConfiguredNetworks.Clear()
			}
		}

		if svcCtx.IsActive && !p.config.EnableServicesElection {
			// If this is an active service then and additional leaderElection will handle stopping
			err = p.deleteService(svcCtx.Ctx, svc.UID)
			if err != nil {
				log.Error(err.Error())
			}
			svcCtx.IsActive = false
		}

		// Calls the cancel function of the context
		log.Warn("(svcs) The load balancer was deleted, cancelling context", "namespace", svc.Namespace, "name", svc.Name, "uid", svc.UID)
		svcCtx.Cancel()
		log.Warn("(svcs) waiting for load balancer to finish", "namespace", svc.Namespace, "name", svc.Name, "uid", svc.UID)
		<-svcCtx.Ctx.Done()
		p.svcMap.Delete(svc.UID)
	}

	log.Info("(svcs) deleted", "service name", svc.Name, "namespace", svc.Namespace)

	return true, nil
}

func (p *Processor) Stop() {
	for _, instance := range p.ServiceInstances {
		for _, cluster := range instance.Clusters {
			cluster.Stop()
		}
	}
}

func (p *Processor) getServiceContext(uid types.UID) (*servicecontext.Context, error) {
	svcCtx, ok := p.svcMap.Load(uid)
	if !ok {
		return nil, nil
	}
	ctx, ok := svcCtx.(*servicecontext.Context)
	if !ok {
		return nil, fmt.Errorf("failed to cast service context pointer - UID: %s", uid)
	}
	return ctx, nil
}

func (p *Processor) CountRouteReferences(route *netlink.Route) int {
	return endpoints.CountRouteReferences(route, &p.ServiceInstances)
}