From 5e2220fd4dac9d736f0a6bfd15356360a417f484 Mon Sep 17 00:00:00 2001 From: Marcel Fest Date: Thu, 20 Aug 2026 13:03:08 +0200 Subject: [PATCH] Fix/watch err (#1697) * refactor(errors): centralize fatal error handling Detect wrapped PanicError values consistently and preserve their underlying causes when adding fatal context. Apply the helpers to manager, cluster, and IPVS error paths. Assisted-by: GitHub-Copilot:unspecified Signed-off-by: Marcel Fest * fix(watchers): restart after terminal watch failures Propagate fatal endpoint watcher failures through the owning service watcher so kube-vip releases leadership instead of remaining active with a stale watch. Treat terminal service, node, and annotation watch failures as errors while preserving clean context cancellation. Return exhausted authorization failures to RetryWatcher, safely decode watch error objects, and replace direct go-spew diagnostics with structured logging. Fixes #1685 Assisted-by: GitHub-Copilot:unspecified Signed-off-by: Marcel Fest * fix(services): replace state after traffic policy changes Recreate the service context and instance as one generation when a Service change requires teardown. Ignore delayed leadership cleanup from superseded contexts so it cannot remove replacement state. This prevents a stale Cluster-policy endpoint watcher from winning the service lease after externalTrafficPolicy changes to Local. Assisted-by: GitHub-Copilot:unspecified Signed-off-by: Marcel Fest * fix(cli): return command errors to container runtime Propagate manager and service command failures through Cobra so the process exits with status 1. Show usage for invocation errors while keeping runtime failures concise. Assisted-by: GitHub-Copilot:unspecified Signed-off-by: Marcel Fest * refactor(logging): use structured errors Replace direct stdout error output with slog records for command failures and traffic mirror qdisc lookup failures. Assisted-by: GitHub-Copilot:unspecified Signed-off-by: Marcel Fest * fix(watchers): continue after endpoint deletion Keep EndpointSlice watchers active when an individual endpoint object is deleted so replacement objects can be observed and service traffic can recover. Assisted-by: GitHub-Copilot:unspecified Signed-off-by: Marcel Fest --------- Signed-off-by: Marcel Fest --- cmd/kube-vip.go | 102 ++++++++++++--------------- go.mod | 2 +- main.go | 4 +- pkg/cluster/service.go | 4 +- pkg/election/election.go | 26 +++---- pkg/loadbalancer/ipvs.go | 2 +- pkg/manager/manager.go | 3 +- pkg/manager/watch_annotations.go | 20 ++---- pkg/services/leader.go | 17 +++-- pkg/services/processor.go | 36 +++++----- pkg/services/processor_lease_test.go | 35 +++++++++ pkg/services/watch_endpoints.go | 36 ++++++---- pkg/services/watch_services.go | 44 +++++------- pkg/trafficmirror/trafficmirror.go | 2 +- pkg/utils/panic.go | 24 +++++-- pkg/utils/panic_test.go | 41 +++++++++++ pkg/utils/watcher.go | 19 ++++- pkg/utils/watcher_test.go | 37 ++++++++++ 18 files changed, 295 insertions(+), 159 deletions(-) create mode 100644 pkg/utils/panic_test.go create mode 100644 pkg/utils/watcher_test.go 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) + } +}