Merge branch 'usage_reporting_interval' of github.com:openziti/fabric into usage_reporting_interval

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.

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

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

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

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

controller/handler_mgmt/connect.go

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

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

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

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

router/env/env.go

@@ -35,4 +35,5 @@ type RouterEnv interface {
 	GetCloseNotify() <-chan struct{}
 	GetMetricsRegistry() metrics.UsageRegistry
 	RenderJsonConfig() (string, error)
+	GetHeartbeatOptions() HeartbeatOptions
 }

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

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)

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

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)