GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-28 14:07:20 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat:core subscription supports filtering error messages

Browse Source
This commit is contained in:
hxy7yx
2024-07-26 15:30:51 +08:00
parent 384676b522
commit 8e3ee3fde4
11 changed files with 300 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/check.yml
+1 -3
View File
@@ -90,9 +90,7 @@ jobs:
- name: setup dependencies
run: |
if [ "${{ matrix.plugin }}" = "mqtt" ] || [ "${{ matrix.plugin }}" = "azure" ]; then
sudo apt-get install -y mosquitto
fi
sudo apt-get install -y mosquitto
pip install -U pytest requests prometheus-client gmqtt pynng sniffio
sudo apt-get install -y socat
neuron.json
+2 -1
View File
@@ -3,5 +3,6 @@
"port": 7000,
"disable_auth": 0,
"syslog_host": "",
"syslog_port": 541
"syslog_port": 541,
"sub_filter_error": 0
}
src/adapter/driver/cache.c
+95 -1
View File
@@ -29,6 +29,8 @@
#include "cache.h"
extern bool sub_filter_err;
typedef struct {
char group[NEU_GROUP_NAME_LEN];
char tag[NEU_TAG_NAME_LEN];
@@ -39,6 +41,7 @@ struct elem {
bool changed;
neu_dvalue_t value;
neu_dvalue_t value_old;
neu_tag_meta_t metas[NEU_TAG_META_SIZE];
@@ -142,8 +145,91 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
HASH_FIND(hh, cache->table, &key, sizeof(tkey_t), elem);
if (elem != NULL) {
elem->timestamp = timestamp;
if (elem->value.type != value.type) {
if (sub_filter_err && value.type == NEU_TYPE_ERROR) {
goto error_not_report;
}
if ((!sub_filter_err && elem->value.type != value.type) ||
(sub_filter_err && elem->value.type != value.type &&
elem->value.type != NEU_TYPE_ERROR)) {
elem->changed = true;
} else if (sub_filter_err && elem->value.type != value.type &&
elem->value.type == NEU_TYPE_ERROR) {
switch (value.type) {
case NEU_TYPE_INT8:
case NEU_TYPE_UINT8:
case NEU_TYPE_INT16:
case NEU_TYPE_UINT16:
case NEU_TYPE_INT32:
case NEU_TYPE_UINT32:
case NEU_TYPE_INT64:
case NEU_TYPE_UINT64:
case NEU_TYPE_BIT:
case NEU_TYPE_BOOL:
case NEU_TYPE_STRING:
case NEU_TYPE_TIME:
case NEU_TYPE_DATA_AND_TIME:
case NEU_TYPE_WORD:
case NEU_TYPE_DWORD:
case NEU_TYPE_LWORD:
if (memcmp(&elem->value_old.value, &value.value,
sizeof(value.value)) != 0) {
elem->changed = true;
}
break;
case NEU_TYPE_BYTES:
if (elem->value_old.value.bytes.length !=
value.value.bytes.length) {
elem->changed = true;
} else {
if (memcpy(elem->value_old.value.bytes.bytes,
value.value.bytes.bytes,
value.value.bytes.length) != 0) {
elem->changed = true;
}
}
break;
case NEU_TYPE_PTR: {
if (elem->value_old.value.ptr.length !=
value.value.ptr.length) {
elem->changed = true;
} else {
if (memcmp(elem->value_old.value.ptr.ptr,
value.value.ptr.ptr,
value.value.ptr.length) != 0) {
elem->changed = true;
}
}
break;
}
case NEU_TYPE_FLOAT:
if (elem->value_old.precision == 0) {
elem->changed =
elem->value_old.value.f32 != value.value.f32;
} else {
if (fabs(elem->value_old.value.f32 - value.value.f32) >
pow(0.1, elem->value_old.precision)) {
elem->changed = true;
}
}
break;
case NEU_TYPE_DOUBLE:
if (elem->value_old.precision == 0) {
elem->changed =
elem->value_old.value.d64 != value.value.d64;
} else {
if (fabs(elem->value_old.value.d64 - value.value.d64) >
pow(0.1, elem->value_old.precision)) {
elem->changed = true;
}
}
break;
case NEU_TYPE_ERROR:
break;
}
} else {
switch (value.type) {
case NEU_TYPE_INT8:
@@ -218,6 +304,14 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
}
}
if (sub_filter_err && value.type != NEU_TYPE_ERROR) {
elem->value_old.type = value.type;
elem->value_old.value = value.value;
elem->value_old.precision = value.precision;
}
error_not_report:
if (change) {
elem->changed = true;
}
src/argparse.c
+29
View File
@@ -80,6 +80,7 @@ const char *usage_text =
" --plugin_dir <DIR> directory from which neuron loads plugin lib files\n"
" --syslog_host <HOST> syslog server host to which neuron will send logs\n"
" --syslog_port <PORT> syslog server port (default 541 if not provided)\n"
" --sub_filter_error The subscribe attribute only detects the last read value and does not report any error tags\n"
"\n";
// clang-format on
@@ -181,6 +182,19 @@ static inline int load_env(neu_cli_args_t *args, char **log_level_out,
}
}
char *sub_filter_e = getenv(NEU_ENV_SUB_FILTER_ERROR);
if (sub_filter_e != NULL) {
if (strcmp(sub_filter_e, "1") == 0) {
args->sub_filter_err = true;
} else if (strcmp(sub_filter_e, "0") == 0) {
args->sub_filter_err = false;
} else {
printf("neuron NEURON_SUB_FILTER_ERROR setting error!\n");
ret = -1;
break;
}
}
char *log_level = getenv(NEU_ENV_LOG_LEVEL);
if (log_level != NULL) {
if (*log_level_out != NULL) {
@@ -289,6 +303,7 @@ static inline int load_config_file(int argc, char *argv[],
{ .name = "port", .t = NEU_JSON_INT },
{ .name = "syslog_host", .t = NEU_JSON_STR },
{ .name = "syslog_port", .t = NEU_JSON_INT },
{ .name = "sub_filter_error", .t = NEU_JSON_INT },
};
resolve_config_file_path(argc, argv, long_options, opts, &config_file);
@@ -369,6 +384,16 @@ static inline int load_config_file(int argc, char *argv[],
break;
}
if (elems[5].v.val_int == 1) {
args->sub_filter_err = true;
} else if (elems[5].v.val_int == 0) {
args->sub_filter_err = false;
} else {
printf("config file %s sub_filter_error setting error!\n",
config_file);
break;
}
ret = 0;
} else {
printf("config file %s elems error! must had ip, port and "
@@ -424,6 +449,7 @@ void neu_cli_args_init(neu_cli_args_t *args, int argc, char *argv[])
{ "stop", no_argument, NULL, 's' },
{ "syslog_host", required_argument, NULL, 'S' },
{ "syslog_port", required_argument, NULL, 'P' },
{ "sub_filter_error", no_argument, NULL, 'f' },
{ NULL, 0, NULL, 0 },
};
@@ -517,6 +543,9 @@ void neu_cli_args_init(neu_cli_args_t *args, int argc, char *argv[])
}
args->syslog_host = strdup(optarg);
break;
case 'f':
args->sub_filter_err = true;
break;
case '?':
default:
usage();
src/argparse.h
+2
View File
@@ -35,6 +35,7 @@
#define NEU_ENV_PLUGIN_DIR "NEURON_PLUGIN_DIR"
#define NEU_ENV_SYSLOG_HOST "NEURON_SYSLOG_HOST"
#define NEU_ENV_SYSLOG_PORT "NEURON_SYSLOG_PORT"
#define NEU_ENV_SUB_FILTER_ERROR "NEURON_SUB_FILTER_ERROR"
#define NEURON_CONFIG_FNAME "./config/neuron.json"
@@ -64,6 +65,7 @@ typedef struct {
int port;
char * syslog_host;
uint16_t syslog_port;
bool sub_filter_err;
} neu_cli_args_t;
/** Parse command line arguments.
src/main.c
+3 -1
View File
@@ -39,6 +39,7 @@ static bool exit_flag = false;
static neu_manager_t *g_manager = NULL;
zlog_category_t * neuron = NULL;
bool disable_jwt = false;
bool sub_filter_err = false;
int default_log_level = ZLOG_LEVEL_NOTICE;
char host_port[32] = { 0 };
char g_status[32] = { 0 };
@@ -165,7 +166,8 @@ int main(int argc, char *argv[])
global_timestamp = neu_time_ms();
neu_cli_args_init(&args, argc, argv);
disable_jwt = args.disable_auth;
disable_jwt = args.disable_auth;
sub_filter_err = args.sub_filter_err;
snprintf(host_port, sizeof(host_port), "http://%s:%d", args.ip, args.port);
if (args.daemonized) {
tests/ft/login/neuron.json
+2 -1
View File
@@ -3,5 +3,6 @@
"port": 7000,
"disable_auth": 1,
"syslog_host": "",
"syslog_port": 541
"syslog_port": 541,
"sub_filter_error": 0
}
tests/ft/login/sub_filter_error.json
+8
View File
@@ -0,0 +1,8 @@
{
"ip": "0.0.0.0",
"port": 7000,
"disable_auth": 0,
"syslog_host": "",
"syslog_port": 541,
"sub_filter_error": 1
}
tests/ft/login/test_launch.py
+144 -1
View File
@@ -1,14 +1,157 @@
import pytest
import neuron.api as api
from neuron.config import *
from neuron.common import description
from neuron.common import (
description,
random_port,
)
import neuron.process as process
import neuron.config as config
from neuron.error import *
import shutil
import os
from neuron.mqtt import Mock
import time
mqtt_node = "mqtt"
upload_topic = "/neuron/test/sub_filter"
modbus_port = random_port()
modbus_node = "modbus_tcp"
hold_bit = [{"name": "hold_bit", "address": "1!400001.15",
"attribute": config.NEU_TAG_ATTRIBUTE_READ_SUBSCRIBE, "type": config.NEU_TYPE_BIT}]
hold_int16 = [{"name": "hold_int16", "address": "1!400001",
"attribute": config.NEU_TAG_ATTRIBUTE_RW_SUBSCRIBE, "type": config.NEU_TYPE_INT16}]
hold_uint16 = [{"name": "hold_uint16", "address": "1!400002",
"attribute": config.NEU_TAG_ATTRIBUTE_RW_SUBSCRIBE, "type": config.NEU_TYPE_UINT16}]
hold_int32 = [{"name": "hold_int32", "address": "1!400003",
"attribute": config.NEU_TAG_ATTRIBUTE_RW_SUBSCRIBE, "type": config.NEU_TYPE_INT32}]
hold_uint32 = [{"name": "hold_uint32", "address": "1!400015",
"attribute": config.NEU_TAG_ATTRIBUTE_RW_SUBSCRIBE, "type": config.NEU_TYPE_UINT32}]
hold_float = [{"name": "hold_float", "address": "1!400017",
"attribute": config.NEU_TAG_ATTRIBUTE_RW_SUBSCRIBE, "type": config.NEU_TYPE_FLOAT}]
hold_string = [{"name": "hold_string", "address": "1!400020.10",
"attribute": config.NEU_TAG_ATTRIBUTE_RW_SUBSCRIBE, "type": config.NEU_TYPE_STRING}]
@pytest.fixture(autouse=True, scope="class")
def mocker():
mock = Mock()
yield mock
mock.done()
class TestLaunch:
@description(
given="sub_filter_error(command line)",
when="start neuron",
then="should make sense",
)
def test_sub_filter_error_command(self, mocker):
process.remove_persistence()
p = process.NeuronProcess()
p.start_sub_filter_err()
api.add_node(node=mqtt_node, plugin=config.PLUGIN_MQTT)
api.mqtt_node_setting(node=mqtt_node, client_id=mocker.rand_id(), host="127.0.0.1", port=mocker.port)
api.add_node(node=modbus_node, plugin=config.PLUGIN_MODBUS_TCP)
api.modbus_tcp_node_setting(node=modbus_node, port=modbus_port)
api.add_group(node=modbus_node, group='group')
api.add_tags_check(node=modbus_node, group='group', tags=hold_bit)
api.add_tags_check(node=modbus_node, group='group', tags=hold_int16)
api.add_tags_check(node=modbus_node, group='group', tags=hold_uint16)
api.add_tags_check(node=modbus_node, group='group', tags=hold_int32)
api.add_tags_check(node=modbus_node, group='group', tags=hold_uint32)
api.add_tags_check(node=modbus_node, group='group', tags=hold_float)
api.add_tags_check(node=modbus_node, group='group', tags=hold_string)
api.subscribe_group(app=mqtt_node, driver=modbus_node, group='group', params={"topic":upload_topic})
msg = mocker.get(upload_topic, timeout=3)
assert msg is None
p_s = process.start_simulator(
["./modbus_simulator", "tcp", f"{modbus_port}", "ip_v4"]
)
time.sleep(5)
msg = mocker.get(upload_topic, timeout=3)
assert msg is None
p.stop()
process.stop_simulator(p_s)
@description(
given="sub_filter_error(environment variable)",
when="start neuron",
then="should make sense",
)
def test_sub_filter_error_environment(self, mocker):
os.environ['NEURON_SUB_FILTER_ERROR'] = '1'
p = process.NeuronProcess()
p.start()
msg = mocker.get(upload_topic, timeout=3)
assert msg is None
p_s = process.start_simulator(
["./modbus_simulator", "tcp", f"{modbus_port}", "ip_v4"]
)
time.sleep(5)
msg = mocker.get(upload_topic, timeout=3)
assert msg is None
del os.environ['NEURON_SUB_FILTER_ERROR']
p.stop()
process.stop_simulator(p_s)
@description(
given="sub_filter_error(configuration file)",
when="start neuron",
then="should make sense",
)
def test_sub_filter_error_file(self, mocker):
dst_file = './build/config/neuron.json'
dst_backup = dst_file + '_backup'
os.rename(dst_file, dst_backup)
try:
shutil.copy2('./tests/ft/login/sub_filter_error.json', dst_file)
except Exception as e:
if os.path.exists(dst_backup):
os.rename(dst_backup, dst_file)
return
p = process.NeuronProcess()
p.start()
msg = mocker.get(upload_topic, timeout=3)
assert msg is None
p_s = process.start_simulator(
["./modbus_simulator", "tcp", f"{modbus_port}", "ip_v4"]
)
time.sleep(5)
msg = mocker.get(upload_topic, timeout=3)
assert msg is None
p.stop()
process.stop_simulator(p_s)
os.remove(dst_file)
os.rename(dst_backup, dst_file)
@description(given="encironment variables 0, configuration file 0 , command line 1",
when="start neuron",
then="command line has the highest priority")
tests/ft/neuron/api.py
+3 -3
View File
@@ -375,10 +375,10 @@ def modbus_rtu_node_setting_base(node, port=502, connection_mode=0, transport_mo
"parity": parity, "baud": baud, "data": data, "address_base": base})
def mqtt_node_setting(node):
return node_setting(node, json={"client-id": "neuron_aBcDeF", "qos": 0, "format": 0,
def mqtt_node_setting(node, client_id="neuron_aBcDeF", host="broker.emqx.io", port=1883):
return node_setting(node, json={"client-id": client_id, "qos": 0, "format": 0,
"write-req-topic": f"/neuron/{node}/write/req", "write-resp-topic": f"/neuron/{node}/write/resp",
"offline-cache": False, "cache-sync-interval": 100, "host": "broker.emqx.io", "port": 1883, "username": "", "password": "", "ssl": False})
"offline-cache": False, "cache-sync-interval": 100, "host": host, "port": port, "username": "", "password": "", "ssl": False})
def get_nodes_disable_auth(type):
tests/ft/neuron/process.py
+11
View File
@@ -30,6 +30,9 @@ class NeuronProcess:
def start_debug(self):
self.p = start_neuron_debug(self.dir)
def start_sub_filter_err(self):
self.p = start_neuron_sub_filter_err(self.dir)
def start_neuron_disable_auth(dir='build/'):
command = ['./neuron', '--disable_auth']
process = subprocess.Popen(
@@ -46,6 +49,14 @@ def start_neuron_debug(dir='build/'):
assert process.poll() is None
return process
def start_neuron_sub_filter_err(dir='build/'):
command = ['./neuron', '--sub_filter_error']
process = subprocess.Popen(
command, stderr=subprocess.PIPE, cwd=dir)
time.sleep(1)
assert process.poll() is None
return process
def start_neuron(dir='build/'):
process = subprocess.Popen(
['./neuron'], stderr=subprocess.PIPE, cwd=dir)