path: root/infra/session_manager/session_manager.c

#include <assert.h>

#include "utils_internal.h"
#include "tcp_reassembly.h"
#include "session_internal.h"
#include "session_manager.h"
#include "session_manager_log.h"
#include "session_manager_cfg.h"
#include "session_manager_rte.h"
#include "session_manager_stat.h"
#include "fieldstat/fieldstat_easy.h"

#define CLEAN_SESSION_BURST 1024

#pragma GCC diagnostic ignored "-Wunused-parameter"

struct session_manager
{
     int pkt_ex_to_get_sess;
     int pkt_ex_to_free_sess;
     int pkt_ex_to_tcp_segment;
     struct exdata_schema *ex_sche;

     int stat_idx[SESS_MGR_STAT_MAX];
     struct fieldstat_easy *fs;
     struct session_manager_cfg *cfg;
     struct session_manager_rte *rte[MAX_THREAD_NUM];
     struct module_manager *mod_mgr;
     struct packet_manager *pkt_mgr;
};

static void free_session(int idx, void *ex_ptr, void *arg)
{
     struct session *sess = (struct session *)ex_ptr;
     struct session_manager *sess_mgr = (struct session_manager *)arg;
     assert(idx == sess_mgr->pkt_ex_to_free_sess);

     if (sess && session_get_current_state(sess) == SESSION_STATE_CLOSED)
     {
          int thread_id = module_manager_get_thread_id(sess_mgr->mod_mgr);
          struct session_manager_rte *sess_mgr_rte = session_manager_get_rte(sess_mgr, thread_id);

          char buffer[4096] = {0};
          session_to_str(sess, 0, buffer, sizeof(buffer));
          SESSION_MANAGER_LOG_INFO("session free: %s", buffer);

          struct exdata_runtime *exdata_rt = (struct exdata_runtime *)session_get_user_data(sess);
          exdata_runtime_free(exdata_rt);
          session_manager_rte_free_session(sess_mgr_rte, sess);
     }
}

static void free_tcp_segment(int idx, void *ex_ptr, void *arg)
{
     struct tcp_segment *segs = (struct tcp_segment *)ex_ptr;
     while (segs)
     {
          struct session *sess = (struct session *)tcp_segment_get_user_data(segs);
          struct tcp_segment *next = segs->next;
          session_free_tcp_segment(sess, segs);
          segs = next;
     }
}

static void notify_sess_closed_by_pseudo_pkt(struct session_manager *sess_mgr, int thread_id, struct session *sess)
{
     struct packet *pseudo = NULL;
     struct packet_manager *pkt_mgr = sess_mgr->pkt_mgr;
     assert(session_get_current_state(sess) == SESSION_STATE_CLOSED);

     if (session_get_first_packet(sess, FLOW_TYPE_C2S))
     {
          pseudo = packet_manager_dup_packet(pkt_mgr, thread_id, session_get_first_packet(sess, FLOW_TYPE_C2S));
     }
     else
     {
          pseudo = packet_manager_dup_packet(pkt_mgr, thread_id, session_get_first_packet(sess, FLOW_TYPE_S2C));
     }

     assert(pseudo);
     packet_tag_set(pseudo, PKT_TAG_KEY_SESS, PKT_TAG_VAL_SESS_FREE);
     packet_set_type(pseudo, PACKET_TYPE_PSEUDO);
     packet_set_action(pseudo, PACKET_ACTION_DROP);
     packet_set_exdata(pseudo, sess_mgr->pkt_ex_to_get_sess, sess);
     packet_set_exdata(pseudo, sess_mgr->pkt_ex_to_free_sess, sess);
     packet_manager_schedule_packet(pkt_mgr, thread_id, pseudo, PACKET_STAGE_FORWARD);
     SESSION_MANAGER_LOG_INFO("notify session %lu %s closed by pseudo packet: %p", session_get_id(sess), session_get_readable_addr(sess), pseudo);
}

static void on_polling(struct module_manager *mod_mgr, void *args)
{
     struct session *sess = NULL;
     uint64_t now_ms = clock_get_real_time_ms();
     struct session_manager *sess_mgr = (struct session_manager *)args;
     int thread_id = module_manager_get_thread_id(mod_mgr);
     struct session_manager_rte *sess_mgr_rte = session_manager_get_rte(sess_mgr, thread_id);

     static __thread uint64_t last_clean_expired_sess_ts = 0;
     if (now_ms - last_clean_expired_sess_ts > sess_mgr->cfg->expire_period_ms)
     {
          for (uint64_t i = 0; i < sess_mgr->cfg->expire_batch_max; i++)
          {
               sess = session_manager_rte_get_expired_session(sess_mgr_rte, now_ms);
               if (sess)
               {
                    notify_sess_closed_by_pseudo_pkt(sess_mgr, thread_id, sess);
               }
               else
               {
                    break;
               }
          }
          last_clean_expired_sess_ts = now_ms;
     }

     static __thread uint64_t last_sync_stat_ms = 0;
     static __thread struct session_manager_stat last_stat = {0};
     if (now_ms - last_sync_stat_ms >= SYNC_STAT_INTERVAL_MS)
     {
          struct session_manager_stat *sess_mgr_stat = session_manager_rte_get_stat(sess_mgr_rte);
          for (int i = 0; i < SESS_MGR_STAT_MAX; i++)
          {
               uint64_t val = session_manager_stat_get(sess_mgr_stat, i) - session_manager_stat_get(&last_stat, i);
               fieldstat_easy_counter_incrby(sess_mgr->fs, thread_id, sess_mgr->stat_idx[i], NULL, 0, val);
          }
          last_stat = *sess_mgr_stat;
          last_sync_stat_ms = now_ms;
     }
}

/******************************************************************************
 * session manager
 ******************************************************************************/

void session_manager_on_packet_forward(struct packet *pkt, struct module *mod)
{
     struct session_manager *sess_mgr = (struct session_manager *)module_get_ctx(mod);
     int thread_id = module_manager_get_thread_id(sess_mgr->mod_mgr);
     struct session_manager_rte *sess_mgr_rte = session_manager_get_rte(sess_mgr, thread_id);

     /*
      * We use the system's real time instead of monotonic time for the following reasons:
      *     -> Session creation/closure times require real time (e.g., for logging session activities).
      *     -> Session ID generation relies on real time        (e.g., for reverse calculating session creation time from the session ID).
      *
      * Note: Modifying the system time will affect the timing wheel, impacting session expiration, and TCP reassembly expiration.
      * Suggestion: After modifying the system time, restart the service to ensure consistent timing.
      */
     uint64_t now_ms = 0;
     struct tuple6 key;
     struct session *sess = NULL;
     if (packet_get_type(pkt) == PACKET_TYPE_PSEUDO)
     {
          sess = packet_exdata_to_session(sess_mgr, pkt);
          if (sess || (sess = session_manager_rte_lookup_session_by_packet(sess_mgr_rte, pkt)))
          {
               session_set_current_packet(sess, pkt);
               packet_get_innermost_tuple6(pkt, &key);
               if (tuple6_cmp(session_get_tuple6(sess), &key) == 0)
               {
                    session_set_flow_type(sess, FLOW_TYPE_C2S);
                    sess->stats[FLOW_TYPE_C2S][STAT_PSEUDO_PACKETS_RECEIVED]++;
                    sess->stats[FLOW_TYPE_C2S][STAT_PSEUDO_BYTES_RECEIVED] += packet_get_raw_len(pkt);
               }
               else
               {
                    session_set_flow_type(sess, FLOW_TYPE_S2C);
                    sess->stats[FLOW_TYPE_S2C][STAT_PSEUDO_PACKETS_RECEIVED]++;
                    sess->stats[FLOW_TYPE_S2C][STAT_PSEUDO_BYTES_RECEIVED] += packet_get_raw_len(pkt);
               }
               packet_tag_set(pkt, PKT_TAG_KEY_SESS, PKT_TAG_VAL_SESS_PSEUDOPKT);
               packet_set_exdata(pkt, sess_mgr->pkt_ex_to_get_sess, sess);
          }
          else
          {
               // TODO new session by pseudo packet
          }
     }
     else
     {
          now_ms = clock_get_real_time_ms();
          sess = session_manager_rte_lookup_session_by_packet(sess_mgr_rte, pkt);
          if (sess || (sess = session_manager_rte_new_session(sess_mgr_rte, pkt, now_ms)))
          {
               if (session_get_user_data(sess) == NULL)
               {
                    session_set_user_data(sess, exdata_runtime_new(sess_mgr->ex_sche));
                    packet_tag_set(pkt, PKT_TAG_KEY_SESS, PKT_TAG_VAL_SESS_NEW);
               }
               session_manager_rte_update_session(sess_mgr_rte, sess, pkt, now_ms);
               packet_tag_set(pkt, PKT_TAG_KEY_SESS, PKT_TAG_VAL_SESS_RAWPKT);
               packet_set_exdata(pkt, sess_mgr->pkt_ex_to_get_sess, sess);
          }
          else
          {
               // miss sess
          }
     }

     if (sess && session_get_type(sess) == SESSION_TYPE_TCP)
     {
          struct tcp_segment *prev = NULL;
          struct tcp_segment *next = NULL;
          while ((next = session_get_tcp_segment(sess)))
          {
               tcp_segment_set_user_data(next, sess);
               if (prev == NULL)
               {
                    prev = next;
                    prev->next = NULL;
                    packet_set_exdata(pkt, sess_mgr->pkt_ex_to_tcp_segment, prev);
                    packet_tag_set(pkt, PKT_TAG_KEY_SESS, PKT_TAG_VAL_SESS_TCP_STREAM);
               }
               else
               {
                    prev->next = next;
                    prev = next;
                    prev->next = NULL;
               }
          }
     }

     while ((sess = session_manager_rte_get_evicted_session(sess_mgr_rte)))
     {
          notify_sess_closed_by_pseudo_pkt(sess_mgr, thread_id, sess);
     }
}

void session_manager_on_packet_output(struct packet *pkt, struct module *mod)
{
     struct session_manager *sess_mgr = (struct session_manager *)module_get_ctx(mod);
     int thread_id = module_manager_get_thread_id(sess_mgr->mod_mgr);
     struct session_manager_rte *sess_mgr_rte = session_manager_get_rte(sess_mgr, thread_id);

     struct session *sess = (struct session *)packet_get_exdata(pkt, sess_mgr->pkt_ex_to_get_sess);
     if (sess)
     {
          struct tuple6 key;
          enum flow_type flow = FLOW_TYPE_NONE;
          packet_get_innermost_tuple6(pkt, &key);
          if (tuple6_cmp(session_get_tuple6(sess), &key) == 0)
          {
               flow = FLOW_TYPE_C2S;
          }
          else
          {
               flow = FLOW_TYPE_S2C;
          }

          int is_pseudo = (packet_get_type(pkt) == PACKET_TYPE_PSEUDO);
          uint16_t len = packet_get_raw_len(pkt);
          switch (packet_get_action(pkt))
          {
          case PACKET_ACTION_DROP:
               session_inc_stat(sess, flow, (is_pseudo ? STAT_PSEUDO_PACKETS_DROPPED : STAT_RAW_PACKETS_DROPPED), 1);
               session_inc_stat(sess, flow, (is_pseudo ? STAT_PSEUDO_BYTES_DROPPED : STAT_RAW_BYTES_DROPPED), len);
               break;
          case PACKET_ACTION_FORWARD:
               session_inc_stat(sess, flow, (is_pseudo ? STAT_PSEUDO_PACKETS_TRANSMITTED : STAT_RAW_PACKETS_TRANSMITTED), 1);
               session_inc_stat(sess, flow, (is_pseudo ? STAT_PSEUDO_BYTES_TRANSMITTED : STAT_RAW_BYTES_TRANSMITTED), len);
               break;
          default:
               assert(0);
               break;
          }

          session_set_current_packet(sess, NULL);
     }

     if (packet_get_origin(pkt) == NULL)
     {
          session_manager_rte_record_duplicated_packet(sess_mgr_rte, pkt);
     }
}

struct session_manager *session_manager_new(struct packet_manager *pkt_mgr, const char *toml_file)
{
     struct session_manager *sess_mgr = calloc(1, sizeof(struct session_manager));
     if (sess_mgr == NULL)
     {
          SESSION_MANAGER_LOG_ERROR("failed to allocate memory for session_manager");
          return NULL;
     }

     sess_mgr->cfg = session_manager_cfg_new(toml_file);
     if (sess_mgr->cfg == NULL)
     {
          SESSION_MANAGER_LOG_ERROR("failed to create session_manager_cfg");
          goto error_out;
     }
     session_manager_cfg_print(sess_mgr->cfg);

     sess_mgr->fs = fieldstat_easy_new(sess_mgr->cfg->thread_num, "session_manager", NULL, 0);
     if (sess_mgr->fs == NULL)
     {
          SESSION_MANAGER_LOG_ERROR("failed to create fieldstat_easy");
          goto error_out;
     }
     for (int i = 0; i < SESS_MGR_STAT_MAX; i++)
     {
          sess_mgr->stat_idx[i] = fieldstat_easy_register_counter(sess_mgr->fs, sess_mgr_stat_str[i]);
     }
     if (fieldstat_easy_enable_auto_output(sess_mgr->fs, "metrics/session_manager.json", 2) != 0)
     {
          SESSION_MANAGER_LOG_ERROR("failed to enable auto output");
          goto error_out;
     }
#if 0
     if (packet_manager_register_node(pkt_mgr, "session_manager", PACKET_STAGE_FORWARD, PKT_TAG_KEY_IPPROTO, PKT_TAG_VAL_IPPROTO_TCP | PKT_TAG_VAL_IPPROTO_UDP, session_manager_on_packet_forward, sess_mgr))
     {
          SESSION_MANAGER_LOG_ERROR("failed to subscribe PACKET_STAGE_FORWARD");
          goto error_out;
     }
     if (packet_manager_register_node(pkt_mgr, "session_manager", PACKET_STAGE_OUTPUT, PKT_TAG_KEY_IPPROTO, PKT_TAG_VAL_IPPROTO_TCP | PKT_TAG_VAL_IPPROTO_UDP, session_manager_on_packet_output, sess_mgr))
     {
          SESSION_MANAGER_LOG_ERROR("failed to subscribe PACKET_STAGE_OUTPUT");
          goto error_out;
     }
#endif

     sess_mgr->ex_sche = exdata_schema_new();
     if (sess_mgr->ex_sche == NULL)
     {
          SESSION_MANAGER_LOG_ERROR("failed to create exdata_schema");
          goto error_out;
     }

     sess_mgr->pkt_ex_to_get_sess = packet_manager_new_packet_exdata_index(pkt_mgr, "pkt_ex_key_for_get_sess", NULL, NULL);
     if (sess_mgr->pkt_ex_to_get_sess == -1)
     {
          SESSION_MANAGER_LOG_ERROR("failed to create packet exdata index");
          goto error_out;
     }

     sess_mgr->pkt_ex_to_free_sess = packet_manager_new_packet_exdata_index(pkt_mgr, "pkt_ex_key_for_free_sess", free_session, sess_mgr);
     if (sess_mgr->pkt_ex_to_free_sess == -1)
     {
          SESSION_MANAGER_LOG_ERROR("failed to create packet exdata index");
          goto error_out;
     }

     sess_mgr->pkt_ex_to_tcp_segment = packet_manager_new_packet_exdata_index(pkt_mgr, "pkt_ex_key_for_tcp_stream", free_tcp_segment, sess_mgr);
     if (sess_mgr->pkt_ex_to_tcp_segment == -1)
     {
          SESSION_MANAGER_LOG_ERROR("failed to create packet exdata index");
          goto error_out;
     }

     sess_mgr->pkt_mgr = pkt_mgr;
     return sess_mgr;

error_out:
     session_manager_free(sess_mgr);
     return NULL;
}

void session_manager_free(struct session_manager *sess_mgr)
{
     if (sess_mgr)
     {
          if (sess_mgr->ex_sche)
          {
               exdata_schema_free(sess_mgr->ex_sche);
          }
          if (sess_mgr->fs)
          {
               fieldstat_easy_free(sess_mgr->fs);
          }
          if (sess_mgr->cfg)
          {
               session_manager_cfg_free(sess_mgr->cfg);
          }
          free(sess_mgr);
     }
}

struct session_manager_cfg *session_manager_get_cfg(const struct session_manager *sess_mgr)
{
     assert(sess_mgr);
     return sess_mgr->cfg;
}

struct session_manager_rte *session_manager_get_rte(const struct session_manager *sess_mgr, uint16_t thread_id)
{
     assert(sess_mgr);
     if (thread_id < sess_mgr->cfg->thread_num)
     {
          return sess_mgr->rte[thread_id];
     }
     else
     {
          return NULL;
     }
}

int session_manager_new_session_exdata_index(struct session_manager *sess_mgr, const char *name, exdata_free *func, void *arg)
{
     assert(sess_mgr);
     assert(name);
     assert(func);
     return exdata_schema_new_index(sess_mgr->ex_sche, name, func, arg);
}

int session_manager_init(struct session_manager *sess_mgr, uint16_t thread_id)
{
     assert(sess_mgr);
     uint64_t now_ms = clock_get_real_time_ms();

     sess_mgr->cfg->session_id_seed = sess_mgr->cfg->instance_id << 8 | thread_id;
     sess_mgr->rte[thread_id] = session_manager_rte_new(sess_mgr->cfg, now_ms);
     if (sess_mgr->rte[thread_id] == NULL)
     {
          SESSION_MANAGER_LOG_ERROR("failed to create session_manager_rte");
          return -1;
     }
     else
     {
          return 0;
     }
}

void session_manager_clean(struct session_manager *sess_mgr, uint16_t thread_id)
{
     assert(sess_mgr);

     struct session_manager_rte *rte = session_manager_get_rte(sess_mgr, thread_id);

     if (rte == NULL)
     {
          return;
     }

     struct session *sess = NULL;
     struct session_manager_stat *stat = session_manager_rte_get_stat(rte);

     while (stat->tcp_sess_used || stat->udp_sess_used)
     {
          while ((sess = session_manager_rte_get_expired_session(rte, UINT64_MAX)))
          {
               exdata_runtime_free((struct exdata_runtime *)session_get_user_data(sess));
               session_manager_rte_free_session(rte, sess);
          }
     }

     session_manager_rte_free(rte);
}

struct session *session_manager_lookup_session_by_packet(struct session_manager *sess_mgr, uint16_t thread_id, const struct packet *pkt)
{
     struct session_manager_rte *rte = session_manager_get_rte(sess_mgr, thread_id);
     if (rte == NULL)
     {
          return NULL;
     }
     return session_manager_rte_lookup_session_by_packet(rte, pkt);
}

struct session *session_manager_lookup_session_by_id(struct session_manager *sess_mgr, uint16_t thread_id, uint64_t sess_id)
{
     struct session_manager_rte *rte = session_manager_get_rte(sess_mgr, thread_id);
     if (rte == NULL)
     {
          return NULL;
     }
     return session_manager_rte_lookup_session_by_id(rte, sess_id);
}

void session_manager_discard_session(struct session_manager *sess_mgr, uint16_t thread_id, struct session *sess)
{
     struct session_manager_rte *rte = session_manager_get_rte(sess_mgr, thread_id);
     if (rte == NULL)
     {
          return;
     }
     session_manager_rte_discard_session(rte, sess);
}

struct session *packet_exdata_to_session(const struct session_manager *sess_mgr, const struct packet *pkt)
{
     return (struct session *)packet_get_exdata(pkt, sess_mgr->pkt_ex_to_get_sess);
}

struct tcp_segment *packet_exdata_to_tcp_segment(const struct session_manager *sess_mgr, const struct packet *pkt)
{
     return (struct tcp_segment *)packet_get_exdata(pkt, sess_mgr->pkt_ex_to_tcp_segment);
}

/******************************************************************************
 * session manager module
 ******************************************************************************/

struct session_manager *module_to_session_manager(struct module *mod)
{
     assert(mod);
     assert(strcmp(module_get_name(mod), SESSION_MANAGER_MODULE_NAME) == 0);
     return module_get_ctx(mod);
}

struct module *session_manager_on_init(struct module_manager *mod_mgr)
{
     assert(mod_mgr);
     struct module *pkt_mgr_mod = module_manager_get_module(mod_mgr, PACKET_MANAGER_MODULE_NAME);
     struct packet_manager *pkt_mgr = module_to_packet_manager(pkt_mgr_mod);
     assert(pkt_mgr);
     const char *toml_file = module_manager_get_toml_path(mod_mgr);
     assert(toml_file);

     struct session_manager *sess_mgr = session_manager_new(pkt_mgr, toml_file);
     if (sess_mgr == NULL)
     {
          return NULL;
     }
     sess_mgr->mod_mgr = mod_mgr;
     module_manager_register_polling_node(mod_mgr, on_polling, sess_mgr);

     struct module *sess_mgr_mod = module_new(SESSION_MANAGER_MODULE_NAME, NULL);
     if (sess_mgr_mod == NULL)
     {
          SESSION_MANAGER_LOG_ERROR("failed to create session_manager");
          session_manager_free(sess_mgr);
          return NULL;
     }
     module_set_ctx(sess_mgr_mod, sess_mgr);

     SESSION_MANAGER_LOG_FATAL("session_manager init");
     return sess_mgr_mod;
}

void session_manager_on_exit(struct module_manager *mod_mgr, struct module *mod)
{
     if (mod)
     {
          struct session_manager *sess_mgr = module_get_ctx(mod);
          session_manager_free(sess_mgr);
          module_free(mod);
          SESSION_MANAGER_LOG_FATAL("session_manager exit");
     }
}

struct module *session_manager_on_thread_init(struct module_manager *mod_mgr, int thread_id, struct module *mod)
{
     struct session_manager *sess_mgr = module_get_ctx(mod);
     assert(sess_mgr);

     if (session_manager_init(sess_mgr, thread_id) != 0)
     {
          SESSION_MANAGER_LOG_ERROR("failed to int session_manager_init");
          return NULL;
     }
     else
     {
          return mod;
     }
}

void session_manager_on_thread_exit(struct module_manager *mod_mgr, int thread_id, struct module *mod)
{
     struct session_manager *sess_mgr = module_get_ctx(mod);
     if (sess_mgr)
     {
          assert(thread_id < (int)sess_mgr->cfg->thread_num);
          session_manager_clean(sess_mgr, thread_id);
     }
}