From 661be7876d746a36bfd996454001e220bb48dd00 Mon Sep 17 00:00:00 2001 From: Andrew Martinez Date: Mon, 1 May 2023 10:02:48 -0400 Subject: [PATCH 1/2] fixes openziti/ziti#1093 - heart beat system was never being configured with a control channel --- controller/env/appenv.go | 2 +- controller/handler_edge_ctrl/common_tunnel.go | 4 +- controller/handler_edge_ctrl/heartbeat.go | 6 +- .../internal/routes/identity_api_model.go | 2 +- controller/model/api_session_manager.go | 30 +++++-- controller/model/identity_manager.go | 52 ++++++++---- controller/model/identity_model.go | 4 +- router/fabric/heartbeat.go | 21 +++-- router/fabric/manager.go | 7 +- router/internal/edgerouter/config.go | 2 +- router/xgress_edge/factory.go | 4 +- tests/ats-edge.router.yml | 1 + tests/identity_api_session_activity_test.go | 82 +++++++++++++++++++ 13 files changed, 169 insertions(+), 48 deletions(-) create mode 100644 tests/identity_api_session_activity_test.go diff --git a/controller/env/appenv.go b/controller/env/appenv.go index 13217e56b..68812ec7f 100644 --- a/controller/env/appenv.go +++ b/controller/env/appenv.go @@ -265,7 +265,7 @@ func (ae *AppEnv) FillRequestContext(rc *response.RequestContext) error { if rc.ApiSession != nil { //updates for api session timeouts - ae.GetManagers().ApiSession.MarkActivityById(rc.ApiSession.Id) + ae.GetManagers().ApiSession.MarkLastActivityById(rc.ApiSession.Id) var err error rc.Identity, err = ae.GetManagers().Identity.Read(rc.ApiSession.IdentityId) diff --git a/controller/handler_edge_ctrl/common_tunnel.go b/controller/handler_edge_ctrl/common_tunnel.go index 40556b25e..0fb508e1c 100644 --- a/controller/handler_edge_ctrl/common_tunnel.go +++ b/controller/handler_edge_ctrl/common_tunnel.go @@ -101,7 +101,7 @@ func (self *baseTunnelRequestContext) ensureApiSessionLocking(configTypes []stri if apiSession != nil && apiSession.IdentityId == self.identity.Id { self.apiSession = apiSession - if _, err := self.handler.getAppEnv().GetManagers().ApiSession.MarkActivityByTokens(self.apiSession.Token); err != nil { + if _, _, err := self.handler.getAppEnv().GetManagers().ApiSession.MarkLastActivityByTokens(self.apiSession.Token); err != nil { logger.WithError(err).Error("unexpected error while marking api session activity") } return false @@ -136,7 +136,7 @@ func (self *baseTunnelRequestContext) ensureApiSessionLocking(configTypes []stri apiSession, _ := self.handler.getAppEnv().Managers.ApiSession.Read(*cachedApiSessionId) if apiSession != nil && apiSession.IdentityId == self.identity.Id { self.apiSession = apiSession - if _, err := self.handler.getAppEnv().GetManagers().ApiSession.MarkActivityByTokens(self.apiSession.Token); err != nil { + if _, _, err := self.handler.getAppEnv().GetManagers().ApiSession.MarkLastActivityByTokens(self.apiSession.Token); err != nil { logger.WithError(err).Error("unexpected error while marking api session activity") } return true diff --git a/controller/handler_edge_ctrl/heartbeat.go b/controller/handler_edge_ctrl/heartbeat.go index 29c8f9a5c..107f4cbfb 100644 --- a/controller/handler_edge_ctrl/heartbeat.go +++ b/controller/handler_edge_ctrl/heartbeat.go @@ -44,7 +44,7 @@ func (h *sessionHeartbeatHandler) HandleReceive(msg *channel.Message, ch channel routerId := ch.Id() if err := proto.Unmarshal(msg.Body, req); err == nil { - notFoundTokens, err := h.appEnv.GetManagers().ApiSession.MarkActivityByTokens(req.Tokens...) + identityIds, notFoundTokens, err := h.appEnv.GetManagers().ApiSession.MarkLastActivityByTokens(req.Tokens...) if err != nil { pfxlog.Logger(). @@ -53,6 +53,10 @@ func (h *sessionHeartbeatHandler) HandleReceive(msg *channel.Message, ch channel Errorf("unable to set activity for heartbeat: %v", err) } + for _, identityId := range identityIds { + h.appEnv.GetManagers().Identity.SetHasErConnection(identityId) + } + if len(notFoundTokens) > 0 { pfxlog.Logger(). WithField("routerId", routerId). diff --git a/controller/internal/routes/identity_api_model.go b/controller/internal/routes/identity_api_model.go index 4a957987e..62d9a0371 100644 --- a/controller/internal/routes/identity_api_model.go +++ b/controller/internal/routes/identity_api_model.go @@ -282,7 +282,7 @@ func MapIdentityToRestModel(ae *env.AppEnv, identity *model.Identity) (*rest_mod RoleAttributes: &roleAttributes, Type: ToEntityRef(identityType.Name, identityType, IdentityTypeLinkFactory), TypeID: &identityType.Id, - HasEdgeRouterConnection: &identity.HasHeartbeat, + HasEdgeRouterConnection: &identity.HasErConnection, HasAPISession: &hasApiSession, DefaultHostingPrecedence: rest_model.TerminatorPrecedence(identity.DefaultHostingPrecedence.String()), DefaultHostingCost: &cost, diff --git a/controller/model/api_session_manager.go b/controller/model/api_session_manager.go index 3492def3f..612c36866 100644 --- a/controller/model/api_session_manager.go +++ b/controller/model/api_session_manager.go @@ -79,7 +79,7 @@ func (self *ApiSessionManager) CreateInCtx(ctx boltz.MutateContext, entity *ApiS } } - self.MarkActivityById(apiSessionId) + self.MarkLastActivityById(apiSessionId) return apiSessionId, err } @@ -134,16 +134,23 @@ func (self *ApiSessionManager) DeleteBatch(id []string, ctx *change.Context) err return self.deleteEntityBatch(id, ctx) } -func (self *ApiSessionManager) MarkActivityById(apiSessionId string) { +// MarkLastActivityById marks the "last activity" of an API Session. This will store a cached "LastUpdatedAt" value for +// an API Session. This data will be used to populate information for API Sessions and will be persisted to the data +// store at a future time in bulk. +func (self *ApiSessionManager) MarkLastActivityById(apiSessionId string) { self.HeartbeatCollector.Mark(apiSessionId) } -// MarkActivityByTokens returns tokens that were not found if any and/or an error. -func (self *ApiSessionManager) MarkActivityByTokens(tokens ...string) ([]string, error) { +// MarkLastActivityByTokens returns the ids of identities that were affected, tokens that were not found if any or an error. +// Marking "last activity" will store a cached "LastUpdatedAt" value for an API Session. This data will be used to +// populate information for API Sessions and will be persisted to the data store at a future time in bulk. +func (self *ApiSessionManager) MarkLastActivityByTokens(tokens ...string) ([]string, []string, error) { var notFoundTokens []string - store := self.Store.(persistence.ApiSessionStore) + store := self.env.GetStores().ApiSession var apiSessions []*persistence.ApiSession + identityIds := map[string]struct{}{} + err := self.GetDb().View(func(tx *bbolt.Tx) error { for _, token := range tokens { apiSession, err := store.LoadOneByToken(tx, token) @@ -161,17 +168,22 @@ func (self *ApiSessionManager) MarkActivityByTokens(tokens ...string) ([]string, }) for _, apiSession := range apiSessions { - self.HeartbeatCollector.Mark(apiSession.Id) - self.env.GetManagers().Identity.SetActive(apiSession.IdentityId) + self.MarkLastActivityById(apiSession.Id) + identityIds[apiSession.IdentityId] = struct{}{} } - return notFoundTokens, err + var uniqueIdentityIds []string + for identityId := range identityIds { + uniqueIdentityIds = append(uniqueIdentityIds, identityId) + } + + return uniqueIdentityIds, notFoundTokens, err } func (self *ApiSessionManager) heartbeatFlush(beats []*Heartbeat) { changeCtx := change.New().SetSourceType("heartbeat.flush").SetChangeAuthorType(change.AuthorTypeController) err := self.GetDb().Batch(changeCtx.NewMutateContext(), func(ctx boltz.MutateContext) error { - store := self.Store.(persistence.ApiSessionStore) + store := self.env.GetStores().ApiSession for _, beat := range beats { err := store.Update(ctx, &persistence.ApiSession{ diff --git a/controller/model/identity_manager.go b/controller/model/identity_manager.go index 9c2958be7..f8a33d9ac 100644 --- a/controller/model/identity_manager.go +++ b/controller/model/identity_manager.go @@ -463,12 +463,15 @@ func (self *IdentityManager) PatchInfo(identity *Identity, changeCtx *change.Con return err } -func (self *IdentityManager) SetActive(id string) { - self.identityStatusMap.SetActive(id) +// SetHasErConnection will register an identity as having an ER connection. The registration has a TTL depending on +// how the status map was configured. +func (self *IdentityManager) SetHasErConnection(identityId string) { + self.identityStatusMap.SetHasEdgeRouterConnection(identityId) } -func (self *IdentityManager) IsActive(id string) bool { - return self.identityStatusMap.IsActive(id) +// HasErConnection will return true if the supplied identity id has a current an active ER connection registered. +func (self *IdentityManager) HasErConnection(id string) bool { + return self.identityStatusMap.HasEdgeRouterConnection(id) } func (self *IdentityManager) VisitIdentityAuthenticatorFingerprints(tx *bbolt.Tx, identityId string, visitor func(string) bool) (bool, error) { @@ -760,9 +763,9 @@ func (self *CreateIdentityWithEnrollmentsCmd) GetChangeContext() *change.Context } type identityStatusMap struct { - identityIdToStatus cmap.ConcurrentMap[string, *status] - initOnce sync.Once - activeDuration time.Duration + identityIdToErConStatus cmap.ConcurrentMap[string, *status] + initOnce sync.Once + activeDuration time.Duration } type status struct { @@ -771,23 +774,34 @@ type status struct { func newIdentityStatusMap(activeDuration time.Duration) *identityStatusMap { return &identityStatusMap{ - identityIdToStatus: cmap.New[*status](), - activeDuration: activeDuration, + identityIdToErConStatus: cmap.New[*status](), + activeDuration: activeDuration, } } -func (statusMap *identityStatusMap) SetActive(identityId string) { +func (statusMap *identityStatusMap) SetHasEdgeRouterConnection(identityId string) { statusMap.initOnce.Do(statusMap.start) - statusMap.identityIdToStatus.Set(identityId, &status{ + statusMap.identityIdToErConStatus.Set(identityId, &status{ expiresAt: time.Now().Add(statusMap.activeDuration), }) } -func (statusMap *identityStatusMap) IsActive(identityId string) bool { - if stat, ok := statusMap.identityIdToStatus.Get(identityId); ok { - return stat.expiresAt.After(time.Now()) +func (statusMap *identityStatusMap) HasEdgeRouterConnection(identityId string) bool { + if stat, ok := statusMap.identityIdToErConStatus.Get(identityId); ok { + now := time.Now() + ret := stat.expiresAt.After(now) + pfxlog.Logger(). + WithField("identityId", identityId). + WithField("expiresAt", stat.expiresAt). + WithField("now", now). + Debugf("reporting identity from active ER conn pool: timedout") + return ret } + + pfxlog.Logger(). + WithField("identityId", identityId). + Debugf("reporting identity from active ER conn pool: not found") return false } @@ -797,14 +811,20 @@ func (statusMap *identityStatusMap) start() { for range ticker.C { var toRemove []string now := time.Now() - statusMap.identityIdToStatus.IterCb(func(key string, stat *status) { + statusMap.identityIdToErConStatus.IterCb(func(key string, stat *status) { if stat.expiresAt.Before(now) { + pfxlog.Logger(). + WithField("identityId", key). + WithField("expiresAt", stat.expiresAt). + WithField("now", now). + Debugf("removing identity from active ER conn pool: not found") toRemove = append(toRemove, key) } }) for _, identityId := range toRemove { - statusMap.identityIdToStatus.Remove(identityId) + + statusMap.identityIdToErConStatus.Remove(identityId) } } }() diff --git a/controller/model/identity_model.go b/controller/model/identity_model.go index bd9fe5000..bfbe5d1e9 100644 --- a/controller/model/identity_model.go +++ b/controller/model/identity_model.go @@ -51,7 +51,7 @@ type Identity struct { RoleAttributes []string EnvInfo *EnvInfo SdkInfo *SdkInfo - HasHeartbeat bool + HasErConnection bool DefaultHostingPrecedence ziti.Precedence DefaultHostingCost uint16 ServiceHostingPrecedences map[string]ziti.Precedence @@ -237,7 +237,7 @@ func (entity *Identity) fillFrom(env Env, _ *bbolt.Tx, boltIdentity *persistence entity.IsDefaultAdmin = boltIdentity.IsDefaultAdmin entity.IsAdmin = boltIdentity.IsAdmin entity.RoleAttributes = boltIdentity.RoleAttributes - entity.HasHeartbeat = env.GetManagers().Identity.IsActive(entity.Id) + entity.HasErConnection = env.GetManagers().Identity.HasErConnection(entity.Id) entity.DefaultHostingPrecedence = boltIdentity.DefaultHostingPrecedence entity.DefaultHostingCost = boltIdentity.DefaultHostingCost entity.ServiceHostingPrecedences = boltIdentity.ServiceHostingPrecedences diff --git a/router/fabric/heartbeat.go b/router/fabric/heartbeat.go index b439f23fb..765b2d207 100644 --- a/router/fabric/heartbeat.go +++ b/router/fabric/heartbeat.go @@ -17,19 +17,20 @@ package fabric import ( - "google.golang.org/protobuf/proto" "github.com/michaelquigley/pfxlog" "github.com/openziti/channel/v2" "github.com/openziti/edge/controller/env" "github.com/openziti/edge/pb/edge_ctrl_pb" "github.com/openziti/edge/runner" + routerEnv "github.com/openziti/fabric/router/env" + "google.golang.org/protobuf/proto" "time" ) const maxTokensPerMessage = 10000 type heartbeatOperation struct { - ctrl channel.Channel + env routerEnv.RouterEnv *runner.BaseOperation tokenProvider TokenProvider } @@ -39,9 +40,9 @@ type TokenProvider interface { flushRecentlyRemoved() } -func newHeartbeatOperation(ctrl channel.Channel, frequency time.Duration, tokenProvider TokenProvider) *heartbeatOperation { +func newHeartbeatOperation(env routerEnv.RouterEnv, frequency time.Duration, tokenProvider TokenProvider) *heartbeatOperation { return &heartbeatOperation{ - ctrl: ctrl, + env: env, tokenProvider: tokenProvider, BaseOperation: runner.NewBaseOperation("Heartbeat Operation", frequency)} } @@ -91,11 +92,13 @@ func (operation *heartbeatOperation) beat(tokens []string) { } } - for _, msg := range msgs { - if err := operation.ctrl.Send(msg); err != nil { - pfxlog.Logger().WithError(err).Error("could not send heartbeats on control channel") - } + operation.env.GetNetworkControllers().ForEach(func(ctrlId string, ch channel.Channel) { + for _, msg := range msgs { + if err := ch.Send(msg); err != nil { + pfxlog.Logger().WithError(err).Error("could not send heartbeats on control channel") + } - } + } + }) } diff --git a/router/fabric/manager.go b/router/fabric/manager.go index 12b5380e9..7745207b0 100644 --- a/router/fabric/manager.go +++ b/router/fabric/manager.go @@ -24,6 +24,7 @@ import ( "github.com/openziti/channel/v2" "github.com/openziti/edge/pb/edge_ctrl_pb" "github.com/openziti/edge/runner" + "github.com/openziti/fabric/router/env" cmap "github.com/orcaman/concurrent-map/v2" "github.com/sirupsen/logrus" "google.golang.org/protobuf/proto" @@ -62,7 +63,7 @@ type StateManager interface { RemoveConnectedApiSessionWithChannel(token string, underlay channel.Channel) AddApiSessionRemovedListener(token string, callBack func(token string)) RemoveListener - StartHeartbeat(channel channel.Channel, seconds int, closeNotify <-chan struct{}) + StartHeartbeat(env env.RouterEnv, seconds int, closeNotify <-chan struct{}) ValidateSessions(ch channel.Channel, chunkSize uint32, minInterval, maxInterval time.Duration) DumpApiSessions(c *bufio.ReadWriter) error @@ -288,8 +289,8 @@ func (sm *StateManagerImpl) getApiSessionRemovedEventName(token string) events.E return events.EventName(eventName) } -func (sm *StateManagerImpl) StartHeartbeat(ctrl channel.Channel, intervalSeconds int, closeNotify <-chan struct{}) { - sm.heartbeatOperation = newHeartbeatOperation(ctrl, time.Duration(intervalSeconds)*time.Second, sm) +func (sm *StateManagerImpl) StartHeartbeat(env env.RouterEnv, intervalSeconds int, closeNotify <-chan struct{}) { + sm.heartbeatOperation = newHeartbeatOperation(env, time.Duration(intervalSeconds)*time.Second, sm) var err error sm.heartbeatRunner, err = runner.NewRunner(1*time.Second, 24*time.Hour, func(e error, operation runner.Operation) { diff --git a/router/internal/edgerouter/config.go b/router/internal/edgerouter/config.go index 03a5ede29..bbf2339af 100644 --- a/router/internal/edgerouter/config.go +++ b/router/internal/edgerouter/config.go @@ -135,7 +135,7 @@ func (config *Config) LoadConfigFromMap(configMap map[interface{}]interface{}) e config.HeartbeatIntervalSeconds = val.(int) } - if config.HeartbeatIntervalSeconds > DefaultHeartbeatIntervalSeconds || config.HeartbeatIntervalSeconds <= MinHeartbeatIntervalSeconds { + if config.HeartbeatIntervalSeconds > DefaultHeartbeatIntervalSeconds || config.HeartbeatIntervalSeconds < MinHeartbeatIntervalSeconds { pfxlog.Logger().Warnf("Invalid heartbeat interval [%v] (min: %v, max: %v), setting to default [%v]", config.HeartbeatIntervalSeconds, MaxHeartbeatIntervalSeconds, MinHeartbeatIntervalSeconds, DefaultHeartbeatIntervalSeconds) config.HeartbeatIntervalSeconds = DefaultHeartbeatIntervalSeconds } diff --git a/router/xgress_edge/factory.go b/router/xgress_edge/factory.go index f65ba5751..2a3bece78 100644 --- a/router/xgress_edge/factory.go +++ b/router/xgress_edge/factory.go @@ -87,9 +87,7 @@ func (factory *Factory) GetTraceDecoders() []channel.TraceMessageDecoder { func (factory *Factory) Run(env env.RouterEnv) error { factory.ctrls = env.GetNetworkControllers() - env.GetNetworkControllers().ForEach(func(_ string, ch channel.Channel) { - factory.stateManager.StartHeartbeat(ch, factory.edgeRouterConfig.HeartbeatIntervalSeconds, env.GetCloseNotify()) - }) + factory.stateManager.StartHeartbeat(env, factory.edgeRouterConfig.HeartbeatIntervalSeconds, env.GetCloseNotify()) factory.certChecker = NewCertExpirationChecker(factory.routerConfig.Id, factory.edgeRouterConfig, env.GetNetworkControllers(), env.GetCloseNotify()) diff --git a/tests/ats-edge.router.yml b/tests/ats-edge.router.yml index 397218949..4581a8be3 100644 --- a/tests/ats-edge.router.yml +++ b/tests/ats-edge.router.yml @@ -36,6 +36,7 @@ ctrl: # section, commenting out, or altering the name of the section will cause the router to no longer operate as an Edge # Router. edge: + heartbeatIntervalSeconds: 10 # (required) Information used to generate the initial registration CSR. For documentation on these fields please # refer to the openssl documentation. These values MUST be supplied and have no defaults. csr: diff --git a/tests/identity_api_session_activity_test.go b/tests/identity_api_session_activity_test.go new file mode 100644 index 000000000..ed7ede59f --- /dev/null +++ b/tests/identity_api_session_activity_test.go @@ -0,0 +1,82 @@ +package tests + +import ( + rest_identity "github.com/openziti/edge-api/rest_management_api_client/identity" + "github.com/openziti/edge-api/rest_model" + "github.com/openziti/edge/controller" + edge_apis "github.com/openziti/sdk-golang/edge-apis" + "github.com/openziti/sdk-golang/ziti" + "net/url" + "testing" + "time" +) + +func Test_Identity_HasErConnection(t *testing.T) { + ctx := NewTestContext(t) + defer ctx.Teardown() + ctx.StartServer() + ctx.RequireAdminManagementApiLogin() + + ctx.CreateEnrollAndStartEdgeRouter() + + service := ctx.AdminManagementSession.RequireNewServiceAccessibleToAll("smartrouting") + + sdkIdentity, context := ctx.AdminManagementSession.RequireCreateSdkContext() + defer context.Close() + + listener, err := context.Listen(service.Name) + ctx.Req.NoError(err) + + isRunning := true + + defer func() { + isRunning = false + ctx.Req.NoError(listener.Close()) + }() + managementStr := "https://" + ctx.ApiHost + controller.ManagementRestApiBaseUrlV1 + managementUrl, err := url.Parse(managementStr) + ctx.Req.NoError(err) + + creds := edge_apis.NewUpdbCredentials(ctx.AdminAuthenticator.Username, ctx.AdminAuthenticator.Password) + + caPool, err := ziti.GetControllerWellKnownCaPool("https://" + ctx.ApiHost) + ctx.Req.NoError(err) + + managementClient := edge_apis.NewManagementApiClient(managementUrl, caPool) + + curSession, err := managementClient.Authenticate(creds) + ctx.Req.NoError(err) + ctx.Req.NotNil(curSession) + + result := make(chan *rest_model.IdentityDetail) + detailIdentityParams := rest_identity.NewDetailIdentityParams() + detailIdentityParams.ID = sdkIdentity.Id + + //HasEdgeRouterConnection can take up to the minimum heartbeat interval (default 60s, configured in tests for 10s) + //Check every 1s for an update + go func() { + for isRunning { + resp, err := managementClient.API.Identity.DetailIdentity(detailIdentityParams, nil) + + ctx.Req.NoError(err) + ctx.NotNil(resp) + + if *resp.Payload.Data.HasAPISession && *resp.Payload.Data.HasEdgeRouterConnection { + result <- resp.Payload.Data + return + } + + time.Sleep(1 * time.Second) + } + }() + + //Should receive a valid result no later than ~10s later based on the heartbeat interval. + select { + case id := <-result: + ctx.Req.True(*id.HasAPISession) + ctx.Req.True(*id.HasEdgeRouterConnection) + case _ = <-time.After(15 * time.Second): + ctx.Fail("timed out") + } + +} From 652d98e2f0bbf7d998a168550a50b366bdbe16c8 Mon Sep 17 00:00:00 2001 From: Andrew Martinez Date: Mon, 1 May 2023 13:48:10 -0400 Subject: [PATCH 2/2] remove empty assignment --- tests/identity_api_session_activity_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/identity_api_session_activity_test.go b/tests/identity_api_session_activity_test.go index ed7ede59f..84c03687f 100644 --- a/tests/identity_api_session_activity_test.go +++ b/tests/identity_api_session_activity_test.go @@ -75,7 +75,7 @@ func Test_Identity_HasErConnection(t *testing.T) { case id := <-result: ctx.Req.True(*id.HasAPISession) ctx.Req.True(*id.HasEdgeRouterConnection) - case _ = <-time.After(15 * time.Second): + case <-time.After(15 * time.Second): ctx.Fail("timed out") }