GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-17 04:40:42 +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-23 14:00:01 +08:00
parent 8b8bb2a285
commit aa12e32274
5 changed files with 131 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
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 };
@@ -164,7 +165,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) {