Incremental Configuration Updates

Previously, the entire configuration stored in the database was
synchronized every second. With the growth of configurations in live
environments on the horizon, this would simply not scale well. This
brings us to incremental updates.

By introducing two new columns - "changed_at" as a Unix millisecond
timestamp and "deleted" as a boolean - for all tables referenced in the
ConfigSet structure, SQL queries can be modified to retrieve only those
rows with a more recent timestamp. The "deleted" column became necessary
to detect disappearances, since the synchronization now only takes newer
items into account. Some additional fields needed to be added to the
ConfigSet to track relationships.

Even though the codebase served well at the time, there was some code
that did almost the same thing as other code, just in different ways.
So a huge refactoring was done. This resulted in an internal generic
function that handles all synchronization with custom callbacks.

The web counterpart is being developed in
<Icinga/icinga-notifications-web#187>.

Closes #5.

Author: oxzi

internal/channel/channel.go

@@ -4,16 +4,19 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/icinga/icinga-notifications/internal/config/baseconf"
 	"github.com/icinga/icinga-notifications/internal/contracts"
 	"github.com/icinga/icinga-notifications/internal/event"
 	"github.com/icinga/icinga-notifications/internal/recipient"
 	"github.com/icinga/icinga-notifications/pkg/plugin"
 	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
 	"net/url"
 )
 
 type Channel struct {
-	ID     int64  `db:"id"`
+	baseconf.IncrementalPkDbEntry[int64] `db:",inline"`
+
 	Name   string `db:"name"`
 	Type   string `db:"type"`
 	Config string `db:"config" json:"-"` // excluded from JSON config dump as this may contain sensitive information
@@ -27,6 +30,19 @@ type Channel struct {
 	pluginCtxCancel func()
 }
 
+// MarshalLogObject implements the zapcore.ObjectMarshaler interface.
+func (c *Channel) MarshalLogObject(encoder zapcore.ObjectEncoder) error {
+	encoder.AddInt64("id", c.ID)
+	encoder.AddString("name", c.Name)
+	encoder.AddString("type", c.Type)
+	return nil
+}
+
+// IncrementalInitAndValidate implements the config.IncrementalConfigurableInitAndValidatable interface.
+func (c *Channel) IncrementalInitAndValidate() error {
+	return ValidateType(c.Type)
+}
+
 // newConfig helps to store the channel's updated properties
 type newConfig struct {
 	ctype  string

internal/config/baseconf/incremental_config.go

@@ -0,0 +1,45 @@
+package baseconf
+
+import (
+	"github.com/icinga/icinga-go-library/types"
+)
+
+// IncrementalDbEntry contains the changed_at and deleted columns as struct fields.
+//
+// This type partially implements config.IncrementalConfigurable with GetChangedAt and IsDeleted. Thus, it can be
+// embedded in other types with the _`db:",inline"`_ struct tag. However, most structs might want to embed the
+// IncrementalPkDbEntry struct instead.
+type IncrementalDbEntry struct {
+	ChangedAt types.UnixMilli `db:"changed_at"`
+	Deleted   types.Bool      `db:"deleted"`
+}
+
+// GetChangedAt returns the changed_at value of this entry from the database.
+//
+// It is required by the config.IncrementalConfigurable interface.
+func (i IncrementalDbEntry) GetChangedAt() types.UnixMilli {
+	return i.ChangedAt
+}
+
+// IsDeleted indicates if this entry is marked as deleted in the database.
+//
+// It is required by the config.IncrementalConfigurable interface.
+func (i IncrementalDbEntry) IsDeleted() bool {
+	return i.Deleted.Valid && i.Deleted.Bool
+}
+
+// IncrementalPkDbEntry implements a single primary key named id of a generic type next to IncrementalDbEntry.
+//
+// This type embeds IncrementalDbEntry and adds a single column/value id field, getting one step closer to implementing
+// the config.IncrementalConfigurable interface. Thus, it needs to be embedded with the _`db:",inline"`_ struct tag.
+type IncrementalPkDbEntry[PK comparable] struct {
+	IncrementalDbEntry `db:",inline"`
+	ID                 PK `db:"id"`
+}
+
+// GetPrimaryKey returns the id of this entry from the database.
+//
+// It is required by the config.IncrementalConfigurable interface.
+func (i IncrementalPkDbEntry[PK]) GetPrimaryKey() PK {
+	return i.ID
+}

internal/config/channel.go

@@ -3,84 +3,30 @@ package config
 import (
 	"context"
 	"github.com/icinga/icinga-notifications/internal/channel"
-	"github.com/jmoiron/sqlx"
 	"go.uber.org/zap"
 )
 
-func (r *RuntimeConfig) fetchChannels(ctx context.Context, tx *sqlx.Tx) error {
-	var channelPtr *channel.Channel
-	stmt := r.db.BuildSelectStmt(channelPtr, channelPtr)
-	r.logger.Debugf("Executing query %q", stmt)
-
-	var channels []*channel.Channel
-	if err := tx.SelectContext(ctx, &channels, stmt); err != nil {
-		r.logger.Errorln(err)
-		return err
-	}
-
-	channelsById := make(map[int64]*channel.Channel)
-	for _, c := range channels {
-		channelLogger := r.logger.With(
-			zap.Int64("id", c.ID),
-			zap.String("name", c.Name),
-			zap.String("type", c.Type),
-		)
-		if channelsById[c.ID] != nil {
-			channelLogger.Warnw("ignoring duplicate config for channel type")
-		} else if err := channel.ValidateType(c.Type); err != nil {
-			channelLogger.Errorw("Cannot load channel config", zap.Error(err))
-		} else {
-			channelsById[c.ID] = c
-
-			channelLogger.Debugw("loaded channel config")
-		}
-	}
-
-	if r.Channels != nil {
-		// mark no longer existing channels for deletion
-		for id := range r.Channels {
-			if _, ok := channelsById[id]; !ok {
-				channelsById[id] = nil
-			}
-		}
-	}
-
-	r.pending.Channels = channelsById
-
-	return nil
-}
-
+// applyPendingChannels synchronizes changed channels.
 func (r *RuntimeConfig) applyPendingChannels() {
-	if r.Channels == nil {
-		r.Channels = make(map[int64]*channel.Channel)
-	}
-
-	for id, pendingChannel := range r.pending.Channels {
-		if pendingChannel == nil {
-			r.Channels[id].Logger.Info("Channel has been removed")
-			r.Channels[id].Stop()
-
-			delete(r.Channels, id)
-		} else if currentChannel := r.Channels[id]; currentChannel != nil {
-			// Currently, the whole config is reloaded from the database frequently, replacing everything.
-			// Prevent restarting the plugin processes every time by explicitly checking for config changes.
-			// The if condition should no longer be necessary when https://github.com/Icinga/icinga-notifications/issues/5
-			// is solved properly.
-			if currentChannel.Type != pendingChannel.Type || currentChannel.Name != pendingChannel.Name || currentChannel.Config != pendingChannel.Config {
-				currentChannel.Type = pendingChannel.Type
-				currentChannel.Name = pendingChannel.Name
-				currentChannel.Config = pendingChannel.Config
-
-				currentChannel.Restart()
-			}
-		} else {
-			pendingChannel.Start(context.TODO(), r.logs.GetChildLogger("channel").With(
-				zap.Int64("id", pendingChannel.ID),
-				zap.String("name", pendingChannel.Name)))
-
-			r.Channels[id] = pendingChannel
-		}
-	}
-
-	r.pending.Channels = nil
+	incrementalApplyPending(
+		r,
+		&r.Channels, &r.configChange.Channels,
+		func(newElement *channel.Channel) error {
+			newElement.Start(context.TODO(), r.logs.GetChildLogger("channel").With(
+				zap.Int64("id", newElement.ID),
+				zap.String("name", newElement.Name)))
+			return nil
+		},
+		func(curElement, update *channel.Channel) error {
+			curElement.ChangedAt = update.ChangedAt
+			curElement.Name = update.Name
+			curElement.Type = update.Type
+			curElement.Config = update.Config
+			curElement.Restart()
+			return nil
+		},
+		func(delElement *channel.Channel) error {
+			delElement.Stop()
+			return nil
+		})
 }

internal/config/contact.go

@@ -1,62 +1,57 @@
 package config
 
 import (
-	"context"
+	"fmt"
 	"github.com/icinga/icinga-notifications/internal/recipient"
-	"github.com/jmoiron/sqlx"
-	"go.uber.org/zap"
+	"slices"
 )
 
-func (r *RuntimeConfig) fetchContacts(ctx context.Context, tx *sqlx.Tx) error {
-	var contactPtr *recipient.Contact
-	stmt := r.db.BuildSelectStmt(contactPtr, contactPtr)
-	r.logger.Debugf("Executing query %q", stmt)
-
-	var contacts []*recipient.Contact
-	if err := tx.SelectContext(ctx, &contacts, stmt); err != nil {
-		r.logger.Errorln(err)
-		return err
-	}
-
-	contactsByID := make(map[int64]*recipient.Contact)
-	for _, c := range contacts {
-		contactsByID[c.ID] = c
-
-		r.logger.Debugw("loaded contact config",
-			zap.Int64("id", c.ID),
-			zap.String("name", c.FullName))
-	}
-
-	if r.Contacts != nil {
-		// mark no longer existing contacts for deletion
-		for id := range r.Contacts {
-			if _, ok := contactsByID[id]; !ok {
-				contactsByID[id] = nil
+// applyPendingContacts synchronizes changed contacts
+func (r *RuntimeConfig) applyPendingContacts() {
+	incrementalApplyPending(
+		r,
+		&r.Contacts, &r.configChange.Contacts,
+		nil,
+		func(curElement, update *recipient.Contact) error {
+			curElement.ChangedAt = update.ChangedAt
+			curElement.FullName = update.FullName
+			curElement.Username = update.Username
+			curElement.DefaultChannelID = update.DefaultChannelID
+			return nil
+		},
+		nil)
+
+	incrementalApplyPending(
+		r,
+		&r.ContactAddresses, &r.configChange.ContactAddresses,
+		func(newElement *recipient.Address) error {
+			contact, ok := r.Contacts[newElement.ContactID]
+			if !ok {
+				return fmt.Errorf("contact address refers unknown contact %d", newElement.ContactID)
 			}
-		}
-	}
-
-	r.pending.Contacts = contactsByID
 
-	return nil
-}
-
-func (r *RuntimeConfig) applyPendingContacts() {
-	if r.Contacts == nil {
-		r.Contacts = make(map[int64]*recipient.Contact)
-	}
+			contact.Addresses = append(contact.Addresses, newElement)
+			return nil
+		},
+		func(curElement, update *recipient.Address) error {
+			if curElement.ContactID != update.ContactID {
+				return errRemoveAndAddInstead
+			}
 
-	for id, pendingContact := range r.pending.Contacts {
-		if pendingContact == nil {
-			delete(r.Contacts, id)
-		} else if currentContact := r.Contacts[id]; currentContact != nil {
-			currentContact.FullName = pendingContact.FullName
-			currentContact.Username = pendingContact.Username
-			currentContact.DefaultChannelID = pendingContact.DefaultChannelID
-		} else {
-			r.Contacts[id] = pendingContact
-		}
-	}
+			curElement.ChangedAt = update.ChangedAt
+			curElement.Type = update.Type
+			curElement.Address = update.Address
+			return nil
+		},
+		func(delElement *recipient.Address) error {
+			contact, ok := r.Contacts[delElement.ContactID]
+			if !ok {
+				return nil
+			}
 
-	r.pending.Contacts = nil
+			contact.Addresses = slices.DeleteFunc(contact.Addresses, func(address *recipient.Address) bool {
+				return address.ID == delElement.ID
+			})
+			return nil
+		})
 }