diff --git a/router/env/ctrls.go b/router/env/ctrls.go index 7811f4396..6034b3798 100644 --- a/router/env/ctrls.go +++ b/router/env/ctrls.go @@ -94,11 +94,32 @@ func NewNetworkControllers(defaultRequestTimeout time.Duration, dialer CtrlDiale heartbeatOptions: heartbeatOptions, defaultRequestTimeout: defaultRequestTimeout, ctrlEndpoints: cmap.New[struct{}](), + dialing: map[string]struct{}{}, + duplicates: map[string]string{}, } } +// errCtrlEndpointRemoved ends the dial of an endpoint that is no longer wanted. +var errCtrlEndpointRemoved = errors.New("controller removed before connection established") + +// duplicateCtrlError reports that an endpoint reached a controller the router is already connected to. +type duplicateCtrlError struct { + ctrlId string +} + +func (self *duplicateCtrlError) Error() string { + return fmt.Sprintf("duplicate channel with id %v", self.ctrlId) +} + type networkControllers struct { - lock sync.Mutex + // lock serializes changes to the wanted endpoints with registering a dialed controller, and guards + // dialing and duplicates. + lock sync.Mutex + // dialing holds the endpoints a dial is under way for. + dialing map[string]struct{} + // duplicates maps an endpoint whose dial reached an already connected controller to that + // controller's id. The endpoint is dialed again once that controller is no longer connected. + duplicates map[string]string ctrlDialer CtrlDialer heartbeatOptions *HeartbeatOptions defaultRequestTimeout time.Duration @@ -150,18 +171,28 @@ func (self *networkControllers) UpdateControllerEndpoints(addresses []string) bo delete(endpoints, endpoint) } else { // existing endpoint is no longer valid, close and remove it - log.WithField("endpoint", endpoints).Info("removing old ctrl endpoint") + log.WithField("endpoint", endpoint).Info("removing old ctrl endpoint") changed = true + delete(self.duplicates, endpoint) self.CloseAndRemoveByAddress(endpoint) } } for endpoint := range endpoints { - log.WithField("endpoint", endpoints).Info("adding new ctrl endpoint") + log.WithField("endpoint", endpoint).Info("adding new ctrl endpoint") changed = true self.connectToControllerWithBackoff(endpoint) } + for endpoint, ctrlId := range self.duplicates { + if ctrl := self.ctrls.Get(ctrlId); ctrl == nil || ctrl.Channel().IsClosed() { + log.WithField("endpoint", endpoint).WithField("ctrlId", ctrlId). + Info("controller the endpoint duplicated is no longer connected, dialing endpoint") + delete(self.duplicates, endpoint) + self.connectToControllerWithBackoff(endpoint) + } + } + return changed } @@ -177,6 +208,10 @@ func (self *networkControllers) UpdateLeader(leaderId string) { } } +// connectToControllerWithBackoff adds endpoint to the wanted endpoints and dials it until a controller +// is registered on it, unless a dial for it is already under way. The dial stops once the endpoint is +// no longer wanted, or when it reaches a controller the router is already connected to, recording the +// endpoint in duplicates. The caller holds lock. func (self *networkControllers) connectToControllerWithBackoff(endpoint string) { log := pfxlog.Logger().WithField("endpoint", endpoint) @@ -188,11 +223,15 @@ func (self *networkControllers) connectToControllerWithBackoff(endpoint string) 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 + if _, ok := self.dialing[endpoint]; ok { + log.Info("connection attempts already under way") + return + } + self.dialing[endpoint] = struct{}{} + // bindErr is the error from registering the controller on the last attempt. The bind runs on the + // dialing goroutine, within ctrlDialer. + var bindErr error bindHandler := channel.BindHandlerF(func(binding channel.Binding) error { id := binding.GetChannel().Id() binding.AddReceiveHandlerF(int32(edge_ctrl_pb.ContentType_CurrentIndexMessageType), func(m *channel.Message, ch channel.Channel) { @@ -203,15 +242,24 @@ func (self *networkControllers) connectToControllerWithBackoff(endpoint string) } }) - return self.Add(endpoint, binding.GetChannel()) + bindErr = self.Add(endpoint, binding.GetChannel()) + return bindErr }) operation := func() error { if !self.ctrlEndpoints.Has(endpoint) { - return backoff.Permanent(errors.New("controller removed before connection established")) + return backoff.Permanent(errCtrlEndpointRemoved) } + bindErr = nil err = self.ctrlDialer(addr, bindHandler) if err != nil { + // The dialer may not keep the bind error in its chain, so a refusal that ends the dial is + // taken from the bind itself. + var permanent *backoff.PermanentError + if errors.As(bindErr, &permanent) { + log.WithError(bindErr).Info("controller connection refused, not retrying") + return bindErr + } log.WithError(err).Error("unable to connect controller") } return err @@ -220,14 +268,52 @@ func (self *networkControllers) connectToControllerWithBackoff(endpoint string) 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") + for { + expBackoff := backoff.NewExponentialBackOff() + expBackoff.InitialInterval = 50 * time.Millisecond + expBackoff.MaxInterval = 5 * time.Minute + expBackoff.MaxElapsedTime = 100 * 365 * 24 * time.Hour + + err := backoff.Retry(operation, expBackoff) + if self.endDial(endpoint, err) { + if err != nil { + log.WithError(err).Error("unable to connect controller, stopping retries.") + } else { + log.Info("successfully connected to controller") + } + return + } + log.Info("endpoint still needs a connection, restarting connection attempts") } }() } +// endDial records that the dial of endpoint ended with err, and reports whether it should stop. It +// shouldn't if the endpoint was removed and then wanted again before the dial noticed, or if the +// controller it duplicated is already gone. +func (self *networkControllers) endDial(endpoint string, err error) bool { + self.lock.Lock() + defer self.lock.Unlock() + + wanted := self.ctrlEndpoints.Has(endpoint) + if errors.Is(err, errCtrlEndpointRemoved) && wanted { + return false + } + + var duplicate *duplicateCtrlError + if errors.As(err, &duplicate) && wanted { + if ctrl := self.ctrls.Get(duplicate.ctrlId); ctrl == nil || ctrl.Channel().IsClosed() { + return false + } + self.duplicates[endpoint] = duplicate.ctrlId + } + + delete(self.dialing, endpoint) + return true +} + +// Add registers ch as the controller dialed on address. It fails permanently if address is no longer +// wanted, or if a controller with ch's id is already registered on an open channel. func (self *networkControllers) Add(address string, ch channel.Channel) error { ctrl := newNetworkCtrl(ch, address, self.heartbeatOptions) @@ -241,19 +327,33 @@ func (self *networkControllers) Add(address string, ch channel.Channel) error { return errors.New("no version header provided") } - if existing := self.ctrls.Get(ch.Id()); existing != nil { - if !existing.Channel().IsClosed() { - // if an existing channel exists, don't keep trying to dial one - return backoff.Permanent(fmt.Errorf("duplicate channel with id %v", ctrl.Channel().Id())) - } + if err := self.register(address, ctrl); err != nil { + return err } - self.ctrls.Put(ch.Id(), ctrl) self.notifyOfChange(ctrl, ControllerAdded) return nil } +// register adds ctrl under lock, so it is either registered before its endpoint is removed, and closed +// by the removal, or refused after it. +func (self *networkControllers) register(address string, ctrl *networkCtrl) error { + self.lock.Lock() + defer self.lock.Unlock() + + if !self.ctrlEndpoints.Has(address) { + return backoff.Permanent(errCtrlEndpointRemoved) + } + + id := ctrl.Channel().Id() + if existing := self.ctrls.Get(id); existing != nil && !existing.Channel().IsClosed() { + return backoff.Permanent(&duplicateCtrlError{ctrlId: id}) + } + self.ctrls.Put(id, ctrl) + return nil +} + func (self *networkControllers) NotifyOfDisconnect(ctrlId string) { if ctrl := self.GetNetworkController(ctrlId); ctrl != nil { self.notifyOfChange(ctrl, ControllerDisconnected) diff --git a/router/env/ctrls_test.go b/router/env/ctrls_test.go new file mode 100644 index 000000000..8d1b6f1a3 --- /dev/null +++ b/router/env/ctrls_test.go @@ -0,0 +1,261 @@ +/* + 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" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/openziti/channel/v4" + "github.com/openziti/foundation/v2/versions" + "github.com/openziti/transport/v2" + "github.com/openziti/transport/v2/tls" + "github.com/stretchr/testify/require" +) + +type testCtrlUnderlay struct { + channel.Underlay + headers map[int32][]byte +} + +func (u *testCtrlUnderlay) Headers() map[int32][]byte { + return u.headers +} + +type testCtrlChannel struct { + channel.Channel + id string + closed atomic.Bool + underlay *testCtrlUnderlay +} + +func (c *testCtrlChannel) Id() string { + return c.id +} + +func (c *testCtrlChannel) Label() string { + return c.id +} + +func (c *testCtrlChannel) IsClosed() bool { + return c.closed.Load() +} + +func (c *testCtrlChannel) Close() error { + c.closed.Store(true) + return nil +} + +func (c *testCtrlChannel) Underlay() channel.Underlay { + return c.underlay +} + +type testCtrlBinding struct { + channel.Binding + ch channel.Channel +} + +func (b *testCtrlBinding) GetChannel() channel.Channel { + return b.ch +} + +func (b *testCtrlBinding) AddReceiveHandlerF(int32, channel.ReceiveHandlerF) {} + +// testCtrlDialer reaches the controller each endpoint names, counting dials. A gated endpoint's first +// dial waits until the gate is released. Bind errors are wrapped without keeping them in the chain. +type testCtrlDialer struct { + t *testing.T + lock sync.Mutex + ctrlIds map[string]string + dials map[string]int + gates map[string]chan struct{} + started map[string]chan struct{} +} + +func newTestCtrlDialer(t *testing.T) *testCtrlDialer { + return &testCtrlDialer{ + t: t, + ctrlIds: map[string]string{}, + dials: map[string]int{}, + gates: map[string]chan struct{}{}, + started: map[string]chan struct{}{}, + } +} + +// gate makes endpoint's first dial wait for release, and returns a channel closed once it starts. +func (d *testCtrlDialer) gate(endpoint string) (started <-chan struct{}, release func()) { + d.lock.Lock() + defer d.lock.Unlock() + gateC := make(chan struct{}) + startedC := make(chan struct{}) + d.gates[endpoint] = gateC + d.started[endpoint] = startedC + var once sync.Once + release = func() { + once.Do(func() { + close(gateC) + }) + } + d.t.Cleanup(release) + return startedC, release +} + +func (d *testCtrlDialer) dialCount(endpoint string) int { + d.lock.Lock() + defer d.lock.Unlock() + return d.dials[endpoint] +} + +func (d *testCtrlDialer) dial(address transport.Address, bindHandler channel.BindHandler) error { + endpoint := address.String() + + d.lock.Lock() + d.dials[endpoint]++ + ctrlId := d.ctrlIds[endpoint] + gateC := d.gates[endpoint] + startedC := d.started[endpoint] + delete(d.gates, endpoint) + d.lock.Unlock() + + if gateC != nil { + close(startedC) + <-gateC + } + + version, err := versions.StdVersionEncDec.Encode(&versions.VersionInfo{Version: "v1.6.0"}) + if err != nil { + return err + } + ch := &testCtrlChannel{ + id: ctrlId, + underlay: &testCtrlUnderlay{headers: map[int32][]byte{channel.HelloVersionHeader: version}}, + } + + if err = bindHandler.BindChannel(&testCtrlBinding{ch: ch}); err != nil { + _ = ch.Close() + return fmt.Errorf("error connecting ctrl (%v)", err) + } + return nil +} + +func newTestNetworkControllers(t *testing.T) (*networkControllers, *testCtrlDialer) { + transport.AddAddressParser(tls.AddressParser{}) + dialer := newTestCtrlDialer(t) + ctrls := NewNetworkControllers(time.Minute, dialer.dial, NewDefaultHeartbeatOptions()).(*networkControllers) + return ctrls, dialer +} + +func testCtrlEndpoint(name string) string { + return "tls:" + name + ":6262" +} + +func (self *networkControllers) isDialing(endpoint string) bool { + self.lock.Lock() + defer self.lock.Unlock() + _, ok := self.dialing[endpoint] + return ok +} + +func requireRegisteredAt(t *testing.T, ctrls *networkControllers, ctrlId string, endpoint string) { + require.Eventually(t, func() bool { + ctrl := ctrls.GetNetworkController(ctrlId) + return ctrl != nil && ctrl.Address() == endpoint && !ctrl.Channel().IsClosed() + }, 5*time.Second, time.Millisecond, "%s was not registered at %s", ctrlId, endpoint) +} + +// Test_NetworkControllers_ReaddedEndpointIsNotDialedTwice: an endpoint removed and added again while its +// dial is under way keeps that dial, rather than starting a second one that would reach the same +// controller. +func Test_NetworkControllers_ReaddedEndpointIsNotDialedTwice(t *testing.T) { + req := require.New(t) + ctrls, dialer := newTestNetworkControllers(t) + ep1, ep2 := testCtrlEndpoint("ctrl1"), testCtrlEndpoint("ctrl2") + dialer.ctrlIds[ep1] = "ctrl1" + dialer.ctrlIds[ep2] = "ctrl2" + started, release := dialer.gate(ep1) + + ctrls.UpdateControllerEndpoints([]string{ep1, ep2}) + select { + case <-started: + case <-time.After(5 * time.Second): + req.FailNow("the first dial did not start") + } + + ctrls.UpdateControllerEndpoints([]string{ep2}) + ctrls.UpdateControllerEndpoints([]string{ep1, ep2}) + release() + + requireRegisteredAt(t, ctrls, "ctrl1", ep1) + requireRegisteredAt(t, ctrls, "ctrl2", ep2) + req.Eventually(func() bool { return !ctrls.isDialing(ep1) }, 5*time.Second, time.Millisecond) + time.Sleep(200 * time.Millisecond) + req.Equal(1, dialer.dialCount(ep1), "an endpoint added back during its dial must not be dialed again") +} + +// Test_NetworkControllers_DuplicateEndpointStopsAndIsDialedWhenItsControllerGoes: an endpoint reaching a +// controller already connected on another endpoint stops dialing, and is dialed again once the endpoint +// that controller was connected on is removed. +func Test_NetworkControllers_DuplicateEndpointStopsAndIsDialedWhenItsControllerGoes(t *testing.T) { + req := require.New(t) + ctrls, dialer := newTestNetworkControllers(t) + viaName, viaAddress := testCtrlEndpoint("ctrl1.example.com"), testCtrlEndpoint("10.0.0.1") + dialer.ctrlIds[viaName] = "ctrl1" + dialer.ctrlIds[viaAddress] = "ctrl1" + + ctrls.UpdateControllerEndpoints([]string{viaName}) + requireRegisteredAt(t, ctrls, "ctrl1", viaName) + + ctrls.UpdateControllerEndpoints([]string{viaName, viaAddress}) + req.Eventually(func() bool { return !ctrls.isDialing(viaAddress) }, 5*time.Second, time.Millisecond, + "a dial reaching an already connected controller must stop") + time.Sleep(300 * time.Millisecond) + req.Equal(1, dialer.dialCount(viaAddress), "a dial reaching an already connected controller must not retry") + requireRegisteredAt(t, ctrls, "ctrl1", viaName) + + ctrls.UpdateControllerEndpoints([]string{viaAddress}) + requireRegisteredAt(t, ctrls, "ctrl1", viaAddress) + req.Equal(2, dialer.dialCount(viaAddress)) +} + +// Test_NetworkControllers_RemovedEndpointIsNotRegistered: a dial that completes after its endpoint was +// removed doesn't register the controller it reached. +func Test_NetworkControllers_RemovedEndpointIsNotRegistered(t *testing.T) { + req := require.New(t) + ctrls, dialer := newTestNetworkControllers(t) + ep1, ep2 := testCtrlEndpoint("ctrl1"), testCtrlEndpoint("ctrl2") + dialer.ctrlIds[ep1] = "ctrl1" + dialer.ctrlIds[ep2] = "ctrl2" + started, release := dialer.gate(ep1) + + ctrls.UpdateControllerEndpoints([]string{ep1}) + select { + case <-started: + case <-time.After(5 * time.Second): + req.FailNow("the first dial did not start") + } + + ctrls.UpdateControllerEndpoints([]string{ep2}) + release() + + requireRegisteredAt(t, ctrls, "ctrl2", ep2) + req.Eventually(func() bool { return !ctrls.isDialing(ep1) }, 5*time.Second, time.Millisecond) + req.Nil(ctrls.GetNetworkController("ctrl1"), "a controller reached on a removed endpoint must not be registered") + req.Equal([]string{ep2}, ctrls.GetEndpoints()) +} diff --git a/router/router.go b/router/router.go index c26f66f13..828dd5b1d 100644 --- a/router/router.go +++ b/router/router.go @@ -875,7 +875,7 @@ func (self *Router) connectToController(addr transport.Address, bindHandler chan 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) + return fmt.Errorf("error connecting ctrl (%w)", err) } channelRef.Store(ch)