Merge pull request #684 from openziti/api-enforcer-delete-batching

Delete expired api sessions in batches. Remove some unused methods

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

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

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)

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

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 `+