gsadmin/ziti
1
0
Fork 0
mirror of https://github.com/openziti/ziti.git synced 2026-10-12 07:18:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add startupTimeout for routers

Browse Source
This commit is contained in:
Paul Lorenz
2024-01-30 22:28:46 -05:00
parent f31b4ce3c3
commit 89cc496a0a
7 changed files with 53 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
router/config.go
+8
View File
@@ -114,6 +114,7 @@ type Config struct {
Options *channel.Options
DataDir string
Heartbeats env.HeartbeatOptions
StartupTimeout time.Duration
}
Link struct {
Listeners []map[interface{}]interface{}
@@ -447,6 +448,7 @@ func LoadConfig(path string) (*Config, error) {
cfg.Ctrl.DefaultRequestTimeout = 5 * time.Second
cfg.Ctrl.Options = channel.DefaultOptions()
cfg.Ctrl.Heartbeats = *env.NewDefaultHeartbeatOptions()
cfg.Ctrl.StartupTimeout = 30 * time.Second
if value, found := cfgmap[CtrlMapKey]; found {
if submap, ok := value.(map[interface{}]interface{}); ok {
@@ -514,6 +516,12 @@ func LoadConfig(path string) (*Config, error) {
return nil, errors.Wrap(err, "invalid value for ctrl.defaultRequestTimeout")
}
}
if value, found := submap["startupTimeout"]; found {
var err error
if cfg.Ctrl.StartupTimeout, err = time.ParseDuration(value.(string)); err != nil {
return nil, errors.Wrap(err, "invalid value for ctrl.startupTimeout")
}
}
if value, found := submap["dataDir"]; found {
cfg.Ctrl.DataDir = value.(string)
} else {
router/router.go
+9 -7
View File
@@ -534,14 +534,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 !self.isShutdown.Load() && len(self.ctrls.GetAll()) == 0 {
if os.Getenv("STACKDUMP_ON_FAILED_STARTUP") == "true" {
debugz.DumpStack()
if self.config.Ctrl.StartupTimeout > 0 {
time.AfterFunc(self.config.Ctrl.StartupTimeout, func() {
if !self.isShutdown.Load() && len(self.ctrls.GetAll()) == 0 {
if os.Getenv("STACKDUMP_ON_FAILED_STARTUP") == "true" {
debugz.DumpStack()
}
pfxlog.Logger().Fatal("unable to connect to any controllers before timeout")
}
pfxlog.Logger().Fatal("unable to connect to any controllers before timeout")
}
})
})
}
_ = self.ctrls.AnyValidCtrlChannel()
for _, x := range self.xrctrls {
router/router_test.go
+2
View File
@@ -49,6 +49,7 @@ func Test_initializeCtrlEndpoints(t *testing.T) {
Options *channel.Options
DataDir string
Heartbeats env.HeartbeatOptions
StartupTimeout time.Duration
}{
DataDir: tmpDir,
InitialEndpoints: []*UpdatableAddress{NewUpdatableAddress(addr)},
@@ -89,6 +90,7 @@ func Test_updateCtrlEndpoints(t *testing.T) {
Options *channel.Options
DataDir string
Heartbeats env.HeartbeatOptions
StartupTimeout time.Duration
}{
DataDir: tmpDir,
InitialEndpoints: []*UpdatableAddress{NewUpdatableAddress(addr), NewUpdatableAddress(addr2)},
router/xlink_transport/dialer.go
+15 -5
View File
@@ -68,7 +68,7 @@ func (self *dialer) Dial(dial xlink.Dial) (xlink.Xlink, error) {
}
linkId := self.id.ShallowCloneWithNewToken(dial.GetLinkId())
connId := uuid.New().String()
connId := uuid.NewString()
var xli xlink.Xlink
if self.config.split {
@@ -92,7 +92,12 @@ func (self *dialer) Dial(dial xlink.Dial) (xlink.Xlink, error) {
}
func (self *dialer) dialSplit(linkId *identity.TokenId, address transport.Address, connId string, dial xlink.Dial) (xlink.Xlink, error) {
logrus.Debugf("dialing link with split payload/ack channels [l/%s]", linkId.Token)
log := pfxlog.Logger().WithFields(logrus.Fields{
"linkId": linkId.Token,
"connId": connId,
})
log.Info("dialing link with split payload/ack channels")
headers := channel.Headers{
LinkHeaderRouterId: []byte(self.id.Token),
@@ -105,7 +110,7 @@ func (self *dialer) dialSplit(linkId *identity.TokenId, address transport.Addres
payloadDialer := channel.NewClassicDialerWithBindAddress(linkId, address, self.config.localBinding, headers)
logrus.Debugf("dialing payload channel for [l/%s]", linkId.Token)
log.Info("dialing payload channel")
bindHandler := &splitDialBindHandler{
dialer: self,
@@ -126,7 +131,7 @@ func (self *dialer) dialSplit(linkId *identity.TokenId, address transport.Addres
return nil, errors.Wrapf(err, "error dialing payload channel for [l/%s]", linkId.Token)
}
logrus.Debugf("dialing ack channel for [l/%s]", linkId.Token)
log.Info("dialing ack channel")
headers = channel.Headers{
LinkHeaderRouterId: []byte(self.id.Token),
@@ -149,7 +154,12 @@ func (self *dialer) dialSplit(linkId *identity.TokenId, address transport.Addres
}
func (self *dialer) dialSingle(linkId *identity.TokenId, address transport.Address, connId string, dial xlink.Dial) (xlink.Xlink, error) {
logrus.Debugf("dialing link with single channel [l/%s]", linkId.Token)
log := pfxlog.Logger().WithFields(logrus.Fields{
"linkId": linkId.Token,
"connId": connId,
})
log.Info("dialing link with single channel")
headers := channel.Headers{
LinkHeaderRouterId: []byte(self.id.Token),
router/xlink_transport/listener.go
+17 -17
View File
@@ -106,20 +106,17 @@ func (self *listener) BindChannel(binding channel.Binding) error {
var iteration uint32
if headers != nil {
if v, ok := headers[LinkHeaderRouterId]; ok {
routerId = string(v)
var ok bool
if routerId, ok = headers.GetStringHeader(LinkHeaderRouterId); ok {
log = log.WithField("routerId", routerId)
log.Info("accepting link")
}
if val, ok := headers[LinkHeaderType]; ok {
chanType = channelType(val[0])
if val, ok := headers.GetByteHeader(LinkHeaderType); ok {
chanType = channelType(val)
}
if val, ok := headers[LinkHeaderRouterVersion]; ok {
routerVersion = string(val)
if routerVersion, ok = headers.GetStringHeader(LinkHeaderRouterVersion); ok {
log = log.WithField("routerVersion", routerVersion)
}
if val, ok := headers[LinkHeaderBinding]; ok {
dialerBinding = string(val)
if dialerBinding, ok = headers.GetStringHeader(LinkHeaderBinding); ok {
log = log.WithField("dialerBinding", dialerBinding)
}
if val, ok := headers.GetUint32Header(LinkHeaderIteration); ok {
@@ -128,6 +125,8 @@ func (self *listener) BindChannel(binding channel.Binding) error {
}
}
log.Info("binding link channel")
linkMeta := &linkMetadata{
routerId: routerId,
routerVersion: routerVersion,
@@ -145,14 +144,15 @@ func (self *listener) BindChannel(binding channel.Binding) error {
func (self *listener) bindSplitChannel(binding channel.Binding, chanType channelType, linkMeta *linkMetadata, log *logrus.Entry) error {
headers := binding.GetChannel().Underlay().Headers()
id, ok := headers[LinkHeaderConnId]
connId, ok := channel.Headers(headers).GetStringHeader(LinkHeaderConnId)
if !ok {
return errors.New("split conn received but missing connection id. closing")
}
log = log.WithField("connId", connId)
log.Info("accepted part of split conn")
complete, xli, err := self.getOrCreateSplitLink(string(id), linkMeta, binding, chanType)
complete, xli, err := self.getOrCreateSplitLink(connId, linkMeta, binding, chanType)
if err != nil {
log.WithError(err).Error("error binding link channel")
return err
@@ -160,7 +160,7 @@ func (self *listener) bindSplitChannel(binding channel.Binding, chanType channel
latencyPing := chanType == PayloadChannel
if err = self.bindHandlerFactory.NewBindHandler(xli, latencyPing, true).BindChannel(binding); err != nil {
self.cleanupDeadPartialLink(xli.id)
self.cleanupDeadPartialLink(connId)
if closeErr := xli.Close(); closeErr != nil {
log.WithError(closeErr).Error("error closing partial split link")
}
@@ -168,7 +168,7 @@ func (self *listener) bindSplitChannel(binding channel.Binding, chanType channel
}
if complete && xli.payloadCh != nil && xli.ackCh != nil {
if err := self.accepter.Accept(xli); err != nil {
if err = self.accepter.Accept(xli); err != nil {
log.WithError(err).Error("error accepting incoming Xlink")
if err := xli.Close(); err != nil {
@@ -195,15 +195,15 @@ func (self *listener) cleanupDeadPartialLink(id string) {
delete(self.pendingLinks, id)
}
func (self *listener) getOrCreateSplitLink(id string, linkMeta *linkMetadata, binding channel.Binding, chanType channelType) (bool, *splitImpl, error) {
func (self *listener) getOrCreateSplitLink(connId string, linkMeta *linkMetadata, binding channel.Binding, chanType channelType) (bool, *splitImpl, error) {
self.lock.Lock()
defer self.lock.Unlock()
complete := false
var link *splitImpl
if pending, found := self.pendingLinks[id]; found {
delete(self.pendingLinks, id)
if pending, found := self.pendingLinks[connId]; found {
delete(self.pendingLinks, connId)
link = pending.link
complete = true
} else {
@@ -220,7 +220,7 @@ func (self *listener) getOrCreateSplitLink(id string, linkMeta *linkMetadata, bi
},
eventTime: time.Now(),
}
self.pendingLinks[id] = pending
self.pendingLinks[connId] = pending
link = pending.link
}
ziti/constants/constants.go
+1
View File
@@ -20,6 +20,7 @@ import "time"
const (
ZITI = "ziti"
ZROK = "zrok"
ZITI_CONTROLLER = "ziti-controller"
ZITI_ROUTER = "ziti-router"
ZITI_TUNNEL = "ziti-tunnel"
zititest/models/links-test/configs/router.yml.tmpl
+1
View File
@@ -18,6 +18,7 @@ tls:
ctrl:
endpoints: {{ range $host := .Model.MustSelectHosts "component.ctrl" 1 }}
- tls:{{ $host.PublicIp }}:6262{{end}}
startupTimeout: 5m
healthChecks:
ctrlPingCheck: