diff --git a/controller/internal/policy/api_session_enforcer.go b/controller/internal/policy/api_session_enforcer.go index 0fb2f06a3..f16613ccb 100644 --- a/controller/internal/policy/api_session_enforcer.go +++ b/controller/internal/policy/api_session_enforcer.go @@ -21,6 +21,7 @@ import ( "github.com/michaelquigley/pfxlog" "github.com/openziti/edge/controller/env" "github.com/openziti/edge/runner" + "github.com/sirupsen/logrus" "time" ) @@ -75,8 +76,16 @@ func (s *ApiSessionEnforcer) Run() error { break } - for _, id := range ids { - _ = s.appEnv.GetHandlers().ApiSession.Delete(id) + logrus.Debugf("found %v expired api-sessions to remove", len(ids)) + + if err = s.appEnv.GetHandlers().ApiSession.DeleteBatch(ids); err != nil { + logrus.WithError(err).Error("failure while batch deleting expired api sessions") + + for _, id := range ids { + if err = s.appEnv.GetHandlers().ApiSession.Delete(id); err != nil { + logrus.WithError(err).Errorf("failure while deleting expired api session: %v", id) + } + } } } diff --git a/controller/model/api_session_handlers.go b/controller/model/api_session_handlers.go index 683059299..ea78753da 100644 --- a/controller/model/api_session_handlers.go +++ b/controller/model/api_session_handlers.go @@ -101,6 +101,10 @@ func (handler *ApiSessionHandler) Delete(id string) error { return handler.deleteEntity(id) } +func (handler *ApiSessionHandler) DeleteBatch(id []string) error { + return handler.deleteEntityBatch(id) +} + func (handler *ApiSessionHandler) MarkActivityById(apiSessionId string) { handler.HeartbeatCollector.Mark(apiSessionId) } diff --git a/controller/model/base_handler.go b/controller/model/base_handler.go index 71a174a40..c00c32eef 100644 --- a/controller/model/base_handler.go +++ b/controller/model/base_handler.go @@ -347,6 +347,17 @@ func (handler *baseHandler) deleteEntity(id string) error { }) } +func (handler *baseHandler) deleteEntityBatch(ids []string) error { + return handler.GetDb().Update(func(tx *bbolt.Tx) error { + for _, id := range ids { + if err := handler.GetStore().DeleteById(boltz.NewMutateContext(tx), id); err != nil { + return err + } + } + return nil + }) +} + func (handler *baseHandler) list(queryString string, resultHandler models.ListResultHandler) error { return handler.GetDb().View(func(tx *bbolt.Tx) error { return handler.ListWithTx(tx, queryString, resultHandler) diff --git a/controller/model/service_handlers.go b/controller/model/service_handlers.go index 75bc6a5cd..8c4f45acf 100644 --- a/controller/model/service_handlers.go +++ b/controller/model/service_handlers.go @@ -258,32 +258,6 @@ func (handler *EdgeServiceHandler) mergeConfigs(tx *bbolt.Tx, configTypes map[st } } -func (handler *EdgeServiceHandler) StreamByServicePolicyId(servicePolicyId string, collect func(*ServiceDetail, error) error) error { - query := fmt.Sprintf(`anyOf(servicePolicies) = "%v"`, servicePolicyId) - - return handler.Stream(query, collect) -} - -func (handler *EdgeServiceHandler) Stream(query string, collect func(*ServiceDetail, error) error) error { - filter, err := ast.Parse(handler.Store, query) - - if err != nil { - return fmt.Errorf("could not parse query for streaming edge services: %v", err) - } - - return handler.env.GetDbProvider().GetDb().View(func(tx *bbolt.Tx) error { - for cursor := handler.Store.IterateIds(tx, filter); cursor.IsValid(); cursor.Next() { - current := cursor.Current() - - service, err := handler.readInTx(tx, string(current)) - if err := collect(service, err); err != nil { - return err - } - } - return nil - }) -} - type PolicyPostureChecks struct { PostureChecks []*PostureCheck PolicyType persistence.PolicyType diff --git a/controller/model/session_handlers.go b/controller/model/session_handlers.go index 2542d52d7..8ba969009 100644 --- a/controller/model/session_handlers.go +++ b/controller/model/session_handlers.go @@ -288,19 +288,6 @@ func (handler *SessionHandler) querySessions(query ast.Query) (*SessionListResul return result, nil } -func (handler *SessionHandler) StreamAll(collect func(*Session, error) error) error { - return handler.env.GetDbProvider().GetDb().View(func(tx *bbolt.Tx) error { - for cursor := handler.Store.IterateIds(tx, ast.BoolNodeTrue); cursor.IsValid(); cursor.Next() { - current := cursor.Current() - apiSession, err := handler.readInTx(tx, string(current)) - if err := collect(apiSession, err); err != nil { - return err - } - } - return collect(nil, nil) - }) -} - func (handler *SessionHandler) ListSessionsForEdgeRouter(edgeRouterId string) (*SessionListResult, error) { result := &SessionListResult{handler: handler} query := fmt.Sprintf(`anyOf(apiSession.identity.edgeRouterPolicies.routers) = "%v" and `+