This is an automated email from the ASF dual-hosted git repository.
wilfred-s pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/yunikorn-core.git
The following commit(s) were added to refs/heads/master by this push:
new a86f91db [YUNIKORN-3336] event system shutdown leaks go routine or
panics (#1111)
a86f91db is described below
commit a86f91db7e5d4aabd0428a56f20c7567b28180af
Author: Wilfred Spiegelenburg <[email protected]>
AuthorDate: Wed Jul 29 11:29:23 2026 +1000
[YUNIKORN-3336] event system shutdown leaks go routine or panics (#1111)
Calling StartService for an existing stopped EventPublisher does not
create a new stop channel. Event publisher will ignore start request
when already running. Calling stop twice on the EventPublisher panics as
it tries to close an already closed channel.
Move config callback setup for event system into the Init() allowing a
start from config even when system start did not start the event system.
Move stop channel creation into start and stop functions
Cleanup layout in event system and publisher
Remove exports from functions and objects where possible.
Fix event loss after restart and race in atomic stop flag
Change stop flag check to `CompareAndSwap` removing the small race that
could occur using `Load` & `Save`.
Remove reset of the callback as the callback never changes and is needed
for turning on event system if turned off via config.
Additional metrics for dropped events (created but not channelled)
Not channelled events are events generated when the event system is
turned off. Dropped events are created when the channel is not open
or full while the event system is running.
created == channelled + not channelled + dropped
Fixed new data race seen during config reload. Move configuration
settings to Init() instead of start for consistent processing.
Closes: #1111
Signed-off-by: Wilfred Spiegelenburg <[email protected]>
---
pkg/events/event_publisher.go | 42 ++++--
pkg/events/event_publisher_test.go | 36 +++--
pkg/events/event_system.go | 284 ++++++++++++++++++++-----------------
pkg/events/event_system_test.go | 75 ++++++++--
pkg/metrics/event.go | 88 +++++++++++-
5 files changed, 365 insertions(+), 160 deletions(-)
diff --git a/pkg/events/event_publisher.go b/pkg/events/event_publisher.go
index a1d95f6a..41e90690 100644
--- a/pkg/events/event_publisher.go
+++ b/pkg/events/event_publisher.go
@@ -19,6 +19,7 @@
package events
import (
+ "sync/atomic"
"time"
"go.uber.org/zap"
@@ -30,33 +31,42 @@ import (
// stores the push event internal
var defaultPushEventInterval = 2 * time.Second
-type EventPublisher struct {
+type eventPublisher struct {
store *EventStore
pushEventInterval time.Duration
- stop chan struct{}
+ stopCh chan struct{}
+ stopped atomic.Bool
}
-func CreateShimPublisher(store *EventStore) *EventPublisher {
- publisher := &EventPublisher{
+func createShimPublisher(store *EventStore) *eventPublisher {
+ publisher := &eventPublisher{
store: store,
pushEventInterval: defaultPushEventInterval,
- stop: make(chan struct{}),
}
+ publisher.stopped.Store(true)
return publisher
}
-func (sp *EventPublisher) StartService() {
- log.Log(log.Events).Info("Starting shim event publisher")
+func (sp *eventPublisher) start() {
+ log.Log(log.Events).Info("Starting event publisher")
+ // handle a restart correctly
+ if !sp.stopped.CompareAndSwap(true, false) {
+ log.Log(log.Events).Info("Event publisher already running")
+ return
+ }
+ sp.stopCh = make(chan struct{})
go func() {
for {
select {
- case <-sp.stop:
+ case <-sp.stopCh:
+ log.Log(log.Events).Info("Event publisher
exiting")
return
case <-time.After(sp.pushEventInterval):
messages := sp.store.CollectEvents()
if len(messages) > 0 {
if eventPlugin :=
plugins.GetResourceManagerCallbackPlugin(); eventPlugin != nil {
-
log.Log(log.Events).Debug("Sending eventChannel", zap.Int("number of messages",
len(messages)))
+
log.Log(log.Events).Debug("Sending eventChannel",
+ zap.Int("number of
messages", len(messages)))
eventPlugin.SendEvent(messages)
}
}
@@ -65,11 +75,17 @@ func (sp *EventPublisher) StartService() {
}()
}
-func (sp *EventPublisher) Stop() {
- log.Log(log.Events).Info("Stopping shim event publisher")
- close(sp.stop)
+func (sp *eventPublisher) stop() {
+ if !sp.stopped.CompareAndSwap(false, true) {
+ log.Log(log.Events).Info("Event publisher already stopped")
+ return
+ }
+ log.Log(log.Events).Info("Stopping event publisher")
+ sp.stopCh <- struct{}{}
+ close(sp.stopCh)
+ sp.stopCh = nil
}
-func (sp *EventPublisher) getEventStore() *EventStore {
+func (sp *eventPublisher) getEventStore() *EventStore {
return sp.store
}
diff --git a/pkg/events/event_publisher_test.go
b/pkg/events/event_publisher_test.go
index 78680b59..722c18f4 100644
--- a/pkg/events/event_publisher_test.go
+++ b/pkg/events/event_publisher_test.go
@@ -19,6 +19,7 @@
package events
import (
+ "runtime"
"testing"
"time"
@@ -32,25 +33,38 @@ import (
// creating a Publisher with nil store should still provide a non-nil object
func TestCreateShimPublisher(t *testing.T) {
- publisher := CreateShimPublisher(nil)
+ publisher := createShimPublisher(nil)
assert.Assert(t, publisher != nil, "publisher should not be nil")
}
-// StartService() and Stop() functions should not cause panic
+// StartService() and stop() functions should not cause panic
func TestServiceStartStopInternal(t *testing.T) {
store := newEventStore(1000)
- publisher := CreateShimPublisher(store)
- publisher.StartService()
- defer publisher.Stop()
+ publisher := createShimPublisher(store)
+ defer publisher.stop()
assert.Equal(t, publisher.getEventStore(), store)
+ // start and stop simulate a restart
+ before := runtime.NumGoroutine()
+ publisher.start()
+ publisher.stop()
+ time.Sleep(10 * time.Millisecond) // tiny sleep to yield
+ assert.Equal(t, before, runtime.NumGoroutine(), "expected no new go
routine after start and stop")
+
+ // start should not fail or panic
+ before = runtime.NumGoroutine()
+ publisher.start()
+ after := runtime.NumGoroutine()
+ assert.Equal(t, before+1, after, "expected 1 new go routine")
+ publisher.start()
+ assert.Equal(t, after, runtime.NumGoroutine(), "Already started should
not create new go routine")
}
func TestNoFillWithoutEventPluginRegistered(t *testing.T) {
store := newEventStore(1000)
- publisher := CreateShimPublisher(store)
+ publisher := createShimPublisher(store)
publisher.pushEventInterval = time.Millisecond
- publisher.StartService()
- defer publisher.Stop()
+ publisher.start()
+ defer publisher.stop()
event := &si.EventRecord{
Type: si.EventRecord_REQUEST,
@@ -80,10 +94,10 @@ func TestPublisherSendsEvent(t *testing.T) {
}
store := newEventStore(1000)
- publisher := CreateShimPublisher(store)
+ publisher := createShimPublisher(store)
publisher.pushEventInterval = time.Millisecond
- publisher.StartService()
- defer publisher.Stop()
+ publisher.start()
+ defer publisher.stop()
event := &si.EventRecord{
Type: si.EventRecord_REQUEST,
diff --git a/pkg/events/event_system.go b/pkg/events/event_system.go
index 117669e1..631db365 100644
--- a/pkg/events/event_system.go
+++ b/pkg/events/event_system.go
@@ -21,6 +21,7 @@ package events
import (
"fmt"
"sync"
+ "sync/atomic"
"time"
"go.uber.org/zap"
@@ -33,8 +34,10 @@ import (
"github.com/apache/yunikorn-scheduler-interface/lib/go/si"
)
-var once sync.Once
-var ev EventSystem
+var (
+ once sync.Once
+ ev EventSystem
+)
type EventSystem interface {
// AddEvent adds an event record to the event system for processing:
@@ -77,17 +80,54 @@ type EventSystem interface {
GetEventStreams() []EventStreamData
}
+// GetEventSystem returns the event system instance. Initialization happens
during the first call.
+// This does not start the service or publisher. Call StartService or
StartServiceWithPublisher
+// on the returned system to start the service.
+func GetEventSystem() EventSystem {
+ once.Do(func() {
+ Init()
+ })
+ return ev
+}
+
+// Init Initializes the event system.
+// Only exported for testing.
+func Init() {
+ // load from config for setting in two places
+ confRequestCapacity := getRequestCapacity()
+ confRingBufferCapacity := getRingBufferCapacity()
+
+ store := newEventStore(confRequestCapacity)
+ buffer := newEventRingBuffer(confRingBufferCapacity)
+ evImpl := &EventSystemImpl{
+ Store: store,
+ eventBuffer: buffer,
+ eventSystemId: fmt.Sprintf("event-system-%d",
time.Now().Unix()),
+ streaming: NewEventStreaming(buffer),
+ trackingEnabled: isTrackingEnabled(),
+ requestCapacity: confRequestCapacity,
+ ringBufferCapacity: confRingBufferCapacity,
+ }
+ evImpl.stopped.Store(true)
+ // start the callback for this instance: must be done
+ configs.AddConfigMapCallback(evImpl.eventSystemId, func() {
+ go evImpl.reloadConfig()
+ })
+
+ ev = evImpl
+}
+
// EventSystemImpl main implementation of the event system which is used for
history tracking.
type EventSystemImpl struct {
eventSystemId string
- Store *EventStore // storing eventChannel
- publisher *EventPublisher
+ Store *EventStore // storing eventChannel, exported for test
+ publisher *eventPublisher
eventBuffer *eventRingBuffer
streaming *EventStreaming
channel chan *si.EventRecord // channelling input eventChannel
- stop chan bool // whether the service is stopped
- stopped bool
+ stop chan struct{} // channel to stop the system
+ stopped atomic.Bool // whether the service is stopped
trackingEnabled bool
requestCapacity uint64
@@ -111,14 +151,6 @@ func (ec *EventSystemImpl) GetEventsFromID(id, count
uint64) ([]*si.EventRecord,
return ec.eventBuffer.GetEventsFromID(id, count)
}
-// GetEventSystem returns the event system instance. Initialization happens
during the first call.
-func GetEventSystem() EventSystem {
- once.Do(func() {
- Init()
- })
- return ev
-}
-
// IsEventTrackingEnabled whether history tracking is currently enabled or not.
func (ec *EventSystemImpl) IsEventTrackingEnabled() bool {
ec.RLock()
@@ -126,40 +158,61 @@ func (ec *EventSystemImpl) IsEventTrackingEnabled() bool {
return ec.trackingEnabled
}
-// GetRequestCapacity returns the capacity of an intermediate storage which is
used by the shim publisher.
-func (ec *EventSystemImpl) GetRequestCapacity() uint64 {
- ec.RLock()
- defer ec.RUnlock()
- return ec.requestCapacity
+// StartService starts the event processing in the background. See the
interface for details.
+func (ec *EventSystemImpl) StartService() {
+ ec.StartServiceWithPublisher(true)
}
-// GetRingBufferCapacity returns the capacity of the buffer which stores
historical elements.
-func (ec *EventSystemImpl) GetRingBufferCapacity() uint64 {
- ec.RLock()
- defer ec.RUnlock()
- return ec.ringBufferCapacity
-}
+// Stop stops the event system, including the shim publisher if it was started.
+func (ec *EventSystemImpl) Stop() {
+ ec.Lock()
+ defer ec.Unlock()
+ // no need to stop twice
+ if !ec.stopped.CompareAndSwap(false, true) {
+ return
+ }
+ log.Log(log.Events).Info("Stopping event system handler")
-// Init Initializes the event system.
-// Only exported for testing.
-func Init() {
- store := newEventStore(getRequestCapacity())
- buffer := newEventRingBuffer(getRingBufferCapacity())
- ev = &EventSystemImpl{
- Store: store,
- channel: make(chan *si.EventRecord,
configs.DefaultEventChannelSize),
- stop: make(chan bool),
- stopped: false,
- eventBuffer: buffer,
- eventSystemId: fmt.Sprintf("event-system-%d",
time.Now().Unix()),
- publisher: CreateShimPublisher(store),
- streaming: NewEventStreaming(buffer),
+ ec.stop <- struct{}{}
+ if ec.channel != nil {
+ close(ec.channel)
+ ec.channel = nil
+ }
+ if ec.publisher != nil {
+ ec.publisher.stop()
+ ec.publisher = nil
}
}
-// StartService starts the event processing in the background. See the
interface for details.
-func (ec *EventSystemImpl) StartService() {
- ec.StartServiceWithPublisher(true)
+// GetEventStreams returns the current active event streams.
+func (ec *EventSystemImpl) GetEventStreams() []EventStreamData {
+ return ec.streaming.GetEventStreams()
+}
+
+// AddEvent adds an event record to the event system. See the interface for
details.
+func (ec *EventSystemImpl) AddEvent(event *si.EventRecord) {
+ if event != nil {
+ event.Message = truncateEventMessage(event.Message)
+ }
+ // all events get tracked, even ones we drop
+ metrics.GetEventMetrics().IncEventsCreated()
+ ec.RLock()
+ defer ec.RUnlock()
+ // not running just track the metric
+ if ec.stopped.Load() {
+ metrics.GetEventMetrics().IncEventsNotChanneled()
+ return
+ }
+
+ select {
+ case ec.channel <- event:
+ metrics.GetEventMetrics().IncEventsChanneled()
+ default:
+ // make sure we do not drop events when running. events
generated while turned off are not "dropped"
+ log.Log(log.Events).Info("Event dropped due to channel full or
closed",
+ zap.Int("channelSize", len(ec.channel)))
+ metrics.GetEventMetrics().IncEventsDropped()
+ }
}
// StartServiceWithPublisher starts the event processing background routines.
@@ -167,14 +220,13 @@ func (ec *EventSystemImpl) StartService() {
func (ec *EventSystemImpl) StartServiceWithPublisher(withPublisher bool) {
ec.Lock()
defer ec.Unlock()
-
- configs.AddConfigMapCallback(ec.eventSystemId, func() {
- go ec.reloadConfig()
- })
-
+ if !ec.stopped.CompareAndSwap(true, false) {
+ log.Log(log.Events).Info("Event system is already running")
+ return
+ }
ec.trackingEnabled = isTrackingEnabled()
- ec.ringBufferCapacity = getRingBufferCapacity()
- ec.requestCapacity = getRequestCapacity()
+ ec.stop = make(chan struct{})
+ ec.channel = make(chan *si.EventRecord, configs.DefaultEventChannelSize)
go func() {
log.Log(log.Events).Info("Starting event system handler")
@@ -196,60 +248,74 @@ func (ec *EventSystemImpl)
StartServiceWithPublisher(withPublisher bool) {
}
}()
if withPublisher {
- ec.publisher.StartService()
+ ec.publisher = createShimPublisher(ec.Store)
+ ec.publisher.start()
}
}
-// Stop stops the event system.
-func (ec *EventSystemImpl) Stop() {
- ec.Lock()
- defer ec.Unlock()
-
- configs.RemoveConfigMapCallback(ec.eventSystemId)
+// getRequestCapacity returns the capacity of an intermediate storage which is
used by the shim publisher.
+func (ec *EventSystemImpl) getRequestCapacity() uint64 {
+ ec.RLock()
+ defer ec.RUnlock()
+ return ec.requestCapacity
+}
- if ec.stopped {
- return
- }
+// getRingBufferCapacity returns the capacity of the buffer which stores
historical elements.
+func (ec *EventSystemImpl) getRingBufferCapacity() uint64 {
+ ec.RLock()
+ defer ec.RUnlock()
+ return ec.ringBufferCapacity
+}
- ec.stop <- true
- if ec.channel != nil {
- close(ec.channel)
- ec.channel = nil
- }
- ec.publisher.Stop()
- ec.stopped = true
+// isRestartNeeded returns true if the tracking mode has switched from off to
on or vice versa.
+// Returns false if tracking mode has not changed
+func (ec *EventSystemImpl) isRestartNeeded() bool {
+ ec.RLock()
+ defer ec.RUnlock()
+ return isTrackingEnabled() != ec.trackingEnabled
}
-func (ec *EventSystemImpl) isStopped() bool {
- return ec.stopped || ec.channel == nil
+// restart restarts the event system, used during config update.
+func (ec *EventSystemImpl) restart() {
+ ec.Stop()
+ ec.StartServiceWithPublisher(true)
}
-// AddEvent adds an event record to the event system. See the interface for
details.
-func (ec *EventSystemImpl) AddEvent(event *si.EventRecord) {
- if event != nil {
- event.Message = truncateEventMessage(event.Message)
+// CloseAllStreams closes all existing streams.
+// VisibleForTesting
+func (ec *EventSystemImpl) CloseAllStreams() {
+ ec.streaming.Lock()
+ defer ec.streaming.Unlock()
+ for consumer := range ec.streaming.eventStreams {
+ ec.streaming.removeEventStream(consumer)
}
+}
- metrics.GetEventMetrics().IncEventsCreated()
+// reloadConfig function called by the config
+func (ec *EventSystemImpl) reloadConfig() {
+ // load from config for setting in two places
+ confRequestCapacity := getRequestCapacity()
+ confRingBufferCapacity := getRingBufferCapacity()
- ec.RLock()
- defer ec.RUnlock()
+ ec.Lock()
+ ec.requestCapacity = confRequestCapacity
+ ec.ringBufferCapacity = confRingBufferCapacity
+ ec.Unlock()
- if ec.isStopped() {
- metrics.GetEventMetrics().IncEventsNotChanneled()
- return
- }
+ // resize the ring buffer & event store with new capacity
+ ec.Store.SetStoreSize(confRequestCapacity)
+ ec.eventBuffer.Resize(confRingBufferCapacity)
- select {
- case ec.channel <- event:
- metrics.GetEventMetrics().IncEventsChanneled()
- default:
- log.Log(log.Events).Debug("could not add Event to channel")
- metrics.GetEventMetrics().IncEventsNotChanneled()
+ if ec.isRestartNeeded() {
+ log.Log(log.Events).Info("Restarting event system handler on
config reload")
+ ec.Lock()
+ ec.trackingEnabled = isTrackingEnabled()
+ ec.Unlock()
+ ec.restart()
}
}
-// truncates event message to 1024 characters for k8s compatibility
+// truncateEventMessage limits the event message to 1024 characters for k8s
compatibility
func truncateEventMessage(message string) string {
const k8sEventMessageLimit = 1024
if len(message) <= k8sEventMessageLimit {
@@ -258,10 +324,13 @@ func truncateEventMessage(message string) string {
return message[:k8sEventMessageLimit-3] + "..."
}
+// isTrackingEnabled gets the current state of tracking from the configuration.
func isTrackingEnabled() bool {
return common.GetConfigurationBool(configs.GetConfigMap(),
configs.CMEventTrackingEnabled, configs.DefaultEventTrackingEnabled)
}
+// getRequestCapacity returns the size of an intermediate storage from the
configuration, using the
+// configs.DefaultEventRequestCapacity if 0 or not defined.
func getRequestCapacity() uint64 {
capacity := common.GetConfigurationUint(configs.GetConfigMap(),
configs.CMEventRequestCapacity, configs.DefaultEventRequestCapacity)
if capacity == 0 {
@@ -273,6 +342,8 @@ func getRequestCapacity() uint64 {
return capacity
}
+// getRingBufferCapacity returns the ring buffer capacity from the
configuration, using the
+// configs.DefaultEventRingBufferCapacity if 0 or not defined.
func getRingBufferCapacity() uint64 {
capacity := common.GetConfigurationUint(configs.GetConfigMap(),
configs.CMEventRingBufferCapacity, configs.DefaultEventRingBufferCapacity)
if capacity == 0 {
@@ -283,44 +354,3 @@ func getRingBufferCapacity() uint64 {
}
return capacity
}
-
-func (ec *EventSystemImpl) isRestartNeeded() bool {
- ec.RLock()
- defer ec.RUnlock()
- return isTrackingEnabled() != ec.trackingEnabled
-}
-
-// Restart restarts the event system, used during config update.
-func (ec *EventSystemImpl) Restart() {
- ec.Stop()
- ec.StartServiceWithPublisher(true)
-}
-
-// GetEventStreams returns the current active event streams.
-func (ec *EventSystemImpl) GetEventStreams() []EventStreamData {
- return ec.streaming.GetEventStreams()
-}
-
-// VisibleForTesting
-func (ec *EventSystemImpl) CloseAllStreams() {
- ec.streaming.Lock()
- defer ec.streaming.Unlock()
- for consumer := range ec.streaming.eventStreams {
- ec.streaming.removeEventStream(consumer)
- }
-}
-
-func (ec *EventSystemImpl) reloadConfig() {
- ec.Lock()
- ec.requestCapacity = getRequestCapacity()
- ec.ringBufferCapacity = getRingBufferCapacity()
- ec.Unlock()
-
- // resize the ring buffer & event store with new capacity
- ec.Store.SetStoreSize(ec.requestCapacity)
- ec.eventBuffer.Resize(ec.ringBufferCapacity)
-
- if ec.isRestartNeeded() {
- ec.Restart()
- }
-}
diff --git a/pkg/events/event_system_test.go b/pkg/events/event_system_test.go
index 18b24ae4..ff3d1320 100644
--- a/pkg/events/event_system_test.go
+++ b/pkg/events/event_system_test.go
@@ -19,6 +19,7 @@
package events
import (
+ "runtime"
"strconv"
"strings"
"sync"
@@ -29,6 +30,7 @@ import (
"github.com/apache/yunikorn-core/pkg/common"
"github.com/apache/yunikorn-core/pkg/common/configs"
+ "github.com/apache/yunikorn-core/pkg/metrics"
"github.com/apache/yunikorn-scheduler-interface/lib/go/si"
)
@@ -38,11 +40,18 @@ func TestSimpleStartAndStop(t *testing.T) {
eventSystem := GetEventSystem()
// adding event to stopped eventSystem does not cause panic
eventSystem.AddEvent(nil)
+ time.Sleep(10 * time.Millisecond) // tiny sleep to yield
+ before := runtime.NumGoroutine()
eventSystem.StartService()
defer eventSystem.Stop()
+ after := runtime.NumGoroutine()
+ assert.Equal(t, before+2, after, "expected 2 new go routines: handler
and publisher")
// add an event
eventSystem.AddEvent(nil)
eventSystem.Stop()
+ time.Sleep(10 * time.Millisecond) // tiny sleep to yield
+ after = runtime.NumGoroutine()
+ assert.Equal(t, before, after, "expected all go routines to exit")
// adding event to stopped eventSystem does not cause panic
eventSystem.AddEvent(nil)
}
@@ -51,7 +60,8 @@ func TestSimpleStartAndStop(t *testing.T) {
// should be retrieved from the EventStore
func TestSingleEventStoredCorrectly(t *testing.T) {
Init()
- eventSystem := GetEventSystem().(*EventSystemImpl) //nolint:errcheck
+ eventSystem, ok := GetEventSystem().(*EventSystemImpl)
+ assert.Assert(t, ok, "expected an EventSystemImpl")
// don't run publisher, because it can collect the event while we're
waiting
eventSystem.StartServiceWithPublisher(false)
defer eventSystem.Stop()
@@ -84,7 +94,8 @@ func TestSingleEventStoredCorrectly(t *testing.T) {
func TestGetEvents(t *testing.T) {
Init()
- eventSystem := GetEventSystem().(*EventSystemImpl) //nolint:errcheck
+ eventSystem, ok := GetEventSystem().(*EventSystemImpl)
+ assert.Assert(t, ok, "expected an EventSystemImpl")
eventSystem.StartServiceWithPublisher(false)
defer eventSystem.Stop()
@@ -116,13 +127,14 @@ func TestConfigUpdate(t *testing.T) {
defer configs.SetConfigMap(map[string]string{})
Init()
- eventSystem := GetEventSystem().(*EventSystemImpl) //nolint:errcheck
+ eventSystem, ok := GetEventSystem().(*EventSystemImpl)
+ assert.Assert(t, ok, "expected an EventSystemImpl")
eventSystem.StartService()
defer eventSystem.Stop()
assert.Assert(t, eventSystem.IsEventTrackingEnabled())
- assert.Equal(t, eventSystem.GetRingBufferCapacity(),
uint64(configs.DefaultEventRingBufferCapacity))
- assert.Equal(t, eventSystem.GetRequestCapacity(),
uint64(configs.DefaultEventRequestCapacity))
+ assert.Equal(t, eventSystem.getRingBufferCapacity(),
uint64(configs.DefaultEventRingBufferCapacity))
+ assert.Equal(t, eventSystem.getRequestCapacity(),
uint64(configs.DefaultEventRequestCapacity))
assert.Equal(t, eventSystem.eventBuffer.capacity,
uint64(configs.DefaultEventRingBufferCapacity))
// update config and wait for refresh
@@ -142,8 +154,8 @@ func TestConfigUpdate(t *testing.T) {
)
assert.NilError(t, err, "timed out waiting for config refresh")
- assert.Equal(t, eventSystem.GetRingBufferCapacity(),
newRingBufferCapacity)
- assert.Equal(t, eventSystem.GetRequestCapacity(), newRequestCapacity)
+ assert.Equal(t, eventSystem.getRingBufferCapacity(),
newRingBufferCapacity)
+ assert.Equal(t, eventSystem.getRequestCapacity(), newRequestCapacity)
assert.Equal(t, eventSystem.eventBuffer.capacity, newRingBufferCapacity)
}
@@ -182,6 +194,9 @@ func TestRequestCapacity(t *testing.T) {
configs.SetConfigMap(config)
capacity = getRequestCapacity()
assert.Equal(t, uint64(configs.DefaultEventRequestCapacity), capacity)
+
+ // reset to empty
+ configs.SetConfigMap(map[string]string{})
}
func TestRingBufferCapacity(t *testing.T) {
@@ -206,6 +221,9 @@ func TestRingBufferCapacity(t *testing.T) {
configs.SetConfigMap(config)
capacity = getRingBufferCapacity()
assert.Equal(t, uint64(configs.DefaultEventRingBufferCapacity),
capacity)
+
+ // reset to empty
+ configs.SetConfigMap(map[string]string{})
}
func TestTruncateEventMessage(t *testing.T) {
@@ -246,8 +264,10 @@ func getTestString(stringLength int) string {
// TestAddEventConcurrentStop verifies AddEvent and Stop can run concurrently
without data races.
func TestAddEventConcurrentStop(t *testing.T) {
Init()
- eventSystem := GetEventSystem().(*EventSystemImpl) //nolint:errcheck
+ eventSystem, ok := GetEventSystem().(*EventSystemImpl)
+ assert.Assert(t, ok, "expected an EventSystemImpl")
eventSystem.StartServiceWithPublisher(false)
+ defer eventSystem.Stop()
var wg sync.WaitGroup
wg.Add(2)
@@ -270,3 +290,42 @@ func TestAddEventConcurrentStop(t *testing.T) {
wg.Wait()
}
+
+// TestAddEventConcurrentStop verifies AddEvent and Stop can run concurrently
without data races.
+func TestAddEventAfterRestart(t *testing.T) {
+ Init()
+ eventSystem, ok := GetEventSystem().(*EventSystemImpl)
+ assert.Assert(t, ok, "expected an EventSystemImpl")
+ metrics.GetEventMetrics().Reset()
+ eventSystem.StartServiceWithPublisher(false)
+ defer eventSystem.Stop()
+
+ var wg sync.WaitGroup
+ wg.Add(2)
+
+ eventCount := 10000
+ go func() {
+ defer wg.Done()
+ for i := range eventCount {
+ eventSystem.AddEvent(&si.EventRecord{
+ Type: si.EventRecord_REQUEST,
+ Message: strconv.Itoa(i),
+ })
+ }
+ }()
+
+ go func() {
+ defer wg.Done()
+ time.Sleep(time.Millisecond)
+ eventSystem.restart()
+ }()
+
+ wg.Wait()
+ counted := metrics.GetEventMetrics().GetEventsDropped()
+ assert.Equal(t, counted, 0, "should not have dropped an event")
+ counted = metrics.GetEventMetrics().GetEventsCreated()
+ assert.Equal(t, counted, eventCount, "number of created events
incorrect")
+ counted = metrics.GetEventMetrics().GetEventsChanneled()
+ notCounted := metrics.GetEventMetrics().GetEventsNotChanneled()
+ assert.Equal(t, counted+notCounted, eventCount, "total number of
(not)channeled events incorrect")
+}
diff --git a/pkg/metrics/event.go b/pkg/metrics/event.go
index d54a007d..281b2937 100644
--- a/pkg/metrics/event.go
+++ b/pkg/metrics/event.go
@@ -18,12 +18,16 @@
package metrics
-import "github.com/prometheus/client_golang/prometheus"
+import (
+ "github.com/prometheus/client_golang/prometheus"
+ dto "github.com/prometheus/client_model/go"
+)
type EventMetrics struct {
totalEventsCreated prometheus.Gauge
totalEventsChanneled prometheus.Gauge
totalEventsNotChanneled prometheus.Gauge
+ totalEventsDropped prometheus.Gauge
totalEventsProcessed prometheus.Gauge
totalEventsStored prometheus.Gauge
totalEventsNotStored prometheus.Gauge
@@ -54,6 +58,13 @@ func initEventMetrics() *EventMetrics {
Name: "total_not_channeled",
Help: "total events not channeled",
})
+ metrics.totalEventsDropped = prometheus.NewGauge(
+ prometheus.GaugeOpts{
+ Namespace: Namespace,
+ Subsystem: EventSubsystem,
+ Name: "total_dropped",
+ Help: "total events dropped",
+ })
metrics.totalEventsProcessed = prometheus.NewGauge(
prometheus.GaugeOpts{
Namespace: Namespace,
@@ -93,6 +104,7 @@ func (em *EventMetrics) Reset() {
em.totalEventsCreated.Set(0)
em.totalEventsChanneled.Set(0)
em.totalEventsNotChanneled.Set(0)
+ em.totalEventsDropped.Set(0)
em.totalEventsStored.Set(0)
em.totalEventsNotStored.Set(0)
em.totalEventsProcessed.Set(0)
@@ -110,6 +122,10 @@ func (em *EventMetrics) IncEventsNotChanneled() {
em.totalEventsNotChanneled.Inc()
}
+func (em *EventMetrics) IncEventsDropped() {
+ em.totalEventsDropped.Inc()
+}
+
func (em *EventMetrics) IncEventsProcessed() {
em.totalEventsProcessed.Inc()
}
@@ -125,3 +141,73 @@ func (em *EventMetrics) IncEventsNotStored() {
func (em *EventMetrics) AddEventsCollected(collectedEvents int) {
em.totalEventsCollected.Add(float64(collectedEvents))
}
+
+// Event system metrics
+
+func (em *EventMetrics) GetEventsCreated() int {
+ metricDto := &dto.Metric{}
+ if err := em.totalEventsCreated.Write(metricDto); err == nil {
+ return int(*metricDto.Gauge.Value)
+ }
+ return -1
+}
+
+func (em *EventMetrics) GetEventsChanneled() int {
+ metricDto := &dto.Metric{}
+ if err := em.totalEventsChanneled.Write(metricDto); err == nil {
+ return int(*metricDto.Gauge.Value)
+ }
+ return -1
+}
+
+func (em *EventMetrics) GetEventsNotChanneled() int {
+ metricDto := &dto.Metric{}
+ if err := em.totalEventsNotChanneled.Write(metricDto); err == nil {
+ return int(*metricDto.Gauge.Value)
+ }
+ return -1
+}
+
+func (em *EventMetrics) GetEventsDropped() int {
+ metricDto := &dto.Metric{}
+ if err := em.totalEventsDropped.Write(metricDto); err == nil {
+ return int(*metricDto.Gauge.Value)
+ }
+ return -1
+}
+
+// Publisher metrics
+
+func (em *EventMetrics) GetEventsProcessed() int {
+ metricDto := &dto.Metric{}
+ if err := em.totalEventsProcessed.Write(metricDto); err == nil {
+ return int(*metricDto.Gauge.Value)
+ }
+ return -1
+}
+
+// Event store metrics
+
+func (em *EventMetrics) GetEventsStored() int {
+ metricDto := &dto.Metric{}
+ if err := em.totalEventsStored.Write(metricDto); err == nil {
+ return int(*metricDto.Gauge.Value)
+ }
+ return -1
+}
+
+func (em *EventMetrics) GetEventsNotStored() int {
+ metricDto := &dto.Metric{}
+ if err := em.totalEventsNotStored.Write(metricDto); err == nil {
+ return int(*metricDto.Gauge.Value)
+ }
+ return -1
+}
+
+func (em *EventMetrics) GetEventsCollected() int {
+ metricDto := &dto.Metric{}
+ if err := em.totalEventsCollected.Write(metricDto); err == nil {
+ return int(*metricDto.Gauge.Value)
+ }
+ return -1
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
