UnionNode buffers points without warning

[hraftery]

After 4 hours on this one, I concede defeat.

Consider this TICKscript:

var one = stream
    |from()
        .measurement('one')

var two = stream
    |from()
        .measurement('two')

one
    |union(two)
        .rename('input')
    @myUDF()
    |log()

It works perfectly on recorded data provided the record uses a query to select historical data. However calling kapacitor record stream does not return and no data is recorded even if a point is inserted into either one or two.

Further, enabling the task produces no log output until x points are added to either one or two. x is about 6, but I couldn't see consistent behaviour to determine a pattern. From there, every time a point is added to either one or two, the point x entries old gets processed. That is, it acts like a FIFO with x elements.

I could find no explanation of this behaviour in the docs, in the source code or via Google searches.

I finally traced it down to the UnionNode by looking at the DOT output of kapacitor show my_task:

digraph my_task {
graph [throughput="0.00 points/s"];

stream0 [avg_exec_time_ns="0s" ];
stream0 -> from2 [processed="15"];
stream0 -> from1 [processed="15"];

from2 [avg_exec_time_ns="2.67µs" ];
from2 -> union4 [processed="11"];

from1 [avg_exec_time_ns="1.894µs" ];
from1 -> union4 [processed="4"];

union4 [avg_exec_time_ns="0s" ];
union4 -> myTask5 [processed="9"];

calcOccupancy5 [avg_exec_time_ns="0s" ];
calcOccupancy5 -> log6 [processed="9"];

log6 [avg_exec_time_ns="0s" ];
log6 -> influxdb_out7 [processed="9"];

influxdb_out7 [avg_exec_time_ns="0s" points_written="8" write_errors="0" ];
}

This seems to indicate that the FromNodes are processing the points immediately, but the UnionNode is 6 points behind. Sure enough, if I remove the UnionNode and connect one of the FromNode's directly, and then add a point to that FromNode, it is processed immediately:

digraph my_task {
graph [throughput="0.00 points/s"];

stream0 [avg_exec_time_ns="0s" ];
stream0 -> from2 [processed="1"];
stream0 -> from1 [processed="1"];

from2 [avg_exec_time_ns="0s" ];
from2 -> calcOccupancy3 [processed="1"];

calcOccupancy3 [avg_exec_time_ns="0s" ];
calcOccupancy3 -> log4 [processed="1"];

log4 [avg_exec_time_ns="0s" ];
log4 -> influxdb_out5 [processed="1"];

influxdb_out5 [avg_exec_time_ns="0s" points_written="1" write_errors="0" ];

from1 [avg_exec_time_ns="1.15µs" ];
}

Is there an explanation for this mysterious behaviour. Can it be disabled so that points are processed immediately?

URL for relevant page?

https://docs.influxdata.com/kapacitor/v1.1//nodes/union_node/

What products and version are you using?

InfluxDB 1.1.1
Kapacitor 1.1.1

Where did you look before opening the issue?

https://docs.influxdata.com/kapacitor/v1.1//nodes/union_node/
https://github.com/influxdata/kapacitor/blob/master/pipeline/union.go
#361

[nathanielc]

@hraftery I also discovered this and started working on a fix (see #1038) but I ran into some significant road blocks. I am transitioning back into a bug fix phase at the moment so I'll take another stab at this.

Thanks for the detailed write up.

[hraftery]

@nathanielc oh that's great! Good luck with it. I'm unfamiliar with the Kapacitor source, but let me know if I can help.