fix: Ensure all MongoDB cursors are being closed (#7940)

mongodb-datastore/db/mongodb_event_repo.go

@@ -476,17 +476,20 @@ func (mr *MongoDBEventRepo) aggregateFromDB(collectionName string, pipeline mong
 	defer cancel()
 
 	cur, err := collection.Aggregate(ctx, pipeline)
-
-	if err != nil {
-		logger.WithError(err).Error("Could not retrieve events from collectiong elements in events collection")
-		return nil, err
-	}
 	// close the cursor after the function has completed to avoid memory leaks
 	defer func() {
+		if cur == nil {
+			return
+		}
 		if err := cur.Close(ctx); err != nil {
 			logger.WithError(err).Error("Could not close cursor")
 		}
 	}()
+	if err != nil {
+		logger.WithError(err).Error("Could not retrieve events from collectiong elements in events collection")
+		return nil, err
+	}
+
 	result.Events = formatEventResults(ctx, cur)
 
 	return result, nil
@@ -532,17 +535,20 @@ func (mr *MongoDBEventRepo) findInDB(collectionName string, pageSize int64, next
 	}
 
 	cur, err := collection.Find(ctx, searchOptions, sortOptions)
-
-	if err != nil {
-		logger.WithError(err).Error("Could not retrieve elements from events collection")
-		return nil, err
-	}
 	// close the cursor after the function has completed to avoid memory leaks
 	defer func() {
+		if cur == nil {
+			return
+		}
 		if err := cur.Close(ctx); err != nil {
 			logger.WithError(err).Error("Could not close cursor.")
 		}
 	}()
+	if err != nil {
+		logger.WithError(err).Error("Could not retrieve elements from events collection")
+		return nil, err
+	}
+
 	result.Events = formatEventResults(ctx, cur)
 
 	result.PageSize = pageSize

shipyard-controller/db/common.go

@@ -0,0 +1,16 @@
+package db
+
+import (
+	"context"
+	log "github.com/sirupsen/logrus"
+	"go.mongodb.org/mongo-driver/mongo"
+)
+
+func closeCursor(ctx context.Context, cur *mongo.Cursor) {
+	if cur == nil {
+		return
+	}
+	if err := cur.Close(ctx); err != nil {
+		log.Errorf("Could not close cursor: %v", err)
+	}
+}

shipyard-controller/db/mongodb.go

@@ -13,6 +13,8 @@ func SetupTTLIndex(ctx context.Context, propertyName string, duration time.Durat
 	ttlInSeconds := int32(duration.Seconds())
 	indexName := propertyName + "_1"
 	cur, err := collection.Indexes().List(ctx)
+	defer closeCursor(ctx, cur)
+
 	if err != nil {
 		return fmt.Errorf("could not load list of indexes of collection %s: %w", collection.Name(), err)
 	}

shipyard-controller/db/mongodb_event_repo.go

@@ -51,6 +51,7 @@ func (mdbrepo *MongoDBEventsRepo) GetEvents(project string, filter common.EventF
 	sortOptions := options.Find().SetSort(bson.D{{Key: "time", Value: -1}})
 
 	cur, err := collection.Find(ctx, searchOptions, sortOptions)
+	defer closeCursor(ctx, cur)
 	if err != nil && err == mongo.ErrNoDocuments {
 		return nil, ErrNoEventFound
 	} else if err != nil {
@@ -61,7 +62,6 @@ func (mdbrepo *MongoDBEventsRepo) GetEvents(project string, filter common.EventF
 
 	events := []models.Event{}
 
-	defer cur.Close(ctx)
 	for cur.Next(ctx) {
 		event, err := decodeKeptnEvent(cur)
 		if err != nil {
@@ -114,6 +114,7 @@ func (mdbrepo *MongoDBEventsRepo) GetRootEvents(getRootParams models.GetRootEven
 	}
 
 	cur, err := collection.Find(ctx, searchOptions, sortOptions)
+	defer closeCursor(ctx, cur)
 	if err != nil && err != mongo.ErrNoDocuments {
 		return nil, err
 	}

shipyard-controller/db/mongodb_eventqueue_repo.go

@@ -184,6 +184,7 @@ func (m *MongoDBEventQueueRepo) GetEventQueueSequenceStates(filter models.EventQ
 		searchOptions[stageScope] = filter.Scope.Stage
 	}
 	cur, err := collection.Find(ctx, searchOptions)
+	defer closeCursor(ctx, cur)
 	if err != nil && err == mongo.ErrNoDocuments {
 		return nil, ErrNoEventFound
 	} else if err != nil {
@@ -194,7 +195,6 @@ func (m *MongoDBEventQueueRepo) GetEventQueueSequenceStates(filter models.EventQ
 
 	stateItems := []models.EventQueueSequenceState{}
 
-	defer cur.Close(ctx)
 	for cur.Next(ctx) {
 		stateItem := models.EventQueueSequenceState{}
 		err := cur.Decode(&stateItem)
@@ -249,6 +249,7 @@ func insertQueueItemIntoCollection(ctx context.Context, collection *mongo.Collec
 
 func getQueueItemsFromCollection(collection *mongo.Collection, ctx context.Context, searchOptions bson.M, opts ...*options.FindOptions) ([]models.QueueItem, error) {
 	cur, err := collection.Find(ctx, searchOptions, opts...)
+	defer closeCursor(ctx, cur)
 	if err != nil && err == mongo.ErrNoDocuments {
 		return nil, ErrNoEventFound
 	} else if err != nil {
@@ -259,7 +260,6 @@ func getQueueItemsFromCollection(collection *mongo.Collection, ctx context.Conte
 
 	queuedItems := []models.QueueItem{}
 
-	defer cur.Close(ctx)
 	for cur.Next(ctx) {
 		queueItem := models.QueueItem{}
 		err := cur.Decode(&queueItem)

shipyard-controller/db/mongodb_log_repo.go

@@ -80,6 +80,7 @@ func (mdbrepo *MongoDBLogRepo) GetLogEntries(params models.GetLogParams) (*model
 	}
 
 	cur, err := collection.Find(ctx, searchOptions, sortOptions)
+	defer closeCursor(ctx, cur)
 	if err != nil && err != mongo.ErrNoDocuments {
 		return nil, err
 	}

shipyard-controller/db/mongodb_state_repo.go

@@ -83,6 +83,7 @@ func (mdbrepo *MongoDBStateRepo) FindSequenceStates(filter models.StateFilter) (
 	}
 
 	cur, err := collection.Find(ctx, searchOptions, sortOptions)
+	defer closeCursor(ctx, cur)
 	if err != nil && err != mongo.ErrNoDocuments {
 		return nil, err
 	}

shipyard-controller/db/mongodb_task_sequence_repo.go

@@ -32,6 +32,7 @@ func (mdbrepo *TaskSequenceMongoDBRepo) GetTaskExecutions(project string, filter
 
 	collection := mdbrepo.getTaskSequenceCollection(project)
 	cur, err := collection.Find(ctx, mdbrepo.getTaskSequenceMappingSearchOptions(filter))
+	defer closeCursor(ctx, cur)
 	if err != nil && err != mongo.ErrNoDocuments {
 		return nil, err
 	}

shipyard-controller/db/mongodb_uniform_repo.go

@@ -328,6 +328,7 @@ func (mdbrepo *MongoDBUniformRepo) getCollectionAndContext() (*mongo.Collection,
 func (mdbrepo *MongoDBUniformRepo) findIntegrations(searchParams models.GetUniformIntegrationsParams, collection *mongo.Collection, ctx context.Context) ([]models.Integration, error) {
 	searchOptions := mdbrepo.getSearchOptions(searchParams)
 	cur, err := collection.Find(ctx, searchOptions)
+	defer closeCursor(ctx, cur)
 	if err != nil && err != mongo.ErrNoDocuments {
 		return nil, err
 	}
@@ -361,7 +362,7 @@ func (mdbrepo *MongoDBUniformRepo) DeleteServiceFromSubscriptions(subscriptionNa
 	}
 
 	cur, err := collection.Find(ctx, filter)
-
+	defer closeCursor(ctx, cur)
 	if err != nil && err != mongo.ErrNoDocuments {
 		return err
 	}

statistics-service/db/migrator.go

@@ -63,10 +63,15 @@ func (m *Migrator) migrateBatch(ctx context.Context) (bool, error) {
 		Limit: crateInt64P(m.batchSize),
 		Skip:  crateInt64P(skips),
 	})
+	defer func() {
+		if cur == nil {
+			return
+		}
+		cur.Close(ctx)
+	}()
 	if err != nil {
 		return false, err
 	}
-	defer cur.Close(ctx)
 
 	currBatchSize := 0
 	for cur.Next(ctx) {

statistics-service/db/statistics_mongodb_repo.go

@@ -42,12 +42,18 @@ func (s *StatisticsMongoDBRepo) GetStatistics(from, to time.Time) ([]operations.
 	}
 
 	cur, err := s.statsCollection.Find(ctx, searchOptions)
+	defer func() {
+		if cur == nil {
+			return
+		}
+		cur.Close(ctx)
+	}()
 	if err != nil {
 		return nil, err
 	}
 
 	result := []operations.Statistics{}
-	defer cur.Close(ctx)
+
 	if cur.RemainingBatchLength() == 0 {
 		return nil, ErrNoStatisticsFound
 	}