tango-tfe/platform/src/proxy.cpp

/*
 * Proxy engine, built around libevent 2.x.
 */

#include <sys/types.h>
#include <sys/socket.h>
#include <sys/prctl.h>
#include <netinet/in.h>
#include <sys/un.h>
#include <sys/stat.h>
#include <assert.h>
#include <signal.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <errno.h>
#include <pthread.h>
#include <getopt.h>
#include <libgen.h>

#include <event2/event.h>
#include <event2/dns.h>
#include <event2/listener.h>
#include <event2/bufferevent.h>
#include <event2/bufferevent_ssl.h>
#include <event2/buffer.h>
#include <event2/thread.h>

#include <MESA/MESA_handle_logger.h>
#include <MESA/MESA_prof_load.h>
#include <MESA/field_stat2.h>

#include <tango_cache_client.h>
#include <tfe_utils.h>
#include <tfe_future.h>
#include <tfe_stream.h>
#include <tfe_proxy.h>
#include <tfe_plugin.h>
#include <tfe_cmsg.h>

#include <platform.h>
#include <proxy.h>
#include <tcp_stream.h>
#include <acceptor_kni_v1.h>
#include <acceptor_kni_v2.h>
#include <watchdog_kni.h>
#include <key_keeper.h>

/* Breakpad */
#include <client/linux/handler/exception_handler.h>
#include <common/linux/http_upload.h>

/* Systemd */
#include <systemd/sd-daemon.h>

extern struct ssl_policy_enforcer* ssl_policy_enforcer_create(void* logger);
extern enum ssl_stream_action ssl_policy_enforce(struct ssl_stream *upstream, void* u_para);

static int signals[] = {SIGHUP, SIGPIPE, SIGUSR1};

/* Global Resource */
void * g_default_logger = NULL;
struct tfe_proxy * g_default_proxy = NULL;
bool g_print_to_stderr = true;

/* Per thread resource */
thread_local unsigned int __currect_thread_id = 0;
thread_local void * __currect_default_logger = NULL;

#define TFE_VAR_VERSION_CATTER(v) __attribute__((__used__)) const char * TFE_VERSION_version_##v = NULL
#define TFE_VAR_VERSION_EXPEND(v) TFE_VAR_VERSION_CATTER(v)

extern "C"
{
/* VERSION TAG */
#ifdef TFE_VAR_VERSION
TFE_VAR_VERSION_EXPEND(TFE_VAR_VERSION);
#else
static __attribute__((__used__)) const char * TFE_VERSION_version_UNKNOWN = NULL;
#endif
#undef TFE_VAR_VERSION_CATTER
#undef TFE_VAR_VERSION_EXPEND
}

/* VERSION STRING */
#ifdef TFE_GIT_VERSION
static __attribute__((__used__)) const char * __tfe_version = TFE_GIT_VERSION;
#else
static __attribute__((__used__)) const char * tfe_version = "Unknown";
#endif

struct tfe_thread_ctx * tfe_proxy_thread_ctx_acquire(struct tfe_proxy * ctx)
{
	unsigned int min_thread_id = 0;
	static unsigned int counter=0;
	counter++;
	/*
	for (unsigned int tid = 0; tid < ctx->nr_work_threads; tid++)
	{
		struct tfe_thread_ctx * thread_ctx = ctx->work_threads[tid];
		unsigned int thread_load = ATOMIC_READ(&thread_ctx->load);

		min_thread_id = min_load > thread_load ? tid : min_thread_id;
		min_load = min_load > thread_load ? thread_load : min_load;
	}
*/
	min_thread_id=counter%ctx->nr_work_threads;
	ATOMIC_INC(&ctx->work_threads[min_thread_id]->load);
	return ctx->work_threads[min_thread_id];
}

void tfe_proxy_thread_ctx_release(struct tfe_thread_ctx * thread_ctx)
{
	ATOMIC_DEC(&thread_ctx->load);
}

/* 检查本进程是否通过SYSTEMD启动 */
static int check_is_started_by_notify()
{
	char * notify_socket = getenv("NOTIFY_SOCKET");
	return notify_socket == NULL ? 0 : 1;
}

int tfe_proxy_fds_accept(struct tfe_proxy * ctx, int fd_downstream, int fd_upstream, struct tfe_cmsg * cmsg)
{
	struct tfe_thread_ctx * worker_thread_ctx = tfe_proxy_thread_ctx_acquire(ctx);
	struct tfe_stream * stream = tfe_stream_create(ctx, worker_thread_ctx);

    enum tfe_stream_proto stream_protocol;
    uint8_t stream_protocol_in_char = 0;
	uint16_t size = 0;

	int result = tfe_cmsg_get_value(cmsg, TFE_CMSG_TCP_RESTORE_PROTOCOL, (unsigned char *)&stream_protocol_in_char,
        sizeof(stream_protocol_in_char), &size);

	if (unlikely(result < 0))
    {
	    TFE_LOG_ERROR(ctx->logger, "failed at fetch connection's protocol from cmsg: %s", strerror(-result));
	    goto __errout;
    }

    stream_protocol = (enum tfe_stream_proto)stream_protocol_in_char;
	tfe_stream_option_set(stream, TFE_STREAM_OPT_SESSION_TYPE, &stream_protocol, sizeof(stream_protocol));
	tfe_stream_cmsg_setup(stream, cmsg);

	/* FOR DEBUG */
	if (unlikely(ctx->tcp_all_passthrough))
	{
		bool __true = true;
		enum tfe_stream_proto __session_type = STREAM_PROTO_PLAIN;

		tfe_stream_option_set(stream, TFE_STREAM_OPT_PASSTHROUGH, &__true, sizeof(__true));
		tfe_stream_option_set(stream, TFE_STREAM_OPT_SESSION_TYPE, &__session_type, sizeof(__session_type));
	}

	result = tfe_stream_init_by_fds(stream, fd_downstream, fd_upstream);
	if (result < 0)
	{
		TFE_LOG_ERROR(ctx->logger, "%p, Fds(downstream = %d, upstream = %d, type = %d) accept failed.",
			stream, fd_downstream, fd_upstream, stream_protocol); goto __errout;
	}
	else
	{
		TFE_LOG_DEBUG(ctx->logger, "%p, Fds(downstream = %d, upstream = %d, type = %d) accepted.",
			stream, fd_downstream, fd_upstream, stream_protocol);
	}

	return 0;

__errout:
	if(stream != NULL) tfe_stream_destory((struct tfe_stream_private *)stream);
	return -1;
}

void tfe_proxy_loopbreak(tfe_proxy * ctx)
{
	event_base_loopbreak(ctx->evbase);
}

void tfe_proxy_free(tfe_proxy * ctx)
{
	return;
}

static void __dummy_event_handler(evutil_socket_t fd, short what, void * arg)
{
	//printf("%s alive\n",__FUNCTION__);
	return;
}

static void __signal_handler_cb(evutil_socket_t fd, short what, void * arg)
{
	tfe_proxy * ctx = (tfe_proxy *) arg;
	switch (fd)
	{
		case SIGTERM:
		case SIGQUIT:
		case SIGHUP: break;
		case SIGUSR1: break;
		case SIGPIPE:
			TFE_PROXY_STAT_INCREASE(STAT_SIGPIPE, 1);
			TFE_LOG_ERROR(ctx->logger, "Warning: Received SIGPIPE; ignoring.\n");
			break;
		default: TFE_LOG_ERROR(ctx->logger, "Warning: Received unexpected signal %i\n", fd);
			break;
	}
}

static void __gc_handler_cb(evutil_socket_t fd, short what, void * arg)
{
	tfe_proxy * ctx = (tfe_proxy *) arg;
	int i = 0;
	for (i = 0; i < TFE_STAT_MAX; i++)
	{
		FS_operate(ctx->fs_handle, ctx->fs_id[i], 0, FS_OP_SET, ATOMIC_READ(&(ctx->stat_val[i])));
	}

	FS_passive_output(ctx->fs_handle);
	return;
}

static void * tfe_work_thread(void * arg)
{
	struct tfe_thread_ctx * ctx = (struct tfe_thread_ctx *) arg;
	struct timeval timer_delay = {60, 0};

	struct event * ev = event_new(ctx->evbase, -1, EV_PERSIST, __dummy_event_handler, NULL);
	if (unlikely(ev == NULL))
	{
		TFE_LOG_ERROR(g_default_logger, "Failed at creating dummy event for thread %u", ctx->thread_id);
		exit(EXIT_FAILURE);
	}

	evtimer_add(ev, &timer_delay);
	ctx->running = 1;
	__currect_thread_id = ctx->thread_id;
	char thread_name[16];
	snprintf(thread_name, sizeof(thread_name), "tfe:worker-%d", ctx->thread_id);
	prctl(PR_SET_NAME, (unsigned long long) thread_name, NULL, NULL, NULL);

	TFE_LOG_INFO(g_default_logger, "Work thread %u is running...", ctx->thread_id);
	event_base_dispatch(ctx->evbase);
	assert(0);
	event_free(ev);
	TFE_LOG_ERROR(g_default_logger, "Work thread %u is exit...", ctx->thread_id);
	return (void *) NULL;
}

void tfe_proxy_work_thread_create_ctx(struct tfe_proxy * proxy)
{
	unsigned int i = 0;
	for (i = 0; i < proxy->nr_work_threads; i++)
	{
		proxy->work_threads[i] = ALLOC(struct tfe_thread_ctx, 1);
		proxy->work_threads[i]->thread_id = i;
		proxy->work_threads[i]->evbase = event_base_new();
	    proxy->work_threads[i]->dnsbase = evdns_base_new(proxy->work_threads[i]->evbase, EVDNS_BASE_INITIALIZE_NAMESERVERS);
		proxy->work_threads[i]->evhttp = key_keeper_evhttp_init(proxy->work_threads[i]->evbase, proxy->work_threads[i]->dnsbase, proxy->key_keeper_handler);
	}
	return;
}
int tfe_proxy_work_thread_run(struct tfe_proxy * proxy)
{
	struct tfe_thread_ctx * __thread_ctx = NULL;
	unsigned int i = 0;
	int ret = 0;
	for (i = 0; i < proxy->nr_work_threads; i++)
	{
		__thread_ctx = proxy->work_threads[i];
		ret = pthread_create(&__thread_ctx->thr, NULL, tfe_work_thread, (void *) __thread_ctx);
		if (unlikely(ret < 0))
		{
			TFE_LOG_ERROR(proxy->logger, "Failed at pthread_create() for thread %d, error %d: %s", i, errno,
				strerror(errno));
			return -1;
		}
	}
	return 0;
}
int tfe_proxy_config(struct tfe_proxy * proxy, const char * profile)
{
	/* Worker threads */
	MESA_load_profile_uint_def(profile, "system", "nr_worker_threads", &proxy->nr_work_threads, 1);
	MESA_load_profile_uint_def(profile, "system", "buffer_output_limit", &proxy->buffer_output_limit, 0);

	/* Debug */
	MESA_load_profile_uint_def(profile, "debug", "passthrough_all_tcp", &proxy->tcp_all_passthrough, 0);

	/* ratelimit */
	MESA_load_profile_uint_def(profile, "ratelimit", "read_rate", &proxy->rate_limit_options.read_rate, 0);
	MESA_load_profile_uint_def(profile, "ratelimit", "read_burst", &proxy->rate_limit_options.read_burst, 0);
	MESA_load_profile_uint_def(profile, "ratelimit", "write_rate", &proxy->rate_limit_options.write_rate, 0);
	MESA_load_profile_uint_def(profile, "ratelimit", "write_burst", &proxy->rate_limit_options.write_burst, 0);

	if(proxy->rate_limit_options.read_rate != 0
		|| proxy->rate_limit_options.read_burst != 0
		|| proxy->rate_limit_options.write_rate != 0
		|| proxy->rate_limit_options.write_burst != 0)
	{
		proxy->en_rate_limit = 1;
	}

	/* TCP options, -1 means unset, we shall not call setsockopt */
	MESA_load_profile_int_def(profile, "tcp", "sz_rcv_buffer", &proxy->tcp_options.sz_rcv_buffer, -1);
	MESA_load_profile_int_def(profile, "tcp", "sz_snd_buffer", &proxy->tcp_options.sz_snd_buffer, -1);
	MESA_load_profile_int_def(profile, "tcp", "so_keepalive", &proxy->tcp_options.so_keepalive, -1);
	MESA_load_profile_int_def(profile, "tcp", "tcp_keepidle", &proxy->tcp_options.tcp_keepidle, -1);
	MESA_load_profile_int_def(profile, "tcp", "tcp_keepintvl", &proxy->tcp_options.tcp_keepintvl, -1);
	MESA_load_profile_int_def(profile, "tcp", "tcp_keepcnt", &proxy->tcp_options.tcp_keepcnt, -1);
	MESA_load_profile_int_def(profile, "tcp", "tcp_user_timeout", &proxy->tcp_options.tcp_user_timeout, -1);
	MESA_load_profile_int_def(profile, "tcp", "tcp_ttl_upstream", &proxy->tcp_options.tcp_ttl_upstream, -1);
	MESA_load_profile_int_def(profile, "tcp", "tcp_ttl_downstream", &proxy->tcp_options.tcp_ttl_downstream, -1);

	return 0;
}

static const char * __str_stat_spec_map[] =
	{
		[STAT_SIGPIPE] = "SIGPIPE",
		[STAT_FD_OPEN_BY_KNI_ACCEPT] = "fd_rx",
		[STAT_FD_CLOSE_BY_KNI_ACCEPT_FAIL] = "fd_rx_err",
		[STAT_FD_CLOSE] = "fd_inst_cls",
		[STAT_STREAM_OPEN] = "stm_open",
		[STAT_STREAM_CLS] =  "stm_cls",
		[STAT_STREAM_CLS_DOWN_EOF] = "dstm_eof",
		[STAT_STREAM_CLS_UP_EOF] = "ustm_eof",
		[STAT_STREAM_CLS_DOWN_ERR] = "dstm_err",
		[STAT_STREAM_CLS_UP_ERR] = "ustm_err",
		[STAT_STREAM_CLS_KILL] = "stm_kill",
		[STAT_STREAM_INTERCEPT] = "stm_incpt",
		[STAT_STREAM_BYPASS] = "stm_byp",
		[STAT_STREAM_INCPT_BYTES] = "stm_incpt_B",
		[STAT_STREAM_INCPT_DOWN_BYTES] = "dstm_incpt_B",
		[STAT_STREAM_INCPT_UP_BYTES] = "ustm_incpt_B",
		[STAT_STREAM_TCP_PLAIN] = "plain",
		[STAT_STREAM_TCP_SSL] = "ssl",
		[TFE_STAT_MAX] = NULL
	};

int tfe_stat_init(struct tfe_proxy * proxy, const char * profile)
{
	static const char * fieldstat_output = "log/tfe.fs2";
	static const char * app_name = "tfe3a";

	int value = 0, i = 0;
	screen_stat_handle_t fs_handle = NULL;
	char statsd_server_ip[TFE_SYMBOL_MAX]={0};
	char histogram_bins[TFE_SYMBOL_MAX]={0};
	int statsd_server_port=0;
	MESA_load_profile_string_def(profile, "STAT", "statsd_server", statsd_server_ip,
		sizeof(statsd_server_ip), "");
	MESA_load_profile_int_def(profile, "STAT", "statsd_port", &(statsd_server_port), 0);
	MESA_load_profile_string_def(profile, "STAT", "histogram_bins",
		histogram_bins, sizeof(histogram_bins), "0.5,0.8,0.9,0.95");

	fs_handle = FS_create_handle();
	FS_set_para(fs_handle, OUTPUT_DEVICE, fieldstat_output, (int)strlen(fieldstat_output) + 1);
	FS_set_para(fs_handle, APP_NAME, app_name, (int)strlen(app_name) + 1);

	value = 1;
	FS_set_para(fs_handle, PRINT_MODE, &value, sizeof(value));
	value = 0;
	FS_set_para(fs_handle, CREATE_THREAD, &value, sizeof(value));
	if(strlen(statsd_server_ip)>0 && statsd_server_port!=0)
	{
		FS_set_para(fs_handle, STATS_SERVER_IP, statsd_server_ip, strlen(statsd_server_ip)+1);
		FS_set_para(fs_handle, STATS_SERVER_PORT, &(statsd_server_port), sizeof(statsd_server_port));
	}
	FS_set_para(fs_handle, HISTOGRAM_GLOBAL_BINS, histogram_bins, strlen(histogram_bins)+1);

	for (i = 0; i < TFE_STAT_MAX; i++)
	{
		proxy->fs_id[i] = FS_register(fs_handle, FS_STYLE_FIELD, FS_CALC_CURRENT, __str_stat_spec_map[i]);
	}

	FS_start(fs_handle);
	proxy->fs_handle = fs_handle;
	return 0;
}

void tfe_proxy_acceptor_init(struct tfe_proxy * proxy, const char * profile)
{
    MESA_load_profile_uint_def(profile, "system", "enable_kni_v1", &proxy->en_kni_v1_acceptor, 0);
    MESA_load_profile_uint_def(profile, "system", "enable_kni_v2", &proxy->en_kni_v2_acceptor, 1);

    if (proxy->en_kni_v1_acceptor)
    {
        g_default_proxy->kni_v1_acceptor = acceptor_kni_v1_create(proxy, profile, proxy->logger);
        CHECK_OR_EXIT(g_default_proxy->kni_v1_acceptor, "Failed at init KNIv1 acceptor. Exit. ");
    }

    if (proxy->en_kni_v2_acceptor)
    {
        g_default_proxy->kni_v2_acceptor = acceptor_kni_v2_create(g_default_proxy, profile, g_default_logger);
        CHECK_OR_EXIT(g_default_proxy->kni_v2_acceptor, "Failed at init KNIv2 acceptor. Exit. ");
    }

    return;
}

struct breakpad_instance
{
	unsigned int en_breakpad;
	char minidump_dir_prefix[TFE_STRING_MAX];
	google_breakpad::ExceptionHandler * exceptionHandler;

	/* Upload to crash server */
	unsigned int en_breakpad_upload;
	char minidump_sentry_upload_url[TFE_STRING_MAX];

	/* Upload tools name */
	char upload_tools_filename[TFE_STRING_MAX];

	/* Upload tools exec command */
	char * upload_tools_exec_argv[64];
	char * minidump_filename;
};

static bool tfe_breakpad_dump_to_file(const google_breakpad::MinidumpDescriptor& descriptor,
	void* context, bool succeeded)
{
	fprintf(stderr, "Crash happened, minidump path:  %s\n", descriptor.path());
	return succeeded;
}

static bool tfe_breakpad_dump_and_report(const google_breakpad::MinidumpDescriptor& descriptor,
	void* context, bool succeeded)
{
	struct breakpad_instance * instance = g_default_proxy->breakpad;
	int ret = 0;

	strncpy(instance->minidump_filename, descriptor.path(),  PATH_MAX - 1);
	fprintf(stderr, "Crash happened, prepare upload the minidump file:  %s\n", descriptor.path());

	ret = access(instance->minidump_filename, F_OK | R_OK);
	if (ret < 0)
	{
		fprintf(stderr, "minidump file is not existed, cannot upload minidump file");
		return succeeded;
	}

	/* Firstly, fork an child process */
	pid_t exec_child_pid = fork();
	if (exec_child_pid == 0)
	{
		/* As a child, exec minidump upload tools */
		ret = execv(instance->upload_tools_filename, instance->upload_tools_exec_argv);
		if (ret < 0)
		{
			fprintf(stderr, "Failed at exec the upload program %s: %s\n",
				instance->upload_tools_filename, strerror(errno));
		}

		exit(EXIT_FAILURE);
	}
	else if (exec_child_pid > 0)
	{
		fprintf(stderr, "Starting upload minidump, PID = %d. \n", exec_child_pid);
		return succeeded;
	}
	else
	{
		/* failed at fork, cannot upload the minidump */
		fprintf(stderr, "Failed at fork(), cannot upload minidump file. :  %s\n", strerror(errno));
		return succeeded;
	}
}

/* COREDUMP GENERATE TEST */
static void segv_generate()
{
	volatile char * _NULLPTR = nullptr;
	(*_NULLPTR) = 0;
}

static void _mkdir(const char *dir)
{
	char tmp[PATH_MAX];
	char * p = NULL;
	size_t len;

	snprintf(tmp, sizeof(tmp), "%s", dir);
	len = strlen(tmp);
	if (tmp[len - 1] == '/')
		tmp[len - 1] = 0;
	for (p = tmp + 1; *p; p++)
	{
		if (*p == '/')
		{
			*p = 0;
			mkdir(tmp, S_IRWXU);
			*p = '/';
		}
	}
	mkdir(tmp, S_IRWXU);
}

int breakpad_init_minidump_upload(struct breakpad_instance * instance, const char * profile)
{
	int ret = 0;
	char execpath[PATH_MAX] = {};
	char * execdirname = NULL;

	ret = MESA_load_profile_string_nodef(profile, "system", "breakpad_upload_url",
		instance->minidump_sentry_upload_url, sizeof(instance->minidump_sentry_upload_url));

	if (unlikely(ret < 0))
	{
		TFE_LOG_ERROR(g_default_logger, "breakpad_upload_url is necessary, failed. ");
		goto errout;
	}

	ret = readlink("/proc/self/exe", execpath, sizeof(execpath));
	if(unlikely(ret < 0))
	{
		TFE_LOG_ERROR(g_default_logger, "Failed at readlink /proc/self/exec: %s", strerror(errno));
		goto errout;
	}

	execdirname = dirname(execpath);
	snprintf(instance->upload_tools_filename, sizeof(instance->upload_tools_filename) - 1,
		"%s/%s", execdirname, "minidump_upload");


	/* Execfile */
	instance->upload_tools_exec_argv[0] = tfe_strdup(instance->upload_tools_filename);

	/* Firstly, Product Name and Product Version */
	instance->upload_tools_exec_argv[1] = tfe_strdup("-p");
	instance->upload_tools_exec_argv[2] = tfe_strdup("tfe");
	instance->upload_tools_exec_argv[3] = tfe_strdup("-v");
	instance->upload_tools_exec_argv[4] = tfe_strdup(tfe_version());

	/* Minidump file location, now we don't know it */
	instance->minidump_filename = (char *)ALLOC(char, PATH_MAX);
	instance->upload_tools_exec_argv[5] = instance->minidump_filename;

	/* Minidup upload url */
	instance->upload_tools_exec_argv[6] = tfe_strdup(instance->minidump_sentry_upload_url);
	instance->upload_tools_exec_argv[7] = NULL;
	return 0;

errout:
	return -1;
}

struct breakpad_instance * breakpad_init(const char * profile)
{
	struct breakpad_instance * instance = ALLOC(struct breakpad_instance, 1);
	assert(instance != nullptr);

	int ret = 0;
	unsigned int disable_coredump;
	MESA_load_profile_uint_def(profile, "system", "disable_coredump", &disable_coredump, 0);
	if (disable_coredump > 0)
	{
		const struct rlimit __rlimit_vars = {.rlim_cur = 0, .rlim_max = 0};
		ret = setrlimit(RLIMIT_CORE, &__rlimit_vars);
		if (ret < 0)
		{
			TFE_LOG_ERROR(g_default_logger, "setrlimit(RLIMIT_CORE, 0) failed: %s", strerror(errno));
		}
	}

	MESA_load_profile_uint_def(profile, "system", "enable_breakpad", &instance->en_breakpad, 1);
	if (instance->en_breakpad <= 0)
	{
		TFE_LOG_ERROR(g_default_logger, "Breakpad Crash Reporting System is disabled. ");
		return instance;
	}

	MESA_load_profile_string_def(profile, "system", "breakpad_minidump_dir",
		instance->minidump_dir_prefix, sizeof(instance->minidump_dir_prefix), "/tmp/crashreport");

	MESA_load_profile_uint_def(profile, "system", "enable_breakpad_upload",
		&instance->en_breakpad_upload, 0);

	/* Create the minidump dir if it is not existed */
	_mkdir(instance->minidump_dir_prefix);

	if (instance->en_breakpad_upload)
	{
		/* Try to init the breakpad upload */
		ret = breakpad_init_minidump_upload(instance, profile);
		if (ret < 0)
		{
			TFE_LOG_ERROR(g_default_logger, "Breakpad upload init failed, using local breakpad dumpfile");
			instance->en_breakpad_upload = 0;
		}

		/* When we use breakpad, do not generate any coredump file */
		const struct rlimit __rlimit_vars = {.rlim_cur = 0, .rlim_max = 0};
		ret = setrlimit(RLIMIT_CORE, &__rlimit_vars);
		if (ret < 0)
		{
			TFE_LOG_ERROR(g_default_logger, "setrlimit(RLIMIT_CORE, 0) failed: %s", strerror(errno));
		}
	}

	if (instance->en_breakpad_upload)
	{
		instance->exceptionHandler = new google_breakpad::ExceptionHandler(
			google_breakpad::MinidumpDescriptor(instance->minidump_dir_prefix), NULL,
			tfe_breakpad_dump_and_report, NULL, true, -1);
	}
	else
	{
		instance->exceptionHandler = new google_breakpad::ExceptionHandler(
			google_breakpad::MinidumpDescriptor(instance->minidump_dir_prefix), NULL,
			tfe_breakpad_dump_to_file, NULL, true, -1);
	}

	TFE_LOG_INFO(g_default_logger, "Breakpad Crash Report is enable. ");
	TFE_LOG_INFO(g_default_logger, "Minidump Dir: %s", instance->minidump_dir_prefix);
	return instance;
}

int main(int argc, char * argv[])
{
	const char * main_profile = "./conf/tfe/tfe.conf";
	const char * future_profile= "./conf/tfe/future.conf";

    int ret = 0;
    int opt = 0;
    bool to_generate_a_segv = false;

    while ((opt = getopt(argc, argv, "vg")) != -1)
    {
        switch (opt)
        {
            case 'v':
                fprintf(stderr, "Tango Frontend Engine, Version: %s\n", tfe_version());
                return 0;

        	case 'g':
        		fprintf(stderr, "Tango Frontend Engine, prepare to generate a coredump.");
				to_generate_a_segv = true;
				break;

            default:
                break;
        }
    }

    fprintf(stderr, "Tango Frontend Engine, Version: %s", __tfe_version);

    /* adds locking, only required if accessed from separate threads */
	evthread_use_pthreads();
	unsigned int __log_level = RLOG_LV_INFO;
	MESA_load_profile_uint_def(main_profile, "log", "level", &__log_level, RLOG_LV_INFO);

	char __log_path[TFE_STRING_MAX]= {};
	MESA_load_profile_string_def(main_profile, "log", "location", __log_path, sizeof(__log_path), "log/tfe.log");

	g_default_logger = MESA_create_runtime_log_handle(__log_path, __log_level);
	if (unlikely(g_default_logger == NULL))
	{
		TFE_LOG_ERROR(g_default_logger, "Failed at creating default logger: %s", "log/tfe.log");
		exit(EXIT_FAILURE);
	}

	/* PROXY INSTANCE */
	g_default_proxy = ALLOC(struct tfe_proxy, 1);
	assert(g_default_proxy);
	strcpy(g_default_proxy->name, "tfe3a");

	g_default_proxy->breakpad = breakpad_init(main_profile);
	CHECK_OR_EXIT(g_default_proxy->breakpad, "Failed at starting breakpad. Exit.");

	if (to_generate_a_segv)
	{
		segv_generate();
	}

	future_promise_library_init(future_profile);
	tango_cache_global_init();

	/* CONFIG */
	ret = tfe_proxy_config(g_default_proxy, main_profile);
	CHECK_OR_EXIT(ret == 0, "Failed at loading profile %s, Exit.", main_profile);

	/* PERFOMANCE MONITOR */
	tfe_stat_init(g_default_proxy, main_profile);

	/* LOGGER */
	g_default_proxy->logger = g_default_logger;

	/* MAIN THREAD EVBASE */
	g_default_proxy->evbase = event_base_new();
	CHECK_OR_EXIT(g_default_proxy->evbase, "Failed at creating evbase for main thread. Exit.");

	/* GC EVENT */
	g_default_proxy->gcev = event_new(g_default_proxy->evbase, -1, EV_PERSIST, __gc_handler_cb, g_default_proxy);
	CHECK_OR_EXIT(g_default_proxy->gcev, "Failed at creating GC event. Exit. ");

	/* KEY_KEEP INIT */
	g_default_proxy->key_keeper_handler = key_keeper_init(main_profile, "key_keeper", g_default_logger);
	CHECK_OR_EXIT(g_default_proxy->key_keeper_handler, "Failed at init Key keeper. Exit.");

	/* SSL INIT */
	g_default_proxy->ssl_mgr_handler = ssl_manager_init(main_profile, "ssl", g_default_proxy->evbase, g_default_proxy->key_keeper_handler, g_default_logger);
	CHECK_OR_EXIT(g_default_proxy->ssl_mgr_handler, "Failed at init SSL manager. Exit.");

	for (size_t i = 0; i < (sizeof(signals) / sizeof(int)); i++)
	{
		g_default_proxy->sev[i] = evsignal_new(g_default_proxy->evbase, signals[i], __signal_handler_cb, g_default_proxy);
		CHECK_OR_EXIT(g_default_proxy->sev[i], "Failed at create signal event. Exit.");
		evsignal_add(g_default_proxy->sev[i], NULL);
	}

	struct timeval gc_delay = {2, 0};
	evtimer_add(g_default_proxy->gcev, &gc_delay);

	/* WORKER THREAD CTX Create */
	tfe_proxy_work_thread_create_ctx(g_default_proxy);
    tfe_proxy_acceptor_init(g_default_proxy, main_profile);

    /* SCM Sender */
    g_default_proxy->scm_sender = sender_scm_init(main_profile, "kni", g_default_logger);
    CHECK_OR_EXIT(g_default_proxy->scm_sender != NULL, "Failed at creating scm sender, Exit.");

	/* PLUGIN INIT */
	unsigned int plugin_iterator = 0;
	for (struct tfe_plugin * plugin_iter = tfe_plugin_iterate(&plugin_iterator);
		 plugin_iter != NULL; plugin_iter = tfe_plugin_iterate(&plugin_iterator))
	{
		ret = plugin_iter->on_init(g_default_proxy);
		CHECK_OR_EXIT(ret >= 0, "Plugin %s init failed. Exit. ", plugin_iter->symbol);
		TFE_LOG_INFO(g_default_logger, "Plugin %s initialized. ", plugin_iter->symbol);
	}

	//ugly here. g_business_maat is available after plugin initiate.
	g_default_proxy->ssl_ply_enforcer=ssl_policy_enforcer_create(g_default_logger);
	ssl_manager_set_new_upstream_cb(g_default_proxy->ssl_mgr_handler, ssl_policy_enforce, g_default_proxy->ssl_ply_enforcer);
	ret = tfe_proxy_work_thread_run(g_default_proxy);
	CHECK_OR_EXIT(ret == 0, "Failed at creating thread. Exit.");

	/* Watchdog KNI */
	g_default_proxy->watchdog_kni = watchdog_kni_create(g_default_proxy, main_profile, g_default_logger);
	CHECK_OR_EXIT(g_default_proxy->watchdog_kni != NULL, "Failed at creating KNI watchdog, Exit.");

	TFE_LOG_ERROR(g_default_logger, "Tango Frontend Engine initialized, Version: %s.", __tfe_version);

	/* If TFE is run by systemd's notify, then tell the systemd our tfe is ready.
	 * and disable the stderr log, only print logs into files */
	if(check_is_started_by_notify())
	{
		sd_notify(0, "READY=1");
		g_print_to_stderr = false;
		sleep(1);
	}

	event_base_dispatch(g_default_proxy->evbase);
	return 0;
}

const char * tfe_version()
{
	return __tfe_version;
}

unsigned int tfe_proxy_get_work_thread_count(void)
{
	return g_default_proxy->nr_work_threads;
}

struct event_base * tfe_proxy_get_work_thread_evbase(unsigned int thread_id)
{
	assert(thread_id < g_default_proxy->nr_work_threads);
	return g_default_proxy->work_threads[thread_id]->evbase;
}

struct evdns_base* tfe_proxy_get_work_thread_dnsbase(unsigned int thread_id)
{
	assert(thread_id < g_default_proxy->nr_work_threads);
	return g_default_proxy->work_threads[thread_id]->dnsbase;
}

struct evhttp_connection* tfe_proxy_get_work_thread_evhttp(unsigned int thread_id)
{
	assert(thread_id < g_default_proxy->nr_work_threads);
	return g_default_proxy->work_threads[thread_id]->evhttp;
}

struct event_base * tfe_proxy_get_gc_evbase(void)
{
	return g_default_proxy->evbase;
}

screen_stat_handle_t tfe_proxy_get_fs_handle(void)
{
	return g_default_proxy->fs_handle;
}

void * tfe_proxy_get_error_logger(void)
{
    return g_default_logger;
}

int tfe_proxy_ssl_add_trust_ca(const char* pem_file)
{
	return ssl_manager_add_trust_ca(g_default_proxy->ssl_mgr_handler, pem_file);
}
int tfe_proxy_ssl_del_trust_ca(const char* pem_file)
{
	return ssl_manager_del_trust_ca(g_default_proxy->ssl_mgr_handler, pem_file);
}
int tfe_proxy_ssl_add_crl(const char* pem_file)
{
	return ssl_manager_add_crl(g_default_proxy->ssl_mgr_handler, pem_file);

}
int tfe_proxy_ssl_del_crl(const char* pem_file)
{
	return ssl_manager_del_crl(g_default_proxy->ssl_mgr_handler, pem_file);
}
void tfe_proxy_ssl_reset_trust_ca(void)
{
	ssl_manager_reset_trust_ca(g_default_proxy->ssl_mgr_handler);
	return;
}