feature: session mananger publish ctrl message; Enhance session debugger module

1 files changed, 152 insertions, 31 deletions

diff --git a/test/session_debugger/session_debugger.c b/test/session_debugger/session_debugger.c
index 0af4081..48c028a 100644
--- a/test/session_debugger/session_debugger.c
+++ b/test/session_debugger/session_debugger.c
@@ -30,20 +30,41 @@ struct session_debugger_exdata
     struct session_debugger *dbg;
     struct session *sess;
 
-    uint64_t c2s_rx_pkts;
-    uint64_t s2c_rx_pkts;
+    // data packet
+    uint64_t c2s_rx_data_pkts;
+    uint64_t s2c_rx_data_pkts;
 
-    uint64_t c2s_rx_bytes;
-    uint64_t s2c_rx_bytes;
+    uint64_t c2s_rx_data_bytes;
+    uint64_t s2c_rx_data_bytes;
 
+    // control packet
+    uint64_t c2s_rx_ctrl_pkts;
+    uint64_t s2c_rx_ctrl_pkts;
+
+    uint64_t c2s_rx_ctrl_bytes;
+    uint64_t s2c_rx_ctrl_bytes;
+
+    // TCP segment
     uint64_t c2s_rx_tcp_seg;
     uint64_t s2c_rx_tcp_seg;
 
     uint64_t c2s_rx_tcp_bytes;
     uint64_t s2c_rx_tcp_bytes;
 
+    // UDP payload
+    uint64_t c2s_rx_udp_payload;
+    uint64_t s2c_rx_udp_payload;
+
+    uint64_t c2s_rx_udp_bytes;
+    uint64_t s2c_rx_udp_bytes;
+
+    // hexdump TCP segment
     int c2s_tcp_seg_hexdump_fd;
     int s2c_tcp_seg_hexdump_fd;
+
+    // hexdump UDP payload
+    int c2s_udp_payload_hexdump_fd;
+    int s2c_udp_payload_hexdump_fd;
 };
 
 static void session_debugger_log(int fd, const char *fmt, ...)
@@ -88,14 +109,25 @@ static struct session_debugger_exdata *session_debugger_exdata_new(struct sessio
     if (session_get_type(sess) == SESSION_TYPE_TCP)
     {
         memset(buff, 0, sizeof(buff));
-        sprintf(buff, "./log/session_debugger.%s_c2s.hexdump", session_get0_readable_addr(sess));
+        sprintf(buff, "./log/session_debugger.TCP_%s_C2S.hexdump", session_get0_readable_addr(sess));
         exdata->c2s_tcp_seg_hexdump_fd = open(buff, O_WRONLY | O_APPEND | O_CREAT, 0644);
 
         memset(buff, 0, sizeof(buff));
-        sprintf(buff, "./log/session_debugger.%s_s2c.hexdump", session_get0_readable_addr(sess));
+        sprintf(buff, "./log/session_debugger.TCP_%s_S2C.hexdump", session_get0_readable_addr(sess));
         exdata->s2c_tcp_seg_hexdump_fd = open(buff, O_WRONLY | O_APPEND | O_CREAT, 0644);
     }
 
+    if (session_get_type(sess) == SESSION_TYPE_UDP)
+    {
+        memset(buff, 0, sizeof(buff));
+        sprintf(buff, "./log/session_debugger.UDP_%s_C2S.hexdump", session_get0_readable_addr(sess));
+        exdata->c2s_udp_payload_hexdump_fd = open(buff, O_WRONLY | O_APPEND | O_CREAT, 0644);
+
+        memset(buff, 0, sizeof(buff));
+        sprintf(buff, "./log/session_debugger.UDP_%s_S2C.hexdump", session_get0_readable_addr(sess));
+        exdata->s2c_udp_payload_hexdump_fd = open(buff, O_WRONLY | O_APPEND | O_CREAT, 0644);
+    }
+
     memset(buff, 0, sizeof(buff));
     session_to_str(sess, 1, buff, sizeof(buff) - 1);
     session_debugger_log(dbg->fd, "sess new: %s", buff);
@@ -115,6 +147,14 @@ static void session_debugger_exdata_free(struct session_debugger_exdata *exdata)
         {
             close(exdata->s2c_tcp_seg_hexdump_fd);
         }
+        if (exdata->c2s_udp_payload_hexdump_fd > 0)
+        {
+            close(exdata->c2s_udp_payload_hexdump_fd);
+        }
+        if (exdata->s2c_udp_payload_hexdump_fd > 0)
+        {
+            close(exdata->s2c_udp_payload_hexdump_fd);
+        }
 
         free(exdata);
     }
@@ -128,7 +168,7 @@ static void session_debugger_exdata_free_callback(int idx, void *ex_ptr, void *a
     session_debugger_exdata_free((struct session_debugger_exdata *)ex_ptr);
 }
 
-static void on_sess_free(struct session *sess, void *arg)
+static void on_session_free(struct session *sess, void *arg)
 {
     struct session_debugger *dbg = (struct session_debugger *)arg;
     struct session_debugger_exdata *exdata = (struct session_debugger_exdata *)session_get_exdata(sess, dbg->sess_exdata_idx);
@@ -136,23 +176,39 @@ static void on_sess_free(struct session *sess, void *arg)
     char buff[PATH_MAX] = {0};
     session_to_str(exdata->sess, 0, buff, sizeof(buff) - 1);
     session_debugger_log(exdata->dbg->fd, "sess free: %s", buff);
-    session_debugger_log(exdata->dbg->fd, "session %lu %s stat:\n"
-                                          "C2S rx      packets: %6lu, C2S rx     bytes: %6lu\n"
-                                          "S2C rx      packets: %6lu, S2C rx     bytes: %6lu\n"
-                                          "C2S rx TCP segments: %6lu, C2S rx TCP bytes: %6lu\n"
-                                          "S2C rx TCP segments: %6lu, S2C rx TCP bytes: %6lu\n",
-                         session_get_id(exdata->sess), session_get0_readable_addr(exdata->sess),
-                         exdata->c2s_rx_pkts, exdata->c2s_rx_bytes,
-                         exdata->s2c_rx_pkts, exdata->s2c_rx_bytes,
-                         exdata->c2s_rx_tcp_seg, exdata->c2s_rx_tcp_bytes,
-                         exdata->s2c_rx_tcp_seg, exdata->s2c_rx_tcp_bytes);
+
+    snprintf(buff, sizeof(buff),
+             "==========================================================\n"
+             "C2S Data Packets    : %6lu | C2S Data Bytes     : %6lu\n"
+             "S2C Data Packets    : %6lu | S2C Data Bytes     : %6lu\n"
+             "----------------------------------------------------------\n"
+             "C2S Control Packets : %6lu | C2S Control Bytes : %6lu\n"
+             "S2C Control Packets : %6lu | S2C Control Bytes : %6lu\n"
+             "----------------------------------------------------------\n"
+             "C2S TCP Segments    : %6lu | C2S TCP Bytes     : %6lu\n"
+             "S2C TCP Segments    : %6lu | S2C TCP Bytes     : %6lu\n"
+             "----------------------------------------------------------\n"
+             "C2S UDP Payload     : %6lu | C2S UDP Bytes     : %6lu\n"
+             "S2C UDP Payload     : %6lu | S2C UDP Bytes     : %6lu\n",
+             exdata->c2s_rx_data_pkts, exdata->c2s_rx_data_bytes,
+             exdata->s2c_rx_data_pkts, exdata->s2c_rx_data_bytes,
+             exdata->c2s_rx_ctrl_pkts, exdata->c2s_rx_ctrl_bytes,
+             exdata->s2c_rx_ctrl_pkts, exdata->s2c_rx_ctrl_bytes,
+             exdata->c2s_rx_tcp_seg, exdata->c2s_rx_tcp_bytes,
+             exdata->s2c_rx_tcp_seg, exdata->s2c_rx_tcp_bytes,
+             exdata->c2s_rx_udp_payload, exdata->c2s_rx_udp_bytes,
+             exdata->s2c_rx_udp_payload, exdata->s2c_rx_udp_bytes);
+    session_debugger_log(exdata->dbg->fd, "session %lu %s statistics:\n%s", session_get_id(exdata->sess), session_get0_readable_addr(exdata->sess), buff);
 }
 
-static void on_sess_packet(struct session *sess, struct packet *pkt, void *arg)
+static void on_session_packet(struct session *sess, struct packet *pkt, void *arg)
 {
     struct session_debugger *dbg = (struct session_debugger *)arg;
+    int is_ctrl = packet_is_ctrl(pkt);
 
     char buff[PATH_MAX];
+    enum flow_type flow = session_get_flow_type(sess);
+    assert(flow == FLOW_TYPE_C2S || flow == FLOW_TYPE_S2C);
     struct session_debugger_exdata *exdata = (struct session_debugger_exdata *)session_get_exdata(sess, dbg->sess_exdata_idx);
     if (exdata == NULL)
     {
@@ -160,24 +216,40 @@ static void on_sess_packet(struct session *sess, struct packet *pkt, void *arg)
         session_set_exdata(sess, dbg->sess_exdata_idx, exdata);
     }
 
-    if (session_get_flow_type(sess) == FLOW_TYPE_C2S)
+    if (flow == FLOW_TYPE_C2S)
     {
-        exdata->c2s_rx_pkts++;
-        exdata->c2s_rx_bytes += packet_get_raw_len(pkt);
+        if (is_ctrl)
+        {
+            exdata->c2s_rx_ctrl_pkts++;
+            exdata->c2s_rx_ctrl_bytes += packet_get_raw_len(pkt);
+        }
+        else
+        {
+            exdata->c2s_rx_data_pkts++;
+            exdata->c2s_rx_data_bytes += packet_get_raw_len(pkt);
+        }
     }
     else
     {
-        exdata->s2c_rx_pkts++;
-        exdata->s2c_rx_bytes += packet_get_raw_len(pkt);
+        if (is_ctrl)
+        {
+            exdata->s2c_rx_ctrl_pkts++;
+            exdata->s2c_rx_ctrl_bytes += packet_get_raw_len(pkt);
+        }
+        else
+        {
+            exdata->s2c_rx_data_pkts++;
+            exdata->s2c_rx_data_bytes += packet_get_raw_len(pkt);
+        }
     }
 
     memset(buff, 0, sizeof(buff));
     session_to_str(sess, 1, buff, sizeof(buff) - 1);
-    session_debugger_log(dbg->fd, "on %s msg: %s", session_type_to_str(session_get_type(sess)), buff);
+    session_debugger_log(dbg->fd, "on %s %s packet: %s", session_type_to_str(session_get_type(sess)), (is_ctrl ? "ctrl" : "data"), buff);
 
     memset(buff, 0, sizeof(buff));
     packet_dump_str(pkt, buff, sizeof(buff) - 1);
-    session_debugger_log(dbg->fd, "rx %s packet\n%s", session_type_to_str(session_get_type(sess)), buff);
+    session_debugger_log(dbg->fd, "rx %s %s packet\n%s", session_type_to_str(session_get_type(sess)), (is_ctrl ? "ctrl" : "data"), buff);
 
     pthread_spin_lock(&dbg->lock);
     packet_dump_hex(pkt, dbg->fd);
@@ -189,14 +261,17 @@ static void on_tcp_stream(struct session *sess, const char *tcp_payload, uint32_
     struct session_debugger *dbg = (struct session_debugger *)arg;
 
     char buff[PATH_MAX];
+    enum flow_type flow = session_get_flow_type(sess);
+    assert(flow == FLOW_TYPE_C2S || flow == FLOW_TYPE_S2C);
     struct session_debugger_exdata *exdata = (struct session_debugger_exdata *)session_get_exdata(sess, dbg->sess_exdata_idx);
+    assert(exdata);
 
     memset(buff, 0, sizeof(buff));
     session_to_str(sess, 1, buff, sizeof(buff) - 1);
-    session_debugger_log(dbg->fd, "on TCP stream msg: %s", buff);
+    session_debugger_log(dbg->fd, "on TCP stream: %s", buff);
 
     pthread_spin_lock(&dbg->lock);
-    if (session_get_flow_type(sess) == FLOW_TYPE_C2S)
+    if (flow == FLOW_TYPE_C2S)
     {
         session_debugger_log(dbg->fd, "rx C2S TCP segment: len: %d, data: %p", tcp_payload_len, tcp_payload);
         hexdump_to_fd(dbg->fd, exdata->c2s_rx_tcp_bytes, tcp_payload, tcp_payload_len);
@@ -217,6 +292,47 @@ static void on_tcp_stream(struct session *sess, const char *tcp_payload, uint32_
     pthread_spin_unlock(&dbg->lock);
 }
 
+static void on_udp_payload(struct session *sess, struct packet *pkt, void *arg)
+{
+    struct session_debugger *dbg = (struct session_debugger *)arg;
+
+    char buff[PATH_MAX];
+    enum flow_type flow = session_get_flow_type(sess);
+    assert(flow == FLOW_TYPE_C2S || flow == FLOW_TYPE_S2C);
+    struct session_debugger_exdata *exdata = (struct session_debugger_exdata *)session_get_exdata(sess, dbg->sess_exdata_idx);
+    assert(exdata);
+
+    const char *udp_payload = packet_get_payload_data(pkt);
+    uint32_t udp_payload_len = packet_get_payload_len(pkt);
+    if (udp_payload_len == 0)
+    {
+        return;
+    }
+
+    memset(buff, 0, sizeof(buff));
+    session_to_str(sess, 1, buff, sizeof(buff) - 1);
+    session_debugger_log(dbg->fd, "on UDP payload: %s", buff);
+
+    pthread_spin_lock(&dbg->lock);
+    if (flow == FLOW_TYPE_C2S)
+    {
+        session_debugger_log(dbg->fd, "rx C2S UDP payload: len: %d, data: %p", udp_payload_len, udp_payload);
+        hexdump_to_fd(dbg->fd, exdata->c2s_rx_udp_bytes, udp_payload, udp_payload_len);
+        hexdump_to_fd(exdata->c2s_udp_payload_hexdump_fd, exdata->c2s_rx_udp_bytes, udp_payload, udp_payload_len);
+        exdata->c2s_rx_udp_payload++;
+        exdata->c2s_rx_udp_bytes += udp_payload_len;
+    }
+    else
+    {
+        session_debugger_log(dbg->fd, "rx S2C UDP payload: len: %d, data: %p", udp_payload_len, udp_payload);
+        hexdump_to_fd(dbg->fd, exdata->s2c_rx_udp_bytes, udp_payload, udp_payload_len);
+        hexdump_to_fd(exdata->s2c_udp_payload_hexdump_fd, exdata->s2c_rx_udp_bytes, udp_payload, udp_payload_len);
+        exdata->s2c_rx_udp_payload++;
+        exdata->s2c_rx_udp_bytes += udp_payload_len;
+    }
+    pthread_spin_unlock(&dbg->lock);
+}
+
 static void session_debugger_free(struct session_debugger *dbg)
 {
     if (dbg)
@@ -257,22 +373,22 @@ static struct session_debugger *session_debugger_new(struct session_manager *ses
         goto error_out;
     }
 
-    if (session_manager_subscribe_free(sess_mgr, on_sess_free, dbg) == -1)
+    if (session_manager_subscribe_free(sess_mgr, on_session_free, dbg) == -1)
     {
         session_debugger_log(STDERR_FILENO, "subscribe free failed\n");
         goto error_out;
     }
-    if (session_manager_subscribe_tcp(sess_mgr, on_sess_packet, dbg) == -1)
+    if (session_manager_subscribe_tcp(sess_mgr, on_session_packet, dbg) == -1)
     {
         session_debugger_log(STDERR_FILENO, "subscribe tcp failed\n");
         goto error_out;
     }
-    if (session_manager_subscribe_udp(sess_mgr, on_sess_packet, dbg) == -1)
+    if (session_manager_subscribe_udp(sess_mgr, on_session_packet, dbg) == -1)
     {
         session_debugger_log(STDERR_FILENO, "subscribe udp failed\n");
         goto error_out;
     }
-    if (session_manager_subscribe_control_packet(sess_mgr, on_sess_packet, dbg) == -1)
+    if (session_manager_subscribe_control_packet(sess_mgr, on_session_packet, dbg) == -1)
     {
         session_debugger_log(STDERR_FILENO, "subscribe control packet failed\n");
         goto error_out;
@@ -282,6 +398,11 @@ static struct session_debugger *session_debugger_new(struct session_manager *ses
         session_debugger_log(STDERR_FILENO, "subscribe tcp stream failed\n");
         goto error_out;
     }
+    if (session_manager_subscribe_udp(sess_mgr, on_udp_payload, dbg) == -1)
+    {
+        session_debugger_log(STDERR_FILENO, "subscribe udp failed\n");
+        goto error_out;
+    }
 
     return dbg;