Node to detect changes between consecutive values in a series field #1844
Conversation
change_detect.go
Outdated
| if s := begin.SizeHint(); s > 0 { | ||
| begin = begin.ShallowCopy() | ||
| begin.SetSizeHint(s - 1) | ||
| } |
There was a problem hiding this comment.
This size hinting is not relevant to the change detect node, meaning its now know that the batch size will be s - 1, but rather its unknown. As such the size hint should be set to 0 since an unknown number of points will be emitted.
change_detect.go
Outdated
| } | ||
|
|
||
| fields := n.Fields().Copy() | ||
| fields[g.n.d.Field] = value |
There was a problem hiding this comment.
Looking over the code I believe this can be simplified. The value here is guaranteed to be the existing value for the field correct? Meaning that all the logic to mutate the point can be removed, since changeDetect is just a special case of a filter node.
That means that doChangeDetect need only accept a edge.FieldTagsTimeGetter and does not need the setter.
If this is not the case, what code path actually changes the value of the field?
change_detect.go
Outdated
| // changeDetect calculates the changeDetect between prev and cur. | ||
| // Return is the resulting changeDetect, whether the current point should be | ||
| // stored as previous, and whether the point result should be emitted. | ||
| func (n *ChangeDetectNode) changeDetect(prev, curr models.Fields, prevTime, currTime time.Time) (interface{}, bool, bool) { |
There was a problem hiding this comment.
It looks like {prev,curr}Time are unused, we should remove them from the function args.
integrations/streamer_test.go
Outdated
| stream | ||
| |from().measurement('packets') | ||
| |changeDetect('value') | ||
| |window() |
There was a problem hiding this comment.
nit, could you fix the whitespace difference here?
| // The changeDetect is computed on a single field, | ||
| // discarding consecutive duplicate values, detecting | ||
| // detects the points at which the series field | ||
| // changes from one value to another. |
There was a problem hiding this comment.
I think it would be good to mention the output of the changeDetect node, i.e. that it just passes the data through unmodified and therefore acts like a special kind of filter.
There was a problem hiding this comment.
FYI, the Kapacitor docs here are autogenerated from these comments on the types in the pipeline package.
pipeline/tick/change_detect_test.go
Outdated
| |changeDetect('work') | ||
| .as('very important') | ||
| .unit(1h) | ||
| .nonNegative() |
There was a problem hiding this comment.
This test needs to be updated.
Required for all non-trivial PRs