-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
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.
- Loading branch information
Showing
24 changed files
with
1,121 additions
and
929 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 | ||
| }) | ||
| } |
Oops, something went wrong.