kube-vip/pkg/manager/manager.go

package manager

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"os"
	"os/signal"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"syscall"
	"time"

	log "log/slog"

	"github.com/kube-vip/kube-vip/pkg/arp"
	"github.com/kube-vip/kube-vip/pkg/bgp"
	"github.com/kube-vip/kube-vip/pkg/cluster"
	"github.com/kube-vip/kube-vip/pkg/election"
	"github.com/kube-vip/kube-vip/pkg/iptables"
	"github.com/kube-vip/kube-vip/pkg/k8s"
	"github.com/kube-vip/kube-vip/pkg/kubevip"
	"github.com/kube-vip/kube-vip/pkg/lease"
	"github.com/kube-vip/kube-vip/pkg/manager/worker"
	"github.com/kube-vip/kube-vip/pkg/networkinterface"
	"github.com/kube-vip/kube-vip/pkg/nftables"
	"github.com/kube-vip/kube-vip/pkg/node"
	"github.com/kube-vip/kube-vip/pkg/route"
	"github.com/kube-vip/kube-vip/pkg/services"
	"github.com/kube-vip/kube-vip/pkg/upnp"
	"github.com/kube-vip/kube-vip/pkg/utils"
	"github.com/kube-vip/kube-vip/pkg/vip"
	"github.com/prometheus/client_golang/prometheus"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
)

// Manager degines the manager of the load-balancing services
type Manager struct {
	clientSet   *kubernetes.Clientset
	rwClientSet *kubernetes.Clientset
	configMap   string
	config      *kubevip.Config

	// Manager services
	// service bool

	// BGP Manager, this is a singleton that manages all BGP advertisements
	bgpServer *bgp.Server

	// This channel is used to catch an OS signal and trigger a shutdown
	signalChan chan os.Signal

	sigint sync.Once

	svcProcessor *services.Processor

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

	// This mutex is to protect calls from various goroutines
	mutex sync.Mutex

	// This tracks used network interfaces and guards them with mutex for concurrent changes.
	intfMgr *networkinterface.Manager

	// This tracks VIPs and performs ARP/NDP advertisement.
	arpMgr *arp.Manager

	// This tracks node labels and performs label management
	// implementation will be decided in constructor
	// based on config.EnableNodeLabeling
	nodeLabelManager node.LabelManager

	// This variable reports if manager is being closed
	closing atomic.Bool

	// Will be used for leaderelection when required
	electionMgr *election.Manager

	// Will handle leases
	leaseMgr *lease.Manager

	// Will handle routes
	routeMgr *route.Manager
}

// New will create a new managing object
func New(ctx context.Context, configMap string, config *kubevip.Config) (*Manager, error) {

	// Instance identity should be the same as k8s node name to ensure better compatibility.
	// By default k8s sets node name to `hostname -s`,
	// so if node name is not provided in the config,
	// we set it to hostname as a fallback.
	// This mimics legacy behavior and should work on old kube-vip installations.
	if config.NodeName == "" {
		log.Warn("Node name is missing from the config, fall back to hostname")
		hostname, err := os.Hostname()
		if err != nil {
			return nil, fmt.Errorf("could not get hostname: %v", err)
		}
		config.NodeName = hostname
	}
	config.NodeName = normalizeNodeName(config.NodeName)
	log.Info("using node name", "name", config.NodeName)

	adminConfigPath := "/etc/kubernetes/admin.conf"
	homeConfigPath := filepath.Join(os.Getenv("HOME"), ".kube", "config")

	var clientset *kubernetes.Clientset
	var clientConfig *rest.Config
	var err error

	switch {
	case config.LeaderElectionType == "etcd":
		// Do nothing, we don't construct a k8s client for etcd leader election
	case config.K8sConfigFile != "" && utils.FileExists(config.K8sConfigFile):
		// An explicitly configured kubeconfig (k8s_config_file env or
		// --k8sConfigPath) takes precedence over the well-known host paths.
		// KubernetesAddr, when set, overrides the API endpoint - static pods
		// on control plane hosts use it to reach their local API server
		// instead of a VIP that may not be up yet.
		clientConfig, err = k8s.NewRestConfig(config.K8sConfigFile, false, config.KubernetesAddr)
		if err != nil {
			return nil, fmt.Errorf("could not create k8s REST config from file %q: %w", config.K8sConfigFile, err)
		}
		if clientset, err = k8s.NewClientset(clientConfig); err != nil {
			return nil, fmt.Errorf("could not create k8s clientset: %w", err)
		}
		log.Info("Using Kubernetes configuration from explicit file", "path", config.K8sConfigFile, "address", config.KubernetesAddr)
	case utils.FileExists(adminConfigPath):
		if config.KubernetesAddr != "" {
			log.Info("k8s address", "address", config.KubernetesAddr)
			clientConfig, err = k8s.NewRestConfig(adminConfigPath, false, config.KubernetesAddr)
		} else if config.EnableControlPlane {
			// If this is a control plane host it will likely have started as a static pod or won't have the
			// VIP up before trying to connect to the API server, we set the API endpoint to this machine to
			// ensure connectivity.
			if config.DetectControlPlane {
				clientConfig, err = k8s.FindWorkingKubernetesAddress(adminConfigPath, false)
			} else {
				// This will attempt to use kubernetes as the hostname (this should be passed as a host alias) in the pod manifest
				clientConfig, err = k8s.NewRestConfig(adminConfigPath, false, fmt.Sprintf("kubernetes:%v", config.Port))
			}
		} else {
			clientConfig, err = k8s.NewRestConfig(adminConfigPath, false, "")
		}
		if err != nil {
			return nil, fmt.Errorf("could not create k8s REST config from external file: %q: %w", adminConfigPath, err)
		}
		if clientset, err = k8s.NewClientset(clientConfig); err != nil {
			return nil, fmt.Errorf("could not create k8s clientset: %w", err)
		}
		log.Debug("Using external Kubernetes configuration from file", "path", adminConfigPath)
	case utils.FileExists(homeConfigPath):
		clientConfig, err = k8s.NewRestConfig(homeConfigPath, false, "")
		if err != nil {
			return nil, fmt.Errorf("could not create k8s REST config from external file: %q: %w", homeConfigPath, err)
		}
		clientset, err = k8s.NewClientset(clientConfig)
		if err != nil {
			return nil, fmt.Errorf("could not create k8s clientset from external file: %q: %w", homeConfigPath, err)
		}
		log.Debug("Using external Kubernetes configuration from file", "path", adminConfigPath)
	default:
		clientConfig, err = k8s.NewRestConfig("", true, "")
		if err != nil {
			return nil, fmt.Errorf("could not create k8s REST config from incluster file: %q: %w", homeConfigPath, err)
		}
		clientset, err = k8s.NewClientset(clientConfig)
		if err != nil {
			return nil, fmt.Errorf("could not create k8s clientset from incluster config: %w", err)
		}
		log.Debug("Using external Kubernetes configuration from incluster config.")
	}

	var rwClientSet *kubernetes.Clientset
	// if clientConfig is not nil, then we are not using etcd leader election
	// we need to create non-timeout clientset for RetryWatcher
	if clientConfig != nil {
		rwConfig := *clientConfig
		rwConfig.Timeout = 0
		rwClientSet, err = k8s.NewClientset(&rwConfig)
		if err != nil {
			return nil, fmt.Errorf("could not create k8s clientset for retry watcher: %w", err)
		}
	}

	// Flip this to something else
	// if config.DetectControlPlane {
	// 	log.Info("[k8s client] flipping to internal service account")
	// 	_, err = clientset.CoreV1().ServiceAccounts("kube-system").Apply(context.TODO(), kubevip.GenerateSA(), v1.ApplyOptions{FieldManager: "application/apply-patch"})
	// 	if err != nil {
	// 		return nil, fmt.Errorf("could not create k8s clientset from incluster config: %v", err)
	// 	}
	// 	_, err = clientset.RbacV1().ClusterRoles().Apply(context.TODO(), kubevip.GenerateCR(), v1.ApplyOptions{FieldManager: "application/apply-patch"})
	// 	if err != nil {
	// 		return nil, fmt.Errorf("could not create k8s clientset from incluster config: %v", err)
	// 	}
	// 	_, err = clientset.RbacV1().ClusterRoleBindings().Apply(context.TODO(), kubevip.GenerateCRB(), v1.ApplyOptions{FieldManager: "application/apply-patch"})
	// 	if err != nil {
	// 		return nil, fmt.Errorf("could not create k8s clientset from incluster config: %v", err)
	// 	}
	// }

	// listen for interrupts or the Linux SIGTERM signal and cancel
	// our context, which the leader election code will observe and
	// step down
	signalChan := make(chan os.Signal, 1)
	// Add Notification for Userland interrupt
	signal.Notify(signalChan, syscall.SIGINT)

	// Add Notification for SIGTERM (sent from Kubernetes)
	signal.Notify(signalChan, syscall.SIGTERM)

	// Add Notification for SIGUSR1 (for configuration dump)
	signal.Notify(signalChan, syscall.SIGUSR1)

	intfMgr := networkinterface.NewManager()
	arpMgr := arp.NewManager(config)

	// create the node label manager
	// constructor will decide if it should be a noop or not
	nodeLabelManager := node.NewManager(config, clientset)

	var bgpServer *bgp.Server
	// If BGP is enabled then we start a server instance that will broadcast VIPs
	if config.EnableBGP {
		var err error
		// If Annotations have been set then we will look them up
		if config.Annotations != "" {
			err = annotationsWatcher(ctx, clientset, rwClientSet, config)
		} else {
			log.Debug("No Node annotations to parse")
		}

		if err != nil {
			return nil, err
		}
		bgpServer, err = bgp.NewBGPServer(config.BGPConfig, log.Level(config.Logging))
		if err != nil {
			return nil, fmt.Errorf("creating BGP server: %w", err)
		}
	}

	electionMgr, err := election.NewManager(config, clientset, rwClientSet)
	if err != nil {
		return nil, fmt.Errorf("creating election manager: %w", err)
	}

	leaseMgr := lease.NewManager()
	routeMgr := route.NewManager()

	svcProcessor := services.NewServicesProcessor(config, bgpServer, clientset, rwClientSet,
		intfMgr, arpMgr, nodeLabelManager, electionMgr, leaseMgr, routeMgr)

	return &Manager{
		clientSet:   clientset,
		rwClientSet: rwClientSet,
		configMap:   configMap,
		config:      config,
		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"}),
		signalChan:       signalChan,
		svcProcessor:     svcProcessor,
		intfMgr:          intfMgr,
		arpMgr:           arpMgr,
		bgpServer:        bgpServer,
		nodeLabelManager: nodeLabelManager,
		electionMgr:      electionMgr,
		leaseMgr:         leaseMgr,
		routeMgr:         routeMgr,
	}, nil
}

// Start will begin the Manager, which will start services and watch the configmap
func (sm *Manager) Start(ctx context.Context) error {
	wg := sync.WaitGroup{}
	defer wg.Wait()

	// HealthCheck
	if sm.config.HealthCheckPort != 0 {
		if sm.config.HealthCheckPort < 1024 {
			return fmt.Errorf("healthcheck port is using a port that is less than 1024 [%d]", sm.config.HealthCheckPort)
		}
		http.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) {
			fmt.Fprintf(w, "OK")
		})
		wg.Go(func() {
			server := &http.Server{
				Addr:              fmt.Sprintf(":%d", sm.config.HealthCheckPort),
				ReadHeaderTimeout: 3 * time.Second,
			}
			err := server.ListenAndServe()
			if err != nil {
				log.Error("healthcheck", "unable to start", err)
			}
		})
	}

	// on exit, clean up the node labels
	defer func() {
		if err := sm.nodeLabelManager.CleanUpLabels(10 * time.Second); err != nil {
			log.Error("CleanUpNodeLabels", "unable to cleanup node labels", err)
		}
	}()

	if sm.config.EnableARP || sm.config.EnableWireguard {
		if sm.config.EnableUPNP {
			clients := upnp.GetConnectionClients(ctx)
			if len(clients) == 0 {
				log.Error("Error Enabling UPNP. No Clients found")
				// Set the struct to false so nothing should use it in future
				sm.config.EnableUPNP = false
			} else {
				for _, c := range clients {
					ip, err := c.GetExternalIPAddress()
					if err != nil {
						log.Error("unable to find IGD2 Gateway address", "err", err)
					}
					log.Info("Found UPNP IGD2 Gateway address", "ip", ip)
				}
			}
			// TODO: It would be nice to run the UPNP refresh only on the leader.
			wg.Go(func() {
				sm.svcProcessor.RefreshUPNPForwards(ctx)
			})
		}
	}

	return sm.startMode(ctx)
}

// Start will begin the Manager, which will start services and watch the configmap
func (sm *Manager) startMode(ctx context.Context) error {
	var cpCluster *cluster.Cluster
	var err error

	w := worker.New(sm.arpMgr, sm.intfMgr, sm.config, &sm.closing, sm.Kill,
		sm.svcProcessor, &sm.mutex, sm.clientSet, sm.bgpServer, sm.electionMgr,
		sm.leaseMgr, sm.routeMgr, sm.nodeLabelManager)

	// use a Go context so we can tell the leaderelection code when we
	// want to step down
	wg := sync.WaitGroup{}
	modeCtx, cancel := context.WithCancel(ctx)
	defer func() {

		wg.Wait()
		w.Cleanup()
		cancel()
		log.Info("Shutting down Kube-Vip")
	}()

	log.Info("starting Kube-vip Manager", "mode", w.Name())
	if err := w.Configure(modeCtx, &wg); err != nil {
		defer cancel()
		return fmt.Errorf("failed to configure %s mode: %w", w.Name(), err)
	}

	if sm.config.EnableControlPlane {
		err = w.InitControlPlane()
		if err != nil {
			defer cancel()
			return err
		}
	}

	// Shutdown function that will wait on this signal, unless we call it ourselves
	wg.Go(func() {
		sm.waitForShutdown(modeCtx, cancel, cpCluster)
	})

	if sm.config.EnableControlPlane {
		wg.Go(func() {
			w.StartControlPlane(modeCtx, sm.electionMgr)
		})
	}

	if sm.config.EnableServices {
		// This will tidy any dangling kube-vip iptables rules
		if sm.config.EgressClean {
			tableName := nftables.EgressTableBaseNameForInstance(sm.config.InstanceName)
			err := nftables.ClearTablesWithName(tableName)
			if err != nil {
				log.Warn("[egress]", "mode", "nftables-internal", "clearing error", err)
			} else {
				log.Info("[egress]", "mode", "nftables-internal", "tables", "cleared")
			}
			// TODO: Deprecate the iptables code v1.2.x
			err = vip.ClearIPTables(sm.config.EgressWithNftables, sm.config.ServiceNamespace, iptables.ProtocolIPv4)
			if err != nil {
				log.Info("[egress]", "legacy-iptables", sm.config.EgressWithNftables, "mode", "IPv4", "error", err)
			}
			err = vip.ClearIPTables(sm.config.EgressWithNftables, sm.config.ServiceNamespace, iptables.ProtocolIPv6)
			if err != nil {
				log.Info("[egress]", "legacy-iptables", sm.config.EgressWithNftables, "mode", "IPv6", "error", err)
			}
		}
		w.ConfigureServices()

		for {
			select {
			case <-modeCtx.Done():
				return nil
			default:
				if err = w.StartServices(modeCtx); err != nil {
					if errors.Is(err, &utils.PanicError{}) {
						sm.Kill()
						return fmt.Errorf("failed to reconcile services, non-recoverable error: %w", err)
					} else {
						log.Error("failed to reconcile services, restarting", "error", err)
					}
				}
			}
		}
	}

	return nil
}

func (sm *Manager) waitForShutdown(ctx context.Context, cancel context.CancelFunc, cpCluster *cluster.Cluster) {
	for {
		sig := <-sm.signalChan
		switch sig {
		case syscall.SIGUSR1:
			log.Info("Received SIGUSR1, dumping configuration")
			sm.dumpConfiguration(ctx)
		case syscall.SIGINT, syscall.SIGTERM:
			sm.closing.Store(true)
			log.Info("Received kube-vip termination, signaling shutdown")
			if cpCluster != nil {
				cpCluster.Stop()
			}
			// Cancel the context, which will in turn cancel the leadership and all goroutines
			cancel()
			return
		}
	}
}

func (sm *Manager) Kill() {
	sm.sigint.Do(func() {
		sm.signalChan <- syscall.SIGINT
	})
}

// normalizeNodeName ensures the local machine hostname conforms to
// Kubernetes RFC1123 node naming conventions (lowercase).
func normalizeNodeName(name string) string {
	return strings.ToLower(name)
}