Merge pull request openziti/fabric#625 from openziti/endpoint-mgmt-refactor

Refactor endpoint management code. Fixes openziti/fabric#610

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
 }

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")
+			}
+		}
+	}
+}

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())

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")
-
-}

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")
 	}
 }
 

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())
 	}

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"`
 }

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])
 }

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")
 }

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)