Don't bulk execute sla lifecycle updates

Author: yhabteab

pkg/icingadb/runtime_updates.go

@@ -204,7 +204,7 @@ func (r *RuntimeUpdates) Sync(
 			g.Go(upsertEntityFunc(entities))
 
 			// Start the regular Icinga DB checkables delete stream.
-			g.Go(deleteEntityFunc(StreamIDsFromUpdatedSlaLifecycles(ctx, r.db, s.Entity(), g, r.logger, deleteEntities, upsertCount)))
+			g.Go(deleteEntityFunc(StreamIDsFromUpdatedSlaLifecycles(ctx, r.db, s.Entity(), g, r.logger, deleteEntities)))
 		default:
 			// For non-checkables runtime updates of upsert event
 			g.Go(upsertEntityFunc(upsertEntities))

pkg/icingadb/sla_lifecycle.go

@@ -101,7 +101,7 @@ func CreateSlaLifecyclesFromCheckables(
 // It's unlikely, but when a given Checkable doesn't already have a `create_time` entry in the database, the update
 // query won't update anything. Either way the entities IDs are streamed into the returned chan.
 func StreamIDsFromUpdatedSlaLifecycles(
-	ctx context.Context, db *database.DB, subject database.Entity, g *errgroup.Group, logger *logging.Logger, entities <-chan database.Entity, bulkSize int,
+	ctx context.Context, db *database.DB, subject database.Entity, g *errgroup.Group, logger *logging.Logger, entities <-chan database.Entity,
 ) <-chan any {
 	deleteEntityIDs := make(chan any, 1)
 
@@ -118,15 +118,11 @@ func StreamIDsFromUpdatedSlaLifecycles(
 		sem := db.GetSemaphoreForTable(tableName)
 		stmt := fmt.Sprintf(`UPDATE %s SET delete_time = :delete_time WHERE "id" = :id AND "delete_time" = 0`, tableName)
 
-		if bulkSize <= 0 {
-			bulkSize = db.Options.MaxPlaceholdersPerStatement
-		}
-
 		// extractEntityId is used as a callback for the on success mechanism to extract the checkables id.
 		extractEntityId := func(e database.Entity) any { return e.(*v1.SlaLifecycle).SourceEntity.ID() }
 
 		return db.NamedBulkExec(
-			ctx, stmt, bulkSize, sem, CreateSlaLifecyclesFromCheckables(ctx, subject, g, entities, true),
+			ctx, stmt, 1, sem, CreateSlaLifecyclesFromCheckables(ctx, subject, g, entities, true),
 			com.NeverSplit[database.Entity], OnSuccessApplyAndSendTo[database.Entity, any](deleteEntityIDs, extractEntityId))
 	})
 

pkg/icingadb/sync.go

@@ -186,7 +186,7 @@ func (s Sync) ApplyDelta(ctx context.Context, delta *Delta) error {
 		case *v1.Host, *v1.Service:
 			g.Go(func() error {
 				return s.db.DeleteStreamed(
-					ctx, entity, StreamIDsFromUpdatedSlaLifecycles(ctx, s.db, entity, g, s.logger, delta.Delete.Entities(ctx), 0),
+					ctx, entity, StreamIDsFromUpdatedSlaLifecycles(ctx, s.db, entity, g, s.logger, delta.Delete.Entities(ctx)),
 					database.OnSuccessIncrement[any](stat))
 			})
 		default: