Write checkable sla history events in runtime updates

Author: yhabteab

pkg/icingadb/runtime_updates.go

@@ -58,7 +58,7 @@ func (r *RuntimeUpdates) ClearStreams(ctx context.Context) (config, state redis.
 }
 
 // Sync synchronizes runtime update streams from s.redis to s.db and deletes the original data on success.
-// Note that Sync must be only be called configuration synchronization has been completed.
+// Note that Sync must only be called once configuration synchronization has been completed.
 // allowParallel allows synchronizing out of order (not FIFO).
 func (r *RuntimeUpdates) Sync(
 	ctx context.Context, factoryFuncs []database.EntityFactoryFunc, streams redis.Streams, allowParallel bool,
@@ -73,7 +73,7 @@ func (r *RuntimeUpdates) Sync(
 
 		updateMessages := make(chan redis.XMessage, r.redis.Options.XReadCount)
 		upsertEntities := make(chan database.Entity, r.redis.Options.XReadCount)
-		deleteIds := make(chan interface{}, r.redis.Options.XReadCount)
+		deleteEntities := make(chan database.Entity, r.redis.Options.XReadCount)
 
 		var upsertedFifo chan database.Entity
 		var deletedFifo chan interface{}
@@ -95,60 +95,176 @@ func (r *RuntimeUpdates) Sync(
 		r.logger.Debugf("Syncing runtime updates of %s", s.Name())
 
 		g.Go(structifyStream(
-			ctx, updateMessages, upsertEntities, upsertedFifo, deleteIds, deletedFifo,
+			ctx, updateMessages, upsertEntities, upsertedFifo, deleteEntities, deletedFifo,
 			structify.MakeMapStructifier(
 				reflect.TypeOf(s.Entity()).Elem(),
 				"json",
 				contracts.SafeInit),
 		))
 
-		g.Go(func() error {
-			var counter com.Counter
-			defer periodic.Start(ctx, r.logger.Interval(), func(_ periodic.Tick) {
-				if count := counter.Reset(); count > 0 {
-					r.logger.Infof("Upserted %d %s items", count, s.Name())
+		// upsertEntityFunc returns a closure that is used to upsert the regular Icinga DB entities.
+		// The returned func is used to directly start a separate goroutine that selects events
+		// sequentially (!allowParallel) from the given chan.
+		upsertEntityFunc := func(entities <-chan database.Entity) func() error {
+			return func() error {
+				var counter com.Counter
+				defer periodic.Start(ctx, r.logger.Interval(), func(_ periodic.Tick) {
+					if count := counter.Reset(); count > 0 {
+						r.logger.Infof("Upserted %d %s items", count, s.Name())
+					}
+				}).Stop()
+
+				onSuccess := []database.OnSuccess[database.Entity]{
+					database.OnSuccessIncrement[database.Entity](&counter),
+					database.OnSuccessIncrement[database.Entity](stat),
+				}
+				if !allowParallel {
+					onSuccess = append(onSuccess, database.OnSuccessSendTo(upsertedFifo))
 				}
-			}).Stop()
 
-			// Updates must be executed in order, ensure this by using a semaphore with maximum 1.
-			sem := semaphore.NewWeighted(1)
+				// Updates must be executed in order, ensure this by using a semaphore with maximum 1.
+				sem := semaphore.NewWeighted(1)
 
-			onSuccess := []database.OnSuccess[database.Entity]{
-				database.OnSuccessIncrement[database.Entity](&counter), database.OnSuccessIncrement[database.Entity](stat),
-			}
-			if !allowParallel {
-				onSuccess = append(onSuccess, database.OnSuccessSendTo(upsertedFifo))
+				return r.db.NamedBulkExec(
+					ctx, upsertStmt, upsertCount, sem, entities, database.SplitOnDupId[database.Entity], onSuccess...,
+				)
 			}
+		}
 
-			return r.db.NamedBulkExec(
-				ctx, upsertStmt, upsertCount, sem, upsertEntities, database.SplitOnDupId[database.Entity], onSuccess...,
-			)
-		})
+		// deleteEntityFunc returns a closure that is used to delete the regular Icinga DB entities
+		// based on their ids. The returned func is used to directly start a separate goroutine that
+		// selects events sequentially (!allowParallel) from the given chan.
+		deleteEntityFunc := func(deleteIds <-chan any) func() error {
+			return func() error {
+				var counter com.Counter
+				defer periodic.Start(ctx, r.logger.Interval(), func(_ periodic.Tick) {
+					if count := counter.Reset(); count > 0 {
+						r.logger.Infof("Deleted %d %s items", count, s.Name())
+					}
+				}).Stop()
 
-		g.Go(func() error {
-			var counter com.Counter
-			defer periodic.Start(ctx, r.logger.Interval(), func(_ periodic.Tick) {
-				if count := counter.Reset(); count > 0 {
-					r.logger.Infof("Deleted %d %s items", count, s.Name())
+				onSuccess := []database.OnSuccess[any]{database.OnSuccessIncrement[any](&counter), database.OnSuccessIncrement[any](stat)}
+				if !allowParallel {
+					onSuccess = append(onSuccess, database.OnSuccessSendTo(deletedFifo))
 				}
-			}).Stop()
 
-			sem := r.db.GetSemaphoreForTable(database.TableName(s.Entity()))
+				sem := r.db.GetSemaphoreForTable(database.TableName(s.Entity()))
 
-			onSuccess := []database.OnSuccess[any]{database.OnSuccessIncrement[any](&counter), database.OnSuccessIncrement[any](stat)}
-			if !allowParallel {
-				onSuccess = append(onSuccess, database.OnSuccessSendTo(deletedFifo))
+				return r.db.BulkExec(ctx, r.db.BuildDeleteStmt(s.Entity()), deleteCount, sem, deleteIds, onSuccess...)
 			}
+		}
 
-			return r.db.BulkExec(ctx, r.db.BuildDeleteStmt(s.Entity()), deleteCount, sem, deleteIds, onSuccess...)
-		})
+		// In order to always get the sla entries written even in case of system errors, we need to process these
+		// first. Otherwise, Icinga DB may be stopped after the regular queries have been processed, and deleted
+		// from the Redis stream, thus we won't be able to generate sla lifecycle for these entities.
+		//
+		// The general event process flow looks as follows:
+		//     structifyStream()  ->  Reads `upsert` & `delete` events from redis and streams the entities to the
+		//                            respective chans `upsertEntities`, `deleteEntities` and waits for `upserted`
+		//                            and `deleted` chans (!allowParallel) before consuming the next one from redis.
+		// - Starts a goroutine that consumes from `upsertEntities` (when the current sync subject is of type checkable,
+		//   this bulk inserts into the sla lifecycle table with semaphore 1 in a `INSERT ... IGNORE ON ERROR` fashion
+		//   and forwards the entities to the next one, which then inserts the checkables into the regular Icinga DB
+		//   tables). After successfully upserting the entities, (!allowParallel) they are passed sequentially to
+		//   the `upserted` stream.
+		//
+		// - Starts another goroutine that consumes from `deleteEntities` concurrently. When the current sync subject is
+		//   of type checkable, this performs sla lifecycle updates matching the checkables id and `delete_time` 0.
+		//   Since the upgrade script should've generated a sla entry for all exiting Checkables, it's unlikely to
+		//   happen, but when there is no tracked `created_at` event for a given checkable, this update is essentially
+		//   a no-op, but forwards the original entities IDs nonetheless to the next one. And finally, the original
+		//   checkables are deleted from the database and (!allowParallel) they are passed sequentially to the
+		//   `deleted` stream.
+		switch s.Entity().(type) {
+		case *v1.Host, *v1.Service:
+			entities := make(chan database.Entity, 1)
+			g.Go(func() error {
+				defer close(entities)
+
+				var counter com.Counter
+				defer periodic.Start(ctx, r.logger.Interval(), func(_ periodic.Tick) {
+					if count := counter.Reset(); count > 0 {
+						r.logger.Infof("Upserted %d %s sla lifecycles", count, s.Name())
+					}
+				}).Stop()
+
+				stmt, _ := r.db.BuildInsertIgnoreStmt(v1.NewSlaLifecycle())
+
+				// Not to mess up the already existing FIFO mechanism, we have to perform only a single query
+				// (semaphore 1) at a time, even though, the sla queries could be executed concurrently.
+				// After successfully upserting a lifecycle entity, the original checkable entity is streamed to "entities".
+				fromSlaEntities := CreateSlaLifecyclesFromCheckables(ctx, g, upsertEntities, false)
+				return r.db.NamedBulkExec(
+					ctx, stmt, upsertCount, semaphore.NewWeighted(1), fromSlaEntities,
+					com.NeverSplit[database.Entity], database.OnSuccessIncrement[database.Entity](&counter),
+					OnSuccessApplyAndSendTo(entities, GetCheckableFromSlaLifecycle))
+			})
+
+			// Start the regular Icinga DB checkables upsert stream.
+			g.Go(upsertEntityFunc(entities))
+
+			deletedIds := make(chan any, r.redis.Options.XReadCount)
+			g.Go(func() error {
+				defer close(deletedIds)
+
+				var counter com.Counter
+				defer periodic.Start(ctx, r.logger.Interval(), func(_ periodic.Tick) {
+					if count := counter.Reset(); count > 0 {
+						r.logger.Infof("Updating %d %s sla lifecycles", count, s.Name())
+					}
+				}).Stop()
+
+				sl := v1.NewSlaLifecycle()
+				stmt := fmt.Sprintf(`UPDATE %s SET delete_time = :delete_time WHERE "id" = :id AND "delete_time" = 0`, database.TableName(sl))
+				sem := r.db.GetSemaphoreForTable(database.TableName(sl))
+
+				// 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 r.db.NamedBulkExec(
+					ctx, stmt, upsertCount, sem, CreateSlaLifecyclesFromCheckables(ctx, g, deleteEntities, true),
+					com.NeverSplit[database.Entity], database.OnSuccessIncrement[database.Entity](&counter),
+					OnSuccessApplyAndSendTo[database.Entity, any](deletedIds, extractEntityId))
+			})
+
+			// Start the regular Icinga DB checkables delete stream.
+			g.Go(deleteEntityFunc(deletedIds))
+		default:
+			// For non-checkables runtime updates of upsert event
+			g.Go(upsertEntityFunc(upsertEntities))
+
+			// For non-checkables runtime updates of delete event
+			deleteIds := make(chan any, r.redis.Options.XReadCount)
+			g.Go(func() error {
+				defer close(deleteIds)
+
+				for {
+					select {
+					case <-ctx.Done():
+						return ctx.Err()
+					case entity, ok := <-deleteEntities:
+						if !ok {
+							return nil
+						}
+
+						select {
+						case deleteIds <- entity.ID():
+						case <-ctx.Done():
+							return ctx.Err()
+						}
+					}
+				}
+			})
+
+			g.Go(deleteEntityFunc(deleteIds))
+		}
 	}
 
 	// customvar and customvar_flat sync.
 	{
 		updateMessages := make(chan redis.XMessage, r.redis.Options.XReadCount)
 		upsertEntities := make(chan database.Entity, r.redis.Options.XReadCount)
-		deleteIds := make(chan interface{}, r.redis.Options.XReadCount)
+		deleteEntities := make(chan database.Entity, r.redis.Options.XReadCount)
 
 		cv := common.NewSyncSubject(v1.NewCustomvar)
 		cvFlat := common.NewSyncSubject(v1.NewCustomvarFlat)
@@ -158,7 +274,7 @@ func (r *RuntimeUpdates) Sync(
 
 		updateMessagesByKey["icinga:"+strcase.Delimited(cv.Name(), ':')] = updateMessages
 		g.Go(structifyStream(
-			ctx, updateMessages, upsertEntities, nil, deleteIds, nil,
+			ctx, updateMessages, upsertEntities, nil, deleteEntities, nil,
 			structify.MakeMapStructifier(
 				reflect.TypeOf(cv.Entity()).Elem(),
 				"json",
@@ -212,7 +328,7 @@ func (r *RuntimeUpdates) Sync(
 			var once sync.Once
 			for {
 				select {
-				case _, ok := <-deleteIds:
+				case _, ok := <-deleteEntities:
 					if !ok {
 						return nil
 					}
@@ -302,7 +418,7 @@ func (r *RuntimeUpdates) xRead(ctx context.Context, updateMessagesByKey map[stri
 // Converted entities are inserted into the upsertEntities or deleteIds channel depending on the "runtime_type" message field.
 func structifyStream(
 	ctx context.Context, updateMessages <-chan redis.XMessage, upsertEntities, upserted chan database.Entity,
-	deleteIds, deleted chan interface{}, structifier structify.MapStructifier,
+	deleteEntities chan database.Entity, deleted chan interface{}, structifier structify.MapStructifier,
 ) func() error {
 	if upserted == nil {
 		upserted = make(chan database.Entity)
@@ -317,7 +433,7 @@ func structifyStream(
 	return func() error {
 		defer func() {
 			close(upsertEntities)
-			close(deleteIds)
+			close(deleteEntities)
 		}()
 
 		for {
@@ -353,7 +469,7 @@ func structifyStream(
 					}
 				} else if runtimeType == "delete" {
 					select {
-					case deleteIds <- entity.ID():
+					case deleteEntities <- entity:
 					case <-ctx.Done():
 						return ctx.Err()
 					}