Merge pull request #1452 from openziti/fix.has.er.conn.ziti.1093

fixes openziti/ziti#1093

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)

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

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).

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,

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{

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)
 			}
 		}
 	}()

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

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")
+			}
 
-	}
+		}
+	})
 
 }

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) {

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
 	}

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())
 

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:

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")
+	}
+
+}