From 0786862a351d27a3904366beaac2461210f79ff1 Mon Sep 17 00:00:00 2001 From: Kevin Wan Date: Mon, 4 Jan 2021 11:17:59 +0800 Subject: [PATCH] align bucket boundary to interval in rolling window (#345) --- core/collection/rollingwindow.go | 4 +++- core/collection/rollingwindow_test.go | 5 +++++ 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/core/collection/rollingwindow.go b/core/collection/rollingwindow.go index 189469c3..63203090 100644 --- a/core/collection/rollingwindow.go +++ b/core/collection/rollingwindow.go @@ -96,7 +96,9 @@ func (rw *RollingWindow) updateOffset() { } rw.offset = (offset + span) % rw.size - rw.lastTime = rw.lastTime + rw.interval*time.Duration(span) + now := timex.Now() + // align to interval time boundary + rw.lastTime = now - (now-rw.lastTime)%rw.interval } type Bucket struct { diff --git a/core/collection/rollingwindow_test.go b/core/collection/rollingwindow_test.go index bbdc7821..0d1b2569 100644 --- a/core/collection/rollingwindow_test.go +++ b/core/collection/rollingwindow_test.go @@ -129,6 +129,11 @@ func TestRollingWindowBucketTimeBoundary(t *testing.T) { r.Add(5) r.Add(6) assert.Equal(t, []float64{1, 5, 15}, listBuckets()) + time.Sleep(time.Millisecond * 100) + r.Add(7) + r.Add(8) + r.Add(9) + assert.Equal(t, []float64{0, 0, 24}, listBuckets()) } func TestRollingWindowDataRace(t *testing.T) {