influxdata / influxdata/kapacitor

Barrier doesn't seem to generate points

Open
#2,314 0 comments 0 reactions 0 assignees View on GitHub
question
Dominant language
Go
Stars
2.4k
Forks
479
Avg merge
4d 16h
Merged PRs (30d)
4

Description

Hi,

I have a situation where I have a network device that provides me with throughput metrics for its two interfaces (they're collected independently, so have different timestamp). I came up with the following script to get the summary of throughput:

```
var source_akl_for_nat01_25 = stream
|from()
.measurement('a10InterfaceStats')
.where(lambda: "host" == 'akl-for-nat01' AND "interface" == '25')
|barrier()
.period(1m)
| default()
.field('rate_byte_rcvd', 0.0)
.field('rate_byte_sent', 0.0)
| log()
|window()
.period(1m)
.every(1m)
.align()

var akl_for_nat01_25_rate_byte_rcvd = source_akl_for_nat01_25
| mean('rate_byte_rcvd')
| log()

var akl_for_nat01_25_rate_byte_sent = source_akl_for_nat01_25
| mean('rate_byte_sent')
| log()

var source_akl_for_nat01_26 = stream
|from()
.measurement('a10InterfaceStats')
.where(lambda: "host" == 'akl-for-nat01' AND "interface" == '26')
|barrier()
.period(1m)
| default()
.field('rate_byte_rcvd', 0.0)
.field('rate_byte_sent', 0.0)
| log()
|window()
.period(1m)
.every(1m)
.align()

var akl_for_nat01_26_rate_byte_rcvd = source_akl_for_nat01_26
| mean('rate_byte_rcvd')
| log()

var akl_for_nat01_26_rate_byte_sent = source_akl_for_nat01_26
| mean('rate_byte_sent')
| log()

akl_for_nat01_25_rate_byte_rcvd
|join(akl_for_nat01_25_rate_byte_sent, akl_for_nat01_26_rate_byte_rcvd, akl_for_nat01_26_rate_byte_sent)
.as('akl_for_nat01_25_rate_byte_rcvd', 'akl_for_nat01_25_rate_byte_sent', 'akl_for_nat01_26_rate_byte_rcvd', 'akl_for_nat01_26_rate_byte_sent')
.fill('null')
|default()
.field('akl_for_nat01_25_rate_byte_rcvd.mean', 0.0)
.field('akl_for_nat01_25_rate_byte_sent.mean', 0.0)
.field('akl_for_nat01_26_rate_byte_rcvd.mean', 0.0)
.field('akl_for_nat01_26_rate_byte_sent.mean', 0.0)
|eval(lambda: "akl_for_nat01_25_rate_byte_rcvd.mean" + "akl_for_nat01_26_rate_byte_rcvd.mean", lambda: "akl_for_nat01_25_rate_byte_sent.mean" + "akl_for_nat01_26_rate_byte_sent.mean" )
.as('rate_byte_rcvd','rate_byte_sent')
|default()
.tag('host', 'akl-for-nat01')

|influxDBOut()
.create()
.database('intfs')
.measurement('a10InteraceStats_summary')
.precision('s')
```

This works ok, as long as the data comes in. If one of the interfaces goes down the device stops reporting it completely, resulting in one of the streams stalling, resulting in the `join` waiting forever.
I though using `|barrier()` would fix this, but I can't see the point emitted by `barrier()` at all.

I created a very simple script to verify it (there's no interface 27 so the data never comes in)
```
var source_akl_for_nat01_27 = stream
|from()
.measurement('a10InterfaceStats')
.where(lambda: "host" == 'akl-for-nat01' AND "interface" == '27')
|barrier()
.period(1m)
| default()
.field('rate_byte_rcvd', 0.0)
.field('rate_byte_sent', 0.0)
| log()
```

but it never logs anything, looking at the output it just sits there, not doing anything:
```
DOT:
digraph barrierTest {
graph [throughput="0.00 points/s"];

stream0 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
stream0 -> from1 [processed="560"];

from1 [avg_exec_time_ns="22.199µs" errors="0" working_cardinality="0" ];
from1 -> barrier2 [processed="0"];

barrier2 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
barrier2 -> default3 [processed="0"];

default3 [avg_exec_time_ns="0s" errors="0" fields_defaulted="0" tags_defaulted="0" working_cardinality="0" ];
default3 -> log4 [processed="0"];

log4 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
}
```

Is this how `barrier` supposed to work, or is my understanding incorrect?

*edit: typos

Contributor guide

Open the contributing guide

Research direction

Reproduce the barrierTest script using the stream, from, barrier, default, and log nodes, then inspect the DOT output and barrier behavior when no points arrive. Compare this with the join scenario involving the missing interface; done means the expected behavior of barrier with an idle input is established and the reported non-emission is either explained or corrected.

Written by the indexing model from the issue text.

Assessment

Tech stack
go
Domain
stream-processing
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.