mirror of
https://hubproxy.babadafafafafa.cn/https://github.com/kube-vip/kube-vip.git
synced 2026-09-20 08:03:47 +08:00
348 lines
12 KiB
Go
348 lines
12 KiB
Go
package endpoints
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"net"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
log "log/slog"
|
|
|
|
"github.com/kube-vip/kube-vip/pkg/bgp"
|
|
"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/metrics"
|
|
"github.com/kube-vip/kube-vip/pkg/route"
|
|
"github.com/kube-vip/kube-vip/pkg/servicecontext"
|
|
"github.com/kube-vip/kube-vip/pkg/utils"
|
|
"github.com/kube-vip/kube-vip/pkg/wireguard"
|
|
v1 "k8s.io/api/core/v1"
|
|
"k8s.io/apimachinery/pkg/watch"
|
|
"k8s.io/client-go/kubernetes"
|
|
)
|
|
|
|
type Processor struct {
|
|
config *kubevip.Config
|
|
provider providers.Provider
|
|
bgpServer *bgp.Server
|
|
worker endpointWorker
|
|
instances *[]*instance.Instance
|
|
leaseMgr *lease.Manager
|
|
}
|
|
|
|
func NewEndpointProcessor(config *kubevip.Config, provider providers.Provider, bgpServer *bgp.Server,
|
|
instances *[]*instance.Instance, leaseMgr *lease.Manager, tunnelMgr *wireguard.TunnelManager, routeMgr *route.Manager) *Processor {
|
|
return &Processor{
|
|
config: config,
|
|
provider: provider,
|
|
bgpServer: bgpServer,
|
|
instances: instances,
|
|
leaseMgr: leaseMgr,
|
|
worker: newEndpointWorker(config, provider, bgpServer, instances, leaseMgr, tunnelMgr, routeMgr),
|
|
}
|
|
}
|
|
|
|
func (p *Processor) AddOrModify(svcCtx *servicecontext.Context, event watch.Event,
|
|
lastKnownGoodEndpoint *string, service *v1.Service, id string,
|
|
serviceFunc func(*servicecontext.Context, *v1.Service, *sync.WaitGroup, bool) error, wg *sync.WaitGroup,
|
|
clientSet *kubernetes.Clientset,
|
|
egressUpdateFunc func(context.Context, *v1.Service) error) (bool, error) {
|
|
|
|
var err error
|
|
if err = p.provider.LoadObject(event.Object, svcCtx.Cancel); err != nil {
|
|
return false, fmt.Errorf("[%s] error loading k8s object: %w", p.provider.GetLabel(), err)
|
|
}
|
|
|
|
endpoints, err := p.worker.getEndpoints(service, id)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
|
|
if err := p.worker.setInstanceEndpointsStatus(svcCtx.Ctx, service, endpoints); err != nil {
|
|
log.Error("updating instance", "err", err)
|
|
}
|
|
|
|
allowReconcileWithoutEndpoints := shouldAllowReconcileWithoutEndpoints(service)
|
|
|
|
// Find out if we have any local endpoints
|
|
// if out endpoint is empty then populate it
|
|
// if not, go through the endpoints and see if ours still exists
|
|
// If we have a local endpoint then begin the leader Election, unless it's already running
|
|
//
|
|
|
|
// Check that we have local endpoints
|
|
if len(endpoints) != 0 {
|
|
// Ignore IPv4
|
|
if service.Annotations[kubevip.EgressIPv6] == "true" && !hasV6(endpoints) {
|
|
return true, nil
|
|
}
|
|
|
|
p.updateLastKnownGoodEndpoint(lastKnownGoodEndpoint, endpoints, service)
|
|
|
|
if err := p.startServiceHandlingIfNeeded(svcCtx, service, serviceFunc, wg); err != nil {
|
|
return true, err
|
|
}
|
|
|
|
svcCtx.SignalReadiness()
|
|
|
|
// There are local endpoints available on the node
|
|
// Process immediately if:
|
|
// - No services/leader election is enabled, OR
|
|
// - WireGuard is enabled (it always needs immediate DNAT rule updates)
|
|
if (!p.config.EnableServicesElection && !p.config.EnableLeaderElection) || p.config.EnableWireguard {
|
|
if err := p.worker.processInstance(svcCtx, service); err != nil {
|
|
return false, fmt.Errorf("failed to process non-empty instance: %w", err)
|
|
}
|
|
}
|
|
} else {
|
|
if allowReconcileWithoutEndpoints {
|
|
// Explicit opt-in for controllers that create LoadBalancer services without endpoints
|
|
if err := p.startServiceHandlingIfNeeded(svcCtx, service, serviceFunc, wg); err != nil {
|
|
return true, err
|
|
}
|
|
svcCtx.SignalReadiness()
|
|
|
|
if (!p.config.EnableServicesElection && !p.config.EnableLeaderElection) || p.config.EnableWireguard {
|
|
if err := p.worker.processInstance(svcCtx, service); err != nil {
|
|
return false, fmt.Errorf("failed to process endpointless instance: %w", err)
|
|
}
|
|
}
|
|
} else if svcCtx.Signalled.Load() {
|
|
// There are no local endpoints
|
|
svcCtx.ResetReadiness()
|
|
p.worker.clear(svcCtx, lastKnownGoodEndpoint, service)
|
|
if p.config.EnableARP && !p.config.EnableServicesElection {
|
|
i := instance.FindServiceInstance(service, *p.instances)
|
|
for _, c := range i.Clusters {
|
|
c.Stop()
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Set the service accordingly
|
|
p.updateAnnotations(service, lastKnownGoodEndpoint, clientSet, egressUpdateFunc)
|
|
|
|
log.Debug("watcher", "provider",
|
|
p.provider.GetLabel(), "service name", service.Name, "namespace", service.Namespace, "endpoints", len(endpoints), "last endpoint", *lastKnownGoodEndpoint)
|
|
|
|
return false, nil
|
|
}
|
|
|
|
func (p *Processor) Delete(ctx context.Context, service *v1.Service, id string) error {
|
|
if err := p.worker.delete(ctx, service, id); err != nil {
|
|
return fmt.Errorf("[%s] error deleting service: %w", p.provider.GetLabel(), err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (p *Processor) updateLastKnownGoodEndpoint(lastKnownGoodEndpoint *string, endpoints []string, service *v1.Service) {
|
|
// if we haven't populated one, then do so
|
|
family := utils.IPv4Family
|
|
if service.Annotations[kubevip.EgressIPv6] == "true" {
|
|
family = utils.IPv6Family
|
|
}
|
|
|
|
ep := getEndpoint(endpoints, family)
|
|
|
|
if *lastKnownGoodEndpoint == "" {
|
|
*lastKnownGoodEndpoint = ep
|
|
return
|
|
}
|
|
|
|
// check out previous endpoint exists
|
|
stillExists := false
|
|
|
|
for x := range endpoints {
|
|
if endpoints[x] == *lastKnownGoodEndpoint {
|
|
stillExists = true
|
|
}
|
|
}
|
|
// If the last endpoint no longer exists, we cancel our leader Election, and set another endpoint as last known good
|
|
if !stillExists {
|
|
ip := net.ParseIP(*lastKnownGoodEndpoint)
|
|
if (ip.To4() != nil && service.Annotations[kubevip.Egress] == "true") ||
|
|
(ip.To4() == nil && service.Annotations[kubevip.EgressIPv6] == "true") {
|
|
p.worker.removeEgress(service, lastKnownGoodEndpoint)
|
|
}
|
|
// Set our active endpoint to an existing one
|
|
*lastKnownGoodEndpoint = ep
|
|
}
|
|
}
|
|
|
|
func (p *Processor) updateAnnotations(service *v1.Service, lastKnownGoodEndpoint *string,
|
|
clientSet *kubernetes.Clientset,
|
|
egressUpdateFunc func(context.Context, *v1.Service) error) {
|
|
// Set the service accordingly
|
|
if service.Annotations[kubevip.Egress] == "true" {
|
|
ip := net.ParseIP(*lastKnownGoodEndpoint)
|
|
|
|
// Store old values from ServiceSnapshot to detect if annotation actually changed
|
|
// We use the ServiceSnapshot instead of the service parameter because the service parameter
|
|
// may have stale annotations if the last update failed
|
|
var oldEndpoint, oldEndpointIPv6 string
|
|
if p.instances != nil {
|
|
serviceInstance := instance.FindServiceInstance(service, *p.instances)
|
|
if serviceInstance != nil {
|
|
oldEndpoint = serviceInstance.ServiceSnapshot.Annotations[kubevip.ActiveEndpoint]
|
|
oldEndpointIPv6 = serviceInstance.ServiceSnapshot.Annotations[kubevip.ActiveEndpointIPv6]
|
|
}
|
|
}
|
|
// Fall back to service annotations if we couldn't find the instance
|
|
if oldEndpoint == "" && oldEndpointIPv6 == "" {
|
|
oldEndpoint = service.Annotations[kubevip.ActiveEndpoint]
|
|
oldEndpointIPv6 = service.Annotations[kubevip.ActiveEndpointIPv6]
|
|
}
|
|
|
|
// Determine which annotation to update based on IP version
|
|
var endpoint, endpointIPv6 string
|
|
if ip.To4() == nil && !p.config.EnableEndpoints {
|
|
// IPv6
|
|
endpointIPv6 = *lastKnownGoodEndpoint
|
|
endpoint = oldEndpoint // Preserve existing IPv4 if any
|
|
} else {
|
|
// IPv4
|
|
endpoint = *lastKnownGoodEndpoint
|
|
endpointIPv6 = oldEndpointIPv6 // Preserve existing IPv6 if any
|
|
}
|
|
|
|
// Check if annotation actually changed
|
|
annotationChanged := (oldEndpoint != endpoint) || (oldEndpointIPv6 != endpointIPv6)
|
|
if !annotationChanged {
|
|
return // Nothing to do
|
|
}
|
|
|
|
// Persist to Kubernetes
|
|
ctx := context.Background()
|
|
|
|
if err := p.provider.UpdateServiceAnnotation(ctx, endpoint, endpointIPv6, service, clientSet); err != nil {
|
|
log.Warn("failed to update service annotation", "service", service.Name, "namespace", service.Namespace, "err", err)
|
|
return
|
|
}
|
|
|
|
log.Debug("updated active endpoint annotation", "service", service.Name, "namespace", service.Namespace, "endpoint", *lastKnownGoodEndpoint)
|
|
|
|
// Trigger egress reconfiguration
|
|
// For services with leader election, the service watcher doesn't process Modified events
|
|
// after initial setup, so we need to directly call the update function
|
|
if egressUpdateFunc != nil {
|
|
// Create a copy of service with updated annotations
|
|
svcCopy := service.DeepCopy()
|
|
svcCopy.Annotations[kubevip.ActiveEndpoint] = endpoint
|
|
svcCopy.Annotations[kubevip.ActiveEndpointIPv6] = endpointIPv6
|
|
|
|
if err := egressUpdateFunc(ctx, svcCopy); err != nil {
|
|
log.Error("failed to reconfigure egress", "service", service.Name, "namespace", service.Namespace, "err", err)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (p *Processor) startServiceHandlingIfNeeded(svcCtx *servicecontext.Context, service *v1.Service,
|
|
serviceFunc func(*servicecontext.Context, *v1.Service, *sync.WaitGroup, bool) error, wg *sync.WaitGroup) error {
|
|
if p.config.EnableServicesElection {
|
|
// startLeaderElection restarts itself until the service context is cancelled,
|
|
// so start it only once instead of on every endpoint event.
|
|
svcCtx.StartLeaderElectionOnce(func() {
|
|
wg.Go(func() {
|
|
p.startLeaderElection(svcCtx, service, serviceFunc, wg)
|
|
})
|
|
})
|
|
return nil
|
|
}
|
|
|
|
if p.config.EnableARP || (p.config.EnableRoutingTable && p.config.EnableLeaderElection) {
|
|
if !svcCtx.Signalled.Load() {
|
|
inst := instance.FindServiceInstance(service, *p.instances)
|
|
if inst == nil {
|
|
return fmt.Errorf("[%s] failed to find an instance for service %s/%s", p.provider.GetLabel(), service.Namespace, service.Name)
|
|
}
|
|
for x := range inst.VIPConfigs {
|
|
log.Debug("starting loadbalancer for service", "provider", p.provider.GetLabel(), "name", service.Name, "namespace", service.Namespace, "uid", service.UID)
|
|
if err := inst.Clusters[x].StartLoadBalancerService(svcCtx.Ctx, inst.VIPConfigs[x], p.bgpServer, lease.ServiceNamespacedName(service), wg); err != nil {
|
|
return fmt.Errorf("failed to start lb: %w", err)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (p *Processor) startLeaderElection(svcCtx *servicecontext.Context, service *v1.Service, serviceFunc func(*servicecontext.Context, *v1.Service, *sync.WaitGroup, bool) error, wg *sync.WaitGroup) {
|
|
// Track this loop for the lifetime of the goroutine. There has to be at most
|
|
// one per service, so a value above 1 means loops leaked.
|
|
loops := metrics.ServiceElectionLoops.WithLabelValues(service.Namespace, service.Name)
|
|
loops.Inc()
|
|
defer loops.Dec()
|
|
|
|
attempts := metrics.ServiceElectionAttemptsTotal.WithLabelValues(service.Namespace, service.Name)
|
|
|
|
// This is a blocking function, that will restart (in the event of failure)
|
|
for {
|
|
select {
|
|
case <-svcCtx.Ctx.Done():
|
|
return
|
|
default:
|
|
leaseNamespace, serviceLease := lease.ServiceName(service)
|
|
id := lease.NewID(p.config.LeaderElectionType, leaseNamespace, serviceLease)
|
|
// The lease is retired once its last service is gone, so an absent one means
|
|
// this loop has nothing left to elect for.
|
|
l := p.leaseMgr.Get(id)
|
|
if l == nil {
|
|
return
|
|
}
|
|
l.Lock()
|
|
|
|
if !l.Elected.Load() {
|
|
l.Unlock()
|
|
attempts.Inc()
|
|
err := serviceFunc(svcCtx, service, wg, true)
|
|
if err != nil {
|
|
log.Error(err.Error())
|
|
}
|
|
} else {
|
|
l.Unlock()
|
|
time.Sleep(time.Millisecond * 200)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func shouldAllowReconcileWithoutEndpoints(service *v1.Service) bool {
|
|
if service == nil || service.Spec.ExternalTrafficPolicy != v1.ServiceExternalTrafficPolicyTypeCluster {
|
|
return false
|
|
}
|
|
|
|
return strings.EqualFold(service.Annotations[kubevip.AllowReconcileWithoutEndpoints], "true")
|
|
}
|
|
|
|
func hasV6(endpoints []string) bool {
|
|
for _, e := range endpoints {
|
|
ip := net.ParseIP(e)
|
|
if ip != nil {
|
|
if ip.To4() == nil {
|
|
return true
|
|
}
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func getEndpoint(endpoints []string, family string) string {
|
|
for _, e := range endpoints {
|
|
ip := net.ParseIP(e)
|
|
if family == utils.IPv4Family && ip.To4() != nil {
|
|
return e
|
|
}
|
|
if family == utils.IPv6Family && ip.To4() == nil {
|
|
return e
|
|
}
|
|
}
|
|
return ""
|
|
}
|