Skip to content

Commit

Permalink
plugin: simplify notifiers queue handling
Browse files Browse the repository at this point in the history 
Signed-off-by: Nicola Murino <nicola.murino@gmail.com>
  • Loading branch information
@drakkan
drakkan committed Dec 25, 2024
1 parent 22f80b9 commit e689d52
Show file tree
Hide file tree
Showing 2 changed files with 97 additions and 147 deletions.
240 changes: 94 additions & 146 deletions internal/plugin/notifier.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -50,97 +50,19 @@ func (c *NotifierConfig) hasActions() bool {
return false
}

type eventsQueue struct {
sync.RWMutex
type notifierPlugin struct {
config Config
notifier notifier.Notifier
client *plugin.Client
mu sync.RWMutex
fsEvents []*notifier.FsEvent
providerEvents []*notifier.ProviderEvent
logEvents []*notifier.LogEvent
}

func (q *eventsQueue) addFsEvent(event *notifier.FsEvent) {
q.Lock()
defer q.Unlock()

q.fsEvents = append(q.fsEvents, event)
}

func (q *eventsQueue) addProviderEvent(event *notifier.ProviderEvent) {
q.Lock()
defer q.Unlock()

q.providerEvents = append(q.providerEvents, event)
}

func (q *eventsQueue) addLogEvent(event *notifier.LogEvent) {
q.Lock()
defer q.Unlock()

q.logEvents = append(q.logEvents, event)
}

func (q *eventsQueue) popFsEvent() *notifier.FsEvent {
q.Lock()
defer q.Unlock()

if len(q.fsEvents) == 0 {
return nil
}
truncLen := len(q.fsEvents) - 1
ev := q.fsEvents[truncLen]
q.fsEvents[truncLen] = nil
q.fsEvents = q.fsEvents[:truncLen]

return ev
}

func (q *eventsQueue) popProviderEvent() *notifier.ProviderEvent {
q.Lock()
defer q.Unlock()

if len(q.providerEvents) == 0 {
return nil
}
truncLen := len(q.providerEvents) - 1
ev := q.providerEvents[truncLen]
q.providerEvents[truncLen] = nil
q.providerEvents = q.providerEvents[:truncLen]

return ev
}

func (q *eventsQueue) popLogEvent() *notifier.LogEvent {
q.Lock()
defer q.Unlock()

if len(q.logEvents) == 0 {
return nil
}
truncLen := len(q.logEvents) - 1
ev := q.logEvents[truncLen]
q.logEvents[truncLen] = nil
q.logEvents = q.logEvents[:truncLen]

return ev
}

func (q *eventsQueue) getSize() int {
q.RLock()
defer q.RUnlock()

return len(q.providerEvents) + len(q.fsEvents) + len(q.logEvents)
}

type notifierPlugin struct {
config Config
notifier notifier.Notifier
client *plugin.Client
queue *eventsQueue
}

func newNotifierPlugin(config Config) (*notifierPlugin, error) {
p := &notifierPlugin{
config: config,
queue: &eventsQueue{},
}
if err := p.initialize(); err != nil {
logger.Warn(logSender, "", "unable to create notifier plugin: %v, config %+v", err, config)
Expand Down Expand Up @@ -180,7 +102,7 @@ func (p *notifierPlugin) initialize() error {
Managed: false,
Logger: &logger.HCLogAdapter{
Logger: hclog.New(&hclog.LoggerOptions{
Name: fmt.Sprintf("%v.%v", logSender, notifier.PluginName),
Name: fmt.Sprintf("%s.%s", logSender, notifier.PluginName),
Level: pluginsLogLevel,
DisableTime: true,
}),
Expand All @@ -204,6 +126,34 @@ func (p *notifierPlugin) initialize() error {
return nil
}

func (p *notifierPlugin) queueSize() int {
p.mu.RLock()
defer p.mu.RUnlock()

return len(p.providerEvents) + len(p.fsEvents) + len(p.logEvents)
}

func (p *notifierPlugin) queueFsEvent(ev *notifier.FsEvent) {
p.mu.Lock()
defer p.mu.Unlock()

p.fsEvents = append(p.fsEvents, ev)
}

func (p *notifierPlugin) queueProviderEvent(ev *notifier.ProviderEvent) {
p.mu.Lock()
defer p.mu.Unlock()

p.providerEvents = append(p.providerEvents, ev)
}

func (p *notifierPlugin) queueLogEvent(ev *notifier.LogEvent) {
p.mu.Lock()
defer p.mu.Unlock()

p.logEvents = append(p.logEvents, ev)
}

func (p *notifierPlugin) canQueueEvent(timestamp int64) bool {
if p.config.NotifierOptions.RetryMaxTime == 0 {
return false
Expand All @@ -214,7 +164,7 @@ func (p *notifierPlugin) canQueueEvent(timestamp int64) bool {
return false
}
if p.config.NotifierOptions.RetryQueueMaxSize > 0 {
return p.queue.getSize() < p.config.NotifierOptions.RetryQueueMaxSize
return p.queueSize() < p.config.NotifierOptions.RetryQueueMaxSize
}
return true
}
Expand All @@ -223,98 +173,96 @@ func (p *notifierPlugin) notifyFsAction(event *notifier.FsEvent) {
if !slices.Contains(p.config.NotifierOptions.FsEvents, event.Action) {
return
}

go func() {
Handler.addTask()
defer Handler.removeTask()

p.sendFsEvent(event)
}()
p.sendFsEvent(event)
}

func (p *notifierPlugin) notifyProviderAction(event *notifier.ProviderEvent, object Renderer) {
if !slices.Contains(p.config.NotifierOptions.ProviderEvents, event.Action) ||
!slices.Contains(p.config.NotifierOptions.ProviderObjects, event.ObjectType) {
return
}
p.sendProviderEvent(event, object)
}

func (p *notifierPlugin) notifyLogEvent(event *notifier.LogEvent) {
p.sendLogEvent(event)
}

go func() {
func (p *notifierPlugin) sendFsEvent(ev *notifier.FsEvent) {
go func(event *notifier.FsEvent) {
Handler.addTask()
defer Handler.removeTask()

objectAsJSON, err := object.RenderAsJSON(event.Action != "delete")
if err != nil {
logger.Warn(logSender, "", "unable to render user as json for action %v: %v", event.Action, err)
return
if err := p.notifier.NotifyFsEvent(event); err != nil {
logger.Warn(logSender, "", "unable to send fs action notification to plugin %v: %v", p.config.Cmd, err)
if p.canQueueEvent(event.Timestamp) {
p.queueFsEvent(event)
}
}
event.ObjectData = objectAsJSON
p.sendProviderEvent(event)
}()
}(ev)
}

func (p *notifierPlugin) notifyLogEvent(event *notifier.LogEvent) {
go func() {
func (p *notifierPlugin) sendProviderEvent(ev *notifier.ProviderEvent, object Renderer) {
go func(event *notifier.ProviderEvent) {
Handler.addTask()
defer Handler.removeTask()

p.sendLogEvent(event)
}()
}

func (p *notifierPlugin) sendFsEvent(event *notifier.FsEvent) {
if err := p.notifier.NotifyFsEvent(event); err != nil {
logger.Warn(logSender, "", "unable to send fs action notification to plugin %v: %v", p.config.Cmd, err)
if p.canQueueEvent(event.Timestamp) {
p.queue.addFsEvent(event)
if object != nil {
objectAsJSON, err := object.RenderAsJSON(event.Action != "delete")
if err != nil {
logger.Error(logSender, "", "unable to render user as json for action %q: %v", event.Action, err)
} else {
event.ObjectData = objectAsJSON
}
}
}
}

func (p *notifierPlugin) sendProviderEvent(event *notifier.ProviderEvent) {
if err := p.notifier.NotifyProviderEvent(event); err != nil {
logger.Warn(logSender, "", "unable to send user action notification to plugin %v: %v", p.config.Cmd, err)
if p.canQueueEvent(event.Timestamp) {
p.queue.addProviderEvent(event)
if err := p.notifier.NotifyProviderEvent(event); err != nil {
logger.Warn(logSender, "", "unable to send user action notification to plugin %v: %v", p.config.Cmd, err)
if p.canQueueEvent(event.Timestamp) {
p.queueProviderEvent(event)
}
}
}
}(ev)
}

func (p *notifierPlugin) sendLogEvent(event *notifier.LogEvent) {
if err := p.notifier.NotifyLogEvent(event); err != nil {
logger.Warn(logSender, "", "unable to send log event to plugin %v: %v", p.config.Cmd, err)
if p.canQueueEvent(event.Timestamp) {
p.queue.addLogEvent(event)
func (p *notifierPlugin) sendLogEvent(ev *notifier.LogEvent) {
go func(event *notifier.LogEvent) {
Handler.addTask()
defer Handler.removeTask()

if err := p.notifier.NotifyLogEvent(event); err != nil {
logger.Warn(logSender, "", "unable to send log event to plugin %v: %v", p.config.Cmd, err)
if p.canQueueEvent(event.Timestamp) {
p.queueLogEvent(event)
}
}
}
}(ev)
}

func (p *notifierPlugin) sendQueuedEvents() {
queueSize := p.queue.getSize()
queueSize := p.queueSize()
if queueSize == 0 {
return
}
logger.Debug(logSender, "", "check queued events for notifier %q, events size: %v", p.config.Cmd, queueSize)
fsEv := p.queue.popFsEvent()
for fsEv != nil {
go func(ev *notifier.FsEvent) {
p.sendFsEvent(ev)
}(fsEv)
fsEv = p.queue.popFsEvent()
p.mu.Lock()
defer p.mu.Unlock()

logger.Debug(logSender, "", "send queued events for notifier %q, events size: %v", p.config.Cmd, queueSize)

for _, ev := range p.fsEvents {
p.sendFsEvent(ev)
}
p.fsEvents = nil

providerEv := p.queue.popProviderEvent()
for providerEv != nil {
go func(ev *notifier.ProviderEvent) {
p.sendProviderEvent(ev)
}(providerEv)
providerEv = p.queue.popProviderEvent()
for _, ev := range p.providerEvents {
p.sendProviderEvent(ev, nil)
}
logEv := p.queue.popLogEvent()
for logEv != nil {
go func(ev *notifier.LogEvent) {
p.sendLogEvent(ev)
}(logEv)
logEv = p.queue.popLogEvent()
p.providerEvents = nil

for _, ev := range p.logEvents {
p.sendLogEvent(ev)
}
logger.Debug(logSender, "", "queued events sent for notifier %q, new events size: %v", p.config.Cmd, p.queue.getSize())
p.logEvents = nil

logger.Debug(logSender, "", "%d queued events sent for notifier %q,", queueSize, p.config.Cmd)
}
4 changes: 3 additions & 1 deletion internal/plugin/plugin.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -642,7 +642,9 @@ func (m *Manager) restartNotifierPlugin(config Config, idx int) {
}

m.notifLock.Lock()
plugin.queue = m.notifiers[idx].queue
plugin.fsEvents = m.notifiers[idx].fsEvents
plugin.providerEvents = m.notifiers[idx].providerEvents
plugin.logEvents = m.notifiers[idx].logEvents
m.notifiers[idx] = plugin
m.notifLock.Unlock()
plugin.sendQueuedEvents()
Expand Down

0 comments on commit e689d52

Please sign in to comment.