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

TSG-935 重构 kafka log 接口

Browse Source
This commit is contained in:
luwenpeng
2020-03-09 16:28:56 +08:00
parent 12e8003d29
commit 8079c4142f
5 changed files with 237 additions and 240 deletions
Download Patch File Download Diff File Expand all files Collapse all files

114
plugin/business/pangu-http/src/pangu_logger.cpp
View File

@@ -1,19 +1,6 @@
#include <cjson/cJSON.h>
#include <librdkafka/rdkafka.h>
#include <MESA/MESA_handle_logger.h>
#include <MESA/MESA_prof_load.h>
#include <assert.h>
#include <arpa/inet.h>
#include <time.h>
#include <stdio.h>
#include <unistd.h>
#include <sys/ioctl.h>
#include <net/if.h>
#include <pthread.h>
#include <errno.h>
#include <tfe_utils.h>
#include <tfe_kafka_logger.h>
#include <cache_evbase_client.h>
#include "pangu_logger.h"
@@ -25,27 +12,20 @@ struct json_spec
};
struct pangu_logger
{
char local_ip_str[TFE_SYMBOL_MAX];
int entry_id;
unsigned int en_sendlog;
unsigned int en_sendlog_meta;
unsigned int en_sendlog_body;
unsigned int local_ip_nr;
void* global_logger;
rd_kafka_t *kafka_handle;
rd_kafka_topic_t* kafka_topic;
pthread_mutex_t mutex;
char brokerlist[TFE_STRING_MAX];
char topic_name[TFE_STRING_MAX];
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;
};
@@ -63,65 +43,12 @@ enum _log_action //Bigger action number is prior.
__LG_ACTION_MAX
};
static unsigned int get_ip_by_eth_name(const char *ifname)
{
int sockfd;
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) < 0)
{
goto error;
}
ip = ((struct sockaddr_in*)&(ifr.ifr_addr))->sin_addr.s_addr;
close(sockfd);
return ip;
error:
close(sockfd);
return INADDR_NONE;
}
static rd_kafka_t * create_kafka_handle(const char* brokerlist)
{
char kafka_errstr[1024];
rd_kafka_t *handle=NULL;
rd_kafka_conf_t *rdkafka_conf = NULL;
rdkafka_conf = rd_kafka_conf_new();
rd_kafka_conf_set(rdkafka_conf, "queue.buffering.max.messages", "1000000", kafka_errstr, sizeof(kafka_errstr));
rd_kafka_conf_set(rdkafka_conf, "topic.metadata.refresh.interval.ms", "600000",kafka_errstr, sizeof(kafka_errstr));
rd_kafka_conf_set(rdkafka_conf, "security.protocol", "MG", kafka_errstr, sizeof(kafka_errstr));
//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, rdkafka_conf, kafka_errstr, sizeof(kafka_errstr));
rdkafka_conf=NULL;
if (handle==NULL)
{
return NULL;
}
if (rd_kafka_brokers_add(handle, brokerlist) == 0)
{
rd_kafka_destroy(handle);
return NULL;
}
return handle;
}
struct pangu_logger* pangu_log_handle_create(const char* profile, const char* section, void* local_logger)
{
int ret=-1;
char nic_name[64]={0};
char brokerlist[TFE_STRING_MAX] = { 0 };
char topic_name[TFE_STRING_MAX] = { 0 };
struct tango_cache_parameter *log_file_upload_para=NULL;
struct pangu_logger* instance=ALLOC(struct pangu_logger,1);
@@ -149,39 +76,27 @@ struct pangu_logger* pangu_log_handle_create(const char* profile, const char* s
}
MESA_load_profile_string_def(profile, section, "NIC_NAME",nic_name,sizeof(nic_name),"eth0");
instance->local_ip_nr=get_ip_by_eth_name(nic_name);
if(instance->local_ip_nr==INADDR_NONE)
{
TFE_LOG_ERROR(local_logger, "%s get NIC_NAME: %s error.", __FUNCTION__, nic_name);
goto error_out;
}
inet_ntop(AF_INET,&(instance->local_ip_nr),instance->local_ip_str,sizeof(instance->local_ip_str));
MESA_load_profile_int_def(profile, section, "ENTRANCE_ID",&(instance->entry_id),0);
ret=MESA_load_profile_string_def(profile, section,"KAFKA_BROKERLIST", instance->brokerlist, sizeof(instance->brokerlist), NULL);
ret=MESA_load_profile_string_def(profile, section,"KAFKA_BROKERLIST", brokerlist, sizeof(brokerlist), NULL);
if(ret<0)
{
TFE_LOG_ERROR(local_logger,"Pangu log init failed, no brokerlist in profile %s section %s.", profile, section);
goto error_out;
}
MESA_load_profile_string_def(profile, section,"KAFKA_TOPIC", topic_name, sizeof(topic_name), "POLICY-EVENT-LOG");
instance->kafka_handle=create_kafka_handle(instance->brokerlist);
if(instance->kafka_handle==NULL)
TFE_LOG_INFO(local_logger, "Pangu kafka brokerlist : %s", brokerlist);
TFE_LOG_INFO(local_logger, "Pangu kafka topic : %s", topic_name);
instance->kafka_logger = tfe_kafka_logger_create(instance->en_sendlog, nic_name, brokerlist, topic_name, local_logger);
if (instance->kafka_logger)
{
TFE_LOG_ERROR(local_logger,"Pangu log init failed. Cannot create lafka handle with brokerlist: %s.", instance->brokerlist);
TFE_LOG_ERROR(local_logger,"Pangu log init failed, error to create kafka logger.");
goto error_out;
}
MESA_load_profile_string_def(profile, section,"KAFKA_TOPIC", instance->topic_name, sizeof(instance->topic_name), "POLICY-EVENT-LOG");
TFE_LOG_INFO(local_logger, "Pangu kafka brokerlist : %s", instance->brokerlist);
TFE_LOG_INFO(local_logger, "Pangu kafka topic : %s", instance->topic_name);
instance->kafka_topic = rd_kafka_topic_new(instance->kafka_handle,instance->topic_name, NULL);
log_file_upload_para=cache_evbase_parameter_new(profile, section, local_logger);
instance->log_file_upload_instance=cache_evbase_instance_new(log_file_upload_para, local_logger);
pthread_mutex_init(&(instance->mutex), NULL);
return instance;
error_out:
@@ -290,7 +205,7 @@ int pangu_send_log(struct pangu_logger* handle, const struct pangu_log* log_msg)
cJSON_AddNumberToObject(common_obj, "common_direction", 0); //0：域内->域外，1：域外->域内，描述的是CLIENT_IP信息
cJSON_AddNumberToObject(common_obj, "common_link_id", 0);
cJSON_AddNumberToObject(common_obj, "common_stream_dir", 3); //1:c2s, 2:s2c, 3:double
cJSON_AddStringToObject(common_obj, "common_sled_ip", handle->local_ip_str);
cJSON_AddStringToObject(common_obj, "common_sled_ip", handle->kafka_logger->local_ip_str);
cJSON_AddNumberToObject(common_obj, "common_entrance_id", handle->entry_id);
cJSON_AddNumberToObject(common_obj, "common_device_id", 0);
cJSON_AddStringToObject(common_obj, "http_url", http->req->req_spec.url);
@@ -398,8 +313,7 @@ int pangu_send_log(struct pangu_logger* handle, const struct pangu_log* log_msg)
TFE_LOG_DEBUG(handle->local_logger, "%s", log_payload);
kafka_status = rd_kafka_produce(handle->kafka_topic, RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_COPY,
log_payload, strlen(log_payload), NULL, 0, NULL);
kafka_status = tfe_kafka_logger_send(handle->kafka_logger, log_payload, strlen(log_payload));
free(log_payload);
cJSON_Delete(per_hit_obj);
if(kafka_status<0)