Skip to content

Commit 461a6b2

Browse files
pawels-optimizelypawels-optimizely
authored
feat: Add observability metrics for event queue (#211)
* feat: Add observability metrics for event queue
1 parent 688ec51 commit 461a6b2

File tree

8 files changed

+220
-7
lines changed

8 files changed

+220
-7
lines changed

‎pkg/client/factory_test.go

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,10 @@ func (f *MockDispatcher) DispatchEvent(event event.LogEvent) (bool, error) {
5151
return true, nil
5252
}
5353

54+
func (f *MockDispatcher) GetMetrics() event.Metrics {
55+
return nil
56+
}
57+
5458
func TestFactoryClientReturnsDefaultClient(t *testing.T) {
5559
factory := OptimizelyFactory{}
5660

‎pkg/event/dispatcher.go

Lines changed: 28 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -37,13 +37,19 @@ var dispatcherLogger = logging.GetLogger("EventDispatcher")
3737
// Dispatcher dispatches events
3838
type Dispatcher interface {
3939
DispatchEvent(event LogEvent) (bool, error)
40+
GetMetrics() Metrics
4041
}
4142

4243
// HTTPEventDispatcher is the HTTP implementation of the Dispatcher interface
4344
type HTTPEventDispatcher struct {
4445
requester *utils.HTTPRequester
4546
}
4647

48+
// GetMetrics is the metric accessor
49+
func (ed *HTTPEventDispatcher) GetMetrics() Metrics {
50+
return nil
51+
}
52+
4753
// DispatchEvent dispatches event with callback
4854
func (ed *HTTPEventDispatcher) DispatchEvent(event LogEvent) (bool, error) {
4955

@@ -71,6 +77,8 @@ type QueueEventDispatcher struct {
7177
eventQueue Queue
7278
eventFlushLock sync.Mutex
7379
Dispatcher Dispatcher
80+
81+
metrics Metrics
7482
}
7583

7684
// DispatchEvent queues event with callback and calls flush in a go routine.
@@ -82,6 +90,17 @@ func (ed *QueueEventDispatcher) DispatchEvent(event LogEvent) (bool, error) {
8290
return true, nil
8391
}
8492

93+
// GetMetrics is the metric accessor
94+
func (ed *QueueEventDispatcher) GetMetrics() Metrics {
95+
ed.eventFlushLock.Lock()
96+
97+
defer func() {
98+
ed.eventFlushLock.Unlock()
99+
}()
100+
return ed.metrics
101+
102+
}
103+
85104
// flush the events
86105
func (ed *QueueEventDispatcher) flushEvents() {
87106

@@ -93,9 +112,11 @@ func (ed *QueueEventDispatcher) flushEvents() {
93112

94113
retryCount := 0
95114

115+
ed.metrics.SetQueueSize(ed.eventQueue.Size())
96116
for ed.eventQueue.Size() > 0 {
97117
if retryCount > maxRetries {
98118
dispatcherLogger.Error(fmt.Sprintf("event failed to send %d times. It will retry on next event sent", maxRetries), nil)
119+
ed.metrics.IncrFailFlushCount()
99120
break
100121
}
101122

@@ -109,6 +130,7 @@ func (ed *QueueEventDispatcher) flushEvents() {
109130
// remove it
110131
dispatcherLogger.Error("invalid type passed to event Dispatcher", nil)
111132
ed.eventQueue.Remove(1)
133+
ed.metrics.IncrFailFlushCount()
112134
continue
113135
}
114136

@@ -119,13 +141,16 @@ func (ed *QueueEventDispatcher) flushEvents() {
119141
dispatcherLogger.Debug(fmt.Sprintf("Dispatched log event %+v", event))
120142
ed.eventQueue.Remove(1)
121143
retryCount = 0
144+
ed.metrics.IncrSuccessFlushCount()
122145
} else {
123146
dispatcherLogger.Warning("dispatch event failed")
124147
// we failed. Sleep some seconds and try again.
125148
time.Sleep(sleepTime)
126149
// increase retryCount. We exit if we have retried x times.
127150
// we will retry again next event that is added.
128151
retryCount++
152+
ed.metrics.IncrRetryFlushCount()
153+
129154
}
130155
} else {
131156
dispatcherLogger.Error("Error dispatching ", err)
@@ -134,13 +159,15 @@ func (ed *QueueEventDispatcher) flushEvents() {
134159
// increase retryCount. We exit if we have retried x times.
135160
// we will retry again next event that is added.
136161
retryCount++
162+
ed.metrics.IncrRetryFlushCount()
137163
}
138164
}
165+
ed.metrics.SetQueueSize(ed.eventQueue.Size())
139166
}
140167

141168
// NewQueueEventDispatcher creates a Dispatcher that queues in memory and then sends via go routine.
142169
func NewQueueEventDispatcher(ctx context.Context) Dispatcher {
143-
dispatcher := &QueueEventDispatcher{eventQueue: NewInMemoryQueue(defaultQueueSize), Dispatcher: &HTTPEventDispatcher{requester: utils.NewHTTPRequester()}}
170+
dispatcher := &QueueEventDispatcher{eventQueue: NewInMemoryQueue(defaultQueueSize), Dispatcher: &HTTPEventDispatcher{requester: utils.NewHTTPRequester()}, metrics: &DefaultMetrics{}}
144171

145172
go func() {
146173
<-ctx.Done()

‎pkg/event/dispatcher_metrics.go

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
/****************************************************************************
2+
* Copyright 2019, Optimizely, Inc. and contributors *
3+
* *
4+
* Licensed under the Apache License, Version 2.0 (the "License"); *
5+
* you may not use this file except in compliance with the License. *
6+
* You may obtain a copy of the License at *
7+
* *
8+
* http://www.apache.org/licenses/LICENSE-2.0 *
9+
* *
10+
* Unless required by applicable law or agreed to in writing, software *
11+
* distributed under the License is distributed on an "AS IS" BASIS, *
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. *
13+
* See the License for the specific language governing permissions and *
14+
* limitations under the License. *
15+
***************************************************************************/
16+
17+
// Package event //
18+
package event
19+
20+
// Metrics is the interface for event processor
21+
type Metrics interface {
22+
SetQueueSize(queueSize int)
23+
IncrSuccessFlushCount()
24+
IncrFailFlushCount()
25+
IncrRetryFlushCount()
26+
}
27+
28+
// DefaultMetrics stores the actual metrics
29+
type DefaultMetrics struct {
30+
QueueSize int
31+
SuccessFlushCount int64
32+
FailFlushCount int64
33+
RetryFlushCount int64
34+
}
35+
36+
// SetQueueSize sets the queue size
37+
func (m *DefaultMetrics) SetQueueSize(queueSize int) {
38+
m.QueueSize = queueSize
39+
}
40+
41+
// IncrSuccessFlushCount increments counter for successful flush
42+
func (m *DefaultMetrics) IncrSuccessFlushCount() {
43+
m.SuccessFlushCount++
44+
}
45+
46+
// IncrFailFlushCount increments counter for failed flush
47+
func (m *DefaultMetrics) IncrFailFlushCount() {
48+
m.FailFlushCount++
49+
}
50+
51+
// IncrRetryFlushCount increments counter for retried flush
52+
func (m *DefaultMetrics) IncrRetryFlushCount() {
53+
m.RetryFlushCount++
54+
}
55+
56+
// Add a metric collection to existing metrics
57+
func (m *DefaultMetrics) Add(v *DefaultMetrics) {
58+
m.QueueSize += v.QueueSize
59+
m.FailFlushCount += v.FailFlushCount
60+
m.SuccessFlushCount += v.SuccessFlushCount
61+
m.RetryFlushCount += v.RetryFlushCount
62+
}

‎pkg/event/dispatcher_metrics_test.go

Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,95 @@
1+
/****************************************************************************
2+
* Copyright 2019, Optimizely, Inc. and contributors *
3+
* *
4+
* Licensed under the Apache License, Version 2.0 (the "License"); *
5+
* you may not use this file except in compliance with the License. *
6+
* You may obtain a copy of the License at *
7+
* *
8+
* http://www.apache.org/licenses/LICENSE-2.0 *
9+
* *
10+
* Unless required by applicable law or agreed to in writing, software *
11+
* distributed under the License is distributed on an "AS IS" BASIS, *
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. *
13+
* See the License for the specific language governing permissions and *
14+
* limitations under the License. *
15+
***************************************************************************/
16+
17+
// Package event //
18+
package event
19+
20+
import (
21+
"testing"
22+
23+
"github.com/stretchr/testify/assert"
24+
)
25+
26+
func TestNewDefaultMetrics(t *testing.T) {
27+
28+
metric := DefaultMetrics{}
29+
assert.Equal(t, 0, metric.QueueSize)
30+
assert.Equal(t, int64(0), metric.SuccessFlushCount)
31+
assert.Equal(t, int64(0), metric.FailFlushCount)
32+
assert.Equal(t, int64(0), metric.RetryFlushCount)
33+
34+
}
35+
36+
func TestSetQueueSize(t *testing.T) {
37+
38+
metric := DefaultMetrics{}
39+
metric.SetQueueSize(24)
40+
assert.Equal(t, 24, metric.QueueSize)
41+
assert.Equal(t, int64(0), metric.SuccessFlushCount)
42+
assert.Equal(t, int64(0), metric.FailFlushCount)
43+
assert.Equal(t, int64(0), metric.RetryFlushCount)
44+
45+
}
46+
47+
func TestIncrSuccessFlushCount(t *testing.T) {
48+
49+
metric := DefaultMetrics{}
50+
metric.IncrSuccessFlushCount()
51+
assert.Equal(t, 0, metric.QueueSize)
52+
assert.Equal(t, int64(1), metric.SuccessFlushCount)
53+
assert.Equal(t, int64(0), metric.FailFlushCount)
54+
assert.Equal(t, int64(0), metric.RetryFlushCount)
55+
56+
}
57+
58+
func TestIncrFailFlushCount(t *testing.T) {
59+
60+
metric := DefaultMetrics{}
61+
metric.IncrFailFlushCount()
62+
assert.Equal(t, 0, metric.QueueSize)
63+
assert.Equal(t, int64(0), metric.SuccessFlushCount)
64+
assert.Equal(t, int64(1), metric.FailFlushCount)
65+
assert.Equal(t, int64(0), metric.RetryFlushCount)
66+
67+
}
68+
69+
func TestIncrRetryFlushCount(t *testing.T) {
70+
71+
metric := DefaultMetrics{}
72+
metric.IncrRetryFlushCount()
73+
assert.Equal(t, 0, metric.QueueSize)
74+
assert.Equal(t, int64(0), metric.SuccessFlushCount)
75+
assert.Equal(t, int64(0), metric.FailFlushCount)
76+
assert.Equal(t, int64(1), metric.RetryFlushCount)
77+
78+
}
79+
80+
func TestAdd(t *testing.T) {
81+
82+
metric := &DefaultMetrics{}
83+
metric.IncrRetryFlushCount()
84+
metric.IncrSuccessFlushCount()
85+
metric.IncrSuccessFlushCount()
86+
metric.IncrFailFlushCount()
87+
metric.SetQueueSize(24)
88+
89+
metric.Add(metric)
90+
assert.Equal(t, 48, metric.QueueSize)
91+
assert.Equal(t, int64(4), metric.SuccessFlushCount)
92+
assert.Equal(t, int64(2), metric.FailFlushCount)
93+
assert.Equal(t, int64(2), metric.RetryFlushCount)
94+
95+
}

‎pkg/event/dispatcher_test.go

Lines changed: 17 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -57,13 +57,18 @@ func TestQueueEventDispatcher_DispatchEvent(t *testing.T) {
5757
// check the queue
5858
assert.Equal(t, 0, qd.eventQueue.Size())
5959

60+
metric := qd.GetMetrics().(*DefaultMetrics)
61+
assert.Equal(t, 0, metric.QueueSize)
62+
assert.Equal(t, int64(1), metric.SuccessFlushCount)
63+
assert.Equal(t, int64(0), metric.FailFlushCount)
64+
assert.Equal(t, int64(0), metric.RetryFlushCount)
65+
6066
}
6167

6268
func TestQueueEventDispatcher_InvalidEvent(t *testing.T) {
6369
ctx := context.TODO()
6470
q := NewQueueEventDispatcher(ctx)
6571

66-
6772
config := TestConfig{}
6873

6974
if qed, ok := q.(*QueueEventDispatcher); ok {
@@ -81,6 +86,12 @@ func TestQueueEventDispatcher_InvalidEvent(t *testing.T) {
8186
// check the queue. bad event type should be removed. but, not sent.
8287
assert.Equal(t, 0, qd.eventQueue.Size())
8388

89+
metric := qd.GetMetrics().(*DefaultMetrics)
90+
assert.Equal(t, 0, metric.QueueSize)
91+
assert.Equal(t, int64(0), metric.SuccessFlushCount)
92+
assert.Equal(t, int64(1), metric.FailFlushCount)
93+
assert.Equal(t, int64(0), metric.RetryFlushCount)
94+
8495
}
8596

8697
func TestQueueEventDispatcher_FailDispath(t *testing.T) {
@@ -108,9 +119,13 @@ func TestQueueEventDispatcher_FailDispath(t *testing.T) {
108119

109120
// give the queue a chance to run
110121
qd.flushEvents()
122+
time.Sleep(1 * time.Second)
111123

112124
// check the queue. bad event type should be removed. but, not sent.
113125
assert.Equal(t, 1, qd.eventQueue.Size())
126+
metric := qd.GetMetrics().(*DefaultMetrics)
127+
assert.Equal(t, 1, metric.QueueSize)
128+
assert.Equal(t, int64(0), metric.SuccessFlushCount)
129+
assert.True(t, metric.RetryFlushCount > 1)
114130

115131
}
116-

‎pkg/event/processor_test.go

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@ import (
2929
)
3030

3131
type CountingDispatcher struct {
32-
eventCount int
32+
eventCount int
3333
visitorCount int
3434
}
3535

@@ -38,7 +38,9 @@ func (c *CountingDispatcher) DispatchEvent(event LogEvent) (bool, error) {
3838
c.visitorCount += len(event.Event.Visitors)
3939
return true, nil
4040
}
41-
41+
func (c *CountingDispatcher) GetMetrics() Metrics {
42+
return nil
43+
}
4244

4345
type MockDispatcher struct {
4446
ShouldFail bool
@@ -54,6 +56,9 @@ func (m *MockDispatcher) DispatchEvent(event LogEvent) (bool, error) {
5456
return true, nil
5557
}
5658

59+
func (f *MockDispatcher) GetMetrics() Metrics {
60+
return nil
61+
}
5762
func NewMockDispatcher(queueSize int, shouldFail bool) *MockDispatcher {
5863
return &MockDispatcher{Events: NewInMemoryQueue(queueSize), ShouldFail: shouldFail}
5964
}

‎scripts/coverage.sh

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,15 +17,15 @@ set -e
1717

1818
workdir=.cover
1919
profile="$workdir/cover.out"
20-
mode=count
20+
mode=atomic
2121

2222
generate_cover_data() {
2323
rm -rf "$workdir"
2424
mkdir "$workdir"
2525

2626
for pkg in "$@"; do
2727
f="$workdir/$(echo $pkg | tr / -).cover"
28-
go test -covermode="$mode" -coverprofile="$f" "$pkg"
28+
go test --race -covermode="$mode" -coverprofile="$f" "$pkg"
2929
done
3030

3131
echo "mode: $mode" >"$profile"

‎tests/integration/optlyplugins/proxy_dispatcher.go

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,11 @@ func (d *ProxyEventDispatcher) DispatchEvent(event event.LogEvent) (bool, error)
3636
return true, nil
3737
}
3838

39+
// DispatchEvent dispatches event with callback
40+
func (d *ProxyEventDispatcher) GetMetrics() event.Metrics {
41+
return nil
42+
}
43+
3944
// GetEvents returns dispatched events
4045
func (d *ProxyEventDispatcher) GetEvents() []event.Batch {
4146
if d.events == nil {

0 commit comments

Comments
 (0)