diff --git a/CODE_OF_CONDUCT.md b/CODE_OF_CONDUCT.md index 9eee7241c..367f70984 100644 --- a/CODE_OF_CONDUCT.md +++ b/CODE_OF_CONDUCT.md @@ -1,17 +1 @@ -# Code of Conduct - -All open source projects managed by NetFoundry share a common [code of -conduct](https://netfoundry.github.io/policies/CODE_OF_CONDUCT.html) which all contributors are expected to follow. -Please be sure you read, understand and adhere to the thoughts expressed therein. - -Individuals acting in any way that is considered in violation of the Code of Conduct will receive corrective actions -appropriate for the violation. It is possible to act in ways which technically do not violate the Code of Conduct but -are clearly against the spirit of the project. These situations will also be considered on a case by case basis and if -necessary the Code of Conduct will be revised accordingly and approrpiate notifications will be made to the community -and to the violator. - -## Violations - -Contributors violating the Code of Conduct will receive public notification of the violation. Contined violations or -aggregious violations of the Code of Conduct will result in the contributor being removed from the contributors list as -well as further contributions being declined. +All open source projects managed by OpenZiti share a common [code of conduct](https://docs.openziti.io/policies/CODE_OF_CONDUCT.html) which all contributors are expected to follow. Please be sure you read, understand and adhere to the guidelines expressed therein. diff --git a/controller/config.go b/controller/config.go index 32bc8ff46..03f11ebbb 100644 --- a/controller/config.go +++ b/controller/config.go @@ -86,8 +86,10 @@ type Config struct { // (e.g. NewListener) type CtrlOptions struct { *channel.Options - NewListener *transport.Address - AdvertiseAddress *transport.Address + NewListener *transport.Address + AdvertiseAddress *transport.Address + RouterHeartbeatOptions *channel.HeartbeatOptions + PeerHeartbeatOptions *channel.HeartbeatOptions } func (config *Config) Configure(sub config.Subconfig) error { @@ -321,7 +323,9 @@ func LoadConfig(path string) (*Config, error) { } controllerConfig.Ctrl.Options = &CtrlOptions{ - Options: channel.DefaultOptions(), + Options: channel.DefaultOptions(), + PeerHeartbeatOptions: channel.DefaultHeartbeatOptions(), + RouterHeartbeatOptions: channel.DefaultHeartbeatOptions(), } if value, found := submap["options"]; found { @@ -369,6 +373,26 @@ func LoadConfig(path string) (*Config, error) { } } + if value, found := submap["routerHeartbeats"]; found { + if submap, ok := value.(map[interface{}]interface{}); ok { + options, err := channel.LoadHeartbeatOptions(submap) + if err != nil { + return nil, err + } + controllerConfig.Ctrl.Options.RouterHeartbeatOptions = options + } + } + + if value, found := submap["peerHeartbeats"]; found { + if submap, ok := value.(map[interface{}]interface{}); ok { + options, err := channel.LoadHeartbeatOptions(submap) + if err != nil { + return nil, err + } + controllerConfig.Ctrl.Options.PeerHeartbeatOptions = options + } + } + if err := controllerConfig.Ctrl.Options.Validate(); err != nil { return nil, fmt.Errorf("error loading channel options for [ctrl/options] (%v)", err) } diff --git a/controller/controller.go b/controller/controller.go index a3d1f99cf..ef93ee7e8 100644 --- a/controller/controller.go +++ b/controller/controller.go @@ -251,11 +251,11 @@ func (c *Controller) Run() error { panic(err) } - ctrlAccepter := handler_ctrl.NewCtrlAccepter(c.network, c.xctrls, c.config.Ctrl.Options.Options, c.config.Trace.Handler) + ctrlAccepter := handler_ctrl.NewCtrlAccepter(c.network, c.xctrls, c.config.Ctrl.Options.Options, c.config.Ctrl.Options.RouterHeartbeatOptions, c.config.Trace.Handler) ctrlAcceptors := map[string]channel.UnderlayAcceptor{} if c.raftController != nil { - c.raftController.ConfigureMeshHandlers(handler_peer_ctrl.NewBindHandler(c.network, c.raftController)) + c.raftController.ConfigureMeshHandlers(handler_peer_ctrl.NewBindHandler(c.network, c.raftController, c.config.Ctrl.Options.PeerHeartbeatOptions)) ctrlAcceptors[mesh.ChannelTypeMesh] = c.raftController.GetMesh() } diff --git a/controller/handler_ctrl/accept.go b/controller/handler_ctrl/accept.go index 6e87aa11b..3852fcf2a 100644 --- a/controller/handler_ctrl/accept.go +++ b/controller/handler_ctrl/accept.go @@ -29,21 +29,24 @@ import ( ) type CtrlAccepter struct { - network *network.Network - xctrls []xctrl.Xctrl - options *channel.Options - traceHandler *channel.TraceHandler + network *network.Network + xctrls []xctrl.Xctrl + options *channel.Options + heartbeatOptions *channel.HeartbeatOptions + traceHandler *channel.TraceHandler } func NewCtrlAccepter(network *network.Network, xctrls []xctrl.Xctrl, options *channel.Options, + heartbeatOptions *channel.HeartbeatOptions, traceHandler *channel.TraceHandler) *CtrlAccepter { return &CtrlAccepter{ - network: network, - xctrls: xctrls, - options: options, - traceHandler: traceHandler, + network: network, + xctrls: xctrls, + options: options, + heartbeatOptions: heartbeatOptions, + traceHandler: traceHandler, } } @@ -101,7 +104,7 @@ func (self *CtrlAccepter) Bind(binding channel.Binding) error { r.Control = ch r.ConnectTime = time.Now() - if err := binding.Bind(newBindHandler(r, self.network, self.xctrls)); err != nil { + if err := binding.Bind(newBindHandler(self.heartbeatOptions, r, self.network, self.xctrls)); err != nil { return errors.Wrap(err, "error binding router") } diff --git a/controller/handler_ctrl/bind.go b/controller/handler_ctrl/bind.go index 034065ba2..13cd29d76 100644 --- a/controller/handler_ctrl/bind.go +++ b/controller/handler_ctrl/bind.go @@ -17,6 +17,7 @@ package handler_ctrl import ( + "github.com/sirupsen/logrus" "time" "github.com/michaelquigley/pfxlog" @@ -31,13 +32,19 @@ import ( ) type bindHandler struct { - router *network.Router - network *network.Network - xctrls []xctrl.Xctrl + heartbeatOptions *channel.HeartbeatOptions + router *network.Router + network *network.Network + xctrls []xctrl.Xctrl } -func newBindHandler(router *network.Router, network *network.Network, xctrls []xctrl.Xctrl) channel.BindHandler { - return &bindHandler{router: router, network: network, xctrls: xctrls} +func newBindHandler(heartbeatOptions *channel.HeartbeatOptions, router *network.Router, network *network.Network, xctrls []xctrl.Xctrl) channel.BindHandler { + return &bindHandler{ + heartbeatOptions: heartbeatOptions, + router: router, + network: network, + xctrls: xctrls, + } } func (self *bindHandler) BindChannel(binding channel.Binding) error { @@ -81,12 +88,14 @@ func (self *bindHandler) BindChannel(binding channel.Binding) error { if doHeartbeat { log.Info("router supports heartbeats") cb := &heartbeatCallback{ - latencyMetric: roundTripHistogram, - queueTimeMetric: queueTimeHistogram, - ch: binding.GetChannel(), - latencySemaphore: concurrenz.NewSemaphore(2), + latencyMetric: roundTripHistogram, + queueTimeMetric: queueTimeHistogram, + ch: binding.GetChannel(), + latencySemaphore: concurrenz.NewSemaphore(2), + closeUnresponsiveTimeout: self.heartbeatOptions.CloseUnresponsiveTimeout, + lastResponse: time.Now().Add(self.heartbeatOptions.CloseUnresponsiveTimeout * 2).UnixMilli(), // wait at least 2x timeout before closing } - channel.ConfigureHeartbeat(binding, 10*time.Second, time.Second, cb) + channel.ConfigureHeartbeat(binding, self.heartbeatOptions.SendInterval, self.heartbeatOptions.CheckInterval, cb) } else if supportLatency, err := self.router.VersionInfo.HasMinimumVersion("0.18.7"); supportLatency && err == nil { log.Info("router does not support heartbeats, using latency probe") latencyHandler := &ctrlChannelLatencyHandler{ @@ -129,19 +138,15 @@ func (self *ctrlChannelLatencyHandler) HandleLatency(latencyType latency.Type, e } type heartbeatCallback struct { - latencyMetric metrics.Histogram - queueTimeMetric metrics.Histogram - firstSent int64 - lastResponse int64 - ch channel.Channel - latencySemaphore concurrenz.Semaphore + latencyMetric metrics.Histogram + queueTimeMetric metrics.Histogram + lastResponse int64 + ch channel.Channel + latencySemaphore concurrenz.Semaphore + closeUnresponsiveTimeout time.Duration } -func (self *heartbeatCallback) HeartbeatTx(int64) { - if self.firstSent == 0 { - self.firstSent = time.Now().UnixMilli() - } -} +func (self *heartbeatCallback) HeartbeatTx(int64) {} func (self *heartbeatCallback) HeartbeatRx(int64) {} @@ -153,10 +158,14 @@ func (self *heartbeatCallback) HeartbeatRespRx(ts int64) { self.latencyMetric.Update(now.UnixNano() - ts) } +func (self *heartbeatCallback) timeSinceLastResponse(nowUnixMillis int64) time.Duration { + return time.Duration(nowUnixMillis-self.lastResponse) * time.Millisecond +} + func (self *heartbeatCallback) CheckHeartBeat() { - log := pfxlog.Logger().WithField("channelId", self.ch.Label()) now := time.Now().UnixMilli() - if self.firstSent != 0 && (now-self.firstSent > 30000) && (now-self.lastResponse > 30000) { + if self.timeSinceLastResponse(now) > self.closeUnresponsiveTimeout { + log := self.logger() log.Error("heartbeat not received in time, closing link") if err := self.ch.Close(); err != nil { log.WithError(err).Error("error while closing link") @@ -166,9 +175,8 @@ func (self *heartbeatCallback) CheckHeartBeat() { } func (self *heartbeatCallback) checkQueueTime() { - log := pfxlog.Logger().WithField("bindingId", self.ch.Id()) if !self.latencySemaphore.TryAcquire() { - log.Warn("unable to check queue time, too many check already running") + self.logger().Warn("unable to check queue time, too many check already running") return } @@ -181,6 +189,10 @@ func (self *heartbeatCallback) checkQueueTime() { StartTime: time.Now(), } if err := self.ch.Send(sendTracker); err != nil && !self.ch.IsClosed() { - log.WithError(err).Error("unable to send queue time tracer") + self.logger().WithError(err).Error("unable to send queue time tracer") } } + +func (self *heartbeatCallback) logger() *logrus.Entry { + return pfxlog.Logger().WithField("channelType", "router").WithField("channelId", self.ch.Id()) +} diff --git a/controller/handler_mgmt/connect.go b/controller/handler_mgmt/connect.go deleted file mode 100644 index 10c6ab8ec..000000000 --- a/controller/handler_mgmt/connect.go +++ /dev/null @@ -1,65 +0,0 @@ -/* - 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 handler_mgmt - -import ( - "crypto/x509" - "github.com/openziti/channel/v2" - "github.com/openziti/fabric/controller/network" - "github.com/openziti/foundation/v2/errorz" - "github.com/openziti/identity" - "github.com/pkg/errors" -) - -type ConnectHandler struct { - identity identity.Identity - network *network.Network -} - -func NewConnectHandler(identity identity.Identity, network *network.Network) *ConnectHandler { - return &ConnectHandler{ - identity: identity, - network: network, - } -} - -func (self *ConnectHandler) HandleConnection(hello *channel.Hello, certificates []*x509.Certificate) error { - if len(certificates) == 0 { - return errors.New("no certificates provided, unable to verify dialer") - } - - config := self.identity.ServerTLSConfig() - - opts := x509.VerifyOptions{ - Roots: config.RootCAs, - Intermediates: x509.NewCertPool(), - KeyUsages: []x509.ExtKeyUsage{x509.ExtKeyUsageAny}, - } - - var errorList errorz.MultipleErrors - - for _, cert := range certificates { - if _, err := cert.Verify(opts); err == nil { - return nil - } else { - errorList = append(errorList, err) - } - } - - //goland:noinspection GoNilness - return errorList.ToError() -} diff --git a/controller/handler_peer_ctrl/bind.go b/controller/handler_peer_ctrl/bind.go index eb809c14e..50483ed4b 100644 --- a/controller/handler_peer_ctrl/bind.go +++ b/controller/handler_peer_ctrl/bind.go @@ -17,20 +17,104 @@ package handler_peer_ctrl import ( + "github.com/michaelquigley/pfxlog" "github.com/openziti/channel/v2" + "github.com/openziti/channel/v2/latency" "github.com/openziti/fabric/controller/network" "github.com/openziti/fabric/controller/raft" + "github.com/openziti/foundation/v2/concurrenz" + "github.com/openziti/metrics" + "github.com/sirupsen/logrus" + "time" ) -func NewBindHandler(n *network.Network, raftCtrl *raft.Controller) channel.BindHandler { +func NewBindHandler(n *network.Network, raftCtrl *raft.Controller, heartbeatOptions *channel.HeartbeatOptions) channel.BindHandler { bindHandler := func(binding channel.Binding) error { binding.AddTypedReceiveHandler(newCommandHandler(raftCtrl)) binding.AddTypedReceiveHandler(newAddPeerHandler(raftCtrl)) binding.AddTypedReceiveHandler(newRemovePeerHandler(raftCtrl)) binding.AddTypedReceiveHandler(newTransferLeadershipHandler(raftCtrl)) binding.AddTypedReceiveHandler(newInspectHandler(n)) + + roundTripHistogram := n.GetMetricsRegistry().Histogram("peer.latency:" + binding.GetChannel().Id()) + queueTimeHistogram := n.GetMetricsRegistry().Histogram("peer.queue_time:" + binding.GetChannel().Id()) + binding.AddCloseHandler(channel.CloseHandlerF(func(ch channel.Channel) { + roundTripHistogram.Dispose() + queueTimeHistogram.Dispose() + })) + + cb := &heartbeatCallback{ + latencyMetric: roundTripHistogram, + queueTimeMetric: queueTimeHistogram, + ch: binding.GetChannel(), + latencySemaphore: concurrenz.NewSemaphore(2), + closeUnresponsiveTimeout: heartbeatOptions.CloseUnresponsiveTimeout, + lastResponse: time.Now().Add(heartbeatOptions.CloseUnresponsiveTimeout * 2).UnixMilli(), // wait at least 2x timeout before closing + } + + channel.ConfigureHeartbeat(binding, heartbeatOptions.SendInterval, heartbeatOptions.CheckInterval, cb) return nil } return channel.BindHandlerF(bindHandler) } + +type heartbeatCallback struct { + latencyMetric metrics.Histogram + queueTimeMetric metrics.Histogram + lastResponse int64 + ch channel.Channel + latencySemaphore concurrenz.Semaphore + closeUnresponsiveTimeout time.Duration +} + +func (self *heartbeatCallback) HeartbeatTx(int64) {} + +func (self *heartbeatCallback) HeartbeatRx(int64) {} + +func (self *heartbeatCallback) HeartbeatRespTx(int64) {} + +func (self *heartbeatCallback) HeartbeatRespRx(ts int64) { + now := time.Now() + self.lastResponse = now.UnixMilli() + self.latencyMetric.Update(now.UnixNano() - ts) +} + +func (self *heartbeatCallback) timeSinceLastResponse(nowUnixMillis int64) time.Duration { + return time.Duration(nowUnixMillis-self.lastResponse) * time.Millisecond +} + +func (self *heartbeatCallback) CheckHeartBeat() { + now := time.Now().UnixMilli() + if self.timeSinceLastResponse(now) > self.closeUnresponsiveTimeout { + log := self.logger() + log.Error("heartbeat not received in time, closing link") + if err := self.ch.Close(); err != nil { + log.WithError(err).Error("error while closing link") + } + } + go self.checkQueueTime() +} + +func (self *heartbeatCallback) checkQueueTime() { + if !self.latencySemaphore.TryAcquire() { + self.logger().Warn("unable to check queue time, too many check already running") + return + } + + defer self.latencySemaphore.Release() + + sendTracker := &latency.SendTimeTracker{ + Handler: func(latencyType latency.Type, latency time.Duration) { + self.queueTimeMetric.Update(latency.Nanoseconds()) + }, + StartTime: time.Now(), + } + if err := self.ch.Send(sendTracker); err != nil && !self.ch.IsClosed() { + self.logger().WithError(err).Error("unable to send queue time tracer") + } +} + +func (self *heartbeatCallback) logger() *logrus.Entry { + return pfxlog.Logger().WithField("channelType", "router").WithField("channelId", self.ch.Id()) +} diff --git a/router/config.go b/router/config.go index 027ce599f..5c6310598 100644 --- a/router/config.go +++ b/router/config.go @@ -19,6 +19,7 @@ package router import ( "bytes" "fmt" + "github.com/openziti/fabric/router/env" "io" "os" "path/filepath" @@ -111,6 +112,7 @@ type Config struct { DefaultRequestTimeout time.Duration Options *channel.Options DataDir string + Heartbeats env.HeartbeatOptions } Link struct { Listeners []map[interface{}]interface{} @@ -433,6 +435,8 @@ func LoadConfig(path string) (*Config, error) { cfg.Ctrl.DefaultRequestTimeout = 5 * time.Second cfg.Ctrl.Options = channel.DefaultOptions() + cfg.Ctrl.Heartbeats = *env.NewDefaultHeartbeatOptions() + if value, found := cfgmap[CtrlMapKey]; found { if submap, ok := value.(map[interface{}]interface{}); ok { if value, found := submap[CtrlEndpointBindMapKey]; found { @@ -478,6 +482,20 @@ func LoadConfig(path string) (*Config, error) { return nil, fmt.Errorf("error loading channel options for [ctrl/options] (%v)", err) } } + + if value, found := submap["heartbeats"]; found { + if submap, ok := value.(map[interface{}]interface{}); ok { + options, err := channel.LoadHeartbeatOptions(submap) + if err != nil { + return nil, err + } + heartbeats, err := env.NewHeartbeatOptions(options) + if err != nil { + return nil, err + } + cfg.Ctrl.Heartbeats = *heartbeats + } + } } if value, found := submap["defaultRequestTimeout"]; found { var err error diff --git a/router/env/ctrl.go b/router/env/ctrl.go index 93369b4bd..3b4e362a6 100644 --- a/router/env/ctrl.go +++ b/router/env/ctrl.go @@ -26,6 +26,31 @@ import ( "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 @@ -52,11 +77,12 @@ type NetworkControllers interface { } type networkCtrl struct { - ch channel.Channel - lastTx int64 - lastRx int64 - latency atomic.Int64 - unresponsive atomic.Bool + ch channel.Channel + heartbeatOptions *HeartbeatOptions + lastTx int64 + lastRx int64 + latency atomic.Int64 + unresponsive atomic.Bool } func (self *networkCtrl) HeartbeatCallback() channel.HeartbeatCallback { @@ -103,8 +129,7 @@ func (self *networkCtrl) HeartbeatRespRx(ts int64) { } func (self *networkCtrl) CheckHeartBeat() { - // TODO: Make configurable, see fabric#507 - if time.Duration(self.latency.Load()) > 5*time.Second { + if time.Duration(self.latency.Load()) > self.heartbeatOptions.UnresponsiveAfter { // if latency is greater than 5 seconds, consider this channel unresponsive self.unresponsive.Store(true) } else if self.lastTx > 0 && self.lastRx < self.lastTx && (time.Now().UnixMilli()-self.lastTx) > 5000 { @@ -115,20 +140,23 @@ func (self *networkCtrl) CheckHeartBeat() { } } -func NewNetworkControllers(defaultRequestTimeout time.Duration) NetworkControllers { +func NewNetworkControllers(defaultRequestTimeout time.Duration, heartbeatOptions *HeartbeatOptions) NetworkControllers { return &networkControllers{ + heartbeatOptions: heartbeatOptions, defaultRequestTimeout: defaultRequestTimeout, } } 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, + ch: ch, + heartbeatOptions: self.heartbeatOptions, } if existing := self.ctrls.Get(ctrl.Channel().Id()); existing != nil { return nil, fmt.Errorf("duplicate channel with id %v", ctrl.Channel().Id()) diff --git a/router/env/env.go b/router/env/env.go index 40e86aeb2..17cdecb44 100644 --- a/router/env/env.go +++ b/router/env/env.go @@ -35,4 +35,5 @@ type RouterEnv interface { GetCloseNotify() <-chan struct{} GetMetricsRegistry() metrics.UsageRegistry RenderJsonConfig() (string, error) + GetHeartbeatOptions() HeartbeatOptions } diff --git a/router/handler_ctrl/bind.go b/router/handler_ctrl/bind.go index 8b068aea2..3d34a812f 100644 --- a/router/handler_ctrl/bind.go +++ b/router/handler_ctrl/bind.go @@ -107,8 +107,7 @@ func (self *bindHandler) BindChannel(binding channel.Binding) error { return err } - // make configurable. see fabric#507 - channel.ConfigureHeartbeat(binding, 10*time.Second, time.Second, ctrl.HeartbeatCallback()) + channel.ConfigureHeartbeat(binding, self.env.GetHeartbeatOptions().SendInterval, self.env.GetHeartbeatOptions().CheckInterval, ctrl.HeartbeatCallback()) if self.env.GetTraceHandler() != nil { binding.AddPeekHandler(self.env.GetTraceHandler()) diff --git a/router/router.go b/router/router.go index 5399f503c..ea38179c9 100644 --- a/router/router.go +++ b/router/router.go @@ -154,6 +154,10 @@ func (self *Router) DefaultRequestTimeout() time.Duration { return self.config.Ctrl.DefaultRequestTimeout } +func (self *Router) GetHeartbeatOptions() env.HeartbeatOptions { + return self.config.Ctrl.Heartbeats +} + func Create(config *Config, versionProvider versions.VersionProvider) *Router { closeNotify := make(chan struct{}) @@ -165,7 +169,7 @@ 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) + 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) diff --git a/router/router_test.go b/router/router_test.go index 4721092c0..d619d6080 100644 --- a/router/router_test.go +++ b/router/router_test.go @@ -49,6 +49,7 @@ func Test_initializeCtrlEndpoints(t *testing.T) { DefaultRequestTimeout time.Duration Options *channel.Options DataDir string + Heartbeats env.HeartbeatOptions }{ DataDir: tmpDir, InitialEndpoints: []*UpdatableAddress{NewUpdatableAddress(addr)}, @@ -85,12 +86,13 @@ func Test_updateCtrlEndpoints(t *testing.T) { DefaultRequestTimeout time.Duration Options *channel.Options DataDir string + Heartbeats env.HeartbeatOptions }{ DataDir: tmpDir, InitialEndpoints: []*UpdatableAddress{NewUpdatableAddress(addr), NewUpdatableAddress(addr2)}, }, }, - ctrls: env.NewNetworkControllers(time.Minute), + ctrls: env.NewNetworkControllers(time.Minute, env.NewDefaultHeartbeatOptions()), ctrlEndpoints: newCtrlEndpoints(), controllersToConnect: struct { controllers map[*UpdatableAddress]bool diff --git a/tests/link_test.go b/tests/link_test.go index 35d91df29..64ba8ae63 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) + ctrls := env.NewNetworkControllers(time.Second, 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) + ctrls := env.NewNetworkControllers(time.Second, env.NewDefaultHeartbeatOptions()) factory := xlink_transport.NewFactory(xla, testBindHandlerFactory{}, tcfg, router.NewLinkRegistry(ctrls), metricsRegistery) dialer, err := factory.CreateDialer(router2Id, nil, tcfg) ctx.Req.NoError(err)