diff --git a/cmd/kube-vip.go b/cmd/kube-vip.go index e6037cc3..0c3d1385 100644 --- a/cmd/kube-vip.go +++ b/cmd/kube-vip.go @@ -52,8 +52,9 @@ var ( ) var kubeVipCmd = &cobra.Command{ - Use: "kube-vip", - Short: "This is a server for providing a Virtual IP and load-balancer for the Kubernetes control-plane", + Use: "kube-vip", + Short: "This is a server for providing a Virtual IP and load-balancer for the Kubernetes control-plane", + SilenceErrors: true, } func init() { @@ -187,11 +188,16 @@ func init() { } // Execute - starts the command parsing process -func Execute() { - if err := kubeVipCmd.Execute(); err != nil { - fmt.Println(err) - os.Exit(1) +func Execute() int { + cmd, err := kubeVipCmd.ExecuteC() + if err != nil { + log.Error("command failed", "err", err) + if cmd == kubeVipCmd { + _ = cmd.Usage() + } + return 1 } + return 0 } var kubeVipVersion = &cobra.Command{ @@ -215,26 +221,24 @@ var kubeVipSample = &cobra.Command{ var kubeVipService = &cobra.Command{ Use: "service", Short: "Start the Virtual IP / Load balancer as a service within a Kubernetes cluster", - Run: func(cmd *cobra.Command, args []string) { //nolint TODO + RunE: func(cmd *cobra.Command, args []string) error { //nolint TODO + cmd.SilenceUsage = true // Load configuration from file if specified (lowest priority) if initConfig.ConfigFile != "" { err := kubevip.MergeConfigFromFile(&initConfig, initConfig.ConfigFile) if err != nil { - log.Error("loading config file", "err", err) - return + return fmt.Errorf("loading config file: %w", err) } } // parse environment variables, these will overwrite anything loaded from config file err := kubevip.ParseEnvironment(&initConfig) if err != nil { - log.Error("parsing env", "err", err) - return + return fmt.Errorf("parsing environment: %w", err) } if err := initConfig.Validate(); err != nil { - log.Error("validating configuration", "err", err) - return + return fmt.Errorf("validating configuration: %w", err) } // Change RTN_UNSPEC to default type @@ -246,8 +250,7 @@ var kubeVipService = &cobra.Command{ log.SetLogLoggerLevel(log.Level(initConfig.Logging)) if err := initConfig.CheckInterface(); err != nil { - log.Error("checking interface", "err", err) - return + return fmt.Errorf("checking interface: %w", err) } // User Environment variables as an option to make manifest clearer @@ -260,8 +263,7 @@ var kubeVipService = &cobra.Command{ if initConfig.EnableControlPlane && (initConfig.EnableARP || initConfig.EnableBGP || initConfig.EnableRoutingTable) { if err := initConfig.CheckSubnetExists(); err != nil { - log.Error("checking subnet exists if vip_address defined", "err", err) - return + return fmt.Errorf("checking subnet exists if vip_address defined: %w", err) } } @@ -269,8 +271,7 @@ var kubeVipService = &cobra.Command{ if initConfig.VIPSubnet == "" && initConfig.Address != "" { initConfig.VIPSubnet, err = GenerateCidrRange(initConfig.Address, initConfig.DNSMode) if err != nil { - log.Error("generating CIDR", "err", err) - return + return fmt.Errorf("generating CIDR: %w", err) } } @@ -280,41 +281,39 @@ var kubeVipService = &cobra.Command{ // Define the new service manager mgr, err := manager.New(ctx, configMap, &initConfig) if err != nil { - log.Error("new manager", "err", err) - return + return fmt.Errorf("new manager: %w", err) } // Start the service manager, this will watch the config Map and construct kube-vip services for it err = mgr.Start(ctx) if err != nil { - log.Error("manager start", "err", err) - return + return fmt.Errorf("manager start: %w", err) } + return nil }, } var kubeVipManager = &cobra.Command{ Use: "manager", Short: "Start the kube-vip manager", - Run: func(cmd *cobra.Command, args []string) { //nolint TODO + RunE: func(cmd *cobra.Command, args []string) error { //nolint TODO + cmd.SilenceUsage = true + // Load configuration from file if specified (lowest priority) if initConfig.ConfigFile != "" { err := kubevip.MergeConfigFromFile(&initConfig, initConfig.ConfigFile) if err != nil { - log.Error("loading config file", "err", err) - return + return fmt.Errorf("loading config file: %w", err) } } // parse environment variables, these will overwrite anything loaded from config file err := kubevip.ParseEnvironment(&initConfig) if err != nil { - log.Error("parsing environment", "err", err) - return + return fmt.Errorf("parsing environment: %w", err) } if err := initConfig.Validate(); err != nil { - log.Error("validating configuration", "err", err) - return + return fmt.Errorf("validating configuration: %w", err) } // Change RTN_UNSPEC to default type @@ -329,8 +328,7 @@ var kubeVipManager = &cobra.Command{ if initConfig.EnableControlPlane && (initConfig.EnableARP || initConfig.EnableBGP || initConfig.EnableRoutingTable) { if err := initConfig.CheckSubnetExists(); err != nil { - log.Error("checking subnet exists if vip_address defined", "err", err) - return + return fmt.Errorf("checking subnet exists if vip_address defined: %w", err) } } @@ -338,8 +336,7 @@ var kubeVipManager = &cobra.Command{ if initConfig.VIPSubnet == "" && initConfig.Address != "" { initConfig.VIPSubnet, err = GenerateCidrRange(initConfig.Address, initConfig.DNSMode) if err != nil { - log.Error("No interface is specified for kube-vip to bind to") - return + return fmt.Errorf("generating CIDR: %w", err) } } @@ -388,13 +385,11 @@ var kubeVipManager = &cobra.Command{ } if mode == "" { - log.Error("no valid kube-vip mode detected, ensure a supported mode is configured") - return + return fmt.Errorf("no valid kube-vip mode detected, ensure a supported mode is configured") } if modesEnabled > 1 { - log.Error("multiple kube-vip modes detected, ensure only one mode is configured") - return + return fmt.Errorf("multiple kube-vip modes detected, ensure only one mode is configured") } // Provide configuration to output/logging @@ -402,18 +397,15 @@ var kubeVipManager = &cobra.Command{ // End if nothing is enabled if !initConfig.EnableServices && !initConfig.EnableControlPlane { - log.Error("no features are enabled") - return + return fmt.Errorf("no features are enabled") } if !initConfig.EnableARP && strings.Contains(initConfig.VIPSubnet, kubevip.Auto) { - log.Error("auto subnet discovery cannot be used outside ARP mode") - return + return fmt.Errorf("auto subnet discovery cannot be used outside ARP mode") } if strings.Contains(initConfig.VIPSubnet, kubevip.Auto) && initConfig.Address != "" { - log.Error("auto subnet discovery cannot be used if VIP address was provided") - return + return fmt.Errorf("auto subnet discovery cannot be used if VIP address was provided") } // If we're using wireguard then all traffic goes through the wg0 interface @@ -430,20 +422,17 @@ var kubeVipManager = &cobra.Command{ log.Warn("attempting to create wireguard interface", "interface not found", initConfig.Interface) err = netlink.LinkAdd(&netlink.Wireguard{LinkAttrs: netlink.LinkAttrs{Name: initConfig.Interface}}) if err != nil { - log.Error("adding link", "err", err) - return + return fmt.Errorf("adding link: %w", err) } l, err = netlink.LinkByName(initConfig.Interface) if err != nil { - log.Error("finding link", "err", err) - return + return fmt.Errorf("finding link: %w", err) } } } err = netlink.LinkSetUp(l) if err != nil { - log.Error("setting link UP", "err", err) - return + return fmt.Errorf("setting link UP: %w", err) } } else { // if we're not using Wireguard then we'll need to use an actual interface @@ -453,8 +442,7 @@ var kubeVipManager = &cobra.Command{ defaultIF, err := vip.GetDefaultGatewayInterface() if err != nil { _ = cmd.Help() - log.Error("detecting interface", "err", err) - return + return fmt.Errorf("detecting interface: %w", err) } initConfig.Interface = defaultIF.Name log.Info("kube-vip bind", "interface", initConfig.Interface) @@ -470,8 +458,7 @@ var kubeVipManager = &cobra.Command{ } // Perform a check on the state of the interface if err := initConfig.CheckInterface(); err != nil { - log.Error("checking interface", "err", err) - return + return fmt.Errorf("checking interface: %w", err) } // User Environment variables as an option to make manifest clearer @@ -483,8 +470,7 @@ var kubeVipManager = &cobra.Command{ // Define the new service manager mgr, err := manager.New(ctx, configMap, &initConfig) if err != nil { - log.Error("new manager", "err", err) - return + return fmt.Errorf("new manager: %w", err) } metrics.RegisterPrometheusMetrics() @@ -493,9 +479,9 @@ var kubeVipManager = &cobra.Command{ // Start the service manager, this will watch the config Map and construct kube-vip services for it err = mgr.Start(ctx) if err != nil { - log.Error("start manager", "err", err) - return + return fmt.Errorf("start manager: %w", err) } + return nil }, } diff --git a/go.mod b/go.mod index a6e0f7bf..66d1e594 100644 --- a/go.mod +++ b/go.mod @@ -5,7 +5,6 @@ go 1.26.4 require ( github.com/cloudflare/ipvs v0.12.0 github.com/containernetworking/plugins v1.9.1 - github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc github.com/docker/docker v28.5.2+incompatible github.com/florianl/go-conntrack v0.7.0 github.com/google/go-cmp v0.7.0 @@ -54,6 +53,7 @@ require ( github.com/containerd/log v0.1.0 // indirect github.com/coreos/go-semver v0.3.1 // indirect github.com/coreos/go-systemd/v22 v22.7.0 // indirect + github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/dgryski/go-farm v0.0.0-20240924180020-3414d57e47da // indirect github.com/distribution/reference v0.6.0 // indirect github.com/docker/go-connections v0.7.0 // indirect diff --git a/main.go b/main.go index 2d66f9cd..8d90dc5d 100644 --- a/main.go +++ b/main.go @@ -1,6 +1,8 @@ package main import ( + "os" + "github.com/kube-vip/kube-vip/cmd" ) @@ -14,5 +16,5 @@ func main() { cmd.Release.Version = Version cmd.Release.Build = Build - cmd.Execute() + os.Exit(cmd.Execute()) } diff --git a/pkg/cluster/service.go b/pkg/cluster/service.go index 0c2a7086..39fd25cf 100644 --- a/pkg/cluster/service.go +++ b/pkg/cluster/service.go @@ -113,7 +113,7 @@ func (cluster *Cluster) StartVipService(ctx context.Context, c *kubevip.Config, err = em.NodeWatcher(ctx, lb, c.Port) if err != nil { log.Error("Error watching node labels", "err", err) - if errors.Is(err, &utils.PanicError{}) { + if utils.IsPanicError(err) { killFunc() return } @@ -429,7 +429,7 @@ func (cluster *Cluster) StartLoadBalancerService(ctx context.Context, c *kubevip if err := network.SetMask(c.VIPSubnet); err != nil { log.Error("failed to set mask", "subnet", c.VIPSubnet, "err", err) lbCancel() - return utils.NewPanicError(fmt.Sprintf("failed to set mask for subnet %q: %s", c.VIPSubnet, err.Error())) + return utils.WrapPanicError(err, "failed to set mask for subnet %q", c.VIPSubnet) } _, err := network.DeleteIP() if err != nil { diff --git a/pkg/election/election.go b/pkg/election/election.go index 3e713f01..d019be8c 100644 --- a/pkg/election/election.go +++ b/pkg/election/election.go @@ -2,14 +2,12 @@ package election import ( "context" - "errors" "fmt" "sync" "time" log "log/slog" - "github.com/davecgh/go-spew/spew" "github.com/kube-vip/kube-vip/pkg/etcd" "github.com/kube-vip/kube-vip/pkg/kubevip" "github.com/kube-vip/kube-vip/pkg/lease" @@ -17,7 +15,6 @@ import ( "github.com/kube-vip/kube-vip/pkg/utils" clientv3 "go.etcd.io/etcd/client/v3" v1 "k8s.io/api/core/v1" - apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/watch" "k8s.io/client-go/kubernetes" @@ -199,7 +196,7 @@ func (em *Manager) NodeWatcher(ctx context.Context, lb *loadbalancer.IPVSLoadBal err = lb.AddBackend(node.Status.Addresses[x].Address, port) if err != nil { log.Error("adding node to load balancer", "node", node.Name, "ip", node.Status.Addresses[x].Address, "err", err) - if errors.Is(err, &utils.PanicError{}) { + if utils.IsPanicError(err) { return fmt.Errorf("add IPVS backend: %w", err) } } @@ -233,23 +230,20 @@ func (em *Manager) NodeWatcher(ctx context.Context, lb *loadbalancer.IPVSLoadBal // Un-used case watch.Error: log.Error("Error attempting to watch Kubernetes Nodes") - - // This round trip allows us to handle unstructured status - errObject := apierrors.FromObject(event.Object) - statusErr, ok := errObject.(*apierrors.StatusError) - if !ok { - log.Error(spew.Sprintf("Received an error which is not *metav1.Status but %#+v", event.Object)) - } - - status := statusErr.ErrStatus - log.Error("watcher", "status", status) - watchErr = fmt.Errorf("node watcher error, status: %s", status.String()) + watchErr = fmt.Errorf("node watcher error: %w", utils.WatchError(event.Object)) + log.Error("watcher", "err", watchErr) default: } } log.Info("Exiting Node watcher") - return watchErr + if watchErr != nil { + return watchErr + } + if ctx.Err() != nil { + return nil + } + return utils.NewPanicError("node watcher channel closed unexpectedly") } func checkIfNodeIsReady(node *v1.Node) bool { diff --git a/pkg/loadbalancer/ipvs.go b/pkg/loadbalancer/ipvs.go index 403a4292..fd222938 100644 --- a/pkg/loadbalancer/ipvs.go +++ b/pkg/loadbalancer/ipvs.go @@ -240,7 +240,7 @@ func (lb *IPVSLoadBalancer) addBackend(address string, port uint16) error { // Fatal error at this point as IPVS is probably not working log.Error("Unable to create an IPVS service, ensure IPVS kernel modules are loaded") log.Error("IPVS service", "err", err) - return utils.NewPanicError(fmt.Sprintf("unable to create an IPVS service - %s", err)) + return utils.WrapPanicError(err, "unable to create an IPVS service") } log.Info("load-Balancer services created", "address", lb.addrString(), "port", lb.Port) diff --git a/pkg/manager/manager.go b/pkg/manager/manager.go index 6144733d..9802bbc0 100644 --- a/pkg/manager/manager.go +++ b/pkg/manager/manager.go @@ -2,7 +2,6 @@ package manager import ( "context" - "errors" "fmt" "net/http" "os" @@ -414,7 +413,7 @@ func (sm *Manager) startMode(ctx context.Context) error { return nil default: if err = w.StartServices(modeCtx); err != nil { - if errors.Is(err, &utils.PanicError{}) { + if utils.IsPanicError(err) { sm.Kill() return fmt.Errorf("failed to reconcile services, non-recoverable error: %w", err) } else { diff --git a/pkg/manager/watch_annotations.go b/pkg/manager/watch_annotations.go index d5c5c233..30e2b78a 100644 --- a/pkg/manager/watch_annotations.go +++ b/pkg/manager/watch_annotations.go @@ -11,10 +11,9 @@ import ( log "log/slog" "github.com/kube-vip/kube-vip/pkg/kubevip" + "github.com/kube-vip/kube-vip/pkg/utils" - "github.com/davecgh/go-spew/spew" v1 "k8s.io/api/core/v1" - apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" "k8s.io/client-go/kubernetes" @@ -108,22 +107,15 @@ func annotationsWatcher(ctx context.Context, clientSet, // Un-used case watch.Error: log.Error("Error attempting to watch Kubernetes Nodes") - - // This round trip allows us to handle unstructured status - errObject := apierrors.FromObject(event.Object) - statusErr, ok := errObject.(*apierrors.StatusError) - if !ok { - log.Error(spew.Sprintf("Received an error which is not *metav1.Status but %#+v", event.Object)) - - } - - status := statusErr.ErrStatus - log.Error(status.String()) + log.Error("annotations watcher failed", "err", utils.WatchError(event.Object)) default: } } log.Info("[annotations] exiting annotations watcher") - return nil + if ctx.Err() != nil { + return nil + } + return utils.NewPanicError("annotations watcher channel closed unexpectedly") } // parseNodeAnnotations parses the annotations on the node and updates the configuration diff --git a/pkg/services/leader.go b/pkg/services/leader.go index 7a1fbd21..6589e0dd 100644 --- a/pkg/services/leader.go +++ b/pkg/services/leader.go @@ -122,7 +122,7 @@ func (p *Processor) StartServicesLeaderElection(svcCtx *servicecontext.Context, // Block until service context is cancelled <-svcCtx.Ctx.Done() - if err := p.onStoppedLeading(svcLease, service); err != nil { + if err := p.onStoppedLeading(svcCtx, svcLease, service); err != nil { log.Error("error on stopped leading", "error", err) } @@ -159,7 +159,7 @@ func (p *Processor) StartServicesLeaderElection(svcCtx *servicecontext.Context, // we can do cleanup here svcLease.Elected.Store(false) log.Info("leadership lost", "service", service.Name, "uid", service.UID, "leader", p.config.NodeName) - if err := p.onStoppedLeading(svcLease, service); err != nil { + if err := p.onStoppedLeading(svcCtx, svcLease, service); err != nil { metrics.ServiceReconcileErrorsTotal.WithLabelValues(service.Namespace, service.Name, "delete_service").Inc() leaderCancel() } @@ -195,9 +195,18 @@ func (p *Processor) onStartedLeading(svcCtx *servicecontext.Context, service *v1 return nil } -func (p *Processor) onStoppedLeading(svcLease *lease.Lease, service *v1.Service) error { +func (p *Processor) onStoppedLeading(svcCtx *servicecontext.Context, svcLease *lease.Lease, service *v1.Service) error { + currentSvcCtx, err := p.getServiceContext(service.UID) + if err != nil { + return err + } + if currentSvcCtx != nil && currentSvcCtx != svcCtx { + log.Debug("skipping cleanup from superseded service context", "service", service.Name, "uid", service.UID) + return nil + } + log.Debug("deleting service due to lost leadership", "uid", service.UID) - err := p.deleteService(svcLease.Ctx, service.UID) + err = p.deleteService(svcLease.Ctx, service.UID) if err != nil { log.Error("service deletion", "err", err) return err diff --git a/pkg/services/processor.go b/pkg/services/processor.go index 47b9e589..b96a5395 100644 --- a/pkg/services/processor.go +++ b/pkg/services/processor.go @@ -2,7 +2,6 @@ package services import ( "context" - "errors" "fmt" log "log/slog" "reflect" @@ -91,7 +90,8 @@ func NewServicesProcessor(config *kubevip.Config, bgpServer *bgp.Server, } } -func (p *Processor) AddOrModify(ctx context.Context, event watch.Event, serviceFunc *Callback, forcedOnly bool, wg *sync.WaitGroup) error { +func (p *Processor) AddOrModify(ctx context.Context, event watch.Event, serviceFunc *Callback, forcedOnly bool, + wg *sync.WaitGroup, cancelWatcher context.CancelCauseFunc) error { svc, ok := event.Object.(*v1.Service) if !ok { return fmt.Errorf("unable to parse Kubernetes services from API watcher") @@ -142,16 +142,6 @@ func (p *Processor) AddOrModify(ctx context.Context, event watch.Event, serviceF } svcInstance := instance.FindServiceInstance(svc, p.ServiceInstances) - var err error - if svcInstance == nil { - svcInstance, err = instance.NewInstance(ctx, svc, p.config, p.intfMgr, p.arpMgr, p.routeMgr, p.nodeLabelManager, wg) - if err != nil { - metrics.ServiceReconcileErrorsTotal.WithLabelValues(svc.Namespace, svc.Name, "new_instance").Inc() - return fmt.Errorf("unable to create instance for service %s/%s", svc.Namespace, svc.Name) - } - p.ServiceInstances = append(p.ServiceInstances, svcInstance) - p.updateActiveServicesMetric() - } _, usesCommonLease := svc.Annotations[kubevip.ServiceLease] if usesCommonLease && svc.Spec.ExternalTrafficPolicy != v1.ServiceExternalTrafficPolicyTypeCluster { @@ -194,8 +184,6 @@ func (p *Processor) AddOrModify(ctx context.Context, event watch.Event, serviceF metrics.ServiceReconcileErrorsTotal.WithLabelValues(svc.Namespace, svc.Name, "delete_service").Inc() log.Error("(svc) unable to remove", "service", svc.UID) } - // in theory this should never fail - p.svcMap.Delete(svc.UID) // Retire the lease before the replacement context is built, so Add below // cannot hand back an instance the pending cleanup is about to cancel. // A lease shared with other services keeps their references and survives. @@ -205,6 +193,7 @@ func (p *Processor) AddOrModify(ctx context.Context, event watch.Event, serviceF // Reset the the svcCtx when it was garbage collected // As the next function will create a new context when nil svcCtx = nil + svcInstance = nil p.updateActiveServicesMetric() } } @@ -223,6 +212,16 @@ func (p *Processor) AddOrModify(ctx context.Context, event watch.Event, serviceF p.svcMap.Store(svc.UID, svcCtx) } + if svcInstance == nil { + svcInstance, err = instance.NewInstance(ctx, svc, p.config, p.intfMgr, p.arpMgr, p.routeMgr, p.nodeLabelManager, wg) + if err != nil { + metrics.ServiceReconcileErrorsTotal.WithLabelValues(svc.Namespace, svc.Name, "new_instance").Inc() + return fmt.Errorf("unable to create instance for service %s/%s", svc.Namespace, svc.Name) + } + p.ServiceInstances = append(p.ServiceInstances, svcInstance) + p.updateActiveServicesMetric() + } + // this goroutine starts service handling function (with or without leaderelection) if !svcCtx.IsWatched { wg.Go(func() { @@ -240,7 +239,7 @@ func (p *Processor) AddOrModify(ctx context.Context, event watch.Event, serviceF err = serviceFunc.Run(svcCtx, svc, wg) if err != nil { log.Error(err.Error()) - if errors.Is(err, &utils.PanicError{}) { + if utils.IsPanicError(err) { // cancel service context on panic error // TODO: should we quit kube-vip altogether here? svcCtx.Cancel() @@ -258,8 +257,11 @@ func (p *Processor) AddOrModify(ctx context.Context, event watch.Event, serviceF } else { provider = providers.NewEndpointslices() } - if err = p.watchEndpoint(svcCtx, p.config.NodeName, svc, provider); err != nil { - log.Error(err.Error()) + if err := p.watchEndpoint(svcCtx, p.config.NodeName, svc, provider, cancelWatcher); err != nil { + log.Error("endpoint watcher failed", "service", svc.Name, "namespace", svc.Namespace, "err", err) + if utils.IsPanicError(err) { + cancelWatcher(err) + } } }) diff --git a/pkg/services/processor_lease_test.go b/pkg/services/processor_lease_test.go index e6e5d0fe..9c311da1 100644 --- a/pkg/services/processor_lease_test.go +++ b/pkg/services/processor_lease_test.go @@ -4,6 +4,7 @@ import ( "context" "testing" + "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/servicecontext" @@ -117,3 +118,37 @@ func TestDropCancelledServiceContextAllowsLeaseRecreation(t *testing.T) { t.Fatal("expected a new lease to be created once the cancelled service context was dropped") } } + +func TestOnStoppedLeadingDoesNotDeleteReplacementContext(t *testing.T) { + p := &Processor{ + config: &kubevip.Config{}, + leaseMgr: lease.NewManager(), + } + + service := &v1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: "example", + Namespace: "default", + UID: types.UID("service-uid"), + }, + } + + oldCtx := servicecontext.New(context.Background()) + replacementCtx := servicecontext.New(context.Background()) + p.svcMap.Store(service.UID, replacementCtx) + replacementInstance := &instance.Instance{ServiceSnapshot: service.DeepCopy()} + p.ServiceInstances = []*instance.Instance{replacementInstance} + + leaseNamespace, serviceLease := lease.ServiceName(service) + svcLease := p.leaseMgr.Add(context.Background(), lease.NewID(p.config.LeaderElectionType, leaseNamespace, serviceLease)) + + if err := p.onStoppedLeading(oldCtx, svcLease, service); err != nil { + t.Fatalf("onStoppedLeading returned an error: %v", err) + } + if got, err := p.getServiceContext(service.UID); err != nil || got != replacementCtx { + t.Fatalf("replacement context was changed: got %v, err %v", got, err) + } + if len(p.ServiceInstances) != 1 || p.ServiceInstances[0] != replacementInstance { + t.Fatal("replacement service instance was removed by superseded cleanup") + } +} diff --git a/pkg/services/watch_endpoints.go b/pkg/services/watch_endpoints.go index 5ccd8334..eda191a9 100644 --- a/pkg/services/watch_endpoints.go +++ b/pkg/services/watch_endpoints.go @@ -1,6 +1,7 @@ package services import ( + "context" "fmt" "sync" @@ -10,28 +11,31 @@ import ( "github.com/kube-vip/kube-vip/pkg/endpoints" "github.com/kube-vip/kube-vip/pkg/endpoints/providers" "github.com/kube-vip/kube-vip/pkg/servicecontext" + "github.com/kube-vip/kube-vip/pkg/utils" v1 "k8s.io/api/core/v1" - apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/watch" ) -func (p *Processor) watchEndpoint(svcCtx *servicecontext.Context, id string, service *v1.Service, provider providers.Provider) error { +func (p *Processor) watchEndpoint(svcCtx *servicecontext.Context, id string, service *v1.Service, + provider providers.Provider, cancelWatcher context.CancelCauseFunc) error { log.Info("watching", "provider", provider.GetLabel(), "service_name", service.Name, "namespace", service.Namespace) // Use a restartable watcher, as this should help in the event of etcd or timeout issues rw, err := provider.CreateRetryWatcher(svcCtx.Ctx, p.rwClientSet, service) if err != nil { - return fmt.Errorf("[%s] error watching endpoints: %w", provider.GetLabel(), err) + if svcCtx.Ctx.Err() != nil { + return nil + } + return utils.WrapPanicError(err, "[%s] error watching endpoints", provider.GetLabel()) } d, err := debouncer.New(rw.ResultChan(), p.config.DebounceTime) if err != nil { rw.Stop() - return fmt.Errorf("failed to create debouncer for endpoints event: %w", err) + return utils.WrapPanicError(err, "failed to create debouncer for endpoints event") } - wg := sync.WaitGroup{} - + var wg sync.WaitGroup stopChan := make(chan any) defer func() { @@ -46,6 +50,9 @@ func (p *Processor) watchEndpoint(svcCtx *servicecontext.Context, id string, ser if d != nil { if err := d.Start(svcCtx.Ctx); err != nil { log.Error("[endpoint watcher] debouncer, cancelling context", "error", err.Error()) + if svcCtx.Ctx.Err() == nil { + cancelWatcher(utils.WrapPanicError(err, "[%s] endpoint debouncer failed", provider.GetLabel())) + } svcCtx.Cancel() } } @@ -85,14 +92,19 @@ func (p *Processor) watchEndpoint(svcCtx *servicecontext.Context, id string, ser return fmt.Errorf("[%s] error while processing delete event: %w", provider.GetLabel(), err) } - log.Info("[endpoint watcher] stopping watching - endpoint object deleted", "provider", provider.GetLabel(), "service name", service.Name, "namespace", service.Namespace) - return nil + log.Info("[endpoint watcher] endpoint object deleted", "provider", provider.GetLabel(), "service name", service.Name, "namespace", service.Namespace) case watch.Error: - errObject := apierrors.FromObject(event.Object) - statusErr, _ := errObject.(*apierrors.StatusError) - log.Error("watch error", "provider", provider.GetLabel(), "err", statusErr) + if svcCtx.Ctx.Err() != nil { + return nil + } + watchErr := utils.WatchError(event.Object) + log.Error("watch error", "provider", provider.GetLabel(), "err", watchErr) + return utils.WrapPanicError(watchErr, "[%s] endpoint watch failed", provider.GetLabel()) } } + if svcCtx.Ctx.Err() != nil { + return nil + } log.Info("[endpoint watcher] stopping watching", "provider", provider.GetLabel(), "service name", service.Name, "namespace", service.Namespace) - return nil //nolint:govet + return utils.NewPanicError("[%s] endpoint watch channel closed unexpectedly for service %s/%s", provider.GetLabel(), service.Namespace, service.Name) } diff --git a/pkg/services/watch_services.go b/pkg/services/watch_services.go index b6266eb9..401cb320 100644 --- a/pkg/services/watch_services.go +++ b/pkg/services/watch_services.go @@ -2,13 +2,11 @@ package services import ( "context" - "errors" "fmt" "sync" log "log/slog" - "github.com/davecgh/go-spew/spew" "github.com/kube-vip/kube-vip/pkg/debouncer" "github.com/kube-vip/kube-vip/pkg/kubevip" "github.com/kube-vip/kube-vip/pkg/metrics" @@ -16,7 +14,6 @@ import ( "github.com/kube-vip/kube-vip/pkg/utils" "github.com/prometheus/client_golang/prometheus" v1 "k8s.io/api/core/v1" - apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/watch" "k8s.io/client-go/tools/cache" @@ -62,7 +59,7 @@ func (p *Processor) ServicesWatcher(ctx context.Context, serviceFunc *Callback, return fmt.Errorf("failed to create debouncer for endpoints event: %w", err) } - wg := sync.WaitGroup{} + var wg sync.WaitGroup defer func() { if d != nil { d.Stop() @@ -71,14 +68,14 @@ func (p *Processor) ServicesWatcher(ctx context.Context, serviceFunc *Callback, wg.Wait() }() - watcherCtx, watcherCancel := context.WithCancel(ctx) - defer watcherCancel() + 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()) - watcherCancel() + cancelWatcher(utils.WrapPanicError(err, "service debouncer failed")) } } <-watcherCtx.Done() @@ -102,41 +99,38 @@ func (p *Processor) ServicesWatcher(ctx context.Context, serviceFunc *Callback, // 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); err != nil { - if errors.Is(err, &utils.PanicError{}) { + 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) - } else { - log.Error("service watcher event failed", "type", event.Type, "error", err) } + log.Error("service watcher event failed", "type", event.Type, "error", err) } case watch.Deleted: if err := p.Delete(event, forcedOnly); err != nil { - if errors.Is(err, &utils.PanicError{}) { + if utils.IsPanicError(err) { return fmt.Errorf("delete service error: %w", err) - } else { - log.Error("service watcher event failed", "type", event.Type, "error", 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") - - // This round trip allows us to handle unstructured status - errObject := apierrors.FromObject(event.Object) - statusErr, ok := errObject.(*apierrors.StatusError) - if !ok { - log.Error(spew.Sprintf("Received an error which is not *metav1.Status but %#+v", event.Object)) - } - - status := statusErr.ErrStatus - log.Error("services", "err", status) + 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 nil + return utils.NewPanicError("service watch channel closed unexpectedly") } func lbClassFilterLegacy(svc *v1.Service, config *kubevip.Config) bool { diff --git a/pkg/trafficmirror/trafficmirror.go b/pkg/trafficmirror/trafficmirror.go index 4f899502..e1982973 100644 --- a/pkg/trafficmirror/trafficmirror.go +++ b/pkg/trafficmirror/trafficmirror.go @@ -146,7 +146,7 @@ func getQdiscFromInterfaceByType(nicID int, nicName string, qType string) (uint3 // get id through tc qdisc show dev fromNICName qs, err := netlink.QdiscList(&netlink.Ifb{LinkAttrs: netlink.LinkAttrs{Index: nicID}}) if err != nil { - fmt.Printf("Failed to list qdisc for interface %s: %v", nicName, err) + log.Error("failed to list qdisc", "interface", nicName, "err", err) return 0, err } for _, q := range qs { diff --git a/pkg/utils/panic.go b/pkg/utils/panic.go index 3553613b..b890ff79 100644 --- a/pkg/utils/panic.go +++ b/pkg/utils/panic.go @@ -1,15 +1,31 @@ package utils -import "fmt" +import ( + "errors" + "fmt" +) type PanicError struct { - cause string + cause error } func (e *PanicError) Error() string { return fmt.Sprintf("%s - unrecoverable error", e.cause) } -func NewPanicError(cause string) error { - return &PanicError{cause: cause} +func (e *PanicError) Unwrap() error { + return e.cause +} + +func NewPanicError(format string, args ...any) error { + return &PanicError{cause: fmt.Errorf(format, args...)} +} + +func WrapPanicError(err error, format string, args ...any) error { + return &PanicError{cause: fmt.Errorf("%s: %w", fmt.Sprintf(format, args...), err)} +} + +func IsPanicError(err error) bool { + var panicErr *PanicError + return errors.As(err, &panicErr) } diff --git a/pkg/utils/panic_test.go b/pkg/utils/panic_test.go new file mode 100644 index 00000000..c78245ec --- /dev/null +++ b/pkg/utils/panic_test.go @@ -0,0 +1,41 @@ +package utils + +import ( + "errors" + "fmt" + "testing" +) + +func TestIsPanicError(t *testing.T) { + panicErr := NewPanicError("endpoint watch stopped") + tests := []struct { + name string + err error + want bool + }{ + {name: "direct", err: panicErr, want: true}, + {name: "wrapped", err: fmt.Errorf("watch failed: %w", panicErr), want: true}, + {name: "ordinary", err: errors.New("watch failed"), want: false}, + {name: "nil", err: nil, want: false}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if got := IsPanicError(test.err); got != test.want { + t.Fatalf("IsPanicError() = %t, want %t", got, test.want) + } + }) + } +} + +func TestWrapPanicErrorPreservesCause(t *testing.T) { + cause := errors.New("endpointslices is forbidden") + err := WrapPanicError(cause, "endpoint watch failed") + + if !IsPanicError(err) { + t.Fatal("expected wrapped error to be classified as PanicError") + } + if !errors.Is(err, cause) { + t.Fatal("expected wrapped PanicError to preserve its cause") + } +} diff --git a/pkg/utils/watcher.go b/pkg/utils/watcher.go index 9858e5f5..35663b3f 100644 --- a/pkg/utils/watcher.go +++ b/pkg/utils/watcher.go @@ -7,10 +7,20 @@ import ( "time" apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/util/wait" "k8s.io/apimachinery/pkg/watch" ) +// WatchError converts a Kubernetes watch error object into a safe Go error. +func WatchError(object runtime.Object) error { + errObject := apierrors.FromObject(object) + if statusErr, ok := errObject.(*apierrors.StatusError); ok { + return statusErr + } + return fmt.Errorf("unknown watch error object of type %T: %v", object, object) +} + // watchWithAuthRetry retries watchFn with exponential backoff on transient 403 Forbidden // and 401 Unauthorized errors. On joining control plane nodes with K8s 1.34+, the local // etcd may still be a learner when kube-vip starts, causing RBAC data to be unavailable. @@ -39,7 +49,14 @@ func WatchWithAuthRetry(ctx context.Context, watchFn func(context.Context) (watc return false, nil }) if err != nil { - return nil, NewPanicError(fmt.Sprintf("watch failed after retries: %q (last: %v)", err.Error(), lastErr)) + if ctx.Err() != nil { + return nil, ctx.Err() + } + if lastErr != nil { + log.Error("watch auth retries exhausted", "err", lastErr) + return nil, lastErr + } + return nil, WrapPanicError(err, "watch failed after retries (last: %v)", lastErr) } return w, nil } diff --git a/pkg/utils/watcher_test.go b/pkg/utils/watcher_test.go new file mode 100644 index 00000000..088cc11c --- /dev/null +++ b/pkg/utils/watcher_test.go @@ -0,0 +1,37 @@ +package utils + +import ( + "errors" + "strings" + "testing" + + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +func TestWatchErrorPreservesStatusError(t *testing.T) { + status := &metav1.Status{ + Status: metav1.StatusFailure, + Reason: metav1.StatusReasonForbidden, + Message: "endpointslices is forbidden", + } + + err := WatchError(status) + var statusErr *apierrors.StatusError + if !errors.As(err, &statusErr) { + t.Fatalf("expected a status error, got %T: %v", err, err) + } + if !strings.Contains(err.Error(), status.Message) { + t.Fatalf("expected error to contain %q, got %q", status.Message, err) + } +} + +func TestWatchErrorHandlesUnexpectedObject(t *testing.T) { + err := WatchError(&metav1.APIGroup{}) + if err == nil { + t.Fatal("expected an error for an unexpected watch object") + } + if !strings.Contains(err.Error(), "unknown watch error object") { + t.Fatalf("expected unexpected-object context, got %q", err) + } +}