diff --git a/controller/env/appenv.go b/controller/env/appenv.go index d78cb69e6..a5c7d5a02 100644 --- a/controller/env/appenv.go +++ b/controller/env/appenv.go @@ -207,6 +207,10 @@ type AppMiddleware func(*AppEnv, http.Handler) http.Handler type authorizer struct { } +const ( + EventualEventsGauge = "eventual.events" +) + func (a authorizer) Authorize(request *http.Request, principal interface{}) error { //principal is an API Session _, ok := principal.(*model.ApiSession) @@ -378,6 +382,33 @@ func (ae *AppEnv) InitPersistence() error { return err } + ae.BoltStores.EventualEventer.AddListener(persistence.EventualEventAddedName, func(i ...interface{}) { + if len(i) == 0 { + pfxlog.Logger().Errorf("could not update metrics for %s gauge on add, event argument length was 0", EventualEventsGauge) + return + } + + if event, ok := i[0].(*persistence.EventualEventAdded); ok { + gauge := ae.GetHostController().GetNetwork().GetMetricsRegistry().Gauge(EventualEventsGauge) + gauge.Update(event.Total) + } else { + pfxlog.Logger().Errorf("could not update metrics for %s gauge on add, event argument was %T expected *EventualEventAdded", EventualEventsGauge, i[0]) + } + }) + ae.BoltStores.EventualEventer.AddListener(persistence.EventualEventRemovedName, func(i ...interface{}) { + if len(i) == 0 { + pfxlog.Logger().Errorf("could not update metrics for %s gauge on remove, event argument length was 0", EventualEventsGauge) + return + } + + if event, ok := i[0].(*persistence.EventualEventRemoved); ok { + gauge := ae.GetHostController().GetNetwork().GetMetricsRegistry().Gauge(EventualEventsGauge) + gauge.Update(event.Total) + } else { + pfxlog.Logger().Errorf("could not update metrics for %s gauge on remove, event argument was %T expected *EventualEventRemoved", EventualEventsGauge, i[0]) + } + }) + ae.Handlers = model.InitHandlers(ae) events.Init(ae.GetDbProvider(), ae.BoltStores, ae.GetHostController().GetCloseNotifyChannel()) diff --git a/controller/internal/policy/session_enforcer_test.go b/controller/internal/policy/session_enforcer_test.go index 611e16756..38f8d5fee 100644 --- a/controller/internal/policy/session_enforcer_test.go +++ b/controller/internal/policy/session_enforcer_test.go @@ -31,9 +31,8 @@ func Test_SessionEnforcer(t *testing.T) { ctx := &enforcerTestContext{ TestContext: model.NewTestContext(t), } - - defer ctx.Cleanup() ctx.Init() + defer ctx.Cleanup() ctx.testSessionsCleanup() } @@ -75,6 +74,16 @@ func (ctx *enforcerTestContext) testSessionsCleanup() { } ctx.NoError(enforcer.Run()) + + done, err := ctx.GetStores().EventualEventer.Trigger() + ctx.NoError(err) + + select { + case <-done: + case <-time.After(5 * time.Second): + ctx.Fail("did not receive done notification from eventual eventer") + } + ctx.ValidateDeleted(apiSession.Id) ctx.ValidateDeleted(session.Id) ctx.ValidateDeleted(session2.Id) diff --git a/controller/model/testing.go b/controller/model/testing.go index 3e68c6f8a..bbe596693 100644 --- a/controller/model/testing.go +++ b/controller/model/testing.go @@ -120,10 +120,6 @@ func (ctx *TestContext) GetFingerprintGenerator() cert.FingerprintGenerator { } func NewTestContext(t *testing.T) *TestContext { - return NewTestContextWithDb(t, "") -} - -func NewTestContextWithDb(t *testing.T, dbPath string) *TestContext { context := &TestContext{ TestContext: persistence.NewTestContext(t), metricsRegistry: metrics.NewRegistry("test", nil), @@ -131,9 +127,11 @@ func NewTestContextWithDb(t *testing.T, dbPath string) *TestContext { closeNotify: make(chan struct{}), }, } - context.InitWithDbFile(dbPath) return context } +func (ctx *TestContext) Init() { + ctx.InitWithDbFile("") +} func (ctx *TestContext) InitWithDbFile(dbPath string) { ctx.TestContext.InitWithDbFile(dbPath) diff --git a/controller/persistence/api_session_store.go b/controller/persistence/api_session_store.go index bdd3b7853..b3ff40929 100644 --- a/controller/persistence/api_session_store.go +++ b/controller/persistence/api_session_store.go @@ -17,7 +17,9 @@ package persistence import ( + "fmt" "github.com/kataras/go-events" + "github.com/michaelquigley/pfxlog" "github.com/openziti/edge/eid" "github.com/openziti/foundation/storage/ast" "github.com/openziti/foundation/storage/boltz" @@ -34,7 +36,8 @@ const ( FieldApiSessionMfaRequired = "mfaRequired" FieldApiSessionLastActivityAt = "lastActivityAt" - EventFullyAuthenticated events.EventName = "FULLY_AUTHENTICATED" + EventFullyAuthenticated events.EventName = "FULLY_AUTHENTICATED" + EventualEventApiSessionDelete = "ApiSessionDelete" ) type ApiSession struct { @@ -98,6 +101,8 @@ func newApiSessionStore(stores *stores) *apiSessionStoreImpl { store := &apiSessionStoreImpl{ baseStore: newBaseStore(stores, EntityTypeApiSessions), } + + stores.EventualEventer.AddEventualListener(EventualEventApiSessionDelete, store.onEventualDelete) store.InitImpl(store) return store } @@ -109,6 +114,40 @@ type apiSessionStoreImpl struct { symbolIdentity boltz.EntitySymbol } +func (store *apiSessionStoreImpl) onEventualDelete(name string, data []byte) { + var ids []string + err := store.stores.DbProvider.GetDb().View(func(tx *bbolt.Tx) error { + query := fmt.Sprintf(`%s = "%s"`, FieldSessionApiSession, string(data)) + var err error + ids, _, err = store.stores.session.QueryIds(tx, query) + return err + }) + + if err != nil { + pfxlog.Logger().WithError(err).WithFields(map[string]interface{}{ + "eventName": name, + "apiSessionId": string(data), + }).Error("error querying for session associated to an api session during onEventualDelete") + } + + for _, id := range ids { + _ = store.stores.DbProvider.GetDb().Update(func(tx *bbolt.Tx) error { + ctx := boltz.NewMutateContext(tx) + err := store.stores.session.DeleteById(ctx, id) + + if err != nil { + pfxlog.Logger().WithError(err).WithFields(map[string]interface{}{ + "eventName": name, + "apiSessionId": string(data), + "sessionId": id, + }).Error("error deleting for session associated to an api session during onEventualDelete") + } + + return nil + }) + } +} + func (store *apiSessionStoreImpl) Create(ctx boltz.MutateContext, entity boltz.Entity) error { err := store.baseStore.Create(ctx, entity) @@ -136,6 +175,20 @@ func (store *apiSessionStoreImpl) Update(ctx boltz.MutateContext, entity boltz.E return err } +func (store *apiSessionStoreImpl) DeleteById(ctx boltz.MutateContext, id string) error { + err := store.baseStore.DeleteById(ctx, id) + + if err == nil { + if bboltEventualEventer, ok := store.baseStore.stores.EventualEventer.(*EventualEventerBbolt); ok { + bboltEventualEventer.AddEventualEventWithCtx(ctx, EventualEventApiSessionDelete, []byte(id)) + } else { + store.baseStore.stores.EventualEventer.AddEventualEvent(EventualEventApiSessionDelete, []byte(id)) + } + } + + return err +} + func (store *apiSessionStoreImpl) NewStoreEntity() boltz.Entity { return &ApiSession{} } diff --git a/controller/persistence/api_session_store_test.go b/controller/persistence/api_session_store_test.go index 07eac3202..f756ab4ea 100644 --- a/controller/persistence/api_session_store_test.go +++ b/controller/persistence/api_session_store_test.go @@ -39,7 +39,8 @@ func Test_ApiSessionStore(t *testing.T) { t.Run("test delete api sessions", ctx.testDeleteApiSessions) } -func (ctx *TestContext) testCreateInvalidApiSessions(_ *testing.T) { +func (ctx *TestContext) testCreateInvalidApiSessions(t *testing.T) { + ctx.BaseTestContext.NextTest(t) defer ctx.CleanupAll() apiSession := NewApiSession(eid.New()) @@ -64,7 +65,8 @@ func (ctx *TestContext) testCreateInvalidApiSessions(_ *testing.T) { ctx.EqualError(err, fmt.Sprintf("an entity of type apiSession already exists with id %v", apiSession.GetId())) } -func (ctx *TestContext) testCreateApiSessions(_ *testing.T) { +func (ctx *TestContext) testCreateApiSessions(t *testing.T) { + ctx.BaseTestContext.NextTest(t) ctx.CleanupAll() identity := ctx.RequireNewIdentity("Jojo", false) @@ -80,9 +82,20 @@ func (ctx *TestContext) testCreateApiSessions(_ *testing.T) { ctx.ValidateBaseline(apiSession2) - ctx.RequireDelete(apiSession) - + err := ctx.Delete(apiSession) + ctx.NoError(err) ctx.RequireDelete(identity) + + done, err := ctx.GetStores().EventualEventer.Trigger() + ctx.NoError(err) + + select { + case <-done: + case <-time.After(5 * time.Second): + ctx.Fail("did not receive done notification from eventual eventer") + + } + ctx.ValidateDeleted(apiSession.Id) ctx.ValidateDeleted(apiSession2.Id) } @@ -130,7 +143,8 @@ func (ctx *TestContext) createApiSessionTestEntities() *apiSessionTestEntities { } } -func (ctx *TestContext) testLoadQueryApiSessions(_ *testing.T) { +func (ctx *TestContext) testLoadQueryApiSessions(t *testing.T) { + ctx.BaseTestContext.NextTest(t) ctx.CleanupAll() entities := ctx.createApiSessionTestEntities() @@ -159,11 +173,11 @@ func (ctx *TestContext) testLoadQueryApiSessions(_ *testing.T) { ctx.NoError(err) } -func (ctx *TestContext) testUpdateApiSessions(_ *testing.T) { +func (ctx *TestContext) testUpdateApiSessions(t *testing.T) { + ctx.BaseTestContext.NextTest(t) ctx.CleanupAll() entities := ctx.createApiSessionTestEntities() earlier := time.Now() - time.Sleep(time.Millisecond * 50) err := ctx.GetDb().Update(func(tx *bbolt.Tx) error { original, err := ctx.stores.ApiSession.LoadOneById(tx, entities.apiSession1.Id) @@ -197,10 +211,31 @@ func (ctx *TestContext) testUpdateApiSessions(_ *testing.T) { ctx.NoError(err) } -func (ctx *TestContext) testDeleteApiSessions(_ *testing.T) { +func (ctx *TestContext) testDeleteApiSessions(t *testing.T) { + ctx.BaseTestContext.NextTest(t) ctx.CleanupAll() entities := ctx.createApiSessionTestEntities() - ctx.RequireDelete(entities.apiSession1) - ctx.RequireDelete(entities.apiSession2) - ctx.RequireDelete(entities.apiSession3) + + err := ctx.Delete(entities.apiSession1) + ctx.NoError(err) + + err = ctx.Delete(entities.apiSession2) + ctx.NoError(err) + + err = ctx.Delete(entities.apiSession3) + ctx.NoError(err) + + done, err := ctx.GetStores().EventualEventer.Trigger() + ctx.NoError(err) + + select { + case <-done: + case <-time.After(5 * time.Second): + ctx.Fail("did not receive done notification from eventual eventer") + + } + + ctx.ValidateDeleted(entities.apiSession1.GetId()) + ctx.ValidateDeleted(entities.apiSession2.GetId()) + ctx.ValidateDeleted(entities.apiSession3.GetId()) } diff --git a/controller/persistence/base_entity.go b/controller/persistence/base_entity.go index 6284c96bb..30fefa380 100644 --- a/controller/persistence/base_entity.go +++ b/controller/persistence/base_entity.go @@ -24,6 +24,7 @@ import ( const ( EntityTypeApiSessions = "apiSessions" EntityTypeApiSessionCertificates = "apiSessionCertificates" + EntityTypeEventualEvents = "eventualEvents" EntityTypeCas = "cas" EntityTypeConfigs = "configs" EntityTypeConfigTypes = "configTypes" diff --git a/controller/persistence/eventual_event_store.go b/controller/persistence/eventual_event_store.go new file mode 100644 index 000000000..057b2885e --- /dev/null +++ b/controller/persistence/eventual_event_store.go @@ -0,0 +1,99 @@ +/* + Copyright NetFoundry, Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + https://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package persistence + +import ( + "github.com/openziti/foundation/storage/ast" + "github.com/openziti/foundation/storage/boltz" + "go.etcd.io/bbolt" +) + +const ( + FieldEventualEventType = "type" + FieldEventualEventData = "data" +) + +type EventualEvent struct { + boltz.BaseExtEntity + Type string + Data []byte +} + +func (entity *EventualEvent) LoadValues(_ boltz.CrudStore, bucket *boltz.TypedBucket) { + entity.LoadBaseValues(bucket) + entity.Type = bucket.GetStringOrError(FieldEventualEventType) + entity.Data = bucket.Get([]byte(FieldEventualEventData)) + +} + +func (entity *EventualEvent) SetValues(ctx *boltz.PersistContext) { + entity.SetBaseValues(ctx) + ctx.SetString(FieldEventualEventType, entity.Type) + ctx.Bucket.SetError(ctx.Bucket.Put([]byte(FieldEventualEventData), entity.Data)) +} + +func (entity *EventualEvent) GetEntityType() string { + return EntityTypeEventualEvents +} + +type EventualEventStore interface { + Store + LoadOneById(tx *bbolt.Tx, id string) (*EventualEvent, error) + LoadOneByQuery(tx *bbolt.Tx, query string) (*EventualEvent, error) +} + +func newEventualEventStore(stores *stores) *eventualEventStoreImpl { + store := &eventualEventStoreImpl{ + baseStore: newBaseStore(stores, EntityTypeEventualEvents), + } + store.InitImpl(store) + return store +} + +type eventualEventStoreImpl struct { + *baseStore + indexName boltz.ReadIndex + symbolEnrollments boltz.EntitySetSymbol +} + +func (store *eventualEventStoreImpl) LoadOneById(tx *bbolt.Tx, id string) (*EventualEvent, error) { + entity := &EventualEvent{} + if err := store.baseLoadOneById(tx, id, entity); err != nil { + return nil, err + } + return entity, nil +} + +func (store *eventualEventStoreImpl) LoadOneByQuery(tx *bbolt.Tx, query string) (*EventualEvent, error) { + entity := &EventualEvent{} + if found, err := store.BaseLoadOneByQuery(tx, query, entity); !found || err != nil { + return nil, err + } + return entity, nil +} + +func (store *eventualEventStoreImpl) NewStoreEntity() boltz.Entity { + return &EventualEvent{} +} + +func (store *eventualEventStoreImpl) initializeLocal() { + store.AddExtEntitySymbols() + store.AddSymbol(FieldEventualEventData, ast.NodeTypeOther) +} + +func (store *eventualEventStoreImpl) initializeLinked() { +} diff --git a/controller/persistence/eventual_eventer.go b/controller/persistence/eventual_eventer.go new file mode 100644 index 000000000..7bccc3fe5 --- /dev/null +++ b/controller/persistence/eventual_eventer.go @@ -0,0 +1,696 @@ +/* + Copyright NetFoundry, Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + https://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package persistence + +import ( + "fmt" + "github.com/kataras/go-events" + "github.com/lucsky/cuid" + "github.com/michaelquigley/pfxlog" + "github.com/openziti/foundation/storage/boltz" + "github.com/openziti/foundation/util/concurrenz" + cmap "github.com/orcaman/concurrent-map" + "github.com/pkg/errors" + "go.etcd.io/bbolt" + "sync" + "sync/atomic" + "time" +) + +// An EventualEventer provides a method for storing events in a persistent manner +// that will be processed at a later date. Processing may include time intensive processing such +// as bulk deletion of other entities. Event persistence strategy, processing order, and processing +// synchronization are up to the implementation to decide. +// +// EventualEventers are also required to emit a series of events via the events.EventEmitter +// interface. See EventualEventAdded and subsequent events for more details. +type EventualEventer interface { + // EventEmmiter is used to provide processing event status on processing state, which is useful + // for instrumenting an EventualEventer for metric purposes (process runtime, process batch runtime, + // event counts, etc.) + events.EventEmmiter + + // AddEventualEvent adds an eventual event with a specific name and byte array data payload. Interpretation + // of the event's data payload is upto the event emitter and consumer. + AddEventualEvent(eventType string, data []byte) + + // AddEventualListener adds a function as call back when an eventual event is processed. + AddEventualListener(eventType string, handler EventListenerFunc) + + // Start should be called at the start of the lifetime of the EventualEventer. + // A closeNotify channel must be supplied for application shutdown eventing. + // + // If an EventualEventer has already been started, it will return an error. + // Errors may be returned for other reasons causing Start to fail. + Start(closeNotify <-chan struct{}) error + + // Stop may be called to manually end of the lifetime of the EventualEventer outside the + // closeNotify signaling provided in the Start call. If not started, an error will be returned. + // Errors may be returned for other reasons causing Stop to fail. + Stop() error + + // Trigger forces an EventualEventer to check for work to be processed. Beyond this method, + // it is the implementation's responsibility to provide other mechanisms or logic to determine + // when work is performed (timers, events, etc.) which may be setup/torn down during Start/Stop. + // + // If the EventualEventer is not currently running or can't process work and error will + // be returned. If it is running a channel will be returned which will be closed after + // the current or next iteration of the event processor has completed. + Trigger() (<-chan struct{}, error) +} + +// EventListenerFunc is a function handler that will be triggered asynchronously some point in the future +type EventListenerFunc func(name string, data []byte) + +type EventualEventAdded struct { + // Id is a unique id for the event created + Id string + + // Total is the total number of eventual events awaiting processing + Total int64 +} + +type EventualEventRemoved struct { + // Id is a unique id for the event deleted + Id string + + // Total is the total number of eventual events awaiting processing + Total int64 +} + +type EventualEventProcessingStart struct { + // Id is a unique id for processing run + Id string + + // StartTime is the time the processing began + StartTime time.Time +} + +type EventualEventProcessingBatchStart struct { + // Id is a unique id for the batch + Id string + + // Id is the unique processing run this batch is a member of + ProcessId string + + // Count is the number of events in the current batch + Count int + + // BatchSize is the batch size for the current batch (the maximum value of Count) + BatchSize int + + // StartTime the time when the batch started processing + StartTime time.Time +} + +type EventualEventProcessingListenerStart struct { + // Id is a unique id for the triggering of a listener + Id string + + // BatchId is the unique id of the batch being processed + BatchId string + + // ProcessId is the unique id of the currently executing process + ProcessId string + + // ListenerFunc is the listener that was executed + ListenerFunc EventListenerFunc + + // BatchEventIndex is the zero based offset of the currently executing event + BatchEventIndex int64 + + // TotalEventIndex is the total index across all batches of the currently executing event + TotalEventIndex int64 + + // EventType is the typeof the event that is triggering the listener + EventType string + + // StartTime is the time when the listener was started + StartTime time.Time +} + +type EventualEventProcessingListenerDone struct { + // Id is a unique id for the triggering of a listener + Id string + + // BatchId is the unique id of the batch being processed + BatchId string + + // ProcessId is the unique id of the currently executing process + ProcessId string + + // ListenerFunc is the listener that was executed + ListenerFunc EventListenerFunc + + // BatchEventIndex is the zero based offset of the currently executing event + BatchEventIndex int64 + + // TotalEventIndex is the total index across all batches of the currently executing event + TotalEventIndex int64 + + // Error is nil if no error occurred during execution, otherwise an error value + Error error + + // EventType is the typeof the event that triggered the listener + EventType string + + // StartTime is the time when the listener started execution + StartTime time.Time + + // EndTime is the time when the listener ended execution + EndTime time.Time +} + +type EventualEventProcessingBatchDone struct { + // Id is a unique id for the batch + Id string + + // Id is the unique processing run this batch is a member of + ProcessId string + + // Count is the number of events in the current batch + Count int + + // BatchSize is the batch size for the current batch (the maximum value of Count) + BatchSize int + + // StartTime the time the batch was started + StartTime time.Time + + // EndTime the time the batch ended + EndTime time.Time +} + +type EventualEventProcessingDone struct { + // Id is a unique id for processing run + Id string + + // TotalBatches is the total number of batches executed during processing + TotalBatches int64 + + // TotalEvent is the total number of events processed + TotalEvents int64 + + // TotalListenersExecuted is the total number of listeners executed during processing + TotalListenersExecuted int64 + + // StartTime is the time when the processing began + StartTime time.Time + + // EndTime is the time when the processing ended + EndTime time.Time +} + +const ( + // EventualEventAddedName is emitted when a new event is added via AddEventualEvent(). + // + // Event arguments: + // 0 - an EventualEventAdded struct + EventualEventAddedName = events.EventName("EventualEventAdded") + + // EventualEventRemovedName is emitted when a previously added eventual event is processed + // + // Event arguments: + // 0 - an EventualEventRemoved struct + EventualEventRemovedName = events.EventName("EventualEventRemoved") + + // EventualEventProcessingStartName is emitted as the first action during processing + // Event arguments: + // 0 - an EventualEventProcessingStart struct + EventualEventProcessingStartName = events.EventName("EventualEventProcessingStart") + + // EventualEventProcessingBatchStartName is emitted as the first set of events are processed + // after EventualEventProcessingStartName. It is possible for 0+ batches to be processed. Each + // patch should contain 1+ events. + // + // Event arguments: + // 0 - an EventualEventProcessingBatchStart struct + EventualEventProcessingBatchStartName = events.EventName("EventualEventProcessingBatchStart") + + // EventualEventProcessingListenerStartName is emitted for each function listener invoked + // on each event. + // + // Event arguments: + // 0 - an EventualEventProcessingListenerStart struct + EventualEventProcessingListenerStartName = events.EventName("EventualEventProcessingListenerStart") + + // EventualEventProcessingListenerDoneName is emitted for each function listener after invocation + // + // Event arguments: + // 0 - an EventualEventProcessingListenerDone struct + EventualEventProcessingListenerDoneName = events.EventName("EventualEventProcessingListenerDone") + + // EventualEventProcessingBatchDoneName is emitted after the last event processed in a batch. + // + // Event arguments: + // 0 - an EventualEventProcessingBatchDone struct + EventualEventProcessingBatchDoneName = events.EventName("EventualEventProcessingBatchDone") + + // EventualEventProcessingDoneName is emitted as the last action during processing after + // all events and batches. + // + // Event arguments: + // 0 - an EventualEventProcessingDone struct + EventualEventProcessingDoneName = events.EventName("EventualEventProcessingDone") +) + +// EventualEventerBbolt implements EventualEventer with a bbolt back storage mechanism. +// Work is performed on a configurable basis via the Interval property in FIFO order. +// +// Events are stored in the following format: +// id - CUID - a monotonic reference id +// name - string - an event name, used for log output +// data - []byte - a string array of arguments +type EventualEventerBbolt struct { + events.EventEmmiter + handlerMap cmap.ConcurrentMap //eventName -> handler func(name, []byte) + Interval time.Duration + closeNotify <-chan struct{} + stopNotify chan struct{} + trigger chan struct{} + outstandingEvents *int64 + + waiters sync.Map //id -> chan struct{} + + running concurrenz.AtomicBoolean + batchSize int + dbProvider DbProvider + store EventualEventStore +} + +var _ EventualEventer = &EventualEventerBbolt{} + +// NewEventualEventerBbolt creates a new bbolt backed asynchronous eventer that will check for new events at the given interval +// or when triggered. On each interval/trigger, the number of events processed is determined by batchSize. +func NewEventualEventerBbolt(dbProvider DbProvider, store EventualEventStore, interval time.Duration, batchSize int) *EventualEventerBbolt { + outstanding := int64(0) + return &EventualEventerBbolt{ + EventEmmiter: events.New(), + Interval: interval, + dbProvider: dbProvider, + store: store, + batchSize: batchSize, + trigger: make(chan struct{}, 1), + handlerMap: cmap.New(), + outstandingEvents: &outstanding, + } +} + +func (a *EventualEventerBbolt) AddEventualEventWithCtx(ctx boltz.MutateContext, eventType string, data []byte) { + newId := cuid.New() + total := atomic.AddInt64(a.outstandingEvents, 1) + a.Emit(EventualEventAddedName, &EventualEventAdded{ + Id: newId, + Total: total, + }) + + event := &EventualEvent{ + BaseExtEntity: boltz.BaseExtEntity{ + Id: newId, + CreatedAt: time.Now(), + UpdatedAt: time.Now(), + }, + Type: eventType, + Data: data, + } + + var err error + if ctx == nil { + err = a.dbProvider.GetDb().Update(func(tx *bbolt.Tx) error { + ctx := boltz.NewMutateContext(tx) + return a.store.Create(ctx, event) + }) + } else { + err = a.store.Create(ctx, event) + } + + if err != nil { + total := atomic.AddInt64(a.outstandingEvents, -1) + a.Emit(EventualEventRemovedName, &EventualEventRemoved{ + Id: newId, + Total: total, + }) + pfxlog.Logger().WithError(err).Error("error adding event for EventualEventerBbolt") + return + } +} + +func (a *EventualEventerBbolt) AddEventualEvent(eventType string, data []byte) { + a.AddEventualEventWithCtx(nil, eventType, data) +} + +func (a *EventualEventerBbolt) AddEventualListener(eventType string, listener EventListenerFunc) { + a.handlerMap.Upsert(eventType, nil, func(exist bool, valueInMap interface{}, newValue interface{}) interface{} { + var handlers []EventListenerFunc + if exist { + handlers = valueInMap.([]EventListenerFunc) + } + + handlers = append(handlers, listener) + return handlers + }) +} + +func (a *EventualEventerBbolt) Start(closeNotify <-chan struct{}) error { + if !a.running.CompareAndSwap(false, true) { + return errors.New("already started") + } + a.stopNotify = make(chan struct{}, 0) + a.closeNotify = closeNotify + go a.run() + + return nil +} + +// run should be executed once on start and is the main +// processing loop. Exiting is based on close/stop notify +// channels provided during creation. Processing is triggered +// when Trigger() is called or on the configured interval. +func (a *EventualEventerBbolt) run() { + for { + select { + case <-a.closeNotify: + a.running.Set(false) + return + case <-a.stopNotify: + a.running.Set(false) + return + case <-a.trigger: + a.process() + case <-time.After(a.Interval): + a.trigger <- struct{}{} + } + } +} + +func (a *EventualEventerBbolt) Stop() error { + if a.running.CompareAndSwap(true, false) { + close(a.stopNotify) + return nil + } + + return nil +} + +func (a *EventualEventerBbolt) Trigger() (<-chan struct{}, error) { + if a.running.Get() { + doneNotify := make(chan struct{}) + a.waiters.Store(cuid.New(), doneNotify) + + select { + case a.trigger <- struct{}{}: + default: + //channel full, already queued for processing + } + + return doneNotify, nil + } + + return nil, errors.New("triggering is impossible, eventual eventer not started") +} + +// deleteEventualEvent removes an eventual event by id from the bbolt backend store. +func (a *EventualEventerBbolt) deleteEventualEvent(id string) error { + err := a.dbProvider.GetDb().Update(func(tx *bbolt.Tx) error { + ctx := boltz.NewMutateContext(tx) + return a.store.DeleteById(ctx, id) + }) + + if err == nil { + total := atomic.AddInt64(a.outstandingEvents, -1) + a.Emit(EventualEventRemovedName, &EventualEventRemoved{ + Id: id, + Total: total, + }) + } + + return err +} + +// notifyWaiters closes all signaling channels returned in calls +// to Trigger(). +func (a *EventualEventerBbolt) notifyWaiters() { + var waiterIds []string + + a.waiters.Range(func(key, value interface{}) bool { + id := key.(string) + waiterChan := value.(chan struct{}) + if waiterChan != nil { + close(waiterChan) + } + + waiterIds = append(waiterIds, id) + return true + }) + + for _, id := range waiterIds { + a.waiters.Delete(id) + } +} + +// getEventualEvents returns eventual events up from the persistent bbolt store. The number +// returned is determined by batchSize and the order of the events are based on key sorting +func (a *EventualEventerBbolt) getEventualEvents() ([]string, []*EventualEvent, error) { + var ids []string + var eventualEvents []*EventualEvent + err := a.dbProvider.GetDb().View(func(tx *bbolt.Tx) error { + var err error + ids, _, err = a.store.QueryIds(tx, fmt.Sprintf("limit %d", a.batchSize)) + + if err != nil { + return err + } + + for _, id := range ids { + event, err := a.store.LoadOneById(tx, id) + + if err != nil { + pfxlog.Logger().WithField("id", id).WithError(err).Errorf("error could not load event id %s", id) + } else { + eventualEvents = append(eventualEvents, event) + } + } + return nil + }) + + return ids, eventualEvents, err +} + +type runInfo struct { + processId string + totalBatches int64 + totalEvents int64 + totalListenersExecuted int64 + totalEventIndex int64 + batchEventIndex int64 + batchId string + numIds int +} + +// resolveEventualEvents will wait if there are newly added eventual events +// that have a transaction that has not cleared the event store yet. This covers +// situations where tests are expecting the next run to resolve their event. resolveEvents +// intentionally adds 2500 microseconds (2.5ms) when new events have been added, but +// none are returned when queried +func (a *EventualEventerBbolt) resolveEventualEvents() (int, []string, []*EventualEvent) { + var numIds int + var ids []string + var eventualEvents []*EventualEvent + + // if we know we have outstanding events query for them up to 5 times then give up + numEvents := atomic.LoadInt64(a.outstandingEvents) + + for attempt := 0; attempt < 5 && numEvents > 0; attempt++ { + var err error + ids, eventualEvents, err = a.getEventualEvents() + numIds = len(ids) + + if err != nil { + pfxlog.Logger().WithError(err).Error("error during EventualEventerBbolt processing, could not query events") + } + + if numIds > 0 { + break + } + + time.Sleep(500 * time.Microsecond) + numEvents = atomic.LoadInt64(a.outstandingEvents) + } + + return numIds, ids, eventualEvents +} + +// process is the main execution look for dealing with batches of eventual events, triggering listeners for +// the eventual events and emitting normal processing events +func (a *EventualEventerBbolt) process() { + + info := &runInfo{ + processId: cuid.New(), + } + + startTime := time.Now() + + a.Emit(EventualEventProcessingStartName, &EventualEventProcessingStart{ + Id: info.processId, + StartTime: startTime, + }) + + defer func() { + a.notifyWaiters() + + a.Emit(EventualEventProcessingDoneName, &EventualEventProcessingDone{ + Id: info.processId, + TotalBatches: info.totalBatches, + TotalEvents: info.totalEvents, + TotalListenersExecuted: info.totalListenersExecuted, + StartTime: startTime, + EndTime: time.Now(), + }) + }() + + for { + info.batchId = cuid.New() + info.numIds = 0 + + var ids []string + var eventualEvents []*EventualEvent + + info.numIds, ids, eventualEvents = a.resolveEventualEvents() + + if info.numIds == 0 { + return + } + + if info.numIds != len(eventualEvents) { + pfxlog.Logger(). + WithFields(map[string]interface{}{ + "queriedCount": info.numIds, + "eventCount": len(eventualEvents), + }). + Warnf("%d event records were queried and only %d events were loaded", info.numIds, len(eventualEvents)) + } + + a.processBatch(info, eventualEvents) + a.deleteEvents(ids) + } +} + +func (a *EventualEventerBbolt) processBatch(info *runInfo, eventualEvents []*EventualEvent) { + info.totalBatches++ + startTime := time.Now() + a.Emit(EventualEventProcessingBatchStartName, &EventualEventProcessingBatchStart{ + Id: info.batchId, + ProcessId: info.processId, + Count: info.numIds, + BatchSize: a.batchSize, + StartTime: startTime, + }) + + info.batchEventIndex = 0 + for _, eventualEvent := range eventualEvents { + info.batchEventIndex++ + info.totalEventIndex++ + + if handlerVals, found := a.handlerMap.Get(eventualEvent.Type); found { + handlers := handlerVals.([]EventListenerFunc) + for _, handler := range handlers { + a.executeHandler(info, eventualEvent, handler) + } + } else { + pfxlog.Logger(). + WithFields(map[string]interface{}{ + "id": eventualEvent.Id, + "type": eventualEvent.Type, + }). + Debugf("event id %s with type %s has no handlers", eventualEvent.Id, eventualEvent.Type) + } + } + + a.Emit(EventualEventProcessingBatchDoneName, &EventualEventProcessingBatchDone{ + Id: info.batchId, + ProcessId: info.processId, + Count: info.numIds, + BatchSize: a.batchSize, + StartTime: startTime, + EndTime: time.Now(), + }) +} + +func (a *EventualEventerBbolt) deleteEvents(ids []string) { + for _, id := range ids { + if err := a.deleteEventualEvent(id); err != nil { + pfxlog.Logger().WithError(err). + WithField("id", id). + Errorf("could not delete event id %s", id) + } + } +} + +func (a *EventualEventerBbolt) executeHandler(info *runInfo, eventualEvent *EventualEvent, handler EventListenerFunc) { + listenerExecId := cuid.New() + info.totalListenersExecuted++ + + var startTime time.Time + var endTime time.Time + + defer func() { + endTime = time.Now() + + var err error + if r := recover(); r != nil { + + switch x := r.(type) { + case string: + err = errors.New(x) + case error: + err = x + default: + err = errors.New("unknown panic") + } + + pfxlog.Logger().WithError(err).Errorf("panic caught during asynchronous event %s handler %v", eventualEvent.Type, handler) + } + + a.Emit(EventualEventProcessingListenerDoneName, &EventualEventProcessingListenerDone{ + Id: listenerExecId, + BatchId: info.batchId, + ProcessId: info.processId, + ListenerFunc: handler, + BatchEventIndex: info.batchEventIndex, + TotalEventIndex: info.totalEventIndex, + EventType: eventualEvent.Type, + Error: err, + StartTime: startTime, + EndTime: endTime, + }) + }() + + startTime = time.Now() + + a.Emit(EventualEventProcessingListenerStartName, &EventualEventProcessingListenerStart{ + Id: listenerExecId, + BatchId: info.batchId, + ProcessId: info.processId, + ListenerFunc: handler, + BatchEventIndex: info.batchEventIndex, + TotalEventIndex: info.totalEventIndex, + EventType: eventualEvent.Type, + StartTime: startTime, + }) + + handler(eventualEvent.Type, eventualEvent.Data) +} diff --git a/controller/persistence/session_store.go b/controller/persistence/session_store.go index 09b5cd2fe..1134f84d4 100644 --- a/controller/persistence/session_store.go +++ b/controller/persistence/session_store.go @@ -179,7 +179,7 @@ func (store *sessionStoreImpl) initializeLocal() { store.symbolServicePolicies = store.AddFkSetSymbol(FieldSessionServicePolicies, store.stores.servicePolicy) store.AddSymbol(FieldSessionType, ast.NodeTypeString) - store.AddFkConstraint(store.symbolApiSession, false, boltz.CascadeDelete) + store.AddFkConstraint(store.symbolApiSession, false, boltz.CascadeCreateUpdate) store.AddFkConstraint(store.symbolService, false, boltz.CascadeDelete) } diff --git a/controller/persistence/session_store_test.go b/controller/persistence/session_store_test.go index 3abd6288c..a18487d6e 100644 --- a/controller/persistence/session_store_test.go +++ b/controller/persistence/session_store_test.go @@ -135,7 +135,21 @@ func (ctx *TestContext) testCreateSessions(_ *testing.T) { ctx.RequireReload(session) ctx.RequireReload(session2) - ctx.RequireDelete(apiSession) + err := ctx.Delete(apiSession) + ctx.NoError(err) + + done, err := ctx.GetStores().EventualEventer.Trigger() + ctx.NoError(err) + + select { + case <-done: + case <-time.After(5 * time.Second): + ctx.Fail("did not receive done notification from eventual eventer") + + } + + ctx.ValidateDeleted(apiSession.GetId()) + ctx.ValidateDeleted(session.Id) ctx.ValidateDeleted(session2.Id) diff --git a/controller/persistence/stores.go b/controller/persistence/stores.go index 65bb298b8..fd028e5d2 100644 --- a/controller/persistence/stores.go +++ b/controller/persistence/stores.go @@ -25,10 +25,12 @@ import ( "github.com/pkg/errors" "go.etcd.io/bbolt" "reflect" + "time" ) type Stores struct { - DbProvider DbProvider + DbProvider DbProvider + EventualEventer EventualEventer // fabric stores Router db.RouterStore @@ -37,6 +39,7 @@ type Stores struct { ApiSession ApiSessionStore ApiSessionCertificate ApiSessionCertificateStore + EventualEvent EventualEventStore Ca CaStore Config ConfigStore ConfigType ConfigTypeStore @@ -116,7 +119,8 @@ func (stores *Stores) CheckIntegrityInTx(tx *bbolt.Tx, fix bool, errorHandler fu } type stores struct { - DbProvider DbProvider + DbProvider DbProvider + EventualEventer EventualEventer // fabric stores Router db.RouterStore @@ -124,6 +128,7 @@ type stores struct { Terminator db.TerminatorStore apiSession *apiSessionStoreImpl + eventualEvent *eventualEventStoreImpl ca *caStoreImpl config *configStoreImpl configType *configTypeStoreImpl @@ -153,10 +158,12 @@ func NewBoltStores(dbProvider DbProvider) (*Stores, error) { DbProvider: dbProvider, } + internalStores.eventualEvent = newEventualEventStore(internalStores) + internalStores.EventualEventer = NewEventualEventerBbolt(dbProvider, internalStores.eventualEvent, 2*time.Second, 1000) + internalStores.Terminator = dbProvider.GetStores().Terminator internalStores.Router = dbProvider.GetStores().Router internalStores.Service = dbProvider.GetStores().Service - internalStores.apiSession = newApiSessionStore(internalStores) internalStores.apiSessionCertificate = newApiSessionCertificateStore(internalStores) internalStores.authenticator = newAuthenticatorStore(internalStores) @@ -188,6 +195,7 @@ func NewBoltStores(dbProvider DbProvider) (*Stores, error) { ApiSession: internalStores.apiSession, ApiSessionCertificate: internalStores.apiSessionCertificate, + EventualEvent: internalStores.eventualEvent, Ca: internalStores.ca, Config: internalStores.config, ConfigType: internalStores.configType, @@ -211,6 +219,8 @@ func NewBoltStores(dbProvider DbProvider) (*Stores, error) { storeMap: make(map[reflect.Type]boltz.CrudStore), } + externalStores.EventualEventer = internalStores.EventualEventer + // The Index store is used for querying indexes. It's a convenient store with only a single value (id), which // is only ever queried using an index set cursor externalStores.Index = boltz.NewBaseStore("invalid", func(id string) error { diff --git a/controller/persistence/testing.go b/controller/persistence/testing.go index 4ef628e0f..5c9977b55 100644 --- a/controller/persistence/testing.go +++ b/controller/persistence/testing.go @@ -60,6 +60,7 @@ type TestContext struct { fabricStores *db.Stores stores *Stores controllers *network.Controllers + closeNotify chan struct{} } func NewTestContext(t *testing.T) *TestContext { @@ -67,11 +68,17 @@ func NewTestContext(t *testing.T) *TestContext { result := &TestContext{ BaseTestContext: *boltz.NewTestContext(t), + closeNotify: make(chan struct{}, 1), } result.Impl = result return result } +func (ctx *TestContext) Cleanup() { + close(ctx.closeNotify) + ctx.BaseTestContext.Cleanup() +} + func (ctx *TestContext) GetStores() *Stores { return ctx.stores } @@ -116,6 +123,9 @@ func (ctx *TestContext) InitWithDbFile(path string) { ctx.NoError(err) ctx.NoError(RunMigrations(ctx.GetDb(), ctx.stores)) + + ctx.NoError(ctx.stores.EventualEventer.Start(ctx.closeNotify)) + } func (ctx *TestContext) requireNewServicePolicy(policyType PolicyType, identityRoles []string, serviceRoles []string) *ServicePolicy { diff --git a/controller/server/controller.go b/controller/server/controller.go index 50299a799..56a52eb75 100644 --- a/controller/server/controller.go +++ b/controller/server/controller.go @@ -247,6 +247,10 @@ func (c *Controller) Initialize() { } + if err := c.AppEnv.GetStores().EventualEventer.Start(c.AppEnv.GetHostController().GetCloseNotifyChannel()); err != nil { + log.WithError(err).Panic("could not start EventualEventer") + } + c.initialized = true } diff --git a/tests/api_error_test.go b/tests/api_error_test.go index a788e41c0..c48749196 100644 --- a/tests/api_error_test.go +++ b/tests/api_error_test.go @@ -1,3 +1,4 @@ +//go:build apitests // +build apitests /* @@ -89,7 +90,6 @@ func Test_Api_Errors(t *testing.T) { resp, err := ctx.newAnonymousClientApiRequest(). SetHeader("accept", "appliaction/x-pem-file, application/json"). - Post("enroll?token=" + madeUpToken) ctx.Req.NoError(err) diff --git a/tests/api_session_certificates_test.go b/tests/api_session_certificates_test.go index 10fb5b233..57e0714b2 100644 --- a/tests/api_session_certificates_test.go +++ b/tests/api_session_certificates_test.go @@ -1,3 +1,4 @@ +//go:build apitests // +build apitests /* @@ -206,4 +207,3 @@ func generateCsr() ([]byte, crypto.PrivateKey, error) { return csr, privateKey, nil } - diff --git a/tests/authenticator_test.go b/tests/authenticator_test.go index 5cc3edaaf..a20f46535 100644 --- a/tests/authenticator_test.go +++ b/tests/authenticator_test.go @@ -1,3 +1,4 @@ +//go:build apitests // +build apitests /* @@ -436,7 +437,6 @@ func Test_Authenticators_NonAdminUsingSelfServiceEndpoints(t *testing.T) { certNonAdminUserSession, err := certNonAdminUserAuthenticator.AuthenticateClientApi(ctx) ctx.Req.NoError(err, "expected no error during non-admin cert authentication") - t.Run("can access their authenticators", func(t *testing.T) { req := require.New(t) resp, err := updbNonAdminSession.newAuthenticatedRequest().Get("/current-identity/authenticators") diff --git a/tests/ca_test.go b/tests/ca_test.go index c557517b7..b27eeecc4 100644 --- a/tests/ca_test.go +++ b/tests/ca_test.go @@ -1,3 +1,4 @@ +//go:build apitests // +build apitests package tests diff --git a/tests/config_test.go b/tests/config_test.go index f5979f699..ea9160470 100644 --- a/tests/config_test.go +++ b/tests/config_test.go @@ -1,3 +1,4 @@ +//go:build apitests // +build apitests /* diff --git a/tests/config_type_test.go b/tests/config_type_test.go index 4c5d715b8..5e23714e2 100644 --- a/tests/config_type_test.go +++ b/tests/config_type_test.go @@ -1,3 +1,4 @@ +//go:build apitests // +build apitests /* diff --git a/tests/current_identity_test.go b/tests/current_identity_test.go index e6db4e339..b5980b38b 100644 --- a/tests/current_identity_test.go +++ b/tests/current_identity_test.go @@ -1,3 +1,4 @@ +//go:build apitests // +build apitests /* diff --git a/tests/edge_router_policy_test.go b/tests/edge_router_policy_test.go index b493028e7..dd9380932 100644 --- a/tests/edge_router_policy_test.go +++ b/tests/edge_router_policy_test.go @@ -1,3 +1,4 @@ +//go:build apitests // +build apitests /* diff --git a/tests/edge_router_test.go b/tests/edge_router_test.go index 55ed4d185..c2be9a3ff 100644 --- a/tests/edge_router_test.go +++ b/tests/edge_router_test.go @@ -1,3 +1,4 @@ +//go:build apitests // +build apitests /* diff --git a/tests/enrollment_ca_auto_test.go b/tests/enrollment_ca_auto_test.go index 6eda8e9d2..13f981f12 100644 --- a/tests/enrollment_ca_auto_test.go +++ b/tests/enrollment_ca_auto_test.go @@ -1,3 +1,4 @@ +//go:build apitests // +build apitests /* diff --git a/tests/enrollment_router_test.go b/tests/enrollment_router_test.go index be84b2024..8dd4b0176 100644 --- a/tests/enrollment_router_test.go +++ b/tests/enrollment_router_test.go @@ -1,3 +1,4 @@ +//go:build apitests // +build apitests /* @@ -371,7 +372,7 @@ func Test_RouterEnrollment(t *testing.T) { ch, err := channel2.NewChannel("apitest", channel2.NewClassicDialer(id, ctx.ControllerConfig.Ctrl.Listener, nil), nil) - defer func(){ + defer func() { if ch != nil { _ = ch.Close() } @@ -383,7 +384,7 @@ func Test_RouterEnrollment(t *testing.T) { t.Run("requesting enrollment extension with", func(t *testing.T) { ctx.testContextChanged(t) - + t.Run("no existing router client certificate fails", func(t *testing.T) { ctx.testContextChanged(t) body := gabs.New() @@ -534,7 +535,6 @@ func Test_RouterEnrollment(t *testing.T) { extensionClientCerts[0].NotBefore.Before(time.Now()) }) - t.Run("the new server cert has its NotAfter date increased", func(t *testing.T) { ctx.testContextChanged(t) extensionServerCert[0].NotAfter.After(serverCerts[0].NotAfter) @@ -553,7 +553,7 @@ func Test_RouterEnrollment(t *testing.T) { ch, err := channel2.NewChannel("apitest", channel2.NewClassicDialer(id, ctx.ControllerConfig.Ctrl.Listener, nil), nil) - defer func(){ + defer func() { if ch != nil { _ = ch.Close() } @@ -570,7 +570,7 @@ func Test_RouterEnrollment(t *testing.T) { ch, err := channel2.NewChannel("apitestextension", channel2.NewClassicDialer(id, ctx.ControllerConfig.Ctrl.Listener, nil), nil) - defer func(){ + defer func() { if ch != nil { _ = ch.Close() } diff --git a/tests/enrollment_updb_test.go b/tests/enrollment_updb_test.go index 6e6781ff3..529c1ce44 100644 --- a/tests/enrollment_updb_test.go +++ b/tests/enrollment_updb_test.go @@ -1,3 +1,4 @@ +//go:build apitests // +build apitests /* @@ -47,7 +48,7 @@ func Test_UpdbEnrollment(t *testing.T) { "updb": updbUsername, }, "enrollment") updbCreate.Set(false, "isAdmin") - + resp, err := ctx.AdminManagementSession.newAuthenticatedRequest().SetBody(updbCreate.String()).Post("identities") ctx.Req.NoError(err) diff --git a/tests/identity_test.go b/tests/identity_test.go index 07d614c34..3fe1cbd3f 100644 --- a/tests/identity_test.go +++ b/tests/identity_test.go @@ -1,3 +1,4 @@ +//go:build apitests // +build apitests /* diff --git a/tests/mfa_ziti_test.go b/tests/mfa_ziti_test.go index 5cb1c02c4..e959d4005 100644 --- a/tests/mfa_ziti_test.go +++ b/tests/mfa_ziti_test.go @@ -1,3 +1,4 @@ +//go:build apitests // +build apitests /* diff --git a/tests/posture_check_domain_test.go b/tests/posture_check_domain_test.go index 389da46a8..c8c123e41 100644 --- a/tests/posture_check_domain_test.go +++ b/tests/posture_check_domain_test.go @@ -1,3 +1,4 @@ +//go:build apitests // +build apitests /* @@ -127,7 +128,6 @@ func Test_PostureChecks_Domain(t *testing.T) { ctx.Req.True(enrolledIdentitySession.isServiceVisibleToUser(service.Id)) }) - t.Run("service has the posture check in its queries", func(t *testing.T) { ctx.testContextChanged(t) code, body := enrolledIdentitySession.query("/services/" + service.Id) @@ -144,7 +144,7 @@ func Test_PostureChecks_Domain(t *testing.T) { if querySet[0].Path("policyId").Data().(string) == dialPolicy.id { dialSet = querySet[0] bindSet = querySet[1] - } else { + } else { dialSet = querySet[1] bindSet = querySet[0] } @@ -159,7 +159,6 @@ func Test_PostureChecks_Domain(t *testing.T) { ctx.Req.Equal(postureCheck.id, postureQueries[0].Path("id").Data().(string)) ctx.Req.Equal(postureCheck.typeId, postureQueries[0].Path("queryType").Data().(string)) - t.Run("query is currently failing", func(t *testing.T) { ctx.testContextChanged(t) ctx.Req.False(dialSet.Path("isPassing").Data().(bool)) @@ -205,7 +204,7 @@ func Test_PostureChecks_Domain(t *testing.T) { if querySet[0].Path("policyId").Data().(string) == dialPolicy.id { dialSet = querySet[0] bindSet = querySet[1] - } else { + } else { dialSet = querySet[1] bindSet = querySet[0] } @@ -264,7 +263,7 @@ func Test_PostureChecks_Domain(t *testing.T) { if querySet[0].Path("policyId").Data().(string) == dialPolicy.id { dialSet = querySet[0] bindSet = querySet[1] - } else { + } else { dialSet = querySet[1] bindSet = querySet[0] } @@ -328,7 +327,7 @@ func Test_PostureChecks_Domain(t *testing.T) { if querySet[0].Path("policyId").Data().(string) == dialPolicy.id { dialSet = querySet[0] bindSet = querySet[1] - } else { + } else { dialSet = querySet[1] bindSet = querySet[0] } @@ -395,7 +394,6 @@ func Test_PostureChecks_Domain(t *testing.T) { ctx.Req.True(enrolledIdentitySession.isServiceVisibleToUser(service.Id)) }) - t.Run("can create session with failing queries via one service policy by having another service policy with no checks", func(t *testing.T) { ctx.testContextChanged(t) resp, err := enrolledIdentitySession.createNewSession(service.Id) diff --git a/tests/posture_check_mfa_test.go b/tests/posture_check_mfa_test.go index 034714385..cc691840c 100644 --- a/tests/posture_check_mfa_test.go +++ b/tests/posture_check_mfa_test.go @@ -1,3 +1,4 @@ +//go:build apitests // +build apitests /* diff --git a/tests/posture_check_os_test.go b/tests/posture_check_os_test.go index 4444be702..863e9244b 100644 --- a/tests/posture_check_os_test.go +++ b/tests/posture_check_os_test.go @@ -1,3 +1,4 @@ +//go:build apitests // +build apitests /* diff --git a/tests/posture_check_process_multi_test.go b/tests/posture_check_process_multi_test.go index e5f23a9b4..7e06be1b0 100644 --- a/tests/posture_check_process_multi_test.go +++ b/tests/posture_check_process_multi_test.go @@ -1,3 +1,4 @@ +//go:build apitests // +build apitests /* diff --git a/tests/posture_check_session_test.go b/tests/posture_check_session_test.go index cfc4e8f59..308db5986 100644 --- a/tests/posture_check_session_test.go +++ b/tests/posture_check_session_test.go @@ -1,3 +1,4 @@ +//go:build apitests // +build apitests /* @@ -166,4 +167,4 @@ func Test_PostureChecks_Sessions(t *testing.T) { }) testServer.waitForDone(ctx, 5*time.Second) -} \ No newline at end of file +} diff --git a/tests/service_edge_router_policy_test.go b/tests/service_edge_router_policy_test.go index 160495495..ae0c24182 100644 --- a/tests/service_edge_router_policy_test.go +++ b/tests/service_edge_router_policy_test.go @@ -1,3 +1,4 @@ +//go:build apitests // +build apitests /* diff --git a/tests/service_list_refresh_test.go b/tests/service_list_refresh_test.go index 50a8e27dd..96363115e 100644 --- a/tests/service_list_refresh_test.go +++ b/tests/service_list_refresh_test.go @@ -1,3 +1,4 @@ +//go:build apitests // +build apitests package tests diff --git a/tests/service_policy_test.go b/tests/service_policy_test.go index b118d25d1..80704dcf8 100644 --- a/tests/service_policy_test.go +++ b/tests/service_policy_test.go @@ -1,3 +1,4 @@ +//go:build apitests // +build apitests /* diff --git a/tests/service_test.go b/tests/service_test.go index 1b801cf87..06554460b 100644 --- a/tests/service_test.go +++ b/tests/service_test.go @@ -1,3 +1,4 @@ +//go:build apitests // +build apitests /* diff --git a/tests/specs_test.go b/tests/specs_test.go index e37390f5c..5da96553e 100644 --- a/tests/specs_test.go +++ b/tests/specs_test.go @@ -1,3 +1,4 @@ +//go:build apitests // +build apitests /* diff --git a/tests/transit_router_test.go b/tests/transit_router_test.go index d65f8e112..37c2c613d 100644 --- a/tests/transit_router_test.go +++ b/tests/transit_router_test.go @@ -1,3 +1,4 @@ +//go:build apitests // +build apitests /*