align bucket boundary to interval in rolling window (#345)

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 {

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) {