From 0f0a157e7259347bf2a25b9280eacbcd9daa72dc Mon Sep 17 00:00:00 2001 From: Brad Davidson Date: Mon, 10 Mar 2025 23:47:33 +0000 Subject: [PATCH] Fix syncing empty list of apiserver addresses during initial startup Also add more debug logging to the sync process. Signed-off-by: Brad Davidson (cherry picked from commit 781640ecea12e0cd83541c99a066baadf47038ef) Signed-off-by: Brad Davidson --- pkg/agent/tunnel/tunnel.go | 124 ++++++++++++++++++++++--------------- 1 file changed, 74 insertions(+), 50 deletions(-) diff --git a/pkg/agent/tunnel/tunnel.go b/pkg/agent/tunnel/tunnel.go index 98094a8d02..ae2d1cba09 100644 --- a/pkg/agent/tunnel/tunnel.go +++ b/pkg/agent/tunnel/tunnel.go @@ -38,7 +38,11 @@ import ( ) var ( - endpointDebounceDelay = time.Second + // endpointDebounceDelay sets how long we wait before updating apiserver + // addresses when the kubernetes endpoint list changes. When the apiserver is + // starting up it adds then removes then re-adds itself a few times in quick + // succession, and we want to avoid closing connections unnecessarily. + endpointDebounceDelay = 3 * time.Second defaultDialer = net.Dialer{} ) @@ -335,14 +339,9 @@ func (a *agentTunnel) watchEndpoints(ctx context.Context, apiServerReady <-chan <-done }() - var cancelUpdate context.CancelFunc - for { select { case <-ctx.Done(): - if cancelUpdate != nil { - cancelUpdate() - } return case ev, ok := <-watch.ResultChan(): endpoint, ok := ev.Object.(*v1.Endpoints) @@ -351,20 +350,15 @@ func (a *agentTunnel) watchEndpoints(ctx context.Context, apiServerReady <-chan continue } - if cancelUpdate != nil { - cancelUpdate() - } - - var debounceCtx context.Context - debounceCtx, cancelUpdate = context.WithCancel(ctx) - // When joining the cluster, the apiserver adds, removes, and then re-adds itself to // the endpoint list several times. This causes a bit of thrashing if we react to // endpoint changes immediately. Instead, perform the endpoint update in a // goroutine that sleeps for a short period before checking for changes and updating // the proxy addresses. If another update occurs, the previous update operation // will be cancelled and a new one queued. - go syncProxyAddresses(debounceCtx, util.GetAddresses(endpoint)) + addresses := util.GetAddresses(endpoint) + logrus.Debugf("Syncing apiserver addresses from tunnel watch: %v", addresses) + syncProxyAddresses(addresses) } } } @@ -478,15 +472,15 @@ func (a *agentTunnel) dialContext(ctx context.Context, network, address string) } // proxySyncer is a common signature for functions that sync the proxy address list with a context -type proxySyncer func(ctx context.Context, addresses []string) +type proxySyncer func(addresses []string) // getProxySyncer returns a function that can be called to update the list of supervisors. // This function is responsible for connecting to or disconnecting websocket tunnels, // as well as updating the proxy loadbalancer server list. func (a *agentTunnel) getProxySyncer(ctx context.Context, wg *sync.WaitGroup, tlsConfig *tls.Config, proxy proxy.Proxy) proxySyncer { disconnect := map[string]context.CancelFunc{} - // Attempt to connect to supervisors, storing their cancellation function for later when we - // need to disconnect. + // Attempt to connect to inital list of addresses, storing their cancellation + // function for later when we need to disconnect. for _, address := range proxy.SupervisorAddresses() { if _, ok := disconnect[address]; !ok { conn := a.connect(ctx, wg, address, tlsConfig) @@ -495,43 +489,72 @@ func (a *agentTunnel) getProxySyncer(ctx context.Context, wg *sync.WaitGroup, tl } } - // return a function that can be called to update the address list. - // servers will be connected to or disconnected from as necessary, - // and the proxy addresses updated. - return func(debounceCtx context.Context, addresses []string) { - select { - case <-time.After(endpointDebounceDelay): - case <-debounceCtx.Done(): + var cancelUpdate context.CancelFunc + + // return a function that can be called to update the address list. servers will be + // connected to or disconnected from as necessary, and the proxy addresses updated. + // The update is done in a goroutine that waits a short period in order to reduce + // thrashing during apiserver startup. Each time the function is called, the context for + // the goroutine started by the previous call is cancelled to prevent it from updating + // if the delay has not yet expired. + return func(addresses []string) { + if len(addresses) == 0 { + logrus.Debugf("Skipping apiserver addresses sync: %v", addresses) return } - // Compare list of supervisor addresses before and after syncing apiserver - // endpoints into the proxy to figure out which supervisors we need to connect to - // or disconnect from. Note that the addresses we were passed will not match - // the supervisor addresses if the supervisor and apiserver are on different ports - - // they must be round-tripped through proxy.Update before comparing. - curAddresses := sets.New(proxy.SupervisorAddresses()...) - proxy.Update(addresses) - newAddresses := sets.New(proxy.SupervisorAddresses()...) - - // add new servers - for address := range newAddresses.Difference(curAddresses) { - if _, ok := disconnect[address]; !ok { - conn := a.connect(ctx, nil, address, tlsConfig) - logrus.Infof("Started tunnel to %s", address) - disconnect[address] = conn.cancel - proxy.SetHealthCheck(address, conn.healthCheck) - } + if cancelUpdate != nil { + cancelUpdate() } - // remove old servers - for address := range curAddresses.Difference(newAddresses) { - if cancel, ok := disconnect[address]; ok { - cancel() - delete(disconnect, address) - logrus.Infof("Stopped tunnel to %s", address) + var debounceCtx context.Context + debounceCtx, cancelUpdate = context.WithCancel(ctx) + + go func() { + select { + case <-time.After(endpointDebounceDelay): + logrus.Debugf("Settled apiserver addresses sync: %v", addresses) + case <-debounceCtx.Done(): + logrus.Debugf("Cancelled apiserver addresses sync: %v", addresses) + return } - } + + // Compare list of supervisor addresses before and after syncing apiserver + // endpoints into the proxy to figure out which supervisors we need to connect to + // or disconnect from. Note that the addresses we were passed will not match + // the supervisor addresses if the supervisor and apiserver are on different ports - + // they must be round-tripped through proxy.Update before comparing. + curAddresses := sets.New(proxy.SupervisorAddresses()...) + proxy.Update(addresses) + newAddresses := sets.New(proxy.SupervisorAddresses()...) + + addedAddresses := newAddresses.Difference(curAddresses) + removedAddresses := curAddresses.Difference(newAddresses) + if addedAddresses.Len() == 0 && removedAddresses.Len() == 0 { + return + } + + logrus.Debugf("Sync apiserver addresses - connecting: %v, disconnecting: %v", addedAddresses.UnsortedList(), removedAddresses.UnsortedList()) + + // add new servers + for address := range addedAddresses { + if _, ok := disconnect[address]; !ok { + conn := a.connect(ctx, nil, address, tlsConfig) + logrus.Infof("Started tunnel to %s", address) + disconnect[address] = conn.cancel + proxy.SetHealthCheck(address, conn.healthCheck) + } + } + + // remove old servers + for address := range removedAddresses { + if cancel, ok := disconnect[address]; ok { + cancel() + delete(disconnect, address) + logrus.Infof("Stopped tunnel to %s", address) + } + } + }() } } @@ -550,10 +573,11 @@ func getAPIServersRequester(node *daemonconfig.Node, proxy proxy.Proxy, syncProx } } - if addresses, err := agentconfig.GetAPIServers(ctx, info); err != nil { + if addresses, err := agentconfig.GetAPIServers(ctx, info); err != nil || len(addresses) == 0 { logrus.Warnf("Failed to get apiserver addresses from supervisor: %v", err) } else { - syncProxyAddresses(ctx, addresses) + logrus.Debugf("Syncing apiserver addresses from server: %v", addresses) + syncProxyAddresses(addresses) } } }