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

feature： TSG-21853 Refactoring TFE Kafka infrastructure

Browse Source
This commit is contained in:
luwenpeng
2024-07-19 18:20:04 +08:00
parent 88a7a8c5c4
commit 2045d517ca
25 changed files with 484 additions and 662 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
plugin/business/chaining-policy/src/chaining_policy.cpp
View File

@@ -144,7 +144,7 @@ struct chaining_policy_enforcer *chaining_policy_enforcer_create(void *logger)
{
int ret = 0;
struct chaining_policy_enforcer *enforcer = ALLOC(struct chaining_policy_enforcer, 1);
enforcer->maat = (struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT);
enforcer->maat = tfe_get_maat_handle();
enforcer->logger = logger;
enforcer->table_id = maat_get_table_id(enforcer->maat, "SERVICE_CHAINING_COMPILE");
if (enforcer->table_id < 0)

4
plugin/business/doh/src/doh.cpp
View File

@@ -385,7 +385,7 @@ static void doh_maat_scan(const struct tfe_stream *stream, const struct tfe_http
static int doh_maat_init(const char *profile, const char *section)
{
g_doh_conf->maat = (struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT);
g_doh_conf->maat = tfe_get_maat_handle();
MESA_load_profile_string_def(profile, section, "table_appid", g_doh_conf->tables[TYPE_APPID].name, TFE_STRING_MAX, "ATTR_APP_ID");
MESA_load_profile_string_def(profile, section, "table_qname", g_doh_conf->tables[TYPE_QNAME].name, TFE_STRING_MAX, "ATTR_DOH_QNAME");
MESA_load_profile_string_def(profile, section, "table_host", g_doh_conf->tables[TYPE_HOST].name, TFE_STRING_MAX, "ATTR_SERVER_FQDN");
@@ -824,7 +824,7 @@ int doh_on_data(const struct tfe_stream *stream, const struct tfe_http_session *
void doh_send_metric_log(const struct tfe_stream * stream, struct doh_ctx *ctx, unsigned int thread_id)
{
size_t c2s_byte_num = 0, s2c_byte_num =0;
struct tfe_fieldstat_metric_t *fieldstat = (struct tfe_fieldstat_metric_t *)tfe_bussiness_resouce_get(DYNAMIC_FIELDSTAT);
struct tfe_fieldstat_metric_t *fieldstat = tfe_get_fieldstat_handle();
fieldstat->tags[thread_id][TAG_VSYS_ID].value_int = ctx->result->vsys_id;
fieldstat->tags[thread_id][TAG_RULE_ID].value_int = ctx->result->config_id;

33
plugin/business/doh/src/logger.cpp
View File

@@ -1,4 +1,5 @@
#include "logger.h"
#include "kafka.h"
struct json_spec
{
@@ -287,14 +288,6 @@ int doh_kafka_init(const char *profile, struct doh_conf *conf)
{
return 0;
}
conf->device_id = (const char *)tfe_bussiness_resouce_get(DEVICE_ID);
conf->effective_device_tag = (const char *)tfe_bussiness_resouce_get(EFFECTIVE_DEVICE_TAG);
conf->kafka_logger = (tfe_kafka_logger_t *)tfe_bussiness_resouce_get(KAFKA_LOGGER);
if (conf->kafka_logger && !conf->kafka_logger->enable)
{
TFE_LOG_ERROR(conf->local_logger, "Doh sendlog ENABLE, but tfe kafka logger DISABLED.");
return -1;
}
return 0;
}
@@ -357,7 +350,6 @@ int doh_send_log(struct doh_conf *handle, const struct tfe_http_session *http, c
const char *tmp_val = NULL;
cJSON *common_obj = NULL, *per_hit_obj = NULL;
char *log_payload = NULL;
int kafka_status = 0;
int send_cnt = 0;
struct timeval cur_time;
char src_ip_str[MAX(INET6_ADDRSTRLEN, INET_ADDRSTRLEN)] = {0};
@@ -450,18 +442,18 @@ int doh_send_log(struct doh_conf *handle, const struct tfe_http_session *http, c
cJSON_AddStringToObject(common_obj, "ip_protocol", "tcp");
cJSON_AddNumberToObject(common_obj, "out_link_id", 0);
cJSON_AddNumberToObject(common_obj, "in_link_id", 0);
cJSON_AddStringToObject(common_obj, "sled_ip", handle->kafka_logger->local_ip_str);
cJSON_AddNumberToObject(common_obj, "t_vsys_id", handle->kafka_logger->t_vsys_id);
cJSON_AddStringToObject(common_obj, "sled_ip", tfe_get_sled_ip());
cJSON_AddNumberToObject(common_obj, "t_vsys_id", tfe_get_vsys_id());
cJSON_AddNumberToObject(common_obj, "vsys_id", ctx->vsys_id);
cJSON_AddStringToObject(common_obj, "device_id", handle->device_id);
cJSON_AddStringToObject(common_obj, "device_id", tfe_get_device_id());
cJSON_AddNumberToObject(common_obj, "sent_bytes", c2s_byte_num);
cJSON_AddNumberToObject(common_obj, "received_bytes", s2c_byte_num);
cJSON_AddStringToObject(common_obj, "doh_url", http->req->req_spec.url);
doh_add_host_to_object(common_obj, http->req->req_spec.host);
if(handle->effective_device_tag)
if(tfe_get_device_tag())
{
cJSON_AddStringToObject(common_obj, "device_tag", handle->effective_device_tag);
cJSON_AddStringToObject(common_obj, "device_tag", tfe_get_device_tag());
}
for (size_t i = 0; i < sizeof(req_fields) / sizeof(struct json_spec); i++)
@@ -524,19 +516,12 @@ int doh_send_log(struct doh_conf *handle, const struct tfe_http_session *http, c
log_payload = cJSON_PrintUnformatted(per_hit_obj);
TFE_LOG_DEBUG(handle->local_logger, "%s", log_payload);
kafka_status = tfe_kafka_logger_send(handle->kafka_logger, TOPIC_LOGGER, log_payload, strlen(log_payload));
free(log_payload);
cJSON_Delete(per_hit_obj);
if (kafka_status < 0)
{
TFE_LOG_ERROR(handle->local_logger, "Kafka produce failed: %s", rd_kafka_err2name(rd_kafka_last_error()));
}
else
if (kafka_send(tfe_get_kafka_handle(), TOPIC_PROXY_EVENT, log_payload, strlen(log_payload)) == 0)
{
send_cnt++;
}
free(log_payload);
cJSON_Delete(per_hit_obj);
}
cJSON_Delete(common_obj);

3
plugin/business/doh/src/pub.h
View File

@@ -13,7 +13,6 @@ extern "C"
#include <tfe_plugin.h>
#include <MESA/maat.h>
#include <MESA/MESA_prof_load.h>
#include <tfe_kafka_logger.h>
#include "dns.h"
@@ -57,9 +56,7 @@ struct doh_conf
int entry_id;
int en_sendlog;
const char *device_id;
const char *effective_device_tag;
tfe_kafka_logger_t *kafka_logger;
int fs_id[DOH_STAT_MAX];
long long stat_val[DOH_STAT_MAX];

2
plugin/business/ssl-policy/src/ssl_policy.cpp
View File

@@ -171,7 +171,7 @@ struct ssl_policy_enforcer* ssl_policy_enforcer_create(void* logger)
{
UNUSED int ret=0;
struct ssl_policy_enforcer* enforcer=ALLOC(struct ssl_policy_enforcer, 1);
enforcer->maat=(struct maat*)tfe_bussiness_resouce_get(STATIC_MAAT);;
enforcer->maat=tfe_get_maat_handle();
enforcer->logger=logger;
enforcer->profile_table_id=maat_get_table_id(enforcer->maat, "PXY_PROFILE_DECRYPTION");
assert(enforcer->profile_table_id >= 0);

2
plugin/business/tcp-policy/src/tcp_policy.cpp
View File

@@ -209,7 +209,7 @@ struct tcp_policy_enforcer *tcp_policy_enforcer_create(void *logger)
{
int ret = 0;
struct tcp_policy_enforcer *enforcer = ALLOC(struct tcp_policy_enforcer, 1);
enforcer->maat = (struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT);
enforcer->maat = tfe_get_maat_handle();
enforcer->logger = logger;
enforcer->table_id = maat_get_table_id(enforcer->maat, "PXY_PROFILE_TCP_OPTION");
if (enforcer->table_id < 0)

4
plugin/business/tsg-http/src/tsg_http.cpp
View File

@@ -1004,7 +1004,7 @@ int maat_table_ex_init(int profile_idx,
int proxy_policy_init(const char* profile_path, const char* static_section, const char* dynamic_section)
{
int ret = 0;
g_proxy_rt->feather = (struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT);
g_proxy_rt->feather = tfe_get_maat_handle();
const char * table_name[__SCAN_TABLE_MAX];
table_name[PXY_CTRL_HTTP_URL] = "ATTR_HTTP_URL";
@@ -1404,7 +1404,7 @@ void proxy_send_metric_log(const struct tfe_stream * stream, struct proxy_http_c
proxy_action_map[PX_ACTION_REJECT]="deny";
proxy_action_map[PX_ACTION_WHITELIST]="allow";
const char *manipulate_action_map[]= {"redirect","block","replace","hijack","insert","edit_element","run_script"};
struct tfe_fieldstat_metric_t *fieldstat = (struct tfe_fieldstat_metric_t *)tfe_bussiness_resouce_get(DYNAMIC_FIELDSTAT);
struct tfe_fieldstat_metric_t *fieldstat = tfe_get_fieldstat_handle();
for(i=0; i< ctx->n_enforce; i++)
{

52
plugin/business/tsg-http/src/tsg_logger.cpp
View File

@@ -1,12 +1,12 @@
#include <cjson/cJSON.h>
#include <MESA/MESA_prof_load.h>
#include <tfe_kafka_logger.h>
#include <tfe_utils.h>
#include <tfe_resource.h>
#include <event2/event.h>
#include <event.h>
#include "kafka.h"
#include "mpack.h"
#include "tsg_proxy_logger.h"
@@ -19,15 +19,12 @@ struct proxy_logger
{
int entry_id;
unsigned int en_sendlog;
const char *device_id;
const char *effective_device_tag;
void* local_logger;
unsigned long long send_cnt;
unsigned long long random_drop;
unsigned long long user_abort;
char local_log_path[TFE_STRING_MAX];
tfe_kafka_logger_t *kafka_logger;
struct cache_evbase_instance * log_file_upload_instance;
};
@@ -59,7 +56,6 @@ void get_http_body_uuid(char *uuid)
size_t file_bucket_upload_once(struct proxy_logger* handle, char *uuid, struct evbuffer *http_body)
{
int kafka_status=0;
mpack_writer_t writer;
char *mpack_data=NULL, *data=NULL;
size_t mpack_size=0, datalen=0;
@@ -94,11 +90,7 @@ size_t file_bucket_upload_once(struct proxy_logger* handle, char *uuid, struct e
{
TFE_LOG_ERROR(handle->local_logger, "Mpack writer destroy is error(%s), uuid: %s", mpack_error_to_string(errorno), uuid);
}
kafka_status = tfe_kafka_logger_send(handle->kafka_logger, TOPIC_BUCKET, mpack_data, mpack_size);
if(kafka_status<0)
{
TFE_LOG_ERROR(handle->local_logger, "Kafka produce failed: %s", rd_kafka_err2name(rd_kafka_last_error()));
}
kafka_send(tfe_get_kafka_handle(), TOPIC_FILE_STREAM, mpack_data, mpack_size);
free(mpack_data);
mpack_data = NULL;
@@ -112,29 +104,10 @@ struct proxy_logger* proxy_log_handle_create(const char* profile, const char* se
struct proxy_logger* instance=ALLOC(struct proxy_logger,1);
instance->local_logger=local_logger;
TFE_LOG_INFO(local_logger,"Tsg-Pxy log is inititating from %s section %s.", profile, section);
MESA_load_profile_uint_def(profile, section, "en_sendlog", &instance->en_sendlog, 1);
TFE_LOG_INFO(local_logger, "Tsg-Pxy sendlog : %s", instance->en_sendlog ? "ENABLE" : "DISABLE");
if (!instance->en_sendlog)
{
return instance;
}
instance->device_id = (const char *)tfe_bussiness_resouce_get(DEVICE_ID);
instance->effective_device_tag = (const char *)tfe_bussiness_resouce_get(EFFECTIVE_DEVICE_TAG);
instance->kafka_logger = (tfe_kafka_logger_t *)tfe_bussiness_resouce_get(KAFKA_LOGGER);
if (instance->kafka_logger && !instance->kafka_logger->enable)
{
TFE_LOG_ERROR(local_logger, "Tsg-Pxy sendlog ENABLE, but tfe kafka logger DISABLED.");
goto error_out;
}
return instance;
error_out:
free(instance);
return NULL;
}
static int get_ip_client_geolocation(struct tfe_cmsg * cmsg, cJSON *per_hit_obj)
@@ -193,7 +166,6 @@ int proxy_send_log(struct proxy_logger* handle, const struct proxy_log* log_msg)
const char* tmp_val=NULL;
cJSON *common_obj=NULL, *per_hit_obj=NULL;
char* log_payload=NULL;
int kafka_status=0;
int send_cnt=0;
struct timeval cur_time;
char src_ip_str[MAX(INET6_ADDRSTRLEN,INET_ADDRSTRLEN)] = {0};
@@ -305,17 +277,17 @@ int proxy_send_log(struct proxy_logger* handle, const struct proxy_log* log_msg)
cJSON_AddStringToObject(common_obj, "ip_protocol", "tcp");
cJSON_AddNumberToObject(common_obj, "out_link_id", 0);
cJSON_AddNumberToObject(common_obj, "in_link_id", 0);
cJSON_AddStringToObject(common_obj, "sled_ip", handle->kafka_logger->local_ip_str);
cJSON_AddNumberToObject(common_obj, "t_vsys_id", handle->kafka_logger->t_vsys_id);
cJSON_AddStringToObject(common_obj, "device_id", handle->device_id);
cJSON_AddStringToObject(common_obj, "sled_ip", tfe_get_sled_ip());
cJSON_AddNumberToObject(common_obj, "t_vsys_id", tfe_get_vsys_id());
cJSON_AddStringToObject(common_obj, "device_id", tfe_get_device_id());
cJSON_AddNumberToObject(common_obj, "sent_bytes", c2s_byte_num);
cJSON_AddNumberToObject(common_obj, "received_bytes", s2c_byte_num);
cJSON_AddStringToObject(common_obj, "http_url", http->req->req_spec.url);
proxy_add_host_to_object(common_obj, http->req->req_spec.host);
if(handle->effective_device_tag)
if (tfe_get_device_tag())
{
cJSON_AddStringToObject(common_obj, "device_tag", handle->effective_device_tag);
cJSON_AddStringToObject(common_obj, "device_tag", tfe_get_device_tag());
}
for(size_t i=0;i<sizeof(req_fields)/sizeof(struct json_spec);i++)
@@ -459,14 +431,12 @@ int proxy_send_log(struct proxy_logger* handle, const struct proxy_log* log_msg)
TFE_LOG_DEBUG(handle->local_logger, "%s", log_payload);
kafka_status = tfe_kafka_logger_send(handle->kafka_logger, TOPIC_LOGGER, log_payload, strlen(log_payload));
if (kafka_send(tfe_get_kafka_handle(), TOPIC_PROXY_EVENT, log_payload, strlen(log_payload)) == 0)
{
send_cnt++;
}
free(log_payload);
cJSON_Delete(per_hit_obj);
if(kafka_status<0)
{
TFE_LOG_ERROR(handle->local_logger, "Kafka produce failed: %s", rd_kafka_err2name(rd_kafka_last_error()));
}
send_cnt++;
}
cJSON_Delete(common_obj);