From 8e3ee3fde4fa4cfb1326b1bd2abdac26b1c2f1ca Mon Sep 17 00:00:00 2001 From: hxy7yx <1595670487@qq.com> Date: Tue, 23 Jul 2024 14:00:01 +0800 Subject: [PATCH] feat:core subscription supports filtering error messages --- .github/workflows/check.yml | 4 +- neuron.json | 3 +- src/adapter/driver/cache.c | 96 +++++++++++++++++- src/argparse.c | 29 ++++++ src/argparse.h | 2 + src/main.c | 4 +- tests/ft/login/neuron.json | 3 +- tests/ft/login/sub_filter_error.json | 8 ++ tests/ft/login/test_launch.py | 145 ++++++++++++++++++++++++++- tests/ft/neuron/api.py | 6 +- tests/ft/neuron/process.py | 11 ++ 11 files changed, 300 insertions(+), 11 deletions(-) create mode 100644 tests/ft/login/sub_filter_error.json diff --git a/.github/workflows/check.yml b/.github/workflows/check.yml index a7487af0..5669d3bd 100644 --- a/.github/workflows/check.yml +++ b/.github/workflows/check.yml @@ -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 diff --git a/neuron.json b/neuron.json index 51b70c27..79621cb7 100644 --- a/neuron.json +++ b/neuron.json @@ -3,5 +3,6 @@ "port": 7000, "disable_auth": 0, "syslog_host": "", - "syslog_port": 541 + "syslog_port": 541, + "sub_filter_error": 0 } diff --git a/src/adapter/driver/cache.c b/src/adapter/driver/cache.c index ed9c76ef..8fbf1bef 100644 --- a/src/adapter/driver/cache.c +++ b/src/adapter/driver/cache.c @@ -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; } diff --git a/src/argparse.c b/src/argparse.c index 25be4d13..2cfc91e6 100644 --- a/src/argparse.c +++ b/src/argparse.c @@ -80,6 +80,7 @@ const char *usage_text = " --plugin_dir directory from which neuron loads plugin lib files\n" " --syslog_host syslog server host to which neuron will send logs\n" " --syslog_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(); diff --git a/src/argparse.h b/src/argparse.h index 8c0ab506..e844c164 100644 --- a/src/argparse.h +++ b/src/argparse.h @@ -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. diff --git a/src/main.c b/src/main.c index c16a4512..d5a4bf34 100644 --- a/src/main.c +++ b/src/main.c @@ -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) { diff --git a/tests/ft/login/neuron.json b/tests/ft/login/neuron.json index bf1c72b4..677cb000 100644 --- a/tests/ft/login/neuron.json +++ b/tests/ft/login/neuron.json @@ -3,5 +3,6 @@ "port": 7000, "disable_auth": 1, "syslog_host": "", - "syslog_port": 541 + "syslog_port": 541, + "sub_filter_error": 0 } diff --git a/tests/ft/login/sub_filter_error.json b/tests/ft/login/sub_filter_error.json new file mode 100644 index 00000000..366a372b --- /dev/null +++ b/tests/ft/login/sub_filter_error.json @@ -0,0 +1,8 @@ +{ + "ip": "0.0.0.0", + "port": 7000, + "disable_auth": 0, + "syslog_host": "", + "syslog_port": 541, + "sub_filter_error": 1 +} diff --git a/tests/ft/login/test_launch.py b/tests/ft/login/test_launch.py index 1e42211e..21c28dcb 100644 --- a/tests/ft/login/test_launch.py +++ b/tests/ft/login/test_launch.py @@ -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") diff --git a/tests/ft/neuron/api.py b/tests/ft/neuron/api.py index cfef5e69..27f6847e 100644 --- a/tests/ft/neuron/api.py +++ b/tests/ft/neuron/api.py @@ -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): diff --git a/tests/ft/neuron/process.py b/tests/ft/neuron/process.py index 98b603a4..7e2f2a39 100644 --- a/tests/ft/neuron/process.py +++ b/tests/ft/neuron/process.py @@ -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)