Atlas › Test
full-buckets
Exact test identity: go.temporal.io/server/common/stats/TestWindowedDigest/full-buckets
- Package
go.temporal.io/server/common/stats- Suite / test hierarchy
TestWindowedDigest/full-buckets- Test
full-buckets- Introduced at
- windowed_tdigest.go ×1 Frontier kind: Joint frontier
- Covered ranges
- 18
- Covered lines
- 56
- Covered files
- 1
Covered source
Expand a file to inspect source; the > gutter marks covered lines.
go.temporal.io/server/common/stats/windowed_tdigest.go 56 covered LOC · 18 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
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
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
>
}
176
}
178
>
candidate := &w.windows[idx]
179
>
// Window start is inclusive, end is exclusive. We're iterating
180
>
// backwards in time, so the first window that matches is the one we want.
181
>
if !timestamp.Before(candidate.start) {
182
>
// The first window that matches might be too short to include this timestamp.
windowed_tdigest.go
183
>
// Make sure the timestamp is actually in the window.
184
>
if !timestamp.Before(w.windows[idx].end) {
185
return nil, errInGap
186
}
188
}
189
}
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 {
220
curr.tdigest.Reset()
221
curr.start = start
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.
243
>
return (idx - 1 + w.cfg.WindowCount) % w.cfg.WindowCount
244
>
}