wip: working prototype. needs rebasing tho

Author: oxzi

pkg/icingadb/sync.go

@@ -173,34 +173,29 @@ func (s Sync) ApplyDelta(ctx context.Context, delta *Delta) error {
 	if len(delta.Delete) > 0 {
 		s.logger.Infof("Deleting %d items of type %s", len(delta.Delete), strcase.Delimited(types.Name(delta.Subject.Entity()), ' '))
 
-		// TODO: generalize, make errCh easier to read
 		ids := delta.Delete.IDs()
-		if _, ok := delta.Subject.Entity().(*v1.Downtime); ok {
-			// Those downtimes are probably removed from the configuration files, i.e. Icinga 2 won't send
-			// the corresponding downtime end/removed events for them. So try to mark them as cancelled manually,
-			// so that they don't show up as if they were still active in the UI.
-			//
-			// The reason why we don't perform this in a separate goroutine like the other ones is that we don't
-			// want to delete the downtimes until we've successfully updated the downtime history records. So,
-			// if we fail to update the downtime history records, for whatever reason, forward the error to the
-			// config sync group and abort the config sync.
-			if err := s.mockDowntimeEnd(ctx, ids); err != nil {
-				errCh := make(chan error, 1)
-				errCh <- err
-				close(errCh)
-				com.ErrgroupReceive(g, errCh)
-			}
-		} else if _, ok := delta.Subject.Entity().(*v1.Comment); ok {
-			if err := s.mockCommentEnd(ctx, ids); err != nil {
-				errCh := make(chan error, 1)
-				errCh <- err
-				close(errCh)
-				com.ErrgroupReceive(g, errCh)
-			}
+
+		// Those objects are probably removed from the configuration files, i.e., Icinga 2 won't send the
+		// corresponding end/removed events for them. So try to mark them as cancelled manually, so that
+		// they don't show up as if they were still active in the UI.
+		//
+		// This is not performed in a separate goroutine to ensure that the relevant Comment/Downtime
+		// object is not deleted before closing the history.
+
+		var err error
+		switch delta.Subject.Entity().(type) {
+		case *v1.Comment:
+			err = s.mockCommentEnd(ctx, ids)
+		case *v1.Downtime:
+			err = s.mockDowntimeEnd(ctx, ids)
+		}
+
+		if ctx.Err() == nil {
+			g.Go(func() error { return err })
 		}
 
-		// Start the deletion process only if we haven't aborted the config sync
-		// due to an error in the downtime history update above.
+		// Start the deletion process only if we haven't aborted the config sync due to an error in the
+		// downtime history update above.
 		if ctx.Err() == nil {
 			g.Go(func() error {
 				return s.db.Delete(ctx, delta.Subject.Entity(), ids, database.OnSuccessIncrement[any](stat))

pkg/icingadb/sync_history_mock.go

@@ -7,29 +7,15 @@ import (
 	"time"
 
 	"github.com/icinga/icinga-go-library/backoff"
-	"github.com/icinga/icinga-go-library/database"
 	"github.com/icinga/icinga-go-library/objectpacker"
 	"github.com/icinga/icinga-go-library/retry"
 	"github.com/icinga/icinga-go-library/types"
 	"github.com/icinga/icingadb/pkg/icingadb/v1/history"
 	"github.com/jmoiron/sqlx"
 	"github.com/pkg/errors"
+	"go.uber.org/zap"
 )
 
-// calcEventId mocks Icinga 2's IcingaDB::CalcEventID method for missing history entries.
-//
-// This is a helper function used below to simulate a new history.id primary key.
-func calcEventId(environmentId, eventType, objectName string) (types.Binary, error) {
-	h := sha1.New() // #nosec G401 -- required by Icinga 2's HashValue function
-
-	err := objectpacker.PackAny([]string{environmentId, eventType, objectName}, h)
-	if err != nil {
-		return nil, err
-	}
-
-	return types.Binary(h.Sum(nil)), nil
-}
-
 // mkGenericHistoryMockStmts creates SQL queries to acquire data and to create a new history entry.
 //
 // This helper function is used in the methods implementing closeObjectAndMockHistory below.
@@ -57,37 +43,57 @@ func mkGenericHistoryMockStmts(tableName string) (stmtPopulate, stmtHistoryInser
 	return
 }
 
+// calcEventId mocks Icinga 2's IcingaDB::CalcEventID method for missing history entries.
+//
+// This is a helper function used below to simulate a new history.id primary key.
+func calcEventId(environmentId, eventType, objectName string) (types.Binary, error) {
+	h := sha1.New() // #nosec G401 -- required by Icinga 2's HashValue function
+
+	err := objectpacker.PackAny([]string{environmentId, eventType, objectName}, h)
+	if err != nil {
+		return nil, err
+	}
+
+	return types.Binary(h.Sum(nil)), nil
+}
+
 // closeObjectAndMockHistory manually closes vanished Icinga 2 objects, e.g., Downtimes of vanished Hosts.
 //
 // This function provides a common stub for each type-specific implementation below.
-func closeObjectAndMockHistory(
+func (s Sync) closeObjectAndMockHistory(
 	ctx context.Context,
-	db *database.DB,
 	ids []any,
+	objectType string,
 	action func(ctx context.Context, tx *sqlx.Tx, id types.Binary) error,
 ) error {
+	startTime := time.Now()
+
 	for _, id := range ids {
 		binaryId, ok := id.(types.Binary)
 		if !ok {
 			return errors.Errorf("ID is not types.Binary, but %T", id)
 		}
 
-		// TODO: Pipeline multiple action calls into one transaction.
 		err := retry.WithBackoff(
 			ctx,
 			func(ctx context.Context) error {
-				return db.ExecTx(ctx, func(ctx context.Context, tx *sqlx.Tx) error {
+				return s.db.ExecTx(ctx, func(ctx context.Context, tx *sqlx.Tx) error {
 					return action(ctx, tx, binaryId)
 				})
 			},
 			retry.Retryable,
 			backoff.DefaultBackoff,
-			db.GetDefaultRetrySettings())
+			s.db.GetDefaultRetrySettings())
 		if err != nil {
 			return errors.Wrap(err, "can't mock object end")
 		}
 	}
 
+	s.logger.Infow("Manually closed vanished objects",
+		zap.String("type", objectType),
+		zap.Int("amount", len(ids)),
+		zap.Duration("duration", time.Since(startTime)))
+
 	return nil
 }
 
@@ -114,11 +120,13 @@ func (s Sync) mockCommentEnd(ctx context.Context, ids []any) error {
 			has_been_removed = :has_been_removed
 		WHERE comment_id = :comment_id`
 
-	return closeObjectAndMockHistory(
+	return s.closeObjectAndMockHistory(
 		ctx,
-		s.db,
 		ids,
+		"comment",
 		func(ctx context.Context, tx *sqlx.Tx, id types.Binary) error {
+			eventTime := time.Now()
+
 			comment := &Comment{
 				CommentHistory: history.CommentHistory{
 					CommentHistoryEntity: history.CommentHistoryEntity{
@@ -141,7 +149,7 @@ func (s Sync) mockCommentEnd(ctx context.Context, ids []any) error {
 
 			// Fill fields to update comment_history
 			comment.RemovedBy = types.MakeString("Comment Config Removed")
-			comment.RemoveTime = types.UnixMilli(time.Now())
+			comment.RemoveTime = types.UnixMilli(eventTime)
 			comment.HasBeenRemoved = types.MakeBool(true)
 
 			_, err = tx.NamedExecContext(ctx, stmtCommentHistoryUpdate, comment)
@@ -156,7 +164,7 @@ func (s Sync) mockCommentEnd(ctx context.Context, ids []any) error {
 			}
 
 			comment.EventType = "comment_remove"
-			comment.EventTime = types.UnixMilli(time.Now())
+			comment.EventTime = types.UnixMilli(eventTime)
 
 			_, err = tx.NamedExecContext(ctx, stmtHistoryInsert, comment)
 			if err != nil {
@@ -196,11 +204,13 @@ func (s Sync) mockDowntimeEnd(ctx context.Context, ids []any) error {
 		SET downtime_end = :cancel_time
 		WHERE downtime_id = :downtime_id`
 
-	return closeObjectAndMockHistory(
+	return s.closeObjectAndMockHistory(
 		ctx,
-		s.db,
 		ids,
+		"downtime",
 		func(ctx context.Context, tx *sqlx.Tx, id types.Binary) error {
+			eventTime := time.Now()
+
 			downtime := &Downtime{
 				DowntimeHistory: history.DowntimeHistory{
 					DowntimeHistoryEntity: history.DowntimeHistoryEntity{
@@ -224,7 +234,7 @@ func (s Sync) mockDowntimeEnd(ctx context.Context, ids []any) error {
 			// Fill fields to update downtime_history and sla_history_downtime.
 			downtime.CancelledBy = types.MakeString("Downtime Config Removed")
 			downtime.HasBeenCancelled = types.MakeBool(true)
-			downtime.CancelTime = types.UnixMilli(time.Now())
+			downtime.CancelTime = types.UnixMilli(eventTime)
 
 			_, err = tx.NamedExecContext(ctx, stmtDowntimeHistoryUpdate, downtime)
 			if err != nil {
@@ -243,7 +253,7 @@ func (s Sync) mockDowntimeEnd(ctx context.Context, ids []any) error {
 			}
 
 			downtime.EventType = "downtime_end"
-			downtime.EventTime = types.UnixMilli(time.Now())
+			downtime.EventTime = types.UnixMilli(eventTime)
 
 			_, err = tx.NamedExecContext(ctx, stmtHistoryInsert, downtime)
 			if err != nil {