Merge pull request #683 from openziti/session-sync-fixes

Session sync fixes

controller/internal/routes/identity_api_model.go

@@ -176,13 +176,15 @@ func MapPatchIdentityToModel(id string, identity *rest_model.IdentityPatch, iden
 			Tags: identity.Tags,
 			Id:   id,
 		},
-		Name:                     identity.Name,
-		IdentityTypeId:           identityTypeId,
-		IsAdmin:                  identity.IsAdmin,
-		RoleAttributes:           identity.RoleAttributes,
-		DefaultHostingPrecedence: ziti.GetPrecedenceForLabel(string(identity.DefaultHostingPrecedence)),
-		DefaultHostingCost:       getDefaultHostingCost(identity.DefaultHostingCost),
-		AppData:                  identity.AppData,
+		Name:                      identity.Name,
+		IdentityTypeId:            identityTypeId,
+		IsAdmin:                   identity.IsAdmin,
+		RoleAttributes:            identity.RoleAttributes,
+		DefaultHostingPrecedence:  ziti.GetPrecedenceForLabel(string(identity.DefaultHostingPrecedence)),
+		DefaultHostingCost:        getDefaultHostingCost(identity.DefaultHostingCost),
+		ServiceHostingPrecedences: getServiceHostingPrecedences(identity.ServiceHostingPrecedences),
+		ServiceHostingCosts:       getServiceHostingCosts(identity.ServiceHostingCosts),
+		AppData:                   identity.AppData,
 	}
 
 	return ret

controller/internal/routes/identity_router.go

@@ -23,6 +23,7 @@ import (
 	"github.com/openziti/edge/controller/env"
 	"github.com/openziti/edge/controller/internal/permissions"
 	"github.com/openziti/edge/controller/model"
+	"github.com/openziti/edge/controller/persistence"
 	"github.com/openziti/edge/controller/response"
 	"github.com/openziti/edge/rest_management_api_server/operations/identity"
 	"github.com/openziti/edge/rest_model"
@@ -181,7 +182,8 @@ func (r *IdentityRouter) Update(ae *env.AppEnv, rc *response.RequestContext, par
 
 func (r *IdentityRouter) Patch(ae *env.AppEnv, rc *response.RequestContext, params identity.PatchIdentityParams) {
 	Patch(rc, func(id string, fields JsonFields) error {
-		return ae.Handlers.Identity.Patch(MapPatchIdentityToModel(params.ID, params.Identity, getIdentityTypeId(ae, params.Identity.Type)), fields.FilterMaps("tags", "appData"))
+		fields = fields.FilterMaps(boltz.FieldTags, persistence.FieldIdentityAppData, persistence.FieldIdentityServiceHostingCosts, persistence.FieldIdentityServiceHostingPrecedences)
+		return ae.Handlers.Identity.Patch(MapPatchIdentityToModel(params.ID, params.Identity, getIdentityTypeId(ae, params.Identity.Type)), fields)
 	})
 }
 

controller/server/controller.go

@@ -262,6 +262,10 @@ func (c *Controller) Run() {
 	if err := c.AppEnv.HostController.RegisterXWebHandlerFactory(clientApiFactory); err != nil {
 		pfxlog.Logger().Fatalf("failed to create Edge Client API factory: %v", err)
 	}
+
+	if err = c.policyEngine.Start(c.AppEnv.HostController.GetCloseNotifyChannel()); err != nil {
+		log.WithError(err).Fatalf("error starting policy engine")
+	}
 }
 
 func (c *Controller) Shutdown() {

controller/sync_strats/rtx.go

@@ -24,6 +24,7 @@ import (
 	"github.com/openziti/fabric/controller/network"
 	"github.com/openziti/foundation/channel2"
 	"github.com/openziti/foundation/util/concurrenz"
+	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"sync"
 )
@@ -78,9 +79,7 @@ func (rtx *RouterSender) run() {
 		case <-rtx.closeNotify:
 			return
 		case msg := <-rtx.send:
-			if !rtx.Router.Control.IsClosed() {
-				_ = rtx.Router.Control.Send(msg)
-			}
+			_ = rtx.Router.Control.Send(msg)
 		}
 	}
 }
@@ -94,16 +93,16 @@ func (rtx *RouterSender) logger() *logrus.Entry {
 		WithField("routerChannelIsOpen", !rtx.Router.Control.IsClosed())
 }
 
-func (rtx *RouterSender) Send(msg *channel2.Message) {
+func (rtx *RouterSender) Send(msg *channel2.Message) error {
 	if !rtx.running.Get() {
-		pfxlog.Logger().Errorf("cannot send to router [%s], rtx'er is stopped", rtx.Router.Id)
-		return
+		rtx.logger().Errorf("cannot send to router [%s], rtx'er is stopped", rtx.Router.Id)
+		return errors.Errorf("cannot send to router [%s], rtx'er is stopped", rtx.Router.Id)
 	}
 
 	if rtx.Router.Control.IsClosed() {
-		pfxlog.Logger().Errorf("cannot send to router [%s], rtx's channel is closed", rtx.Router.Id)
+		rtx.logger().Errorf("cannot send to router [%s], rtx's channel is closed", rtx.Router.Id)
 		rtx.Stop()
-		return
+		return errors.Errorf("cannot send to router [%s], rtx's channel is closed", rtx.Router.Id)
 	}
 
 	select {
@@ -111,10 +110,7 @@ func (rtx *RouterSender) Send(msg *channel2.Message) {
 	case <-rtx.closeNotify:
 	}
 
-}
-
-func (rtx *RouterSender) IsConnected() bool {
-	return !rtx.Router.Control.IsClosed()
+	return nil
 }
 
 // Map used make working with internal RouterSender easier as sync.Map accepts and returns interface{}
@@ -140,16 +136,15 @@ func (m *routerTxMap) GetState(id string) env.RouterStateValues {
 }
 
 func (m *routerTxMap) Remove(id string) {
-	entry := m.Get(id)
-	if entry != nil {
+	if val, found := m.internalMap.LoadAndDelete(id); found {
+		entry := val.(*RouterSender)
 		entry.Stop()
-		m.internalMap.Delete(id)
 	}
 }
 
 // Range creates a snapshot of the rtx's to loop over and will execute callback
 // with each rtx.
-func (m *routerTxMap) Range(callback func(entries *RouterSender) bool) {
+func (m *routerTxMap) Range(callback func(entries *RouterSender)) {
 	var rtxs []*RouterSender
 
 	m.internalMap.Range(func(edgeRouterId, value interface{}) bool {
@@ -159,7 +154,7 @@ func (m *routerTxMap) Range(callback func(entries *RouterSender) bool) {
 			pfxlog.Logger().Errorf("could not convert edge router entry got %t: %v", value, value)
 		}
 
-		return false
+		return true
 	})
 
 	for _, rtx := range rtxs {

controller/sync_strats/sync_instant.go

@@ -199,9 +199,8 @@ func (strategy *InstantStrategy) ApiSessionAdded(apiSession *persistence.ApiSess
 		Sequence: 0,
 	}
 
-	strategy.rtxMap.Range(func(rtx *RouterSender) bool {
-		strategy.sendApiSessionAdded(rtx, false, state, []*edge_ctrl_pb.ApiSession{apiSessionProto})
-		return true
+	strategy.rtxMap.Range(func(rtx *RouterSender) {
+		_ = strategy.sendApiSessionAdded(rtx, false, state, []*edge_ctrl_pb.ApiSession{apiSessionProto})
 	})
 }
 
@@ -221,13 +220,12 @@ func (strategy *InstantStrategy) ApiSessionUpdated(apiSession *persistence.ApiSe
 		ApiSessions: []*edge_ctrl_pb.ApiSession{apiSessionProto},
 	}
 
-	strategy.rtxMap.Range(func(rtx *RouterSender) bool {
+	strategy.rtxMap.Range(func(rtx *RouterSender) {
 		content, _ := proto.Marshal(apiSessionAdded)
 		msg := channel2.NewMessage(env.ApiSessionUpdatedType, content)
 		msg.Headers[env.SyncStrategyTypeHeader] = []byte(strategy.Type())
 		msg.Headers[env.SyncStrategyStateHeader] = nil
-		rtx.Send(msg)
-		return true
+		_ = rtx.Send(msg)
 	})
 }
 
@@ -236,11 +234,10 @@ func (strategy *InstantStrategy) ApiSessionDeleted(apiSession *persistence.ApiSe
 		Tokens: []string{apiSession.Token},
 	}
 
-	strategy.rtxMap.Range(func(rtx *RouterSender) bool {
+	strategy.rtxMap.Range(func(rtx *RouterSender) {
 		content, _ := proto.Marshal(sessionRemoved)
 		msg := channel2.NewMessage(env.ApiSessionRemovedType, content)
-		rtx.Send(msg)
-		return true
+		_ = rtx.Send(msg)
 	})
 }
 
@@ -249,11 +246,10 @@ func (strategy *InstantStrategy) SessionDeleted(session *persistence.Session) {
 		Tokens: []string{session.Token},
 	}
 
-	strategy.rtxMap.Range(func(rtx *RouterSender) bool {
+	strategy.rtxMap.Range(func(rtx *RouterSender) {
 		content, _ := proto.Marshal(sessionRemoved)
 		msg := channel2.NewMessage(env.SessionRemovedType, content)
-		rtx.Send(msg)
-		return true
+		_ = rtx.Send(msg)
 	})
 }
 
@@ -362,7 +358,6 @@ func (strategy *InstantStrategy) ReceiveResync(r *network.Router, _ *edge_ctrl_p
 }
 
 func (strategy *InstantStrategy) ReceiveClientHello(r *network.Router, respHello *edge_ctrl_pb.ClientHello) {
-
 	rtx := strategy.rtxMap.Get(r.Id)
 
 	if rtx == nil {
@@ -451,7 +446,9 @@ func (strategy *InstantStrategy) synchronize(rtx *RouterSender) {
 
 			if len(apiSessions) >= chunkSize {
 				state.IsLast = !cursor.IsValid()
-				strategy.sendApiSessionAdded(rtx, true, state, apiSessions)
+				if err = strategy.sendApiSessionAdded(rtx, true, state, apiSessions); err != nil {
+					return err
+				}
 
 				state.Sequence = state.Sequence + 1
 				apiSessions = []*edge_ctrl_pb.ApiSession{}
@@ -460,19 +457,22 @@ func (strategy *InstantStrategy) synchronize(rtx *RouterSender) {
 
 		if len(apiSessions) > 0 {
 			state.IsLast = true
-			strategy.sendApiSessionAdded(rtx, true, state, apiSessions)
+			if err := strategy.sendApiSessionAdded(rtx, true, state, apiSessions); err != nil {
+				return err
+			}
 		}
 		return nil
 	})
 
 	if err != nil {
 		logger.WithError(err).Error("failure synchronizing api sessions")
+		rtx.SetSyncStatus(env.RouterSyncError)
+	} else {
+		rtx.SetSyncStatus(env.RouterSyncDone)
 	}
-
-	rtx.SetSyncStatus(env.RouterSyncDone)
 }
 
-func (strategy *InstantStrategy) sendApiSessionAdded(rtx *RouterSender, isFullState bool, state *InstantSyncState, apiSessions []*edge_ctrl_pb.ApiSession) {
+func (strategy *InstantStrategy) sendApiSessionAdded(rtx *RouterSender, isFullState bool, state *InstantSyncState, apiSessions []*edge_ctrl_pb.ApiSession) error {
 	stateBytes, _ := json.Marshal(state)
 
 	msgContent := &edge_ctrl_pb.ApiSessionAdded{
@@ -487,7 +487,7 @@ func (strategy *InstantStrategy) sendApiSessionAdded(rtx *RouterSender, isFullSt
 	msg.Headers[env.SyncStrategyTypeHeader] = []byte(strategy.Type())
 	msg.Headers[env.SyncStrategyStateHeader] = stateBytes
 
-	rtx.Send(msg)
+	return rtx.Send(msg)
 }
 
 type InstantSyncState struct {

go.mod

@@ -40,9 +40,9 @@ require (
 	github.com/miekg/dns v1.1.42
 	github.com/mitchellh/mapstructure v1.4.1
 	github.com/netfoundry/secretstream v0.1.2
-	github.com/openziti/fabric v0.16.64
+	github.com/openziti/fabric v0.16.65
 	github.com/openziti/foundation v0.15.53
-	github.com/openziti/sdk-golang v0.15.50
+	github.com/openziti/sdk-golang v0.15.51
 	github.com/orcaman/concurrent-map v0.0.0-20210106121528-16402b402231
 	github.com/pkg/errors v0.9.1
 	github.com/rcrowley/go-metrics v0.0.0-20200313005456-10cdbea86bc0

go.sum

@@ -487,15 +487,12 @@ github.com/openzipkin/zipkin-go v0.1.1/go.mod h1:NtoC/o8u3JlF1lSlyPNswIbeQH9bJTm
 github.com/openziti/bbolt v1.3.6-0.20210317142109-547da822475e h1:ST9+54UtCr96S5+Wa7LQtpGS6/xK7Z+KQXvOAtYgJlA=
 github.com/openziti/bbolt v1.3.6-0.20210317142109-547da822475e/go.mod h1:G5EMThwa9y8QZGBClrRx5EY+Yw9kAhnjy3bSjsnlVTQ=
 github.com/openziti/dilithium v0.3.3/go.mod h1:vsCjI2AU/hon9e+dLhUFbCNGesJDj2ASgkySOcpmvjo=
-github.com/openziti/fabric v0.16.64 h1:arJPbmAzTJ0nF3jCyYU6AH4J94QXoz0uHYMErKssMgc=
-github.com/openziti/fabric v0.16.64/go.mod h1:WX5RRBP4jEMaF9R7yHK/q2C+qFuzgN7k/Y97tfO5XQE=
 github.com/openziti/fabric v0.16.65 h1:cAt+wvp7a1gUUCbtHFVdw827z+alW7JJER33mKWJbUk=
 github.com/openziti/fabric v0.16.65/go.mod h1:WX5RRBP4jEMaF9R7yHK/q2C+qFuzgN7k/Y97tfO5XQE=
-github.com/openziti/foundation v0.15.52/go.mod h1:NnSBTp9ZkcYsv6nYxZR0/NzoHfkIo6tjpBHDvBsPFqw=
 github.com/openziti/foundation v0.15.53 h1:r5DTm1G/L6thdKAVjnh58RrksOSlN+9hcvzFoecGezI=
 github.com/openziti/foundation v0.15.53/go.mod h1:h8d47dCV3vgwAkY4pkQrgpu3wBblDwUrNb/uvDAs0sU=
-github.com/openziti/sdk-golang v0.15.50 h1:BeKlp17UpQHGFDUvXn/Yn+q9r/wxSdhCs0TubQOp4Es=
-github.com/openziti/sdk-golang v0.15.50/go.mod h1:nA/ik6YZPTcLNdiq5tKjIPjjLCTtOEsfFYVl8NpAF2Q=
+github.com/openziti/sdk-golang v0.15.51 h1:QOt4G/MVH6sAQ84hreul8pVIlSIqdWaih7s4GHBo76Q=
+github.com/openziti/sdk-golang v0.15.51/go.mod h1:BuUe+RiRfR/cVVetvVROMOia2Qlq+EQSjQZ+UItQKwg=
 github.com/orcaman/concurrent-map v0.0.0-20190826125027-8c72a8bb44f6/go.mod h1:Lu3tH6HLW3feq74c2GC+jIMS/K2CFcDWnWD9XkenwhI=
 github.com/orcaman/concurrent-map v0.0.0-20210106121528-16402b402231 h1:fa50YL1pzKW+1SsBnJDOHppJN9stOEwS+CRWyUtyYGU=
 github.com/orcaman/concurrent-map v0.0.0-20210106121528-16402b402231/go.mod h1:Lu3tH6HLW3feq74c2GC+jIMS/K2CFcDWnWD9XkenwhI=