diff --git a/controller/handler_edge_ctrl/heartbeat.go b/controller/handler_edge_ctrl/heartbeat.go index 8da447f0e..e5a020e38 100644 --- a/controller/handler_edge_ctrl/heartbeat.go +++ b/controller/handler_edge_ctrl/heartbeat.go @@ -38,17 +38,34 @@ 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). + Errorf("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) + + _ = ch.Send(msg) } } 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/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 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)