packet manager support output fieldstat

4 files changed, 238 insertions, 129 deletions

diff --git a/infra/packet_manager/CMakeLists.txt b/infra/packet_manager/CMakeLists.txt
index a665944..de5b0bb 100644
--- a/infra/packet_manager/CMakeLists.txt
+++ b/infra/packet_manager/CMakeLists.txt
@@ -12,6 +12,6 @@ target_include_directories(packet_manager PUBLIC ${CMAKE_SOURCE_DIR}/deps/uthash
 target_include_directories(packet_manager PUBLIC ${CMAKE_SOURCE_DIR}/deps/logger)
 target_include_directories(packet_manager PUBLIC ${CMAKE_SOURCE_DIR}/include)
 target_include_directories(packet_manager PUBLIC ${CMAKE_SOURCE_DIR}/infra)
-target_link_libraries(packet_manager tuple logger dablooms mq exdata module_manager)
+target_link_libraries(packet_manager tuple logger dablooms mq exdata module_manager fieldstat4)
 
 add_subdirectory(test)
\ No newline at end of file
diff --git a/infra/packet_manager/packet_manager.c b/infra/packet_manager/packet_manager.c
index be36103..f132536 100644
--- a/infra/packet_manager/packet_manager.c
+++ b/infra/packet_manager/packet_manager.c
@@ -3,6 +3,7 @@
 #include "utils_internal.h"
 #include "packet_internal.h"
 #include "packet_manager.h"
+#include "fieldstat/fieldstat_easy.h"
 
 #define PACKET_MANAGER_MODULE_NAME "packet_manager_module"
 
@@ -10,7 +11,7 @@
 #define PACKET_MANAGER_LOG_FATAL(format, ...) STELLAR_LOG_FATAL(__thread_local_logger, "packet manager", format, ##__VA_ARGS__)
 #define PACKET_MANAGER_LOG_INFO(format, ...) STELLAR_LOG_INFO(__thread_local_logger, "packet manager", format, ##__VA_ARGS__)
 
-struct packet_manager_runtime
+struct packet_manager_rte
 {
     enum packet_stage curr_stage;
     struct packet_queue queue[PACKET_QUEUE_MAX];
@@ -18,26 +19,29 @@ struct packet_manager_runtime
     void *claim_arg;
     on_packet_claimed_callback *claim_cb;
 
-    struct mq_runtime *mq;
+    struct mq_runtime *mq_rte;
     struct packet_manager_stat stat;
 };
 
-struct packet_manager_schema
+struct packet_manager_sche
 {
-    struct exdata_schema *exdata;
-    struct mq_schema *mq;
-    int topic_id[PACKET_STAGE_MAX];
+    struct exdata_schema *ex_sche;
+    struct mq_schema *mq_sche;
+    int pkt_msg_id[PACKET_STAGE_MAX];
 };
 
 struct packet_manager
 {
     uint16_t thread_num;
-    struct packet_manager_schema *schema;
-    struct packet_manager_runtime *runtime[MAX_THREAD_NUM];
+    struct packet_manager_sche *sche;
+    struct packet_manager_rte *rte[MAX_THREAD_NUM];
+
+    struct fieldstat_easy *fs;
+    int pkt_mgr_fs_idx[PKT_MGR_STAT_MAX];
 };
 
 /******************************************************************************
- * packet manager runtime
+ * utils
  ******************************************************************************/
 
 const char *packet_stage_to_str(enum packet_stage stage)
@@ -59,58 +63,76 @@ const char *packet_stage_to_str(enum packet_stage stage)
     }
 }
 
-void packet_manager_runtime_free(struct packet_manager_runtime *pkt_mgr_rt)
+uint64_t packet_manager_stat_get(struct packet_manager_stat *stat, enum pkt_mgr_stat_type type)
+{
+    switch (type)
+    {
+#define XX(_type, _name, _val) case _type: return stat->_val;
+    PKT_MGR_STAT_MAP(XX)
+#undef XX
+    default:
+        return 0;
+    }
+}
+
+/******************************************************************************
+ * packet manager rte
+ ******************************************************************************/
+
+static void packet_manager_rte_free(struct packet_manager_rte *pkt_mgr_rte)
 {
-    if (pkt_mgr_rt)
+    struct packet *pkt = NULL;
+
+    if (pkt_mgr_rte)
     {
         for (int i = 0; i < PACKET_QUEUE_MAX; i++)
         {
-            struct packet *pkt = NULL;
-            while ((pkt = TAILQ_FIRST(&pkt_mgr_rt->queue[i])))
+            while ((pkt = TAILQ_FIRST(&pkt_mgr_rte->queue[i])))
             {
-                TAILQ_REMOVE(&pkt_mgr_rt->queue[i], pkt, stage_tqe);
+                TAILQ_REMOVE(&pkt_mgr_rte->queue[i], pkt, stage_tqe);
                 packet_free(pkt);
             }
         }
+
+        free(pkt_mgr_rte);
+        pkt_mgr_rte = NULL;
     }
-    free(pkt_mgr_rt);
-    pkt_mgr_rt = NULL;
 }
 
-struct packet_manager_runtime *packet_manager_runtime_new(struct mq_runtime *mq_rt)
+static struct packet_manager_rte *packet_manager_rte_new(struct mq_runtime *mq_rte)
 {
-    struct packet_manager_runtime *pkt_mgr_rt = calloc(1, sizeof(struct packet_manager_runtime));
-    if (pkt_mgr_rt == NULL)
+    struct packet_manager_rte *pkt_mgr_rte = calloc(1, sizeof(struct packet_manager_rte));
+    if (pkt_mgr_rte == NULL)
     {
-        PACKET_MANAGER_LOG_ERROR("failed to allocate memory for packet_manager_runtime");
+        PACKET_MANAGER_LOG_ERROR("failed to allocate memory for packet_manager_rte");
         return NULL;
     }
 
     for (int i = 0; i < PACKET_QUEUE_MAX; i++)
     {
-        TAILQ_INIT(&pkt_mgr_rt->queue[i]);
+        TAILQ_INIT(&pkt_mgr_rte->queue[i]);
     }
-    pkt_mgr_rt->mq = mq_rt;
+    pkt_mgr_rte->mq_rte = mq_rte;
 
-    return pkt_mgr_rt;
+    return pkt_mgr_rte;
 }
 
 /******************************************************************************
- * packet manager schema
+ * packet manager sche
  ******************************************************************************/
 
-static void on_packet_stage_dispatch(int topic_id, void *msg, on_msg_cb_func *cb, void *cb_arg, void *dispatch_arg)
+static void on_packet_stage_dispatch(int pkt_msg_id, void *msg, on_msg_cb_func *cb, void *cb_arg, void *dispatch_arg)
 {
     assert(msg);
     assert(dispatch_arg);
 
-    struct packet_manager_schema *schema = (struct packet_manager_schema *)dispatch_arg;
+    enum packet_stage stage = PACKET_STAGE_MAX;
     struct packet *pkt = (struct packet *)msg;
+    struct packet_manager_sche *pkt_mgr_sche = (struct packet_manager_sche *)dispatch_arg;
 
-    enum packet_stage stage = PACKET_STAGE_MAX;
     for (int i = 0; i < PACKET_STAGE_MAX; i++)
     {
-        if (schema->topic_id[i] == topic_id)
+        if (pkt_mgr_sche->pkt_msg_id[i] == pkt_msg_id)
         {
             stage = i;
             break;
@@ -120,62 +142,62 @@ static void on_packet_stage_dispatch(int topic_id, void *msg, on_msg_cb_func *cb
     ((on_packet_stage_callback *)(void *)cb)(pkt, stage, cb_arg);
 }
 
-static void packet_schema_free(struct packet_manager_schema *pkt_mgr_schema)
+static void packet_manager_sche_free(struct packet_manager_sche *pkt_mgr_sche)
 {
-    if (pkt_mgr_schema)
+    if (pkt_mgr_sche)
     {
-        if (pkt_mgr_schema->mq)
+        if (pkt_mgr_sche->mq_sche)
         {
             for (int i = 0; i < PACKET_STAGE_MAX; i++)
             {
-                if (pkt_mgr_schema->topic_id[i] >= 0)
+                if (pkt_mgr_sche->pkt_msg_id[i] >= 0)
                 {
-                    mq_schema_destroy_topic(pkt_mgr_schema->mq, pkt_mgr_schema->topic_id[i]);
+                    mq_schema_destroy_topic(pkt_mgr_sche->mq_sche, pkt_mgr_sche->pkt_msg_id[i]);
                 }
             }
         }
 
-        if (pkt_mgr_schema->exdata)
+        if (pkt_mgr_sche->ex_sche)
         {
-            exdata_schema_free(pkt_mgr_schema->exdata);
+            exdata_schema_free(pkt_mgr_sche->ex_sche);
         }
 
-        free(pkt_mgr_schema);
-        pkt_mgr_schema = NULL;
+        free(pkt_mgr_sche);
+        pkt_mgr_sche = NULL;
     }
 }
 
-static struct packet_manager_schema *packet_schema_new(struct mq_schema *mq)
+static struct packet_manager_sche *packet_manager_sche_new(struct mq_schema *mq_sche)
 {
-    struct packet_manager_schema *pkt_mgr_schema = calloc(1, sizeof(struct packet_manager_schema));
-    if (pkt_mgr_schema == NULL)
+    struct packet_manager_sche *pkt_mgr_sche = calloc(1, sizeof(struct packet_manager_sche));
+    if (pkt_mgr_sche == NULL)
     {
         PACKET_MANAGER_LOG_ERROR("failed to allocate memory for packet_schema");
         return NULL;
     }
 
-    pkt_mgr_schema->exdata = exdata_schema_new();
-    if (pkt_mgr_schema->exdata == NULL)
+    pkt_mgr_sche->mq_sche = mq_sche;
+    pkt_mgr_sche->ex_sche = exdata_schema_new();
+    if (pkt_mgr_sche->ex_sche == NULL)
     {
         PACKET_MANAGER_LOG_ERROR("failed to create exdata_schema");
         goto error_out;
     }
 
-    pkt_mgr_schema->mq = mq;
     for (int i = 0; i < PACKET_STAGE_MAX; i++)
     {
-        pkt_mgr_schema->topic_id[i] = mq_schema_create_topic(pkt_mgr_schema->mq, packet_stage_to_str(i), &on_packet_stage_dispatch, pkt_mgr_schema, NULL, NULL);
-        if (pkt_mgr_schema->topic_id[i] < 0)
+        pkt_mgr_sche->pkt_msg_id[i] = mq_schema_create_topic(pkt_mgr_sche->mq_sche, packet_stage_to_str(i), &on_packet_stage_dispatch, pkt_mgr_sche, NULL, NULL);
+        if (pkt_mgr_sche->pkt_msg_id[i] < 0)
         {
             PACKET_MANAGER_LOG_ERROR("failed to create topic %s", packet_stage_to_str(i));
             goto error_out;
         }
     }
 
-    return pkt_mgr_schema;
+    return pkt_mgr_sche;
 
 error_out:
-    packet_schema_free(pkt_mgr_schema);
+    packet_manager_sche_free(pkt_mgr_sche);
     return NULL;
 }
 
@@ -183,7 +205,7 @@ error_out:
  *  packet manager
  ******************************************************************************/
 
-struct packet_manager *packet_manager_new(struct mq_schema *mq_schema, uint16_t thread_num)
+struct packet_manager *packet_manager_new(struct mq_schema *mq_sche, uint16_t thread_num)
 {
     struct packet_manager *pkt_mgr = calloc(1, sizeof(struct packet_manager));
     if (pkt_mgr == NULL)
@@ -193,13 +215,29 @@ struct packet_manager *packet_manager_new(struct mq_schema *mq_schema, uint16_t
     }
 
     pkt_mgr->thread_num = thread_num;
-    pkt_mgr->schema = packet_schema_new(mq_schema);
-    if (pkt_mgr->schema == NULL)
+    pkt_mgr->sche = packet_manager_sche_new(mq_sche);
+    if (pkt_mgr->sche == NULL)
     {
         PACKET_MANAGER_LOG_ERROR("failed to create packet_schema");
         goto error_out;
     }
 
+    pkt_mgr->fs = fieldstat_easy_new(pkt_mgr->thread_num, "packet_manager", NULL, 0);
+    if (pkt_mgr->fs == NULL)
+    {
+        PACKET_MANAGER_LOG_ERROR("failed to create fieldstat_easy");
+        goto error_out;
+    }
+    if (fieldstat_easy_enable_auto_output(pkt_mgr->fs, "packet_manager.fs4", 2) != 0)
+    {
+        PACKET_MANAGER_LOG_ERROR("failed to enable auto output for fieldstat_easy");
+        goto error_out;
+    }
+    for (int i = 0; i < PKT_MGR_STAT_MAX; i++)
+    {
+        pkt_mgr->pkt_mgr_fs_idx[i] = fieldstat_easy_register_counter(pkt_mgr->fs, pkt_mgr_stat_str[i]);
+    }
+
     return pkt_mgr;
 
 error_out:
@@ -211,7 +249,12 @@ void packet_manager_free(struct packet_manager *pkt_mgr)
 {
     if (pkt_mgr)
     {
-        packet_schema_free(pkt_mgr->schema);
+        if (pkt_mgr->fs)
+        {
+            fieldstat_easy_free(pkt_mgr->fs);
+        }
+
+        packet_manager_sche_free(pkt_mgr->sche);
 
         free(pkt_mgr);
         pkt_mgr = NULL;
@@ -221,30 +264,29 @@ void packet_manager_free(struct packet_manager *pkt_mgr)
 int packet_manager_new_packet_exdata_index(struct packet_manager *pkt_mgr, const char *name, exdata_free *func, void *arg)
 {
     assert(pkt_mgr);
-    return exdata_schema_new_index(pkt_mgr->schema->exdata, name, func, arg);
+    return exdata_schema_new_index(pkt_mgr->sche->ex_sche, name, func, arg);
 }
 
 int packet_manager_subscribe(struct packet_manager *pkt_mgr, enum packet_stage stage, on_packet_stage_callback *cb, void *arg)
 {
     assert(pkt_mgr);
-    return mq_schema_subscribe(pkt_mgr->schema->mq, pkt_mgr->schema->topic_id[stage], (on_msg_cb_func *)(void *)cb, arg);
+    return mq_schema_subscribe(pkt_mgr->sche->mq_sche, pkt_mgr->sche->pkt_msg_id[stage], (on_msg_cb_func *)(void *)cb, arg);
 }
 
-int packet_manager_init(struct packet_manager *pkt_mgr, uint16_t thread_id, struct mq_runtime *mq_rt)
+int packet_manager_init(struct packet_manager *pkt_mgr, uint16_t thread_id, struct mq_runtime *mq_rte)
 {
     assert(pkt_mgr);
     assert(thread_id < pkt_mgr->thread_num);
-    assert(mq_rt);
+    assert(mq_rte);
 
-    struct packet_manager_runtime *pkt_mgr_rt = packet_manager_runtime_new(mq_rt);
-    if (pkt_mgr_rt == NULL)
+    pkt_mgr->rte[thread_id] = packet_manager_rte_new(mq_rte);
+    if (pkt_mgr->rte[thread_id] == NULL)
     {
-        PACKET_MANAGER_LOG_ERROR("failed to create packet_manager_runtime");
+        PACKET_MANAGER_LOG_ERROR("failed to create packet_manager_rte");
         return -1;
     }
     else
     {
-        pkt_mgr->runtime[thread_id] = pkt_mgr_rt;
         return 0;
     }
 }
@@ -254,38 +296,35 @@ void packet_manager_clean(struct packet_manager *pkt_mgr, uint16_t thread_id)
     assert(pkt_mgr);
     assert(thread_id < pkt_mgr->thread_num);
 
-    struct packet_manager_runtime *pkt_mgr_rt = pkt_mgr->runtime[thread_id];
-    PACKET_MANAGER_LOG_INFO("runtime: %p, idx: %d, will be cleaned", pkt_mgr_rt, thread_id);
-    packet_manager_print_stat(pkt_mgr, thread_id);
-    packet_manager_runtime_free(pkt_mgr_rt);
-    pkt_mgr_rt = NULL;
+    struct packet_manager_rte *pkt_mgr_rte = pkt_mgr->rte[thread_id];
+    packet_manager_rte_free(pkt_mgr_rte);
+    pkt_mgr_rte = NULL;
 }
 
 void packet_manager_ingress(struct packet_manager *pkt_mgr, uint16_t thread_id, struct packet *pkt)
 {
-    struct packet_manager_runtime *runtime = pkt_mgr->runtime[thread_id];
+    struct packet_manager_rte *pkt_mgr_rte = pkt_mgr->rte[thread_id];
+    struct exdata_runtime *ex_rte = exdata_runtime_new(pkt_mgr->sche->ex_sche);
+    packet_set_user_data(pkt, ex_rte);
 
-    struct exdata_runtime *exdata_rt = exdata_runtime_new(pkt_mgr->schema->exdata);
-    packet_set_user_data(pkt, exdata_rt);
-
-    runtime->stat.total.pkts_ingress++;
-    runtime->stat.queue[PACKET_STAGE_PREROUTING].pkts_in++;
-    TAILQ_INSERT_TAIL(&runtime->queue[PACKET_STAGE_PREROUTING], pkt, stage_tqe);
+    pkt_mgr_rte->stat.pkts_ingress++;
+    pkt_mgr_rte->stat.queue[PACKET_STAGE_PREROUTING].pkts_in++;
+    TAILQ_INSERT_TAIL(&pkt_mgr_rte->queue[PACKET_STAGE_PREROUTING], pkt, stage_tqe);
 }
 
 struct packet *packet_manager_egress(struct packet_manager *pkt_mgr, uint16_t thread_id)
 {
-    struct packet_manager_runtime *runtime = pkt_mgr->runtime[thread_id];
+    struct packet_manager_rte *pkt_mgr_rte = pkt_mgr->rte[thread_id];
+    struct packet *pkt = TAILQ_FIRST(&pkt_mgr_rte->queue[PACKET_STAGE_MAX]);
 
-    struct packet *pkt = TAILQ_FIRST(&runtime->queue[PACKET_STAGE_MAX]);
     if (pkt)
     {
-        runtime->stat.total.pkts_egress++;
-        runtime->stat.queue[PACKET_STAGE_MAX].pkts_out++;
-        TAILQ_REMOVE(&runtime->queue[PACKET_STAGE_MAX], pkt, stage_tqe);
+        pkt_mgr_rte->stat.pkts_egress++;
+        pkt_mgr_rte->stat.queue[PACKET_STAGE_MAX].pkts_out++;
+        TAILQ_REMOVE(&pkt_mgr_rte->queue[PACKET_STAGE_MAX], pkt, stage_tqe);
 
-        struct exdata_runtime *exdata_rt = packet_get_user_data(pkt);
-        exdata_runtime_free(exdata_rt);
+        struct exdata_runtime *ex_rte = packet_get_user_data(pkt);
+        exdata_runtime_free(ex_rte);
         return pkt;
     }
     else
@@ -296,45 +335,45 @@ struct packet *packet_manager_egress(struct packet_manager *pkt_mgr, uint16_t th
 
 void packet_manager_dispatch(struct packet_manager *pkt_mgr, uint16_t thread_id)
 {
-    struct packet_manager_runtime *runtime = pkt_mgr->runtime[thread_id];
+    struct packet_manager_rte *pkt_mgr_rte = pkt_mgr->rte[thread_id];
 
     for (int i = 0; i < PACKET_STAGE_MAX; i++)
     {
-        runtime->curr_stage = i;
+        pkt_mgr_rte->curr_stage = i;
 
         struct packet *pkt = NULL;
-        while ((pkt = TAILQ_FIRST(&runtime->queue[runtime->curr_stage])))
+        while ((pkt = TAILQ_FIRST(&pkt_mgr_rte->queue[pkt_mgr_rte->curr_stage])))
         {
             packet_set_claim(pkt, false);
-            runtime->claim_cb = NULL;
-            runtime->claim_arg = NULL;
+            pkt_mgr_rte->claim_cb = NULL;
+            pkt_mgr_rte->claim_arg = NULL;
 
-            TAILQ_REMOVE(&runtime->queue[runtime->curr_stage], pkt, stage_tqe);
-            runtime->stat.queue[runtime->curr_stage].pkts_out++;
+            TAILQ_REMOVE(&pkt_mgr_rte->queue[pkt_mgr_rte->curr_stage], pkt, stage_tqe);
+            pkt_mgr_rte->stat.queue[pkt_mgr_rte->curr_stage].pkts_out++;
 
-            mq_runtime_publish_message(runtime->mq, runtime->curr_stage, pkt);
-            mq_runtime_dispatch(runtime->mq);
+            mq_runtime_publish_message(pkt_mgr_rte->mq_rte, pkt_mgr_rte->curr_stage, pkt);
+            mq_runtime_dispatch(pkt_mgr_rte->mq_rte);
 
             if (packet_is_claim(pkt))
             {
-                if (runtime->claim_cb)
+                if (pkt_mgr_rte->claim_cb)
                 {
-                    runtime->claim_cb(pkt, runtime->claim_arg);
+                    pkt_mgr_rte->claim_cb(pkt, pkt_mgr_rte->claim_arg);
                 }
                 continue;
             }
 
-            TAILQ_INSERT_TAIL(&runtime->queue[runtime->curr_stage + 1], pkt, stage_tqe);
-            runtime->stat.queue[runtime->curr_stage + 1].pkts_in++;
+            TAILQ_INSERT_TAIL(&pkt_mgr_rte->queue[pkt_mgr_rte->curr_stage + 1], pkt, stage_tqe);
+            pkt_mgr_rte->stat.queue[pkt_mgr_rte->curr_stage + 1].pkts_in++;
         }
     }
-    runtime->curr_stage = -1;
+    pkt_mgr_rte->curr_stage = -1;
 }
 
 int packet_manager_claim_packet(struct packet_manager *pkt_mgr, uint16_t thread_id, struct packet *pkt, on_packet_claimed_callback cb, void *arg)
 {
     assert(pkt_mgr);
-    struct packet_manager_runtime *runtime = pkt_mgr->runtime[thread_id];
+    struct packet_manager_rte *pkt_mgr_rte = pkt_mgr->rte[thread_id];
 
     if (packet_is_claim(pkt))
     {
@@ -343,10 +382,10 @@ int packet_manager_claim_packet(struct packet_manager *pkt_mgr, uint16_t thread_
     }
     else
     {
-        runtime->claim_cb = cb;
-        runtime->claim_arg = arg;
+        pkt_mgr_rte->claim_cb = cb;
+        pkt_mgr_rte->claim_arg = arg;
         packet_set_claim(pkt, true);
-        runtime->stat.queue[runtime->curr_stage].pkts_claim++;
+        pkt_mgr_rte->stat.queue[pkt_mgr_rte->curr_stage].pkts_claim++;
         return 0;
     }
 }
@@ -354,7 +393,7 @@ int packet_manager_claim_packet(struct packet_manager *pkt_mgr, uint16_t thread_
 void packet_manager_schedule_packet(struct packet_manager *pkt_mgr, uint16_t thread_id, struct packet *pkt, enum packet_stage stage)
 {
     assert(pkt_mgr);
-    struct packet_manager_runtime *runtime = pkt_mgr->runtime[thread_id];
+    struct packet_manager_rte *pkt_mgr_rte = pkt_mgr->rte[thread_id];
 
     if (stage >= PACKET_STAGE_MAX)
     {
@@ -363,33 +402,33 @@ void packet_manager_schedule_packet(struct packet_manager *pkt_mgr, uint16_t thr
         return;
     }
 
-    runtime->stat.queue[stage].pkts_schedule++;
-    runtime->stat.queue[stage].pkts_in++;
-    TAILQ_INSERT_TAIL(&runtime->queue[stage], pkt, stage_tqe);
+    pkt_mgr_rte->stat.queue[stage].pkts_schedule++;
+    pkt_mgr_rte->stat.queue[stage].pkts_in++;
+    TAILQ_INSERT_TAIL(&pkt_mgr_rte->queue[stage], pkt, stage_tqe);
 }
 
 struct packet_manager_stat *packet_manager_get_stat(struct packet_manager *pkt_mgr, uint16_t thread_id)
 {
-    struct packet_manager_runtime *runtime = pkt_mgr->runtime[thread_id];
-    return &runtime->stat;
+    struct packet_manager_rte *pkt_mgr_rte = pkt_mgr->rte[thread_id];
+    return &pkt_mgr_rte->stat;
 }
 
 void packet_manager_print_stat(struct packet_manager *pkt_mgr, uint16_t thread_id)
 {
-    struct packet_manager_runtime *runtime = pkt_mgr->runtime[thread_id];
+    struct packet_manager_rte *pkt_mgr_rte = pkt_mgr->rte[thread_id];
 
     PACKET_MANAGER_LOG_INFO("runtime: %p, pkts_ingress: %lu, pkts_egress: %lu",
-                            runtime, runtime->stat.total.pkts_ingress,
-                            runtime->stat.total.pkts_egress);
+                            pkt_mgr_rte, pkt_mgr_rte->stat.pkts_ingress,
+                            pkt_mgr_rte->stat.pkts_egress);
     for (int i = 0; i < PACKET_QUEUE_MAX; i++)
     {
         PACKET_MANAGER_LOG_INFO("runtime: %p, %-24s stat => pkts_in: %lu, pkts_out: %lu, pkts_claim: %lu, pkts_schedule: %lu",
-                                runtime,
+                                pkt_mgr_rte,
                                 packet_stage_to_str(i),
-                                runtime->stat.queue[i].pkts_in,
-                                runtime->stat.queue[i].pkts_out,
-                                runtime->stat.queue[i].pkts_claim,
-                                runtime->stat.queue[i].pkts_schedule);
+                                pkt_mgr_rte->stat.queue[i].pkts_in,
+                                pkt_mgr_rte->stat.queue[i].pkts_out,
+                                pkt_mgr_rte->stat.queue[i].pkts_claim,
+                                pkt_mgr_rte->stat.queue[i].pkts_schedule);
     }
 }
 
@@ -397,6 +436,27 @@ void packet_manager_print_stat(struct packet_manager *pkt_mgr, uint16_t thread_i
  * packet manager module
  ******************************************************************************/
 
+static void on_polling(struct stellar_module_manager *mod_mgr, void *args)
+{
+    uint64_t now_ms = clock_get_real_time_ms();
+    int thread_id = stellar_module_manager_get_thread_id(mod_mgr);
+    struct packet_manager *pkt_mgr = (struct packet_manager *)args;
+
+    static __thread uint64_t last_sync_stat_ms = 0;
+    static __thread struct packet_manager_stat pkt_mgr_last_stat = {0};
+    if (now_ms - last_sync_stat_ms >= SYNC_STAT_INTERVAL_MS)
+    {
+        struct packet_manager_stat *pkt_mgr_curr_stat = &pkt_mgr->rte[thread_id]->stat;
+        for (int i = 0; i < PKT_MGR_STAT_MAX; i++)
+        {
+            uint64_t val = packet_manager_stat_get(pkt_mgr_curr_stat, i) - packet_manager_stat_get(&pkt_mgr_last_stat, i);
+            fieldstat_easy_counter_incrby(pkt_mgr->fs, thread_id, pkt_mgr->pkt_mgr_fs_idx[i], NULL, 0, val);
+        }
+        pkt_mgr_last_stat = *pkt_mgr_curr_stat;
+        last_sync_stat_ms = now_ms;
+    }
+}
+
 struct packet_manager *stellar_module_get_packet_manager(struct stellar_module_manager *mod_mgr)
 {
     assert(mod_mgr);
@@ -411,15 +471,16 @@ struct packet_manager *stellar_module_get_packet_manager(struct stellar_module_m
 struct stellar_module *packet_manager_on_init(struct stellar_module_manager *mod_mgr)
 {
     assert(mod_mgr);
-    struct mq_schema *mq_schema = stellar_module_manager_get_mq_schema(mod_mgr);
-    assert(mq_schema);
+    struct mq_schema *mq_sche = stellar_module_manager_get_mq_schema(mod_mgr);
+    assert(mq_sche);
     uint16_t thread_num = stellar_module_manager_get_max_thread_num(mod_mgr);
 
-    struct packet_manager *pkt_mgr = packet_manager_new(mq_schema, thread_num);
+    struct packet_manager *pkt_mgr = packet_manager_new(mq_sche, thread_num);
     if (pkt_mgr == NULL)
     {
         return NULL;
     }
+    stellar_module_manager_polling_subscribe(mod_mgr, on_polling, pkt_mgr);
 
     struct stellar_module *pkt_mgr_mod = stellar_module_new(PACKET_MANAGER_MODULE_NAME, NULL);
     if (pkt_mgr_mod == NULL)
@@ -450,11 +511,11 @@ struct stellar_module *packet_manager_on_thread_init(struct stellar_module_manag
 {
     struct packet_manager *pkt_mgr = stellar_module_get_ctx(mod);
     assert(pkt_mgr);
-    struct mq_runtime *mq_rt = stellar_module_manager_get_mq_runtime(mod_mgr);
-    assert(mq_rt);
+    struct mq_runtime *mq_rte = stellar_module_manager_get_mq_runtime(mod_mgr);
+    assert(mq_rte);
     assert(thread_id < pkt_mgr->thread_num);
 
-    if (packet_manager_init(pkt_mgr, thread_id, mq_rt) != 0)
+    if (packet_manager_init(pkt_mgr, thread_id, mq_rte) != 0)
     {
         PACKET_MANAGER_LOG_ERROR("failed to init packet_manager_init");
         return NULL;
diff --git a/infra/packet_manager/packet_manager.h b/infra/packet_manager/packet_manager.h
index 7821758..61d5f04 100644
--- a/infra/packet_manager/packet_manager.h
+++ b/infra/packet_manager/packet_manager.h
@@ -12,11 +12,8 @@ extern "C"
 
 struct packet_manager_stat
 {
-    struct
-    {
-        uint64_t pkts_ingress;
-        uint64_t pkts_egress;
-    } total;
+    uint64_t pkts_ingress;
+    uint64_t pkts_egress;
     struct
     {
         uint64_t pkts_in;  // include the packets that are scheduled
@@ -26,10 +23,55 @@ struct packet_manager_stat
     } queue[PACKET_QUEUE_MAX]; // the last queue is for sending packets
 };
 
+// XX(type, name, val)
+#define PKT_MGR_STAT_MAP(XX)                                                                                                \
+    XX(PKT_MGR_STAT_PKTS_INGRESS, pkts_ingress, pkts_ingress)                                                               \
+    XX(PKT_MGR_STAT_PKTS_EGRESS, pkts_egress, pkts_egress)                                                                  \
+    /* PREROUTING */                                                                                                        \
+    XX(PKT_MGR_STAT_PKTS_IN_ON_PREROUTING, pkts_in_on_prerouting, queue[PACKET_STAGE_PREROUTING].pkts_in)                   \
+    XX(PKT_MGR_STAT_PKTS_OUT_ON_PREROUTING, pkts_out_on_prerouting, queue[PACKET_STAGE_PREROUTING].pkts_out)                \
+    XX(PKT_MGR_STAT_PKTS_CLAIM_ON_PREROUTING, pkts_claim_on_prerouting, queue[PACKET_STAGE_PREROUTING].pkts_claim)          \
+    XX(PKT_MGR_STAT_PKTS_SCHEDULE_ON_PREROUTING, pkts_schedule_on_prerouting, queue[PACKET_STAGE_PREROUTING].pkts_schedule) \
+    /* INPUT */                                                                                                             \
+    XX(PKT_MGR_STAT_PKTS_IN_ON_INPUT, pkts_in_on_input, queue[PACKET_STAGE_INPUT].pkts_in)                                  \
+    XX(PKT_MGR_STAT_PKTS_OUT_ON_INPUT, pkts_out_on_input, queue[PACKET_STAGE_INPUT].pkts_out)                               \
+    XX(PKT_MGR_STAT_PKTS_CLAIM_ON_INPUT, pkts_claim_on_input, queue[PACKET_STAGE_INPUT].pkts_claim)                         \
+    XX(PKT_MGR_STAT_PKTS_SCHEDULE_ON_INPUT, pkts_schedule_on_input, queue[PACKET_STAGE_INPUT].pkts_schedule)                \
+    /* FORWARD */                                                                                                           \
+    XX(PKT_MGR_STAT_PKTS_IN_ON_FORWARD, pkts_in_on_forward, queue[PACKET_STAGE_FORWARD].pkts_in)                            \
+    XX(PKT_MGR_STAT_PKTS_OUT_ON_FORWARD, pkts_out_on_forward, queue[PACKET_STAGE_FORWARD].pkts_out)                         \
+    XX(PKT_MGR_STAT_PKTS_CLAIM_ON_FORWARD, pkts_claim_on_forward, queue[PACKET_STAGE_FORWARD].pkts_claim)                   \
+    XX(PKT_MGR_STAT_PKTS_SCHEDULE_ON_FORWARD, pkts_schedule_on_forward, queue[PACKET_STAGE_FORWARD].pkts_schedule)          \
+    /* OUTPUT */                                                                                                            \
+    XX(PKT_MGR_STAT_PKTS_IN_ON_OUTPUT, pkts_in_on_output, queue[PACKET_STAGE_OUTPUT].pkts_in)                               \
+    XX(PKT_MGR_STAT_PKTS_OUT_ON_OUTPUT, pkts_out_on_output, queue[PACKET_STAGE_OUTPUT].pkts_out)                            \
+    XX(PKT_MGR_STAT_PKTS_CLAIM_ON_OUTPUT, pkts_claim_on_output, queue[PACKET_STAGE_OUTPUT].pkts_claim)                      \
+    XX(PKT_MGR_STAT_PKTS_SCHEDULE_ON_OUTPUT, pkts_schedule_on_output, queue[PACKET_STAGE_OUTPUT].pkts_schedule)             \
+    /* POSTROUTING */                                                                                                       \
+    XX(PKT_MGR_STAT_PKTS_IN_ON_POSTROUTING, pkts_in_on_postrouting, queue[PACKET_STAGE_POSTROUTING].pkts_in)                \
+    XX(PKT_MGR_STAT_PKTS_OUT_ON_POSTROUTING, pkts_out_on_postrouting, queue[PACKET_STAGE_POSTROUTING].pkts_out)             \
+    XX(PKT_MGR_STAT_PKTS_CLAIM_ON_POSTROUTING, pkts_claim_on_postrouting, queue[PACKET_STAGE_POSTROUTING].pkts_claim)       \
+    XX(PKT_MGR_STAT_PKTS_SCHEDULE_ON_POSTROUTING, pkts_schedule_on_postrouting, queue[PACKET_STAGE_POSTROUTING].pkts_schedule)
+
+enum pkt_mgr_stat_type
+{
+#define XX(type, name, val) type,
+    PKT_MGR_STAT_MAP(XX)
+#undef XX
+    PKT_MGR_STAT_MAX
+};
+
+__attribute__((unused)) static const char pkt_mgr_stat_str[PKT_MGR_STAT_MAX][64] =
+{
+#define XX(type, name, val) #name,
+    PKT_MGR_STAT_MAP(XX)
+#undef XX
+};
+
 struct packet_manager *packet_manager_new(struct mq_schema *mq_schema, uint16_t thread_num);
 void packet_manager_free(struct packet_manager *pkt_mgr);
 
-int packet_manager_init(struct packet_manager *pkt_mgr, uint16_t thread_id, struct mq_runtime *mq_rt);
+int packet_manager_init(struct packet_manager *pkt_mgr, uint16_t thread_id, struct mq_runtime *mq_rte);
 void packet_manager_clean(struct packet_manager *pkt_mgr, uint16_t thread_id);
 void packet_manager_ingress(struct packet_manager *pkt_mgr, uint16_t thread_id, struct packet *pkt);
 struct packet *packet_manager_egress(struct packet_manager *pkt_mgr, uint16_t thread_id);
@@ -39,6 +81,8 @@ void packet_manager_print_stat(struct packet_manager *pkt_mgr, uint16_t thread_i
 
 const char *packet_stage_to_str(enum packet_stage stage);
 
+uint64_t packet_manager_stat_get(struct packet_manager_stat *stat, enum pkt_mgr_stat_type type);
+
 #ifdef __cplusplus
 }
 #endif
diff --git a/infra/packet_manager/test/gtest_packet_manager.cpp b/infra/packet_manager/test/gtest_packet_manager.cpp
index 902b7c7..30abcdb 100644
--- a/infra/packet_manager/test/gtest_packet_manager.cpp
+++ b/infra/packet_manager/test/gtest_packet_manager.cpp
@@ -71,8 +71,8 @@ static struct packet_manager_stat init_stat = {};
 
 static void check_stat(struct packet_manager_stat *curr_stat, struct packet_manager_stat *expect_stat)
 {
-	EXPECT_TRUE(curr_stat->total.pkts_ingress == expect_stat->total.pkts_ingress);
-	EXPECT_TRUE(curr_stat->total.pkts_egress == expect_stat->total.pkts_egress);
+	EXPECT_TRUE(curr_stat->pkts_ingress == expect_stat->pkts_ingress);
+	EXPECT_TRUE(curr_stat->pkts_egress == expect_stat->pkts_egress);
 
 	for (int i = 0; i < PACKET_QUEUE_MAX; i++)
 	{
@@ -143,7 +143,8 @@ TEST(PACKET_MANAGER, SUBSCRIBER_PACKET_STAGE)
 	EXPECT_TRUE(packet_manager_egress(pkt_mgr, thread_id) == &pkt);
 	EXPECT_TRUE(packet_manager_egress(pkt_mgr, thread_id) == NULL);
 	struct packet_manager_stat expect_stat = {
-		.total = {.pkts_ingress = 1, .pkts_egress = 1},
+		.pkts_ingress = 1,
+		.pkts_egress = 1,
 		.queue = {
 			[PACKET_STAGE_PREROUTING] = {.pkts_in = 1, .pkts_out = 1, .pkts_claim = 0, .pkts_schedule = 0},
 			[PACKET_STAGE_INPUT] = {.pkts_in = 1, .pkts_out = 1, .pkts_claim = 0, .pkts_schedule = 0},
@@ -246,7 +247,8 @@ TEST(PACKET_MANAGER, CLAIM_PACKET)
 	packet_manager_dispatch(pkt_mgr, thread_id);
 	EXPECT_TRUE(packet_manager_egress(pkt_mgr, thread_id) == NULL);
 	struct packet_manager_stat expect_stat = {
-		.total = {.pkts_ingress = 1, .pkts_egress = 0},
+		.pkts_ingress = 1,
+		.pkts_egress = 0,
 		.queue = {
 			[PACKET_STAGE_PREROUTING] = {.pkts_in = 1, .pkts_out = 1, .pkts_claim = 1, .pkts_schedule = 0},
 			[PACKET_STAGE_INPUT] = {.pkts_in = 0, .pkts_out = 0, .pkts_claim = 0, .pkts_schedule = 0},
@@ -329,7 +331,8 @@ TEST(PACKET_MANAGER, SCHEDULE_PACKET)
 	EXPECT_TRUE(packet_manager_egress(pkt_mgr, thread_id) == &pkt);
 	EXPECT_TRUE(packet_manager_egress(pkt_mgr, thread_id) == NULL);
 	struct packet_manager_stat expect_stat = {
-		.total = {.pkts_ingress = 1, .pkts_egress = 5},
+		.pkts_ingress = 1,
+		.pkts_egress = 5,
 		.queue = {
 			[PACKET_STAGE_PREROUTING] = {.pkts_in = 1, .pkts_out = 1, .pkts_claim = 0, .pkts_schedule = 0},
 			[PACKET_STAGE_INPUT] = {.pkts_in = 2, .pkts_out = 2, .pkts_claim = 0, .pkts_schedule = 1},
@@ -423,7 +426,8 @@ TEST(PACKET_MANAGER, SCHEDULE_CLAIMED_PACKET)
 	packet_manager_dispatch(pkt_mgr, thread_id);
 	EXPECT_TRUE(packet_manager_egress(pkt_mgr, thread_id) == &pkt);
 	struct packet_manager_stat expect_stat = {
-		.total = {.pkts_ingress = 1, .pkts_egress = 1},
+		.pkts_ingress = 1,
+		.pkts_egress = 1,
 		.queue = {
 			[PACKET_STAGE_PREROUTING] = {.pkts_in = 1, .pkts_out = 1, .pkts_claim = 1, .pkts_schedule = 0},
 			[PACKET_STAGE_INPUT] = {.pkts_in = 0, .pkts_out = 0, .pkts_claim = 0, .pkts_schedule = 0},