influxdata / influxdata/kapacitor
difference() buffers too many points
- Dominant language
- Go
- Stars
- 2.4k
- Forks
- 479
- Avg merge
- 4d 16h
- Merged PRs (30d)
- 4
Description
```
stream
|from().measurement('system').groupBy('host')
|log()
|difference('value')
|log()
```
```
[task_master:main] 2016/09/14 11:54:24 I! Started task: test
[test:log2] 2016/09/14 11:54:24 I! models.Point{Name:"system", Database:"telegraf", RetentionPolicy:"autogen", Group:"host=foo", Dimensions:models.Dimensions{ByName:false, TagNames:[]string{"host"}}, Tags:models.Tags{"host":"foo"}, Fields:models.Fields{"value":0}, Time:time.Time{sec:63609465264, nsec:224637417, loc:(*time.Location)(0x107adc0)}}
[test:log2] 2016/09/14 11:54:25 I! models.Point{Name:"system", Database:"telegraf", RetentionPolicy:"autogen", Group:"host=foo", Dimensions:models.Dimensions{ByName:false, TagNames:[]string{"host"}}, Tags:models.Tags{"host":"foo"}, Fields:models.Fields{"value":1}, Time:time.Time{sec:63609465265, nsec:251324450, loc:(*time.Location)(0x107adc0)}}
[test:log2] 2016/09/14 11:54:26 I! models.Point{Name:"system", Database:"telegraf", RetentionPolicy:"autogen", Group:"host=foo", Dimensions:models.Dimensions{ByName:false, TagNames:[]string{"host"}}, Tags:models.Tags{"host":"foo"}, Fields:models.Fields{"value":4}, Time:time.Time{sec:63609465266, nsec:275530867, loc:(*time.Location)(0x107adc0)}}
[test:log4] 2016/09/14 11:54:26 I! models.Point{Name:"system", Database:"", RetentionPolicy:"", Group:"host=foo", Dimensions:models.Dimensions{ByName:false, TagNames:[]string{"host"}}, Tags:models.Tags{"host":"foo"}, Fields:models.Fields{"difference":1}, Time:time.Time{sec:63609465265, nsec:251324450, loc:(*time.Location)(0x107adc0)}}
[test:log2] 2016/09/14 11:54:27 I! models.Point{Name:"system", Database:"telegraf", RetentionPolicy:"autogen", Group:"host=foo", Dimensions:models.Dimensions{ByName:false, TagNames:[]string{"host"}}, Tags:models.Tags{"host":"foo"}, Fields:models.Fields{"value":9}, Time:time.Time{sec:63609465267, nsec:300613686, loc:(*time.Location)(0x107adc0)}}
[test:log4] 2016/09/14 11:54:27 I! models.Point{Name:"system", Database:"", RetentionPolicy:"", Group:"host=foo", Dimensions:models.Dimensions{ByName:false, TagNames:[]string{"host"}}, Tags:models.Tags{"host":"foo"}, Fields:models.Fields{"difference":3}, Time:time.Time{sec:63609465266, nsec:275530867, loc:(*time.Location)(0x107adc0)}}
```
Notice how the first `log4` line comes through after the 3rd `log2` line, meaning the `difference()` node has received 3 data points. And the value it emits is the result of the difference of the first 2 points.
This is with kapacitor build from eccff59
Contributor guide
Research direction
Reproduce the issue with the shown stream pipeline and compare the ordering of the log2 and log4 output. Start by locating the implementation behind difference('value'), then verify that the node emits after the second point rather than buffering a third; the observed output should no longer show the extra point being received first.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- go
- Domain
- stream-processing
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 45/100