lf-edge / ekuiper Goto Github PK
View Code? Open in Web Editor NEWLightweight data stream processing engine for IoT edge
Home Page: https://ekuiper.org
License: Apache License 2.0
Lightweight data stream processing engine for IoT edge
Home Page: https://ekuiper.org
License: Apache License 2.0
./cli --version
flag provided but not defined: -version
Usage of ./cli:
-env string
set environment to prod or test (default "prod")
The extended functions are not aggregate functions. Need to support aggregate functions
The command of bin/cli getstatus rule rule1
listed as in below.
{
"source_demo_0_records_in_total": 0,
"source_demo_0_records_out_total": 0,
"source_demo_0_exceptions_total": 0,
"source_demo_0_process_latency_ms": 0,
"source_demo_0_buffer_length": 0,
"op_preprocessor_demo_0_records_in_total": 0,
"op_preprocessor_demo_0_records_out_total": 0,
"op_preprocessor_demo_0_exceptions_total": 0,
"op_preprocessor_demo_0_process_latency_ms": 0,
"op_preprocessor_demo_0_buffer_length": 0,
"op_window_0_records_in_total": 0,
"op_window_0_records_out_total": 0,
"op_window_0_exceptions_total": 0,
"op_window_0_process_latency_ms": 0,
"op_window_0_buffer_length": 0,
"op_filter_0_records_in_total": 0,
"op_filter_0_records_out_total": 0,
"op_filter_0_exceptions_total": 0,
"op_filter_0_process_latency_ms": 0,
"op_filter_0_buffer_length": 0,
"op_project_0_records_in_total": 0,
"op_project_0_records_out_total": 0,
"op_project_0_exceptions_total": 0,
"op_project_0_process_latency_ms": 0,
"op_project_0_buffer_length": 0,
"sink_sink_rest_0_records_in_total": 0,
"sink_sink_rest_0_records_out_total": 0,
"sink_sink_rest_0_exceptions_total": 0,
"sink_sink_rest_0_process_latency_ms": 0,
"sink_sink_rest_0_buffer_length": 0
}
kuiper_
prefix should be added back, otherwise the metrics cannot be easily found in Prometheus console.
Prometheus metrics collection should be disabled by default for two reasons?
i'am already using prometheus blackbox exporter to alert expiry ssl
can i install node-exporter too in the same prometheus server ??
The having
clause should only support avg, count, max, min, sum
functions. The SQL parser need to validate it before execution.
Currently, no protocols are required for running dockers. With this approach, user cannot change to SSL connections.
docker run -d --name kuiper -e MQTT_BROKER_ADDRESS=broker.emqx.io:1883 emqx/kuiper:latest
So Docker image need to let user to specify the protocols, such as in below.
docker run -d --name kuiper -e MQTT_BROKER_ADDRESS=tcp://broker.emqx.io:1883 emqx/kuiper:latest
EdgeX MQTT source support
Be able to create for Schemaless support...
bin/cli create stream demo '() WITH (FORMAT="JSON", DATASOURCE="demo" TYPE="edgex")'
Develop test script for CLI
Need an automatic approach to create following kuiper releases.
engine_x.x.x.darwin-amd64.tar.gz - For Mac
engine_x.x.x.darwin-amd64.zip - For Mac
engine_x.x.x.linux-amd64.tar.gz - For Linux 64
engine_x.x.x.linux-amd64.zip - For Linux 64
engine_x.x.x.linux-386.tar.gz - For Linux 32
engine_x.x.x.linux-386.zip - For Linux 32
Build the binary from branch agg_func_validator
.
bin/cli create stream demo '(temperature float, humidity bigint) WITH (FORMAT="JSON", DATASOURCE="devices/+/messages")'
bin/cli query
command.SELECT avg(temperature) AS t_av, max(temperature) AS t_max, min(temperature) AS t_min, COUNT(*) As t_count, split_value(mqtt(topic), "/", 1) AS device_id FROM demo GROUP BY device_id, TUMBLINGWINDOW(ss, 5);
1,20,30
2,31,40
1,35,50
2,20,30
1,80,90
2,45,20
1,10,90
2,12,30
1,65,35
2,55,32
xstream > [{"device_id":"1","t_av":26.5,"t_count":4,"t_max":35,"t_min":20}]
[{"device_id":"1","t_av":42.4,"t_count":5,"t_max":80,"t_min":10}]
[{"device_id":"2","t_av":55,"t_count":1,"t_max":55,"t_min":55}]
Err cannot be passed to cli when any errors in open func of source
Add a TCP source that can be used for accepting data sending from other tools.
Currently, below message print when creating duplicated streams.
Stream command error: Create stream fails: Item demo already exists.
Below would be better?
Create stream fails: Item demo already exists.
Support the EdgeX message bus sink
Core command service sink support
Follow the steps in link, the analysis result will be sent out for 2 times.
It quit with following error stack trace.
bin/server
Serving Kuiper server on port 20498
panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x1 addr=0x28 pc=0x146bb2c]
goroutine 60 [running]:
engine/xstream/extensions.(*MQTTSource).Close(0xc000103880, 0x164d240, 0xc0001326c0, 0xc005528f38, 0x1)
/Users/rockyjin/Downloads/workspace/edge/src/kuiper/xstream/extensions/mqtt_source.go:150 +0x8c
engine/xstream/nodes.(*SourceNode).Open.func1(0xc00002bc40, 0x164d240, 0xc0001326c0, 0xc0055380e0, 0xc00a8bc000)
/Users/rockyjin/Downloads/workspace/edge/src/kuiper/xstream/nodes/source_node.go:45 +0x1a2
created by engine/xstream/nodes.(*SourceNode).Open
/Users/rockyjin/Downloads/workspace/edge/src/kuiper/xstream/nodes/source_node.go:28 +0x11c
bin/cli query
Connecting to 127.0.0.1:20498...
kuiper > select * from demo;
Failed to create query: fail to get stream demo, please check if stream is created.
Failed to create query: fail to get stream demo, please check if stream is created.
{
"id": "rule1",
"sql": "select rand() as rkey,randomnumber from demo where randomnumber > 80",
"actions": [
{
"rest": {
"url": "http://127.0.0.1:48060/api/v1/notification",
"method": "post",
"dataTemplate": "{\"slug\": \"notice-test-{{.rkey}}\", \"sender\": \"System Management\", \"category\": \"SECURITY\", \"severity\": \"CRITICAL\", \"content\": \"Number is too high: {{.randomnumber}}\", \"labels\": [ \"cool\", \"test\" ] }",
"sendSingle": true
}
}
]
}
Currently, if there are "
in the data template, then it need to be escaped, can we improve this?
Create a Docker image for Kuiper projects.
Update the version # in the conf file...
Now Kuiper need to support dynamically plugin loading, and CGO_ENABLED must be set to 1. But if set the CGO_ENABLED to 1, then cross-compiling is not supported. Need to change the build process?
During testing large scale rules with 8000 rules running and sending 1 message per 10 second per rule for about 2 hours, Kuiper panics for concurrent sink cache save with the below message:
goroutine 53 [running]:
runtime.throw(0xd4b8a4, 0x21)
/usr/local/go/src/runtime/panic.go:774 +0x72 fp=0xc0b33c3388 sp=0xc0b33c3358 pc=0x6af7d2
runtime.mapaccess2(0xc70b80, 0xc02e178cf0, 0xc0a28babc0, 0xc0a28babc0, 0xc0b33c3470)
/usr/local/go/src/runtime/map.go:470 +0x278 fp=0xc0b33c33d0 sp=0xc0b33c3388 pc=0x68ef48
reflect.mapaccess(0xc70b80, 0xc02e178cf0, 0xc0a28babc0, 0xd4378c)
/usr/local/go/src/runtime/map.go:1319 +0x3f fp=0xc0b33c3408 sp=0xc0b33c33d0 pc=0x69125f
reflect.Value.MapIndex(0xc70b80, 0xc005fe3130, 0x195, 0xc4e5e0, 0xc0a28babc0, 0x82, 0x7fe8d0b54fff, 0x400, 0x7fe8d0a85d00)
/usr/local/go/src/reflect/value.go:1176 +0x16d fp=0xc0b33c3480 sp=0xc0b33c3408 pc=0x7132cd
encoding/gob.(*Encoder).encodeMap(0xc0b0ed4be0, 0xc0bd8a2cc0, 0xc70b80, 0xc005fe3130, 0x195, 0xd65260, 0xd65268, 0x0, 0x0)
/usr/local/go/src/encoding/gob/encode.go:375 +0x1e8 fp=0xc0b33c3520 sp=0xc0b33c3480 pc=0x9d6e18
encoding/gob.encOpFor.func3(0xc00016a0c0, 0xc093721bc0, 0xc70b80, 0xc005fe3130, 0x195)
/usr/local/go/src/encoding/gob/encode.go:571 +0xa7 fp=0xc0b33c35a0 sp=0xc0b33c3520 pc=0x9e1d87
encoding/gob.(*Encoder).encodeStruct(0xc0b0ed4be0, 0xc0bd8a2cc0, 0xc049e5f320, 0xcb1aa0, 0xc005fe3130, 0x199)
/usr/local/go/src/encoding/gob/encode.go:328 +0x268 fp=0xc0b33c3678 sp=0xc0b33c35a0 pc=0x9d6508
encoding/gob.(*Encoder).encode(0xc0b0ed4be0, 0xc0bd8a2cc0, 0xca84a0, 0xc005fe3130, 0x16, 0xc000145e80)
/usr/local/go/src/encoding/gob/encode.go:701 +0x16f fp=0xc0b33c3720 sp=0xc0b33c3678 pc=0x9da09f
encoding/gob.(*Encoder).encodeInterface(0xc0b0ed4be0, 0xc0b0ed4c18, 0xc76040, 0xc0b6620400, 0x94)
/usr/local/go/src/encoding/gob/encode.go:419 +0x537 fp=0xc0b33c3860 sp=0xc0b33c3720 pc=0x9d7497
encoding/gob.encOpFor.func5(0xc000154540, 0xc093721b40, 0xc76040, 0xc0b6620400, 0x94)
/usr/local/go/src/encoding/gob/encode.go:589 +0x6a fp=0xc0b33c38a0 sp=0xc0b33c3860 pc=0x9e209a
encoding/gob.(*Encoder).encodeStruct(0xc0b0ed4be0, 0xc0b0ed4c18, 0xc0000974e0, 0xcbec00, 0xc0b6620400, 0x99)
/usr/local/go/src/encoding/gob/encode.go:328 +0x268 fp=0xc0b33c3978 sp=0xc0b33c38a0 pc=0x9d6508
encoding/gob.encOpFor.func4(0x0, 0xc093721b00, 0xcbec00, 0xc0b6620400, 0x99)
/usr/local/go/src/encoding/gob/encode.go:581 +0xac fp=0xc0b33c39d0 sp=0xc0b33c3978 pc=0x9e1fac
encoding/gob.encodeReflectValue(0xc093721b00, 0xcbec00, 0xc0b6620400, 0x99, 0xc000130590, 0x0)
/usr/local/go/src/encoding/gob/encode.go:363 +0xcf fp=0xc0b33c3a50 sp=0xc0b33c39d0 pc=0x9d6b5f
encoding/gob.(*Encoder).encodeMap(0xc0b0ed4be0, 0xc0b0ed4c18, 0xc70a00, 0xc000145e08, 0x195, 0xd65280, 0xc000130590, 0x0, 0x0)
/usr/local/go/src/encoding/gob/encode.go:375 +0x22e fp=0xc0b33c3af0 sp=0xc0b33c3a50 pc=0x9d6e5e
encoding/gob.encOpFor.func3(0xc0000fddd0, 0xc093721ac0, 0xc70a00, 0xc000145e08, 0x195)
/usr/local/go/src/encoding/gob/encode.go:571 +0xa7 fp=0xc0b33c3b70 sp=0xc0b33c3af0 pc=0x9e1d87
encoding/gob.(*Encoder).encodeSingle(0xc0b0ed4be0, 0xc0b0ed4c18, 0xc0000974c0, 0xc70a00, 0xc000145e08, 0x195)
/usr/local/go/src/encoding/gob/encode.go:301 +0x1c5 fp=0xc0b33c3c08 sp=0xc0b33c3b70 pc=0x9d61b5
encoding/gob.(*Encoder).encode(0xc0b0ed4be0, 0xc0b0ed4c18, 0xc2e820, 0xc000145e08, 0x16, 0xc000144180)
/usr/local/go/src/encoding/gob/encode.go:703 +0x1b7 fp=0xc0b33c3cb0 sp=0xc0b33c3c08 pc=0x9da0e7
encoding/gob.(*Encoder).EncodeValue(0xc0b0ed4be0, 0xc2e820, 0xc000145e08, 0x16, 0x0, 0x0)
/usr/local/go/src/encoding/gob/encoder.go:251 +0x3ce fp=0xc0b33c3d88 sp=0xc0b33c3cb0 pc=0x9db77e
encoding/gob.(*Encoder).Encode(0xc0b0ed4be0, 0xc2e820, 0xc000145e08, 0xc78740, 0xc0b659bec0)
/usr/local/go/src/encoding/gob/encoder.go:176 +0xa4 fp=0xc0b33c3dd8 sp=0xc0b33c3d88 pc=0x9db194
github.com/patrickmn/go-cache.(*cache).Save(0xc000145e00, 0xe0bc00, 0xc0b1be4f48, 0x0, 0x0)
/home/ubuntu/go/pkg/mod/github.com/patrickmn/[email protected]+incompatible/cache.go:975 +0x1aa fp=0xc0b33c3ee8 sp=0xc0b33c3dd8 pc=0x9ecf9a
github.com/patrickmn/go-cache.(*cache).SaveFile(0xc000145e00, 0xc0000790e0, 0x5c, 0x0, 0x0)
/home/ubuntu/go/pkg/mod/github.com/patrickmn/[email protected]+incompatible/cache.go:989 +0x86 fp=0xc0b33c3f48 sp=0xc0b33c3ee8 pc=0x9ed0b6
github.com/emqx/kuiper/common.(*SimpleKVStore).run(0xc000145e40)
/home/ubuntu/kuiper/common/kv.go:111 +0xe2 fp=0xc0b33c3fd8 sp=0xc0b33c3f48 pc=0xa21272
runtime.goexit()
/usr/local/go/src/runtime/asm_amd64.s:1357 +0x1 fp=0xc0b33c3fe0 sp=0xc0b33c3fd8 pc=0x6dccb1
created by github.com/emqx/kuiper/common.NewSimpleKVStore
/home/ubuntu/kuiper/common/kv.go:74 +0xd3
The aggregate functions, such as avg, count, max, min, sum
cannot be used in where
clause. The SQL parser need to validate it before execution.
Create an end-2-end automatic framework for functional testing.
GOOS=linux GOARCH=arm make pkg
Ultimate Packer for eXecutables
Copyright (C) 1996 - 2018
UPX 3.95 Markus Oberhumer, Laszlo Molnar & John Reiser Aug 26th 2018
File size Ratio Format Name
-------------------- ------ ----------- -----------
7151768 -> 2870752 40.14% linux/arm cli
Packed 1 file.
Ultimate Packer for eXecutables
Copyright (C) 1996 - 2018
UPX 3.95 Markus Oberhumer, Laszlo Molnar & John Reiser Aug 26th 2018
File size Ratio Format Name
-------------------- ------ ----------- -----------
8663112 -> 3474404 40.11% linux/arm server
Packed 1 file.
Build successfully
mv: rename engine.zip to ../_packages/engine.zip: No such file or directory
mv: rename engine.tar.gz to ../_packages/engine.tar.gz: No such file or directory
make: *** [pkg] Error 1
To support the Windows developers and runtime env, move the EdgeX plugins as internal shipped sources. Several questions to be figured out.
Support to create Schemaless schema.
bin/cli create stream demo '() WITH (FORMAT="JSON", DATASOURCE="demo" TYPE="edgex")'
Currently all of configuration items are specified in rule file, do we need move some of configuration into etc/sinks
folder?
Need to save calculation result into local storage in case of remote system is not available.
Sometimes the MQTT link in source/sink are dis-connected, need to re-connect to server if detecting any such issues.
FVT env for general Linux server
FVT env for Raspberry Pi env
A web sink for sending the result data to remote address.
Need an API for recording the gauge data of system...
I have filled the username and password into the mqtt_source.yaml, buf everytime it said to me the result of query is "Not Authorized" when I query from remote server.
To support deploy the EMQ X Edge & Kuiper at K3s, need a helm file.
Below is an example....
Use GET http://127.0.0.1:8080/rules/rule1/status
to get the rule status, but the response is not pure json.
Running with metrics:
should be removed.
Running with metrics:
{
"source_demo_0_records_in_total": 0,
"source_demo_0_records_out_total": 0,
"source_demo_0_exceptions_total": 0,
"source_demo_0_process_latency_ms": 0,
"source_demo_0_buffer_length": 0,
"source_demo_0_last_invocation": 0,
"op_preprocessor_demo_0_records_in_total": 0,
"op_preprocessor_demo_0_records_out_total": 0,
"op_preprocessor_demo_0_exceptions_total": 0,
"op_preprocessor_demo_0_process_latency_ms": 0,
"op_preprocessor_demo_0_buffer_length": 0,
"op_preprocessor_demo_0_last_invocation": 0,
"op_project_0_records_in_total": 0,
"op_project_0_records_out_total": 0,
"op_project_0_exceptions_total": 0,
"op_project_0_process_latency_ms": 0,
"op_project_0_buffer_length": 0,
"op_project_0_last_invocation": 0,
"sink_sink_log_0_records_in_total": 0,
"sink_sink_log_0_records_out_total": 0,
"sink_sink_log_0_exceptions_total": 0,
"sink_sink_log_0_process_latency_ms": 0,
"sink_sink_log_0_buffer_length": 0,
"sink_sink_log_0_last_invocation": 0,
"sink_sink_mqtt_0_records_in_total": 0,
"sink_sink_mqtt_0_records_out_total": 0,
"sink_sink_mqtt_0_exceptions_total": 0,
"sink_sink_mqtt_0_process_latency_ms": 0,
"sink_sink_mqtt_0_buffer_length": 0,
"sink_sink_mqtt_0_last_invocation": 0
}
Reduce the memory footprint to less than 10MB...
Develop test script for RestAPI interface
demo
demo
. The log displays rule is closing. But if type the command ./cli getstatus rule rule_name
, it prints running
.source > pre-processor > operators > sinks
in_total, out_total, exceptions_total, latency_ms, last_invocation
"kuiper_op_filter_0_last_invocation": "0001-01-01 00:00:00 +0000 UTC",
. If no message are processed, can we just set the value to an empty string?"kuiper_op_filter_0_last_invocation": "2019-12-25 13:36:50.656026 +0800 CST m=+2030.910846243"
. I would suggest to remove +0800 CST m=+2030.910846243
, and just keep 2019-12-25 13:36:50.656026
.It should not be started if the rule was already started.
{
"sql" : "create stream demo (temperature float, humidity bigint) WITH (FORMAT=\"JSON\", DATASOURCE=\"devices/+/messages\" )"
}
light
field does not existed in previous stream definition.{
"id": "rule1",
"sql": "SELECT light FROM demo",
"actions": [
{
"mqtt": {
"server": "tcp://${mqtt_srv}:1883",
"topic": "devices/result",
"qos": 1,
"clientId": "demo_001"
}
}
]
}
light
definition as in below.{
"sql" : "create stream demo (temperature float, humidity bigint, light bigint) WITH (FORMAT=\"JSON\", DATASOURCE=\"devices/+/messages\" )"
}
light
value still cannot be fetch.Benchmark result for max rule support number for Kuiper.
A declarative, efficient, and flexible JavaScript library for building user interfaces.
๐ Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. ๐๐๐
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google โค๏ธ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.