Atlas › Test
TestWindowedDigest_RingBufferWrapPreservesNewest
Exact test identity: go.temporal.io/server/common/stats/TestWindowedDigest_RingBufferWrapPreservesNewest
- Package
go.temporal.io/server/common/stats- Suite / test hierarchy
TestWindowedDigest_RingBufferWrapPreservesNewest- Test
TestWindowedDigest_RingBufferWrapPreservesNewest- Introduced at
- TestWindowedDigest_RingBufferWrapPreservesNewest Frontier kind: Test frontier
- Covered ranges
- 24
- Covered lines
- 76
- Covered files
- 2
Covered source
Expand a file to inspect source; the > gutter marks covered lines.
go.temporal.io/server/common/stats/windowed_tdigest.go 73 covered LOC · 23 ranges
Open complete file
72
// So, if you want 300 seconds of history on an event that records 1k counts/sec, 3 10-second windows
73
// is fine.
74
>
func NewWindowedTDigest(cfg WindowConfig) (TimeWindowedStats, error) {
windowed_tdigest.go
75
>
if cfg.WindowCount <= 0 {
76
return nil, errors.New("windowCount must be non-negative")
77
}
79
return nil, errors.New("probable misconfiguration detected: windowSize is too small, consider increasing it to at least 50ms")
80
}
82
>
windows: make([]timedWindow, cfg.WindowCount),
83
>
cfg: cfg,
84
>
// mu and head both empty
85
>
}, nil
86
}
87
94
}
95
96
>
func (w *timeWindowedTDigest) Record(value float64, timestamp time.Time) {
windowed_tdigest.go
97
>
w.RecordMulti(value, timestamp, 1)
98
>
}
99
100
>
func (w *timeWindowedTDigest) RecordMulti(value float64, timestamp time.Time, count uint64) {
windowed_tdigest.go
101
>
if err := w.addToWindow(timestamp, value, count); err != nil {
102
// Data must have been too old, ignore.
103
return
105
}
106
108
>
merged := w.getMergedWindows()
109
>
if merged == nil || merged.Count() == 0 {
110
return 0
111
}
112
114
}
115
116
>
func (w *timeWindowedTDigest) TrimmedMean(lowerQuantile, upperQuantile float64) float64 {
windowed_tdigest.go
117
>
merged := w.getMergedWindows()
118
>
if merged == nil || merged.Count() == 0 {
119
return 0
120
}
121
123
}
124
126
>
w.mu.Lock()
127
>
defer w.mu.Unlock()
128
>
129
>
var merged *tdigest.TDigest
130
>
for idx := range w.windows {
131
>
td := w.windows[idx].tdigest
132
>
if td == nil {
133
continue
134
}
135
137
>
merged = td.Clone()
138
>
} else {
140
>
}
141
}
143
}
144
145
>
func (w *timeWindowedTDigest) addToWindow(timestamp time.Time, value float64, count uint64) error {
windowed_tdigest.go
146
>
w.mu.Lock()
147
>
defer w.mu.Unlock()
148
>
149
>
candidate, err := w.searchWindowsBackwards(timestamp)
150
>
if err != nil {
151
>
if !errors.Is(err, errTooNew) {
152
// err is errTooOld or errInGap
153
return err
154
}
155
157
}
158
160
}
161
166
// Precondition: w.mu is held.
167
// Returns the window containing the given timestamp, or error if no window exists.
168
>
func (w *timeWindowedTDigest) searchWindowsBackwards(timestamp time.Time) (*timedWindow, error) {
windowed_tdigest.go
169
>
latest := &w.windows[w.head]
170
>
if !timestamp.Before(latest.start) {
171
>
// If the requested timestamp is after the latest window, no point in searching
172
>
if !timestamp.Before(latest.end) {
173
>
return nil, errTooNew
174
>
}
175
return latest, nil
176
}
193
194
// Precondition: w.mu is held.
195
>
func (w *timeWindowedTDigest) advanceWindow(timestamp time.Time) *timedWindow {
windowed_tdigest.go
196
>
if w.cfg.FillBlankIntervals {
197
// Fill in all the intervening blank windows.
198
// The empty digests will not affect the stats, but they will drop old windows.
211
return lastWindow
212
}
214
}
215
216
// Precondition: w.mu is held.
217
>
func (w *timeWindowedTDigest) advanceWindowSimple(start time.Time) *timedWindow {
windowed_tdigest.go
218
>
w.head = w.modInc(w.head)
219
>
if curr := &w.windows[w.head]; curr.tdigest != nil {
221
>
curr.start = start
222
>
curr.end = start.Add(w.cfg.WindowSize)
223
>
return curr
224
>
}
225
226
>
digest, _ := tdigest.New(tdigest.RandomNumberGenerator(fastrand.Rand{}))
windowed_tdigest.go
227
>
window := timedWindow{
228
>
tdigest: digest,
229
>
start: start,
230
>
end: start.Add(w.cfg.WindowSize),
231
>
}
232
>
w.windows[w.head] = window
233
>
return &window
234
}
235
236
// modulo-increment, for wrapping around the ring buffer
238
>
return (idx + 1) % w.cfg.WindowCount
239
>
}
240
241
// modulo-decrement, for wrapping around the ring buffer.
go.temporal.io/server/common/fastrand/fastrand.go 3 covered LOC · 1 range
Open complete file
168
169
// Clone method is a no-op to support use in tdigest library.
171
>
return r
172
>
}