diff --git a/common/CMakeLists.txt b/common/CMakeLists.txt index 35a8c40..0aa8dfc 100644 --- a/common/CMakeLists.txt +++ b/common/CMakeLists.txt @@ -1,6 +1,6 @@ add_library( common src/tfe_utils.cpp src/tfe_types.cpp src/tfe_future.cpp src/tfe_http.cpp src/tfe_plugin.cpp - src/tfe_rpc.cpp src/tfe_cmsg.cpp src/tfe_kafka_logger.cpp src/tfe_resource.cpp src/tfe_scan.cpp + src/tfe_rpc.cpp src/tfe_cmsg.cpp src/kafka.cpp src/tfe_resource.cpp src/tfe_scan.cpp src/tfe_pkt_util.cpp src/tfe_tcp_restore.cpp src/packet_construct.cpp src/tap.cpp src/io_uring.cpp src/intercept_policy.cpp src/tfe_fieldstat.cpp src/tuple.cpp src/tfe_packet_io.cpp src/tfe_session_table.cpp diff --git a/common/include/kafka.h b/common/include/kafka.h new file mode 100644 index 0000000..4a242ed --- /dev/null +++ b/common/include/kafka.h @@ -0,0 +1,31 @@ +#ifndef _KAFKA_H +#define _KAFKA_H + +#ifdef __cplusplus +extern "C" +{ +#endif + +enum topic_idx +{ + TOPIC_RULE_HITS, + TOPIC_PROXY_EVENT, + TOPIC_FILE_STREAM, + TOPIC_EXCH_CERT, + + // add more topic here + + MAX_TOPIC_NUM, +}; + +struct kafka *kafka_create(const char *profile); +void kafka_destroy(struct kafka *handle); +// return 0: if success +// return -1: if failed +int kafka_send(struct kafka *handle, enum topic_idx idx, const char *data, int len); + +#ifdef __cplusplus +} +#endif + +#endif diff --git a/common/include/tfe_kafka_logger.h b/common/include/tfe_kafka_logger.h deleted file mode 100644 index 0ea0d20..0000000 --- a/common/include/tfe_kafka_logger.h +++ /dev/null @@ -1,44 +0,0 @@ -#ifndef _TFE_KAFKA_LOGGER_H -#define _TFE_KAFKA_LOGGER_H - -#ifdef __cpluscplus -extern "C" -{ -#endif - -#include -#include - -enum kafka_topic_type -{ - TOPIC_LOGGER, - TOPIC_BUCKET, - TOPIC_MC_CACHE, - TOPIC_MAX -}; - -typedef struct tfe_kafka_logger_s -{ - int enable; - int t_vsys_id; - - unsigned int local_ip_num; - char local_ip_str[TFE_SYMBOL_MAX]; - - char topic_name[TOPIC_MAX][TFE_STRING_MAX]; - char broker_list[TFE_STRING_MAX]; - - rd_kafka_t *kafka_handle[TOPIC_MAX]; - rd_kafka_topic_t *kafka_topic[TOPIC_MAX]; -} tfe_kafka_logger_t; - -tfe_kafka_logger_t *tfe_kafka_logger_create(int enable, const char *nic_name, const char *brokerlist, void *local_logger); -int tfe_logger_create_kafka_topic(tfe_kafka_logger_t *logger, const char *sasl_username, const char *sasl_passwd, const char *topic_name, int topic_id, void *local_logger); -void tfe_kafka_logger_destroy(tfe_kafka_logger_t *logger); -int tfe_kafka_logger_send(tfe_kafka_logger_t *logger, int topic_id, const char *data, int len); - -#ifdef __cpluscplus -} -#endif - -#endif diff --git a/common/include/tfe_resource.h b/common/include/tfe_resource.h index 8597770..a2f321e 100644 --- a/common/include/tfe_resource.h +++ b/common/include/tfe_resource.h @@ -11,15 +11,6 @@ struct app_id_dict void app_id_dict_free(struct app_id_dict *app_dict); -enum RESOURCE_TYPE -{ - STATIC_MAAT, - KAFKA_LOGGER, - DEVICE_ID, - EFFECTIVE_DEVICE_TAG, - DYNAMIC_FIELDSTAT, -}; - enum scan_common_table { PXY_CTRL_SOURCE_IP, @@ -46,6 +37,15 @@ enum scan_common_table __SCAN_COMMON_TABLE_MAX }; -int tfe_bussiness_resouce_init(); -void *tfe_bussiness_resouce_get(enum RESOURCE_TYPE type); -int tfe_bussiness_tableid_get(enum scan_common_table type); \ No newline at end of file +int tfe_env_init(); +int tfe_bussiness_tableid_get(enum scan_common_table type); + +int tfe_get_vsys_id(); +const char *tfe_get_device_id(); +const char *tfe_get_data_center(); +const char *tfe_get_device_group(); +const char *tfe_get_device_tag(); +const char *tfe_get_sled_ip(); +struct kafka *tfe_get_kafka_handle(); +struct maat *tfe_get_maat_handle(); +struct tfe_fieldstat_metric_t *tfe_get_fieldstat_handle(); \ No newline at end of file diff --git a/common/include/tfe_utils.h b/common/include/tfe_utils.h index bda4958..a8b37a3 100644 --- a/common/include/tfe_utils.h +++ b/common/include/tfe_utils.h @@ -9,18 +9,11 @@ #include #include //scan_dir #include +#include "kafka.h" -#define LOG_TAG_POLICY "POLICY" -#define LOG_TAG_UTILS "UTILS" -#define LOG_TAG_RAWPKT "RAW_PACKET" #define LOG_TAG_CTRLPKT "CTRL_PACKET" #define LOG_TAG_STABLE "SESSION_TABLE" #define LOG_TAG_PKTIO "PACKET_IO" -#define LOG_TAG_METRICS "G_METRICS" -#define LOG_TAG_SF_METRICS "SF_METRICS" -#define LOG_TAG_SF_STATUS "SF_STATUS" -#define LOG_TAG_SCE "SCE" -#define LOG_TAG_TIMESTAMP "TIMESTAMP" #define TFE_STRING_MAX 2048 #define TFE_PATH_MAX 256 diff --git a/common/src/intercept_policy.cpp b/common/src/intercept_policy.cpp index 47c9607..d8f57cb 100644 --- a/common/src/intercept_policy.cpp +++ b/common/src/intercept_policy.cpp @@ -217,7 +217,7 @@ struct intercept_policy_enforcer *intercept_policy_enforcer_create(void *logger) { int ret = 0; struct intercept_policy_enforcer *enforcer = ALLOC(struct intercept_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_INTERCEPT_COMPILE"); diff --git a/common/src/kafka.cpp b/common/src/kafka.cpp new file mode 100644 index 0000000..a82c641 --- /dev/null +++ b/common/src/kafka.cpp @@ -0,0 +1,251 @@ +#include +#include + +#include "kafka.h" +#include "tfe_utils.h" +#include +#include + +#define MAX_SYMBOL_LEN 128 + +struct config +{ + char brokerlist[MAX_SYMBOL_LEN]; + char sasl_username[MAX_SYMBOL_LEN]; + char sasl_passwd[MAX_SYMBOL_LEN]; + char topic_name[MAX_TOPIC_NUM][MAX_SYMBOL_LEN]; +}; + +struct per_producer_per_topic +{ + rd_kafka_t *producer; + rd_kafka_topic_t *topic; +}; + +struct kafka +{ + struct config cfg; + struct per_producer_per_topic *pppt[MAX_TOPIC_NUM]; +}; + +/****************************************************************************** + * Private API + ******************************************************************************/ + +static void per_producer_per_topic_free(struct per_producer_per_topic *pppt) +{ + if (pppt) + { + if (pppt->topic) + { + rd_kafka_topic_destroy(pppt->topic); + pppt->topic = NULL; + } + + if (pppt->producer) + { + rd_kafka_destroy(pppt->producer); + pppt->producer = NULL; + } + + free(pppt); + pppt = NULL; + } +} + +static struct per_producer_per_topic *per_producer_per_topic_new(const char *brokerlist, const char *sasl_username, const char *sasl_passwd, const char *topic_name) +{ + char err_str[1024] = {0}; + struct per_producer_per_topic *pppt = (struct per_producer_per_topic *)calloc(1, sizeof(struct per_producer_per_topic)); + if (!pppt) + { + return NULL; + } + + rd_kafka_conf_t *conf = rd_kafka_conf_new(); + if (!conf) + { + TFE_LOG_ERROR(g_default_logger, "KAFKA: failed to create kafka conf"); + goto error_out; + } + if (rd_kafka_conf_set(conf, "queue.buffering.max.messages", "1000000", err_str, sizeof(err_str)) != RD_KAFKA_CONF_OK) + { + TFE_LOG_ERROR(g_default_logger, "KAFKA: failed to set kafka queue.buffering.max.messages, %s", err_str); + goto error_out; + } + if (rd_kafka_conf_set(conf, "topic.metadata.refresh.interval.ms", "600000", err_str, sizeof(err_str)) != RD_KAFKA_CONF_OK) + { + TFE_LOG_ERROR(g_default_logger, "KAFKA: failed to set kafka topic.metadata.refresh.interval.ms, %s", err_str); + goto error_out; + } + if (rd_kafka_conf_set(conf, "client.id", topic_name, err_str, sizeof(err_str)) != RD_KAFKA_CONF_OK) + { + TFE_LOG_ERROR(g_default_logger, "KAFKA: failed to set kafka client.id, %s", err_str); + goto error_out; + } + if (strlen(sasl_username) > 0 && strlen(sasl_passwd) > 0) + { + if (rd_kafka_conf_set(conf, "security.protocol", "sasl_plaintext", err_str, sizeof(err_str)) != RD_KAFKA_CONF_OK) + { + TFE_LOG_ERROR(g_default_logger, "KAFKA: failed to set kafka security.protocol, %s", err_str); + goto error_out; + } + if (rd_kafka_conf_set(conf, "sasl.mechanisms", "PLAIN", err_str, sizeof(err_str)) != RD_KAFKA_CONF_OK) + { + TFE_LOG_ERROR(g_default_logger, "KAFKA: failed to set kafka sasl.mechanisms, %s", err_str); + goto error_out; + } + if (rd_kafka_conf_set(conf, "sasl.username", sasl_username, err_str, sizeof(err_str)) != RD_KAFKA_CONF_OK) + { + TFE_LOG_ERROR(g_default_logger, "KAFKA: failed to set kafka sasl.username, %s", err_str); + goto error_out; + } + if (rd_kafka_conf_set(conf, "sasl.password", sasl_passwd, err_str, sizeof(err_str)) != RD_KAFKA_CONF_OK) + { + TFE_LOG_ERROR(g_default_logger, "KAFKA: failed to set kafka sasl.password, %s", err_str); + goto error_out; + } + } + else + { + if (rd_kafka_conf_set(conf, "security.protocol", "plaintext", err_str, sizeof(err_str)) != RD_KAFKA_CONF_OK) + { + TFE_LOG_ERROR(g_default_logger, "KAFKA: failed to set kafka security.protocol, %s", err_str); + goto error_out; + } + } + + // The conf object is freed by this function and must not be used or destroyed by the application sub-sequently. + pppt->producer = rd_kafka_new(RD_KAFKA_PRODUCER, conf, err_str, sizeof(err_str)); + conf = NULL; + if (pppt->producer == NULL) + { + TFE_LOG_ERROR(g_default_logger, "KAFKA: failed to create kafka producer, %s", err_str); + goto error_out; + } + + if (rd_kafka_brokers_add(pppt->producer, brokerlist) == 0) + { + TFE_LOG_ERROR(g_default_logger, "KAFKA: failed to add kafka brokers"); + goto error_out; + } + + pppt->topic = rd_kafka_topic_new(pppt->producer, topic_name, NULL); + if (pppt->topic == NULL) + { + TFE_LOG_ERROR(g_default_logger, "KAFKA: failed to create kafka topic: %s", topic_name); + goto error_out; + } + + return pppt; + +error_out: + if (conf) + { + rd_kafka_conf_destroy(conf); + } + + per_producer_per_topic_free(pppt); + return NULL; +} + +/****************************************************************************** + * Public API -- Kafka + ******************************************************************************/ + +// due to limit by client.id, need per producer per topic +struct kafka *kafka_create(const char *profile) +{ + struct kafka *handle = (struct kafka *)calloc(1, sizeof(struct kafka)); + if (!handle) + { + return NULL; + } + + MESA_load_profile_string_def(profile, "kafka", "brokerlist", handle->cfg.brokerlist, sizeof(handle->cfg.brokerlist), ""); + MESA_load_profile_string_def(profile, "kafka", "sasl_username", handle->cfg.sasl_username, sizeof(handle->cfg.sasl_username), ""); + MESA_load_profile_string_def(profile, "kafka", "sasl_passwd", handle->cfg.sasl_passwd, sizeof(handle->cfg.sasl_passwd), ""); + MESA_load_profile_string_def(profile, "kafka", "rule_hits_topic", handle->cfg.topic_name[TOPIC_RULE_HITS], sizeof(handle->cfg.topic_name[TOPIC_RULE_HITS]), ""); + MESA_load_profile_string_def(profile, "kafka", "proxy_event_topic", handle->cfg.topic_name[TOPIC_PROXY_EVENT], sizeof(handle->cfg.topic_name[TOPIC_PROXY_EVENT]), ""); + MESA_load_profile_string_def(profile, "kafka", "file_stream_topic", handle->cfg.topic_name[TOPIC_FILE_STREAM], sizeof(handle->cfg.topic_name[TOPIC_FILE_STREAM]), ""); + MESA_load_profile_string_def(profile, "kafka", "exch_cert_topic", handle->cfg.topic_name[TOPIC_EXCH_CERT], sizeof(handle->cfg.topic_name[TOPIC_EXCH_CERT]), ""); + + if (strlen(handle->cfg.brokerlist) == 0) + { + TFE_LOG_ERROR(g_default_logger, "KAFKA: brokerlist is empty"); + goto error_out; + } + + for (int i = 0; i < MAX_TOPIC_NUM; i++) + { + if (strlen(handle->cfg.topic_name[i]) == 0) + { + TFE_LOG_ERROR(g_default_logger, "KAFKA: topic_name[%d] is empty", i); + goto error_out; + } + } + + for (int i = 0; i < MAX_TOPIC_NUM; i++) + { + handle->pppt[i] = per_producer_per_topic_new(handle->cfg.brokerlist, handle->cfg.sasl_username, handle->cfg.sasl_passwd, handle->cfg.topic_name[i]); + if (!handle->pppt[i]) + { + goto error_out; + } + } + + return handle; + +error_out: + kafka_destroy(handle); + return NULL; +} + +void kafka_destroy(struct kafka *handle) +{ + if (handle) + { + for (int i = 0; i < MAX_TOPIC_NUM; i++) + { + per_producer_per_topic_free(handle->pppt[i]); + handle->pppt[i] = NULL; + } + + free(handle); + handle = NULL; + } +} + +int kafka_send(struct kafka *handle, enum topic_idx idx, const char *data, int len) +{ + if (!handle) + { + TFE_LOG_ERROR(g_default_logger, "KAFKA: handle is NULL"); + return -1; + } + + if (idx < 0 || idx >= MAX_TOPIC_NUM) + { + TFE_LOG_ERROR(g_default_logger, "KAFKA: invalid topic index: %d", idx); + return -1; + } + + if (handle->pppt[idx]) + { + if (rd_kafka_produce(handle->pppt[idx]->topic, RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_COPY, (void *)data, len, NULL, 0, NULL) == -1) + { + TFE_LOG_ERROR(g_default_logger, "KAFKA: failed to produce message with topic [%d], %s", idx, rd_kafka_err2str(rd_kafka_last_error())); + return -1; + } + else + { + TFE_LOG_DEBUG(g_default_logger, "KAFKA: success to produce message with topic [%d], %s", idx, data); + return 0; + } + } + else + { + TFE_LOG_ERROR(g_default_logger, "KAFKA: topic %d not initialized", idx); + return -1; + } +} \ No newline at end of file diff --git a/common/src/tfe_kafka_logger.cpp b/common/src/tfe_kafka_logger.cpp deleted file mode 100644 index 28f4ce8..0000000 --- a/common/src/tfe_kafka_logger.cpp +++ /dev/null @@ -1,212 +0,0 @@ -#include -#include -#include -#include - -#include - -// return INADDR_NONE if error occur -static unsigned int get_ip_by_eth_name(const char *ifname) -{ - int sockfd = -1; - struct ifreq ifr; - unsigned int ip; - - sockfd = socket(AF_INET, SOCK_DGRAM, 0); - if (-1 == sockfd) - { - goto error; - } - - strcpy(ifr.ifr_name, ifname); - if (ioctl(sockfd, SIOCGIFADDR, &ifr) == -1) - { - goto error; - } - close(sockfd); - - ip = ((struct sockaddr_in *)&(ifr.ifr_addr))->sin_addr.s_addr; - return ip; - -error: - if (sockfd > 0) - close(sockfd); - return INADDR_NONE; -} - -static rd_kafka_t *create_kafka_handle(const char *brokerlist, const char *sasl_username, const char *sasl_passwd, const char *topic_name, void *local_logger) -{ - int ret; - char kafka_errstr[1024] = {0}; - rd_kafka_t *handle = NULL; - rd_kafka_conf_t *rconf = NULL; - - rconf = rd_kafka_conf_new(); - - ret = rd_kafka_conf_set(rconf, "queue.buffering.max.messages", "1000000", kafka_errstr, sizeof(kafka_errstr)); - if (ret != RD_KAFKA_CONF_OK) - { - TFE_LOG_ERROR(local_logger, "Error to set kafka \"queue.buffering.max.messages\", %s.", kafka_errstr); - rd_kafka_conf_destroy(rconf); - return NULL; - } - ret = rd_kafka_conf_set(rconf, "topic.metadata.refresh.interval.ms", "600000", kafka_errstr, sizeof(kafka_errstr)); - if (ret != RD_KAFKA_CONF_OK) - { - TFE_LOG_ERROR(local_logger, "Error to set kafka \"topic.metadata.refresh.interval.ms\", %s.", kafka_errstr); - rd_kafka_conf_destroy(rconf); - return NULL; - } - ret = rd_kafka_conf_set(rconf, "security.protocol", "plaintext", kafka_errstr, sizeof(kafka_errstr)); - if (ret != RD_KAFKA_CONF_OK) - { - TFE_LOG_ERROR(local_logger, "Error to set kafka \"security.protocol\", %s.", kafka_errstr); - rd_kafka_conf_destroy(rconf); - return NULL; - } - ret = rd_kafka_conf_set(rconf, "client.id", topic_name, kafka_errstr, sizeof(kafka_errstr)); - if (ret != RD_KAFKA_CONF_OK) - { - TFE_LOG_ERROR(local_logger, "Error to set kafka \"client.id\", %s.", kafka_errstr); - rd_kafka_conf_destroy(rconf); - return NULL; - } - - if (strlen(sasl_username) > 0 && strlen(sasl_passwd) > 0) - { - rd_kafka_conf_set(rconf, "security.protocol", "sasl_plaintext", kafka_errstr, sizeof(kafka_errstr)); - rd_kafka_conf_set(rconf, "sasl.mechanisms", "PLAIN", kafka_errstr, sizeof(kafka_errstr)); - ret = rd_kafka_conf_set(rconf, "sasl.username", sasl_username, kafka_errstr, sizeof(kafka_errstr)); - if (ret != RD_KAFKA_CONF_OK) - { - TFE_LOG_ERROR(local_logger, "Error to set kafka \"sasl.username\", %s.", kafka_errstr); - rd_kafka_conf_destroy(rconf); - return NULL; - } - ret = rd_kafka_conf_set(rconf, "sasl.password", sasl_passwd, kafka_errstr, sizeof(kafka_errstr)); - if (ret != RD_KAFKA_CONF_OK) - { - TFE_LOG_ERROR(local_logger, "Error to set kafka \"sasl.password\", %s.", kafka_errstr); - rd_kafka_conf_destroy(rconf); - return NULL; - } - } - - //The conf object is freed by this function and must not be used or destroyed by the application sub-sequently. - handle = rd_kafka_new(RD_KAFKA_PRODUCER, rconf, kafka_errstr, sizeof(kafka_errstr)); - rconf = NULL; - if (handle == NULL) - { - TFE_LOG_ERROR(local_logger, "Error to new kafka, %s.", kafka_errstr); - return NULL; - } - - if (rd_kafka_brokers_add(handle, brokerlist) == 0) - { - TFE_LOG_ERROR(local_logger, "Error to add kakfa bokers."); - rd_kafka_destroy(handle); - return NULL; - } - - return handle; -} - -int tfe_kafka_logger_topic_new(tfe_kafka_logger_t *logger, const char *topic_name, int topic_id, void *local_logger) -{ - if(logger && logger->enable) - { - strncpy(logger->topic_name[topic_id], topic_name, sizeof(logger->topic_name[topic_id])-1); - logger->kafka_topic[topic_id] = rd_kafka_topic_new(logger->kafka_handle[topic_id], topic_name, NULL); - if (logger->kafka_topic[topic_id] == NULL) - { - TFE_LOG_ERROR(local_logger, "Error to creat kafka topic: %s.", topic_name); - rd_kafka_destroy(logger->kafka_handle[topic_id]); - free(logger); - return -1; - } - } - return 0; -} - -tfe_kafka_logger_t *tfe_kafka_logger_create(int enable, const char *nic_name, const char *brokerlist, void *local_logger) -{ - char *override_sled_ip=NULL; - - tfe_kafka_logger_t *logger = (tfe_kafka_logger_t *)calloc(1, sizeof(tfe_kafka_logger_t)); - if (!logger) - { - return NULL; - } - - logger->enable = enable; - if (!logger->enable) - { - return logger; - } - - override_sled_ip = getenv("OVERRIDE_SLED_IP"); - if(override_sled_ip != NULL) - { - strncpy(logger->local_ip_str, override_sled_ip, sizeof(logger->local_ip_str)-1); - goto finish; - } - - logger->local_ip_num = get_ip_by_eth_name(nic_name); - if (logger->local_ip_num == INADDR_NONE) - { - TFE_LOG_ERROR(local_logger, "Error to get NIC_NAME: %s.", nic_name); - free(logger); - return NULL; - } - inet_ntop(AF_INET, &(logger->local_ip_num), logger->local_ip_str, sizeof(logger->local_ip_str)); -finish: - strncpy(logger->broker_list, brokerlist, sizeof(logger->broker_list)-1); - return logger; -} - -int tfe_logger_create_kafka_topic(tfe_kafka_logger_t *logger, const char *sasl_username, const char *sasl_passwd, const char *topic_name, int topic_id, void *local_logger) -{ - if(!logger->enable) - { - return 0; - } - - logger->kafka_handle[topic_id] = create_kafka_handle(logger->broker_list, sasl_username, sasl_passwd, topic_name, local_logger); - if (logger->kafka_handle[topic_id] == NULL) - { - TFE_LOG_ERROR(local_logger, "Error to creat kafka handler with brokerlist: %s.", logger->broker_list); - free(logger); - return -1; - } - tfe_kafka_logger_topic_new(logger, topic_name, topic_id, logger); - return 0; -} - -void tfe_kafka_logger_destroy(tfe_kafka_logger_t *logger) -{ - if (logger) - { - for(int i=0; ikafka_topic[i]) - { - rd_kafka_topic_destroy(logger->kafka_topic[i]); - } - - if(logger->kafka_handle[i]) - { - rd_kafka_destroy(logger->kafka_handle[i]); - } - } - free(logger); - logger = NULL; - } -} - -int tfe_kafka_logger_send(tfe_kafka_logger_t *logger, int topic_id, const char *data, int len) -{ - if (logger && logger->enable) - return rd_kafka_produce(logger->kafka_topic[topic_id], RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_COPY, (void *)data, len, NULL, 0, NULL); - else - return 0; -} diff --git a/common/src/tfe_resource.cpp b/common/src/tfe_resource.cpp index c0c4e6d..8379a65 100644 --- a/common/src/tfe_resource.cpp +++ b/common/src/tfe_resource.cpp @@ -1,21 +1,73 @@ #include #include #include -#include #include #include #include +#include "kafka.h" #define MAAT_INPUT_JSON 0 #define MAAT_INPUT_REDIS 1 #define MAAT_INPUT_FILE 2 static int scan_table_id[__SCAN_COMMON_TABLE_MAX]; -static struct maat *static_maat = NULL; -static tfe_kafka_logger_t *kafka_logger = NULL; static struct tfe_fieldstat_metric_t *dynamic_fieldstat = NULL; -static char *device_id = NULL; -static char *effective_device_tag=NULL; +static char *device_tag=NULL; + +struct tfe_fieldstat_metric_t *fieldstat_handle = NULL; +struct kafka *kafka_handle = NULL; +struct maat *maat_handle = NULL; + +static int vsys_id = 0; +static char data_center[1024] = {0}; +static char device_group[1024] = {0}; +static char device_id[1024] = {0}; +static char sled_ip[1024] = {0}; + +int tfe_get_vsys_id() +{ + return vsys_id; +} + +const char *tfe_get_device_id() +{ + return device_id; +} + +const char *tfe_get_data_center() +{ + return data_center; +} + +const char *tfe_get_device_group() +{ + return device_group; +} + +const char *tfe_get_device_tag() +{ + return device_tag; +} + +const char *tfe_get_sled_ip() +{ + return sled_ip; +} + +struct kafka *tfe_get_kafka_handle() +{ + return kafka_handle; +} + +struct maat *tfe_get_maat_handle() +{ + return maat_handle; +} + +struct tfe_fieldstat_metric_t *tfe_get_fieldstat_handle() +{ + return fieldstat_handle; +} static struct tfe_fieldstat_metric_t *create_fieldstat_instance(const char *profile, const char *section, int max_thread, void *logger) { @@ -173,129 +225,9 @@ error_out: return NULL; } -static tfe_kafka_logger_t *create_kafka_logger(const char *profile, const char *section, void *logger) +static char* create_device_tag(const char *profile, const char *section, void *logger) { - int ret=0, enable=0, vsystem_id=0; - char nic_name[TFE_SYMBOL_MAX] = {0}; - char brokerlist[TFE_STRING_MAX] = {0}; - char logger_topic[TFE_STRING_MAX] = {0}; - char bucket_topic[TFE_STRING_MAX] = {0}; - char sasl_username[TFE_STRING_MAX] = {0}; - char sasl_passwd[TFE_STRING_MAX] = {0}; - tfe_kafka_logger_t *kafka_logger = NULL; - - MESA_load_profile_int_def(profile, section, "enable", &enable, 1); - MESA_load_profile_int_def(profile, section, "VSYSTEM_ID", &vsystem_id, 1); - MESA_load_profile_string_def(profile, section, "NIC_NAME", nic_name, sizeof(nic_name), "eth0"); - MESA_load_profile_string_def(profile, section, "KAFKA_BROKERLIST", brokerlist, sizeof(brokerlist), ""); - MESA_load_profile_string_def(profile, section, "LOGGER_SEND_TOPIC", logger_topic, sizeof(logger_topic), "PROXY-EVENT"); - MESA_load_profile_string_def(profile, section, "FILE_BUCKET_TOPIC", bucket_topic, sizeof(bucket_topic), "TRAFFIC-FILE-STREAM-RECORD"); - MESA_load_profile_string_def(profile, section, "SASL_USERNAME", sasl_username, sizeof(sasl_username), ""); - MESA_load_profile_string_def(profile, section, "SASL_PASSWD", sasl_passwd, sizeof(sasl_passwd), ""); - - if (!strlen(brokerlist)) - { - TFE_LOG_ERROR(logger, "tfe kafka init failed, no brokerlist in profile %s section %s.", profile, section); - return NULL; - } - - kafka_logger = tfe_kafka_logger_create(enable, nic_name, brokerlist, logger); - if (kafka_logger == NULL) - { - TFE_LOG_ERROR(logger, "tfe kafka init failed, error to create kafka logger."); - return NULL; - } - - ret = tfe_logger_create_kafka_topic(kafka_logger, sasl_username, sasl_passwd, logger_topic, TOPIC_LOGGER, logger); - if(ret < 0) - { - TFE_LOG_ERROR(logger, "tfe kafka init failed, error to create %s topic.", logger_topic); - return NULL; - } - - ret = tfe_logger_create_kafka_topic(kafka_logger, sasl_username, sasl_passwd, bucket_topic, TOPIC_BUCKET, logger); - if(ret < 0) - { - TFE_LOG_ERROR(logger, "tfe kafka init failed, error to create %s topic.", bucket_topic); - return NULL; - } - kafka_logger->t_vsys_id=vsystem_id; - - TFE_LOG_INFO(logger, "tfe kafka logger : %s", enable ? "ENABLE" : "DISABLE"); - TFE_LOG_INFO(logger, "tfe kafka vsystem id : %d", vsystem_id); - TFE_LOG_INFO(logger, "tfe logger kafka topic : %s", logger_topic); - TFE_LOG_INFO(logger, "tfe bucket kafka topic : %s", bucket_topic); - TFE_LOG_INFO(logger, "tfe kafka brokerlist : %s", brokerlist); - - if (strlen(sasl_username) > 0 && strlen(sasl_passwd) > 0) - { - TFE_LOG_INFO(logger, "tfe kafka sasl_username : %s", sasl_username); - TFE_LOG_INFO(logger, "tfe kafka sasl_passwd : %s", sasl_passwd); - } - - return kafka_logger; -} - -static char *cerate_device_id(const char *profile, const char *section, void *logger) -{ - int ret = -1; - size_t device_id_size = 0; - char *tsg_sn_file = NULL, *device_id; - - const char *device_def_id = "DFT2201925000001"; - cJSON *json = NULL, *item = NULL; - char device_id_filepath[TFE_STRING_MAX] = {0}; - - ret = MESA_load_profile_string_def(profile, section, "device_id_filepath", device_id_filepath, sizeof(device_id_filepath), NULL); - if (ret < 0) - { - TFE_LOG_ERROR(logger, "Invalid device parameter: device_id_filepath not existed in profile %s section %s.", profile, section); - goto finish; - } - tsg_sn_file = tfe_read_file(device_id_filepath, &device_id_size); - if (tsg_sn_file == NULL) - { - TFE_LOG_ERROR(logger, "Invalid device parameter: device sn file not existed."); - goto finish; - } - json = cJSON_Parse(tsg_sn_file); - if (json == NULL) - { - TFE_LOG_ERROR(logger, "Invalid device parameter: %s invalid json format", tsg_sn_file); - goto finish; - } - item = cJSON_GetObjectItem(json, "sn"); - if (unlikely(!item || !cJSON_IsString(item))) - { - TFE_LOG_ERROR(logger, "Invalid device parameter: %s invalid json format", tsg_sn_file); - goto finish; - } - device_id = tfe_strdup(item->valuestring); - - if(tsg_sn_file) - { - FREE(&tsg_sn_file); - } - cJSON_Delete(json); - TFE_LOG_INFO(logger, "tfe device id : %s", device_id); - - return device_id; -finish: - TFE_LOG_INFO(logger, "tfe use default device id : %s", device_def_id); - if (json) - { - cJSON_Delete(json); - } - if(tsg_sn_file) - { - FREE(&tsg_sn_file); - } - return (char *)device_def_id; -} - -static char* create_effective_device_tag(const char *profile, const char *section, void *logger) -{ - char *effective_device_tag=NULL; + char *c=NULL; char accept_path[TFE_PATH_MAX] = {0}, accept_tags[TFE_STRING_MAX] = {0}; MESA_load_profile_string_def(profile, section, "accept_path", accept_path, sizeof(accept_path), ""); @@ -307,10 +239,10 @@ static char* create_effective_device_tag(const char *profile, const char *sectio { return NULL; } - effective_device_tag = tfe_strdup(accept_tags); - TFE_LOG_INFO(logger, "tfe device tag : %s", effective_device_tag); + device_tag = tfe_strdup(accept_tags); + TFE_LOG_INFO(logger, "tfe device tag : %s", device_tag); - return effective_device_tag; + return device_tag; } void app_dict_table_new_cb(const char *table_name, int table_id, const char* key, const char* table_line, void **ad, long argl, void* argp) @@ -410,71 +342,62 @@ static int maat_common_table_init() for (int i = 0; i < __SCAN_COMMON_TABLE_MAX; i++) { - scan_table_id[i] = maat_get_table_id(static_maat, table_name[i]); + scan_table_id[i] = maat_get_table_id(maat_handle, table_name[i]); if (scan_table_id[i] < 0) { TFE_LOG_ERROR(g_default_logger, "Maat table %s register failed.", table_name[i]); return -1; } } - maat_plugin_table_ex_schema_register(static_maat, "APP_ID_DICT", app_dict_table_new_cb, app_dict_table_free_cb, app_dict_table_dup_cb, 0, NULL); + maat_plugin_table_ex_schema_register(maat_handle, "APP_ID_DICT", app_dict_table_new_cb, app_dict_table_free_cb, app_dict_table_dup_cb, 0, NULL); return 0; } -int tfe_bussiness_resouce_init() +int tfe_env_init() { - const char *profile_path = "./conf/tfe/tfe.conf"; - unsigned int thread_num = tfe_proxy_get_work_thread_count(); - static_maat = create_maat_feather("static", profile_path, "MAAT", thread_num, g_default_logger); - if (!static_maat) - { - return -1; - } + const char *profile_path = "./conf/tfe/tfe.conf"; - kafka_logger = create_kafka_logger(profile_path, "kafka", g_default_logger); - if (!kafka_logger) - { - return -1; - } + MESA_load_profile_int_def(profile_path, "public", "vsys_id", &vsys_id, 0); + MESA_load_profile_string_def(profile_path, "public", "data_center", data_center, sizeof(data_center), ""); + MESA_load_profile_string_def(profile_path, "public", "device_group", device_group, sizeof(device_group), ""); + MESA_load_profile_string_def(profile_path, "public", "device_id", device_id, sizeof(device_id), ""); - dynamic_fieldstat = create_fieldstat_instance(profile_path, "proxy_hits", thread_num, g_default_logger); - if(!dynamic_fieldstat) - { - return -1; - } + char *ptr = getenv("OVERRIDE_SLED_IP"); + if (ptr == NULL) + { + return -1; + } + strncpy(sled_ip, ptr, strlen(ptr)); - device_id = cerate_device_id(profile_path, "kafka", g_default_logger); - - effective_device_tag = create_effective_device_tag(profile_path, "MAAT", g_default_logger); + kafka_handle = kafka_create(profile_path); + if (!kafka_handle) + { + return -1; + } - if (maat_common_table_init()) - { - return -1; - } + unsigned int thread_num = tfe_proxy_get_work_thread_count(); + maat_handle = create_maat_feather("static", profile_path, "MAAT", thread_num, g_default_logger); + if (!maat_handle) + { + return -1; + } - return 0; -} + dynamic_fieldstat = create_fieldstat_instance(profile_path, "proxy_hits", thread_num, g_default_logger); + if (!dynamic_fieldstat) + { + return -1; + } -void *tfe_bussiness_resouce_get(enum RESOURCE_TYPE type) -{ - switch (type) - { - case STATIC_MAAT: - return static_maat; - case KAFKA_LOGGER: - return kafka_logger; - case DEVICE_ID: - return device_id; - case EFFECTIVE_DEVICE_TAG: - return effective_device_tag; - case DYNAMIC_FIELDSTAT: - return dynamic_fieldstat; - default: - return NULL; - } + device_tag = create_device_tag(profile_path, "MAAT", g_default_logger); + if (maat_common_table_init()) + { + return -1; + } + + return 0; } int tfe_bussiness_tableid_get(enum scan_common_table type) { - return scan_table_id[type]; + return scan_table_id[type]; } \ No newline at end of file diff --git a/common/src/tfe_scan.cpp b/common/src/tfe_scan.cpp index fe77194..a886e68 100644 --- a/common/src/tfe_scan.cpp +++ b/common/src/tfe_scan.cpp @@ -30,7 +30,7 @@ int tfe_scan_subscribe_id(const struct tfe_stream *stream, long long *result, st if (strlen(source_subscribe_id)) { - scan_ret = maat_scan_string((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), tfe_bussiness_tableid_get(PXY_CTRL_SUBSCRIBER_ID), + scan_ret = maat_scan_string(tfe_get_maat_handle(), tfe_bussiness_tableid_get(PXY_CTRL_SUBSCRIBER_ID), source_subscribe_id, strlen(source_subscribe_id),result + hit_cnt + hit_cnt_ip, MAX_SCAN_RESULT - hit_cnt - hit_cnt_ip, &n_hit_result, scan_mid); if (scan_ret == MAAT_SCAN_HIT) @@ -44,7 +44,7 @@ int tfe_scan_subscribe_id(const struct tfe_stream *stream, long long *result, st TFE_LOG_INFO(logger, "Scan src TSG_OBJ_SUBSCRIBER_ID, NO hit subid: %s scan ret: %d addr: %s", source_subscribe_id, scan_ret, stream->str_stream_info); } - scan_ret = maat_scan_not_logic((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), tfe_bussiness_tableid_get(PXY_CTRL_SUBSCRIBER_ID), + scan_ret = maat_scan_not_logic(tfe_get_maat_handle(), tfe_bussiness_tableid_get(PXY_CTRL_SUBSCRIBER_ID), result + hit_cnt + hit_cnt_ip, MAX_SCAN_RESULT - hit_cnt - hit_cnt_ip, &n_hit_result, scan_mid); if (scan_ret == MAAT_SCAN_HIT) { @@ -54,7 +54,7 @@ int tfe_scan_subscribe_id(const struct tfe_stream *stream, long long *result, st if (strlen(dest_subscribe_id)) { - scan_ret = maat_scan_string((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), tfe_bussiness_tableid_get(PXY_CTRL_SUBSCRIBER_ID), + scan_ret = maat_scan_string(tfe_get_maat_handle(), tfe_bussiness_tableid_get(PXY_CTRL_SUBSCRIBER_ID), dest_subscribe_id, strlen(dest_subscribe_id),result + hit_cnt + hit_cnt_ip, MAX_SCAN_RESULT - hit_cnt - hit_cnt_ip,&n_hit_result, scan_mid); if (scan_ret == MAAT_SCAN_HIT) @@ -68,7 +68,7 @@ int tfe_scan_subscribe_id(const struct tfe_stream *stream, long long *result, st TFE_LOG_INFO(logger, "Scan dst TSG_OBJ_SUBSCRIBER_ID, NO hit subid: %s scan ret: %d addr: %s", dest_subscribe_id, scan_ret, stream->str_stream_info); } - scan_ret = maat_scan_not_logic((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), tfe_bussiness_tableid_get(PXY_CTRL_SUBSCRIBER_ID), + scan_ret = maat_scan_not_logic(tfe_get_maat_handle(), tfe_bussiness_tableid_get(PXY_CTRL_SUBSCRIBER_ID), result + hit_cnt + hit_cnt_ip, MAX_SCAN_RESULT - hit_cnt - hit_cnt_ip, &n_hit_result, scan_mid); if (scan_ret == MAAT_SCAN_HIT) { @@ -84,13 +84,13 @@ static int scan_group(struct maat_hit_group hit_group, long long *result, struct size_t n_hit_result=0; int scan_ret=0, hit_cnt_group=0; - scan_ret = maat_scan_group((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), table_id, &hit_group, 1, + scan_ret = maat_scan_group(tfe_get_maat_handle(), table_id, &hit_group, 1, result+hit_cnt+hit_cnt_group, MAX_SCAN_RESULT-hit_cnt-hit_cnt_group, &n_hit_result, scan_mid); if(scan_ret == MAAT_SCAN_HIT) { hit_cnt_group+=n_hit_result; } - scan_ret = maat_scan_not_logic((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), table_id, result+hit_cnt+hit_cnt_group, + scan_ret = maat_scan_not_logic(tfe_get_maat_handle(), table_id, result+hit_cnt+hit_cnt_group, MAX_SCAN_RESULT-hit_cnt-hit_cnt_group, &n_hit_result, scan_mid); if (scan_ret == MAAT_SCAN_HIT) { @@ -242,19 +242,19 @@ int tfe_scan_app_id(long long *result, struct maat_state *scan_mid, int hit_cnt, size_t n_hit_result = 0; struct maat_hit_group hit_group; - struct app_id_dict *app_dict = (struct app_id_dict*)maat_plugin_table_get_ex_data((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), tfe_bussiness_tableid_get(PXY_CTRL_APP_ID_DICT), + struct app_id_dict *app_dict = (struct app_id_dict*)maat_plugin_table_get_ex_data(tfe_get_maat_handle(), tfe_bussiness_tableid_get(PXY_CTRL_APP_ID_DICT), (const char *)&app_id, sizeof(long long)); if(app_dict!=NULL) { memset(&hit_group, 0, sizeof(hit_group)); hit_group.group_id=app_dict->group_id; - scan_ret = maat_scan_group((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), table_id, &hit_group, 1, result+hit_cnt+hit_app_id, + scan_ret = maat_scan_group(tfe_get_maat_handle(), table_id, &hit_group, 1, result+hit_cnt+hit_app_id, MAX_SCAN_RESULT-hit_cnt-hit_app_id, &n_hit_result, scan_mid); if(scan_ret==MAAT_SCAN_HIT) { hit_app_id += n_hit_result; } - scan_ret = maat_scan_not_logic((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), table_id, result+hit_cnt+hit_app_id, MAX_SCAN_RESULT-hit_cnt-hit_app_id, &n_hit_result, scan_mid); + scan_ret = maat_scan_not_logic(tfe_get_maat_handle(), table_id, result+hit_cnt+hit_app_id, MAX_SCAN_RESULT-hit_cnt-hit_app_id, &n_hit_result, scan_mid); if (scan_ret == MAAT_SCAN_HIT) { hit_app_id += n_hit_result; @@ -298,13 +298,13 @@ int tfe_scan_internal_exteral_addr(const struct tfe_stream *stream, long long *r if(n_last_hit_group > 0) { maat_state_get_last_hit_groups(scan_mid, last_hit_groups, array_size); - scan_ret = maat_scan_group((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), table_id, last_hit_groups, array_size, result+hit_cnt+hit_cnt_ip, + scan_ret = maat_scan_group(tfe_get_maat_handle(), table_id, last_hit_groups, array_size, result+hit_cnt+hit_cnt_ip, MAX_SCAN_RESULT-hit_cnt-hit_cnt_ip, &n_hit_result, scan_mid); if(scan_ret == MAAT_SCAN_HIT) { hit_cnt_ip += n_hit_result; } - scan_ret = maat_scan_not_logic((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), table_id, + scan_ret = maat_scan_not_logic(tfe_get_maat_handle(), table_id, result+hit_cnt+hit_cnt_ip, MAX_SCAN_RESULT-hit_cnt-hit_cnt_ip, &n_hit_result, scan_mid); if (scan_ret == MAAT_SCAN_HIT) { @@ -332,13 +332,13 @@ int tfe_scan_internal_exteral_port(const struct tfe_stream *stream, long long *r if(n_last_hit_group > 0) { maat_state_get_last_hit_groups(scan_mid, last_hit_groups, array_size); - scan_ret = maat_scan_group((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), table_id, last_hit_groups, array_size, result+hit_cnt+hit_cnt_port, + scan_ret = maat_scan_group(tfe_get_maat_handle(), table_id, last_hit_groups, array_size, result+hit_cnt+hit_cnt_port, MAX_SCAN_RESULT-hit_cnt-hit_cnt_port, &n_hit_result, scan_mid); if(scan_ret == MAAT_SCAN_HIT) { hit_cnt_port += n_hit_result; } - scan_ret = maat_scan_not_logic((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), table_id, + scan_ret = maat_scan_not_logic(tfe_get_maat_handle(), table_id, result+hit_cnt+hit_cnt_port, MAX_SCAN_RESULT-hit_cnt-hit_cnt_port, &n_hit_result, scan_mid); if (scan_ret == MAAT_SCAN_HIT) { @@ -354,7 +354,7 @@ int tfe_scan_port(const struct tfe_stream *stream, long long *result, struct maa int hit_cnt_port = 0; size_t n_hit_result = 0; - scan_ret=maat_scan_integer((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), tfe_bussiness_tableid_get(PXY_CTRL_SOURCE_PORT), ntohs(source), + scan_ret=maat_scan_integer(tfe_get_maat_handle(), tfe_bussiness_tableid_get(PXY_CTRL_SOURCE_PORT), ntohs(source), result+hit_cnt+hit_cnt_port, MAX_SCAN_RESULT-hit_cnt-hit_cnt_port, &n_hit_result, scan_mid); if(scan_ret == MAAT_SCAN_HIT) { @@ -365,14 +365,14 @@ int tfe_scan_port(const struct tfe_stream *stream, long long *result, struct maa { hit_cnt_port+=scan_ret; } - scan_ret = maat_scan_not_logic((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), tfe_bussiness_tableid_get(PXY_CTRL_SOURCE_PORT), + scan_ret = maat_scan_not_logic(tfe_get_maat_handle(), tfe_bussiness_tableid_get(PXY_CTRL_SOURCE_PORT), result+hit_cnt+hit_cnt_port, MAX_SCAN_RESULT-hit_cnt-hit_cnt_port, &n_hit_result, scan_mid); if (scan_ret == MAAT_SCAN_HIT) { hit_cnt_port+=n_hit_result; } - scan_ret=maat_scan_integer((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), tfe_bussiness_tableid_get(PXY_CTRL_DESTINATION_PORT), ntohs(dest), + scan_ret=maat_scan_integer(tfe_get_maat_handle(), tfe_bussiness_tableid_get(PXY_CTRL_DESTINATION_PORT), ntohs(dest), result+hit_cnt+hit_cnt_port, MAX_SCAN_RESULT-hit_cnt-hit_cnt_port, &n_hit_result, scan_mid); if(scan_ret == MAAT_SCAN_HIT) { @@ -383,7 +383,7 @@ int tfe_scan_port(const struct tfe_stream *stream, long long *result, struct maa { hit_cnt_port+=scan_ret; } - scan_ret = maat_scan_not_logic((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), tfe_bussiness_tableid_get(PXY_CTRL_DESTINATION_PORT), + scan_ret = maat_scan_not_logic(tfe_get_maat_handle(), tfe_bussiness_tableid_get(PXY_CTRL_DESTINATION_PORT), result+hit_cnt+hit_cnt_port, MAX_SCAN_RESULT-hit_cnt-hit_cnt_port, &n_hit_result, scan_mid); if (scan_ret == MAAT_SCAN_HIT) { @@ -403,20 +403,20 @@ int tfe_scan_ipv4_addr(const struct tfe_stream *stream, long long *result, struc memset(&hit_group, 0, sizeof(hit_group)); hit_group.group_id=PROTOCOL_TCP_GROUP_ID; - scan_ret = maat_scan_group((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), tfe_bussiness_tableid_get(PXY_CTRL_IP_PROTOCOL), &hit_group, 1, + scan_ret = maat_scan_group(tfe_get_maat_handle(), tfe_bussiness_tableid_get(PXY_CTRL_IP_PROTOCOL), &hit_group, 1, result+hit_cnt+hit_cnt_ip, MAX_SCAN_RESULT-hit_cnt-hit_cnt_ip, &n_hit_result, scan_mid); if(scan_ret==MAAT_SCAN_HIT) { hit_cnt_ip += n_hit_result; } - scan_ret = maat_scan_not_logic((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), tfe_bussiness_tableid_get(PXY_CTRL_IP_PROTOCOL), + scan_ret = maat_scan_not_logic(tfe_get_maat_handle(), tfe_bussiness_tableid_get(PXY_CTRL_IP_PROTOCOL), result+hit_cnt+hit_cnt_ip, MAX_SCAN_RESULT-hit_cnt-hit_cnt_ip, &n_hit_result, scan_mid); if (scan_ret == MAAT_SCAN_HIT) { hit_cnt_ip += n_hit_result; } - scan_ret = maat_scan_ipv4_port((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), tfe_bussiness_tableid_get(PXY_CTRL_SOURCE_IP), sapp_addr.v4->saddr, ntohs(sapp_addr.v4->source), + scan_ret = maat_scan_ipv4_port(tfe_get_maat_handle(), tfe_bussiness_tableid_get(PXY_CTRL_SOURCE_IP), sapp_addr.v4->saddr, ntohs(sapp_addr.v4->source), result+hit_cnt+hit_cnt_ip, MAX_SCAN_RESULT-hit_cnt-hit_cnt_ip, &n_hit_result, scan_mid); if (scan_ret == MAAT_SCAN_HIT) { @@ -427,14 +427,14 @@ int tfe_scan_ipv4_addr(const struct tfe_stream *stream, long long *result, struc { hit_cnt_ip += scan_ret; } - scan_ret = maat_scan_not_logic((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), tfe_bussiness_tableid_get(PXY_CTRL_SOURCE_IP), + scan_ret = maat_scan_not_logic(tfe_get_maat_handle(), tfe_bussiness_tableid_get(PXY_CTRL_SOURCE_IP), result+hit_cnt+hit_cnt_ip, MAX_SCAN_RESULT-hit_cnt-hit_cnt_ip, &n_hit_result, scan_mid); if (scan_ret == MAAT_SCAN_HIT) { hit_cnt_ip += n_hit_result; } - scan_ret = maat_scan_ipv4_port((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), tfe_bussiness_tableid_get(PXY_CTRL_DESTINATION_IP), sapp_addr.v4->daddr, ntohs(sapp_addr.v4->dest), + scan_ret = maat_scan_ipv4_port(tfe_get_maat_handle(), tfe_bussiness_tableid_get(PXY_CTRL_DESTINATION_IP), sapp_addr.v4->daddr, ntohs(sapp_addr.v4->dest), result+hit_cnt+hit_cnt_ip, MAX_SCAN_RESULT-hit_cnt-hit_cnt_ip, &n_hit_result, scan_mid); if(scan_ret == MAAT_SCAN_HIT) { @@ -445,7 +445,7 @@ int tfe_scan_ipv4_addr(const struct tfe_stream *stream, long long *result, struc { hit_cnt_ip += scan_ret; } - scan_ret = maat_scan_not_logic((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), tfe_bussiness_tableid_get(PXY_CTRL_DESTINATION_IP), + scan_ret = maat_scan_not_logic(tfe_get_maat_handle(), tfe_bussiness_tableid_get(PXY_CTRL_DESTINATION_IP), result+hit_cnt+hit_cnt_ip, MAX_SCAN_RESULT-hit_cnt-hit_cnt_ip, &n_hit_result, scan_mid); if (scan_ret == MAAT_SCAN_HIT) { @@ -464,19 +464,19 @@ int tfe_scan_ipv6_addr(const struct tfe_stream *stream, long long *result, struc memset(&hit_group, 0, sizeof(hit_group)); hit_group.group_id=PROTOCOL_TCP_GROUP_ID; - scan_ret = maat_scan_group((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), tfe_bussiness_tableid_get(PXY_CTRL_IP_PROTOCOL), &hit_group, 1, + scan_ret = maat_scan_group(tfe_get_maat_handle(), tfe_bussiness_tableid_get(PXY_CTRL_IP_PROTOCOL), &hit_group, 1, result+hit_cnt+hit_cnt_ip, MAX_SCAN_RESULT-hit_cnt-hit_cnt_ip, &n_hit_result, scan_mid); if(scan_ret==MAAT_SCAN_HIT) { hit_cnt_ip += n_hit_result; } - scan_ret = maat_scan_not_logic((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), tfe_bussiness_tableid_get(PXY_CTRL_IP_PROTOCOL), + scan_ret = maat_scan_not_logic(tfe_get_maat_handle(), tfe_bussiness_tableid_get(PXY_CTRL_IP_PROTOCOL), result+hit_cnt+hit_cnt_ip, MAX_SCAN_RESULT-hit_cnt-hit_cnt_ip, &n_hit_result, scan_mid); if (scan_ret == MAAT_SCAN_HIT) { hit_cnt_ip += n_hit_result; } - scan_ret = maat_scan_ipv6_port((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), tfe_bussiness_tableid_get(PXY_CTRL_SOURCE_IP), sapp_addr.v6->saddr, ntohs(sapp_addr.v6->source), + scan_ret = maat_scan_ipv6_port(tfe_get_maat_handle(), tfe_bussiness_tableid_get(PXY_CTRL_SOURCE_IP), sapp_addr.v6->saddr, ntohs(sapp_addr.v6->source), result+hit_cnt+hit_cnt_ip, MAX_SCAN_RESULT-hit_cnt-hit_cnt_ip, &n_hit_result, scan_mid); if (scan_ret == MAAT_SCAN_HIT) { @@ -487,14 +487,14 @@ int tfe_scan_ipv6_addr(const struct tfe_stream *stream, long long *result, struc { hit_cnt_ip += scan_ret; } - scan_ret = maat_scan_not_logic((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), tfe_bussiness_tableid_get(PXY_CTRL_SOURCE_IP), + scan_ret = maat_scan_not_logic(tfe_get_maat_handle(), tfe_bussiness_tableid_get(PXY_CTRL_SOURCE_IP), result+hit_cnt+hit_cnt_ip, MAX_SCAN_RESULT-hit_cnt-hit_cnt_ip, &n_hit_result, scan_mid); if (scan_ret == MAAT_SCAN_HIT) { hit_cnt_ip += n_hit_result; } - scan_ret = maat_scan_ipv6_port((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), tfe_bussiness_tableid_get(PXY_CTRL_DESTINATION_IP), sapp_addr.v6->daddr, ntohs(sapp_addr.v6->dest), + scan_ret = maat_scan_ipv6_port(tfe_get_maat_handle(), tfe_bussiness_tableid_get(PXY_CTRL_DESTINATION_IP), sapp_addr.v6->daddr, ntohs(sapp_addr.v6->dest), result+hit_cnt+hit_cnt_ip, MAX_SCAN_RESULT-hit_cnt-hit_cnt_ip, &n_hit_result, scan_mid); if (scan_ret == MAAT_SCAN_HIT) { @@ -505,7 +505,7 @@ int tfe_scan_ipv6_addr(const struct tfe_stream *stream, long long *result, struc { hit_cnt_ip += scan_ret; } - scan_ret = maat_scan_not_logic((struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT), tfe_bussiness_tableid_get(PXY_CTRL_DESTINATION_IP), + scan_ret = maat_scan_not_logic(tfe_get_maat_handle(), tfe_bussiness_tableid_get(PXY_CTRL_DESTINATION_IP), result+hit_cnt+hit_cnt_ip, MAX_SCAN_RESULT-hit_cnt-hit_cnt_ip, &n_hit_result, scan_mid); if (scan_ret == MAAT_SCAN_HIT) { diff --git a/conf/tfe/tfe.conf b/conf/tfe/tfe.conf index 4d56b0b..4f13b38 100644 --- a/conf/tfe/tfe.conf +++ b/conf/tfe/tfe.conf @@ -21,6 +21,11 @@ cpu_affinity_mask=1-9 # LEAST_CONN = 0; ROUND_ROBIN = 1 load_balance=1 +[public] +vsys_id=1 +data_center=center-xxg-tsgx +device_group=group-xxg-tsgx +device_id=9800165603247024 # for enable kni v3 [nfq] @@ -97,9 +102,6 @@ trusted_cert_dir=resource/tfe/trusted_storage log_master_key=0 key_log_file=log/sslkeylog.log -# mid cert cache -mc_cache_enable=1 - [key_keeper] #Mode: debug - generate cert with ca_path, normal - generate cert with cert store #0 on cache 1 off cache @@ -184,16 +186,13 @@ http_keepalive_addr=192.168.41.60 http_keepalive_port=9273 [kafka] -enable=1 -vsystem_id=1 -NIC_NAME=enp2s0 -kafka_brokerlist=192.168.40.224:9092 -logger_send_topic=PROXY-EVENT -file_bucket_topic=TRAFFIC-FILE-STREAM-RECORD -mc_cache_topic=PXY-EXCH-INTERMEDIA-CERT +brokerlist=192.168.40.224:9092 sasl_username=admin sasl_passwd=galaxy2019 -device_id_filepath=/opt/tsg/etc/tsg_sn.json +rule_hits_topic=POLICY-RULE-METRIC +proxy_event_topic=PROXY-EVENT +file_stream_topic=TRAFFIC-FILE-STREAM-RECORD +exch_cert_topic=PXY-EXCH-INTERMEDIA-CERT [maat] # 0:json 1:redis 2:iris @@ -202,7 +201,6 @@ stat_switch=1 perf_switch=1 table_info=resource/pangu/table_info.conf accept_path=/opt/tsg/etc/tsg_device_tag.json -accept_tag_key=device_id stat_file=log/pangu_scan.fs2 effect_interval_s=1 deferred_load_on=0 diff --git a/platform/include/internal/ssl_fetch_cert.h b/platform/include/internal/ssl_fetch_cert.h index cd6d16a..594b3fb 100644 --- a/platform/include/internal/ssl_fetch_cert.h +++ b/platform/include/internal/ssl_fetch_cert.h @@ -1,12 +1,6 @@ -// -// Created by lwp on 2019/10/16. -// - #ifndef TFE_SSL_FETCH_CERT_H #define TFE_SSL_FETCH_CERT_H -// return 0 for success, return -1 for failed -int ssl_mid_cert_kafka_logger_create(const char *profile, const char *section); void ssl_fetch_trusted_cert_from_chain(STACK_OF(X509) *cert_chain, X509_STORE *trusted_store, const char *hostname); -#endif //TFE_SSL_FETCH_CERT_H \ No newline at end of file +#endif \ No newline at end of file diff --git a/platform/src/acceptor_kni_v4.cpp b/platform/src/acceptor_kni_v4.cpp index 49ad2ba..ab73596 100644 --- a/platform/src/acceptor_kni_v4.cpp +++ b/platform/src/acceptor_kni_v4.cpp @@ -171,7 +171,7 @@ static void *worker_thread_cycle(void *arg) io_uring_set_read_cb(io_uring_on_tap_s, handle_decryption_packet_from_tap, thread_ctx); } - TFE_LOG_INFO(logger, "%s: worker thread %d is running", "LOG_TAG_KNI", thread_index); + TFE_LOG_INFO(logger, "worker thread %d is running", thread_index); while (1) { @@ -216,7 +216,7 @@ static void *worker_thread_cycle(void *arg) } error_out: - TFE_LOG_ERROR(logger, "%s: worker thread %d exiting", LOG_TAG_SCE, thread_index); + TFE_LOG_ERROR(logger, "worker thread %d exiting", thread_index); return (void *)NULL; } diff --git a/platform/src/proxy.cpp b/platform/src/proxy.cpp index ddfe5c4..746ba29 100644 --- a/platform/src/proxy.cpp +++ b/platform/src/proxy.cpp @@ -272,7 +272,7 @@ static void __signal_handler_cb(evutil_socket_t fd, short what, void * arg) TFE_LOG_ERROR(ctx->logger, "recv SIGHUP, reload zlog.conf"); MESA_handle_runtime_log_reconstruction(NULL); MESA_load_profile_int_def(profile_path, "maat", "log_level", &(log_level), LOG_LEVEL_FATAL); - maat = (struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT); + maat = tfe_get_maat_handle(); if(maat) { maat_reload_log_level(maat, (enum log_level)log_level); @@ -688,7 +688,7 @@ int main(int argc, char * argv[]) CHECK_OR_EXIT(g_default_proxy->key_keeper_handler, "Failed at init Key keeper. Exit."); /* RESOURCE INIT */ - ret = tfe_bussiness_resouce_init(); + ret = tfe_env_init(); CHECK_OR_EXIT(ret == 0, "TFE bussiness resource init failed. Exit."); /* SSL INIT */ diff --git a/platform/src/ssl_fetch_cert.cpp b/platform/src/ssl_fetch_cert.cpp index 7a4a0de..a206260 100644 --- a/platform/src/ssl_fetch_cert.cpp +++ b/platform/src/ssl_fetch_cert.cpp @@ -1,12 +1,9 @@ -// -// Created by lwp on 2019/10/16. -// - #include #include +#include "kafka.h" +#include "tfe_utils.h" #include -#include #include #include @@ -36,51 +33,8 @@ static char cert_type_desc[MAX_TYPE][64] = { {"Root certificate"}, }; -struct ssl_mid_cert_ctx +static void send_cert_to_kafka(const char *sni, const char *fingerprint, const char *cert) { - int enable; - tfe_kafka_logger_t *g_kafka_logger; -}; -struct ssl_mid_cert_ctx mid_cert_ctx; - -int ssl_mid_cert_kafka_logger_create(const char *profile, const char *section) -{ - int ret=0; - char topic_name[TFE_SYMBOL_MAX] = {0}; - char sasl_username[TFE_STRING_MAX] = {0}; - char sasl_passwd[TFE_STRING_MAX] = {0}; - - MESA_load_profile_int_def(profile, section, "mc_cache_enable", &mid_cert_ctx.enable, 0); - MESA_load_profile_string_def(profile, "kafka", "mc_cache_topic", topic_name, sizeof(topic_name), "PXY-EXCH-INTERMEDIA-CERT"); - MESA_load_profile_string_def(profile, "kafka", "SASL_USERNAME", sasl_username, sizeof(sasl_username), ""); - MESA_load_profile_string_def(profile, "kafka", "SASL_PASSWD", sasl_passwd, sizeof(sasl_passwd), ""); - - if(mid_cert_ctx.enable == 0) - { - return 0; - } - - mid_cert_ctx.g_kafka_logger = (tfe_kafka_logger_t *)tfe_bussiness_resouce_get(KAFKA_LOGGER); - if(!mid_cert_ctx.g_kafka_logger) - { - return -1; - } - - ret = tfe_logger_create_kafka_topic(mid_cert_ctx.g_kafka_logger, sasl_username, sasl_passwd, topic_name, TOPIC_MC_CACHE, g_default_logger); - if(ret < 0) - { - return -1; - } - return 0; -} - -static void ssl_mid_cert_kafka_logger_send(const char *sni, const char *fingerprint, const char *cert) -{ - if (mid_cert_ctx.g_kafka_logger->enable == 0) - { - return; - } - cJSON *obj = NULL; cJSON *dup = NULL; char *msg = NULL; @@ -88,13 +42,12 @@ static void ssl_mid_cert_kafka_logger_send(const char *sni, const char *fingerpr obj = cJSON_CreateObject(); cJSON_AddStringToObject(obj, "sni", sni); cJSON_AddStringToObject(obj, "fingerprint", fingerprint); - cJSON_AddNumberToObject(obj, "vsys_id", mid_cert_ctx.g_kafka_logger->t_vsys_id); + cJSON_AddNumberToObject(obj, "vsys_id", tfe_get_vsys_id()); cJSON_AddStringToObject(obj, "cert", cert); - cJSON_AddStringToObject(obj, "tfe_ip", mid_cert_ctx.g_kafka_logger->local_ip_str); + cJSON_AddStringToObject(obj, "tfe_ip", tfe_get_sled_ip()); dup = cJSON_Duplicate(obj, 1); msg = cJSON_PrintUnformatted(dup); - TFE_LOG_DEBUG(g_default_logger, "log to [%s] msg:%s", mid_cert_ctx.g_kafka_logger->topic_name[TOPIC_MC_CACHE], msg); - tfe_kafka_logger_send(mid_cert_ctx.g_kafka_logger, TOPIC_MC_CACHE, msg, strlen(msg)); + kafka_send(tfe_get_kafka_handle(), TOPIC_EXCH_CERT, msg, strlen(msg)); free(msg); cJSON_Delete(dup); @@ -142,10 +95,6 @@ void ssl_fetch_trusted_cert_from_chain(STACK_OF(X509) * cert_chain, X509_STORE * char *fingerprint = NULL; X509 *cert = NULL; X509_OBJECT *obj = NULL; - if (!mid_cert_ctx.g_kafka_logger || !mid_cert_ctx.enable) - { - return; - } deep = sk_X509_num(cert_chain); for (int i = 0; i < deep; i++) @@ -197,13 +146,12 @@ void ssl_fetch_trusted_cert_from_chain(STACK_OF(X509) * cert_chain, X509_STORE * if (!in_store && fingerprint && pem) { - ssl_mid_cert_kafka_logger_send(hostname, fingerprint, pem); + send_cert_to_kafka(hostname, fingerprint, pem); } end: - TFE_LOG_DEBUG(g_default_logger, "[dep:%d/%d] is %s, in_trusted_store:%d, sin:%s; subject:(%s); issuer:(%s); fingerprint:%s; cert:%s", - i, deep, cert_type_desc[type], in_store, (hostname ? hostname : "NULL"), (subj ? subj : "NULL"), (issuer ? issuer : "NULL"), (fingerprint ? fingerprint : "NULL"), - ((pem && mid_cert_ctx.g_kafka_logger->enable == 0x10) ? pem : " ...")); + TFE_LOG_DEBUG(g_default_logger, "[dep:%d/%d] is %s, in_trusted_store:%d, sin:%s; subject:(%s); issuer:(%s); fingerprint:%s", + i, deep, cert_type_desc[type], in_store, (hostname ? hostname : "NULL"), (subj ? subj : "NULL"), (issuer ? issuer : "NULL"), (fingerprint ? fingerprint : "NULL")); if (pem) { free(pem); diff --git a/platform/src/ssl_service_cache.cpp b/platform/src/ssl_service_cache.cpp index 0561f43..13c7adc 100644 --- a/platform/src/ssl_service_cache.cpp +++ b/platform/src/ssl_service_cache.cpp @@ -3,12 +3,6 @@ #include #include -struct ssl_ja3_enforcer -{ - struct maat *maat; - int table_id; -}; - struct ssl_svc_ja3 { char ja3_hash[33]; @@ -26,7 +20,7 @@ struct ssl_svc_addr const char *dport; }; -static struct ssl_ja3_enforcer g_static_enforcer = {0}; +static int table_id = 0; static void ssl_svc_ja3_param_dup_cb(int table_id, void **to, void **from, long argl, void *argp) { @@ -93,24 +87,23 @@ static void ssl_svc_ja3_param_free(struct ssl_svc_ja3 *param) static int ssl_svc_ja3_init(const char *table_name) { - g_static_enforcer.maat = (struct maat *)tfe_bussiness_resouce_get(STATIC_MAAT); - g_static_enforcer.table_id = maat_get_table_id(g_static_enforcer.maat, table_name); - if (g_static_enforcer.table_id < 0) + table_id = maat_get_table_id(tfe_get_maat_handle(), table_name); + if (table_id < 0) { TFE_LOG_ERROR(g_default_logger, "Maat table %s register failed.", table_name); return 0; } - int ret = maat_plugin_table_ex_schema_register(g_static_enforcer.maat, + int ret = maat_plugin_table_ex_schema_register(tfe_get_maat_handle(), table_name, ssl_svc_ja3_param_new_cb, ssl_svc_ja3_param_free_cb, ssl_svc_ja3_param_dup_cb, 0, - &g_static_enforcer); + NULL); if (ret < 0) { TFE_LOG_ERROR(g_default_logger, "failed at Maat_plugin_EX_register(%s), table_id = %d, ret = %d", - table_name, g_static_enforcer.table_id, ret); + table_name, table_id, ret); return 0; } @@ -122,7 +115,7 @@ enum ssl_ja3_pinning_status ssl_svc_ja3_scan(char *ja3_hash, const char *addr_st enum ssl_ja3_pinning_status ret = JA3_PINNING_STATUS_UNKNOWN; struct ssl_svc_ja3 *param = NULL; - param = (struct ssl_svc_ja3 *)maat_plugin_table_get_ex_data(g_static_enforcer.maat, g_static_enforcer.table_id, ja3_hash, strlen(ja3_hash)); + param = (struct ssl_svc_ja3 *)maat_plugin_table_get_ex_data(tfe_get_maat_handle(), table_id, ja3_hash, strlen(ja3_hash)); if (param == NULL) { ret = JA3_PINNING_STATUS_UNKNOWN; diff --git a/platform/src/ssl_stream.cpp b/platform/src/ssl_stream.cpp index 46fb62a..fbbf6d3 100644 --- a/platform/src/ssl_stream.cpp +++ b/platform/src/ssl_stream.cpp @@ -636,11 +636,6 @@ struct ssl_mgr * ssl_manager_init(const char * ini_profile, const char * section mgr->logger = logger; mgr->ev_base_gc=ev_base_gc; - if (ssl_mid_cert_kafka_logger_create(ini_profile, section)) - { - goto error_out; - } - MESA_load_profile_uint_def(ini_profile, section, "ssl_debug", &(ssl_debug), 0); MESA_load_profile_string_def(ini_profile, section, "ssl_min_version", version_str, sizeof(version_str), "ssl3"); mgr->ssl_min_version = sslver_str2num(version_str); diff --git a/plugin/business/chaining-policy/src/chaining_policy.cpp b/plugin/business/chaining-policy/src/chaining_policy.cpp index ffd3852..1e10910 100644 --- a/plugin/business/chaining-policy/src/chaining_policy.cpp +++ b/plugin/business/chaining-policy/src/chaining_policy.cpp @@ -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) diff --git a/plugin/business/doh/src/doh.cpp b/plugin/business/doh/src/doh.cpp index 4103f69..15032b4 100644 --- a/plugin/business/doh/src/doh.cpp +++ b/plugin/business/doh/src/doh.cpp @@ -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; diff --git a/plugin/business/doh/src/logger.cpp b/plugin/business/doh/src/logger.cpp index a2567c0..279915c 100644 --- a/plugin/business/doh/src/logger.cpp +++ b/plugin/business/doh/src/logger.cpp @@ -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); diff --git a/plugin/business/doh/src/pub.h b/plugin/business/doh/src/pub.h index 491bfe2..a0b7d5b 100644 --- a/plugin/business/doh/src/pub.h +++ b/plugin/business/doh/src/pub.h @@ -13,7 +13,6 @@ extern "C" #include #include #include -#include #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]; diff --git a/plugin/business/ssl-policy/src/ssl_policy.cpp b/plugin/business/ssl-policy/src/ssl_policy.cpp index 625ba48..93dc152 100644 --- a/plugin/business/ssl-policy/src/ssl_policy.cpp +++ b/plugin/business/ssl-policy/src/ssl_policy.cpp @@ -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); diff --git a/plugin/business/tcp-policy/src/tcp_policy.cpp b/plugin/business/tcp-policy/src/tcp_policy.cpp index d5360b0..bba0297 100644 --- a/plugin/business/tcp-policy/src/tcp_policy.cpp +++ b/plugin/business/tcp-policy/src/tcp_policy.cpp @@ -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) diff --git a/plugin/business/tsg-http/src/tsg_http.cpp b/plugin/business/tsg-http/src/tsg_http.cpp index d07d529..e30d25a 100644 --- a/plugin/business/tsg-http/src/tsg_http.cpp +++ b/plugin/business/tsg-http/src/tsg_http.cpp @@ -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++) { diff --git a/plugin/business/tsg-http/src/tsg_logger.cpp b/plugin/business/tsg-http/src/tsg_logger.cpp index 51cefe9..348e011 100644 --- a/plugin/business/tsg-http/src/tsg_logger.cpp +++ b/plugin/business/tsg-http/src/tsg_logger.cpp @@ -1,12 +1,12 @@ #include #include -#include #include #include #include #include +#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;ilocal_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);