tango-tfe/plugin/business/pangu-http/pangu_logger.cpp

#include "pangu_logger.h"
#include <tfe_utils.h>
#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>

struct json_spec
{
	int json_type;
	const char *name;
};

struct pangu_logger
{
	char local_ip_str[TFE_SYMBOL_MAX];
	int entry_id;
	
	unsigned int local_ip_nr;
	void* global_logger;
	rd_kafka_t *kafka_handle;
	rd_kafka_topic_t* kafka_topic;
	
	char brokerlist[TFE_STRING_MAX];
	struct json_spec opt2json[LOG_OPT_MAX];
	const char* topic_name;

	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];
};



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)
{
	int i = 0;	
	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,i=0;
	char addr_string[TFE_SYMBOL_MAX]={0},local_msg_dir[TFE_STRING_MAX]={0};
	char nic_name[64]={0};
	unsigned int ip_buff[TFE_SYMBOL_MAX];

	struct pangu_logger* instance=ALLOC(struct pangu_logger,1);
	instance->global_logger=local_logger;
	
	instance->opt2json[LOG_OPT_HTTP_C2S_ISN] = {cJSON_Number,"isn"};
	instance->opt2json[LOG_OPT_HTTP_PROXY_FLAG] = {cJSON_Number,"proxy_flag"};
	instance->opt2json[LOG_OPT_HTTP_URL] = {cJSON_String,"url"};
	instance->opt2json[LOG_OPT_HTTP_COOKIE] = {cJSON_String,"cookie"};
	instance->opt2json[LOG_OPT_HTTP_REFERER] = {cJSON_String,"referer"};
	instance->opt2json[LOG_OPT_HTTP_UA] = {cJSON_String,"user_agent"};
	instance->opt2json[LOG_OPT_HTTP_REQ_LINE] = {cJSON_String,"req_line"};
	instance->opt2json[LOG_OPT_HTTP_RES_LINE] = {cJSON_String,"res_line"};
	instance->opt2json[LOG_OPT_HTTP_SET_COOKIE] = {cJSON_String,"set_cookie"};
	instance->opt2json[LOG_OPT_HTTP_CONTENT_TYPE] = {cJSON_String,"content_type"};
	instance->opt2json[LOG_OPT_HTTP_CONTENT_LEN] = {cJSON_String,"content_len"};

	TFE_LOG_ERROR(local_logger,"Pangu log is inititating from %s section %s.", profile, section);
	
	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);
	if(ret<0)
	{
		TFE_LOG_ERROR(local_logger,"Pangu log init failed, no brokerlist in profile %s section %s.", profile, section);
		goto error_out;
	}
	instance->kafka_handle=create_kafka_handle(instance->brokerlist);
	if(instance->kafka_handle==NULL)
	{
		TFE_LOG_ERROR(local_logger,"Pangu log init failed. Cannot create lafka handle with brokerlist: %s.", instance->brokerlist);	
		goto error_out;
	}
	instance->topic_name="PXY_HTTP_LOG";
	instance->kafka_topic = rd_kafka_topic_new(instance->kafka_handle,instance->topic_name, NULL);
	
	return instance;
	
error_out:
	free(instance);
	return NULL;
}
int pangu_send_log(struct pangu_logger* logger, const struct pangu_log* log_msg,struct opt_unit* log_opt,int opt_num)
{

}