Merge pull request #1319 from lixiumei123/v2.4

version:2.4.8

.github/workflows/ci.yml

@@ -8,8 +8,8 @@ on:
       - published
 
 env:
-  VERSION: 2.4.5
-  UI_VERSION: 2.4.3
+  VERSION: 2.4.8
+  UI_VERSION: 2.4.6
   BRANCH: v2.4
 
 jobs:

.gitignore

@@ -16,4 +16,7 @@ tmp
 *.deb
 *.rpm
 neuron-sdk
-.vscode
+.vscode
+log.html
+output.xml
+report.html

CMakeLists.txt

@@ -32,7 +32,7 @@ if(NOT DISABLE_WERROR)
 endif()
 
 set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Wall -Wextra -g")
-set(CMAKE_C_FLAGS_RELEASE "${CMAKE_C_FLAGS} -O1")
+set(CMAKE_C_FLAGS_RELEASE "${CMAKE_C_FLAGS} -O0")
 
 if(NOT DISABLE_ASAN)
   set(CMAKE_C_FLAGS_DEBUG "${CMAKE_C_FLAGS} -fsanitize=address")

include/neuron/errcodes.h

@@ -133,7 +133,7 @@ typedef enum {
     NEU_ERR_MQTT_SUSPENDED                      = 4015,
     NEU_ERR_MQTT_PUBLISH_OVER_LENGTH            = 4016,
 
-    NEU_ERR_FILE_TOO_LONG      = 4100,
+    NEU_ERR_STRING_TOO_LONG    = 4100,
     NEU_ERR_FILE_OPEN_FAILURE  = 4101,
     NEU_ERR_FILE_READ_FAILURE  = 4102,
     NEU_ERR_FILE_WRITE_FAILURE = 4103,

include/neuron/type.h

@@ -28,6 +28,7 @@ extern "C" {
 #include <netinet/in.h>
 #include <stdbool.h>
 #include <stdint.h>
+#include <stdio.h>
 
 typedef enum {
     NEU_TYPE_INT8   = 1,
@@ -109,7 +110,80 @@ typedef union {
     double   d64;
     char     str[NEU_VALUE_SIZE];
     uint8_t  bytes[NEU_VALUE_SIZE];
-} neu_value_u;
+} __attribute__((packed)) neu_value_u;
+
+static inline char *neu_value_str(neu_type_e type, neu_value_u value)
+{
+    static __thread char str[32] = { 0 };
+
+    switch (type) {
+    case NEU_TYPE_INT8:
+        snprintf(str, sizeof(str), "type: %s, value: %i", neu_type_string(type),
+                 value.i8);
+        break;
+    case NEU_TYPE_UINT8:
+        snprintf(str, sizeof(str), "type: %s, value: %u", neu_type_string(type),
+                 value.u8);
+        break;
+    case NEU_TYPE_INT16:
+        snprintf(str, sizeof(str), "type: %s, value: %i", neu_type_string(type),
+                 value.i16);
+        break;
+    case NEU_TYPE_WORD:
+    case NEU_TYPE_UINT16:
+        snprintf(str, sizeof(str), "type: %s, value: %u", neu_type_string(type),
+                 value.u16);
+        break;
+    case NEU_TYPE_INT32:
+        snprintf(str, sizeof(str), "type: %s, value: %i", neu_type_string(type),
+                 value.i32);
+        break;
+    case NEU_TYPE_DWORD:
+    case NEU_TYPE_UINT32:
+        snprintf(str, sizeof(str), "type: %s, value: %u", neu_type_string(type),
+                 value.u32);
+        break;
+    case NEU_TYPE_INT64: {
+        long long int i = value.i64;
+        snprintf(str, sizeof(str), "type: %s, value: %lld",
+                 neu_type_string(type), i);
+        break;
+    }
+    case NEU_TYPE_LWORD:
+    case NEU_TYPE_UINT64: {
+        long long unsigned int i = value.u64;
+        snprintf(str, sizeof(str), "type: %s, value: %llu",
+                 neu_type_string(type), i);
+        break;
+    }
+    case NEU_TYPE_FLOAT:
+        snprintf(str, sizeof(str), "type: %s, value: %f", neu_type_string(type),
+                 value.f32);
+        break;
+    case NEU_TYPE_DOUBLE:
+        snprintf(str, sizeof(str), "type: %s, value: %f", neu_type_string(type),
+                 value.d64);
+        break;
+    case NEU_TYPE_BIT:
+        snprintf(str, sizeof(str), "type: %s, value: %u", neu_type_string(type),
+                 value.u8);
+        break;
+    case NEU_TYPE_BOOL:
+        snprintf(str, sizeof(str), "type: %s, value: %u", neu_type_string(type),
+                 value.boolean);
+        break;
+    case NEU_TYPE_STRING: {
+        snprintf(str, sizeof(str), "type: %s, value: %c%c%c",
+                 neu_type_string(type), value.str[0], value.str[1],
+                 value.str[2]);
+        break;
+    }
+    default:
+        break;
+    }
+
+    return str;
+}
 
 typedef struct {
     neu_type_e  type;
@@ -443,6 +517,134 @@ static inline void neu_htons_p(uint16_t *pu)
     neu_ntohs_p(pu);
 }
 
+static inline uint16_t neu_get_u16(uint8_t *bytes)
+{
+    uint16_t ret = 0;
+    uint8_t *t   = NULL;
+
+    t = (uint8_t *) &ret;
+
+    t[0] = bytes[0];
+    t[1] = bytes[1];
+
+    return ret;
+}
+
+static inline int16_t neu_get_i16(uint8_t *bytes)
+{
+    int16_t  ret = 0;
+    uint8_t *t   = NULL;
+
+    t = (uint8_t *) &ret;
+
+    t[0] = bytes[0];
+    t[1] = bytes[1];
+
+    return ret;
+}
+
+static inline uint32_t neu_get_u32(uint8_t *bytes)
+{
+    uint32_t ret = 0;
+    uint8_t *t   = NULL;
+
+    t = (uint8_t *) &ret;
+
+    t[0] = bytes[0];
+    t[1] = bytes[1];
+    t[2] = bytes[2];
+    t[3] = bytes[3];
+
+    return ret;
+}
+
+static inline int32_t neu_get_i32(uint8_t *bytes)
+{
+    int32_t  ret = 0;
+    uint8_t *t   = NULL;
+
+    t = (uint8_t *) &ret;
+
+    t[0] = bytes[0];
+    t[1] = bytes[1];
+    t[2] = bytes[2];
+    t[3] = bytes[3];
+
+    return ret;
+}
+
+static inline float neu_get_f32(uint8_t *bytes)
+{
+    float    ret = 0;
+    uint8_t *t   = NULL;
+
+    t = (uint8_t *) &ret;
+
+    t[0] = bytes[0];
+    t[1] = bytes[1];
+    t[2] = bytes[2];
+    t[3] = bytes[3];
+
+    return ret;
+}
+
+static inline uint64_t neu_get_u64(uint8_t *bytes)
+{
+    uint64_t ret = 0;
+    uint8_t *t   = NULL;
+
+    t = (uint8_t *) &ret;
+
+    t[0] = bytes[0];
+    t[1] = bytes[1];
+    t[2] = bytes[2];
+    t[3] = bytes[3];
+    t[4] = bytes[4];
+    t[5] = bytes[5];
+    t[6] = bytes[6];
+    t[7] = bytes[7];
+
+    return ret;
+}
+
+static inline int64_t neu_get_i64(uint8_t *bytes)
+{
+    int64_t  ret = 0;
+    uint8_t *t   = NULL;
+
+    t = (uint8_t *) &ret;
+
+    t[0] = bytes[0];
+    t[1] = bytes[1];
+    t[2] = bytes[2];
+    t[3] = bytes[3];
+    t[4] = bytes[4];
+    t[5] = bytes[5];
+    t[6] = bytes[6];
+    t[7] = bytes[7];
+
+    return ret;
+}
+
+static inline double neu_get_f64(uint8_t *bytes)
+{
+    double   ret = 0;
+    uint8_t *t   = NULL;
+
+    t = (uint8_t *) &ret;
+
+    t[0] = bytes[0];
+    t[1] = bytes[1];
+    t[2] = bytes[2];
+    t[3] = bytes[3];
+    t[4] = bytes[4];
+    t[5] = bytes[5];
+    t[6] = bytes[6];
+    t[7] = bytes[7];
+
+    return ret;
+}
+
 #ifdef __cplusplus
 }
 #endif

include/neuron/utils/protocol_buf.h

@@ -192,4 +192,16 @@ neu_protocol_unpack_buf_used_size(neu_protocol_unpack_buf_t *buf)
     return buf->offset;
 }
 
+inline static void
+neu_protocol_unpack_buf_use_all(neu_protocol_unpack_buf_t *buf)
+{
+    buf->offset = buf->size;
+}
+
+inline static uint16_t
+neu_protocol_unpack_buf_size(neu_protocol_unpack_buf_t *buf)
+{
+    return buf->size;
+}
+
 #endif

plugins/ekuiper/plugin_ekuiper.c

@@ -34,8 +34,8 @@ static neu_plugin_t *ekuiper_plugin_open(void)
 
     neu_plugin_common_init(&plugin->common);
 
-    zlog_info(neuron, "success to create plugin: %s",
-              neu_plugin_module.module_name);
+    zlog_notice(neuron, "success to create plugin: %s",
+                neu_plugin_module.module_name);
     return plugin;
 }
 
@@ -44,8 +44,8 @@ static int ekuiper_plugin_close(neu_plugin_t *plugin)
     int rv = 0;
 
     free(plugin);
-    zlog_info(neuron, "success to free plugin: %s",
-              neu_plugin_module.module_name);
+    zlog_notice(neuron, "success to free plugin: %s",
+                neu_plugin_module.module_name);
     return rv;
 }
 
@@ -91,7 +91,7 @@ static int ekuiper_plugin_init(neu_plugin_t *plugin)
 
     plugin->recv_aio = recv_aio;
 
-    plog_info(plugin, "plugin initialized");
+    plog_notice(plugin, "plugin initialized");
     return rv;
 }
 
@@ -104,7 +104,7 @@ static int ekuiper_plugin_uninit(neu_plugin_t *plugin)
     free(plugin->host);
     free(plugin->url);
 
-    plog_info(plugin, "plugin uninitialized");
+    plog_notice(plugin, "plugin uninitialized");
     return rv;
 }
 
@@ -172,7 +172,7 @@ static int parse_config(neu_plugin_t *plugin, const char *setting,
     *host_p = host.v.val_str;
     *port_p = port.v.val_int;
 
-    plog_info(plugin, "config host:%s port:%" PRIu16, *host_p, *port_p);
+    plog_notice(plugin, "config host:%s port:%" PRIu16, *host_p, *port_p);
 
     return 0;
 
@@ -201,7 +201,7 @@ static int ekuiper_plugin_config(neu_plugin_t *plugin, const char *setting)
         goto error;
     }
 
-    plog_info(plugin, "config success");
+    plog_notice(plugin, "config success");
 
     free(plugin->host);
     plugin->host = host;
@@ -220,7 +220,7 @@ static int ekuiper_plugin_config(neu_plugin_t *plugin, const char *setting)
 error:
     free(url);
     free(host);
-    plog_info(plugin, "config failure");
+    plog_error(plugin, "config failure");
     return rv;
 }
 
@@ -239,7 +239,7 @@ static int ekuiper_plugin_request(neu_plugin_t *      plugin,
     switch (header->type) {
     case NEU_RESP_ERROR: {
         neu_resp_error_t *error = (neu_resp_error_t *) data;
-        plog_info(plugin, "receive resp errcode: %d", error->error);
+        plog_debug(plugin, "receive resp errcode: %d", error->error);
         break;
     }
     case NEU_REQRESP_TRANS_DATA: {
@@ -259,7 +259,7 @@ static int ekuiper_plugin_request(neu_plugin_t *      plugin,
     case NEU_REQ_UNSUBSCRIBE_GROUP:
         break;
     default:
-        plog_notice(plugin, "unsupported request type: %d", header->type);
+        plog_warn(plugin, "unsupported request type: %d", header->type);
         break;
     }
 

plugins/file/file.json

@@ -15,7 +15,8 @@
         "type": "int",
         "default": 512,
         "valid": {
-            "min": 1
+            "min": 1,
+            "max": 512
         }
     }
 }

plugins/file/file_plugin.c

@@ -74,6 +74,9 @@ static neu_plugin_t *driver_open(void)
 
 static int driver_close(neu_plugin_t *plugin)
 {
+    plog_notice(plugin, "close and free `%s` plugin success",
+                plugin->common.name);
+
     free(plugin);
 
     return 0;
@@ -83,13 +86,17 @@ static int driver_init(neu_plugin_t *plugin)
 {
     plugin->events = neu_event_new();
 
+    plog_notice(plugin, "initialize `%s` plugin success", plugin->common.name);
+
     return 0;
 }
 
 static int driver_uninit(neu_plugin_t *plugin)
 {
     neu_event_close(plugin->events);
-    plog_info(plugin, "node; file uininit ok");
+
+    plog_notice(plugin, "uninitialize `%s` plugin success",
+                plugin->common.name);
 
     return 0;
 }
@@ -98,6 +105,8 @@ static int driver_start(neu_plugin_t *plugin)
 {
     plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED;
 
+    plog_notice(plugin, "start `%s` plugin success", plugin->common.name);
+
     return 0;
 }
 
@@ -105,6 +114,8 @@ static int driver_stop(neu_plugin_t *plugin)
 {
     plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
 
+    plog_notice(plugin, "stop `%s` plugin success", plugin->common.name);
+
     return 0;
 }
 
@@ -115,7 +126,7 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
 
     int ret = neu_parse_param((char *) config, &err_param, 1, &file_length);
     if (ret != 0) {
-        plog_warn(plugin, "config:%s, decode error: %s", config, err_param);
+        plog_error(plugin, "config:%s, decode error: %s", config, err_param);
         free(err_param);
         return -1;
     }
@@ -138,8 +149,8 @@ static int driver_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
 
 static int driver_validate_tag(neu_plugin_t *plugin, neu_datatag_t *tag)
 {
-    plog_debug(plugin, "validate tag success, name:%s, address:%s,type:%d ",
-               tag->name, tag->address, tag->type);
+    plog_notice(plugin, "validate tag success, name:%s, address:%s, type:%d ",
+                tag->name, tag->address, tag->type);
 
     return 0;
 }
@@ -152,5 +163,10 @@ static int driver_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group)
 static int driver_write(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
                         neu_value_u value)
 {
-    return file_write(plugin, req, tag, value);
+    int ret = file_write(plugin, req, tag, value);
+
+    plog_notice(plugin, "write response, ret: %d, tag_name: %s, type: %d", ret,
+                tag->name, tag->type);
+
+    return ret;
 }

plugins/file/file_req.c

@@ -62,7 +62,7 @@ int file_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group)
         }
 
         if ((f = fopen(tag->address, "rb+")) == NULL) {
-            nlog_error("open file:%s", tag->address);
+            nlog_warn("fail open to read and write file: %s", tag->address);
             dvalue.type      = NEU_TYPE_ERROR;
             dvalue.value.i32 = NEU_ERR_FILE_OPEN_FAILURE;
             goto dvalue_result;
@@ -71,8 +71,11 @@ int file_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group)
         fseek(f, 0, SEEK_END);
         int size = ftell(f);
         if (size >= length) {
+            nlog_warn(
+                "the file length exceeds the set value, file: %s, length: %d",
+                tag->address, size);
             dvalue.type      = NEU_TYPE_ERROR;
-            dvalue.value.i32 = NEU_ERR_FILE_TOO_LONG;
+            dvalue.value.i32 = NEU_ERR_STRING_TOO_LONG;
 
             fclose(f);
             goto dvalue_result;
@@ -81,10 +84,18 @@ int file_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group)
         fseek(f, 0, SEEK_SET);
         int ret = fread(buf, sizeof(char), length, f);
         if (ret == 0) {
-            nlog_error("read file failed:%s", tag->address);
+            nlog_warn("read file failed:%s", tag->address);
             dvalue.type      = NEU_TYPE_ERROR;
             dvalue.value.i32 = NEU_ERR_FILE_READ_FAILURE;
 
+            fclose(f);
+            goto dvalue_result;
+        } else if (ret >= NEU_VALUE_SIZE) {
+            nlog_warn("the file length exceeds the maximum, file: %s, ret: %d",
+                      tag->address, ret);
+            dvalue.type      = NEU_TYPE_ERROR;
+            dvalue.value.i32 = NEU_ERR_STRING_TOO_LONG;
+
             fclose(f);
             goto dvalue_result;
         }
@@ -121,19 +132,32 @@ static void plugin_group_free(neu_plugin_group_t *pgp)
 int file_write(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
                neu_value_u value)
 {
-    FILE *fp  = fopen(tag->address, "w");
-    int   ret = 0;
+    FILE *fp     = NULL;
+    int   ret    = 0;
+    int   length = strlen(value.str);
+
+    if (length >= NEU_VALUE_SIZE) {
+        nlog_warn("write file exceeds the maximum value, file: %s, length: %d",
+                  tag->address, length);
+        ret = NEU_ERR_STRING_TOO_LONG;
 
-    if (0 == fwrite(value.str, sizeof(char), strlen(value.str), fp)) {
-        nlog_error("write file failed:%s", tag->address);
-        ret = NEU_ERR_FILE_WRITE_FAILURE;
         goto dvalue_result;
     }
 
+    fp = fopen(tag->address, "w");
+
+    if (0 == fwrite(value.str, sizeof(char), strlen(value.str), fp)) {
+        nlog_warn("write file failed: %s", tag->address);
+        ret = NEU_ERR_FILE_WRITE_FAILURE;
+
+        fclose(fp);
+        goto dvalue_result;
+    }
+
+    fclose(fp);
     neu_datatag_string_ltoh(value.str, strlen(value.str));
 
 dvalue_result:
-    fclose(fp);
     plugin->common.adapter_callbacks->driver.write_response(
         plugin->common.adapter, req, ret);
 

plugins/modbus/modbus_tcp.c

@@ -91,12 +91,14 @@ static int driver_init(neu_plugin_t *plugin)
     plugin->events = neu_event_new();
     plugin->stack  = modbus_stack_create((void *) plugin, modbus_send_msg,
                                         modbus_value_handle, modbus_write_resp);
+    plog_notice(plugin, "%s init success", plugin->common.name);
 
     return 0;
 }
 
 static int driver_uninit(neu_plugin_t *plugin)
 {
+    plog_notice(plugin, "%s uninit start", plugin->common.name);
     if (plugin->conn != NULL) {
         neu_conn_destory(plugin->conn);
     }
@@ -107,7 +109,7 @@ static int driver_uninit(neu_plugin_t *plugin)
 
     neu_event_close(plugin->events);
 
-    plog_info(plugin, "node: modbus uninit ok");
+    plog_notice(plugin, "%s uninit success", plugin->common.name);
 
     return 0;
 }
@@ -115,12 +117,14 @@ static int driver_uninit(neu_plugin_t *plugin)
 static int driver_start(neu_plugin_t *plugin)
 {
     neu_conn_start(plugin->conn);
+    plog_notice(plugin, "%s start success", plugin->common.name);
     return 0;
 }
 
 static int driver_stop(neu_plugin_t *plugin)
 {
     neu_conn_stop(plugin->conn);
+    plog_notice(plugin, "%s stop success", plugin->common.name);
     return 0;
 }
 
@@ -137,8 +141,8 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
         neu_parse_param((char *) config, &err_param, 3, &port, &host, &timeout);
 
     if (ret != 0) {
-        plog_warn(plugin, "config: %s, decode error: %s", (char *) config,
-                  err_param);
+        plog_error(plugin, "config: %s, decode error: %s", (char *) config,
+                   err_param);
         free(err_param);
         free(host.v.val_str);
         return -1;
@@ -150,8 +154,9 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
     param.params.tcp_client.port    = port.v.val_int;
     param.params.tcp_client.timeout = timeout.v.val_int;
 
-    plog_info(plugin, "config: host: %s, port: %" PRId64 ", timeout: %" PRId64,
-              host.v.val_str, port.v.val_int, timeout.v.val_int);
+    plog_notice(plugin,
+                "config: host: %s, port: %" PRId64 ", timeout: %" PRId64,
+                host.v.val_str, port.v.val_int, timeout.v.val_int);
 
     plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
     if (plugin->conn != NULL) {
@@ -181,13 +186,21 @@ static int driver_validate_tag(neu_plugin_t *plugin, neu_datatag_t *tag)
 
     int ret = modbus_tag_to_point(tag, &point);
     if (ret == 0) {
-        plog_debug(
+        plog_notice(
             plugin,
             "validate tag success, name: %s, address: %s, type: %d, slave id: "
             "%d, start address: %d, n register: %d, area: %s",
             tag->name, tag->address, tag->type, point.slave_id,
             point.start_address, point.n_register,
             modbus_area_to_str(point.area));
+    } else {
+        plog_error(
+            plugin,
+            "validate tag error, name: %s, address: %s, type: %d, slave id: "
+            "%d, start address: %d, n register: %d, area: %s",
+            tag->name, tag->address, tag->type, point.slave_id,
+            point.start_address, point.n_register,
+            modbus_area_to_str(point.area));
     }
 
     return ret;

plugins/monitor/metric_handle.c

@@ -109,8 +109,8 @@ static int response(nng_aio *aio, char *content, enum nng_http_status status)
     nng_http_res_set_status(res, status);
 
     nng_http_req *nng_req = nng_aio_get_input(aio, 0);
-    nlog_notice("%s %s [%d]", nng_http_req_get_method(nng_req),
-                nng_http_req_get_uri(nng_req), status);
+    nlog_info("%s %s [%d]", nng_http_req_get_method(nng_req),
+              nng_http_req_get_uri(nng_req), status);
 
     nng_aio_set_output(aio, 0, res);
     nng_aio_finish(aio, 0);
@@ -159,7 +159,14 @@ static inline void gen_global_metrics(const neu_metrics_t *metrics,
 static inline void
 gen_single_node_metrics(const neu_node_metrics_t *node_metrics, FILE *stream)
 {
+    fprintf(stream,
+            "# HELP node_type Driver(1) or APP(2)\n"
+            "# TYPE node_type gauge\n"
+            "node_type{node=\"%s\"} %d\n",
+            node_metrics->name, node_metrics->type);
+
     neu_metric_entry_t *e = NULL;
+
     HASH_LOOP(hh, node_metrics->entries, e)
     {
         fprintf(stream,
@@ -208,9 +215,28 @@ static void gen_all_node_metrics(const neu_metrics_t *metrics, int type_filter,
     neu_group_metrics_t *g = NULL;
     neu_node_metrics_t * n = NULL;
 
+    bool commented = false;
+    HASH_LOOP(hh, metrics->node_metrics, n)
+    {
+        if (!(type_filter & n->type)) {
+            continue;
+        }
+
+        if (!commented) {
+            commented = true;
+            fprintf(stream,
+                    "# HELP node_type Driver(1) or APP(2)\n"
+                    "# TYPE node_type gauge\n");
+        }
+
+        fprintf(stream, "node_type{node=\"%s\"} %d\n", n->name, n->type);
+    }
+
+    n = NULL;
+
     HASH_LOOP(hh, metrics->registered_metrics, r)
     {
-        bool commented = false;
+        commented = false;
         HASH_LOOP(hh, metrics->node_metrics, n)
         {
             if (!(type_filter & n->type)) {
@@ -282,8 +308,8 @@ void handle_get_metric(nng_aio *aio)
     const char *cat_param = neu_http_get_param(aio, "category", &cat_param_len);
     if (NULL != cat_param &&
         !parse_metrics_catgory(cat_param, cat_param_len, &cat)) {
-        plog_info(plugin, "invalid metrics category: %.*s", (int) cat_param_len,
-                  cat_param);
+        plog_error(plugin, "invalid metrics category: %.*s",
+                   (int) cat_param_len, cat_param);
         status = NNG_HTTP_STATUS_BAD_REQUEST;
         goto end;
     }

plugins/monitor/monitor.c

@@ -37,15 +37,15 @@ static void connect_cb(void *data)
 {
     neu_plugin_t *plugin      = data;
     plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED;
-    plog_info(plugin, "plugin `%s` connected", neu_plugin_module.module_name);
+    plog_notice(plugin, "plugin `%s` connected", neu_plugin_module.module_name);
 }
 
 static void disconnect_cb(void *data)
 {
     neu_plugin_t *plugin      = data;
     plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
-    plog_info(plugin, "plugin `%s` disconnected",
-              neu_plugin_module.module_name);
+    plog_notice(plugin, "plugin `%s` disconnected",
+                neu_plugin_module.module_name);
 }
 
 static nng_http_server *server_init()
@@ -54,7 +54,7 @@ static nng_http_server *server_init()
     nng_http_server *server    = NULL;
     const char *     host_port = "http://0.0.0.0:7000";
 
-    nlog_info("monitor bind url: %s", host_port);
+    nlog_notice("monitor bind url: %s", host_port);
 
     int ret = nng_url_parse(&url, host_port);
     if (0 != ret) {
@@ -179,7 +179,7 @@ static inline void stop_heartbeart_timer(neu_plugin_t *plugin)
     if (plugin->heartbeat_timer) {
         neu_event_del_timer(plugin->events, plugin->heartbeat_timer);
         plugin->heartbeat_timer = NULL;
-        plog_info(plugin, "heartbeat timer stopped");
+        plog_notice(plugin, "heartbeat timer stopped");
     }
 }
 
@@ -219,7 +219,7 @@ end:
     }
     plugin->heartbeat_timer = timer;
 
-    plog_info(plugin, "start_hearbeat_timer interval: %" PRIu64, interval);
+    plog_notice(plugin, "start_hearbeat_timer interval: %" PRIu64, interval);
 
     return 0;
 }
@@ -286,8 +286,8 @@ static int monitor_plugin_close(neu_plugin_t *plugin)
     nng_http_server_stop(plugin->api_server);
     nng_http_server_release(plugin->api_server);
     pthread_mutex_destroy(&plugin->mutex);
-    plog_info(plugin, "Success to free plugin: %s",
-              neu_plugin_module.module_name);
+    plog_notice(plugin, "Success to free plugin: %s",
+                neu_plugin_module.module_name);
     free(plugin);
 
     return rv;
@@ -298,7 +298,7 @@ static int monitor_plugin_init(neu_plugin_t *plugin)
     int rv = 0;
     (void) plugin;
 
-    plog_info(plugin, "Initialize plugin: %s", neu_plugin_module.module_name);
+    plog_notice(plugin, "Initialize plugin: %s", neu_plugin_module.module_name);
     return rv;
 }
 
@@ -308,7 +308,8 @@ static int monitor_plugin_uninit(neu_plugin_t *plugin)
 
     uninit_timers_and_mqtt(plugin);
 
-    plog_info(plugin, "Uninitialize plugin: %s", neu_plugin_module.module_name);
+    plog_notice(plugin, "Uninitialize plugin: %s",
+                neu_plugin_module.module_name);
     return rv;
 }
 
@@ -372,7 +373,7 @@ static int monitor_plugin_config(neu_plugin_t *plugin, const char *setting)
     memmove(plugin->config, &config, sizeof(config));
     // `config` moved, do not call monitor_config_fini
 
-    plog_info(plugin, "config plugin `%s` success", plugin_name);
+    plog_notice(plugin, "config plugin `%s` success", plugin_name);
     return 0;
 
 error:
@@ -423,7 +424,7 @@ static int monitor_plugin_start(neu_plugin_t *plugin)
     const char *plugin_name = neu_plugin_module.module_name;
 
     if (NULL == plugin->mqtt_client) {
-        plog_info(plugin, "mqtt client is NULL");
+        plog_notice(plugin, "mqtt client is NULL");
         goto end;
     }
 
@@ -443,7 +444,7 @@ static int monitor_plugin_start(neu_plugin_t *plugin)
 end:
     if (0 == rv) {
         plugin->started = true;
-        plog_info(plugin, "start plugin `%s` success", plugin_name);
+        plog_notice(plugin, "start plugin `%s` success", plugin_name);
     } else {
         plog_error(plugin, "start plugin `%s` failed, error %d", plugin_name,
                    rv);
@@ -456,13 +457,13 @@ static int monitor_plugin_stop(neu_plugin_t *plugin)
 {
     if (plugin->mqtt_client) {
         neu_mqtt_client_close(plugin->mqtt_client);
-        plog_info(plugin, "mqtt client closed");
+        plog_notice(plugin, "mqtt client closed");
     }
 
     stop_heartbeart_timer(plugin);
     plugin->started = false;
-    plog_info(plugin, "stop plugin `%s` success",
-              neu_plugin_module.module_name);
+    plog_notice(plugin, "stop plugin `%s` success",
+                neu_plugin_module.module_name);
     return 0;
 }
 

plugins/monitor/monitor_config.c

@@ -75,12 +75,12 @@ static inline int parse_ssl_params(neu_plugin_t *plugin, const char *setting,
     // ssl, optional
     int ret = neu_parse_param(setting, NULL, 1, ssl);
     if (0 != ret) {
-        plog_info(plugin, "setting no ssl");
+        plog_notice(plugin, "setting no ssl");
         return 0;
     }
 
     if (false == ssl->v.val_bool) {
-        plog_info(plugin, "setting ssl disabled");
+        plog_notice(plugin, "setting ssl disabled");
         return 0;
     }
 
@@ -93,7 +93,7 @@ static inline int parse_ssl_params(neu_plugin_t *plugin, const char *setting,
     // cert, optional
     ret = neu_parse_param(setting, NULL, 1, cert);
     if (0 != ret) {
-        plog_info(plugin, "setting no cert");
+        plog_notice(plugin, "setting no cert");
         return 0;
     }
 
@@ -110,7 +110,7 @@ static inline int parse_ssl_params(neu_plugin_t *plugin, const char *setting,
     // keypass, optional
     ret = neu_parse_param(setting, NULL, 1, keypass);
     if (0 != ret) {
-        plog_info(plugin, "setting no keypass");
+        plog_notice(plugin, "setting no keypass");
     } else if (0 == strlen(keypass->v.val_str)) {
         plog_error(plugin, "setting invalid keypass");
         return -1;
@@ -280,13 +280,13 @@ int monitor_config_parse(neu_plugin_t *plugin, const char *setting,
     // username, optional
     ret = neu_parse_param(setting, NULL, 1, &username);
     if (0 != ret) {
-        plog_info(plugin, "setting no username");
+        plog_notice(plugin, "setting no username");
     }
 
     // password, optional
     ret = neu_parse_param(setting, NULL, 1, &password);
     if (0 != ret) {
-        plog_info(plugin, "setting no password");
+        plog_notice(plugin, "setting no password");
     }
 
     ret = parse_ssl_params(plugin, setting, &ssl, &ca, &cert, &key, &keypass);
@@ -313,48 +313,54 @@ int monitor_config_parse(neu_plugin_t *plugin, const char *setting,
     config->key                = key.v.val_str;
     config->keypass            = keypass.v.val_str;
 
-    plog_info(plugin, "config client-id         : %s", config->client_id);
-    plog_info(plugin, "config event-topic-prefix: %s",
-              config->event_topic_prefix);
-    plog_info(plugin, "config heartbeat-interval: %" PRIu64,
-              config->heartbeat_interval);
-    plog_info(plugin, "config heartbeat-topic   : %s", config->heartbeat_topic);
-    plog_info(plugin, "config host              : %s", config->host);
-    plog_info(plugin, "config port              : %" PRIu16, config->port);
+    plog_notice(plugin, "config client-id         : %s", config->client_id);
+    plog_notice(plugin, "config event-topic-prefix: %s",
+                config->event_topic_prefix);
+    plog_notice(plugin, "config heartbeat-interval: %" PRIu64,
+                config->heartbeat_interval);
+    plog_notice(plugin, "config heartbeat-topic   : %s",
+                config->heartbeat_topic);
+    plog_notice(plugin, "config host              : %s", config->host);
+    plog_notice(plugin, "config port              : %" PRIu16, config->port);
 
     if (config->username) {
-        plog_info(plugin, "config username          : %s", config->username);
+        plog_notice(plugin, "config username          : %s", config->username);
     }
     if (config->password) {
-        plog_info(plugin, "config password          : %s",
-                  0 == strlen(config->password) ? "" : placeholder);
+        plog_notice(plugin, "config password          : %s",
+                    0 == strlen(config->password) ? "" : placeholder);
     }
     if (config->ca) {
-        plog_info(plugin, "config ca                : %s", placeholder);
+        plog_notice(plugin, "config ca                : %s", placeholder);
     }
     if (config->cert) {
-        plog_info(plugin, "config cert              : %s", placeholder);
+        plog_notice(plugin, "config cert              : %s", placeholder);
     }
     if (config->key) {
-        plog_info(plugin, "config key               : %s", placeholder);
+        plog_notice(plugin, "config key               : %s", placeholder);
     }
     if (config->keypass) {
-        plog_info(plugin, "config keypass           : %s", placeholder);
+        plog_notice(plugin, "config keypass           : %s", placeholder);
     }
 
-    plog_info(plugin, "node add event topic     : %s", config->node_add_topic);
-    plog_info(plugin, "node del event topic     : %s", config->node_del_topic);
-    plog_info(plugin, "node ctl event topic     : %s", config->node_ctl_topic);
-    plog_info(plugin, "node setting event topic : %s",
-              config->node_setting_topic);
-    plog_info(plugin, "group add topic          : %s", config->group_add_topic);
-    plog_info(plugin, "group del topic          : %s", config->group_del_topic);
-    plog_info(plugin, "group update topic       : %s",
-              config->group_update_topic);
-    plog_info(plugin, "tag add topic            : %s", config->tag_add_topic);
-    plog_info(plugin, "tag del topic            : %s", config->tag_del_topic);
-    plog_info(plugin, "tag update topic         : %s",
-              config->tag_update_topic);
+    plog_notice(plugin, "node add event topic     : %s",
+                config->node_add_topic);
+    plog_notice(plugin, "node del event topic     : %s",
+                config->node_del_topic);
+    plog_notice(plugin, "node ctl event topic     : %s",
+                config->node_ctl_topic);
+    plog_notice(plugin, "node setting event topic : %s",
+                config->node_setting_topic);
+    plog_notice(plugin, "group add topic          : %s",
+                config->group_add_topic);
+    plog_notice(plugin, "group del topic          : %s",
+                config->group_del_topic);
+    plog_notice(plugin, "group update topic       : %s",
+                config->group_update_topic);
+    plog_notice(plugin, "tag add topic            : %s", config->tag_add_topic);
+    plog_notice(plugin, "tag del topic            : %s", config->tag_del_topic);
+    plog_notice(plugin, "tag update topic         : %s",
+                config->tag_update_topic);
 
     return 0;
 

plugins/mqtt/mqtt_config.c

@@ -72,12 +72,12 @@ static inline int parse_ssl_params(neu_plugin_t *plugin, const char *setting,
     // ssl, optional
     int ret = neu_parse_param(setting, NULL, 1, ssl);
     if (0 != ret) {
-        plog_info(plugin, "setting no ssl");
+        plog_notice(plugin, "setting no ssl");
         return 0;
     }
 
     if (false == ssl->v.val_bool) {
-        plog_info(plugin, "setting ssl disabled");
+        plog_notice(plugin, "setting ssl disabled");
         return 0;
     }
 
@@ -90,7 +90,7 @@ static inline int parse_ssl_params(neu_plugin_t *plugin, const char *setting,
     // cert, optional
     ret = neu_parse_param(setting, NULL, 1, cert);
     if (0 != ret) {
-        plog_info(plugin, "setting no cert");
+        plog_notice(plugin, "setting no cert");
         return 0;
     }
 
@@ -107,7 +107,7 @@ static inline int parse_ssl_params(neu_plugin_t *plugin, const char *setting,
     // keypass, optional
     ret = neu_parse_param(setting, NULL, 1, keypass);
     if (0 != ret) {
-        plog_info(plugin, "setting no keypass");
+        plog_notice(plugin, "setting no keypass");
     } else if (0 == strlen(keypass->v.val_str)) {
         plog_error(plugin, "setting invalid keypass");
         return -1;
@@ -128,7 +128,7 @@ static int parse_cache_params(neu_plugin_t *plugin, const char *setting,
     // offline-cache flag, optional
     ret = neu_parse_param(setting, NULL, 1, offline_cache);
     if (0 != ret) {
-        plog_info(plugin, "setting no offline cache flag");
+        plog_notice(plugin, "setting no offline cache flag");
         flag_present = false; // `offline-cache` not present in setting
     }
 
@@ -276,7 +276,7 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
     if (NULL == write_resp_topic.v.val_str &&
         0 > neu_asprintf(&write_resp_topic.v.val_str, "/neuron/%s/write/resp",
                          plugin->common.name)) {
-        plog_error(plugin, "setting write request topic error");
+        plog_error(plugin, "setting write response topic error");
         goto error;
     }
 
@@ -302,13 +302,13 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
     // username, optional
     ret = neu_parse_param(setting, NULL, 1, &username);
     if (0 != ret) {
-        plog_info(plugin, "setting no username");
+        plog_notice(plugin, "setting no username");
     }
 
     // password, optional
     ret = neu_parse_param(setting, NULL, 1, &password);
     if (0 != ret) {
-        plog_info(plugin, "setting no password");
+        plog_notice(plugin, "setting no password");
     }
 
     ret = parse_ssl_params(plugin, setting, &ssl, &ca, &cert, &key, &keypass);
@@ -333,36 +333,38 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
     config->key              = key.v.val_str;
     config->keypass          = keypass.v.val_str;
 
-    plog_info(plugin, "config client-id       : %s", config->client_id);
-    plog_info(plugin, "config qos             : %d", config->qos);
-    plog_info(plugin, "config format          : %s",
-              mqtt_upload_format_str(config->format));
-    plog_info(plugin, "config write-req-topic : %s", config->write_req_topic);
-    plog_info(plugin, "config write-resp-topic: %s", config->write_req_topic);
-    plog_info(plugin, "config cache           : %zu", config->cache);
-    plog_info(plugin, "config cache-mem-size  : %zu", config->cache_mem_size);
-    plog_info(plugin, "config cache-disk-size : %zu", config->cache_disk_size);
-    plog_info(plugin, "config host            : %s", config->host);
-    plog_info(plugin, "config port            : %" PRIu16, config->port);
+    plog_notice(plugin, "config client-id       : %s", config->client_id);
+    plog_notice(plugin, "config qos             : %d", config->qos);
+    plog_notice(plugin, "config format          : %s",
+                mqtt_upload_format_str(config->format));
+    plog_notice(plugin, "config write-req-topic : %s", config->write_req_topic);
+    plog_notice(plugin, "config write-resp-topic: %s",
+                config->write_resp_topic);
+    plog_notice(plugin, "config cache           : %zu", config->cache);
+    plog_notice(plugin, "config cache-mem-size  : %zu", config->cache_mem_size);
+    plog_notice(plugin, "config cache-disk-size : %zu",
+                config->cache_disk_size);
+    plog_notice(plugin, "config host            : %s", config->host);
+    plog_notice(plugin, "config port            : %" PRIu16, config->port);
 
     if (config->username) {
-        plog_info(plugin, "config username        : %s", config->username);
+        plog_notice(plugin, "config username        : %s", config->username);
     }
     if (config->password) {
-        plog_info(plugin, "config password        : %s",
-                  0 == strlen(config->password) ? "" : placeholder);
+        plog_notice(plugin, "config password        : %s",
+                    0 == strlen(config->password) ? "" : placeholder);
     }
     if (config->ca) {
-        plog_info(plugin, "config ca              : %s", placeholder);
+        plog_notice(plugin, "config ca              : %s", placeholder);
     }
     if (config->cert) {
-        plog_info(plugin, "config cert            : %s", placeholder);
+        plog_notice(plugin, "config cert            : %s", placeholder);
     }
     if (config->key) {
-        plog_info(plugin, "config key             : %s", placeholder);
+        plog_notice(plugin, "config key             : %s", placeholder);
     }
     if (config->keypass) {
-        plog_info(plugin, "config keypass         : %s", placeholder);
+        plog_notice(plugin, "config keypass         : %s", placeholder);
     }
 
     return 0;

plugins/mqtt/mqtt_handle.c

@@ -194,8 +194,8 @@ static char *generate_write_resp_json(neu_plugin_t *    plugin,
 static inline int send_read_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
                                 neu_json_read_req_t *req)
 {
-    plog_info(plugin, "read uuid:%s, group:%s, node:%s", mqtt->uuid, req->group,
-              req->node);
+    plog_notice(plugin, "read uuid:%s, group:%s, node:%s", mqtt->uuid,
+                req->group, req->node);
 
     neu_reqresp_head_t header = { 0 };
     header.ctx                = mqtt;
@@ -214,8 +214,8 @@ static inline int send_read_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
 static int send_write_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
                           neu_json_write_req_t *req)
 {
-    plog_info(plugin, "write uuid:%s, group:%s, node:%s", mqtt->uuid,
-              req->group, req->node);
+    plog_notice(plugin, "write uuid:%s, group:%s, node:%s", mqtt->uuid,
+                req->group, req->node);
 
     neu_reqresp_head_t  header = { 0 };
     neu_req_write_tag_t cmd    = { 0 };
@@ -538,8 +538,8 @@ int handle_subscribe_group(neu_plugin_t *plugin, neu_req_subscribe_t *sub_info)
         goto end;
     }
 
-    plog_info(plugin, "route driver:%s group:%s to topic:%s", sub_info->driver,
-              sub_info->group, topic.v.val_str);
+    plog_notice(plugin, "route driver:%s group:%s to topic:%s",
+                sub_info->driver, sub_info->group, topic.v.val_str);
 
 end:
     free(sub_info->params);
@@ -550,7 +550,7 @@ int handle_unsubscribe_group(neu_plugin_t *         plugin,
                              neu_req_unsubscribe_t *unsub_info)
 {
     route_tbl_del(&plugin->route_tbl, unsub_info->driver, unsub_info->group);
-    plog_info(plugin, "del route driver:%s group:%s", unsub_info->driver,
-              unsub_info->group);
+    plog_notice(plugin, "del route driver:%s group:%s", unsub_info->driver,
+                unsub_info->group);
     return 0;
 }

plugins/mqtt/mqtt_plugin.c

@@ -30,15 +30,15 @@ static void connect_cb(void *data)
 {
     neu_plugin_t *plugin      = data;
     plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED;
-    plog_info(plugin, "plugin `%s` connected", neu_plugin_module.module_name);
+    plog_notice(plugin, "plugin `%s` connected", neu_plugin_module.module_name);
 }
 
 static void disconnect_cb(void *data)
 {
     neu_plugin_t *plugin      = data;
     plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
-    plog_info(plugin, "plugin `%s` disconnected",
-              neu_plugin_module.module_name);
+    plog_notice(plugin, "plugin `%s` disconnected",
+                neu_plugin_module.module_name);
 }
 
 static neu_plugin_t *mqtt_plugin_open(void)
@@ -51,7 +51,7 @@ static neu_plugin_t *mqtt_plugin_open(void)
 static int mqtt_plugin_close(neu_plugin_t *plugin)
 {
     const char *name = neu_plugin_module.module_name;
-    plog_info(plugin, "success to free plugin:%s", name);
+    plog_notice(plugin, "success to free plugin:%s", name);
 
     free(plugin);
     return NEU_ERR_SUCCESS;
@@ -59,8 +59,8 @@ static int mqtt_plugin_close(neu_plugin_t *plugin)
 
 static int mqtt_plugin_init(neu_plugin_t *plugin)
 {
-    plog_info(plugin, "initialize plugin `%s` success",
-              neu_plugin_module.module_name);
+    plog_notice(plugin, "initialize plugin `%s` success",
+                neu_plugin_module.module_name);
     neu_adapter_register_metric_cb_t register_metric =
         plugin->common.adapter_callbacks->register_metric;
     register_metric(plugin->common.adapter, NEU_METRIC_CACHED_MSGS_NUM,
@@ -85,8 +85,8 @@ static int mqtt_plugin_uninit(neu_plugin_t *plugin)
 
     route_tbl_free(plugin->route_tbl);
 
-    plog_info(plugin, "uninitialize plugin `%s` success",
-              neu_plugin_module.module_name);
+    plog_notice(plugin, "uninitialize plugin `%s` success",
+                neu_plugin_module.module_name);
     return NEU_ERR_SUCCESS;
 }
 
@@ -269,11 +269,11 @@ static int mqtt_plugin_config(neu_plugin_t *plugin, const char *setting)
     }
     memmove(&plugin->config, &config, sizeof(config));
 
-    plog_info(plugin, "config plugin `%s` success", plugin_name);
+    plog_notice(plugin, "config plugin `%s` success", plugin_name);
     return 0;
 
 error:
-    plog_info(plugin, "config plugin `%s` fail", plugin_name);
+    plog_error(plugin, "config plugin `%s` fail", plugin_name);
     mqtt_config_fini(&config);
     return rv;
 }
@@ -299,7 +299,7 @@ static int mqtt_plugin_start(neu_plugin_t *plugin)
 
 end:
     if (0 == rv) {
-        plog_info(plugin, "start plugin `%s` success", plugin_name);
+        plog_notice(plugin, "start plugin `%s` success", plugin_name);
     } else {
         plog_error(plugin, "start plugin `%s` failed, error %d", plugin_name,
                    rv);
@@ -313,11 +313,11 @@ static int mqtt_plugin_stop(neu_plugin_t *plugin)
     if (plugin->client) {
         unsubscribe(plugin, &plugin->config);
         neu_mqtt_client_close(plugin->client);
-        plog_info(plugin, "mqtt client closed");
+        plog_notice(plugin, "mqtt client closed");
     }
 
-    plog_info(plugin, "stop plugin `%s` success",
-              neu_plugin_module.module_name);
+    plog_notice(plugin, "stop plugin `%s` success",
+                neu_plugin_module.module_name);
     return NEU_ERR_SUCCESS;
 }
 

plugins/restful/normal_handle.c

@@ -54,12 +54,20 @@ void handle_login(nng_aio *aio)
         aio, neu_json_login_req_t, neu_json_decode_login_req, {
             neu_json_login_resp_t login_resp = { 0 };
             neu_user_t *          user       = neu_load_user(req->name);
+            int                   pass_len   = strlen(req->pass);
 
             if (NULL == user) {
-                nlog_warn("could not find user `%s`", req->name);
+                nlog_error("could not find user `%s`", req->name);
                 NEU_JSON_RESPONSE_ERROR(NEU_ERR_INVALID_USER_OR_PASSWORD, {
                     neu_http_response(aio, error_code.error, result_error);
                 });
+            } else if (pass_len < NEU_USER_PASSWORD_MIN_LEN ||
+                       pass_len > NEU_USER_PASSWORD_MAX_LEN) {
+                nlog_error("user `%s` password too short or too long",
+                           req->name);
+                NEU_JSON_RESPONSE_ERROR(NEU_ERR_INVALID_PASSWORD_LEN, {
+                    neu_http_response(aio, error_code.error, result_error);
+                });
             } else if (neu_user_check_password(user, req->pass)) {
 
                 char *token  = NULL;
@@ -81,7 +89,7 @@ void handle_login(nng_aio *aio)
                 jwt_free_str(token);
                 free(result);
             } else {
-                nlog_warn("user `%s` password check fail", req->name);
+                nlog_error("user `%s` password check fail", req->name);
                 NEU_JSON_RESPONSE_ERROR(NEU_ERR_INVALID_USER_OR_PASSWORD, {
                     neu_http_response(aio, error_code.error, result_error);
                 });

plugins/restful/rest.c

@@ -55,7 +55,7 @@ static nng_http_server *server_init()
     nng_http_server *server;
 
     snprintf(host_port, sizeof(host_port), "http://0.0.0.0:7000");
-    nlog_info("bind url: %s", host_port);
+    nlog_notice("bind url: %s", host_port);
 
     int ret = nng_url_parse(&url, host_port);
     if (ret != 0) {
@@ -105,7 +105,7 @@ static neu_plugin_t *dashb_plugin_open(void)
         goto server_init_fail;
     }
 
-    nlog_info("Success to create plugin: %s", neu_plugin_module.module_name);
+    nlog_notice("Success to create plugin: %s", neu_plugin_module.module_name);
     return plugin;
 
 server_init_fail:
@@ -121,13 +121,11 @@ static int dashb_plugin_close(neu_plugin_t *plugin)
 {
     int rv = 0;
 
-    nlog_info("Success to before free plugin: %s",
-              neu_plugin_module.module_name);
     nng_http_server_stop(plugin->server);
     nng_http_server_release(plugin->server);
 
     free(plugin);
-    nlog_info("Success to free plugin: %s", neu_plugin_module.module_name);
+    nlog_notice("Success to free plugin: %s", neu_plugin_module.module_name);
     return rv;
 }
 
@@ -139,7 +137,7 @@ static int dashb_plugin_init(neu_plugin_t *plugin)
 
     rv = neu_jwt_init(g_config_dir);
 
-    nlog_info("Initialize plugin: %s", neu_plugin_module.module_name);
+    nlog_notice("Initialize plugin: %s", neu_plugin_module.module_name);
     return rv;
 }
 
@@ -149,7 +147,7 @@ static int dashb_plugin_uninit(neu_plugin_t *plugin)
 
     (void) plugin;
 
-    nlog_info("Uninitialize plugin: %s", neu_plugin_module.module_name);
+    nlog_notice("Uninitialize plugin: %s", neu_plugin_module.module_name);
     return rv;
 }
 
@@ -160,7 +158,7 @@ static int dashb_plugin_config(neu_plugin_t *plugin, const char *configs)
     (void) plugin;
     (void) configs;
 
-    nlog_info("config plugin: %s", neu_plugin_module.module_name);
+    nlog_notice("config plugin: %s", neu_plugin_module.module_name);
     return rv;
 }
 

plugins/restful/rw_handle.c

@@ -58,7 +58,6 @@ void handle_write(nng_aio *aio)
 
     NEU_PROCESS_HTTP_REQUEST_VALIDATE_JWT(
         aio, neu_json_write_req_t, neu_json_decode_write_req, {
-            int                 ret    = 0;
             neu_reqresp_head_t  header = { 0 };
             neu_req_write_tag_t cmd    = { 0 };
 
@@ -75,8 +74,13 @@ void handle_write(nng_aio *aio)
                 cmd.value.value.u64 = req->value.val_int;
                 break;
             case NEU_JSON_STR:
-                cmd.value.type = NEU_TYPE_STRING;
-                strcpy(cmd.value.value.str, req->value.val_str);
+                if (strlen(req->value.val_str) >= NEU_VALUE_SIZE) {
+                    cmd.value.type      = NEU_TYPE_ERROR;
+                    cmd.value.value.i32 = NEU_ERR_STRING_TOO_LONG;
+                } else {
+                    cmd.value.type = NEU_TYPE_STRING;
+                    strcpy(cmd.value.value.str, req->value.val_str);
+                }
                 break;
             case NEU_JSON_DOUBLE:
                 cmd.value.type      = NEU_TYPE_DOUBLE;
@@ -91,11 +95,18 @@ void handle_write(nng_aio *aio)
                 break;
             }
 
-            ret = neu_plugin_op(plugin, header, &cmd);
-            if (ret != 0) {
-                NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, {
-                    neu_http_response(aio, NEU_ERR_IS_BUSY, result_error);
+            if (cmd.value.type == NEU_TYPE_ERROR) {
+                NEU_JSON_RESPONSE_ERROR(NEU_ERR_STRING_TOO_LONG, {
+                    neu_http_response(aio, NEU_ERR_STRING_TOO_LONG,
+                                      result_error);
                 });
+            } else {
+                int ret = neu_plugin_op(plugin, header, &cmd);
+                if (ret != 0) {
+                    NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, {
+                        neu_http_response(aio, NEU_ERR_IS_BUSY, result_error);
+                    });
+                }
             }
         })
 }

src/adapter/adapter.c

@@ -124,8 +124,8 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
 
     rv = nng_pair1_open(&adapter->sock);
     assert(rv == 0);
-    nng_socket_set_int(adapter->sock, NNG_OPT_RECVBUF, 128);
-    nng_socket_set_int(adapter->sock, NNG_OPT_SENDBUF, 128);
+    nng_socket_set_int(adapter->sock, NNG_OPT_RECVBUF, 8192);
+    nng_socket_set_int(adapter->sock, NNG_OPT_SENDBUF, 8192);
     nng_socket_set_ms(adapter->sock, NNG_OPT_SENDTIMEO, 1000);
     nng_socket_get_int(adapter->sock, NNG_OPT_RECVFD, &adapter->recv_fd);
 
@@ -181,7 +181,7 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
     timer_level.usr_data = (void *) adapter;
     adapter->timer_lev   = neu_event_add_timer(adapter->events, timer_level);
 
-    nlog_info("Success to create adapter: %s", adapter->name);
+    nlog_notice("Success to create adapter: %s", adapter->name);
 
     adapter_storage_state(adapter->name, adapter->state);
     return adapter;
@@ -371,8 +371,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
 
     nng_recvmsg(adapter->sock, &msg, 0);
     header = (neu_reqresp_head_t *) nng_msg_body(msg);
-    nlog_debug("adapter(%s) recv msg from: %s, type: %s", adapter->name,
-               header->sender, neu_reqresp_type_string(header->type));
+    nlog_info("adapter(%s) recv msg from: %s, type: %s", adapter->name,
+              header->sender, neu_reqresp_type_string(header->type));
 
     switch (header->type) {
     case NEU_RESP_GET_DRIVER_GROUP:
@@ -755,9 +755,10 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
         strcpy(name, adapter->name);
         strcpy(receiver, header->receiver);
         if (nng_sendmsg(adapter->sock, uninit_msg, 0) == 0) {
-            nlog_warn("%s send uninit msg to %s", name, receiver);
+            nlog_notice("%s send uninit msg to %s", name, receiver);
         } else {
             nng_msg_free(uninit_msg);
+            nlog_error("%s send uninit msg to %s failed", name, receiver);
         }
         break;
     }
@@ -822,7 +823,7 @@ int neu_adapter_uninit(neu_adapter_t *adapter)
 
     neu_event_del_timer(adapter->events, adapter->timer_lev);
 
-    nlog_info("Stop the adapter(%s)", adapter->name);
+    nlog_notice("Stop the adapter(%s)", adapter->name);
     return 0;
 }
 

src/adapter/driver/driver.c

@@ -19,6 +19,7 @@
 #include <assert.h>
 #include <math.h>
 #include <netinet/in.h>
+#include <pthread.h>
 #include <stdlib.h>
 
 #include <nng/nng.h>
@@ -37,12 +38,20 @@
 #include "errcodes.h"
 #include "tag.h"
 
+typedef struct to_be_write_tag {
+    neu_datatag_t *tag;
+    neu_value_u    value;
+    void *         req;
+} to_be_write_tag_t;
+
 typedef struct group {
     char *name;
 
-    int64_t      timestamp;
-    neu_group_t *group;
-    UT_array *   static_tags;
+    int64_t         timestamp;
+    neu_group_t *   group;
+    UT_array *      static_tags;
+    UT_array *      wt_tags;
+    pthread_mutex_t wt_mtx;
 
     neu_event_timer_t *report;
     neu_event_timer_t *read;
@@ -74,6 +83,7 @@ static void update(neu_adapter_t *adapter, const char *group, const char *tag,
                    neu_dvalue_t value);
 static void write_response(neu_adapter_t *adapter, void *r, neu_error error);
 static group_t *find_group(neu_adapter_driver_t *driver, const char *name);
+static void     store_write_tag(group_t *group, to_be_write_tag_t *tag);
 
 static void write_response(neu_adapter_t *adapter, void *r, neu_error error)
 {
@@ -122,7 +132,7 @@ static void update(neu_adapter_t *adapter, const char *group, const char *tag,
             &driver->adapter, NEU_METRIC_TAG_READ_ERRORS_TOTAL,
             NEU_TYPE_ERROR == value.type, NULL);
     }
-    nlog_debug(
+    nlog_info(
         "update driver: %s, group: %s, tag: %s, type: %s, timestamp: %" PRId64,
         driver->adapter.name, group, tag, neu_type_string(value.type),
         global_timestamp);
@@ -182,7 +192,13 @@ int neu_adapter_driver_uninit(neu_adapter_driver_t *driver)
         free(el->name);
         utarray_free(el->grp.tags);
 
+        utarray_foreach(el->wt_tags, to_be_write_tag_t *, tag)
+        {
+            neu_tag_free(tag->tag);
+        }
+
         utarray_free(el->static_tags);
+        utarray_free(el->wt_tags);
         neu_group_destroy(el->group);
         free(el);
     }
@@ -414,8 +430,12 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
             driver->adapter.cb_funs.response(&driver->adapter, req, &error);
             free(req);
         } else {
-            driver->adapter.module->intf_funs->driver.write_tag(
-                driver->adapter.plugin, (void *) req, tag, cmd->value.value);
+            to_be_write_tag_t wtag = { 0 };
+            wtag.req               = (void *) req;
+            wtag.value             = cmd->value.value;
+            wtag.tag               = neu_tag_dup(tag);
+
+            store_write_tag(g, &wtag);
         }
 
         neu_tag_free(tag);
@@ -429,6 +449,7 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
 int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name,
                                  uint32_t interval)
 {
+    UT_icd   icd  = { sizeof(to_be_write_tag_t), NULL, NULL, NULL };
     group_t *find = NULL;
     int      ret  = NEU_ERR_GROUP_EXIST;
 
@@ -443,6 +464,10 @@ int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name,
             .type        = NEU_EVENT_TIMER_NOBLOCK,
         };
 
+        pthread_mutex_init(&find->wt_mtx, NULL);
+
+        utarray_new(find->wt_tags, &icd);
+
         find->driver         = driver;
         find->name           = strdup(name);
         find->group          = neu_group_new(name, interval);
@@ -530,11 +555,18 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver, const char *name)
             neu_driver_cache_del(driver->cache, name, tag->name);
         }
 
+        utarray_foreach(find->wt_tags, to_be_write_tag_t *, tag)
+        {
+            neu_tag_free(tag->tag);
+        }
+
         utarray_free(find->static_tags);
         utarray_free(find->grp.tags);
+        utarray_free(find->wt_tags);
         neu_group_destroy(find->group);
         free(find);
 
+        pthread_mutex_destroy(&find->wt_mtx);
         neu_adapter_del_group_metrics(&driver->adapter, name);
         ret = NEU_ERR_SUCCESS;
     }
@@ -881,7 +913,8 @@ static void group_change(void *arg, int64_t timestamp, UT_array *static_tags,
 
     group->static_tags = static_tags;
     group->grp         = grp;
-    nlog_notice("group: %s changed", group->name);
+    nlog_notice("group: %s changed, timestamp: %" PRIi64, group->name,
+                timestamp);
 }
 
 static int read_callback(void *usr_data)
@@ -892,6 +925,17 @@ static int read_callback(void *usr_data)
         return 0;
     }
 
+    pthread_mutex_lock(&group->wt_mtx);
+    utarray_foreach(group->wt_tags, to_be_write_tag_t *, wtag)
+    {
+        group->driver->adapter.module->intf_funs->driver.write_tag(
+            group->driver->adapter.plugin, (void *) wtag->req, wtag->tag,
+            wtag->value);
+        neu_tag_free(wtag->tag);
+    }
+    utarray_clear(group->wt_tags);
+    pthread_mutex_unlock(&group->wt_mtx);
+
     if (neu_group_is_change(group->group, group->timestamp)) {
         neu_group_change_test(group->group, group->timestamp, (void *) group,
                               group_change);
@@ -927,7 +971,7 @@ static int read_report_group(int64_t timestamp, int64_t timeout,
         if (neu_tag_attribute_test(tag, NEU_ATTRIBUTE_SUBSCRIBE)) {
             if (neu_driver_cache_get_changed(cache, group, tag->name, &value) !=
                 0) {
-                nlog_debug("tag: %s not changed", tag->name);
+                nlog_info("tag: %s not changed", tag->name);
                 continue;
             }
         } else {
@@ -1199,3 +1243,10 @@ static void read_group(int64_t timestamp, int64_t timeout,
         }
     }
 }
+
+static void store_write_tag(group_t *group, to_be_write_tag_t *tag)
+{
+    pthread_mutex_lock(&group->wt_mtx);
+    utarray_push_back(group->wt_tags, tag);
+    pthread_mutex_unlock(&group->wt_mtx);
+}

src/adapter/storage.c

@@ -125,7 +125,7 @@ int adapter_load_setting(const char *node, char **setting)
 {
     int rv = neu_persister_load_node_setting(node, (const char **) setting);
     if (0 != rv) {
-        nlog_info("load %s setting fail", node);
+        nlog_error("load %s setting fail", node);
         return -1;
     }
 

src/base/metrics.c

@@ -189,7 +189,7 @@ static inline void metrics_unregister_entry(const char *name)
     HASH_FIND_STR(g_metrics_.registered_metrics, name, e);
     if (0 == --e->value) {
         HASH_DEL(g_metrics_.registered_metrics, e);
-        nlog_info("del entry:%s", e->name);
+        nlog_notice("del entry:%s", e->name);
         neu_metric_entry_free(e);
     }
 }

src/connection/connection.c

@@ -217,10 +217,10 @@ int neu_conn_tcp_server_accept(neu_conn_t *conn)
     conn->connected(conn->data, fd);
     conn->callback_trigger = true;
 
-    zlog_info(conn->param.log, "%s:%d accpet new client: %s:%d, fd: %d",
-              conn->param.params.tcp_server.ip,
-              conn->param.params.tcp_server.port, inet_ntoa(client.sin_addr),
-              ntohs(client.sin_port), fd);
+    zlog_notice(conn->param.log, "%s:%d accpet new client: %s:%d, fd: %d",
+                conn->param.params.tcp_server.ip,
+                conn->param.params.tcp_server.port, inet_ntoa(client.sin_addr),
+                ntohs(client.sin_port), fd);
 
     pthread_mutex_unlock(&conn->mtx);
 
@@ -237,6 +237,8 @@ int neu_conn_tcp_server_close_client(neu_conn_t *conn, int fd)
 
     conn->disconnected(conn->data, fd);
     conn_tcp_server_del_client(conn, fd);
+    conn->offset = 0;
+    memset(conn->buf, 0, conn->buf_size);
 
     pthread_mutex_unlock(&conn->mtx);
     return 0;
@@ -399,15 +401,9 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
         zlog_error(conn->param.log,
                    "conn fd: %d, recv buf len %zd, ret: %zd, errno: %s(%d)",
                    conn->fd, len, ret, strerror(errno), errno);
-    }
-
-    if (errno == EPIPE || ret == -1 || (ret == 0 && errno != 0)) {
-        conn_disconnect(conn);
-    }
-
-    if (ret > 0 && conn->callback_trigger == false) {
-        conn->connected(conn->data, conn->fd);
-        conn->callback_trigger = true;
+        if (errno == EPIPE || (ret == 0 && errno != 0)) {
+            conn_disconnect(conn);
+        }
     }
 
     pthread_mutex_unlock(&conn->mtx);
@@ -552,9 +548,9 @@ static void conn_tcp_server_listen(neu_conn_t *conn)
         conn->tcp_server.is_listen = true;
 
         conn->param.params.tcp_server.start_listen(conn->data, fd);
-        zlog_info(conn->param.log, "tcp server listen %s:%d success, fd: %d",
-                  conn->param.params.tcp_server.ip,
-                  conn->param.params.tcp_server.port, fd);
+        zlog_notice(conn->param.log, "tcp server listen %s:%d success, fd: %d",
+                    conn->param.params.tcp_server.ip,
+                    conn->param.params.tcp_server.port, fd);
     }
 }
 
@@ -562,9 +558,9 @@ static void conn_tcp_server_stop(neu_conn_t *conn)
 {
     if (conn->param.type == NEU_CONN_TCP_SERVER &&
         conn->tcp_server.is_listen == true) {
-        zlog_info(conn->param.log, "tcp server close %s:%d, fd: %d",
-                  conn->param.params.tcp_server.ip,
-                  conn->param.params.tcp_server.port, conn->fd);
+        zlog_notice(conn->param.log, "tcp server close %s:%d, fd: %d",
+                    conn->param.params.tcp_server.ip,
+                    conn->param.params.tcp_server.port, conn->fd);
 
         conn->param.params.tcp_server.stop_listen(conn->data, conn->fd);
 
@@ -624,9 +620,9 @@ static void conn_connect(neu_conn_t *conn)
             conn->is_connected = false;
             return;
         } else {
-            zlog_info(conn->param.log, "connect %s:%d success",
-                      conn->param.params.tcp_client.ip,
-                      conn->param.params.tcp_client.port);
+            zlog_notice(conn->param.log, "connect %s:%d success",
+                        conn->param.params.tcp_client.ip,
+                        conn->param.params.tcp_client.port);
             conn->is_connected = true;
             conn->fd           = fd;
         }
@@ -679,9 +675,9 @@ static void conn_connect(neu_conn_t *conn)
             conn->is_connected = false;
             return;
         } else {
-            zlog_info(conn->param.log, "connect %s:%d success",
-                      conn->param.params.udp.dst_ip,
-                      conn->param.params.udp.dst_port);
+            zlog_notice(conn->param.log, "connect %s:%d success",
+                        conn->param.params.udp.dst_ip,
+                        conn->param.params.udp.dst_port);
             conn->is_connected = true;
             conn->fd           = fd;
         }
@@ -846,6 +842,14 @@ static void conn_connect(neu_conn_t *conn)
 
 static void conn_disconnect(neu_conn_t *conn)
 {
+    conn->is_connected = false;
+    if (conn->callback_trigger == true) {
+        conn->disconnected(conn->data, conn->fd);
+        conn->callback_trigger = false;
+    }
+    conn->offset = 0;
+    memset(conn->buf, 0, conn->buf_size);
+
     switch (conn->param.type) {
     case NEU_CONN_TCP_SERVER:
         for (int i = 0; i < conn->param.params.tcp_server.max_link; i++) {
@@ -865,14 +869,6 @@ static void conn_disconnect(neu_conn_t *conn)
         }
         break;
     }
-
-    conn->is_connected = false;
-    if (conn->callback_trigger == true) {
-        conn->disconnected(conn->data, conn->fd);
-        conn->callback_trigger = false;
-    }
-    conn->offset = 0;
-    memset(conn->buf, 0, conn->buf_size);
 }
 
 static void conn_tcp_server_add_client(neu_conn_t *conn, int fd,
@@ -940,8 +936,7 @@ int neu_conn_stream_consume(neu_conn_t *conn, void *context,
             if (used == 0) {
                 break;
             } else if (used == -1) {
-                memset(conn->buf, 0, conn->buf_size);
-                conn->offset = 0;
+                neu_conn_disconnect(conn);
                 break;
             }
         }
@@ -972,8 +967,7 @@ int neu_conn_stream_tcp_server_consume(neu_conn_t *conn, int fd, void *context,
             if (used == 0) {
                 break;
             } else if (used == -1) {
-                memset(conn->buf, 0, conn->buf_size);
-                conn->offset = 0;
+                neu_conn_tcp_server_close_client(conn, fd);
                 break;
             }
         }
@@ -1001,9 +995,20 @@ int neu_conn_wait_msg(neu_conn_t *conn, void *context, uint16_t n_byte,
         neu_protocol_unpack_buf_init(&pbuf, conn->buf, conn->offset);
         neu_buf_result_t result = fn(context, &pbuf);
         if (result.need > 0) {
-            assert(result.need <= conn->buf_size - conn->offset);
-            ret = neu_conn_recv(conn, conn->buf + conn->offset, result.need);
+            if (result.need <= conn->buf_size - conn->offset) {
+                ret =
+                    neu_conn_recv(conn, conn->buf + conn->offset, result.need);
+            } else {
+                zlog_error(conn->param.log,
+                           "no enough recv buf, need: %" PRIu16
+                           " , buf size: %" PRIu16 " offset: %" PRIu16,
+                           result.need, conn->buf_size, conn->offset);
+                return -1;
+            }
+        } else if (result.need == 0) {
+            return result.used;
         } else {
+            neu_conn_disconnect(conn);
             return result.used;
         }
     }
@@ -1025,9 +1030,20 @@ int neu_conn_tcp_server_wait_msg(neu_conn_t *conn, int fd, void *context,
         neu_buf_result_t result = fn(context, &pbuf);
         if (result.need > 0) {
             assert(result.need <= conn->buf_size - conn->offset);
-            ret = neu_conn_tcp_server_recv(conn, fd, conn->buf + conn->offset,
-                                           result.need);
+            if (result.need <= conn->buf_size - conn->offset) {
+                ret = neu_conn_tcp_server_recv(
+                    conn, fd, conn->buf + conn->offset, result.need);
+            } else {
+                zlog_error(conn->param.log,
+                           "no enough recv buf, need: %" PRIu16
+                           " , buf size: %" PRIu16 " offset: %" PRIu16,
+                           result.need, conn->buf_size, conn->offset);
+                return -1;
+            }
+        } else if (result.need == 0) {
+            return result.used;
         } else {
+            neu_conn_tcp_server_close_client(conn, fd);
             return result.used;
         }
     }

src/connection/mqtt_client.c

@@ -287,14 +287,14 @@ static void task_handle_sub(task_t *task, neu_mqtt_client_t *client)
     }
 
     rc = nng_mqtt_msg_get_suback_return_codes(msg, &n);
-    log(info, "recv SUBACK return_code:%d", *rc);
+    log(debug, "recv SUBACK return_code:%d", *rc);
     rv             = 0x80 == *rc ? -1 : 0;
     task->sub->ack = true;
     client->suback_count += 1;
 
 end:
     // TODO: handle retry subscribe on failure
-    log(info, "sub [%s, QoS%d], %s", task->sub->topic, task->sub->qos,
+    log(notice, "sub [%s, QoS%d], %s", task->sub->topic, task->sub->qos,
         (0 == rv) ? "OK" : "FAIL");
     nng_msg_free(msg);
     return;
@@ -326,10 +326,10 @@ static void task_handle_unsub(task_t *task, neu_mqtt_client_t *client)
 
 end:
     if (task->sub) {
-        log(info, "unsub [%s, QoS%d], %s", task->sub->topic, task->sub->qos,
+        log(notice, "unsub [%s, QoS%d], %s", task->sub->topic, task->sub->qos,
             (0 == rv) ? "OK" : "FAIL");
     } else {
-        log(info, "unsub %s", (0 == rv) ? "OK" : "FAIL");
+        log(notice, "unsub %s", (0 == rv) ? "OK" : "FAIL");
     }
     nng_msg_free(msg);
     return;
@@ -431,7 +431,7 @@ static void connect_cb(nng_pipe p, nng_pipe_ev ev, void *arg)
     neu_mqtt_client_connection_cb_t cb     = NULL;
     void *                          data   = NULL;
 
-    log(info, "mqtt client connected");
+    log(notice, "mqtt client connected");
 
     nng_mtx_lock(client->mtx);
     // start receiving
@@ -457,7 +457,7 @@ static void disconnect_cb(nng_pipe p, nng_pipe_ev ev, void *arg)
     neu_mqtt_client_connection_cb_t cb     = NULL;
     void *                          data   = NULL;
 
-    log(info, "mqtt client disconnected");
+    log(notice, "mqtt client disconnected");
     nng_mtx_lock(client->mtx);
     client->connected = false;
     cb                = client->disconnect_cb;
@@ -511,7 +511,7 @@ static void recv_cb(void *arg)
     uint8_t     qos   = nng_mqtt_msg_get_publish_qos(msg);
 
     (void) payload;
-    log(info, "recv [%.*s, QoS%d] %" PRIu32 " bytes", (unsigned) topic_len,
+    log(debug, "recv [%.*s, QoS%d] %" PRIu32 " bytes", (unsigned) topic_len,
         topic, (int) qos, payload_len);
 
     // find matching subscription
@@ -565,7 +565,7 @@ static inline task_t *client_alloc_task(neu_mqtt_client_t *client)
     // the task queue length should be a small number due to reuse
     // log info every time the task queue length increments by 32
     if (0 == (client->task_count & (32 - 1))) {
-        log(info, "task queue length: %zu", client->task_count);
+        log(debug, "task queue length: %zu", client->task_count);
     }
 
     return task;
@@ -1017,7 +1017,7 @@ int neu_mqtt_client_set_tls(neu_mqtt_client_t *client, const char *ca,
 
     if (NULL == ca) {
         // disable tls
-        log(info, "tls disabled");
+        log(debug, "tls disabled");
         if (client->tls_cfg) {
             nng_tls_config_free(client->tls_cfg);
             client->tls_cfg = NULL;
@@ -1077,7 +1077,7 @@ int neu_mqtt_client_set_cache_size(neu_mqtt_client_t *client,
 
     if (0 == mem_size_bytes && 0 == db_size_bytes) {
         // disable cache
-        log(info, "cache disabled");
+        log(debug, "cache disabled");
         if (client->sqlite_cfg) {
             nng_mqtt_free_sqlite_opt(client->sqlite_cfg);
             client->sqlite_cfg = NULL;

src/core/manager.c

@@ -97,8 +97,9 @@ neu_manager_t *neu_manager_create()
     rv = nng_listen(manager->socket, url, NULL, 0);
     assert(rv == 0);
 
-    nng_socket_set_int(manager->socket, NNG_OPT_RECVBUF, 2048);
-    nng_socket_set_int(manager->socket, NNG_OPT_SENDBUF, 2048);
+    nng_socket_set_int(manager->socket, NNG_OPT_RECVBUF, 8192);
+    nng_socket_set_int(manager->socket, NNG_OPT_SENDBUF, 8292);
+    nng_socket_set_ms(manager->socket, NNG_OPT_SENDTIMEO, 1000);
     nng_socket_get_int(manager->socket, NNG_OPT_RECVFD, &param.fd);
     manager->loop = neu_event_add_io(manager->events, param);
 
@@ -135,6 +136,7 @@ neu_manager_t *neu_manager_create()
     manager->timer_timestamp =
         neu_event_add_timer(manager->events, timestamp_timer_param);
 
+    nlog_notice("manager start");
     return manager;
 }
 
@@ -177,7 +179,7 @@ void neu_manager_destroy(neu_manager_t *manager)
     neu_event_close(manager->events);
 
     free(manager);
-    nlog_warn("manager exit");
+    nlog_notice("manager exit");
 }
 
 const char *neu_manager_get_url()
@@ -228,8 +230,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
 
     header = (neu_reqresp_head_t *) nng_msg_body(msg);
 
-    nlog_debug("manager recv msg from: %s to %s, type: %s", header->sender,
-               header->receiver, neu_reqresp_type_string(header->type));
+    nlog_info("manager recv msg from: %s to %s, type: %s", header->sender,
+              header->receiver, neu_reqresp_type_string(header->type));
     switch (header->type) {
     case NEU_REQRESP_TRANS_DATA: {
         neu_reqresp_trans_data_t *cmd = (neu_reqresp_trans_data_t *) &header[1];
@@ -253,7 +255,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
         utarray_foreach(pipes, nng_pipe *, pipe)
         {
             forward_msg_dup(manager, msg, *pipe);
-            nlog_info("forward license update to pipe: %d", pipe->id);
+            nlog_notice("forward license update to pipe: %d", pipe->id);
         }
         utarray_free(pipes);
 
@@ -729,8 +731,9 @@ inline static void forward_msg_dup(neu_manager_t *manager, nng_msg *msg,
     nng_msg_dup(&out_msg, msg);
     nng_msg_set_pipe(out_msg, pipe);
     if (nng_sendmsg(manager->socket, out_msg, 0) == 0) {
-        nlog_debug("forward msg to pipe %d", pipe.id);
+        nlog_info("forward msg to pipe %d", pipe.id);
     } else {
+        nlog_warn("forward msg to pipe %d fail", pipe.id);
         nng_msg_free(out_msg);
     }
 }
@@ -744,8 +747,9 @@ inline static void forward_msg(neu_manager_t *manager, nng_msg *msg,
     nng_msg_dup(&out_msg, msg);
     nng_msg_set_pipe(out_msg, pipe);
     if (nng_sendmsg(manager->socket, out_msg, 0) == 0) {
-        nlog_debug("forward msg to %s", node);
+        nlog_info("forward msg to %s", node);
     } else {
+        nlog_warn("forward msg to pipe (%s)%d fail", node, pipe.id);
         nng_msg_free(out_msg);
     }
 }

src/core/manager_internal.c

@@ -324,8 +324,10 @@ int neu_manager_send_subscribe(neu_manager_t *manager, const char *app,
     nng_msg_set_pipe(out_msg, pipe);
 
     if (nng_sendmsg(manager->socket, out_msg, 0) == 0) {
-        nlog_info("send %s to %s", neu_reqresp_type_string(header.type), app);
+        nlog_notice("send %s to %s", neu_reqresp_type_string(header.type), app);
     } else {
+        nlog_warn("send %s to %s failed", neu_reqresp_type_string(header.type),
+                  app);
         free(cmd.params);
         nng_msg_free(out_msg);
     }

src/core/storage.c

@@ -96,7 +96,7 @@ int manager_load_plugin(neu_manager_t *manager)
     {
         rv                    = neu_manager_add_plugin(manager, *name);
         const char *ok_or_err = (0 == rv) ? "success" : "fail";
-        nlog_info("load plugin %s, lib:%s", ok_or_err, *name);
+        nlog_notice("load plugin %s, lib:%s", ok_or_err, *name);
     }
 
     utarray_foreach(plugin_infos, char **, name) { free(*name); }
@@ -122,9 +122,9 @@ int manager_load_node(neu_manager_t *manager)
             manager, node_info->name, node_info->plugin_name,
             node_info->state == NEU_NODE_RUNNING_STATE_RUNNING);
         const char *ok_or_err = (0 == rv) ? "success" : "fail";
-        nlog_info("load adapter %s type:%d, name:%s plugin:%s state:%d",
-                  ok_or_err, node_info->type, node_info->name,
-                  node_info->plugin_name, node_info->state);
+        nlog_notice("load adapter %s type:%d, name:%s plugin:%s state:%d",
+                    ok_or_err, node_info->type, node_info->name,
+                    node_info->plugin_name, node_info->state);
     }
 
     utarray_free(node_infos);
@@ -151,9 +151,9 @@ int manager_load_subscribe(neu_manager_t *manager)
                                            info->driver_name, info->group_name,
                                            info->params);
                 const char *ok_or_err = (0 == rv) ? "success" : "fail";
-                nlog_info("%s load subscription app:%s driver:%s grp:%s",
-                          ok_or_err, node->node, info->driver_name,
-                          info->group_name);
+                nlog_notice("%s load subscription app:%s driver:%s grp:%s",
+                            ok_or_err, node->node, info->driver_name,
+                            info->group_name);
                 if (0 == rv) {
                     neu_manager_send_subscribe(manager, node->node,
                                                info->driver_name,

src/daemon.c

@@ -129,6 +129,8 @@ int neuron_already_running()
 
     snprintf(buf, sizeof(buf), "%ld", (long) getpid());
     size = write(fd, buf, strlen(buf) + 1);
+    nlog_warn("write %s, error %s(%d)", NEURON_DAEMON_LOCK_FNAME,
+              strerror(errno), errno);
     assert(size != -1);
 
     return 0;

src/event/event_linux.c

@@ -205,19 +205,20 @@ neu_event_timer_t *neu_event_add_timer(neu_events_t *          events,
 
     ret = epoll_ctl(events->epoll_fd, EPOLL_CTL_ADD, timer_fd, &event);
 
-    zlog_info(neuron,
-              "add timer, second: %" PRId64 ", millisecond: %" PRId64
-              ", timer: %d in epoll %d, "
-              "ret: %d",
-              timer.second, timer.millisecond, timer_fd, events->epoll_fd, ret);
+    zlog_notice(neuron,
+                "add timer, second: %" PRId64 ", millisecond: %" PRId64
+                ", timer: %d in epoll %d, "
+                "ret: %d",
+                timer.second, timer.millisecond, timer_fd, events->epoll_fd,
+                ret);
 
     return timer_ctx;
 }
 
 int neu_event_del_timer(neu_events_t *events, neu_event_timer_t *timer)
 {
-    zlog_info(neuron, "del timer: %d from epoll: %d", timer->fd,
-              events->epoll_fd);
+    zlog_notice(neuron, "del timer: %d from epoll: %d", timer->fd,
+                events->epoll_fd);
 
     timer->stop = true;
     nng_mtx_lock(timer->mtx);
@@ -251,8 +252,8 @@ neu_event_io_t *neu_event_add_io(neu_events_t *events, neu_event_io_param_t io)
 
     ret = epoll_ctl(events->epoll_fd, EPOLL_CTL_ADD, io.fd, &event);
 
-    nlog_info("add io, fd: %d, epoll: %d, ret: %d", io.fd, events->epoll_fd,
-              ret);
+    nlog_notice("add io, fd: %d, epoll: %d, ret: %d", io.fd, events->epoll_fd,
+                ret);
 
     return io_ctx;
 }
@@ -260,7 +261,7 @@ neu_event_io_t *neu_event_add_io(neu_events_t *events, neu_event_io_param_t io)
 int neu_event_del_io(neu_events_t *events, neu_event_io_t *io)
 {
     struct event_data *data = (struct event_data *) io->event_data;
-    zlog_info(neuron, "del io: %d from epoll: %d", io->fd, events->epoll_fd);
+    zlog_notice(neuron, "del io: %d from epoll: %d", io->fd, events->epoll_fd);
 
     epoll_ctl(events->epoll_fd, EPOLL_CTL_DEL, io->fd, NULL);
     switch (data->type) {

src/main.c

@@ -34,7 +34,7 @@ static bool           exit_flag         = false;
 static neu_manager_t *g_manager         = NULL;
 zlog_category_t *     neuron            = NULL;
 bool                  disable_jwt       = false;
-int                   default_log_level = ZLOG_LEVEL_INFO;
+int                   default_log_level = ZLOG_LEVEL_NOTICE;
 
 int64_t global_timestamp = 0;
 

src/parser/neu_json_tag.c

@@ -199,7 +199,8 @@ int neu_json_tag_check_type(neu_json_tag_t *tag)
             return NEU_JSON_BOOL == tag->t;
         case NEU_TYPE_BYTES:
         case NEU_TYPE_STRING:
-            return NEU_JSON_STR == tag->t;
+            return NEU_JSON_STR == tag->t &&
+                strlen(tag->value.val_str) <= NEU_VALUE_SIZE;
         }
     } else {
         // non-static tag should have no initial value

src/persist/persist.c

@@ -383,8 +383,8 @@ static int apply_schema_file(sqlite3 *db, const char *dir, const char *file)
 
 end:
     if (0 == rv) {
-        nlog_info("success apply schema `%s`, version=`%s` description=`%s`",
-                  path, version, description);
+        nlog_notice("success apply schema `%s`, version=`%s` description=`%s`",
+                    path, version, description);
     } else {
         nlog_error("fail apply schema `%s`, version=`%s` description=`%s`",
                    path, version, description);
@@ -441,7 +441,7 @@ static int apply_schemas(sqlite3 *db, const char *dir)
         return NEU_ERR_EINTERNAL;
     }
 
-    nlog_info("schema head version=%s", version ? version : "none");
+    nlog_notice("schema head version=%s", version ? version : "none");
 
     if (dirty) {
         nlog_error("database is dirty, need manual intervention");
@@ -825,7 +825,7 @@ int neu_persister_store_tags(const char *driver_name, const char *group_name,
     return 0;
 
 error:
-    nlog_info("rollback transaction");
+    nlog_warn("rollback transaction");
     sqlite3_exec(global_db, "ROLLBACK", NULL, NULL, NULL);
     sqlite3_finalize(stmt);
     free(val_str);

src/utils/http.c

@@ -295,6 +295,8 @@ int neu_http_response(nng_aio *aio, neu_err_code_e code, char *content)
     case NEU_ERR_MQTT_CONNECT_FAILURE:
     case NEU_ERR_MQTT_SUBSCRIBE_FAILURE:
     case NEU_ERR_MQTT_PUBLISH_FAILURE:
+    case NEU_ERR_STRING_TOO_LONG:
+    case NEU_ERR_FILE_WRITE_FAILURE:
         status = NNG_HTTP_STATUS_OK;
         break;
     case NEU_ERR_EINTERNAL:

version

@@ -1 +1 @@
-2.4.5
+2.4.8