Random lost of events with eKuiper HTTP Push Stream Source

[canob]

[ ORIGINAL DESCRIPTION, THAT FINALLY IS NOT THE PROBLEM, SEE MY NEXT COMMENT]
Hi everyone,
I'm trying to match a field, qname, ended with ".me" or ".xyz" or ".biz", with SQL LIKE operator (reviewed this doc to know how to use LIKE, https://ekuiper.org/docs/en/latest/sqls/query_language_elements.html#where)
I created a rule:
SELECT * FROM dns WHERE qname LIKE "%.me" OR qname LIKE "%.xyz" OR qname LIKE "%.biz"
And the rule has sink to a file, but anything is being written to the file.
For example, my expectation is that this log is being captured by the rule:
{"query_type":"AAAA","source_ip":"10.11.0.45","tag":"dns_detections_uncommon_TLDs","timestamp":"Oct 31 11:09:35","qname":"testing.me"}
Do I need to scape the dot using LIKE? How? Or why is not working?
Thanks in advance for your help.

[canob]

At the end, the problem is with the HTTP Push Source...
Randomly, the HTTP Push Source receive one, two, or the three test events that I generate doing DNS queries (one with a domain ended with .me, one with a domain ended with .biz, and one with a domain ended with .xyz), events that are being collected by FluentBit, and sended using FluentBit HTTP Output to eKuiper HTTP Push Source, and that random behavior is the reason because the rule sends one, two or three events to the destination file, rule that is based on the configured SQL query, SELECT * FROM dns WHERE qname LIKE "%.me" OR qname LIKE "%.xyz" OR qname LIKE "%.biz"
I think that the problem is in eKuiper Push Source, because I'm sending the same logs with the same FluentBit Instance using FluentBit HTTP Output (same output plugin) to OpenObserve and Parseable, and both are receiving the logs correctly, and additionally I'm sending the logs to a file with FluentBit and to Grafana Loki, and the only destination that is not receiving all the logs randomly is eKuiper using HTTP Push Source.
Any advice on what I can review to troubleshoot the behavior of HTTP Push Source?
The describe of the stream that is failing:

root@ekuiper:/kuiper/bin# ./kuiper describe stream dns
Connecting to 127.0.0.1:20498...
Fields
--------------------------------------------------------------------------------

CONF_KEY: default
DATASOURCE: /api/data/dns
FORMAT: json
TIMESTAMP: timestamp
TYPE: httppush

And the describe of the rule:

root@ekuiper:/kuiper/bin# ./kuiper describe rule dns_detections_uncommon_TLDs
Connecting to 127.0.0.1:20498...
{
  "id": "dns_detections_uncommon_TLDs",
  "name": "dns_detections_uncommon_TLDs",
  "triggered": true,
  "sql": "SELECT * FROM dns WHERE qname LIKE \"%.me\" OR qname LIKE \"%.biz\" OR qname LIKE \"%.xyz\"",
  "actions": [
    {
      "file": {
        "resourceId": "dns",
        "checkInterval": 6000,
        "fileType": "lines",
        "hasHeader": false,
        "path": "/kuiper/output/dns-detections.json",
        "rollingCount": 10000,
        "rollingInterval": 36000,
        "rollingNamePattern": "suffix",
        "omitIfEmpty": false,
        "sendSingle": true,
        "format": "json",
        "bufferLength": 1024,
        "enableCache": false,
        "runAsync": false
      }
    },
    {
      "rest": {
        "resourceId": "dns-fluentbit-output",
        "bodyType": "json",
        "debugResp": false,
        "insecureSkipVerify": true,
        "method": "POST",
        "oauth": {
          "access": {},
          "refresh": {}
        },
        "responseType": "code",
        "timeout": 5000,
        "url": "http://fluentbit-linux:8889",
        "omitIfEmpty": false,
        "sendSingle": true,
        "format": "json",
        "bufferLength": 1024,
        "enableCache": false,
        "runAsync": false
      }
    }
  ],
  "options": {
    "debug": false,
    "logFilename": "",
    "isEventTime": false,
    "lateTolerance": 1000,
    "concurrency": 1,
    "bufferLength": 1024,
    "sendMetaToSink": false,
    "sendError": true,
    "qos": 0,
    "checkpointInterval": 300000,
    "cron": "",
    "duration": "",
    "cronDatetimeRange": null,
    "restartStrategy": {
      "attempts": 0,
      "delay": 1000,
      "multiplier": 2,
      "maxDelay": 30000,
      "jitter": 0.1
    }
  }
}

Thanks in advance for your help.

[ngjaying]

What is the throughput of the message sending to http push source? More than 1000 per second? You can enable debug and all received data in httppush will be printed to the log.

[canob]

No, less than 1000 per second. I'm going to review how to enable debug in the ekuiper config file to get the data received by httppush printed to the log, for compare.
Thanks for your help @ngjaying , and sorry for continue bothering you with all my questions.

[ngjaying]

No problem at all. Thanks for reporting problems.

[canob]

@ngjaying, can you tell me how to enable the debbuging you mentioned for httpush to enable all received data to be printed to the log?
Because I changed kuiper.yaml to this:

basic:
  # true|false, with debug level, it prints more debug info
  debug: true
  # true|false, if it's set to true, then the log will be print to console
  consoleLog: true
  # true|false, if it's set to true, then the log will be print to log file
  fileLog: true

But I still don't seeing received data from httpush on stream.log file after ekuiper docker restart.
Thanks in advance.

[canob]

Strange, I modified kuiper.yaml to enable debug (is in a persisted volume) and restarted the docker (previous post), but was not working, and is only working after I added KUIPER__BASIC__DEBUG: "true" to the env of the Docker. Shouldn't the debug be working in both cases?
I'm going to review the logs now and back to you with the conclusions.
Thanks again @ngjaying

[canob]

So I did many tests for this, and every test suggests that eKuiper HTTP Push Source is lossing events randomly.

The tests:

Event pipelines that I created for the same DNS events source:

FluentBit Forward Output -> Fluentd MQTT Output -> EMQX MQTT -> eKuiper MQTT Stream Source named mqtt -> eKuiper Rule with "SELECT * FROM mqtt"
FluentBit HTTP Output -> eKuiper HTTP Push Stream Source named dns -> eKuiper Rule with "SELECT * FROM dns"
FluentBit HTTP Output -> OpenObserve
FluentBit HTTP Output -> Parseable
FluentBit File Output -> TextFile

On every case, FluentBit send all the events perfectly, but not in the case of eKuiper HTTP Push Stream Source, which missed several.

To compare the received events in both eKuiper Sources, I did DNS queries to this domains:
marking.me, marking.biz, and marking.xyz

marketing.me, marketing.biz, and marketing.xyz

After I did that, I review the stream.log file (eKuiper is in debug mode):

root@ar09-ubuntu:/disk/ekuiper/log# cat stream.log | grep marking | grep "received message"
root@ar09-ubuntu:/disk/ekuiper/log# cat stream.log | grep marking | grep "project plan receive"
time="2023-11-01 20:05:31" level=debug msg="project plan receive &{mqtt map[qname:marking.me query_type:A source_ip:10.11.0.45 tag:dns_query timestamp:Nov  1 17:06:22] 1698869131868 map[messageid:0 topic:dns_query] {{{0 0} 0 0 {{} 0} {{} 0}} map[] map[]} {0 0} map[]}" file="operator/project_operator.go:54" rule=mqtt
time="2023-11-01 20:05:32" level=debug msg="project plan receive &{mqtt map[qname:marking.me query_type:AAAA source_ip:10.11.0.45 tag:dns_query timestamp:Nov  1 17:06:23] 1698869132863 map[messageid:0 topic:dns_query] {{{0 0} 0 0 {{} 0} {{} 0}} map[] map[]} {0 0} map[]}" file="operator/project_operator.go:54" rule=mqtt
time="2023-11-01 20:05:35" level=debug msg="project plan receive &{mqtt map[qname:marking.biz query_type:A source_ip:10.11.0.45 tag:dns_query timestamp:Nov  1 17:06:26] 1698869135863 map[messageid:0 topic:dns_query] {{{0 0} 0 0 {{} 0} {{} 0}} map[] map[]} {0 0} map[]}" file="operator/project_operator.go:54" rule=mqtt
time="2023-11-01 20:05:35" level=debug msg="project plan receive &{mqtt map[qname:marking.biz query_type:AAAA source_ip:10.11.0.45 tag:dns_query timestamp:Nov  1 17:06:26] 1698869135867 map[messageid:0 topic:dns_query] {{{0 0} 0 0 {{} 0} {{} 0}} map[] map[]} {0 0} map[]}" file="operator/project_operator.go:54" rule=mqtt
time="2023-11-01 20:05:40" level=debug msg="project plan receive &{mqtt map[qname:marking.xyz query_type:A source_ip:10.11.0.45 tag:dns_query timestamp:Nov  1 17:06:31] 1698869140856 map[messageid:0 topic:dns_query] {{{0 0} 0 0 {{} 0} {{} 0}} map[] map[]} {0 0} map[]}" file="operator/project_operator.go:54" rule=mqtt
time="2023-11-01 20:05:40" level=debug msg="project plan receive &{mqtt map[qname:marking.xyz query_type:AAAA source_ip:10.11.0.45 tag:dns_query timestamp:Nov  1 17:06:31] 1698869140857 map[messageid:0 topic:dns_query] {{{0 0} 0 0 {{} 0} {{} 0}} map[] map[]} {0 0} map[]}" file="operator/project_operator.go:54" rule=mqtt
root@ar09-ubuntu:/disk/ekuiper/log# cat stream.log | grep marketing | grep "received message"
time="2023-11-01 20:11:21" level=debug msg="httppush received message map[date:2023-11-01T20:12:11.000000Z qname:marketing.xyz query_type:AAAA source_ip:10.11.0.45 tag:dns_query timestamp:Nov  1 17:12:11]" file="httpserver/data_server.go:92" httppush_connection=0
root@ar09-ubuntu:/disk/ekuiper/log# cat stream.log | grep marketing | grep "project plan receive"
time="2023-11-01 20:11:10" level=debug msg="project plan receive &{mqtt map[qname:marketing.me query_type:A source_ip:10.11.0.45 tag:dns_query timestamp:Nov  1 17:12:01] 1698869470856 map[messageid:0 topic:dns_query] {{{0 0} 0 0 {{} 0} {{} 0}} map[] map[]} {0 0} map[]}" file="operator/project_operator.go:54" rule=mqtt
time="2023-11-01 20:11:10" level=debug msg="project plan receive &{mqtt map[qname:marketing.me query_type:AAAA source_ip:10.11.0.45 tag:dns_query timestamp:Nov  1 17:12:01] 1698869470856 map[messageid:0 topic:dns_query] {{{0 0} 0 0 {{} 0} {{} 0}} map[] map[]} {0 0} map[]}" file="operator/project_operator.go:54" rule=mqtt
time="2023-11-01 20:11:14" level=debug msg="project plan receive &{mqtt map[qname:marketing.biz query_type:A source_ip:10.11.0.45 tag:dns_query timestamp:Nov  1 17:12:05] 1698869474852 map[messageid:0 topic:dns_query] {{{0 0} 0 0 {{} 0} {{} 0}} map[] map[]} {0 0} map[]}" file="operator/project_operator.go:54" rule=mqtt
time="2023-11-01 20:11:14" level=debug msg="project plan receive &{mqtt map[qname:marketing.biz query_type:AAAA source_ip:10.11.0.45 tag:dns_query timestamp:Nov  1 17:12:05] 1698869474857 map[messageid:0 topic:dns_query] {{{0 0} 0 0 {{} 0} {{} 0}} map[] map[]} {0 0} map[]}" file="operator/project_operator.go:54" rule=mqtt
time="2023-11-01 20:11:20" level=debug msg="project plan receive &{mqtt map[qname:marketing.xyz query_type:A source_ip:10.11.0.45 tag:dns_query timestamp:Nov  1 17:12:11] 1698869480858 map[messageid:0 topic:dns_query] {{{0 0} 0 0 {{} 0} {{} 0}} map[] map[]} {0 0} map[]}" file="operator/project_operator.go:54" rule=mqtt
time="2023-11-01 20:11:21" level=debug msg="project plan receive &{mqtt map[qname:marketing.xyz query_type:AAAA source_ip:10.11.0.45 tag:dns_query timestamp:Nov  1 17:12:11] 1698869481862 map[messageid:0 topic:dns_query] {{{0 0} 0 0 {{} 0} {{} 0}} map[] map[]} {0 0} map[]}" file="operator/project_operator.go:54" rule=mqtt

As you can see, for the eKuiper HTTP Push Source, in the first case, domains starting with marking, I didn't receive any events (to eKuiper MQTT Source I received all the events), and for the second case, domains starting with marketing, I only received one event, instead of the six that I received with eKuiper MQTT Source.

So, it is clear that eKuiper HTTP Push is loosing events, but I don't know which other tests I can do to provide more troubleshooting information.

I'm sure there isn't a problem on FluentBit's side, simply because it is sending the same DNS events to OpenObserve and Parseable using the same output plugin, HTTP Output, without any problem.

@ngjaying, let me know If I can provide more information about the issue.

[canob]

stream.log @ngjaying ?

[ngjaying]

Yes

[canob]

Here is the file
stream.zip

[ngjaying]

In the log, it seems you have several rules to consume the httppush source. Can you stop them and leave only dns_query? Also, what's your metrics like for rule dns_query?

[canob]

Stopped all the other rules.
Here are the metrics for rule dns_query:

[canob]

So I'm add one additional event pipeline for the same DNS events source:
FluentBit Forward Output -> Fluentd HTTP Output -> eKuiper HTTP Push Stream Source named dns_fluentd -> eKuiper Rule with "SELECT * FROM dns_fluentd"
And compared to OpenObserve/Parseable/Grafana Loki, I'm loosing events every time that I used eKuiper HTTP Push Stream Source, even using FluentBit HTTP Output or Fluentd HTTP Output to send the events to eKuiper.
For example, I generated 6 events for the domain espndeportes, and I see that 6 events on OpenObserve/Parseable/Grafana Loki, and on stream.log file I see 1 received event from eKuiper HTTP Push Stream Source.
The 6 events on OpenObserve/Parseable/Grafana Loki:



Only 1 received event by HTTP Push Stream Source:

root@ar09-ubuntu:/disk/ekuiper/log# cat stream.log | grep espndeportes | grep "received message"
time="2023-11-02 01:12:33" level=debug msg="httppush received message map[date:2023-11-02T01:13:24.000000Z qname:espndeportes.com.ar query_type:A source_ip:10.11.0.45 tag:dns_query timestamp:Nov  1 22:13:24]" file="httpserver/data_server.go:92" httppush_connection=0
root@ar09-ubuntu:/disk/ekuiper/log# cat stream.log | grep espndeportes | grep "project plan receive"
time="2023-11-02 01:12:33" level=debug msg="project plan receive &{dns map[date:2023-11-02T01:13:24.000000Z qname:espndeportes.com.ar query_type:A source_ip:10.11.0.45 tag:dns_query timestamp:Nov  1 22:13:24] 1698887553838 map[topic:$$httppush//api/data/dns] {{{0 0} 0 0 {{} 0} {{} 0}} map[] map[]} {0 0} map[]}" file="operator/project_operator.go:54" rule=dns_query

So, with eKuiper HTTP Push Stream Source (FluentBit Forward Output -> Fluentd HTTP Output -> eKuiper HTTP Push Stream Source named dns_fluentd -> eKuiper Rule with "SELECT * FROM dns_fluentd"), I'm loosing events.
And with eKuiper MQTT Stream Source (FluentBit Forward Output -> Fluentd MQTT Output -> EMQX MQTT -> eKuiper MQTT Stream Source named mqtt -> eKuiper Rule with "SELECT * FROM mqtt"), I'm not loosing any events.

I have no idea what is happening with eKuiper HTTP Push Stream Source, or how can I help, :(

[canob]

No, but I can try to put a HTTP Proxy or API Gateway in the middle of FluentBit and eKuiper to get and example for each of them.

[canob]

Hi @ngjaying,
Thanks to https://github.com/socsieng/capture-proxy, attached all the requests of FluentBit and the responses of eKuiper using the four formats of FluentBit HTTP Output: msgpack, json, json_stream, json_lines.
The tree events that I send on every test:

{"qname":"comm-cohort.ess.apple.com","source_ip":"10.11.21.7","timestamp":"Nov  2 10:38:05","tag":"dns_query","query_type":"A"}
{"qname":"roaming.officeapps.live.com","source_ip":"10.199.0.145","timestamp":"Nov  2 10:38:05","tag":"dns_query","query_type":"A"}
{"query_type":"A","timestamp":"Nov  2 10:38:05","tag":"dns_query","source_ip":"10.11.0.50","qname":"ntp.msn.com"}

Let me know if that is what you asked for.

captures.zip

[ngjaying]

What format do you use when encountering random lost? It seems json_stream, json_lines are send out multiple json maps in one request. eKuiper may only decode one of them.

[canob]

I used json_stream and json_lines format, and the result was just what you are saying, got only the first event. With msgpack and json format I got no events and an error on eKuiper logs that cannot decode.
I think FluentBit send more than an event by request because of performance (send only one event by request maybe is slower), and the other tools with HTTP Input that I used (OpenObserve, Parseable, Grafana Loki) can manage that and receive all the events, so it is possible to eKuiper in the future understand and receive all the events with some of these formats too with HTTP Push Source?
Plan B is add EMQX MQTT Broker and Fluentd (to have MQTT Output, because FluentBit doesn't have it) between FluentBit and eKuiper, but it is adding two additional tools to the pipeline to manage and care about. eKuiper File Source is not an option too because what you already told me about that is not designed for "realtime" data transfer, because of how manage the processed files (process files, not reading them as a "tail").

[ngjaying]

It's a good idea to support these formats. You can create a feature request. Thanks.