gfwleak/tango-tsg-service-chaining-engine
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

TSG-14649 tsg-service-chaining-engine适配新的控制报文格式

Browse Source
This commit is contained in:
luwenpeng
2023-04-28 15:00:46 +08:00
parent d1b04d50b8
commit a2d0f40f76
26 changed files with 16188 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files

115
platform/src/packet_io.cpp
View File

@@ -3,7 +3,7 @@
#include <netinet/udp.h>
#include <netinet/ether.h>
#include <cjson/cJSON.h>
#include "mpack.h"
#include <MESA/MESA_prof_load.h>
#include "log.h"
@@ -561,37 +561,81 @@ static int send_event_log(struct session_ctx *session_ctx, struct selected_chain
struct packet_io *packet_io = thread_ctx->ref_io;
int thread_index = thread_ctx->thread_index;
char buffer[32] = {0};
sprintf(buffer, "%lu", session_ctx->session_id);
cJSON *root = cJSON_CreateObject();
cJSON_AddStringToObject(root, "tsync", "1.0");
cJSON_AddStringToObject(root, "session_id", buffer);
cJSON_AddStringToObject(root, "state", "closing");
cJSON_AddStringToObject(root, "method", "log_update");
cJSON *sf_profile_ids = cJSON_CreateArray();
for (int i = 0; i < chaining->chaining_used; i++)
{
struct selected_sf *sf = &(chaining->chaining[i]);
if (sf->sf_need_skip == 0 && sf->sf_action == SESSION_ACTION_FORWARD)
{
cJSON *id = cJSON_CreateNumber(sf->sf_profile_id);
cJSON_AddItemToArray(sf_profile_ids, id);
}
}
cJSON *params = cJSON_CreateObject();
cJSON_AddItemToObject(params, "sf_profile_ids", sf_profile_ids);
cJSON_AddItemToObject(root, "params", params);
char *json_str = cJSON_PrintUnformatted(root);
char *data;
size_t size;
mpack_writer_t writer;
mpack_writer_init_growable(&writer, &data, &size);
LOG_INFO("%s: session %lu %s event log: %s", LOG_TAG_METRICS, session_ctx->session_id, session_ctx->session_addr, json_str);
// write the example on the msgpack homepage
mpack_build_map(&writer); // root begin
// tsync
mpack_write_cstr(&writer, "tsync");
mpack_write_cstr(&writer, "2.0");
// session_id
mpack_write_cstr(&writer, "session_id");
mpack_write_u64(&writer, session_ctx->session_id);
// state
// mpack_write_cstr(&writer, "state");
// mpack_write_cstr(&writer, "closed");
// method
mpack_write_cstr(&writer, "method");
mpack_write_cstr(&writer, "log_update");
// params
{
mpack_write_cstr(&writer, "params");
mpack_build_map(&writer); // params value begin
// sce
{
mpack_write_cstr(&writer, "sce");
mpack_build_map(&writer); // sce value begin
mpack_write_cstr(&writer, "sf_profile_ids");
mpack_build_array(&writer);
for (int i = 0; i < chaining->chaining_used; i++)
{
struct selected_sf *sf = &(chaining->chaining[i]);
if (sf->sf_need_skip == 0 && sf->sf_action == SESSION_ACTION_FORWARD)
{
mpack_write_u32(&writer, sf->sf_profile_id);
}
}
mpack_complete_array(&writer);
mpack_complete_map(&writer); // sce value end
}
mpack_complete_map(&writer); // params value end
}
mpack_complete_map(&writer); // root end
// finish writing
if (mpack_writer_destroy(&writer) != mpack_ok)
{
assert(0);
if (data)
{
free(data);
data = NULL;
}
return 0;
}
LOG_INFO("%s: session %lu %s send event log %ld bytes", LOG_TAG_METRICS, session_ctx->session_id, session_ctx->session_addr, size);
marsio_buff_t *tx_buffs[1];
char *raw_packet_header_data = session_ctx->ctrl_meta->raw_data;
int raw_packet_header_len = session_ctx->ctrl_meta->l7offset;
marsio_buff_malloc_global(packet_io->instance, tx_buffs, 1, 0, thread_index);
char *dst = marsio_buff_append(tx_buffs[0], raw_packet_header_len + strlen(json_str));
char *dst = marsio_buff_append(tx_buffs[0], raw_packet_header_len + size);
memcpy(dst, raw_packet_header_data, raw_packet_header_len);
memcpy(dst + raw_packet_header_len, json_str, strlen(json_str));
memcpy(dst + raw_packet_header_len, data, size);
struct metadata meta = {0};
meta.session_id = session_ctx->session_id;
@@ -603,8 +647,7 @@ static int send_event_log(struct session_ctx *session_ctx, struct selected_chain
mbuff_set_metadata(tx_buffs[0], &meta);
int nsend = marsio_buff_datalen(tx_buffs[0]);
marsio_send_burst(packet_io->dev_nf_interface.mr_path, thread_index, tx_buffs, 1);
free(json_str);
cJSON_Delete(root);
free(data);
return nsend;
}
@@ -712,9 +755,12 @@ static void handle_session_closing(struct metadata *meta, struct ctrl_pkt_parser
if (chaining_raw->chaining_used && sce_ctx->enable_send_log)
{
nsend = send_event_log(s_ctx, chaining_raw, thread_ctx);
ATOMIC_INC(&(g_metrics->sf_session.log));
throughput_metrics_inc(&(g_metrics->ctrl_pkt.tx), 1, nsend);
throughput_metrics_inc(&(g_metrics->device.nf_tx), 1, nsend);
if (nsend > 0)
{
ATOMIC_INC(&(g_metrics->sf_session.log));
throughput_metrics_inc(&(g_metrics->ctrl_pkt.tx), 1, nsend);
throughput_metrics_inc(&(g_metrics->device.nf_tx), 1, nsend);
}
}
struct selected_chaining *chaining_decrypted = s_ctx->chainings.chaining_decrypted;
@@ -722,9 +768,12 @@ static void handle_session_closing(struct metadata *meta, struct ctrl_pkt_parser
if (chaining_decrypted->chaining_used && sce_ctx->enable_send_log)
{
nsend = send_event_log(s_ctx, chaining_decrypted, thread_ctx);
ATOMIC_INC(&(g_metrics->sf_session.log));
throughput_metrics_inc(&(g_metrics->ctrl_pkt.tx), 1, nsend);
throughput_metrics_inc(&(g_metrics->device.nf_tx), 1, nsend);
if (nsend > 0)
{
ATOMIC_INC(&(g_metrics->sf_session.log));
throughput_metrics_inc(&(g_metrics->ctrl_pkt.tx), 1, nsend);
throughput_metrics_inc(&(g_metrics->device.nf_tx), 1, nsend);
}
}
session_table_delete_by_id(session_table, meta->session_id);