diff --git a/router/env/ctrl.go b/router/env/ctrl.go index 3b4e362a6..be951608e 100644 --- a/router/env/ctrl.go +++ b/router/env/ctrl.go @@ -17,67 +17,24 @@ package env import ( - "fmt" "sync/atomic" "time" "github.com/openziti/channel/v2" - "github.com/openziti/foundation/v2/concurrenz" - "github.com/openziti/foundation/v2/errorz" ) -func NewDefaultHeartbeatOptions() *HeartbeatOptions { - return &HeartbeatOptions{ - HeartbeatOptions: *channel.DefaultHeartbeatOptions(), - UnresponsiveAfter: 5 * time.Second, - } -} - -func NewHeartbeatOptions(options *channel.HeartbeatOptions) (*HeartbeatOptions, error) { - unresponsiveAfter, err := options.GetDuration("unresponsiveAfter") - if err != nil { - return nil, err - } - result := NewDefaultHeartbeatOptions() - result.HeartbeatOptions = *options - if unresponsiveAfter != nil { - result.UnresponsiveAfter = *unresponsiveAfter - } - return result, nil -} - -type HeartbeatOptions struct { - channel.HeartbeatOptions - UnresponsiveAfter time.Duration -} - -type CtrlChannel interface { - Channel() channel.Channel - DefaultRequestTimeout() time.Duration -} - type NetworkController interface { Channel() channel.Channel + Address() string Latency() time.Duration HeartbeatCallback() channel.HeartbeatCallback IsUnresponsive() bool isMoreResponsive(other NetworkController) bool } -type NetworkControllers interface { - Add(ch channel.Channel) (NetworkController, error) - GetAll() map[string]NetworkController - AnyCtrlChannel() channel.Channel - GetCtrlChannel(ctrlId string) channel.Channel - GetCtrlChannelByAddress(address string) (ctrlId string, ch channel.Channel) - DefaultRequestTimeout() time.Duration - ForEach(f func(ctrlId string, ch channel.Channel)) - Close() error - CloseAndRemoveByAddress(address string) error -} - type networkCtrl struct { ch channel.Channel + address string heartbeatOptions *HeartbeatOptions lastTx int64 lastRx int64 @@ -93,6 +50,10 @@ func (self *networkCtrl) Channel() channel.Channel { return self.ch } +func (self *networkCtrl) Address() string { + return self.address +} + func (self *networkCtrl) Latency() time.Duration { return time.Duration(self.latency.Load()) } @@ -140,91 +101,27 @@ func (self *networkCtrl) CheckHeartBeat() { } } -func NewNetworkControllers(defaultRequestTimeout time.Duration, heartbeatOptions *HeartbeatOptions) NetworkControllers { - return &networkControllers{ - heartbeatOptions: heartbeatOptions, - defaultRequestTimeout: defaultRequestTimeout, +func NewDefaultHeartbeatOptions() *HeartbeatOptions { + return &HeartbeatOptions{ + HeartbeatOptions: *channel.DefaultHeartbeatOptions(), + UnresponsiveAfter: 5 * time.Second, } } -type networkControllers struct { - heartbeatOptions *HeartbeatOptions - defaultRequestTimeout time.Duration - ctrls concurrenz.CopyOnWriteMap[string, NetworkController] -} - -func (self *networkControllers) Add(ch channel.Channel) (NetworkController, error) { - ctrl := &networkCtrl{ - ch: ch, - heartbeatOptions: self.heartbeatOptions, +func NewHeartbeatOptions(options *channel.HeartbeatOptions) (*HeartbeatOptions, error) { + unresponsiveAfter, err := options.GetDuration("unresponsiveAfter") + if err != nil { + return nil, err } - if existing := self.ctrls.Get(ctrl.Channel().Id()); existing != nil { - return nil, fmt.Errorf("duplicate channel with id %v", ctrl.Channel().Id()) + result := NewDefaultHeartbeatOptions() + result.HeartbeatOptions = *options + if unresponsiveAfter != nil { + result.UnresponsiveAfter = *unresponsiveAfter } - self.ctrls.Put(ctrl.Channel().Id(), ctrl) - return ctrl, nil + return result, nil } -func (self *networkControllers) GetAll() map[string]NetworkController { - return self.ctrls.AsMap() -} - -func (self *networkControllers) AnyCtrlChannel() channel.Channel { - var current NetworkController - for _, ctrl := range self.ctrls.AsMap() { - if current == nil || ctrl.isMoreResponsive(current) { - current = ctrl - } - } - if current == nil { - return nil - } - return current.Channel() -} - -func (self *networkControllers) GetCtrlChannel(controllerId string) channel.Channel { - if ctrl := self.ctrls.Get(controllerId); ctrl != nil { - return ctrl.Channel() - } - return nil -} - -func (self *networkControllers) GetCtrlChannelByAddress(address string) (string, channel.Channel) { - for k, v := range self.ctrls.AsMap() { - if v.Channel().Underlay().GetLocalAddr().String() == address || v.Channel().Underlay().GetRemoteAddr().String() == address { - return k, v.Channel() - } - } - return "", nil -} - -func (self *networkControllers) DefaultRequestTimeout() time.Duration { - return self.defaultRequestTimeout -} - -func (self *networkControllers) ForEach(f func(controllerId string, ch channel.Channel)) { - for controllerId, ctrl := range self.ctrls.AsMap() { - f(controllerId, ctrl.Channel()) - } -} - -func (self *networkControllers) Close() error { - var errors errorz.MultipleErrors - self.ForEach(func(_ string, ch channel.Channel) { - if err := ch.Close(); err != nil { - errors = append(errors, err) - } - }) - return errors.ToError() -} - -func (self *networkControllers) CloseAndRemoveByAddress(address string) error { - ctrlId, ch := self.GetCtrlChannelByAddress(address) - if ch != nil { - delete(self.ctrls.AsMap(), ctrlId) - if err := ch.Close(); err != nil { - return err - } - } - return nil +type HeartbeatOptions struct { + channel.HeartbeatOptions + UnresponsiveAfter time.Duration } diff --git a/router/env/ctrls.go b/router/env/ctrls.go new file mode 100644 index 000000000..be33bef67 --- /dev/null +++ b/router/env/ctrls.go @@ -0,0 +1,214 @@ +/* + Copyright NetFoundry Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + https://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package env + +import ( + "fmt" + "github.com/cenkalti/backoff/v4" + "github.com/michaelquigley/pfxlog" + "github.com/openziti/transport/v2" + cmap "github.com/orcaman/concurrent-map/v2" + "github.com/pkg/errors" + "sync" + "time" + + "github.com/openziti/channel/v2" + "github.com/openziti/foundation/v2/concurrenz" + "github.com/openziti/foundation/v2/errorz" +) + +type NetworkControllers interface { + UpdateControllerEndpoints(endpoints []string) bool + GetAll() map[string]NetworkController + GetNetworkController(ctrlId string) NetworkController + AnyCtrlChannel() channel.Channel + GetCtrlChannel(ctrlId string) channel.Channel + DefaultRequestTimeout() time.Duration + ForEach(f func(ctrlId string, ch channel.Channel)) + Close() error +} + +type CtrlDialer func(address transport.Address, bindHandler channel.BindHandler) error + +func NewNetworkControllers(defaultRequestTimeout time.Duration, dialer CtrlDialer, heartbeatOptions *HeartbeatOptions) NetworkControllers { + return &networkControllers{ + ctrlDialer: dialer, + heartbeatOptions: heartbeatOptions, + defaultRequestTimeout: defaultRequestTimeout, + ctrlEndpoints: cmap.New[struct{}](), + } +} + +type networkControllers struct { + lock sync.Mutex + ctrlDialer CtrlDialer + heartbeatOptions *HeartbeatOptions + defaultRequestTimeout time.Duration + ctrlEndpoints cmap.ConcurrentMap[string, struct{}] + ctrls concurrenz.CopyOnWriteMap[string, NetworkController] +} + +func (self *networkControllers) UpdateControllerEndpoints(addresses []string) bool { + self.lock.Lock() + defer self.lock.Unlock() + + changed := false + + log := pfxlog.Logger() + endpoints := map[string]struct{}{} + for _, endpoint := range addresses { + endpoints[endpoint] = struct{}{} + } + + for _, endpoint := range self.ctrlEndpoints.Keys() { + if _, ok := endpoints[endpoint]; ok { + // already known endpoint, don't need to try and connect in next step + delete(endpoints, endpoint) + } else { + // existing endpoint is no longer valid, close and remove it + log.WithField("endpoint", endpoints).Info("removing old ctrl endpoint") + changed = true + self.CloseAndRemoveByAddress(endpoint) + } + } + + for endpoint := range endpoints { + log.WithField("endpoint", endpoints).Info("adding new ctrl endpoint") + changed = true + self.connectToControllerWithBackoff(endpoint) + } + + return changed +} + +func (self *networkControllers) connectToControllerWithBackoff(endpoint string) { + log := pfxlog.Logger().WithField("endpoint", endpoint) + + addr, err := transport.ParseAddress(endpoint) + if err != nil { + log.WithField("endpoint", endpoint).WithError(err).Error("unable to parse endpoint address, ignoring") + return + } + + self.ctrlEndpoints.Set(endpoint, struct{}{}) + + expBackoff := backoff.NewExponentialBackOff() + expBackoff.InitialInterval = 50 * time.Millisecond + expBackoff.MaxInterval = 5 * time.Minute + expBackoff.MaxElapsedTime = 100 * 365 * 24 * time.Hour + + bindHandler := channel.BindHandlerF(func(binding channel.Binding) error { + return self.Add(endpoint, binding.GetChannel()) + }) + + operation := func() error { + if !self.ctrlEndpoints.Has(endpoint) { + return backoff.Permanent(errors.New("controller removed before connection established")) + } + err := self.ctrlDialer(addr, bindHandler) + if err != nil { + log.WithError(err).Error("unable to connect controller") + } + return err + } + + log.Info("starting connection attempts") + + go func() { + if err := backoff.Retry(operation, expBackoff); err != nil { + log.WithError(err).Error("unable to connect controller, stopping retries.") + } else { + log.Info("successfully connected to controller") + } + }() +} + +func (self *networkControllers) Add(address string, ch channel.Channel) error { + ctrl := &networkCtrl{ + ch: ch, + address: address, + heartbeatOptions: self.heartbeatOptions, + } + if existing := self.ctrls.Get(ch.Id()); existing != nil { + if !existing.Channel().IsClosed() { + return fmt.Errorf("duplicate channel with id %v", ctrl.Channel().Id()) + } + } + self.ctrls.Put(ch.Id(), ctrl) + return nil +} + +func (self *networkControllers) GetAll() map[string]NetworkController { + return self.ctrls.AsMap() +} + +func (self *networkControllers) AnyCtrlChannel() channel.Channel { + var current NetworkController + for _, ctrl := range self.ctrls.AsMap() { + if current == nil || ctrl.isMoreResponsive(current) { + current = ctrl + } + } + if current == nil { + return nil + } + return current.Channel() +} + +func (self *networkControllers) GetCtrlChannel(controllerId string) channel.Channel { + if ctrl := self.ctrls.Get(controllerId); ctrl != nil { + return ctrl.Channel() + } + return nil +} + +func (self *networkControllers) GetNetworkController(controllerId string) NetworkController { + return self.ctrls.Get(controllerId) +} + +func (self *networkControllers) DefaultRequestTimeout() time.Duration { + return self.defaultRequestTimeout +} + +func (self *networkControllers) ForEach(f func(controllerId string, ch channel.Channel)) { + for controllerId, ctrl := range self.ctrls.AsMap() { + f(controllerId, ctrl.Channel()) + } +} + +func (self *networkControllers) Close() error { + var errList errorz.MultipleErrors + self.ForEach(func(_ string, ch channel.Channel) { + if err := ch.Close(); err != nil { + errList = append(errList, err) + } + }) + return errList.ToError() +} + +func (self *networkControllers) CloseAndRemoveByAddress(address string) { + self.ctrlEndpoints.Remove(address) + + for id, ctrl := range self.ctrls.AsMap() { + if ctrl.Address() == address { + self.ctrls.Delete(id) + if err := ctrl.Channel().Close(); err != nil { + pfxlog.Logger().WithField("ctrlId", id).WithField("endpoint", address).WithError(err).Error("error closing channel to controller") + } + } + } +} diff --git a/router/handler_ctrl/bind.go b/router/handler_ctrl/bind.go index 3d34a812f..c8bd31279 100644 --- a/router/handler_ctrl/bind.go +++ b/router/handler_ctrl/bind.go @@ -35,13 +35,12 @@ import ( type bindHandler struct { env env.RouterEnv forwarder *forwarder.Forwarder - ctrlAddressChanger CtrlAddressChanger xgDialerPool goroutines.Pool linkDialerPool goroutines.Pool ctrlAddressUpdater CtrlAddressUpdater } -func NewBindHandler(routerEnv env.RouterEnv, forwarder *forwarder.Forwarder, ctrlAddressChanger CtrlAddressChanger, ctrlAddressUpdater CtrlAddressUpdater) (channel.BindHandler, error) { +func NewBindHandler(routerEnv env.RouterEnv, forwarder *forwarder.Forwarder, ctrlAddressUpdater CtrlAddressUpdater) (channel.BindHandler, error) { linkDialerPoolConfig := goroutines.PoolConfig{ QueueSize: uint32(forwarder.Options.LinkDial.QueueLength), MinWorkers: 0, @@ -81,7 +80,6 @@ func NewBindHandler(routerEnv env.RouterEnv, forwarder *forwarder.Forwarder, ctr return &bindHandler{ env: routerEnv, forwarder: forwarder, - ctrlAddressChanger: ctrlAddressChanger, xgDialerPool: xgDialerPool, linkDialerPool: linkDialerPool, ctrlAddressUpdater: ctrlAddressUpdater, @@ -95,16 +93,16 @@ func (self *bindHandler) BindChannel(binding channel.Binding) error { binding.AddTypedReceiveHandler(newUnrouteHandler(self.forwarder)) binding.AddTypedReceiveHandler(newTraceHandler(self.env.GetRouterId(), self.forwarder.TraceController(), binding.GetChannel())) binding.AddTypedReceiveHandler(newInspectHandler(self.env, self.forwarder)) - binding.AddTypedReceiveHandler(newSettingsHandler(self.ctrlAddressChanger)) + binding.AddTypedReceiveHandler(newSettingsHandler(self.ctrlAddressUpdater)) binding.AddTypedReceiveHandler(newFaultHandler(self.env.GetXlinkRegistry())) binding.AddTypedReceiveHandler(newUpdateCtrlAddressesHandler(self.ctrlAddressUpdater)) binding.AddPeekHandler(trace.NewChannelPeekHandler(self.env.GetRouterId().Token, binding.GetChannel(), self.forwarder.TraceController())) latency.AddLatencyProbeResponder(binding) - ctrl, err := self.env.GetNetworkControllers().Add(binding.GetChannel()) - if err != nil { - return err + ctrl := self.env.GetNetworkControllers().GetNetworkController(binding.GetChannel().Id()) + if ctrl == nil { + return errors.Errorf("controller [%v] not registered, cannot configure", binding.GetChannel().Id()) } channel.ConfigureHeartbeat(binding, self.env.GetHeartbeatOptions().SendInterval, self.env.GetHeartbeatOptions().CheckInterval, ctrl.HeartbeatCallback()) diff --git a/router/handler_ctrl/settings.go b/router/handler_ctrl/settings.go index 925e66eb3..7a7ff5926 100644 --- a/router/handler_ctrl/settings.go +++ b/router/handler_ctrl/settings.go @@ -21,21 +21,11 @@ import ( "github.com/openziti/channel/v2" "github.com/openziti/fabric/pb/ctrl_pb" "google.golang.org/protobuf/proto" - "strings" ) -// CtrlAddressChanger provides indirect access to underlying configuration without creating -// import loops w/ *router.Config. -type CtrlAddressChanger interface { - CreateBackup() (string, error) - UpdateControllerEndpoint(address string) error - Save() error - CurrentCtrlAddress() string -} - // settingsHandler is a catch-all handler for all settings message (currently only sent on router connect). type settingsHandler struct { - config CtrlAddressChanger + updater CtrlAddressUpdater } func (handler *settingsHandler) ContentType() int32 { @@ -54,7 +44,7 @@ func (handler *settingsHandler) HandleReceive(msg *channel.Message, ch channel.C switch settingType { case int32(ctrl_pb.SettingTypes_NewCtrlAddress): newAddress := string(settingValue) - handler.newCtrlAddress(newAddress) + handler.updater.UpdateCtrlEndpoints([]string{newAddress}) default: log.Error("unknown setting type, ignored") } @@ -64,49 +54,8 @@ func (handler *settingsHandler) HandleReceive(msg *channel.Message, ch channel.C } } -func newSettingsHandler(config CtrlAddressChanger) channel.TypedReceiveHandler { +func newSettingsHandler(updater CtrlAddressUpdater) channel.TypedReceiveHandler { return &settingsHandler{ - config: config, + updater: updater, } } - -// newCtrlAddress interrogates the current configuration for controller ctrl address updates. -// If necessary it will create a backup of the current config, alter the runtime ctrl address, -// and save a new version of the router configuration in place. -func (handler *settingsHandler) newCtrlAddress(newAddress string) { - currentAddress := strings.TrimSpace(handler.config.CurrentCtrlAddress()) - newAddress = strings.TrimSpace(newAddress) - - log := pfxlog.Logger().WithFields(map[string]interface{}{ - "newAddress": newAddress, - "currentAddress": currentAddress, - }) - - log.Info("received new controller address header") - - if currentAddress == newAddress { - log.Warn("ignoring new controller address, same as current value") - return - } - - log.Info("new controller address detect creating backup and saving new configuration") - if destBackupPath, err := handler.config.CreateBackup(); err == nil { - log.WithField("backupPath", destBackupPath).Info("backup configuration created") - } else { - log.WithError(err).Error("could not create configuration backup") - return - } - - if err := handler.config.UpdateControllerEndpoint(newAddress); err != nil { - log.WithError(err).Error("could not update controller endpoint address") - return - } - - if err := handler.config.Save(); err != nil { - log.WithError(err).Error("could not save new configuration") - return - } - - log.Info("successfully saved new controller address") - -} diff --git a/router/handler_ctrl/updateCtrlAddresses.go b/router/handler_ctrl/updateCtrlAddresses.go index ba18dd461..03eb90b28 100644 --- a/router/handler_ctrl/updateCtrlAddresses.go +++ b/router/handler_ctrl/updateCtrlAddresses.go @@ -11,12 +11,11 @@ import ( var updateCtrlAddressesHandlerInstance *updateCtrlAddressesHandler type CtrlAddressUpdater interface { - UpdateCtrlEndpoints(endpoints []string) error + UpdateCtrlEndpoints(endpoints []string) } type updateCtrlAddressesHandler struct { - callback CtrlAddressUpdater - + callback CtrlAddressUpdater currentVersion uint64 } @@ -38,12 +37,9 @@ func (handler *updateCtrlAddressesHandler) HandleReceive(msg *channel.Message, c }) if upd.IsLeader || handler.currentVersion == 0 || handler.currentVersion < upd.Index { - if err := handler.callback.UpdateCtrlEndpoints(upd.Addresses); err != nil { - log.WithError(err).Error("Unable to update ctrl addresses") - } + log.Info("updating to controller endpoints to version") + handler.callback.UpdateCtrlEndpoints(upd.Addresses) handler.currentVersion = upd.Index - - log.Info("updated to version") } } diff --git a/router/handler_link/bind.go b/router/handler_link/bind.go index fdc802a0b..327423fa4 100644 --- a/router/handler_link/bind.go +++ b/router/handler_link/bind.go @@ -144,7 +144,12 @@ func (self *bindHandler) verifyRouter(l xlink.Xlink, ch channel.Channel) error { Fingerprints: fingerprints, } - reply, err := protobufs.MarshalTyped(verifyLink).WithTimeout(10 * time.Second).SendForReply(self.ctrl.AnyCtrlChannel()) + ctrlCh := self.ctrl.AnyCtrlChannel() + if ctrlCh == nil { + return errors.Errorf("unable to verify link %v, no controller available", l.Id()) + } + + reply, err := protobufs.MarshalTyped(verifyLink).WithTimeout(10 * time.Second).SendForReply(ctrlCh) if err != nil { return errors.Wrapf(err, "unable to verify router %v for link %v", l.DestinationId(), l.Id()) } diff --git a/router/router.go b/router/router.go index 7a6a62412..c510dfd6c 100644 --- a/router/router.go +++ b/router/router.go @@ -28,14 +28,11 @@ import ( "os" "path" "plugin" - "strings" - "sync" "sync/atomic" "time" gosundheit "github.com/AppsFlyer/go-sundheit" "github.com/AppsFlyer/go-sundheit/checks" - "github.com/cenkalti/backoff/v4" "github.com/michaelquigley/pfxlog" "github.com/openziti/channel/v2" "github.com/openziti/fabric/health" @@ -61,7 +58,6 @@ import ( "github.com/openziti/identity" "github.com/openziti/metrics" "github.com/openziti/transport/v2" - cmap "github.com/orcaman/concurrent-map/v2" "github.com/pkg/errors" "github.com/sirupsen/logrus" "google.golang.org/protobuf/proto" @@ -89,12 +85,6 @@ type Router struct { versionProvider versions.VersionProvider debugOperations map[byte]func(c *bufio.ReadWriter) error - ctrlEndpoints ctrlEndpoints - controllersToConnect struct { - controllers map[*UpdatableAddress]bool - mtx sync.Mutex - } - xwebs []xweb.Instance xwebFactoryRegistry xweb.Registry agentBindHandlers []channel.BindHandler @@ -169,40 +159,28 @@ func Create(config *Config, versionProvider versions.VersionProvider) *Router { metricsRegistry := metrics.NewUsageRegistry(config.Id.Token, map[string]string{}, closeNotify) xgress.InitMetrics(metricsRegistry) - ctrls := env.NewNetworkControllers(config.Ctrl.DefaultRequestTimeout, &config.Ctrl.Heartbeats) - faulter := forwarder.NewFaulter(ctrls, config.Forwarder.FaultTxInterval, closeNotify) - scanner := forwarder.NewScanner(ctrls, config.Forwarder, closeNotify) - fwd := forwarder.NewForwarder(metricsRegistry, faulter, scanner, config.Forwarder, closeNotify) - - xgress.InitPayloadIngester(closeNotify) - xgress.InitAcker(fwd, metricsRegistry, closeNotify) - xgress.InitRetransmitter(fwd, fwd, metricsRegistry, closeNotify) - router := &Router{ config: config, - ctrls: ctrls, - faulter: faulter, - scanner: scanner, - forwarder: fwd, metricsRegistry: metricsRegistry, shutdownC: closeNotify, shutdownDoneC: make(chan struct{}), versionProvider: versionProvider, debugOperations: map[byte]func(c *bufio.ReadWriter) error{}, xwebFactoryRegistry: xweb.NewRegistryMap(), - xlinkRegistry: NewLinkRegistry(ctrls), - ctrlEndpoints: newCtrlEndpoints(), - controllersToConnect: struct { - controllers map[*UpdatableAddress]bool - mtx sync.Mutex - }{ - controllers: make(map[*UpdatableAddress]bool), - mtx: sync.Mutex{}, - }, } + router.ctrls = env.NewNetworkControllers(config.Ctrl.DefaultRequestTimeout, router.connectToController, &config.Ctrl.Heartbeats) + router.xlinkRegistry = NewLinkRegistry(router.ctrls) + router.faulter = forwarder.NewFaulter(router.ctrls, config.Forwarder.FaultTxInterval, closeNotify) + router.scanner = forwarder.NewScanner(router.ctrls, config.Forwarder, closeNotify) + router.forwarder = forwarder.NewForwarder(metricsRegistry, router.faulter, router.scanner, config.Forwarder, closeNotify) + + xgress.InitPayloadIngester(closeNotify) + xgress.InitAcker(router.forwarder, metricsRegistry, closeNotify) + xgress.InitRetransmitter(router.forwarder, router.forwarder, metricsRegistry, closeNotify) + var err error - router.ctrlBindhandler, err = handler_ctrl.NewBindHandler(router, fwd, config, router) + router.ctrlBindhandler, err = handler_ctrl.NewBindHandler(router, router.forwarder, router) if err != nil { panic(err) } @@ -236,10 +214,6 @@ func (self *Router) Start() error { return err } - if err := self.initializeCtrlEndpoints(); err != nil { - return err - } - self.showOptions() self.startProfiling() @@ -269,7 +243,7 @@ func (self *Router) Start() error { go web.Run() } - if err := self.startControlPlane(); err != nil { + if err = self.startControlPlane(); err != nil { return err } return nil @@ -462,21 +436,15 @@ func (self *Router) startXgressListeners() { } func (self *Router) startControlPlane() error { - endpoints := self.ctrlEndpoints.Items() - if len(endpoints) == 0 { - return errors.New("no controller endpoints configured, exiting") + endpoints, err := self.getInitialCtrlEndpoints() + if err != nil { + return err } log := pfxlog.Logger() - log.Infof("router configured with %v controller endpoints", len(endpoints)) - for _, endpoint := range endpoints { - log.Infof("connecting to controller at endpoint [%v]", endpoint.String()) - if err := self.connectToController(endpoint, self.config.Ctrl.LocalBinding); err != nil { - return err - } - } + self.ctrls.UpdateControllerEndpoints(endpoints) for _, x := range self.xrctrls { if err := x.Run(self); err != nil { @@ -487,10 +455,16 @@ func (self *Router) startControlPlane() error { self.metricsReporter = fabricMetrics.NewControllersReporter(self.ctrls) self.metricsRegistry.StartReporting(self.metricsReporter, self.config.Metrics.ReportInterval, self.config.Metrics.MessageQueueSize) + time.AfterFunc(time.Second*15, func() { + if len(self.ctrls.GetAll()) == 0 { + pfxlog.Logger().Fatal("unable to connect to any controllers before timeout") + } + }) + return nil } -func (self *Router) connectToController(addr *UpdatableAddress, localBinding string) error { +func (self *Router) connectToController(addr transport.Address, bindHandler channel.BindHandler) error { attributes := map[int32][]byte{} version, err := self.versionProvider.EncoderDecoder().Encode(self.versionProvider.AsVersionInfo()) @@ -534,9 +508,10 @@ func (self *Router) connectToController(addr *UpdatableAddress, localBinding str if "" != self.config.Ctrl.LocalBinding { logrus.Debugf("Using local interface %s to dial controller", self.config.Ctrl.LocalBinding) } - dialer := channel.NewReconnectingDialerWithHandlerAndLocalBinding(self.config.Id, addr, localBinding, attributes, reconnectHandler) + dialer := channel.NewReconnectingDialerWithHandlerAndLocalBinding(self.config.Id, addr, self.config.Ctrl.LocalBinding, attributes, reconnectHandler) - ch, err := channel.NewChannel("ctrl", dialer, self.ctrlBindhandler, self.config.Ctrl.Options) + bindHandler = channel.BindHandlers(bindHandler, self.ctrlBindhandler) + ch, err := channel.NewChannel("ctrl", dialer, bindHandler, self.config.Ctrl.Options) if err != nil { return fmt.Errorf("error connecting ctrl (%v)", err) } @@ -545,65 +520,6 @@ func (self *Router) connectToController(addr *UpdatableAddress, localBinding str return nil } -func (self *Router) connectToControllerWithBackoff(addr *UpdatableAddress, localBinding string, maxTimeout *time.Duration) error { - self.controllersToConnect.mtx.Lock() - log := pfxlog.Logger() - defer func() { - self.controllersToConnect.mtx.Unlock() - }() - if _, exists := self.controllersToConnect.controllers[addr]; exists { - log.WithField("Controller Addr", addr).Info("Already attempting to connect") - return nil - } - - expBackoff := backoff.NewExponentialBackOff() - expBackoff.InitialInterval = 50 * time.Millisecond - expBackoff.MaxInterval = 5 * time.Minute - expBackoff.MaxElapsedTime = 365 * 24 * time.Hour - if maxTimeout != nil { - expBackoff.MaxElapsedTime = *maxTimeout - } - - operation := func() error { - self.controllersToConnect.mtx.Lock() - defer self.controllersToConnect.mtx.Unlock() - if _, exists := self.controllersToConnect.controllers[addr]; !exists { - return backoff.Permanent(errors.New("controller removed before connection established")) - } - err := self.connectToController(addr, localBinding) - if err != nil { - log. - WithField("Controller Addr", addr). - WithField("Local binding", localBinding). - WithError(err). - Error("Unable to connect controller") - } - return err - } - - self.controllersToConnect.controllers[addr] = true - log.WithField("Controller Addr", addr).Info("Starting connection attempts") - - go func() { - if err := backoff.Retry(operation, expBackoff); err != nil { - log. - WithField("Controller Addr", addr). - WithField("Local binding", localBinding). - WithError(err). - Error("Unable to connect controller. Stopping Retries.") - } else { - log.WithField("Controller Addr", addr). - WithField("Local binding", localBinding). - Info("Successfully connected to controller") - } - self.controllersToConnect.mtx.Lock() - delete(self.controllersToConnect.controllers, addr) - self.controllersToConnect.mtx.Unlock() - }() - - return nil -} - func (self *Router) initializeHealthChecks() (gosundheit.Health, error) { checkConfig := self.config.HealthChecks logrus.Infof("starting health check with ctrl ping initially after %v, then every %v, timing out after %v", @@ -646,96 +562,62 @@ func (self *Router) RegisterXWebHandlerFactory(x xweb.ApiHandlerFactory) error { return self.xwebFactoryRegistry.Add(x) } -func (self *Router) initializeCtrlEndpoints() error { +func (self *Router) getInitialCtrlEndpoints() ([]string, error) { log := pfxlog.Logger() if self.config.Ctrl.DataDir == "" { - return errors.New("ctrl DataDir not configured") + return nil, errors.New("ctrl DataDir not configured") } endpointsFile := path.Join(self.config.Ctrl.DataDir, "endpoints") + var endpoints []string + if _, err := os.Stat(endpointsFile); err != nil && errors.Is(err, fs.ErrNotExist) { log.Infof("controller endpoints file [%v] doesn't exist. Using initial endpoints from config", endpointsFile) for _, ep := range self.config.Ctrl.InitialEndpoints { - self.ctrlEndpoints.Set(ep.String(), ep) + endpoints = append(endpoints, ep.String()) } - return nil + return endpoints, nil } log.Infof("loading controller endpoints from [%v]", endpointsFile) b, err := os.ReadFile(endpointsFile) if err != nil { - return err + return nil, err } - if err = yaml.Unmarshal(b, &self.ctrlEndpoints); err != nil { - return err + endpointCfg := &endpointConfig{} + + if err = yaml.Unmarshal(b, endpointCfg); err != nil { + return nil, err } - if len(self.ctrlEndpoints.Items()) == 0 { - return errors.Errorf("no controller endpoints found in [%v], consider deleting file", endpointsFile) + endpoints = endpointCfg.Endpoints + + if len(endpoints) == 0 { + return nil, errors.Errorf("no controller endpoints found in [%v], consider deleting file", endpointsFile) } - //TODO: Handle mismatches - return nil + return endpoints, nil } -func (self *Router) UpdateCtrlEndpoints(endpoints []string) error { - log := pfxlog.Logger() - save := false - newEps := make(map[string]bool) - for _, ep := range endpoints { - newEps[ep] = true - } - for knownep := range self.ctrlEndpoints.Items() { - if _, ok := newEps[knownep]; !ok { - log.WithField("endpoint", knownep).Info("Removing old ctrl endpoint") - save = true - parsed, err := transport.ParseAddress(knownep) - if err != nil { - return err - } - _, parsedAddr, _ := strings.Cut(parsed.String(), ":") - - self.ctrlEndpoints.Remove(knownep) - self.controllersToConnect.mtx.Lock() - delete(self.controllersToConnect.controllers, NewUpdatableAddress(parsed)) - self.controllersToConnect.mtx.Unlock() - - if ch := self.ctrls.CloseAndRemoveByAddress(parsedAddr); ch != nil { - log.WithField("endpoint", knownep).WithError(err).Error("Unable to close ctrl channel to controller") - return err - } - } - } - for _, ep := range endpoints { - if !self.ctrlEndpoints.Has(ep) { - log.WithField("endpoint", ep).Info("Adding new ctrl endpoint") - save = true - parsed, err := transport.ParseAddress(ep) - if err != nil { - return err - } - upAddr := NewUpdatableAddress(parsed) - if err := self.connectToControllerWithBackoff(upAddr, self.config.Ctrl.LocalBinding, nil); err != nil { - log.WithError(err).Error("Unable to connect controller") - return err - } - self.ctrlEndpoints.Set(ep, upAddr) - } - } - - if save { - log.WithField("filepath", self.config.Ctrl.DataDir).Info("Attempting to save file") +func (self *Router) UpdateCtrlEndpoints(endpoints []string) { + log := pfxlog.Logger().WithField("endpoints", endpoints).WithField("filepath", self.config.Ctrl.DataDir) + if changed := self.ctrls.UpdateControllerEndpoints(endpoints); changed { + log.Info("Attempting to save file") endpointsFile := path.Join(self.config.Ctrl.DataDir, "endpoints") - data, err := self.ctrlEndpoints.MarshalYAML() - if err != nil { - return err + + configData := map[string]interface{}{ + "Endpoints": endpoints, + } + + if data, err := yaml.Marshal(configData); err != nil { + log.WithError(err).Error("unable to marshal updated controller endpoints to yaml") + } else if err = os.WriteFile(endpointsFile, data, 0600); err != nil { + log.WithError(err).Error("unable to write updated controller endpoints to file") } - return os.WriteFile(endpointsFile, data.([]byte), 0600) } - return nil } type connectionToggle interface { @@ -747,7 +629,7 @@ type controllerPinger struct { router *Router } -func (self *controllerPinger) PingContext(ctx context.Context) error { +func (self *controllerPinger) PingContext(context.Context) error { ctrls := self.router.ctrls.GetAll() if len(ctrls) == 0 { @@ -768,37 +650,6 @@ func (self *controllerPinger) PingContext(ctx context.Context) error { return errors.New("control channels are slow") } -type ctrlEndpoints struct { - cmap.ConcurrentMap[string, *UpdatableAddress] `yaml:"Endpoints"` -} - -func newCtrlEndpoints() ctrlEndpoints { - return ctrlEndpoints{cmap.New[*UpdatableAddress]()} -} - -// MarshalYAML handles serialization for the YAML format -func (c *ctrlEndpoints) MarshalYAML() (interface{}, error) { - data := make([]*UpdatableAddress, 0) - for _, ep := range c.Items() { - data = append(data, ep) - } - dat, err := yaml.Marshal(&struct { - Endpoints []*UpdatableAddress `yaml:"Endpoints,flow"` - }{ - Endpoints: data, - }) - return dat, err -} - -func (c *ctrlEndpoints) UnmarshalYAML(value *yaml.Node) error { - var endpoints struct { - Endpoints []*UpdatableAddress `yaml:"Endpoints,flow"` - } - if err := value.Decode(&endpoints); err != nil { - return err - } - for _, ep := range endpoints.Endpoints { - c.ConcurrentMap.Set(ep.String(), ep) - } - return nil +type endpointConfig struct { + Endpoints []string `yaml:"Endpoints"` } diff --git a/router/router_test.go b/router/router_test.go index d619d6080..342d47d7a 100644 --- a/router/router_test.go +++ b/router/router_test.go @@ -5,7 +5,6 @@ import ( "github.com/stretchr/testify/require" "os" "path" - "sync" "testing" "time" @@ -25,7 +24,7 @@ func Test_initializeCtrlEndpoints_ErrorsWithoutDataDir(t *testing.T) { r := Router{ config: &Config{}, } - err = r.initializeCtrlEndpoints() + _, err = r.getInitialCtrlEndpoints() assert.Error(t, err) assert.ErrorContains(t, err, "ctrl DataDir not configured") } @@ -55,12 +54,11 @@ func Test_initializeCtrlEndpoints(t *testing.T) { InitialEndpoints: []*UpdatableAddress{NewUpdatableAddress(addr)}, }, }, - ctrlEndpoints: newCtrlEndpoints(), } - expected := newCtrlEndpoints() - expected.Set(addr.String(), NewUpdatableAddress(addr)) - - assert.NoError(t, r.initializeCtrlEndpoints()) + expected := []string{addr.String()} + endpoints, err := r.getInitialCtrlEndpoints() + assert.NoError(t, err) + assert.Equal(t, expected, endpoints) assert.NoFileExists(t, path.Join(tmpDir, "endpoints")) } @@ -78,6 +76,10 @@ func Test_updateCtrlEndpoints(t *testing.T) { addr2, err := transport.ParseAddress("tls:localhost:6767") req.NoError(err) + ctrlDialer := func(address transport.Address, bindHandler channel.BindHandler) error { + return nil + } + r := Router{ config: &Config{ Ctrl: struct { @@ -92,40 +94,25 @@ func Test_updateCtrlEndpoints(t *testing.T) { InitialEndpoints: []*UpdatableAddress{NewUpdatableAddress(addr), NewUpdatableAddress(addr2)}, }, }, - ctrls: env.NewNetworkControllers(time.Minute, env.NewDefaultHeartbeatOptions()), - ctrlEndpoints: newCtrlEndpoints(), - controllersToConnect: struct { - controllers map[*UpdatableAddress]bool - mtx sync.Mutex - }{controllers: map[*UpdatableAddress]bool{}, mtx: sync.Mutex{}}, + ctrls: env.NewNetworkControllers(time.Minute, ctrlDialer, env.NewDefaultHeartbeatOptions()), } - expected := newCtrlEndpoints() - expected.Set(addr.String(), NewUpdatableAddress(addr)) - req.NoError(r.initializeCtrlEndpoints()) - - err = r.UpdateCtrlEndpoints([]string{"tls:localhost:6565"}) + endpoints, err := r.getInitialCtrlEndpoints() req.NoError(err) + r.UpdateCtrlEndpoints(endpoints) + + r.UpdateCtrlEndpoints([]string{"tls:localhost:6565"}) req.FileExists(path.Join(tmpDir, "endpoints")) b, err := os.ReadFile(path.Join(tmpDir, "endpoints")) req.NoError(err) req.NotEmpty(b) - //TODO: Figure out why we can't just unmarshal directly on struct - var holder = struct { - inner ctrlEndpoints - }{ - inner: newCtrlEndpoints(), - } + endpointCfg := &endpointConfig{} - out := newCtrlEndpoints() - err = yaml.Unmarshal(b, &holder.inner) + err = yaml.Unmarshal(b, &endpointCfg) req.NoError(err) - for k, v := range out.Items() { - req.True(expected.Has(k), "Expected to have addr %s", k) - expectedAddr, _ := expected.Get(k) - req.Equal(expectedAddr, v) - } + req.Equal(1, len(endpointCfg.Endpoints)) + req.Equal(addr.String(), endpointCfg.Endpoints[0]) } diff --git a/router/xlink_transport/perf_test.go b/router/xlink_transport/perf_test.go index 596a4164c..9b53fc249 100644 --- a/router/xlink_transport/perf_test.go +++ b/router/xlink_transport/perf_test.go @@ -61,6 +61,10 @@ func (t testUnderlay) SetWriteTimeout(time.Duration) error { return nil } +func (t testUnderlay) SetWriteDeadline(time.Time) error { + return nil +} + func (t testUnderlay) GetLocalAddr() net.Addr { panic("implement me") } diff --git a/tests/link_test.go b/tests/link_test.go index 64ba8ae63..9dd3eefef 100644 --- a/tests/link_test.go +++ b/tests/link_test.go @@ -69,7 +69,7 @@ func Test_LinkWithValidCertFromUnknownChain(t *testing.T) { "split": false, } metricsRegistery := metrics.NewRegistry("test", nil) - ctrls := env.NewNetworkControllers(time.Second, env.NewDefaultHeartbeatOptions()) + ctrls := env.NewNetworkControllers(time.Second, nil, env.NewDefaultHeartbeatOptions()) factory := xlink_transport.NewFactory(xla, testBindHandlerFactory{}, tcfg, router.NewLinkRegistry(ctrls), metricsRegistery) dialer, err := factory.CreateDialer(badId, nil, tcfg) ctx.Req.NoError(err) @@ -118,7 +118,7 @@ func Test_UnrequestedLinkFromValidRouter(t *testing.T) { } metricsRegistery := metrics.NewRegistry("test", nil) - ctrls := env.NewNetworkControllers(time.Second, env.NewDefaultHeartbeatOptions()) + ctrls := env.NewNetworkControllers(time.Second, nil, env.NewDefaultHeartbeatOptions()) factory := xlink_transport.NewFactory(xla, testBindHandlerFactory{}, tcfg, router.NewLinkRegistry(ctrls), metricsRegistery) dialer, err := factory.CreateDialer(router2Id, nil, tcfg) ctx.Req.NoError(err)