Merge pull request #1364 from lixiumei123/v2.4

version:2.4.9

.github/workflows/ci.yml

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

src/adapter/driver/driver.c

@@ -55,6 +55,7 @@ typedef struct group {
 
     neu_event_timer_t *report;
     neu_event_timer_t *read;
+    neu_event_timer_t *write;
 
     neu_plugin_group_t    grp;
     neu_adapter_driver_t *driver;
@@ -73,6 +74,7 @@ struct neu_adapter_driver {
 
 static int  report_callback(void *usr_data);
 static int  read_callback(void *usr_data);
+static int  write_callback(void *usr_data);
 static void read_group(int64_t timestamp, int64_t timeout,
                        neu_driver_cache_t *cache, const char *group,
                        UT_array *tags, neu_resp_tag_value_t *datas);
@@ -185,6 +187,7 @@ int neu_adapter_driver_uninit(neu_adapter_driver_t *driver)
 
         neu_adapter_del_timer((neu_adapter_t *) driver, el->report);
         neu_event_del_timer(driver->driver_events, el->read);
+        neu_event_del_timer(driver->driver_events, el->write);
         if (el->grp.group_free != NULL) {
             el->grp.group_free(&el->grp);
         }
@@ -481,6 +484,11 @@ int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name,
         param.cb     = read_callback;
         find->read   = neu_event_add_timer(driver->driver_events, param);
 
+        param.second      = 0;
+        param.millisecond = 100;
+        param.cb          = write_callback;
+        find->write       = neu_event_add_timer(driver->driver_events, param);
+
         REGISTER_GROUP_METRIC(&driver->adapter, find->name,
                               NEU_METRIC_GROUP_TAGS_TOTAL,
                               neu_group_tag_size(find->group));
@@ -539,6 +547,7 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver, const char *name)
 
         neu_adapter_del_timer((neu_adapter_t *) driver, find->report);
         neu_event_del_timer(driver->driver_events, find->read);
+        neu_event_del_timer(driver->driver_events, find->write);
         if (find->grp.group_free != NULL) {
             find->grp.group_free(&find->grp);
         }
@@ -917,7 +926,7 @@ static void group_change(void *arg, int64_t timestamp, UT_array *static_tags,
                 timestamp);
 }
 
-static int read_callback(void *usr_data)
+static int write_callback(void *usr_data)
 {
     group_t *                group = (group_t *) usr_data;
     neu_node_running_state_e state = group->driver->adapter.state;
@@ -936,6 +945,17 @@ static int read_callback(void *usr_data)
     utarray_clear(group->wt_tags);
     pthread_mutex_unlock(&group->wt_mtx);
 
+    return 0;
+}
+
+static int read_callback(void *usr_data)
+{
+    group_t *                group = (group_t *) usr_data;
+    neu_node_running_state_e state = group->driver->adapter.state;
+    if (state != NEU_NODE_RUNNING_STATE_RUNNING) {
+        return 0;
+    }
+
     if (neu_group_is_change(group->group, group->timestamp)) {
         neu_group_change_test(group->group, group->timestamp, (void *) group,
                               group_change);

src/connection/connection.c

@@ -47,6 +47,7 @@ struct neu_conn {
     void *           data;
     bool             is_connected;
     bool             stop;
+    bool             connection_ok;
 
     neu_conn_callback connected;
     neu_conn_callback disconnected;
@@ -340,8 +341,14 @@ ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len)
                        conn->fd, len, ret, strerror(errno), errno);
         }
 
-        if (ret == -1 && errno != EAGAIN) {
-            conn_disconnect(conn);
+        if (ret == -1) {
+            if (errno != EAGAIN) {
+                conn_disconnect(conn);
+            } else {
+                if (conn->connection_ok == true) {
+                    conn_disconnect(conn);
+                }
+            }
         }
 
         if (ret > 0 && conn->callback_trigger == false) {
@@ -352,6 +359,7 @@ ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len)
 
     if (ret > 0) {
         conn->state.send_bytes += ret;
+        conn->connection_ok = true;
     }
 
     pthread_mutex_unlock(&conn->mtx);
@@ -397,13 +405,23 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
         }
         break;
     }
-    if (ret <= 0) {
-        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 == 0 && errno != 0)) {
+    if (conn->param.type == NEU_CONN_TTY_CLIENT) {
+        if (ret == -1) {
+            zlog_error(
+                conn->param.log,
+                "tty conn fd: %d, recv buf len %zd, ret: %zd, errno: %s(%d)",
+                conn->fd, len, ret, strerror(errno), errno);
             conn_disconnect(conn);
         }
+    } else {
+        if (ret <= 0) {
+            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 (ret == 0 || (ret == -1 && errno != EAGAIN)) {
+                conn_disconnect(conn);
+            }
+        }
     }
 
     pthread_mutex_unlock(&conn->mtx);
@@ -483,6 +501,10 @@ static void conn_init_param(neu_conn_t *conn, neu_conn_param_t *param)
         } else {
             conn->block = false;
         }
+        zlog_notice(conn->param.log, "tcp server %s:%d, timeout: %d, block: %d",
+                    conn->param.params.tcp_server.ip,
+                    conn->param.params.tcp_server.port,
+                    conn->param.params.tcp_server.timeout, conn->block);
         break;
     case NEU_CONN_TCP_CLIENT:
         conn->param.params.tcp_client.ip = strdup(param->params.tcp_client.ip);
@@ -490,6 +512,10 @@ static void conn_init_param(neu_conn_t *conn, neu_conn_param_t *param)
         conn->param.params.tcp_client.timeout =
             param->params.tcp_client.timeout;
         conn->block = conn->param.params.tcp_client.timeout > 0;
+        zlog_notice(conn->param.log, "tcp client %s:%d, timeout: %d, block: %d",
+                    conn->param.params.tcp_client.ip,
+                    conn->param.params.tcp_client.port,
+                    conn->param.params.tcp_client.timeout, conn->block);
         break;
     case NEU_CONN_UDP:
         conn->param.params.udp.src_ip   = strdup(param->params.udp.src_ip);
@@ -498,6 +524,11 @@ static void conn_init_param(neu_conn_t *conn, neu_conn_param_t *param)
         conn->param.params.udp.dst_port = param->params.udp.dst_port;
         conn->param.params.udp.timeout  = param->params.udp.timeout;
         conn->block                     = conn->param.params.udp.timeout > 0;
+        zlog_notice(
+            conn->param.log, "udp %s:%d -> %s:%d, timeout: %d, block: %d",
+            conn->param.params.udp.src_ip, conn->param.params.udp.src_port,
+            conn->param.params.udp.dst_ip, conn->param.params.udp.dst_port,
+            conn->param.params.udp.timeout, conn->block);
         break;
     case NEU_CONN_TTY_CLIENT:
         conn->param.params.tty_client.device =
@@ -620,9 +651,9 @@ static void conn_connect(neu_conn_t *conn)
             conn->is_connected = false;
             return;
         } else {
-            zlog_notice(conn->param.log, "connect %s:%d success",
+            zlog_notice(conn->param.log, "connect %s:%d, block: %d success",
                         conn->param.params.tcp_client.ip,
-                        conn->param.params.tcp_client.port);
+                        conn->param.params.tcp_client.port, conn->block);
             conn->is_connected = true;
             conn->fd           = fd;
         }
@@ -842,7 +873,8 @@ static void conn_connect(neu_conn_t *conn)
 
 static void conn_disconnect(neu_conn_t *conn)
 {
-    conn->is_connected = false;
+    conn->is_connected  = false;
+    conn->connection_ok = false;
     if (conn->callback_trigger == true) {
         conn->disconnected(conn->data, conn->fd);
         conn->callback_trigger = false;

version

@@ -1 +1 @@
-2.4.8
+2.4.9