Some questions about the eKuiper+Edgex

[littletiger123]

Please describe your problem in detail, if necessary, you can upload the log file through the attachment:
I have tried the quick start and it succeeds. But I read the edgex_rule_engine_tutorial.md which is about the eKuiper and Edgex. I am a new one so I have several questions about it

1. Who create the data? Is the virtual device and where I can find it ?
2. I enter the docker container, and I found that EDGEX__DEFAULT__SERVER=edgex-redis which means we do not use the MQTT server? Also, I change the **CONF_KEY ** and set it mqtt_conf when creating the steam, but it may not have an effect. So I want to know which yaml is used? /kuiper/etc/mqtt_source.yaml or /kuiper/erc/sources/edgex.yaml. And may you exlpain the different because I see the MQTT config occur in both YAML.

application_conf:
    port: 5571
    protocol: tcp
    server: localhost
    topic: application
default:
    optional:
        Password: y3sm8JQBDjAAb3IpEZzxUddzgq/aUMQBhJFIhaAMEwQl
        Username: redis5
    port: 6379
    protocol: redis
    server: edgex-redis
    serviceServer: http://localhost:48080
    topic: rules-events
    type: redis
mqtt_conf:
    optional:
        ClientId: client1
    port: 1883
    protocol: tcp
    server: 127.0.0.1
    topic: events
    type: mqtt

3. I changed the yaml file and I want to know how to apply the new yaml to the service?
   Thanks a lot.

[ngjaying]

@littletiger123

1. The data are emitted by device-virtual. Check https://docs.edgexfoundry.org/1.2/getting-started/Ch-GettingStartedUsers/#verify-edgex-foundry-running, there is a edgex-device-virtual service.
2. Use edgex.yaml. Edgex source hides the internal message bus from eKuiper. The mqtt.yaml is used when you connect to a general/normal mqtt broker. You can modify the type property of edgex.yaml to change the underlying edgex message bus type. However, you must firstly change your edgex configuration and make sure the config in edgex.yaml matches your edgex instance.
3. Restart the rule will pick up the new source config.

[littletiger123]

@ngjaying
Sorry for ask you again.
The first time I erter the edgex-kuiper container, I thought it used the default conf in the edgex.yaml. While after I exit the docker and exec it again.

I enter the command

/kuiper $ bin/kuiper query
Connecting to 127.0.0.1:20498... 
kuiper > select * from demo
Query was submit successfully.
kuiper > found error when connecting to tcp://127.0.0.1:1883: network Error : dial tcp 127.0.0.1:1883: connect: connection refused

After I edit the mqtt_source.yaml, the error disappear, so I think it worked according to the mqtt_source.yaml rather than the edgex.yaml
What should I do to let the service according to the edgex.yaml?

[ngjaying]

What's the definition of stream demo? If the type is mqtt or unsepcified, it will be a mqtt stream, thus mqtt_source.yaml will be used. To use edgex source, you'll need to specify type="edgex" in your stream definition.

{
  "sql" : "create stream demo () WITH (FORMAT=\"JSON\", TYPE=\"edgex\")"
}

[littletiger123]

Yes, it works.
Now I want to change the device-virtual service to let the data be sent to the mqtt broker I set on my own.
May I change the /res/configuration.toml like this in the edgex-device-virtual docker? But the file is read-only, may I need to change the environment variable? It will be better if you have docs about this.

[MessageQueue]           
Enabled = true                                          
Protocol = 'tcp'                                                 
Host = '192.168.100.201'                                                 
Port = 1883
Type = 'mqtt'                                                     
# AuthMode = 'usernamepassword'  # required for redis messagebus (secure or insecure).
# SecretName = "redisdb"                                                              
PublishTopicPrefix = 'edgex/events/device' # /<device-profile-name>/<device-name>/<source-name> will be added to this Publish Topic prefix
  [MessageQueue.Optional]                                                                                                                 
  # Default MQTT Specific options that need to be here to enable environment variable overrides of them                                   
  # Client Identifiers                                                                                                                    
  ClientId = "device-virtual"                                                                                                             
  # Connection information                                                                                                                
  Qos = "0" # Quality of Sevice values are 0 (At most once), 1 (At least once) or 2 (Exactly once)                                        
  KeepAlive = "10" # Seconds (must be 2 or greater)                                                                                       
  Retained = "false"                                                                                                                      
  AutoReconnect = "true"                                                                                                                  
  ConnectTimeout = "5" # Seconds                                                                                                          
  SkipCertVerify = "false" # Only used if Cert/Key file or Cert/Key PEMblock are specified                                                

BTW, the settings will be also changed in the edgex.yaml in edgex-kuiper container.

[ngjaying]

That's not part of eKuiper actually. I think edgeX should have docs about that. Could you try to search their doc and ask in the edgex community? Thanks

[littletiger123]

Supplement for the question before, I never used the CONF_KEY="mqtt_conf" when I created the stream. So if it used the edgex.yaml, it will apply the default settings in the edgex.yaml. However, it did not.

[ngjaying]

What's your stream def? How do you identiy default setting is not used? Did you use edgex docker compose which defines env var and will override the edgex.yaml conf?

[littletiger123]

My target is to use the MQTT broker I created as the messagebus. And I only want to get theRandom-Integer-Devicedata.

First，I change the rulesengine:

   environment:
      API_GATEWAY_HOST: edgex-kong
      API_GATEWAY_STATUS_PORT: '8100'
      EDGEX__DEFAULT__PORT: 6379
      EDGEX__DEFAULT__PROTOCOL: redis
      EDGEX__DEFAULT__SERVER: edgex-redis
      EDGEX__DEFAULT__TOPIC: edgex/events/#/Random-Integer-Device/#
      EDGEX__DEFAULT__MESSAGETYPE: request
      EDGEX__DEFAULT__TYPE: mqtt
      KUIPER__BASIC__CONSOLELOG: "true"
      KUIPER__BASIC__RESTPORT: 59720
      PROXY_SETUP_HOST: edgex-security-proxy-setup
      STAGEGATE_BOOTSTRAPPER_HOST: edgex-security-bootstrapper
      STAGEGATE_BOOTSTRAPPER_STARTPORT: '54321'
      STAGEGATE_DATABASE_HOST: edgex-redis
      STAGEGATE_DATABASE_PORT: '6379'
      STAGEGATE_DATABASE_READYPORT: '6379'
      STAGEGATE_KONGDB_HOST: edgex-kong-db
      STAGEGATE_KONGDB_PORT: '5432'
      STAGEGATE_KONGDB_READYPORT: '54325'
      STAGEGATE_READY_TORUNPORT: '54329'
      STAGEGATE_REGISTRY_HOST: edgex-core-consul
      STAGEGATE_REGISTRY_PORT: '8500'
      STAGEGATE_REGISTRY_READYPORT: '54324'
      STAGEGATE_SECRETSTORESETUP_HOST: edgex-security-secretstore-setup
      STAGEGATE_SECRETSTORESETUP_TOKENS_READYPORT: '54322'
      STAGEGATE_WAITFOR_TIMEOUT: 60s

I added

      EDGEX__DEFAULT__TOPIC: edgex/events/#/Random-Integer-Device/#
      EDGEX__DEFAULT__MESSAGETYPE: request
      EDGEX__DEFAULT__TYPE: mqtt

which means the messagebus only receive the data from Random-Integer-Device
However, I think it may not be added to the env?

docker inspect edgex-app-rules-engine

            "Env": [
                "API_GATEWAY_HOST=edgex-kong",
                "API_GATEWAY_STATUS_PORT=8100",
                "CLIENTS_CORE_COMMAND_HOST=edgex-core-command",
                "CLIENTS_CORE_DATA_HOST=edgex-core-data",
                "CLIENTS_CORE_METADATA_HOST=edgex-core-metadata",
                "CLIENTS_SUPPORT_NOTIFICATIONS_HOST=edgex-support-notifications",
                "CLIENTS_SUPPORT_SCHEDULER_HOST=edgex-support-scheduler",
                "DATABASES_PRIMARY_HOST=edgex-redis",
                "EDGEX_PROFILE=rules-engine",
                "EDGEX_SECURITY_SECRET_STORE=true",
                "MESSAGEQUEUE_HOST=edgex-redis",
                "PROXY_SETUP_HOST=edgex-security-proxy-setup",
                "REGISTRY_HOST=edgex-core-consul",
                "SECRETSTORE_HOST=edgex-vault",
                "SECRETSTORE_PORT=8200",
                "SERVICE_HOST=edgex-app-rules-engine",
                "STAGEGATE_BOOTSTRAPPER_HOST=edgex-security-bootstrapper",
                "STAGEGATE_BOOTSTRAPPER_STARTPORT=54321",
                "STAGEGATE_DATABASE_HOST=edgex-redis",
                "STAGEGATE_DATABASE_PORT=6379",
                "STAGEGATE_DATABASE_READYPORT=6379",
                "STAGEGATE_KONGDB_HOST=edgex-kong-db",
                "STAGEGATE_KONGDB_PORT=5432",
                "STAGEGATE_KONGDB_READYPORT=54325",
                "STAGEGATE_READY_TORUNPORT=54329",
                "STAGEGATE_REGISTRY_HOST=edgex-core-consul",
                "STAGEGATE_REGISTRY_PORT=8500",
                "STAGEGATE_REGISTRY_READYPORT=54324",
                "STAGEGATE_SECRETSTORESETUP_HOST=edgex-security-secretstore-setup",
                "STAGEGATE_SECRETSTORESETUP_TOKENS_READYPORT=54322",
                "STAGEGATE_WAITFOR_TIMEOUT=60s",
                "TRIGGER_EDGEXMESSAGEBUS_PUBLISHHOST_HOST=edgex-redis",
                "TRIGGER_EDGEXMESSAGEBUS_SUBSCRIBEHOST_HOST=edgex-redis",
                "PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
            ],

Also, I changed the edgex.yaml in the edgex-kuiper container:

mqtt_conf:
    optional:
        ClientId: client1
    port: 1883
    protocol: tcp
    server: 192.168.100.201
    topic: edgex/events
    type: mqtt

Create the stream:

bin/kuiper create stream demo '() WITH (FORMAT="JSON", CONF_KEY="mqtt_conf", TYPE="edgex")'

I used` mosquitto_sub -h slave201 -t edgex/events/* to check and I got nothing.

/kuiper $ bin/kuiper getstatus rule rule1
Connecting to 127.0.0.1:20498... 
{
  "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_us": 0,
  "source_demo_0_buffer_length": 0,
  "source_demo_0_last_invocation": 0,
  "op_1_preprocessor_demo_0_records_in_total": 0,
  "op_1_preprocessor_demo_0_records_out_total": 0,
  "op_1_preprocessor_demo_0_exceptions_total": 0,
  "op_1_preprocessor_demo_0_process_latency_us": 0,
  "op_1_preprocessor_demo_0_buffer_length": 0,
  "op_1_preprocessor_demo_0_last_invocation": 0,
  "op_2_project_0_records_in_total": 0,
  "op_2_project_0_records_out_total": 0,
  "op_2_project_0_exceptions_total": 0,
  "op_2_project_0_process_latency_us": 0,
  "op_2_project_0_buffer_length": 0,
  "op_2_project_0_last_invocation": 0,
  "sink_mqtt_0_0_records_in_total": 0,
  "sink_mqtt_0_0_records_out_total": 0,
  "sink_mqtt_0_0_exceptions_total": 0,
  "sink_mqtt_0_0_process_latency_us": 0,
  "sink_mqtt_0_0_buffer_length": 0,
  "sink_mqtt_0_0_last_invocation": 0,
  "sink_log_1_0_records_in_total": 0,
  "sink_log_1_0_records_out_total": 0,
  "sink_log_1_0_exceptions_total": 0,
  "sink_log_1_0_process_latency_us": 0,
  "sink_log_1_0_buffer_length": 0,
  "sink_log_1_0_last_invocation": 0
}

Is the data from Random-Integer-Device sent to the messagebus? I saw the data may not sent to the stream.
Thanks a lot.

Also, I want to ask a question about the topic.
The topic of data from the device is edgex/events/#, why not it will be sent to the stream as the topic is rule-event? I found no docs about it , may you briefly explain it?

[littletiger123]

That's not part of eKuiper actually. I think edgeX should have docs about that. Could you try to search their doc and ask in the edgex community? Thanks

Okay, thanks.