Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Incremental Configuration Updates #191

Merged
merged 1 commit into from
Jul 12, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 17 additions & 1 deletion internal/channel/channel.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down
45 changes: 45 additions & 0 deletions internal/config/baseconf/incremental_config.go
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
}
98 changes: 22 additions & 76 deletions internal/config/channel.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
})
}
97 changes: 46 additions & 51 deletions internal/config/contact.go
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)
oxzi marked this conversation as resolved.
Show resolved Hide resolved
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
})
}
Loading
Loading