Not able to execute basic demo example on Kuiper rules engine with EdgeX

[AnilChoudhury-Eaton]

Hi,
I am using master branch of edgeX docker-compose script to run EdgeX services on RPI running ubuntu.
I have configured the MQTT configuration in docker-compose for 'rulesengine' service to filter a particular device resource events. I can see those events when i do subscribe to mqtt bus.
Later i created a default stream and a default rule but i can get following warning: "No readings are processed for the event, so ignore it."

Container configuration:

rulesengine:
container_name: edgex-kuiper
depends_on:
- database
- mqtt_broker
environment:
EDGEX__DEFAULT__PORT: 1883
EDGEX__DEFAULT__PROTOCOL: tcp
EDGEX__DEFAULT__SERVER: 192.168.0.105
# EDGEX__DEFAULT__TOPIC: 'edgex/events/device/#'
EDGEX__DEFAULT__TOPIC: 'edgex/events/device/Random-Integer-Device/Random-Integer-Device/Int16'
EDGEX__DEFAULT__TYPE: mqtt
EDGEX__DEFAULT__MESSAGETYPE: request
KUIPER__BASIC__CONSOLELOG: "true"
KUIPER__BASIC__RESTPORT: 59720
hostname: edgex-kuiper
image: emqx/kuiper:1.2.1-alpine
networks:
edgex-network: {}
ports:
- 0.0.0.0:59720:59720/tcp
read_only: true
security_opt:
- no-new-privileges:true
user: kuiper:kuiper
volumes:
- kuiper-data:/kuiper/data:z

Stream created:

ubuntu@ubuntu:~/edgex/edgex-compose$ curl -X POST http://192.168.0.105:59720/streams -H 'Content-Type: application/json' -d '{ "sql": "create stream demo() WITH (FORMAT="JSON", TYPE="edgex")"}'
Stream demo is created.

ubuntu@ubuntu:~/edgex/edgex-compose$

Rule created:

ubuntu@ubuntu:~/edgex/edgex-compose$ curl -X POST http://192.168.0.105:59720/rules \

-H 'Content-Type: application/json'
-d '{
"id": "rule1",
"sql": "SELECT * FROM demo",
"actions": [
{
"mqtt": {
"server": "tcp://broker.emqx.io:1883",
"topic": "result-anil-1000",
"clientId": "anil-edgex-kui[er-test-001"
}
},
{
"log":{}
}
]
}'
Rule rule1 was created successfully.

ubuntu@ubuntu:~/edgex/edgex-compose$

Container run log:

ubuntu@ubuntu:~/edgex/edgex-compose$ sudo docker-compose -f docker-compose-no-secty-with-ui-arm64.yml run rulesengine
WARNING: Some services (data) use the 'deploy' key, which will be ignored. Compose does not support 'deploy' configuration - use docker stack deploy to deploy to a swarm.
Starting edgex-mqtt-broker ... done
Starting edgex-redis ... done
/kuiper/etc/sources/edgex.yaml
Find env: EDGEX__DEFAULT__PORT=1883, start to handle it.
Find env: EDGEX__DEFAULT__MESSAGETYPE=request, start to handle it.
Find env: EDGEX__DEFAULT__TOPIC=edgex/events/device/Random-Integer-Device/Random-Integer-Device/Int16, start to handle it.
Find env: EDGEX__DEFAULT__PROTOCOL=tcp, start to handle it.
Find env: KUIPER_HOME=/kuiper, start to handle it.
not concerned env KUIPER_HOME=/kuiper, skip it.
Find env: EDGEX__DEFAULT__SERVER=192.168.0.105, start to handle it.
Find env: KUIPER__BASIC__CONSOLELOG=true, start to handle it.
Find env: EDGEX__DEFAULT__TYPE=mqtt, start to handle it.
Find env: KUIPER__BASIC__RESTPORT=59720, start to handle it.

Conf file kuiper:
basic:
consoleLog: true
debug: false
fileLog: true
ip: 0.0.0.0
maxAge: 72
pluginHosts: https://www.emqx.io/downloads
port: 20498
prometheus: false
prometheusPort: 20499
restIp: 0.0.0.0
restPort: 59720
rotateTime: 24
rule:
checkpointInterval: 300000
qos: 0
sendError: true
sink:
cacheThreshold: 10
cacheTriggerCount: 15
disableCache: true

/kuiper/etc/kuiper.yaml updated

Conf file edgex:
application_conf:
port: 5571
protocol: tcp
server: localhost
topic: application
default:
messagetype: request
port: 1883
protocol: tcp
server: 192.168.0.105
topic: edgex/events/device/Random-Integer-Device/Random-Integer-Device/Int16
type: mqtt
mqtt_conf:
optional:
ClientId: client1
port: 1883
protocol: tcp
server: 127.0.0.1
topic: events
type: mqtt

/kuiper/etc/sources/edgex.yaml updated
time="2021-08-25 11:12:03" level=info msg="db location is /kuiper/data" file="server/server.go:43"
time="2021-08-25 11:12:03" level=info msg="sourceMeta file : edgex.json" file="plugins/sourceMeta.go:135"
time="2021-08-25 11:12:03" level=info msg="sourceMeta file : httppull.json" file="plugins/sourceMeta.go:135"
time="2021-08-25 11:12:03" level=info msg="sourceMeta file : random.json" file="plugins/sourceMeta.go:135"
time="2021-08-25 11:12:03" level=info msg="sourceMeta file : zmq.json" file="plugins/sourceMeta.go:135"
time="2021-08-25 11:12:03" level=info msg="Loading metadata file for sink: edgex.json" file="plugins/sinkMeta.go:216"
time="2021-08-25 11:12:03" level=info msg="Loading metadata file for sink: file.json" file="plugins/sinkMeta.go:216"
time="2021-08-25 11:12:03" level=info msg="Loading metadata file for sink: influx.json" file="plugins/sinkMeta.go:216"
time="2021-08-25 11:12:03" level=info msg="Loading metadata file for sink: log.json" file="plugins/sinkMeta.go:216"
time="2021-08-25 11:12:03" level=info msg="Loading metadata file for sink: mqtt.json" file="plugins/sinkMeta.go:216"
time="2021-08-25 11:12:03" level=info msg="Loading metadata file for sink: nop.json" file="plugins/sinkMeta.go:216"
time="2021-08-25 11:12:03" level=info msg="Loading metadata file for sink: options.json" file="plugins/sinkMeta.go:216"
time="2021-08-25 11:12:03" level=info msg="Loading metadata file for sink: properties.json" file="plugins/sinkMeta.go:216"
time="2021-08-25 11:12:03" level=info msg="Loading metadata file for sink: rest.json" file="plugins/sinkMeta.go:216"
time="2021-08-25 11:12:03" level=info msg="Loading metadata file for sink: tdengine.json" file="plugins/sinkMeta.go:216"
time="2021-08-25 11:12:03" level=info msg="Loading metadata file for sink: zmq.json" file="plugins/sinkMeta.go:216"
time="2021-08-25 11:12:03" level=info msg="funcMeta file : accumulateWordCount.json" file="plugins/funcMeta.go:109"
time="2021-08-25 11:12:03" level=info msg="funcMeta file : countPlusOne.json" file="plugins/funcMeta.go:109"
time="2021-08-25 11:12:03" level=info msg="funcMeta file : echo.json" file="plugins/funcMeta.go:109"
time="2021-08-25 11:12:03" level=info msg="funcMeta file : geohash.json" file="plugins/funcMeta.go:109"
time="2021-08-25 11:12:03" level=info msg="funcMeta file : image.json" file="plugins/funcMeta.go:109"
time="2021-08-25 11:12:03" level=info msg="funcMeta file : internal.json" file="plugins/funcMeta.go:109"
time="2021-08-25 11:12:03" level=info msg="funcMeta file : labelImage.json" file="plugins/funcMeta.go:109"
time="2021-08-25 11:12:03" level=info msg="funcMeta file : windows.json" file="plugins/funcMeta.go:109"
time="2021-08-25 11:12:03" level=info msg="uiMsg file : en_US.ini" file="plugins/msg_util.go:37"
time="2021-08-25 11:12:03" level=info msg="uiMsg file : zh_CN.ini" file="plugins/msg_util.go:37"
time="2021-08-25 11:12:03" level=info msg="Starting rules" file="server/server.go:66"
time="2021-08-25 11:12:03" level=info msg="Init rule with options &{IsEventTime:false LateTol:1000 Concurrency:1 BufferLength:1024 SendMetaToSink:false SendError:true Qos:0 CheckpointInterval:300000}" file="planner/planner.go:26"
time="2021-08-25 11:12:03" level=info msg="fail to get stream EdgexStream, please check if stream is created" file="server/server.go:72"
time="2021-08-25 11:12:03" level=info msg="Rule test-demo was stoped." file="server/server.go:72"
time="2021-08-25 11:12:03" level=info msg="Init rule with options &{IsEventTime:false LateTol:1000 Concurrency:1 BufferLength:1024 SendMetaToSink:false SendError:true Qos:0 CheckpointInterval:300000}" file="planner/planner.go:26"
time="2021-08-25 11:12:03" level=info msg="Rule rule1 is replaced." file="processors/xsql_processor.go:368"
time="2021-08-25 11:12:03" level=info msg="Rule rule1 was started." file="server/server.go:72"
time="2021-08-25 11:12:03" level=info msg="Opening stream" file="xstream/streams.go:132" rule=rule1
time="2021-08-25 11:12:03" level=info msg="open sink node 1 instances" file="nodes/sink_node.go:164" rule=rule1
time="2021-08-25 11:12:03" level=info msg="Opening mqtt sink for rule rule1." file="sinks/mqtt_sink.go:134" rule=rule1
time="2021-08-25 11:12:03" level=info msg="Connect MQTT broker with username and password." file="sinks/mqtt_sink.go:153" rule=rule1
time="2021-08-25 11:12:03" level=info msg="Serving kuiper (version - 1.2.1) on port 20498, and restful api on http://0.0.0.0:59720. \n" file="server/server.go:144"
time="2021-08-25 11:12:03" level=info msg="open source node demo with option &{ JSON edgex false 0}" file="sinks/mqtt_sink.go:153" rule=rule1
time="2021-08-25 11:12:03" level=info msg="open sink node 1 instances" file="nodes/sink_node.go:164" rule=rule1
time="2021-08-25 11:12:03" level=info msg="Opening func collector" file="collectors/func.go:36" rule=rule1
Serving kuiper (version - 1.2.1) on port 20498, and restful api on http://0.0.0.0:59720.
time="2021-08-25 11:12:03" level=info msg="open source node 1 instances" file="nodes/source_node.go:86" rule=rule1
time="2021-08-25 11:12:03" level=info msg="Use configuration for edgex messagebus {{ 0 } {192.168.0.105 1883 tcp} mqtt map[]}" file="extensions/edgex_source.go:91"
time="2021-08-25 11:12:03" level=info msg="Start source demo instance 0 successfully" file="nodes/source_node.go:135" rule=rule1
time="2021-08-25 11:12:03" level=info msg="The connection to edgex messagebus is established successfully." file="extensions/edgex_source.go:102" rule=rule1
time="2021-08-25 11:12:03" level=info msg="Successfully subscribed to edgex messagebus topic edgex/events/device/Random-Integer-Device/Random-Integer-Device/Int16." file="extensions/edgex_source.go:111" rule=rule1
time="2021-08-25 11:12:04" level=info msg="The connection to server tcp://broker.emqx.io:1883 was established successfully" file="sinks/mqtt_sink.go:182" rule=rule1
time="2021-08-25 11:12:16" level=warning msg="No readings are processed for the event, so ignore it." file="extensions/edgex_source.go:177" rule=rule1
time="2021-08-25 11:12:31" level=warning msg="No readings are processed for the event, so ignore it." file="extensions/edgex_source.go:177" rule=rule1
time="2021-08-25 11:12:46" level=warning msg="No readings are processed for the event, so ignore it." file="extensions/edgex_source.go:177" rule=rule1
time="2021-08-25 11:13:01" level=warning msg="No readings are processed for the event, so ignore it." file="extensions/edgex_source.go:177" rule=rule1
time="2021-08-25 11:13:16" level=warning msg="No readings are processed for the event, so ignore it." file="extensions/edgex_source.go:177" rule=rule1
time="2021-08-25 11:13:31" level=warning msg="No readings are processed for the event, so ignore it." file="extensions/edgex_source.go:177" rule=rule1
time="2021-08-25 11:13:46" level=warning msg="No readings are processed for the event, so ignore it." file="extensions/edgex_source.go:177" rule=rule1
time="2021-08-25 11:14:01" level=warning msg="No readings are processed for the event, so ignore it." file="extensions/edgex_source.go:177" rule=rule1
time="2021-08-25 11:14:16" level=warning msg="No readings are processed for the event, so ignore it." file="extensions/edgex_source.go:177" rule=rule1

MQTT messages from device service on the bus which should get consumed by rules engine directly:
ubuntu@ubuntu:~/edgex/edgex-compose$ mosquitto_sub -t edgex/events/device/Random-Integer-Device/Random-Integer-Device/Int16 -d -q 1
Client mosq-OunGlQYGWAT7yr8K1a sending CONNECT
Client mosq-OunGlQYGWAT7yr8K1a received CONNACK (0)
Client mosq-OunGlQYGWAT7yr8K1a sending SUBSCRIBE (Mid: 1, Topic: edgex/events/device/Random-Integer-Device/Random-Integer-Device/Int16, QoS: 1, Options: 0x00)
Client mosq-OunGlQYGWAT7yr8K1a received SUBACK
Subscribed (mid: 1): 1
Client mosq-OunGlQYGWAT7yr8K1a received PUBLISH (d0, q0, r0, m0, 'edgex/events/device/Random-Integer-Device/Random-Integer-Device/Int16', ... (785 bytes))
{"ReceivedTopic":"","CorrelationID":"","Payload":"eyJhcGlWZXJzaW9uIjoidjIiLCJyZXF1ZXN0SWQiOiJhNWVhZjA2NC1lN2YwLTQzNGYtODNiOS01MDc2NWEwZDM3ODciLCJldmVudCI6eyJhcGlWZXJzaW9uIjoidjIiLCJpZCI6ImYxZGI2OGRkLWEyMDYtNDBhOS1iYzI1LWIxMTM0NjZiNGQwMSIsImRldmljZU5hbWUiOiJSYW5kb20tSW50ZWdlci1EZXZpY2UiLCJwcm9maWxlTmFtZSI6IlJhbmRvbS1JbnRlZ2VyLURldmljZSIsInNvdXJjZU5hbWUiOiJJbnQxNiIsIm9yaWdpbiI6MTYyOTg5MDU5NjczMDEzNDk1NiwicmVhZGluZ3MiOlt7ImlkIjoiNGI0NDAxYjEtNzlhOS00MGNkLTg0MTctNjJjNTY5NGFjODk3Iiwib3JpZ2luIjoxNjI5ODkwNTk2NzMwMTM0OTU2LCJkZXZpY2VOYW1lIjoiUmFuZG9tLUludGVnZXItRGV2aWNlIiwicmVzb3VyY2VOYW1lIjoiSW50MTYiLCJwcm9maWxlTmFtZSI6IlJhbmRvbS1JbnRlZ2VyLURldmljZSIsInZhbHVlVHlwZSI6IkludDE2IiwiYmluYXJ5VmFsdWUiOm51bGwsIm1lZGlhVHlwZSI6IiIsInZhbHVlIjoiMTc5NjgifV19fQ==","ContentType":"application/json"}

[ngjaying]

messageType property is supported since 1.3.0. I found you are using 1.2.1. Please use image: lfedge/ekuiper:1.3.0-alpine

[AnilChoudhury-Eaton]

messageType property is supported since 1.3.0. I found you are using 1.2.1. Please use image: lfedge/ekuiper:1.3.0-alpine

Thank you. This worked.