From 12ab2bbb34bec6c771ef2e9ee6321daf7391c9bd Mon Sep 17 00:00:00 2001 From: Andrew Martinez Date: Wed, 17 Feb 2021 10:54:07 -0500 Subject: [PATCH 1/4] do not remove api sessions added after/during sync --- router/handler_edge_ctrl/apiSessionAdded.go | 45 +++++++++++++-------- router/internal/fabric/manager.go | 22 ++++++---- 2 files changed, 43 insertions(+), 24 deletions(-) diff --git a/router/handler_edge_ctrl/apiSessionAdded.go b/router/handler_edge_ctrl/apiSessionAdded.go index 6660fa03e..feb10f8b0 100644 --- a/router/handler_edge_ctrl/apiSessionAdded.go +++ b/router/handler_edge_ctrl/apiSessionAdded.go @@ -37,7 +37,7 @@ type apiSessionAddedHandler struct { syncTracker *apiSessionSyncTracker reqChan chan *apiSessionAddedWithState - syncReady chan []*edge_ctrl_pb.ApiSession + syncReady chan *apiSessionSyncTracker syncFail chan error stop chan struct{} @@ -49,7 +49,7 @@ func NewApiSessionAddedHandler(sm fabric.StateManager, control channel2.Channel) sm: sm, reqChan: make(chan *apiSessionAddedWithState, 100), - syncReady: make(chan []*edge_ctrl_pb.ApiSession, 0), + syncReady: make(chan *apiSessionSyncTracker, 0), syncFail: make(chan error, 0), stop: make(chan struct{}, 0), @@ -109,13 +109,20 @@ func (h *apiSessionAddedHandler) startSyncApplier() { select { case <-h.stop: return - case apiSessions := <-h.syncReady: + case sync := <-h.syncReady: + lastId := "" + apiSessions := sync.all() for _, apiSession := range apiSessions { + if lastId == "" || apiSession.Id > lastId { + lastId = apiSession.Id + } + h.sm.AddApiSession(apiSession) } - h.sm.RemoveMissingApiSessions(apiSessions) - pfxlog.Logger().Infof("finished sychronizing api sessions [count: %d]", len(apiSessions)) + h.sm.RemoveMissingApiSessions(apiSessions, lastId) + + pfxlog.Logger().Infof("finished sychronizing api sessions [count: %d, syncId: %s]", len(apiSessions), sync.syncId) } } } @@ -149,7 +156,7 @@ func (h *apiSessionAddedHandler) legacySync(reqWithState *apiSessionAddedWithSta h.sm.AddApiSession(apiSession) } - h.sm.RemoveMissingApiSessions(reqWithState.ApiSessions) + h.sm.RemoveMissingApiSessions(reqWithState.ApiSessions, "") } func (h *apiSessionAddedHandler) startReceiveSync() { @@ -188,7 +195,7 @@ func (h *apiSessionAddedHandler) instantSync(reqWithState *apiSessionAddedWithSt if h.syncTracker == nil || h.syncTracker.syncId == "" { logger.Infof("first api session syncId [%s], starting", reqWithState.Id) - } else if h.syncTracker.isDone{ + } else if h.syncTracker.isDone { logger.Infof("api session syncId [%s], starting", reqWithState.Id) } else { logger.Infof("api session with newer syncId [old: %s, new: %s], aborting old, starting new", h.syncTracker.syncId, reqWithState.Id) @@ -214,13 +221,13 @@ func (h *apiSessionAddedHandler) instantSync(reqWithState *apiSessionAddedWithSt } type apiSessionSyncTracker struct { - syncId string - reqsWithState map[int]*apiSessionAddedWithState - hasLast bool - lastSeq int - stop chan struct{} - deadline sync.Once - isDone bool + syncId string + reqsWithState map[int]*apiSessionAddedWithState + hasLast bool + lastSeq int + stop chan struct{} + deadline sync.Once + isDone bool } func newApiSessionSyncTracker(id string) *apiSessionSyncTracker { @@ -251,7 +258,7 @@ func (tracker *apiSessionSyncTracker) IsDone() bool { return tracker.isDone } -func (tracker *apiSessionSyncTracker) StartDeadline(syncReady chan []*edge_ctrl_pb.ApiSession, syncFail chan error, timeout time.Duration) { +func (tracker *apiSessionSyncTracker) StartDeadline(syncReady chan *apiSessionSyncTracker, syncFail chan error, timeout time.Duration) { tracker.deadline.Do(func() { go func() { ticker := time.NewTicker(1 * time.Second) @@ -261,7 +268,8 @@ func (tracker *apiSessionSyncTracker) StartDeadline(syncReady chan []*edge_ctrl_ return case <-ticker.C: if tracker.HasAll() { - syncReady <- tracker.all() + syncReady <- tracker + tracker.isDone = true return } @@ -312,6 +320,11 @@ type apiSessionAddedWithState struct { *edge_ctrl_pb.ApiSessionAdded } +type apiSessionSync struct { + syncId string + apiSession []*edge_ctrl_pb.ApiSession +} + func parseInstantSyncHeaders(msg *channel2.Message) (string, *sync_strats.InstantSyncState, error) { if syncStrategyType, ok := msg.Headers[env.SyncStrategyTypeHeader]; ok { if syncStrategyState, ok := msg.Headers[env.SyncStrategyStateHeader]; ok { diff --git a/router/internal/fabric/manager.go b/router/internal/fabric/manager.go index 857562763..57c6dbe49 100644 --- a/router/internal/fabric/manager.go +++ b/router/internal/fabric/manager.go @@ -58,7 +58,7 @@ type StateManager interface { AddApiSession(apiSession *edge_ctrl_pb.ApiSession) UpdateApiSession(apiSession *edge_ctrl_pb.ApiSession) RemoveApiSession(token string) - RemoveMissingApiSessions(knownSessions []*edge_ctrl_pb.ApiSession) + RemoveMissingApiSessions(knownSessions []*edge_ctrl_pb.ApiSession, beforeSessionId string) AddConnectedApiSession(token string, removeCB func(), ch channel2.Channel) RemoveConnectedApiSession(token string, underlay channel2.Channel) AddApiSessionRemovedListener(token string, callBack func(token string)) RemoveListener @@ -67,8 +67,8 @@ type StateManager interface { } type StateManagerImpl struct { - sessionsByToken *sync.Map //"network" sessions - apiSessionsByToken *sync.Map + sessionsByToken *sync.Map //token -> "network" sessions + apiSessionsByToken *sync.Map // token -> api session activeApiSessions cmap.ConcurrentMap Hostname string @@ -141,16 +141,22 @@ func (sm *StateManagerImpl) RemoveApiSession(token string) { pfxlog.Logger().WithField("apiSessionToken", token).Debugf("could not remove api session [token: %s]; not found", token) } } -func (sm *StateManagerImpl) RemoveMissingApiSessions(knownSessions []*edge_ctrl_pb.ApiSession) { + +// Removes API Sessions not present in the knownApiSessions argument. If the beforeSessionId value is not empty string, +// it will be used as a monotonic comparison between it and API session ids. API session ids later than the sync +// will be ignored. +func (sm *StateManagerImpl) RemoveMissingApiSessions(knownApiSessions []*edge_ctrl_pb.ApiSession, beforeSessionId string) { validTokens := map[string]bool{} - for _, session := range knownSessions { - validTokens[session.Token] = true + for _, apiSession := range knownApiSessions { + validTokens[apiSession.Token] = true } var tokensToRemove []string - sm.apiSessionsByToken.Range(func(key, _ interface{}) bool { + sm.apiSessionsByToken.Range(func(key, val interface{}) bool { token, _ := key.(string) - if _, ok := validTokens[token]; !ok { + apiSession, _ := val.(*edge_ctrl_pb.ApiSession) + + if _, ok := validTokens[token]; !ok && (beforeSessionId == "" || apiSession.Id <= beforeSessionId) { tokensToRemove = append(tokensToRemove, token) } return true From 919d7a3d6b1ba84b23984b71dccf1c82f2f39c3a Mon Sep 17 00:00:00 2001 From: Andrew Martinez Date: Wed, 17 Feb 2021 12:19:06 -0500 Subject: [PATCH 2/4] add hearbeat invalid api token delete --- controller/handler_edge_ctrl/heartbeat.go | 25 ++++++++++++++++++++--- controller/model/api_session_handlers.go | 25 +++++++++++++++++++---- tests/service_test.go | 2 +- 3 files changed, 44 insertions(+), 8 deletions(-) diff --git a/controller/handler_edge_ctrl/heartbeat.go b/controller/handler_edge_ctrl/heartbeat.go index 8da447f0e..93be7a587 100644 --- a/controller/handler_edge_ctrl/heartbeat.go +++ b/controller/handler_edge_ctrl/heartbeat.go @@ -22,6 +22,7 @@ import ( "github.com/openziti/edge/controller/env" "github.com/openziti/edge/pb/edge_ctrl_pb" "github.com/openziti/foundation/channel2" + "time" ) type sessionHeartbeatHandler struct { @@ -38,17 +39,35 @@ func (h *sessionHeartbeatHandler) ContentType() int32 { return env.ApiSessionHeartbeatType } -func (h *sessionHeartbeatHandler) HandleReceive(msg *channel2.Message, _ channel2.Channel) { +func (h *sessionHeartbeatHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) { go func() { req := &edge_ctrl_pb.ApiSessionHeartbeat{} + routerId := ch.Id().Token if err := proto.Unmarshal(msg.Body, req); err == nil { - err := h.appEnv.GetHandlers().ApiSession.MarkActivity(req.Tokens) + notFoundTokens, err := h.appEnv.GetHandlers().ApiSession.MarkActivity(req.Tokens) if err != nil { pfxlog.Logger(). WithError(err). - Error("unable to set activity for heartbeat") + WithField("routerId", routerId). + Error("unable to set activity for heartbeat: %v", err) + } + + if len(notFoundTokens) > 0 { + pfxlog.Logger(). + WithField("routerId", routerId). + Debugf("api session tokens not found during heartbeat, sending delete: %v", notFoundTokens) + + msgStruct := &edge_ctrl_pb.ApiSessionRemoved{ + Tokens: notFoundTokens, + } + + msgBytes, _ := proto.Marshal(msgStruct) + msg := channel2.NewMessage(env.ApiSessionRemovedType, msgBytes) + + //if we don't delete it this time, we will try again when we see it + ch.SendWithTimeout(msg, 5*time.Second) } } else { diff --git a/controller/model/api_session_handlers.go b/controller/model/api_session_handlers.go index 6337d4021..7db2a447e 100644 --- a/controller/model/api_session_handlers.go +++ b/controller/model/api_session_handlers.go @@ -90,23 +90,40 @@ func (handler *ApiSessionHandler) Delete(id string) error { return handler.deleteEntity(id) } -func (handler *ApiSessionHandler) MarkActivity(tokens []string) error { - return handler.GetDb().Batch(func(tx *bbolt.Tx) error { +// MarkActivity returns tokens that were not found if any and/or an error. +func (handler *ApiSessionHandler) MarkActivity(tokens []string) ([]string, error) { + var notFoundTokens []string + + err := handler.GetDb().Batch(func(tx *bbolt.Tx) error { store := handler.Store.(persistence.ApiSessionStore) mutCtx := boltz.NewMutateContext(tx) for _, token := range tokens { apiSession, err := store.LoadOneByToken(tx, token) if err != nil { - return err + if boltz.IsErrNotFoundErr(err) { + notFoundTokens = append(notFoundTokens, token) + } else { + return err + } + } if err = store.Update(mutCtx, apiSession, persistence.UpdateTimeOnlyFieldChecker{}); err != nil { - return err + if err != nil { + if boltz.IsErrNotFoundErr(err) { + notFoundTokens = append(notFoundTokens, token) + } else { + return err + } + + } } handler.env.GetHandlers().Identity.SetActive(apiSession.IdentityId) } return nil }) + + return notFoundTokens, err } func (handler *ApiSessionHandler) Stream(query string, collect func(*ApiSession, error) error) error { diff --git a/tests/service_test.go b/tests/service_test.go index 29ea71cea..f61b1bd4c 100644 --- a/tests/service_test.go +++ b/tests/service_test.go @@ -168,7 +168,7 @@ func Test_Services(t *testing.T) { entityJson := ctx.AdminSession.validateEntityWithQuery(service) createdAt := ctx.validateDateFieldsForCreate(now, entityJson) - time.Sleep(time.Millisecond * 10) + time.Sleep(time.Millisecond * 100) now = time.Now() service.terminatorStrategy = "ha" ctx.AdminSession.requireUpdateEntity(service) From d2774ae41d342fdaab78d43d8bcaca517f3a9329 Mon Sep 17 00:00:00 2001 From: Andrew Martinez Date: Wed, 17 Feb 2021 12:55:58 -0500 Subject: [PATCH 3/4] fix lint errors --- controller/handler_edge_ctrl/heartbeat.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/controller/handler_edge_ctrl/heartbeat.go b/controller/handler_edge_ctrl/heartbeat.go index 93be7a587..37484e6f8 100644 --- a/controller/handler_edge_ctrl/heartbeat.go +++ b/controller/handler_edge_ctrl/heartbeat.go @@ -51,7 +51,7 @@ func (h *sessionHeartbeatHandler) HandleReceive(msg *channel2.Message, ch channe pfxlog.Logger(). WithError(err). WithField("routerId", routerId). - Error("unable to set activity for heartbeat: %v", err) + Errorf("unable to set activity for heartbeat: %v", err) } if len(notFoundTokens) > 0 { From 4b75b638da4a5e5a05a645a97ae091803b9e2615 Mon Sep 17 00:00:00 2001 From: Andrew Martinez Date: Wed, 17 Feb 2021 13:02:49 -0500 Subject: [PATCH 4/4] use send instead of send w/ tx --- controller/handler_edge_ctrl/heartbeat.go | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/controller/handler_edge_ctrl/heartbeat.go b/controller/handler_edge_ctrl/heartbeat.go index 37484e6f8..e5a020e38 100644 --- a/controller/handler_edge_ctrl/heartbeat.go +++ b/controller/handler_edge_ctrl/heartbeat.go @@ -22,7 +22,6 @@ import ( "github.com/openziti/edge/controller/env" "github.com/openziti/edge/pb/edge_ctrl_pb" "github.com/openziti/foundation/channel2" - "time" ) type sessionHeartbeatHandler struct { @@ -66,8 +65,7 @@ func (h *sessionHeartbeatHandler) HandleReceive(msg *channel2.Message, ch channe msgBytes, _ := proto.Marshal(msgStruct) msg := channel2.NewMessage(env.ApiSessionRemovedType, msgBytes) - //if we don't delete it this time, we will try again when we see it - ch.SendWithTimeout(msg, 5*time.Second) + _ = ch.Send(msg) } } else {