Kapacitor: How to debug stream task - httpOut always returns series: null

[seb-koch]

I am writing here after i spent hours of trying to debug my stream task and failed. I am out of ideas and hope that you can point me to the mistake i made. I lack ways to debug that stream, even though log nodes were added i see no errors anywhere and believe some major step is missing?

It's a simple test where i want to have a httpOut() node print out some data. At this point i do not even care what data, just some small hello world.

My Problem

I created the stream task in Chronograf but whenever i issue a call to the API endpoint i receive a null series.

URL: http://localhost:9092/kapacitor/v1/tasks/httpout-test/sensor
Response: {"series":null}

Version

~ ❯❯❯ kapacitor version
Kapacitor v1.3.3 (git:  ce586f35e89e75a1779e2b493caba15d66295a15)

Data

Query: SELECT * FROM "azeti"."autogen"."Value" WHERE time > now() - 20s

Raw

11/6/2017 5:25:21 PM,,5e908c8b-75a5-4116-a9b6-2113a9da3b43,0,2b7b223f-6f68-4c0d-beb2-2a4bbe8a90d5,Analogue_Output_1,0,
11/6/2017 5:25:21 PM,,5e908c8b-75a5-4116-a9b6-2113a9da3b43,0,7bbc56a4-e8d6-4c24-b575-7249b76973f3,Analogue_Input_1,0,

The Task

~ ❯❯❯ kapacitor show httpout-test
ID: httpout-test
Error:
Template:
Type: stream
Status: enabled
Executing: true
Created: 06 Nov 17 15:33 CET
Modified: 06 Nov 17 17:19 CET
LastEnabled: 06 Nov 17 17:19 CET
Databases Retention Policies: ["azeti"."autogen"]
TICKscript:
stream
    |from()
        .database('azeti')
        //        .retentionPolicy('autogen')
        .measurement('Value')
    //        .groupBy('sensor_guid', 'location_guid')
    |window()
        .period(2m)
        .every(30s)
        .align()
    |log()
    |last('processing_level')
    //      .where(lambda: "state" == 'WARNING')
    |log()
    |httpOut('sensor')

DOT:
digraph httpout-test {
graph [throughput="0.00 points/s"];

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

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

window2 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
window2 -> log3 [processed="0"];

log3 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
log3 -> last4 [processed="0"];

last4 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
last4 -> log5 [processed="0"];

log5 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
log5 -> http_out6 [processed="0"];

http_out6 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
}

Questions

1. How can i troubleshoot stream tasks? None of my '|log()' logged anything into the logfile, i got DEBUG log level enabled. I can only see the HTTP requests.

[httpd] ::1 - - [06/Nov/2017:17:27:57 +0100] "GET /kapacitor/v1/tasks/httpout-test/sensor HTTP/1.1" 200 39 "-" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/61.0.3163.100 Safari/537.36" 72d2efb9-c30f-11e7-840b-000000000000 260

1. Do stream tasks require a record&replay like batch tasks do? My understanding from the docs is that stream tasks live by the data collection from incoming data, e.g. my influxdb that receives new values every n seconds.
   I found nothing in the docs but wonder if i miss any important enabling step? I did enable the task in Chronograf.

[seb-koch]

Even when adding a log node with level DEBUG nothing shows up in the logs besides the GET request, it's really a black box right now. I believe the stream is not even executed and the GET request goes into a void?
Your help is highly appreciated, i am sure it's some tiny flaw.

This is my tick script now:

    |from()
        .database('azeti')
        .retentionPolicy('autogen')
        .measurement('Value')
    //        .groupBy('sensor_guid', 'location_guid')
    |log()
        .level('DEBUG')
    |window()
        .period(1h)
        .every(10s)
    |log()
        .level('DEBUG')
    |count('value')
    |log()
        .level('DEBUG')
    |httpOut('sensor')
    |log()
        .level('DEBUG')

And the task details (it seems like no data reaches the stream at all?

ubuntu@ARK2151:~$ kapacitor show apitest
ID: apitest
Error:
Template:
Type: stream
Status: enabled
Executing: true
Created: 06 Nov 17 19:36 CET
Modified: 07 Nov 17 10:05 CET
LastEnabled: 07 Nov 17 10:05 CET
Databases Retention Policies: ["azeti"."autogen"]
TICKscript:
stream
    |from()
        .database('azeti')
        .retentionPolicy('autogen')
        .measurement('Value')
    //        .groupBy('sensor_guid', 'location_guid')
    |log()
        .level('DEBUG')
    |window()
        .period(1h)
        .every(10s)
    |log()
        .level('DEBUG')
    |count('value')
    |log()
        .level('DEBUG')
    |httpOut('sensor')
    |log()
        .level('DEBUG')

DOT:
digraph apitest {
graph [throughput="0.00 points/s"];

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

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

log2 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
log2 -> window3 [processed="0"];

window3 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
window3 -> log4 [processed="0"];

log4 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
log4 -> count5 [processed="0"];

count5 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
count5 -> log6 [processed="0"];

log6 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
log6 -> http_out7 [processed="0"];

http_out7 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
http_out7 -> log8 [processed="0"];

log8 [avg_exec_time_ns="0s" errors="0" working_cardinality="0" ];
}

[seb-koch]

We figured out the issue. The influxdb server wasn't able to contact the kapacitor server due to an blocking firewall.

Questions

1. Why there was no helpful debug output at all?
2. Why do batches work but streams are affected by this firewall issue?

Suggestions

1. Maybe you could extend the responsible function with a debug output in case a response wasn't received within a certain timeout?
2. Add a kapacitor status check command that does a send/receive towards the configured influx?
   This would point out quickly to an network issue.

[cub-bsi]

Running into the same problem
@seb-koch could you share some insights on how you solved this? I.e. what port do have to be reachable etc.

[seb-koch]

Hi, unfortunately this happened quite some time ago and we stopped using kapacitor due to the complexity. As far as i remember it was simply our firewall blocking a request and we added an exception that kapacitor was able to reach influx through the API ports. Hope this helps a bit. Regards,

[kszarlej]

Why do batches work but streams are affected by this firewall issue?

It is simply because of how stream works. For streams kapacitor only registers a subscription in InfluxDB and waits for incoming data from it. InfluxDB is issuing HTTP connections to subscribed kapacitor and replicates the data - hence it needs the connectivity to kapacitor because he is the connection issuer. For query tasks kapacitor just issue queries and gets the outputs in the same connection that's why influxdb doesn't need connectivity to kapacitor.

In any case, for stream processing error logs would actually be visible in InfluxDB logs (log lines about being unable to send data to subscriber). Documentation should be a bit more helpful here.