Merge pull request #1063 from fengzeroz/v2.2

version: 2.2.14

.github/workflows/test-helm.yaml

@@ -17,7 +17,7 @@ jobs:
         run: |
           helm install neuron deploy/charts/neuron\
             --set image.repository="emqx/neuron"\
-            --set image.tag="2.0.1"
+            --set image.tag="2.2.14"
       - name: Check deployment
         timeout-minutes: 5
         run: |

include/neuron/define.h

@@ -20,12 +20,12 @@
 #ifndef _NEU_DEFINE_H_
 #define _NEU_DEFINE_H_
 
-#define NEU_TAG_NAME_LEN 32
+#define NEU_TAG_NAME_LEN 64
 #define NEU_TAG_ADDRESS_LEN 128
 #define NEU_TAG_DESCRIPTION_LEN 128
-#define NEU_GROUP_NAME_LEN 32
+#define NEU_GROUP_NAME_LEN 128
 #define NEU_GROUP_INTERVAL_LIMIT 100
-#define NEU_NODE_NAME_LEN 32
+#define NEU_NODE_NAME_LEN 128
 #define NEU_PLUGIN_NAME_LEN 32
 #define NEU_PLUGIN_LIBRARY_LEN 32
 #define NEU_PLUGIN_DESCRIPTION_LEN 512

persistence/0001_2.2.1.sql

@@ -0,0 +1,64 @@
+
+BEGIN TRANSACTION;
+
+ALTER TABLE nodes RENAME TO old_nodes;
+CREATE TABLE
+  nodes (
+    name TEXT PRIMARY KEY check(length(name) <= 128),
+    type integer(1) NOT NULL check(type IN (1, 2)),
+    state integer(1) NOT NULL check(state BETWEEN 1 AND 4),
+    plugin_name TEXT NOT NULL check(length(plugin_name) <= 32)
+  );
+INSERT INTO nodes SELECT * FROM old_nodes;
+
+ALTER TABLE groups RENAME TO old_groups;
+CREATE TABLE IF NOT EXISTS
+  groups (
+    driver_name TEXT NOT NULL,
+    name TEXT NULL check(length(name) <= 128),
+    interval INTEGER NOT NULL check(interval >= 100),
+    UNIQUE (driver_name, name),
+    FOREIGN KEY (driver_name) REFERENCES nodes (name) ON UPDATE CASCADE ON DELETE CASCADE
+  );
+INSERT INTO groups SELECT * FROM old_groups;
+
+ALTER TABLE tags RENAME TO old_tags;
+CREATE TABLE IF NOT EXISTS
+  tags (
+    driver_name TEXT NOT NULL,
+    group_name TEXT NOT NULL,
+    name TEXT NULL check(length(name) <= 64),
+    address TEXT NULL check(length(address) <= 128),
+    attribute INTEGER NOT NULL check(attribute BETWEEN 0 AND 7),
+    precision INTEGER NOT NULL check(precision BETWEEN 0 AND 17),
+    decimal REAL NOT NULL,
+    type INTEGER NOT NULL check(type BETWEEN 0 AND 19),
+    description TEXT NULL check(length(description) <= 128),
+    UNIQUE (driver_name, group_name, name),
+    FOREIGN KEY (driver_name, group_name) REFERENCES groups (driver_name, name) ON UPDATE CASCADE ON DELETE CASCADE
+  );
+INSERT INTO tags SELECT * FROM old_tags;
+
+ALTER TABLE subscriptions RENAME TO old_subscriptions;
+CREATE TABLE IF NOT EXISTS
+  subscriptions (
+    app_name TEXT NOT NULL,
+    driver_name TEXT NOT NULL,
+    group_name TEXT NOT NULL,
+    CHECK (app_name != driver_name),
+    UNIQUE (app_name, driver_name, group_name),
+    FOREIGN KEY (app_name) REFERENCES nodes (name) ON UPDATE CASCADE ON DELETE CASCADE,
+    FOREIGN KEY (driver_name, group_name) REFERENCES groups (driver_name, name) ON UPDATE CASCADE ON DELETE CASCADE
+  );
+INSERT INTO subscriptions SELECT * FROM old_subscriptions;
+
+ALTER TABLE settings RENAME TO old_settings;
+CREATE TABLE IF NOT EXISTS
+  settings (
+    node_name TEXT NOT NULL,
+    setting TEXT NOT NULL,
+    UNIQUE (node_name),
+    FOREIGN KEY (node_name) REFERENCES nodes (name) ON UPDATE CASCADE ON DELETE CASCADE
+  );
+INSERT INTO settings SELECT * FROM old_settings;
+COMMIT;

plugins/ekuiper/plugin_ekuiper.c

@@ -114,6 +114,8 @@ static int ekuiper_plugin_start(neu_plugin_t *plugin)
 
     nng_pipe_notify(plugin->sock, NNG_PIPE_EV_ADD_POST, pipe_add_cb, plugin);
     nng_pipe_notify(plugin->sock, NNG_PIPE_EV_REM_POST, pipe_rm_cb, plugin);
+    nng_socket_set_int(plugin->sock, NNG_OPT_SENDBUF, 2048);
+    nng_socket_set_int(plugin->sock, NNG_OPT_RECVBUF, 2048);
 
     if ((rv = nng_listen(plugin->sock, EKUIPER_PLUGIN_URL, NULL, 0)) != 0) {
         plog_error(plugin, "nng_listen: %s", nng_strerror(rv));

plugins/ekuiper/read_write.c

@@ -56,7 +56,7 @@ void send_data(neu_plugin_t *plugin, neu_reqresp_trans_data_t *trans_data)
     rv = nng_sendmsg(plugin->sock, msg,
                      NNG_FLAG_NONBLOCK); // TODO: use aio to send message
     if (0 != rv) {
-        plog_error(plugin, "nng cannot send msg");
+        plog_error(plugin, "nng cannot send msg: %s", nng_strerror(rv));
         nng_msg_free(msg);
     }
 }

plugins/restful/adapter_handle.c

@@ -37,7 +37,7 @@ void handle_add_adapter(nng_aio *aio)
 
     REST_PROCESS_HTTP_REQUEST_VALIDATE_JWT(
         aio, neu_json_add_node_req_t, neu_json_decode_add_node_req, {
-            if (strlen(req->name) > NEU_NODE_NAME_LEN) {
+            if (strlen(req->name) >= NEU_NODE_NAME_LEN) {
                 NEU_JSON_RESPONSE_ERROR(NEU_ERR_NODE_NAME_TOO_LONG, {
                     http_response(aio, NEU_ERR_NODE_NAME_TOO_LONG,
                                   result_error);

plugins/restful/group_config_handle.c

@@ -36,7 +36,7 @@ void handle_add_group_config(nng_aio *aio)
     REST_PROCESS_HTTP_REQUEST_VALIDATE_JWT(
         aio, neu_json_add_group_config_req_t,
         neu_json_decode_add_group_config_req, {
-            if (strlen(req->group) > NEU_GROUP_NAME_LEN) {
+            if (strlen(req->group) >= NEU_GROUP_NAME_LEN) {
                 NEU_JSON_RESPONSE_ERROR(NEU_ERR_GROUP_NAME_TOO_LONG, {
                     http_response(aio, NEU_ERR_GROUP_NAME_TOO_LONG,
                                   result_error);

plugins/restful/license.c

@@ -36,6 +36,7 @@
 #include <time.h>
 
 #include "errcodes.h"
+#include "utils/asprintf.h"
 #include "utils/log.h"
 
 #include "license.h"
@@ -81,7 +82,7 @@ static uint64_t mktimestamp(struct tm dt)
     return (ret);
 }
 
-static uint64_t get_asn1_ts(ASN1_TIME *time)
+static uint64_t get_asn1_ts(ASN1_TIME *time, char **ts_str)
 {
     struct tm   t;
     const char *str = (const char *) time->data;
@@ -97,38 +98,46 @@ static uint64_t get_asn1_ts(ASN1_TIME *time)
                &t.tm_hour, &t.tm_min, &t.tm_sec);
     }
 
+    neu_asprintf(ts_str, "%4d-%02d-%02d %02d:%02d:%02d", t.tm_year, t.tm_mon,
+                 t.tm_mday, t.tm_hour, t.tm_min, t.tm_sec);
+
     t.tm_mon -= 1;
+
     /* Note: we did not adjust the time based on time zone information */
     return mktimestamp(t);
 }
 
 static int scan_bitvec(uint8_t vec[], size_t size, const char *hex)
 {
-    if ('0' == *hex && 'x' == tolower(*(hex + 1))) {
-        hex += 2;
+    // 0xFFFFFFFFFFFFF
+    // skip 0x
+    char *tmp   = strdup(hex);
+    char *start = tmp;
+    int   index = 0;
+
+    tmp += 2;
+    if (strlen(tmp) % 2 == 1) {
+        tmp -= 1;
+        tmp[0] = '0';
     }
 
-    size_t i   = 0;
-    size_t len = strlen(hex);
-
-    while (i < len && i / 2 < size) {
-        char    c = hex[len - 1 - i];
+    for (int i = strlen(tmp) - 2; i >= 0; i -= 2) {
         uint8_t v = 0;
-        if ('0' <= c && c <= '9') {
-            v = c - '0';
-        } else if ('a' <= c && c <= 'f') {
-            v = 10 + c - 'a';
-        } else if ('A' <= c && c <= 'F') {
-            v = 10 + c - 'A';
-        } else {
-            break;
-        }
 
-        vec[i / 2] += v << ((i & 0x01) * 4);
-        ++i;
+        if (sscanf(tmp + i, "%02hhX", &v) != 1) {
+            free(start);
+            return -1;
+        }
+        if ((size_t) index > size) {
+            free(start);
+            return -1;
+        }
+        vec[index] = v;
+        index += 1;
     }
 
-    return (i == len) ? 0 : -1;
+    free(start);
+    return 0;
 }
 
 void print_bitvec(char *s, size_t size, uint8_t vec[], size_t n)
@@ -245,9 +254,14 @@ static int extract_license(const char *license_fname, const char *pub_key,
 
     not_before      = X509_get_notBefore(cert);
     not_after       = X509_get_notAfter(cert);
-    license->since_ = get_asn1_ts(not_before);
-    license->until_ = get_asn1_ts(not_after);
+    license->since_ = get_asn1_ts(not_before, &license->since_str_);
+    license->until_ = get_asn1_ts(not_after, &license->until_str_);
     nlog_info("certificate expire date: %s", not_after->data);
+    if (NULL == license->since_str_ || NULL == license->until_str_) {
+        nlog_warn("cannot strdup");
+        rv = NEU_ERR_EINTERNAL;
+        goto final;
+    }
 
     subj = X509_get_subject_name(cert);
     for (i = 0; i < X509_NAME_entry_count(subj); i++) {
@@ -273,6 +287,7 @@ static int extract_license(const char *license_fname, const char *pub_key,
         goto final;
     }
 
+    unsigned mark = 0;
     for (i = 0; i < sk_X509_EXTENSION_num(ext_list); i++) {
         ext = sk_X509_EXTENSION_value(ext_list, i);
         obj = X509_EXTENSION_get_object(ext);
@@ -286,25 +301,36 @@ static int extract_license(const char *license_fname, const char *pub_key,
         if (0 == strcmp(buf, LICENSE_KEY_TYPE)) {
             license->type_ = scan_licese_type(val);
             if (LICENSE_TYPE_MAX == license->type_) {
-                zlog_error(neuron, "invalid license type:`%s`", val);
+                nlog_error("invalid license type:`%s`", val);
                 rv = NEU_ERR_LICENSE_INVALID;
             }
+            mark |= 0x01;
         } else if (0 == strcmp(buf, LICENSE_KEY_MAX_NODES)) {
             license->max_nodes_ = strtoul(val, NULL, 0);
+            mark |= 0x02;
         } else if (0 == strcmp(buf, LICENSE_KEY_MAX_NODE_TAGS)) {
             license->max_node_tags_ = strtoul(val, NULL, 0);
+            mark |= 0x04;
         } else if (0 == strcmp(buf, LICENSE_KEY_PLUGIN_FLAG)) {
             if (0 !=
                 scan_bitvec(license->plugin_flag_,
                             sizeof(license->plugin_flag_), val)) {
-                zlog_error(neuron, "fail to scan license plugin flag:`%s`",
-                           val);
+                nlog_error("fail to scan license plugin flag:`%s`", val);
                 rv = NEU_ERR_LICENSE_INVALID;
             }
+            mark |= 0x08;
+        } else if (0 == strcmp(buf, LICENSE_KEY_TOKEN)) {
+            license->token_ = strdup(val);
+            mark |= 0x10;
         }
     }
 
-    nlog_info("certification verification success");
+    if (0x0F == (mark & 0x0F)) {
+        nlog_info("certification verification success");
+    } else {
+        rv = NEU_ERR_LICENSE_INVALID;
+        nlog_error("missing license key fields, mark:%u", mark);
+    }
 
 final:
     if (pkey)
@@ -330,26 +356,15 @@ const char *plugin_bit_str(plugin_bit_e b)
 
 void license_print(license_t *license)
 {
-    struct tm   tm        = {};
-    char        start[16] = {};
-    char        end[16]   = {};
-    time_t      since     = license->since_;
-    time_t      until     = license->until_;
-    const char *fmt       = "%Y%m%dT%H%M%S";
-
-    gmtime_r(&since, &tm);
-    strftime(start, sizeof(start), fmt, &tm);
-    gmtime_r(&until, &tm);
-    strftime(end, sizeof(end), fmt, &tm);
-
     char flag[sizeof(license->plugin_flag_) * 2 + 1] = {};
     print_bitvec(flag, sizeof(flag), license->plugin_flag_,
                  sizeof(license->plugin_flag_));
 
     nlog_info("license `%s` for %s, from %s to %s, max_nodes:%" PRIu32
               ", max_node_tags:%" PRIu32 ", flag:0x%s",
-              license->fname_, license_type_str(license->type_), start, end,
-              license->max_nodes_, license->max_node_tags_, flag);
+              license->fname_, license_type_str(license->type_),
+              license->since_str_, license->until_str_, license->max_nodes_,
+              license->max_node_tags_, flag);
 }
 
 void license_init(license_t *license)
@@ -363,7 +378,10 @@ void license_init(license_t *license)
 
 void license_fini(license_t *license)
 {
+    free(license->since_str_);
+    free(license->until_str_);
     free(license->fname_);
+    free(license->token_);
 }
 
 license_t *license_new()
@@ -410,7 +428,10 @@ int license_read(license_t *license, const char *fname)
 
     license_print(&lic);
 
+    free(license->since_str_);
+    free(license->until_str_);
     free(license->fname_);
+    free(license->token_);
     *license = lic;
 
     return rv;

plugins/restful/license.h

@@ -50,6 +50,8 @@
 //
 #define LICENSE_KEY_PLUGIN_FLAG "1.3.6.1.4.1.52509.13"
 
+#define LICENSE_KEY_TOKEN "1.3.6.1.4.1.52509.14"
+
 #ifdef __cplusplus
 extern "C" {
 #endif
@@ -91,10 +93,13 @@ typedef struct {
     license_type_e type_;                 // license type
     char *         fname_;                // license file name
     uint64_t       since_;                // not before
+    char *         since_str_;            //
     uint64_t       until_;                // not after
+    char *         until_str_;            //
     uint32_t       max_nodes_;            // max number of nodes
     uint32_t       max_node_tags_;        // max number of tags per node
     uint8_t        plugin_flag_[200 / 8]; // plugin enable/disable bits
+    char *         token_;                // hardware token
 } license_t;
 
 // Initialize to be expired with no file.

plugins/restful/license_handle.c

@@ -135,17 +135,12 @@ int copy_license_file_if_necessary()
 
 void handle_get_license(nng_aio *aio)
 {
-    int                         rv              = 0;
-    char *                      license_path    = NULL;
-    license_t                   lic             = {};
-    neu_json_get_license_resp_t resp            = {};
-    time_t                      since           = 0;
-    time_t                      until           = 0;
-    char                        valid_since[20] = {};
-    char                        valid_until[20] = {};
-    struct tm                   tm              = {};
-    UT_array *                  plugin_names    = NULL;
-    char *                      result          = NULL;
+    int                         rv           = 0;
+    char *                      license_path = NULL;
+    license_t                   lic          = {};
+    neu_json_get_license_resp_t resp         = {};
+    UT_array *                  plugin_names = NULL;
+    char *                      result       = NULL;
 
     VALIDATE_JWT(aio);
 
@@ -163,26 +158,10 @@ void handle_get_license(nng_aio *aio)
         goto final;
     }
 
-    since = license_not_before(&lic);
-    gmtime_r(&since, &tm);
-    if (0 ==
-        strftime(valid_since, sizeof(valid_since), "%Y-%m-%d %H:%M:%S", &tm)) {
-        rv = NEU_ERR_EINTERNAL;
-        goto final;
-    }
-
-    until = license_not_after(&lic);
-    gmtime_r(&until, &tm);
-    if (0 ==
-        strftime(valid_until, sizeof(valid_until), "%Y-%m-%d %H:%M:%S", &tm)) {
-        rv = NEU_ERR_EINTERNAL;
-        goto final;
-    }
-
     resp.license_type  = (char *) license_type_str(license_get_type(&lic));
     resp.valid         = !license_is_expired(&lic);
-    resp.valid_since   = valid_since;
-    resp.valid_until   = valid_until;
+    resp.valid_since   = lic.since_str_;
+    resp.valid_until   = lic.until_str_;
     resp.max_nodes     = license_get_max_nodes(&lic);
     resp.max_node_tags = license_get_max_node_tags(&lic);
 

src/adapter/driver/driver.c

@@ -502,15 +502,15 @@ int neu_adapter_driver_add_tag(neu_adapter_driver_t *driver, const char *group,
     int      ret  = NEU_ERR_SUCCESS;
     group_t *find = NULL;
 
-    if (strlen(tag->name) > NEU_TAG_NAME_LEN) {
+    if (strlen(tag->name) >= NEU_TAG_NAME_LEN) {
         return NEU_ERR_TAG_NAME_TOO_LONG;
     }
 
-    if (strlen(tag->address) > NEU_TAG_ADDRESS_LEN) {
+    if (strlen(tag->address) >= NEU_TAG_ADDRESS_LEN) {
         return NEU_ERR_TAG_ADDRESS_TOO_LONG;
     }
 
-    if (strlen(tag->description) > NEU_TAG_DESCRIPTION_LEN) {
+    if (strlen(tag->description) >= NEU_TAG_DESCRIPTION_LEN) {
         return NEU_ERR_TAG_DESCRIPTION_TOO_LONG;
     }
 
@@ -568,7 +568,7 @@ int neu_adapter_driver_update_tag(neu_adapter_driver_t *driver,
     int      ret  = NEU_ERR_SUCCESS;
     group_t *find = NULL;
 
-    if (strlen(tag->name) > NEU_TAG_NAME_LEN) {
+    if (strlen(tag->name) >= NEU_TAG_NAME_LEN) {
         return NEU_ERR_TAG_NAME_TOO_LONG;
     }
 

src/connection/connection.c

@@ -235,6 +235,11 @@ ssize_t neu_conn_tcp_server_send(neu_conn_t *conn, int fd, uint8_t *buf,
     conn_tcp_server_listen(conn);
     ret = send(fd, buf, len, MSG_NOSIGNAL | MSG_DONTWAIT);
 
+    if (ret <= 0 && errno != EAGAIN) {
+        conn->disconnected(conn->data, fd);
+        conn_tcp_server_del_client(conn, fd);
+    }
+
     pthread_mutex_unlock(&conn->mtx);
 
     return ret;
@@ -257,6 +262,11 @@ ssize_t neu_conn_tcp_server_recv(neu_conn_t *conn, int fd, uint8_t *buf,
         ret = recv(fd, buf, len, 0);
     }
 
+    if (ret <= 0) {
+        conn->disconnected(conn->data, fd);
+        conn_tcp_server_del_client(conn, fd);
+    }
+
     pthread_mutex_unlock(&conn->mtx);
 
     return ret;
@@ -509,6 +519,7 @@ static void conn_tcp_server_stop(neu_conn_t *conn)
             conn->fd = 0;
         }
 
+        conn->tcp_server.n_client  = 0;
         conn->tcp_server.is_listen = false;
     }
 }
@@ -758,6 +769,7 @@ static void conn_disconnect(neu_conn_t *conn)
                 conn->tcp_server.clients[i].fd = 0;
             }
         }
+        conn->tcp_server.n_client = 0;
         break;
     case NEU_CONN_TCP_CLIENT:
     case NEU_CONN_UDP:
@@ -804,8 +816,6 @@ static void conn_tcp_server_del_client(neu_conn_t *conn, int fd)
             return;
         }
     }
-
-    assert(1 == 0);
 }
 
 void neu_conn_stream_consume(neu_conn_t *conn, void *context,

version

@@ -1 +1 @@
-2.2.7
+2.2.14