path: root/service/src/node_tera.c

#include "sc_trace.h"
#include <cJSON.h>
#include <netinet/in.h>

#include <MESA_prof_load.h>
#include <link_db.h>
#include <port_adapter_mapping.h>
#include <sc_common.h>
#include <sc_node.h>

/* Tera ingress next node */
enum tera_ingress_next
{
    TERA_INGRESS_NEXT_CLASSIFIER = 0,
    TERA_INGRESS_NEXT_PKT_DROP,
    TERA_INGRESS_NEXT_MAX
};

/* Tera egress next node */
enum tera_egress_next
{
    TERA_EGRESS_NEXT_ETH_EGRESS = 0,
    TERA_EGRESS_NEXT_PKT_DROP,
    TERA_EGRESS_NEXT_MAX
};

/* Tera adapter */
struct tera_adapter
{
    uint8_t mac_flipping;
    uint16_t tera_adapter_id;
    uint16_t vlan_int;
    uint16_t vlan_ext;
    struct mr_dev_desc * listen_device;
};

/* Tera ingress drop reason */
enum tera_ingress_drop_reason
{
    TERA_INGR_DROP_RSN_NO_MATCH = 0,
    TERA_INGR_DROP_RSN_APPEND_SID_ERR,
    TERA_INGR_DROP_RSN_MAX
};

/* Tera ingress drop reason string */
static const char * tera_ingress_drop_reason_str[TERA_INGR_DROP_RSN_MAX] = {"drop_rsn_adapter_lookup_miss",
                                                                            "drop_rsn_append_sid_err"};

/* Tera ingress stats struct */
struct tera_ingress_stats
{
    volatile uint64_t deal_pkts;
    volatile uint64_t pkts_per_batch;
    volatile uint64_t to_classifier;
    volatile uint64_t drop_reason[TERA_INGR_DROP_RSN_MAX];
} __rte_cache_aligned;

/* Tera egress stats struct */
struct tera_egress_stats
{
    volatile uint64_t deal_pkts;
    volatile uint64_t pkts_per_batch;
} __rte_cache_aligned;

/* Tera main struct */
struct node_tera_main
{
    uint16_t nr_adapters;
    uint32_t nr_sids;
    uint32_t sid_start;
    uint32_t sid_end;
    struct link_db_ctx * link_db_ctx;
    struct tera_adapter * adapters;
} __rte_cache_min_aligned;

static struct tera_ingress_stats ingress_stats_per_graph[RTE_MAX_LCORE] = {};
static struct tera_egress_stats egress_stats_per_graph[RTE_MAX_LCORE] = {};

/* Global tera main */
static struct node_tera_main g_tera_main = {};
static uint16_t nr_max_tera_adapters = 256;

/* Tera main get */
static inline struct node_tera_main * node_tera_main_get()
{
    return &g_tera_main;
}

/* Tera adapters num get */
unsigned int nr_max_tera_adapters_get()
{
    return nr_max_tera_adapters;
}

/* Tera adapters id check */
int tera_adapter_id_check(uint32_t tera_adapter_id)
{
    if (tera_adapter_id >= nr_max_tera_adapters)
        return RT_ERR;

    struct tera_adapter * adapter = &node_tera_main_get()->adapters[tera_adapter_id];

    if (adapter->listen_device == NULL)
        return RT_ERR;

    if (adapter->tera_adapter_id != tera_adapter_id)
        return RT_ERR;

    return RT_SUCCESS;
}

/* Tera config parse */
int tera_config_parse(struct sc_main * sc, struct node_tera_main * tera_main)
{
    /* Get sid start */
    uint32_t sid_start;
    MESA_load_profile_uint_def(sc->local_cfgfile, "tera_adapters", "sid_start", &sid_start, 500);

    /* Get sid end */
    uint32_t sid_end;
    MESA_load_profile_uint_def(sc->local_cfgfile, "tera_adapters", "sid_end", &sid_end, 600);

    /* Check the sid range */
    if (sid_start > sid_end)
    {
        MR_ERROR("Tera adapters config 'sid_end' less than 'sid_start' .");
        return RT_ERR;
    }

    /* Parse all config */
    int ret;
    uint16_t nr_adapters = 0;
    char str_conf_section[MR_STRING_MAX] = {};
    char str_listen_device[MR_STRING_MAX] = {};
    for (int index = 0; index < nr_max_tera_adapters; index++)
    {
        snprintf(str_conf_section, sizeof(str_conf_section), "tera_adapter:%d", index);

        /* Get vlan conf */
        uint32_t vlan_int;
        ret = MESA_load_profile_uint_nodef(sc->local_cfgfile, str_conf_section, "vlan_int", &vlan_int);
        if (ret < 0)
            continue;

        uint32_t vlan_ext;
        ret = MESA_load_profile_uint_nodef(sc->local_cfgfile, str_conf_section, "vlan_ext", &vlan_ext);
        if (ret < 0)
        {
            MR_ERROR("The : %s ,No config the 'vlan_ext'.", str_conf_section);
            return RT_ERR;
        }

        /* Get tera adapter id */
        uint32_t tera_adapter_id;
        ret = MESA_load_profile_uint_nodef(sc->local_cfgfile, str_conf_section, "tera_adapter_id", &tera_adapter_id);
        if (ret < 0)
        {
            MR_ERROR("The : %s ,No config the 'tera_adapter_id'.", str_conf_section);
            return RT_ERR;
        }

        if (tera_adapter_id >= nr_max_tera_adapters)
        {
            MR_ERROR("The : %s 'tera_adapter_id' is invalid:  %u", str_conf_section, tera_adapter_id);
            return RT_ERR;
        }

        /* Get mac flipping */
        uint32_t mac_flipping;
        MESA_load_profile_uint_def(sc->local_cfgfile, str_conf_section, "mac_flipping", &mac_flipping, 0);

        /* Get listen device */
        ret = MESA_load_profile_string_nodef(sc->local_cfgfile, str_conf_section, "listen_device", str_listen_device,
                                             sizeof(str_listen_device));
        if (ret < 0)
        {
            MR_ERROR("The : %s ,No config the 'listen_device'.", str_conf_section);
            return RT_ERR;
        }

        struct mr_dev_desc * dev_desc = mr_dev_desc_lookup(sc->devmgr_main, str_listen_device);
        if (dev_desc == NULL)
        {
            MR_ERROR("The : %s 'listen_device' is invalid:  %s", str_conf_section, str_listen_device);
            return RT_ERR;
        }

        /* Port adapter mapping insert */
        port_adapter_mapping_insert(dev_desc->port_id, ADAPTER_TYPE_TERA);

        /* Save the adapter */
        struct tera_adapter * adapter = &tera_main->adapters[tera_adapter_id];
        adapter->vlan_int = htons((uint16_t)vlan_int);
        adapter->vlan_ext = htons((uint16_t)vlan_ext);
        adapter->mac_flipping = (uint8_t)mac_flipping;
        adapter->listen_device = dev_desc;
        adapter->tera_adapter_id = tera_adapter_id;

        nr_adapters++;
    }

    /* Check nr adapter */
    uint32_t nr_sids = sid_end - sid_start + 1;
    if (nr_adapters > nr_sids)
    {
        MR_ERROR("Tera adapters num:%u out of sid num: %u .", nr_adapters, nr_sids);
        return RT_ERR;
    }

    /* Inserter sid to forwarder table */
    for (int i = 0; i < nr_max_tera_adapters; i++)
    {
        struct tera_adapter * adapter = &tera_main->adapters[i];
        if (adapter->listen_device == NULL)
            continue;

        forwarder_table_insert(sid_start + adapter->tera_adapter_id, FORWARDER_TYPE_TERA);
    }

    /* Save the sid and nr adapters */
    tera_main->sid_start = sid_start;
    tera_main->sid_end = sid_end;
    tera_main->nr_sids = nr_sids;
    tera_main->nr_adapters = nr_adapters;

    return RT_SUCCESS;
}

/* Tera information dump */
void tera_info_dump(struct node_tera_main * tera_main)
{
    MR_INFO(" ");
    MR_INFO("Tera adapter, total num: %u, sid num:%u, sid_start:%u, sid_end:%u", tera_main->nr_adapters,
            tera_main->nr_sids, tera_main->sid_start, tera_main->sid_end);

    for (int index = 0; index < nr_max_tera_adapters; index++)
    {
        struct tera_adapter * adapter = &tera_main->adapters[index];
        if (adapter->listen_device == NULL)
            continue;

        MR_INFO("Tera adapter,tera adapter id:%u, listen device: %s, vlan_int: %u, vlan_ext: %u, "
                "mac_flipping:%u",
                adapter->tera_adapter_id, adapter->listen_device->symbol, ntohs(adapter->vlan_int),
                ntohs(adapter->vlan_ext), adapter->mac_flipping);
    }
}

/* Tera init */
int tera_init(struct sc_main * sc)
{
    /* Malloc tera main */
    struct node_tera_main * tera_main = node_tera_main_get();

    /* Load the Tera max entry; default is 256 */
    uint32_t max_rules;
    MESA_load_profile_uint_def(sc->local_cfgfile, "limits", "nr_max_tera_adapters", &max_rules, 256);

    /* According the max entry malloc tera adapter */
    nr_max_tera_adapters = (uint16_t)max_rules;
    tera_main->adapters = ZMALLOC(sizeof(struct tera_adapter) * max_rules);
    MR_VERIFY_MALLOC(tera_main->adapters);

    /* Parse the config */
    if (tera_config_parse(sc, tera_main) != RT_SUCCESS)
    {
        MR_ERROR("Tera config parse error.");
        return RT_ERR;
    }

    /* Dump the information */
    tera_info_dump(tera_main);

    /* Link db ctx create */
    uint32_t nr_max_link_dbs;
    MESA_load_profile_uint_def(sc->local_cfgfile, "limits", "nr_max_link_dbs", &nr_max_link_dbs, 32);
    tera_main->link_db_ctx = link_db_create(LINK_DB_TYPE_TERA, nr_max_link_dbs);
    if (tera_main->link_db_ctx == NULL)
        return RT_ERR;

    if (link_db_config_parse(sc->local_cfgfile, tera_main->link_db_ctx) == RT_ERR)
        return RT_ERR;

    return RT_SUCCESS;
}

/* Tera adapter match */
static inline struct tera_adapter * tera_adapter_match(const struct node_tera_main * tera_main, uint16_t port_id,
                                                       uint16_t pkt_vlan_id, uint8_t * out_dir,
                                                       uint16_t * out_vlan_couple)
{
    for (int index = 0; index < nr_max_tera_adapters; index++)
    {
        struct tera_adapter * adapter = &tera_main->adapters[index];

        if (adapter->listen_device == NULL)
        {
            continue;
        }

        if (adapter->listen_device->port_id != port_id)
        {
            continue;
        }

        if (adapter->vlan_int == pkt_vlan_id)
        {
            /* i to e */
            *out_dir = 0;
            *out_vlan_couple = adapter->vlan_ext;
            return adapter;
        }
        else if (adapter->vlan_ext == pkt_vlan_id)
        {
            /* e to i */
            *out_dir = 1;
            *out_vlan_couple = adapter->vlan_int;
            return adapter;
        }
    }
    return NULL;
}

/* Generate and store the trace information for ingress */
static __rte_always_inline void gen_store_trace_info_ingress(struct rte_node * node, struct rte_mbuf * mbuf,
                                                             uint16_t next_node_index,
                                                             struct tera_ingress_stats * stats, uint16_t vlan_id,
                                                             uint16_t vlan_couple, uint8_t mac_flipping,
                                                             uint16_t prepend_sid,
                                                             enum tera_ingress_drop_reason drop_reason)
{
    /* Populate the next node infomation */
    char str_record[MR_STRING_MAX];
    int len = snprintf(str_record, sizeof(str_record), "next node:%s", node->nodes[next_node_index]->name);

    /* Populate the drop reason */
    if (unlikely(next_node_index == TERA_INGRESS_NEXT_PKT_DROP))
    {
        assert(drop_reason < TERA_INGR_DROP_RSN_MAX);
        len +=
            snprintf(str_record + len, sizeof(str_record) - len, ", rsn:%s", tera_ingress_drop_reason_str[drop_reason]);
    }
    else
    {
        struct mrb_metadata * mrb_meta = mrbuf_cz_data(mbuf, MR_NODE_CTRLZONE_ID);
        len += snprintf(str_record + len, sizeof(str_record) - len,
                        ", tera id:%u, dir:%u, vlan:%u to %u, mac flipping:%u, prep sid:%u", mrb_meta->adapter_id,
                        mrb_meta->dir, ntohs(vlan_id), ntohs(vlan_couple), mac_flipping, prepend_sid);

        /* Populate the sids information */
        len += embed_sid_info(mbuf, str_record + len, sizeof(str_record) - len);
    }

    /* Emit the trace record */
    struct dp_trace_record_meta meta = {
        .measurement_type = DP_TRACE_MEASUREMENT_TYPE_TRACE, .appsym = MR_TRACE_APPSYM, .module = node->name};
    dp_trace_record_emit_str(sc_main_get()->trace, mbuf, rte_lcore_id(), &meta, str_record);
}

/* Tera ingress node process function */
static __rte_always_inline uint16_t tera_ingress_node_process(struct rte_graph * graph, struct rte_node * node,
                                                              void ** objs, uint16_t cnt)
{
    /* Get pkt num and pkt buffer */
    uint16_t last_spec = 0;
    uint16_t n_left_from = cnt;
    uint16_t batch_next_node_index = TERA_INGRESS_NEXT_CLASSIFIER;
    void ** batch_pkts = objs;
    struct rte_mbuf ** pkts = (struct rte_mbuf **)objs;

    /* Single Packet Processing */
    uint8_t prev_dir = UINT8_MAX;
    uint16_t prev_port_id = UINT16_MAX;
    uint16_t prev_vlan_id = UINT16_MAX;
    uint16_t prev_vlan_couple = UINT16_MAX;
    enum tera_ingress_drop_reason drop_reason = TERA_INGR_DROP_RSN_MAX;
    struct tera_ingress_stats stats = {};
    struct tera_adapter * prev_adapter = NULL;
    struct node_tera_main * tera_main = node_tera_main_get();

    while (n_left_from > 0)
    {
        struct rte_mbuf * mbuf = pkts[0];
        pkts += 1;
        n_left_from -= 1;

        /* Prefetch next mbuf */
        if (n_left_from > 0)
        {
            rte_prefetch0_write(pkts[0]);
            rte_prefetch0_write(mrbuf_cz_data(pkts[0], MR_NODE_CTRLZONE_ID));
        }

        struct mrb_metadata * mrb_meta = (struct mrb_metadata *)mrbuf_cz_data(mbuf, MR_NODE_CTRLZONE_ID);
        struct pkt_parser_result * parser_result = &mrb_meta->pkt_parser_result;

        /* Get vlan hdr */
        struct rte_vlan_hdr * vlan_hdr =
            rte_pktmbuf_mtod_offset(mbuf, struct rte_vlan_hdr *, parser_result->layers[1].offset);

        /* Get vlan id */
        uint16_t vlan_id = vlan_hdr->vlan_tci & 0xFF0F;

        /* Tera adapter match */
        if ((prev_vlan_id != vlan_id) || (prev_port_id != mrb_meta->port_ingress))
        {
            prev_port_id = mrb_meta->port_ingress;
            prev_vlan_id = vlan_id;
            prev_adapter = tera_adapter_match(tera_main, prev_port_id, prev_vlan_id, &prev_dir, &prev_vlan_couple);
        }

        uint16_t prepend_sid = 0;
        uint16_t next_node_index;
        if (prev_adapter == NULL)
        {
            drop_reason = TERA_INGR_DROP_RSN_NO_MATCH;
            stats.drop_reason[drop_reason]++;
            next_node_index = TERA_INGRESS_NEXT_PKT_DROP;
            goto node_enqueue;
        }

        /* Mac flipping */
        if (prev_adapter->mac_flipping)
        {
            /* Get eth hdr */
            struct rte_ether_hdr * eth_hdr =
                rte_pktmbuf_mtod_offset(mbuf, struct rte_ether_hdr *, parser_result->layers[0].offset);

            /* Swap mac addr */
            swap_mac_addr(eth_hdr);
        }

        /* Update the vlan id */
        vlan_hdr->vlan_tci = ((vlan_hdr->vlan_tci & 0x00F0) | prev_vlan_couple);

        /* Traffic link id match */
        uint16_t traffic_link_id;
        struct link_db_match_field match_field;
        match_field.tera_adapter_id = prev_adapter->tera_adapter_id;
        link_db_match(tera_main->link_db_ctx, &match_field, 1, &traffic_link_id);

        /* Fill mrb_meta */
        mrb_meta->adapter_type = ADAPTER_TYPE_TERA;
        mrb_meta->adapter_id = prev_adapter->tera_adapter_id;
        mrb_meta->dir = prev_dir;
        mrb_meta->traffic_link_id = traffic_link_id;

        /* Insert sid */
        prepend_sid = tera_main->sid_start + prev_adapter->tera_adapter_id;
        if (unlikely(sid_list_prepend(&mrb_meta->sid_list, &prepend_sid, 1) == RT_ERR))
        {
            drop_reason = TERA_INGR_DROP_RSN_APPEND_SID_ERR;
            stats.drop_reason[drop_reason]++;
            next_node_index = TERA_INGRESS_NEXT_PKT_DROP;
            goto node_enqueue;
        }

        /* Send the pkt to classifier */
        next_node_index = TERA_INGRESS_NEXT_CLASSIFIER;
        stats.to_classifier++;

    node_enqueue:
#if 0
        /* Check if tracing is enabled for the current Mbuf */
        if (unlikely(dp_trace_record_can_emit(mbuf, DP_TRACE_MEASUREMENT_TYPE_TRACE)))
        {
            gen_store_trace_info_ingress(node, mbuf, next_node_index, &stats, vlan_id, prev_vlan_couple,
                                         prev_adapter->mac_flipping, prepend_sid, drop_reason);
            // gen_store_trace_info_sid_list(node, mbuf);
        }
#endif

        /* Check if the next index needs to be changed */
        if (unlikely(batch_next_node_index != next_node_index))
        {
            /* Enqueue the last packets if the next index has changed */
            rte_node_enqueue(graph, node, batch_next_node_index, batch_pkts, last_spec);
            batch_pkts += last_spec;
            last_spec = 1;
            batch_next_node_index = next_node_index;
        }
        else
        {
            /* If the next index hasn't changed, update the last packets */
            last_spec++;
        }
    }

    /* Process any remaining packets */
    if (likely(last_spec > 0))
        rte_node_enqueue(graph, node, batch_next_node_index, batch_pkts, last_spec);

    /* Update graph stats */
    struct tera_ingress_stats * graph_stats = &ingress_stats_per_graph[graph->id];
    graph_stats->deal_pkts += cnt;
    graph_stats->pkts_per_batch = cnt;
    graph_stats->to_classifier += stats.to_classifier;

    for (int i = 0; i < TERA_INGR_DROP_RSN_MAX; i++)
    {
        graph_stats->drop_reason[i] += stats.drop_reason[i];
    }

    return cnt;
}

/* Tera ingress node base */
static struct rte_node_register tera_ingress_node_base = {
    .process = tera_ingress_node_process,
    .name = "tera_ingress",
    .init = NULL,
    .nb_edges = TERA_INGRESS_NEXT_MAX,
    .next_nodes =
        {
            [TERA_INGRESS_NEXT_CLASSIFIER] = "classifier",
            [TERA_INGRESS_NEXT_PKT_DROP] = "pkt_drop_trap",
        },
};

RTE_NODE_REGISTER(tera_ingress_node_base);

/* Generate and store the trace information for egress */
static __rte_always_inline void gen_store_trace_info_egress(struct rte_node * node, struct rte_mbuf * mbuf,
                                                            uint16_t next_node_index)
{

    /* Populate the next node infomation */
    char str_record[MR_STRING_MAX];
    struct mrb_metadata * mrb_meta = mrbuf_cz_data(mbuf, MR_NODE_CTRLZONE_ID);
    snprintf(str_record, sizeof(str_record), "next node:%s, tera id:%u, tx:%u", node->nodes[next_node_index]->name,
             mrb_meta->adapter_id, mrb_meta->port_egress);

    /* Emit the trace record */
    struct dp_trace_record_meta meta = {
        .measurement_type = DP_TRACE_MEASUREMENT_TYPE_TRACE, .appsym = MR_TRACE_APPSYM, .module = node->name};
    dp_trace_record_emit_str(sc_main_get()->trace, mbuf, rte_lcore_id(), &meta, str_record);
}

/* Tera egress node process function */
static __rte_always_inline uint16_t tera_egress_node_process(struct rte_graph * graph, struct rte_node * node,
                                                             void ** objs, uint16_t cnt)
{
    /* Get pkt num and pkt buffer */
    uint16_t n_left_from = cnt;
    struct rte_mbuf ** pkts = (struct rte_mbuf **)objs;

    /* Single Packet Processing */
    uint16_t prev_tera_adapter_id = UINT16_MAX;
    uint16_t prev_port_id = UINT16_MAX;
    struct node_tera_main * tera_main = node_tera_main_get();
    uint32_t sid_start = tera_main->sid_start;

    while (n_left_from > 0)
    {
        struct rte_mbuf * mbuf = pkts[0];
        pkts += 1;
        n_left_from -= 1;

        /* Prefetch next mbuf */
        if (n_left_from > 0)
        {
            rte_prefetch0_write(pkts[0]);
            rte_prefetch0_write(mrbuf_cz_data(pkts[0], MR_NODE_CTRLZONE_ID));
        }

        /* Get tera id  */
        struct mrb_metadata * mrb_meta = (struct mrb_metadata *)mrbuf_cz_data(mbuf, MR_NODE_CTRLZONE_ID);
        uint16_t tera_adapter_id = mrb_meta->cur_sid - sid_start;

        assert(tera_adapter_id < nr_max_tera_adapters);

        if (prev_tera_adapter_id != tera_adapter_id)
        {
            struct tera_adapter * tera_adapter = &tera_main->adapters[tera_adapter_id];

            assert(tera_adapter->listen_device != NULL);

            prev_tera_adapter_id = tera_adapter_id;
            prev_port_id = tera_adapter->listen_device->port_id;
        }

        /* Fill port egress */
        mrb_meta->port_egress = prev_port_id;

        /* Check if tracing is enabled for the current Mbuf */
        if (unlikely(dp_trace_record_can_emit(mbuf, DP_TRACE_MEASUREMENT_TYPE_TRACE)))
        {
            gen_store_trace_info_egress(node, mbuf, TERA_EGRESS_NEXT_ETH_EGRESS);
        }
    }

    /* Update graph stats */
    struct tera_egress_stats * graph_stats = &egress_stats_per_graph[graph->id];
    graph_stats->deal_pkts += cnt;
    graph_stats->pkts_per_batch = cnt;

    /* Move to next node */
    rte_node_next_stream_move(graph, node, TERA_EGRESS_NEXT_ETH_EGRESS);
    return cnt;
}

/* Tera egress node base */
static struct rte_node_register tera_egress_node_base = {
    .process = tera_egress_node_process,
    .name = "tera_egress",
    .init = NULL,
    .nb_edges = TERA_EGRESS_NEXT_MAX,
    .next_nodes =
        {
            [TERA_EGRESS_NEXT_ETH_EGRESS] = "eth_egress",
            [TERA_EGRESS_NEXT_PKT_DROP] = "pkt_drop_trap",
        },
};

RTE_NODE_REGISTER(tera_egress_node_base);

/************************************** Tera Statistics **************************************/
cJSON * tera_ingress_node_monit_loop(struct sc_main * sc)
{
    cJSON * json_root = cJSON_CreateObject();
    unsigned int nr_graphs = sc->nr_io_thread;

    uint64_t total_pkts[nr_graphs];
    uint64_t pkts_per_batch[nr_graphs];
    uint64_t to_classifier[nr_graphs];
    uint64_t drop_reason[TERA_INGR_DROP_RSN_MAX][nr_graphs];

    for (uint32_t graph_id = 0; graph_id < nr_graphs; graph_id++)
    {
        struct tera_ingress_stats * stats = &ingress_stats_per_graph[graph_id];
        if (stats->deal_pkts == 0)
        {
            total_pkts[graph_id] = 0;
            pkts_per_batch[graph_id] = 0;
            to_classifier[graph_id] = 0;

            for (int i = 0; i < TERA_INGR_DROP_RSN_MAX; i++)
            {
                drop_reason[i][graph_id] = 0;
            }

            continue;
        }

        total_pkts[graph_id] = stats->deal_pkts;
        pkts_per_batch[graph_id] = stats->pkts_per_batch;
        to_classifier[graph_id] = stats->to_classifier;

        for (int i = 0; i < TERA_INGR_DROP_RSN_MAX; i++)
        {
            drop_reason[i][graph_id] = stats->drop_reason[i];
        }
    }

    cJSON * json_total_pkts = create_uint64_array(total_pkts, nr_graphs);
    cJSON_AddItemToObject(json_root, "tera ingress, total_pkts", json_total_pkts);

    cJSON * json_pkts_per_batch = create_uint64_array(pkts_per_batch, nr_graphs);
    cJSON_AddItemToObject(json_root, "tera ingress, pkts_per_batch", json_pkts_per_batch);

    cJSON * json_to_classifier = create_uint64_array(to_classifier, nr_graphs);
    cJSON_AddItemToObject(json_root, "tera ingress, to_classifier", json_to_classifier);

    for (int i = 0; i < TERA_INGR_DROP_RSN_MAX; i++)
    {
        char str_title[MR_STRING_MAX];
        snprintf(str_title, sizeof(str_title), "tera ingress, %s", tera_ingress_drop_reason_str[i]);

        cJSON * json_drop_reason = create_uint64_array(drop_reason[i], nr_graphs);
        cJSON_AddItemToObject(json_root, str_title, json_drop_reason);
    }

    return json_root;
}

cJSON * tera_egress_node_monit_loop(struct sc_main * sc)
{
    cJSON * json_root = cJSON_CreateObject();
    unsigned int nr_graphs = sc->nr_io_thread;

    uint64_t total_pkts[nr_graphs];
    uint64_t pkts_per_batch[nr_graphs];

    for (uint32_t graph_id = 0; graph_id < nr_graphs; graph_id++)
    {
        struct tera_egress_stats * stats = &egress_stats_per_graph[graph_id];
        if (stats->deal_pkts == 0)
        {
            total_pkts[graph_id] = 0;
            pkts_per_batch[graph_id] = 0;
            continue;
        }

        total_pkts[graph_id] = stats->deal_pkts;
        pkts_per_batch[graph_id] = stats->pkts_per_batch;
    }

    cJSON * json_total_pkts = create_uint64_array(total_pkts, nr_graphs);
    cJSON_AddItemToObject(json_root, "tera egress, total_pkts", json_total_pkts);

    cJSON * json_pkts_per_batch = create_uint64_array(pkts_per_batch, nr_graphs);
    cJSON_AddItemToObject(json_root, "tera egress, pkts_per_batch", json_pkts_per_batch);
    return json_root;
}