diff --git a/router/xgress_edge_tunnel/factory.go b/router/xgress_edge_tunnel/factory.go index 6ba5a9d0f..53724a8b4 100644 --- a/router/xgress_edge_tunnel/factory.go +++ b/router/xgress_edge_tunnel/factory.go @@ -17,14 +17,15 @@ package xgress_edge_tunnel import ( + "github.com/michaelquigley/pfxlog" "github.com/openziti/channel" "github.com/openziti/edge/router/fabric" "github.com/openziti/edge/router/handler_edge_ctrl" "github.com/openziti/fabric/router" "github.com/openziti/fabric/router/xgress" "github.com/openziti/foundation/identity/identity" - "github.com/openziti/storage/boltz" "github.com/openziti/foundation/util/stringz" + "github.com/openziti/storage/boltz" "github.com/pkg/errors" "time" ) @@ -46,6 +47,8 @@ type Factory struct { } func (self *Factory) NotifyOfReconnect() { + pfxlog.Logger().Info("control channel reconnected, re-establishing hosted services") + self.tunneler.servicePoller.serviceListener.NotifyOfReconnect() } func (self *Factory) GetTraceDecoders() []channel.TraceMessageDecoder { diff --git a/tunnel/entities/service.go b/tunnel/entities/service.go index ece9a1a50..06ed2c07b 100644 --- a/tunnel/entities/service.go +++ b/tunnel/entities/service.go @@ -362,6 +362,7 @@ type Service struct { DialIdentityProvider TemplateFunc SourceAddrProvider TemplateFunc cleanupActions []func() + reconnectAction func() lock sync.Mutex } @@ -371,6 +372,12 @@ func (self *Service) AddCleanupAction(f func()) { self.cleanupActions = append(self.cleanupActions, f) } +func (self *Service) SetReconnectAction(f func()) { + self.lock.Lock() + defer self.lock.Unlock() + self.reconnectAction = f +} + func (self *Service) RunCleanupActions() { self.lock.Lock() defer self.lock.Unlock() @@ -378,6 +385,18 @@ func (self *Service) RunCleanupActions() { for _, action := range self.cleanupActions { action() } + + self.cleanupActions = nil + self.reconnectAction = nil +} + +func (self *Service) RunReconnectAction() { + self.lock.Lock() + reconnectAction := self.reconnectAction + self.lock.Unlock() + if reconnectAction != nil { + reconnectAction() + } } func (self *Service) GetSourceAddr(sourceAddr net.Addr, destAddr net.Addr) string { diff --git a/tunnel/intercept/svcpoll.go b/tunnel/intercept/svcpoll.go index 9c4f87fee..baaee3bef 100644 --- a/tunnel/intercept/svcpoll.go +++ b/tunnel/intercept/svcpoll.go @@ -87,6 +87,9 @@ func (self *ServiceListener) HandleProviderReady(provider tunnel.FabricProvider) } func (self *ServiceListener) HandleServicesChange(eventType ziti.ServiceEventType, service *edge.Service) { + self.Lock() + defer self.Unlock() + tunnelerService := &entities.Service{Service: *service} log := logrus.WithField("service", service.Name) @@ -231,6 +234,12 @@ func (self *ServiceListener) host(svc *entities.Service, tracker AddressTracker) } svc.AddCleanupAction(stopHook) + svc.SetReconnectAction(func() { + logger.Info("re-establishing hosting after reconnect") + stopHook() + self.host(svc, tracker) + }) + for idx, hostContext := range hostContexts { hostControl, err := self.provider.HostService(hostContext) if err != nil { @@ -257,6 +266,14 @@ func (self *ServiceListener) host(svc *entities.Service, tracker AddressTracker) } } +func (self *ServiceListener) NotifyOfReconnect() { + self.Lock() + defer self.Unlock() + for _, svc := range self.services { + svc.RunReconnectAction() + } +} + func (self *ServiceListener) configureSourceAddrProvider(svc *entities.Service) error { var err error if template := svc.GetSourceIpTemplate(); template != "" {