gfwleak/tango-tfe
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

packet io增加fieldstat状态统计

Browse Source
This commit is contained in:
wangmenglan
2023-05-15 16:41:59 +08:00
parent 935aa7235e
commit b931a3dc58
17 changed files with 780 additions and 757 deletions
Download Patch File Download Diff File Expand all files Collapse all files

384
common/src/tfe_packet_io.cpp
View File

@@ -184,12 +184,12 @@ static int is_downstream_keepalive_packet(marsio_buff_t *rx_buff)
}
}
static int tfe_tap_write_per_thread(int tap_fd, const char *data, int data_len, void *logger)
static int tap_write(int tap_fd, const char *data, int data_len, void *logger)
{
int ret = write(tap_fd, data, data_len);
if (ret != data_len)
{
TFE_LOG_ERROR(g_default_logger, "%s: need send %dB, only send %dB, aborting: %s", LOG_TAG_PKTIO, data_len, ret, strerror(errno));
TFE_LOG_ERROR(logger, "%s: need send %dB, only send %dB, aborting: %s", LOG_TAG_PKTIO, data_len, ret, strerror(errno));
}
return ret;
@@ -456,7 +456,7 @@ static int fake_tcp_handshake(struct tfe_proxy *proxy, struct tcp_restore_info *
return 0;
}
static int overwrite_tcp_mss(struct tfe_cmsg *cmsg, struct tcp_restore_info *restore)
static int overwrite_tcp_mss(struct tfe_cmsg *cmsg, struct tcp_restore_info *restore, uint64_t session_id, void *logger)
{
int ret = 0;
uint16_t size = 0;
@@ -468,25 +468,25 @@ static int overwrite_tcp_mss(struct tfe_cmsg *cmsg, struct tcp_restore_info *res
ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_DOWNSTREAM_TCP_MSS_ENABLE, (unsigned char *)&client_side_mss_enable, sizeof(client_side_mss_enable), &size);
if (ret < 0)
{
TFE_LOG_ERROR(g_default_logger, "failed at fetch client side tcp mss from cmsg: %s", strerror(-ret));
TFE_LOG_ERROR(logger, "%s: session %lu failed at fetch client side tcp mss from cmsg: %s", LOG_TAG_PKTIO, session_id, strerror(-ret));
return -1;
}
ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_DOWNSTREAM_TCP_MSS_VALUE, (unsigned char *)&client_side_mss_value, sizeof(client_side_mss_value), &size);
if (ret < 0)
{
TFE_LOG_ERROR(g_default_logger, "failed at fetch client side tcp mss value from cmsg: %s", strerror(-ret));
TFE_LOG_ERROR(logger, "%s: session %lu failed at fetch client side tcp mss value from cmsg: %s", LOG_TAG_PKTIO, session_id, strerror(-ret));
return -1;
}
ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_UPSTREAM_TCP_MSS_ENABLE, (unsigned char *)&server_side_mss_enable, sizeof(server_side_mss_enable), &size);
if (ret < 0)
{
TFE_LOG_ERROR(g_default_logger, "failed at fetch server side tcp mss from cmsg: %s", strerror(-ret));
TFE_LOG_ERROR(logger, "%s: session %lu failed at fetch server side tcp mss from cmsg: %s", LOG_TAG_PKTIO, session_id, strerror(-ret));
return -1;
}
ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_UPSTREAM_TCP_MSS_VALUE, (unsigned char *)&server_side_mss_value, sizeof(server_side_mss_value), &size);
if (ret < 0)
{
TFE_LOG_ERROR(g_default_logger, "failed at fetch server side tcp mss value from cmsg: %s", strerror(-ret));
TFE_LOG_ERROR(logger, "%s: session %lu failed at fetch server side tcp mss value from cmsg: %s", LOG_TAG_PKTIO, session_id, strerror(-ret));
return -1;
}
if (client_side_mss_enable)
@@ -700,7 +700,7 @@ static int tcp_restore_set_from_pkg(struct addr_tuple4 *tuple4, struct tcp_resto
// return 0 : success
// return -1 : error
static int packet_io_config(const char *profile, struct config *config)
static int packet_io_config(const char *profile, struct config *config, void *logger)
{
int ret = 0;
@@ -730,77 +730,77 @@ static int packet_io_config(const char *profile, struct config *config)
char src_mac_addr[18] = {0};
ret = MESA_load_profile_string_nodef(profile, "PACKET_IO", "src_mac_addr", src_mac_addr, sizeof(src_mac_addr));
if(ret < 0){
TFE_LOG_ERROR(g_default_logger, "%s: invalid src_mac_addr: src_mac_addr not set, profile = %s, section = PACKET_IO", LOG_TAG_PKTIO, profile);
TFE_LOG_ERROR(logger, "%s: invalid src_mac_addr: src_mac_addr not set, profile = %s, section = PACKET_IO", LOG_TAG_PKTIO, profile);
return -1;
}
str_to_mac(src_mac_addr, config->src_mac);
ret = get_mac_by_device_name(config->dev_tap, config->tap_mac);
if (ret != 0) {
TFE_LOG_ERROR(g_default_logger, "%s: invalid tap_name: unable get %s mac", LOG_TAG_PKTIO, config->dev_tap);
TFE_LOG_ERROR(logger, "%s: invalid tap_name: unable get %s mac", LOG_TAG_PKTIO, config->dev_tap);
return -1;
}
ret = get_mac_by_device_name(config->dev_tap_c, config->tap_c_mac);
if (ret != 0) {
TFE_LOG_ERROR(g_default_logger, "%s: invalid device_client: unable get %s mac", LOG_TAG_PKTIO, config->dev_tap_c);
TFE_LOG_ERROR(logger, "%s: invalid device_client: unable get %s mac", LOG_TAG_PKTIO, config->dev_tap_c);
return -1;
}
ret = get_mac_by_device_name(config->dev_tap_s, config->tap_s_mac);
if (ret != 0) {
TFE_LOG_ERROR(g_default_logger, "%s: invalid device_server: unable get %s mac", LOG_TAG_PKTIO, config->dev_tap_s);
TFE_LOG_ERROR(logger, "%s: invalid device_server: unable get %s mac", LOG_TAG_PKTIO, config->dev_tap_s);
return -1;
}
if (config->rx_burst_max > RX_BURST_MAX)
{
TFE_LOG_ERROR(g_default_logger, "%s: invalid rx_burst_max, exceeds limit %d", LOG_TAG_PKTIO, RX_BURST_MAX);
TFE_LOG_ERROR(logger, "%s: invalid rx_burst_max, exceeds limit %d", LOG_TAG_PKTIO, RX_BURST_MAX);
return -1;
}
if (strlen(config->app_symbol) == 0)
{
TFE_LOG_ERROR(g_default_logger, "%s: invalid app_symbol in %s", LOG_TAG_PKTIO, profile);
TFE_LOG_ERROR(logger, "%s: invalid app_symbol in %s", LOG_TAG_PKTIO, profile);
return -1;
}
if (strlen(config->dev_nf_interface) == 0)
{
TFE_LOG_ERROR(g_default_logger, "%s: invalid dev_nf_interface in %s", LOG_TAG_PKTIO, profile);
TFE_LOG_ERROR(logger, "%s: invalid dev_nf_interface in %s", LOG_TAG_PKTIO, profile);
return -1;
}
TFE_LOG_DEBUG(g_default_logger, "%s: PACKET_IO->bypass_all_traffic : %d", LOG_TAG_PKTIO, config->bypass_all_traffic);
TFE_LOG_DEBUG(g_default_logger, "%s: PACKET_IO->rx_burst_max : %d", LOG_TAG_PKTIO, config->rx_burst_max);
TFE_LOG_DEBUG(g_default_logger, "%s: PACKET_IO->app_symbol : %s", LOG_TAG_PKTIO, config->app_symbol);
TFE_LOG_DEBUG(g_default_logger, "%s: PACKET_IO->dev_nf_interface : %s", LOG_TAG_PKTIO, config->dev_nf_interface);
TFE_LOG_DEBUG(g_default_logger, "%s: PACKET_IO->tap_name : %s", LOG_TAG_PKTIO, config->tap_rps_mask);
TFE_LOG_DEBUG(g_default_logger, "%s: PACKET_IO->tap_allow_mutilthread : %d", LOG_TAG_PKTIO, config->tap_allow_mutilthread);
TFE_LOG_DEBUG(g_default_logger, "%s: PACKET_IO->bpf_obj : %s", LOG_TAG_PKTIO, config->bpf_obj);
TFE_LOG_DEBUG(g_default_logger, "%s: PACKET_IO->bpf_debug_log : %d", LOG_TAG_PKTIO, config->bpf_debug_log);
TFE_LOG_DEBUG(g_default_logger, "%s: PACKET_IO->bpf_hash_mode : %d", LOG_TAG_PKTIO, config->bpf_hash_mode);
TFE_LOG_DEBUG(g_default_logger, "%s: PACKET_IO->tap_rps_enable : %d", LOG_TAG_PKTIO, config->tap_rps_enable);
TFE_LOG_DEBUG(g_default_logger, "%s: PACKET_IO->tap_rps_mask : %s", LOG_TAG_PKTIO, config->tap_rps_mask);
TFE_LOG_DEBUG(g_default_logger, "%s: PACKET_IO->enable_iouring : %d", LOG_TAG_PKTIO, config->enable_iouring);
TFE_LOG_DEBUG(g_default_logger, "%s: PACKET_IO->enable_debuglog : %d", LOG_TAG_PKTIO, config->enable_debuglog);
TFE_LOG_DEBUG(g_default_logger, "%s: PACKET_IO->ring_size : %d", LOG_TAG_PKTIO, config->ring_size);
TFE_LOG_DEBUG(g_default_logger, "%s: PACKET_IO->buff_size : %d", LOG_TAG_PKTIO, config->buff_size);
TFE_LOG_DEBUG(g_default_logger, "%s: PACKET_IO->flags : %d", LOG_TAG_PKTIO, config->flags);
TFE_LOG_DEBUG(g_default_logger, "%s: PACKET_IO->sq_thread_idle : %d", LOG_TAG_PKTIO, config->sq_thread_idle);
TFE_LOG_DEBUG(g_default_logger, "%s: PACKET_IO->device_client : %s", LOG_TAG_PKTIO, config->dev_tap_c);
TFE_LOG_DEBUG(g_default_logger, "%s: PACKET_IO->device_server : %s", LOG_TAG_PKTIO, config->dev_tap_s);
TFE_LOG_DEBUG(g_default_logger, "%s: PACKET_IO->src_mac_addr : %s", LOG_TAG_PKTIO, src_mac_addr);
TFE_LOG_DEBUG(logger, "%s: PACKET_IO->bypass_all_traffic : %d", LOG_TAG_PKTIO, config->bypass_all_traffic);
TFE_LOG_DEBUG(logger, "%s: PACKET_IO->rx_burst_max : %d", LOG_TAG_PKTIO, config->rx_burst_max);
TFE_LOG_DEBUG(logger, "%s: PACKET_IO->app_symbol : %s", LOG_TAG_PKTIO, config->app_symbol);
TFE_LOG_DEBUG(logger, "%s: PACKET_IO->dev_nf_interface : %s", LOG_TAG_PKTIO, config->dev_nf_interface);
TFE_LOG_DEBUG(logger, "%s: PACKET_IO->tap_name : %s", LOG_TAG_PKTIO, config->tap_rps_mask);
TFE_LOG_DEBUG(logger, "%s: PACKET_IO->tap_allow_mutilthread : %d", LOG_TAG_PKTIO, config->tap_allow_mutilthread);
TFE_LOG_DEBUG(logger, "%s: PACKET_IO->bpf_obj : %s", LOG_TAG_PKTIO, config->bpf_obj);
TFE_LOG_DEBUG(logger, "%s: PACKET_IO->bpf_debug_log : %d", LOG_TAG_PKTIO, config->bpf_debug_log);
TFE_LOG_DEBUG(logger, "%s: PACKET_IO->bpf_hash_mode : %d", LOG_TAG_PKTIO, config->bpf_hash_mode);
TFE_LOG_DEBUG(logger, "%s: PACKET_IO->tap_rps_enable : %d", LOG_TAG_PKTIO, config->tap_rps_enable);
TFE_LOG_DEBUG(logger, "%s: PACKET_IO->tap_rps_mask : %s", LOG_TAG_PKTIO, config->tap_rps_mask);
TFE_LOG_DEBUG(logger, "%s: PACKET_IO->enable_iouring : %d", LOG_TAG_PKTIO, config->enable_iouring);
TFE_LOG_DEBUG(logger, "%s: PACKET_IO->enable_debuglog : %d", LOG_TAG_PKTIO, config->enable_debuglog);
TFE_LOG_DEBUG(logger, "%s: PACKET_IO->ring_size : %d", LOG_TAG_PKTIO, config->ring_size);
TFE_LOG_DEBUG(logger, "%s: PACKET_IO->buff_size : %d", LOG_TAG_PKTIO, config->buff_size);
TFE_LOG_DEBUG(logger, "%s: PACKET_IO->flags : %d", LOG_TAG_PKTIO, config->flags);
TFE_LOG_DEBUG(logger, "%s: PACKET_IO->sq_thread_idle : %d", LOG_TAG_PKTIO, config->sq_thread_idle);
TFE_LOG_DEBUG(logger, "%s: PACKET_IO->device_client : %s", LOG_TAG_PKTIO, config->dev_tap_c);
TFE_LOG_DEBUG(logger, "%s: PACKET_IO->device_server : %s", LOG_TAG_PKTIO, config->dev_tap_s);
TFE_LOG_DEBUG(logger, "%s: PACKET_IO->src_mac_addr : %s", LOG_TAG_PKTIO, src_mac_addr);
return 0;
}
// return 0 : success
// return -1 : error
static int packet_io_get_metadata(marsio_buff_t *rx_buff, struct metadata *meta)
static int packet_io_get_metadata(marsio_buff_t *rx_buff, struct metadata *meta, void *logger)
{
memset(meta, 0, sizeof(struct metadata));
if (marsio_buff_get_metadata(rx_buff, MR_BUFF_SESSION_ID, &(meta->session_id), sizeof(meta->session_id)) <= 0)
{
TFE_LOG_ERROR(g_default_logger, "%s: unable to get session_id from metadata", LOG_TAG_PKTIO);
TFE_LOG_ERROR(logger, "%s: unable to get session_id from metadata", LOG_TAG_PKTIO);
return -1;
}
@@ -808,7 +808,7 @@ static int packet_io_get_metadata(marsio_buff_t *rx_buff, struct metadata *meta)
meta->raw_data = marsio_buff_mtod(rx_buff);
if (meta->raw_data == NULL || meta->raw_len == 0)
{
TFE_LOG_ERROR(g_default_logger, "%s: unable to get raw_data from metadata", LOG_TAG_PKTIO);
TFE_LOG_ERROR(logger, "%s: unable to get raw_data from metadata", LOG_TAG_PKTIO);
return -1;
}
@@ -816,7 +816,7 @@ static int packet_io_get_metadata(marsio_buff_t *rx_buff, struct metadata *meta)
// 0: I2E
if (marsio_buff_get_metadata(rx_buff, MR_BUFF_DIR, &(meta->is_e2i_dir), sizeof(meta->is_e2i_dir)) <= 0)
{
TFE_LOG_ERROR(g_default_logger, "%s: unable to get buff_dir from metadata", LOG_TAG_PKTIO);
TFE_LOG_ERROR(logger, "%s: unable to get buff_dir from metadata", LOG_TAG_PKTIO);
return -1;
}
@@ -826,7 +826,7 @@ static int packet_io_get_metadata(marsio_buff_t *rx_buff, struct metadata *meta)
// only control packet set MR_BUFF_PAYLOAD_OFFSET
if (marsio_buff_get_metadata(rx_buff, MR_BUFF_PAYLOAD_OFFSET, &(meta->l7offset), sizeof(meta->l7offset)) <= 0)
{
TFE_LOG_ERROR(g_default_logger, "%s: unable to get l7offset from metadata", LOG_TAG_PKTIO);
TFE_LOG_ERROR(logger, "%s: unable to get l7offset from metadata", LOG_TAG_PKTIO);
return -1;
}
}
@@ -836,7 +836,7 @@ static int packet_io_get_metadata(marsio_buff_t *rx_buff, struct metadata *meta)
uint16_t user_data = 0;
if (marsio_buff_get_metadata(rx_buff, MR_BUFF_USER_0, &(user_data), sizeof(user_data)) <= 0)
{
TFE_LOG_ERROR(g_default_logger, "%s: unable to get is_decrypted from metadata", LOG_TAG_PKTIO);
TFE_LOG_ERROR(logger, "%s: unable to get is_decrypted from metadata", LOG_TAG_PKTIO);
return -1;
}
if (user_data & TRAFFIC_IS_DECRYPTED)
@@ -852,14 +852,14 @@ static int packet_io_get_metadata(marsio_buff_t *rx_buff, struct metadata *meta)
meta->route_ctx.len = marsio_buff_get_metadata(rx_buff, MR_BUFF_ROUTE_CTX, meta->route_ctx.data, sizeof(meta->route_ctx.data));
if (meta->route_ctx.len <= 0)
{
TFE_LOG_ERROR(g_default_logger, "%s: unable to get route_ctx from metadata", LOG_TAG_PKTIO);
TFE_LOG_ERROR(logger, "%s: unable to get route_ctx from metadata", LOG_TAG_PKTIO);
return -1;
}
meta->sids.num = marsio_buff_get_sid_list(rx_buff, meta->sids.elems, sizeof(meta->sids.elems) / sizeof(meta->sids.elems[0]));
if (meta->sids.num < 0)
{
TFE_LOG_ERROR(g_default_logger, "%s: unable to get sid_list from metadata", LOG_TAG_PKTIO);
TFE_LOG_ERROR(logger, "%s: unable to get sid_list from metadata", LOG_TAG_PKTIO);
return -1;
}
@@ -868,13 +868,13 @@ static int packet_io_get_metadata(marsio_buff_t *rx_buff, struct metadata *meta)
// return 0 : success
// return -1 : error
static int packet_io_set_metadata(marsio_buff_t *tx_buff, struct metadata *meta)
static int packet_io_set_metadata(marsio_buff_t *tx_buff, struct metadata *meta, void *logger)
{
if (meta->session_id)
{
if (marsio_buff_set_metadata(tx_buff, MR_BUFF_SESSION_ID, &(meta->session_id), sizeof(meta->session_id)) != 0)
{
TFE_LOG_ERROR(g_default_logger, "%s: unable to set session_id for metadata", LOG_TAG_PKTIO);
TFE_LOG_ERROR(logger, "%s: unable to set session_id for metadata", LOG_TAG_PKTIO);
return -1;
}
}
@@ -884,7 +884,7 @@ static int packet_io_set_metadata(marsio_buff_t *tx_buff, struct metadata *meta)
marsio_buff_set_ctrlbuf(tx_buff);
if (marsio_buff_set_metadata(tx_buff, MR_BUFF_PAYLOAD_OFFSET, &(meta->l7offset), sizeof(meta->l7offset)) != 0)
{
TFE_LOG_ERROR(g_default_logger, "%s: unable to set l7offset for metadata", LOG_TAG_PKTIO);
TFE_LOG_ERROR(logger, "%s: unable to set l7offset for metadata", LOG_TAG_PKTIO);
return -1;
}
}
@@ -897,7 +897,7 @@ static int packet_io_set_metadata(marsio_buff_t *tx_buff, struct metadata *meta)
}
if (marsio_buff_set_metadata(tx_buff, MR_BUFF_USER_0, &(user_data), sizeof(user_data)) != 0)
{
TFE_LOG_ERROR(g_default_logger, "%s: unable to set is_decrypted for metadata", LOG_TAG_PKTIO);
TFE_LOG_ERROR(logger, "%s: unable to set is_decrypted for metadata", LOG_TAG_PKTIO);
return -1;
}
}
@@ -906,7 +906,7 @@ static int packet_io_set_metadata(marsio_buff_t *tx_buff, struct metadata *meta)
{
if (marsio_buff_set_metadata(tx_buff, MR_BUFF_ROUTE_CTX, meta->route_ctx.data, meta->route_ctx.len) != 0)
{
TFE_LOG_ERROR(g_default_logger, "%s: unable to set route_ctx for metadata", LOG_TAG_PKTIO);
TFE_LOG_ERROR(logger, "%s: unable to set route_ctx for metadata", LOG_TAG_PKTIO);
return -1;
}
}
@@ -915,7 +915,7 @@ static int packet_io_set_metadata(marsio_buff_t *tx_buff, struct metadata *meta)
{
if (marsio_buff_set_sid_list(tx_buff, meta->sids.elems, meta->sids.num) != 0)
{
TFE_LOG_ERROR(g_default_logger, "%s: unable to set sid_list for metadata", LOG_TAG_PKTIO);
TFE_LOG_ERROR(logger, "%s: unable to set sid_list for metadata", LOG_TAG_PKTIO);
return -1;
}
}
@@ -923,9 +923,9 @@ static int packet_io_set_metadata(marsio_buff_t *tx_buff, struct metadata *meta)
return 0;
}
static void packet_io_dump_metadata(marsio_buff_t *tx_buff, struct metadata *meta)
static void packet_io_dump_metadata(marsio_buff_t *tx_buff, struct metadata *meta, void *logger)
{
TFE_LOG_DEBUG(g_default_logger, "%s: META={session_id: %lu, raw_len: %d, is_e2i_dir: %d, is_ctrl_pkt: %d, l7offset: %d, is_decrypted: %u, sids_num: %d}", LOG_TAG_PKTIO, meta->session_id, meta->raw_len, meta->is_e2i_dir, meta->is_ctrl_pkt, meta->l7offset, meta->is_decrypted, meta->sids.num);
TFE_LOG_DEBUG(logger, "%s: META={session_id: %lu, raw_len: %d, is_e2i_dir: %d, is_ctrl_pkt: %d, l7offset: %d, is_decrypted: %u, sids_num: %d}", LOG_TAG_PKTIO, meta->session_id, meta->raw_len, meta->is_e2i_dir, meta->is_ctrl_pkt, meta->l7offset, meta->is_decrypted, meta->sids.num);
}
/*
@@ -948,9 +948,10 @@ static void send_event_log(struct session_ctx *s_ctx, int thread_seq, void *ctx)
struct packet_io_thread_ctx *thread = (struct packet_io_thread_ctx *)ctx;
struct acceptor_kni_v4 *acceptor_ctx = thread->ref_acceptor_ctx;
struct packet_io *packet_io = thread->ref_io;
struct packet_io_fs *packet_io_fs = thread->ret_fs_state;
marsio_buff_t *tx_buffs[1];
struct metadata meta = {0};
void * logger = thread->logger;
uint16_t length = 0;
uint8_t ssl_intercept_status = 0;
@@ -1059,8 +1060,8 @@ static void send_event_log(struct session_ctx *s_ctx, int thread_seq, void *ctx)
meta.sids.num = 1;
meta.sids.elems[0] = acceptor_ctx->firewall_sids;
route_ctx_copy(&meta.route_ctx, &s_ctx->ctrl_meta->route_ctx);
packet_io_set_metadata(tx_buffs[0], &meta);
int nsend = marsio_buff_datalen(tx_buffs[0]);
packet_io_set_metadata(tx_buffs[0], &meta, logger);
ATOMIC_INC(&(packet_io_fs->session_log));
marsio_send_burst(packet_io->dev_nf_interface.mr_path, thread_seq, tx_buffs, 1);
if (data)
@@ -1068,21 +1069,65 @@ static void send_event_log(struct session_ctx *s_ctx, int thread_seq, void *ctx)
return;
}
static void tcp_restore_info_dump(struct tcp_restore_info *info, uint64_t session_id, void *logger)
{
char str_client_addr[64] = { 0 };
char str_server_addr[64] = { 0 };
const struct tcp_restore_endpoint *client = &info->client;
const struct tcp_restore_endpoint *server = &info->server;
assert(client->addr.ss_family == server->addr.ss_family);
if (client->addr.ss_family == AF_INET)
{
struct sockaddr_in *sk_client = (struct sockaddr_in *)&client->addr;
struct sockaddr_in *sk_server = (struct sockaddr_in *)&server->addr;
uint16_t port_client = ntohs(sk_client->sin_port);
uint16_t port_server = ntohs(sk_server->sin_port);
inet_ntop(AF_INET, &sk_client->sin_addr, str_client_addr, sizeof(str_client_addr));
inet_ntop(AF_INET, &sk_server->sin_addr, str_server_addr, sizeof(str_client_addr));
TFE_LOG_DEBUG(logger, "restore_info session %lu %s:%hu %s:%hu: cur_dir=%u\n"
"\tclient={ addr=%s, port=%hu, seq:%u, ack:%u, ts_val:%u, mss=%u, window:%hu, wscale_perm=%u, wscale=%u, timestamp_perm=%u, sack_perm=%u }\n"
"\tserver={ addr=%s, port=%hu, seq:%u, ack:%u, ts_val:%u, mss=%u, window:%hu, wscale_perm=%u, wscale=%u, timestamp_perm=%u, sack_perm=%u }",
session_id, str_client_addr, port_client, str_server_addr, port_server, info->cur_dir,
str_client_addr, port_client, client->seq, client->ack, client->ts_val, client->mss, client->window, (client->wscale_perm ? 1 : 0), client->wscale, (client->timestamp_perm ? 1 : 0), (client->sack_perm ? 1 : 0),
str_server_addr, port_server, server->seq, server->ack, server->ts_val, server->mss, server->window, (server->wscale_perm ? 1 : 0), server->wscale, (server->timestamp_perm ? 1 : 0), (server->sack_perm ? 1 : 0));
}
else if (client->addr.ss_family == AF_INET6)
{
struct sockaddr_in6 *sk_client = (struct sockaddr_in6 *)&client->addr;
struct sockaddr_in6 *sk_server = (struct sockaddr_in6 *)&server->addr;
uint16_t port_client = ntohs(sk_client->sin6_port);
uint16_t port_server = ntohs(sk_server->sin6_port);
inet_ntop(AF_INET6, &sk_client->sin6_addr, str_client_addr, sizeof(str_client_addr));
inet_ntop(AF_INET6, &sk_server->sin6_addr, str_server_addr, sizeof(str_client_addr));
TFE_LOG_DEBUG(logger, "restore_info session %lu %s:%hu %s:%hu: tcp_restore_info %p cur_dir=%u\n"
"\tclient={ addr=%s, port=%hu, seq:%u, ack:%u, ts_val:%u, mss=%u, window:%hu, wscale_perm=%u, wscale=%u, timestamp_perm=%u, sack_perm=%u }\n"
"\tserver={ addr=%s, port=%hu, seq:%u, ack:%u, ts_val:%u, mss=%u, window:%hu, wscale_perm=%u, wscale=%u, timestamp_perm=%u, sack_perm=%u }",
session_id, str_client_addr, port_client, str_server_addr, port_server, info, info->cur_dir,
str_client_addr, port_client, client->seq, client->ack, client->ts_val, client->mss, client->window, (client->wscale_perm ? 1 : 0), client->wscale, (client->timestamp_perm ? 1 : 0), (client->sack_perm ? 1 : 0),
str_server_addr, port_server, server->seq, server->ack, server->ts_val, server->mss, server->window, (server->wscale_perm ? 1 : 0), server->wscale, (server->timestamp_perm ? 1 : 0), (server->sack_perm ? 1 : 0));
}
}
// return 0 : success
// return -1 : error
static int handle_session_opening(struct metadata *meta, struct ctrl_pkt_parser *parser, int thread_seq, void *ctx)
{
uint8_t *iptmp = NULL;
int fd_downstream = 0;
int fd_upstream = 0;
int fd_fake_c = 0;
int fd_fake_s = 0;
uint16_t size = 0;
char *addr_str = NULL;
unsigned int stream_common_direction;
uint8_t stream_protocol_in_char = 0;
uint8_t enalbe_decrypted_traffic_steering = 0;
uint8_t enable_decrypted_traffic_steering = 0;
struct ethhdr *ether_hdr = NULL;
struct session_ctx *s_ctx = NULL;
struct addr_tuple4 inner_tuple4;
@@ -1090,22 +1135,21 @@ static int handle_session_opening(struct metadata *meta, struct ctrl_pkt_parser
memset(&inner_tuple4, 0, sizeof(inner_tuple4));
memset(&restore_info, 0, sizeof(restore_info));
struct sockaddr_in *in_addr_client = (struct sockaddr_in *)&restore_info.client.addr;
struct sockaddr_in *in_addr_server = (struct sockaddr_in *)&restore_info.server.addr;
struct packet_io_thread_ctx *thread = (struct packet_io_thread_ctx *)ctx;
struct packet_io *packet_io = thread->ref_io;
struct packet_io_fs *packet_io_fs = thread->ret_fs_state;
void * logger = thread->logger;
struct raw_pkt_parser raw_parser;
raw_packet_parser_init(&raw_parser, meta->session_id, LAYER_TYPE_ALL, 8);
const void *payload = raw_packet_parser_parse(&raw_parser, (const void *)meta->raw_data, meta->raw_len);
const void *payload = raw_packet_parser_parse(&raw_parser, (const void *)meta->raw_data, meta->raw_len, logger);
if ((char *)payload - meta->raw_data != meta->l7offset)
{
uint16_t offset = (char *)payload - meta->raw_data;
TFE_LOG_ERROR(g_default_logger, "%s: incorrect dataoffset in the control zone of session %lu, offset:%u, l7offset:%u, payload:%p, raw_data:%p", LOG_TAG_PKTIO, meta->session_id, offset, meta->l7offset, payload, meta->raw_data);
TFE_LOG_ERROR(logger, "%s: incorrect dataoffset in the control zone of session %lu, offset:%u, l7offset:%u, payload:%p, raw_data:%p", LOG_TAG_PKTIO, meta->session_id, offset, meta->l7offset, payload, meta->raw_data);
}
raw_packet_parser_get_most_inner_tuple4(&raw_parser, &inner_tuple4);
raw_packet_parser_get_most_inner_tuple4(&raw_parser, &inner_tuple4, logger);
intercept_policy_enforce(thread->ref_proxy->int_ply_enforcer, parser->cmsg);
tcp_policy_enforce(thread->ref_proxy->tcp_ply_enforcer, parser->cmsg);
@@ -1116,51 +1160,17 @@ static int handle_session_opening(struct metadata *meta, struct ctrl_pkt_parser
tcp_restore_set_from_cmsg(parser->cmsg, &restore_info);
tcp_restore_set_from_pkg(&inner_tuple4, &restore_info);
if (overwrite_tcp_mss(parser->cmsg, &restore_info))
if (overwrite_tcp_mss(parser->cmsg, &restore_info, meta->session_id, logger))
{
goto end;
}
tcp_restore_info_dump(&restore_info, meta->session_id, logger);
iptmp = (uint8_t *)&in_addr_client->sin_addr.s_addr;
// tcp repair C2S
addr_str = addr_tuple4_to_str(&inner_tuple4);
TFE_LOG_DEBUG(g_default_logger, "restore_info session %lu %s : client", meta->session_id, addr_str);
TFE_LOG_DEBUG(g_default_logger, "\t addr:%d.%d.%d.%d", iptmp[0], iptmp[1], iptmp[2], iptmp[3]);
TFE_LOG_DEBUG(g_default_logger, "\t port:%u", in_addr_client->sin_port);
TFE_LOG_DEBUG(g_default_logger, "\t seq:%u", restore_info.client.seq);
TFE_LOG_DEBUG(g_default_logger, "\t ack:%u", restore_info.client.ack);
TFE_LOG_DEBUG(g_default_logger, "\t ts_val:%u", restore_info.client.ts_val);
TFE_LOG_DEBUG(g_default_logger, "\t mss:%u", restore_info.client.mss);
TFE_LOG_DEBUG(g_default_logger, "\t window:%u", restore_info.client.window);
TFE_LOG_DEBUG(g_default_logger, "\t wscale:%u", restore_info.client.wscale);
TFE_LOG_DEBUG(g_default_logger, "\t wscale_perm:%s", restore_info.client.wscale_perm > 0?"true":"false");
TFE_LOG_DEBUG(g_default_logger, "\t timestamp_perm:%s", restore_info.client.timestamp_perm > 0?"true":"false");
TFE_LOG_DEBUG(g_default_logger, "\t sack_perm:%s", restore_info.client.sack_perm > 0?"true":"false");
iptmp = (uint8_t *)&in_addr_server->sin_addr.s_addr;
// tcp repair C2S
TFE_LOG_DEBUG(g_default_logger, "restore_info session %lu %s : server", meta->session_id, addr_str);
TFE_LOG_DEBUG(g_default_logger, "\t addr:%d.%d.%d.%d", iptmp[0], iptmp[1], iptmp[2], iptmp[3]);
TFE_LOG_DEBUG(g_default_logger, "\t port:%u", in_addr_server->sin_port);
TFE_LOG_DEBUG(g_default_logger, "\t seq:%u", restore_info.server.seq);
TFE_LOG_DEBUG(g_default_logger, "\t ack:%u", restore_info.server.ack);
TFE_LOG_DEBUG(g_default_logger, "\t ts_val:%u", restore_info.server.ts_val);
TFE_LOG_DEBUG(g_default_logger, "\t mss:%u", restore_info.server.mss);
TFE_LOG_DEBUG(g_default_logger, "\t window:%u", restore_info.server.window);
TFE_LOG_DEBUG(g_default_logger, "\t wscale:%u", restore_info.server.wscale);
TFE_LOG_DEBUG(g_default_logger, "\t wscale_perm:%s", restore_info.server.wscale_perm > 0?"true":"false");
TFE_LOG_DEBUG(g_default_logger, "\t timestamp_perm:%s", restore_info.server.timestamp_perm > 0?"true":"false");
TFE_LOG_DEBUG(g_default_logger, "\t sack_perm:%s", restore_info.server.sack_perm > 0?"true":"false");
free(addr_str);
fd_upstream = tfe_tcp_restore_fd_create(&(restore_info.client), &(restore_info.server), packet_io->config.dev_tap, 0x65);
if (fd_upstream < 0)
{
TFE_LOG_ERROR(g_default_logger, "Failed at tcp_restore_fd_create(UPSTREAM)");
TFE_LOG_ERROR(logger, "%s: session %lu Failed at tcp_restore_fd_create(UPSTREAM)", LOG_TAG_PKTIO, meta->session_id);
goto end;
}
@@ -1168,40 +1178,40 @@ static int handle_session_opening(struct metadata *meta, struct ctrl_pkt_parser
fd_downstream = tfe_tcp_restore_fd_create(&(restore_info.server), &(restore_info.client), packet_io->config.dev_tap, 0x65);
if (fd_downstream < 0)
{
TFE_LOG_ERROR(g_default_logger, "Failed at tcp_restore_fd_create(DOWNSTREAM)");
TFE_LOG_ERROR(logger, "%s: session %lu Failed at tcp_restore_fd_create(DOWNSTREAM)", LOG_TAG_PKTIO, meta->session_id);
goto end;
}
tfe_cmsg_get_value(parser->cmsg, TFE_CMSG_TCP_RESTORE_PROTOCOL, (unsigned char *)&stream_protocol_in_char, sizeof(stream_protocol_in_char), &size);
tfe_cmsg_get_value(parser->cmsg, TFE_CMSG_TCP_DECRYPTED_TRAFFIC_STEERING, (unsigned char *)&enalbe_decrypted_traffic_steering, sizeof(enalbe_decrypted_traffic_steering), &size);
tfe_cmsg_get_value(parser->cmsg, TFE_CMSG_TCP_DECRYPTED_TRAFFIC_STEERING, (unsigned char *)&enable_decrypted_traffic_steering, sizeof(enable_decrypted_traffic_steering), &size);
if ((STREAM_PROTO_PLAIN == (enum tfe_stream_proto)stream_protocol_in_char && thread->ref_proxy->traffic_steering_options.enable_steering_http) ||
(STREAM_PROTO_SSL == (enum tfe_stream_proto)stream_protocol_in_char && thread->ref_proxy->traffic_steering_options.enable_steering_ssl) ||
enalbe_decrypted_traffic_steering == 1)
enable_decrypted_traffic_steering == 1)
{
if (fake_tcp_handshake(thread->ref_proxy, &restore_info) == -1)
{
TFE_LOG_ERROR(g_default_logger, "session %lu Failed at fake_tcp_handshake()", meta->session_id);
TFE_LOG_ERROR(logger, "%s: session %lu Failed at fake_tcp_handshake()", LOG_TAG_PKTIO, meta->session_id);
goto end;
}
fd_fake_c = tfe_tcp_restore_fd_create(&(restore_info.client), &(restore_info.server), thread->ref_proxy->traffic_steering_options.device_client, thread->ref_proxy->traffic_steering_options.so_mask_client);
if (fd_fake_c < 0)
{
TFE_LOG_ERROR(g_default_logger, "session %lu Failed at tcp_restore_fd_create(fd_fake_c)", meta->session_id);
TFE_LOG_ERROR(logger, "%s: session %lu Failed at tcp_restore_fd_create(fd_fake_c)", LOG_TAG_PKTIO, meta->session_id);
goto end;
}
fd_fake_s = tfe_tcp_restore_fd_create(&(restore_info.server), &(restore_info.client), thread->ref_proxy->traffic_steering_options.device_server, thread->ref_proxy->traffic_steering_options.so_mask_server);
if (fd_fake_s < 0)
{
TFE_LOG_ERROR(g_default_logger, "session %lu Failed at tcp_restore_fd_create(fd_fake_s)", meta->session_id);
TFE_LOG_ERROR(logger, "%s: session %lu Failed at tcp_restore_fd_create(fd_fake_s)", LOG_TAG_PKTIO, meta->session_id);
goto end;
}
}
if (tfe_proxy_fds_accept(thread->ref_proxy, fd_downstream, fd_upstream, fd_fake_c, fd_fake_s, parser->cmsg) < 0)
{
TFE_LOG_ERROR(g_default_logger, "session %lu Failed at tfe_proxy_fds_accept()", meta->session_id);
TFE_LOG_ERROR(logger, "%s: session %lu Failed at tfe_proxy_fds_accept()", LOG_TAG_PKTIO, meta->session_id);
goto end;
}
@@ -1254,10 +1264,10 @@ static int handle_session_opening(struct metadata *meta, struct ctrl_pkt_parser
route_ctx_copy(&s_ctx->raw_meta_e2i->route_ctx, &parser->ack_route_ctx);
}
TFE_LOG_INFO(g_default_logger, "%s: session %lu %s active first", LOG_TAG_PKTIO, s_ctx->session_id, s_ctx->session_addr);
TFE_LOG_INFO(logger, "%s: session %lu %s active first", LOG_TAG_PKTIO, s_ctx->session_id, s_ctx->session_addr);
session_table_insert(thread->session_table, s_ctx->session_id, &(s_ctx->c2s_info.tuple4), s_ctx, session_value_free_cb);
ATOMIC_INC(&(packet_io_fs->session_num));
return 0;
end:
if (parser->cmsg)
@@ -1285,13 +1295,16 @@ static int handle_session_active(struct metadata *meta, struct ctrl_pkt_parser *
static int handle_session_closing(struct metadata *meta, struct ctrl_pkt_parser *parser, int thread_seq, void *ctx)
{
struct packet_io_thread_ctx *thread = (struct packet_io_thread_ctx *)ctx;
struct packet_io_fs *packet_io_fs = thread->ret_fs_state;
void * logger = thread->logger;
struct session_node *node = session_table_search_by_id(thread->session_table, meta->session_id);
if (node)
{
struct session_ctx *s_ctx = (struct session_ctx *)node->val_data;
TFE_LOG_INFO(g_default_logger, "%s: session %lu closing", LOG_TAG_PKTIO, s_ctx->session_id);
TFE_LOG_INFO(logger, "%s: session %lu closing", LOG_TAG_PKTIO, s_ctx->session_id);
session_table_delete_by_id(thread->session_table, meta->session_id);
ATOMIC_DEC(&(packet_io_fs->session_num));
return 0;
}
@@ -1304,9 +1317,11 @@ static int handle_session_resetall(struct metadata *meta, struct ctrl_pkt_parser
{
struct packet_io_thread_ctx *thread = (struct packet_io_thread_ctx *)ctx;
struct acceptor_kni_v4 *acceptor_ctx = thread->ref_acceptor_ctx;
struct packet_io_fs *packet_io_fs = thread->ret_fs_state;
void * logger = thread->logger;
TFE_LOG_ERROR(g_default_logger, "%s: session %lu resetall: notification clears all session tables !!!", LOG_TAG_PKTIO, meta->session_id);
TFE_LOG_ERROR(logger, "%s: session %lu resetall: notification clears all session tables !!!", LOG_TAG_PKTIO, meta->session_id);
ATOMIC_ZERO(&(packet_io_fs->session_num));
for (int i = 0; i < acceptor_ctx->nr_worker_threads; i++)
{
struct packet_io_thread_ctx *thread_ctx = &acceptor_ctx->work_threads[i];
@@ -1321,29 +1336,28 @@ static int handle_session_resetall(struct metadata *meta, struct ctrl_pkt_parser
static int handle_control_packet(struct packet_io *handle, marsio_buff_t *rx_buff, int thread_seq, void *ctx)
{
struct packet_io_thread_ctx *thread = (struct packet_io_thread_ctx *)ctx;
struct acceptor_kni_v4 *acceptor_ctx = thread->ref_acceptor_ctx;
struct packet_io_fs *packet_io_fs = thread->ret_fs_state;
void * logger = thread->logger;
struct metadata meta;
if (packet_io_get_metadata(rx_buff, &meta) == -1)
if (packet_io_get_metadata(rx_buff, &meta, logger) == -1)
{
TFE_LOG_ERROR(g_default_logger, "%s: unexpected control packet, unable to get metadata", LOG_TAG_PKTIO);
packet_io_dump_metadata(rx_buff, &meta);
TFE_LOG_ERROR(logger, "%s: unexpected control packet, unable to get metadata\n\tMETA={session_id: %lu, raw_len: %d, is_e2i_dir: %d, is_ctrl_pkt: %d, l7offset: %d, is_decrypted: %u, sids_num: %d}",
LOG_TAG_PKTIO, meta.session_id, meta.raw_len, meta.is_e2i_dir, meta.is_ctrl_pkt, meta.l7offset, meta.is_decrypted, meta.sids.num);
return -1;
}
struct ctrl_pkt_parser ctrl_parser;
ctrl_packet_parser_init(&ctrl_parser);
if (ctrl_packet_parser_parse(&ctrl_parser, meta.raw_data + meta.l7offset, meta.raw_len - meta.l7offset) == -1)
if (ctrl_packet_parser_parse(&ctrl_parser, meta.raw_data + meta.l7offset, meta.raw_len - meta.l7offset, logger) == -1)
{
TFE_LOG_ERROR(g_default_logger, "%s: unexpected control packet, unable to parse data", LOG_TAG_PKTIO);
TFE_LOG_ERROR(logger, "%s: unexpected control packet, metadata's session %lu unable to parse data", LOG_TAG_PKTIO, meta.session_id);
return -1;
}
ctrl_packet_parser_dump(&ctrl_parser);
if (ctrl_parser.session_id != meta.session_id)
{
TFE_LOG_ERROR(g_default_logger, "%s: unexpected control packet, metadata's session %lu != control packet's session %lu", LOG_TAG_PKTIO, meta.session_id, ctrl_parser.session_id);
TFE_LOG_ERROR(logger, "%s: unexpected control packet, metadata's session %lu != control packet's session %lu", LOG_TAG_PKTIO, meta.session_id, ctrl_parser.session_id);
return -1;
}
@@ -1376,17 +1390,18 @@ static int handle_raw_packet_from_nf(struct packet_io *handle, marsio_buff_t *rx
struct packet_io_thread_ctx *thread = (struct packet_io_thread_ctx *)ctx;
struct packet_io *packet_io = thread->ref_io;
struct packet_io_fs *packet_io_fs = thread->ret_fs_state;
void * logger = thread->logger;
int raw_len = marsio_buff_datalen(rx_buff);
char *raw_data = marsio_buff_mtod(rx_buff);
char *buff = NULL;
int buff_size = 0;
struct metadata meta;
if (packet_io_get_metadata(rx_buff, &meta) == -1)
if (packet_io_get_metadata(rx_buff, &meta, logger) == -1)
{
TFE_LOG_ERROR(g_default_logger, "%s: unexpected raw packet, unable to get metadata, bypass !!!", LOG_TAG_PKTIO);
packet_io_dump_metadata(rx_buff, &meta);
TFE_LOG_ERROR(logger, "%s: unexpected control packet, unable to get metadata\n\tMETA={session_id: %lu, raw_len: %d, is_e2i_dir: %d, is_ctrl_pkt: %d, l7offset: %d, is_decrypted: %u, sids_num: %d}",
LOG_TAG_PKTIO, meta.session_id, meta.raw_len, meta.is_e2i_dir, meta.is_ctrl_pkt, meta.l7offset, meta.is_decrypted, meta.sids.num);
throughput_metrics_inc(&packet_io_fs->raw_pkt_rx, 1, raw_len);
throughput_metrics_inc(&packet_io_fs->raw_bypass, 1, raw_len);
marsio_send_burst(handle->dev_nf_interface.mr_path, thread_seq, &rx_buff, 1);
return -1;
}
@@ -1394,7 +1409,8 @@ static int handle_raw_packet_from_nf(struct packet_io *handle, marsio_buff_t *rx
struct session_node *node = session_table_search_by_id(thread->session_table, meta.session_id);
if (node == NULL)
{
TFE_LOG_ERROR(g_default_logger, "%s: unexpected raw packet, unable to find session %lu from session table, bypass !!!", LOG_TAG_PKTIO, meta.session_id);
throughput_metrics_inc(&packet_io_fs->raw_pkt_rx, 1, raw_len);
throughput_metrics_inc(&packet_io_fs->raw_bypass, 1, raw_len);
marsio_send_burst(handle->dev_nf_interface.mr_path, thread_seq, &rx_buff, 1);
return -1;
}
@@ -1403,31 +1419,33 @@ static int handle_raw_packet_from_nf(struct packet_io *handle, marsio_buff_t *rx
if (meta.is_decrypted)
{
throughput_metrics_inc(&packet_io_fs->decrypt_rx, 1, raw_len);
// c2s
if (meta.is_e2i_dir == s_ctx->c2s_info.is_e2i_dir) {
add_ether_header(raw_data, packet_io->config.tap_c_mac, packet_io->config.tap_s_mac);
throughput_metrics_inc(&packet_io_fs->tap_s_pkt_tx, 1, raw_len);
if (packet_io->config.enable_iouring) {
io_uring_submit_write_entry(thread->tap_ctx->io_uring_s, raw_data, raw_len);
}
else {
tfe_tap_write_per_thread(thread->tap_ctx->tap_s, raw_data, raw_len, g_default_logger);
tap_write(thread->tap_ctx->tap_s, raw_data, raw_len, logger);
}
throughput_metrics_inc(&packet_io_fs->tap_s_pkt_tx, 1, raw_len);
}
// s2c
else {
add_ether_header(raw_data, packet_io->config.tap_s_mac, packet_io->config.tap_c_mac);
throughput_metrics_inc(&packet_io_fs->tap_c_pkt_tx, 1, raw_len);
if (packet_io->config.enable_iouring) {
io_uring_submit_write_entry(thread->tap_ctx->io_uring_c, raw_data, raw_len);
}
else {
tfe_tap_write_per_thread(thread->tap_ctx->tap_c, raw_data, raw_len, g_default_logger);
tap_write(thread->tap_ctx->tap_c, raw_data, raw_len, logger);
}
throughput_metrics_inc(&packet_io_fs->tap_c_pkt_tx, 1, raw_len);
}
}
else
{
throughput_metrics_inc(&packet_io_fs->raw_pkt_rx, 1, raw_len);
if (meta.is_e2i_dir)
{
if (metadata_is_empty(s_ctx->raw_meta_e2i))
@@ -1444,20 +1462,13 @@ static int handle_raw_packet_from_nf(struct packet_io *handle, marsio_buff_t *rx
}
s_ctx->raw_meta_i2e->sids = meta.sids;
}
#if 0
struct raw_pkt_parser raw_parser;
raw_packet_parser_init(&raw_parser, meta->session_id, LAYER_TYPE_ALL, 8);
const void *payload = raw_packet_parser_parse(&raw_parser, (const void *)meta->raw_data, meta->raw_len);
buff_size = raw_len - ((char *)payload - meta->raw_data) + sizeof(struct ethhdr) + sizeof(struct ip) + sizeof(struct tcphdr);
#endif
// send to tap0
add_ether_header(raw_data, packet_io->config.src_mac, packet_io->config.tap_mac);
if (packet_io->config.enable_iouring) {
io_uring_submit_write_entry(thread->tap_ctx->io_uring_fd, raw_data, raw_len);
}
else {
tfe_tap_write_per_thread(thread->tap_ctx->tap_fd, raw_data, raw_len, g_default_logger);
tap_write(thread->tap_ctx->tap_fd, raw_data, raw_len, logger);
}
throughput_metrics_inc(&packet_io_fs->tap_pkt_tx, 1, raw_len);
@@ -1503,8 +1514,7 @@ struct tap_ctx *tfe_tap_ctx_create(void *ctx)
{
int ret = 0;
struct packet_io_thread_ctx *thread_ctx = (struct packet_io_thread_ctx *)ctx;
struct acceptor_kni_v4 *acceptor_ctx = thread_ctx->ref_acceptor_ctx;
struct packet_io *packet_io = acceptor_ctx->io;
struct packet_io *packet_io = thread_ctx->ref_io;
struct tap_ctx *tap_ctx = (struct tap_ctx *)calloc(1, sizeof(struct tap_ctx));
assert(tap_ctx != NULL);
@@ -1560,11 +1570,11 @@ error_out:
return NULL;
}
int packet_io_thread_init(struct packet_io *handle, struct packet_io_thread_ctx *thread_ctx)
int packet_io_thread_init(struct packet_io *handle, struct packet_io_thread_ctx *thread_ctx, void *logger)
{
if (marsio_thread_init(handle->instance) != 0)
{
TFE_LOG_ERROR(g_default_logger, "%s: unable to init marsio thread %d", LOG_TAG_PKTIO, thread_ctx->thread_index);
TFE_LOG_ERROR(logger, "%s: unable to init marsio thread %d", LOG_TAG_PKTIO, thread_ctx->thread_index);
return -1;
}
@@ -1611,14 +1621,14 @@ void packet_io_destory(struct packet_io *handle)
}
}
struct packet_io *packet_io_create(const char *profile, int thread_num, cpu_set_t *coremask)
struct packet_io *packet_io_create(const char *profile, int thread_num, cpu_set_t *coremask, void *logger)
{
int opt = 1;
struct packet_io *handle = (struct packet_io *)calloc(1, sizeof(struct packet_io));
assert(handle != NULL);
handle->thread_num = thread_num;
if (packet_io_config(profile, &(handle->config)) != 0)
if (packet_io_config(profile, &(handle->config), logger) != 0)
{
goto error_out;
}
@@ -1632,32 +1642,32 @@ struct packet_io *packet_io_create(const char *profile, int thread_num, cpu_set_
}
}
else if (thread_num > 1){
TFE_LOG_ERROR(g_default_logger, "%s: under tap mode, when disable tap_allow_mutilthread, only support one work thread.", LOG_TAG_PKTIO);
TFE_LOG_ERROR(logger, "%s: under tap mode, when disable tap_allow_mutilthread, only support one work thread.", LOG_TAG_PKTIO);
goto error_out;
}
handle->instance = marsio_create();
if (handle->instance == NULL)
{
TFE_LOG_ERROR(g_default_logger, "%s: unable to create marsio instance", LOG_TAG_PKTIO);
TFE_LOG_ERROR(logger, "%s: unable to create marsio instance", LOG_TAG_PKTIO);
goto error_out;
}
if (marsio_option_set(handle->instance, MARSIO_OPT_THREAD_MASK_IN_CPUSET, coremask, sizeof(cpu_set_t)) != 0)
{
TFE_LOG_ERROR(g_default_logger, "%s: unable to set MARSIO_OPT_EXIT_WHEN_ERR option for marsio instance", LOG_TAG_PKTIO);
TFE_LOG_ERROR(logger, "%s: unable to set MARSIO_OPT_EXIT_WHEN_ERR option for marsio instance", LOG_TAG_PKTIO);
goto error_out;
}
if (marsio_option_set(handle->instance, MARSIO_OPT_EXIT_WHEN_ERR, &opt, sizeof(opt)) != 0)
{
TFE_LOG_ERROR(g_default_logger, "%s: unable to set MARSIO_OPT_EXIT_WHEN_ERR option for marsio instance", LOG_TAG_PKTIO);
TFE_LOG_ERROR(logger, "%s: unable to set MARSIO_OPT_EXIT_WHEN_ERR option for marsio instance", LOG_TAG_PKTIO);
goto error_out;
}
if (marsio_init(handle->instance, handle->config.app_symbol) != 0)
{
TFE_LOG_ERROR(g_default_logger, "%s: unable to initialize marsio instance", LOG_TAG_PKTIO);
TFE_LOG_ERROR(logger, "%s: unable to initialize marsio instance", LOG_TAG_PKTIO);
goto error_out;
}
@@ -1665,14 +1675,14 @@ struct packet_io *packet_io_create(const char *profile, int thread_num, cpu_set_
handle->dev_nf_interface.mr_dev = marsio_open_device(handle->instance, handle->config.dev_nf_interface, handle->thread_num, handle->thread_num);
if (handle->dev_nf_interface.mr_dev == NULL)
{
TFE_LOG_ERROR(g_default_logger, "%s: unable to open device %s", LOG_TAG_PKTIO, handle->config.dev_nf_interface);
TFE_LOG_ERROR(logger, "%s: unable to open device %s", LOG_TAG_PKTIO, handle->config.dev_nf_interface);
goto error_out;
}
handle->dev_nf_interface.mr_path = marsio_sendpath_create_by_vdev(handle->dev_nf_interface.mr_dev);
if (handle->dev_nf_interface.mr_path == NULL)
{
TFE_LOG_ERROR(g_default_logger, "%s: unable to create sendpath for device %s", LOG_TAG_PKTIO, handle->config.dev_nf_interface);
TFE_LOG_ERROR(logger, "%s: unable to create sendpath for device %s", LOG_TAG_PKTIO, handle->config.dev_nf_interface);
goto error_out;
}
@@ -1703,6 +1713,9 @@ int packet_io_polling_nf_interface(struct packet_io *handle, int thread_seq, voi
for (int i = 0; i < nr_recv; i++)
{
int raw_len = marsio_buff_datalen(rx_buffs[i]);
throughput_metrics_inc(&packet_io_fs->raw_pkt_rx, 1, raw_len);
throughput_metrics_inc(&packet_io_fs->raw_pkt_tx, 1, raw_len);
throughput_metrics_inc(&packet_io_fs->raw_bypass, 1, raw_len);
}
marsio_send_burst(handle->dev_nf_interface.mr_path, thread_seq, rx_buffs, nr_recv);
@@ -1716,20 +1729,29 @@ int packet_io_polling_nf_interface(struct packet_io *handle, int thread_seq, voi
if (is_downstream_keepalive_packet(rx_buff))
{
throughput_metrics_inc(&packet_io_fs->raw_pkt_rx, 1, raw_len);
throughput_metrics_inc(&packet_io_fs->raw_pkt_tx, 1, raw_len);
throughput_metrics_inc(&packet_io_fs->keepalived_pkt_rx, 1, raw_len);
throughput_metrics_inc(&packet_io_fs->keepalived_pkt_tx, 1, raw_len);
marsio_send_burst(handle->dev_nf_interface.mr_path, thread_seq, &rx_buff, 1);
continue;
}
if (marsio_buff_is_ctrlbuf(rx_buff))
{
throughput_metrics_inc(&packet_io_fs->raw_pkt_rx, 1, raw_len);
throughput_metrics_inc(&packet_io_fs->raw_pkt_tx, 1, raw_len);
throughput_metrics_inc(&packet_io_fs->ctrl_pkt_rx, 1, raw_len);
throughput_metrics_inc(&packet_io_fs->ctrl_pkt_tx, 1, raw_len);
// all control packet need bypass
handle_control_packet(handle, rx_buff, thread_seq, ctx);
marsio_send_burst(handle->dev_nf_interface.mr_path, thread_seq, &rx_buff, 1);
}
else
{
throughput_metrics_inc(&packet_io_fs->raw_pkt_rx, 1, raw_len);
handle_raw_packet_from_nf(handle, rx_buff, thread_seq, ctx);
}
}
@@ -1742,28 +1764,31 @@ void handle_decryption_packet_from_tap(const char *data, int len, void *args)
struct packet_io_thread_ctx *thread = (struct packet_io_thread_ctx *)args;
struct acceptor_kni_v4 *acceptor_ctx = thread->ref_acceptor_ctx;
struct packet_io *packet_io = thread->ref_io;
struct packet_io_fs *packet_io_fs = thread->ret_fs_state;
struct addr_tuple4 inner_addr;
struct raw_pkt_parser raw_parser;
void * logger = thread->logger;
memset(&inner_addr, 0, sizeof(struct addr_tuple4));
raw_packet_parser_init(&raw_parser, 0, LAYER_TYPE_ALL, 8);
raw_packet_parser_parse(&raw_parser, (const void *)data, len);
raw_packet_parser_get_most_inner_tuple4(&raw_parser, &inner_addr);
raw_packet_parser_parse(&raw_parser, (const void *)data, len, logger);
raw_packet_parser_get_most_inner_tuple4(&raw_parser, &inner_addr, logger);
throughput_metrics_inc(&packet_io_fs->decrypt_rx, 1, len);
struct session_node *node = session_table_search_by_addr(thread->session_table, &inner_addr);
if (node == NULL)
{
char *addr_string = addr_tuple4_to_str(&inner_addr);
TFE_LOG_ERROR(g_default_logger, "%s: unexpected inject packet, unable to find session %s from session table, drop !!!", LOG_TAG_PKTIO, addr_string);
free(addr_string);
if (node == NULL) {
throughput_metrics_inc(&packet_io_fs->decrypt_rxdrop, 1, len);
return;
}
struct session_ctx *s_ctx = (struct session_ctx *)node->val_data;
marsio_buff_t *tx_buffs[1];
int alloc_ret = marsio_buff_malloc_device(packet_io->dev_nf_interface.mr_dev, tx_buffs, 1, 0, thread->thread_index);
if (alloc_ret < 0){
TFE_LOG_ERROR(g_default_logger, "Failed at alloc marsio buffer, ret = %d, thread_seq = %d",
TFE_LOG_ERROR(logger, "Failed at alloc marsio buffer, ret = %d, thread_seq = %d",
alloc_ret, thread->thread_index);
throughput_metrics_inc(&packet_io_fs->decrypt_rxdrop, 1, len);
return;
}
@@ -1781,10 +1806,14 @@ void handle_decryption_packet_from_tap(const char *data, int len, void *args)
meta.sids.elems[0] = acceptor_ctx->sce_sids;
meta.sids.elems[1] = acceptor_ctx->proxy_sids;
if (memcmp(&inner_addr, &s_ctx->c2s_info.tuple4, sizeof(struct addr_tuple4)) == 0)
if (memcmp(&inner_addr, &s_ctx->c2s_info.tuple4, sizeof(struct addr_tuple4)) == 0) {
meta.is_e2i_dir = s_ctx->c2s_info.is_e2i_dir;
else
throughput_metrics_inc(&packet_io_fs->tap_c_pkt_rx, 1, len);
}
else {
meta.is_e2i_dir = s_ctx->s2c_info.is_e2i_dir;
throughput_metrics_inc(&packet_io_fs->tap_s_pkt_rx, 1, len);
}
if (meta.is_e2i_dir)
{
@@ -1794,7 +1823,8 @@ void handle_decryption_packet_from_tap(const char *data, int len, void *args)
{
route_ctx_copy(&meta.route_ctx, &s_ctx->raw_meta_i2e->route_ctx);
}
packet_io_set_metadata(tx_buffs[0], &meta);
packet_io_set_metadata(tx_buffs[0], &meta, logger);
throughput_metrics_inc(&packet_io_fs->decrypt_tx, 1, len);
marsio_send_burst(packet_io->dev_nf_interface.mr_path, thread->thread_index, tx_buffs, 1);
}
@@ -1803,30 +1833,32 @@ void handle_raw_packet_from_tap(const char *data, int len, void *args)
char *src_mac = NULL;
char *dst_mac = NULL;
struct packet_io_thread_ctx *thread = (struct packet_io_thread_ctx *)args;
struct acceptor_kni_v4 *acceptor_ctx = thread->ref_acceptor_ctx;
struct packet_io *packet_io = thread->ref_io;
struct packet_io_fs *packet_io_fs = thread->ret_fs_state;
struct addr_tuple4 inner_addr;
struct raw_pkt_parser raw_parser;
void * logger = thread->logger;
memset(&inner_addr, 0, sizeof(struct addr_tuple4));
raw_packet_parser_init(&raw_parser, 0, LAYER_TYPE_ALL, 8);
raw_packet_parser_parse(&raw_parser, (const void *)data, len);
raw_packet_parser_get_most_inner_tuple4(&raw_parser, &inner_addr);
raw_packet_parser_parse(&raw_parser, (const void *)data, len, logger);
raw_packet_parser_get_most_inner_tuple4(&raw_parser, &inner_addr, logger);
throughput_metrics_inc(&packet_io_fs->tap_pkt_rx, 1, len);
struct session_node *node = session_table_search_by_addr(thread->session_table, &inner_addr);
if (node == NULL)
{
char *addr_string = addr_tuple4_to_str(&inner_addr);
TFE_LOG_ERROR(g_default_logger, "%s: unexpected inject packet, unable to find session %s from session table, drop !!!", LOG_TAG_PKTIO, addr_string);
free(addr_string);
if (node == NULL) {
throughput_metrics_inc(&packet_io_fs->tap_pkt_rxdrop, 1, len);
return;
}
struct session_ctx *s_ctx = (struct session_ctx *)node->val_data;
marsio_buff_t *tx_buffs[1];
int alloc_ret = marsio_buff_malloc_device(packet_io->dev_nf_interface.mr_dev, tx_buffs, 1, 0, thread->thread_index);
if (alloc_ret < 0){
TFE_LOG_ERROR(g_default_logger, "Failed at alloc marsio buffer, ret = %d, thread_seq = %d",
TFE_LOG_ERROR(logger, "Failed at alloc marsio buffer, ret = %d, thread_seq = %d",
alloc_ret, thread->thread_index);
throughput_metrics_inc(&packet_io_fs->tap_pkt_rxdrop, 1, len);
return;
}
@@ -1864,9 +1896,9 @@ void handle_raw_packet_from_tap(const char *data, int len, void *args)
sids_copy(&meta.sids, &s_ctx->raw_meta_i2e->sids);
route_ctx_copy(&meta.route_ctx, &s_ctx->raw_meta_i2e->route_ctx);
}
packet_io_set_metadata(tx_buffs[0], &meta);
packet_io_set_metadata(tx_buffs[0], &meta, logger);
add_ether_header(dst, src_mac, dst_mac);
throughput_metrics_inc(&packet_io_fs->raw_pkt_tx, 1, len);
marsio_send_burst(packet_io->dev_nf_interface.mr_path, thread->thread_index, tx_buffs, 1);
}