From b931a3dc58dead4e02cda48813bdffd2e2d072c0 Mon Sep 17 00:00:00 2001 From: wangmenglan Date: Mon, 15 May 2023 16:41:59 +0800 Subject: packet io增加fieldstat状态统计 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- common/src/tfe_packet_io.cpp | 384 +++++++++++++++++++++++-------------------- 1 file changed, 208 insertions(+), 176 deletions(-) (limited to 'common/src/tfe_packet_io.cpp') diff --git a/common/src/tfe_packet_io.cpp b/common/src/tfe_packet_io.cpp index 766ea48..34bee10 100644 --- a/common/src/tfe_packet_io.cpp +++ b/common/src/tfe_packet_io.cpp @@ -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); } -- cgit v1.2.3