Restablish router/tunneler hosting when the ctrl ch reconnects

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 {

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 {

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 != "" {