mirror of
https://github.com/resmoio/kubernetes-event-exporter.git
synced 2026-08-24 01:06:22 +00:00
Marc Demierre
106e78ae0e
Before the change, the engine didn't call the `Close()` method of the sinks. This is needed in some cases, i.e when a sink implementation is buffered. This change adds a `Close()` method to the registry that will signal sinks to exit and wait for all sinks to exit before returning. This is then used in the engine stop logic. In the channel-based registry, the closing of all sinks is done in parallel (using a `sync.WaitGroup`). In the sync registry, sinks are closed sequentially. Fixes issue #10
37 lines
942 B
Go
37 lines
942 B
Go
package exporter
|
|
|
|
import (
|
|
"context"
|
|
"github.com/opsgenie/kubernetes-event-exporter/pkg/kube"
|
|
"github.com/opsgenie/kubernetes-event-exporter/pkg/sinks"
|
|
"github.com/rs/zerolog/log"
|
|
)
|
|
|
|
// SyncRegistry is for development purposes and performs poorly and blocks when an event is received so it is
|
|
// not suited for high volume & production workloads
|
|
type SyncRegistry struct {
|
|
reg map[string]sinks.Sink
|
|
}
|
|
|
|
func (s *SyncRegistry) SendEvent(name string, event *kube.EnhancedEvent) {
|
|
err := s.reg[name].Send(context.Background(), event)
|
|
if err != nil {
|
|
log.Debug().Err(err).Str("sink", name).Str("event", string(event.UID)).Msg("Cannot send event")
|
|
}
|
|
}
|
|
|
|
func (s *SyncRegistry) Register(name string, sink sinks.Sink) {
|
|
if s.reg == nil {
|
|
s.reg = make(map[string]sinks.Sink)
|
|
}
|
|
|
|
s.reg[name] = sink
|
|
}
|
|
|
|
func (s *SyncRegistry) Close() {
|
|
for name, sink := range s.reg {
|
|
log.Info().Str("sink", name).Msg("Closing sink")
|
|
sink.Close()
|
|
}
|
|
}
|