551 lines
22 KiB
C++
551 lines
22 KiB
C++
#include <sys/stat.h>
|
||
#include <sys/types.h>
|
||
#include <unistd.h>
|
||
#include <stdio.h>
|
||
#include <stdlib.h>
|
||
#include <assert.h>
|
||
#include <errno.h>
|
||
#include <pthread.h>
|
||
#include <string.h>
|
||
#include <sys/ioctl.h>
|
||
#include <sys/socket.h>
|
||
#include <netinet/in.h>
|
||
#include <netinet/tcp.h>
|
||
#include <arpa/inet.h>
|
||
#include <net/if.h>
|
||
|
||
#include <event2/thread.h>
|
||
|
||
#include <MESA/MESA_prof_load.h>
|
||
|
||
#include "doris_consumer_client.h"
|
||
#include "doris_producer_client.h"
|
||
#include "doris_server_main.h"
|
||
#include "doris_server_http.h"
|
||
|
||
struct doris_global_info g_doris_server_info;
|
||
static unsigned long doris_version_20210830=20210830L;
|
||
|
||
int doris_mkdir_according_path(const char * path)
|
||
{
|
||
char buffer[256];
|
||
const char *ps=path, *pc;
|
||
|
||
if(*ps == '/')
|
||
ps += 1;
|
||
|
||
while((pc = strchr(ps, '/')) != NULL)
|
||
{
|
||
while(*(pc+1) == '/')
|
||
pc++;
|
||
|
||
memcpy(buffer, path, pc - path);
|
||
buffer[pc-path] = '\0';
|
||
|
||
if(access(buffer, F_OK))
|
||
{
|
||
if(mkdir(buffer, 0777))
|
||
{
|
||
return -1;
|
||
}
|
||
}
|
||
ps = pc + 1;
|
||
}
|
||
if(access(path, F_OK))
|
||
{
|
||
if(mkdir(path, 0777))
|
||
{
|
||
return -1;
|
||
}
|
||
}
|
||
return 0;
|
||
}
|
||
|
||
int32_t get_ip_by_ifname(const char *ifname)
|
||
{
|
||
int sockfd;
|
||
struct ifreq ifr;
|
||
int32_t ip;
|
||
|
||
sockfd = socket(AF_INET, SOCK_DGRAM, 0);
|
||
if (-1 == sockfd)
|
||
return INADDR_NONE;
|
||
|
||
strcpy(ifr.ifr_name,ifname);
|
||
if (ioctl(sockfd, SIOCGIFADDR, &ifr) < 0)
|
||
{
|
||
close(sockfd);
|
||
return INADDR_NONE;
|
||
}
|
||
|
||
ip = ((struct sockaddr_in*)&(ifr.ifr_addr))->sin_addr.s_addr;
|
||
close(sockfd);
|
||
return ip;
|
||
}
|
||
|
||
static inline void set_sockopt_keepalive(int sd, int keepidle, int keepintvl, int keepcnt)
|
||
{
|
||
int keepalive = 1;
|
||
setsockopt(sd, SOL_SOCKET, SO_KEEPALIVE, (void*)&keepalive, sizeof(keepalive));
|
||
setsockopt(sd, SOL_TCP, TCP_KEEPIDLE, (void*)&keepidle, sizeof(keepidle));
|
||
setsockopt(sd, SOL_TCP, TCP_KEEPINTVL, (void*)&keepintvl, sizeof(keepintvl));
|
||
setsockopt(sd, SOL_TCP, TCP_KEEPCNT, (void*)&keepcnt, sizeof(keepcnt));
|
||
}
|
||
|
||
static inline void set_listen_sockopt(int sd)
|
||
{
|
||
if(g_doris_server_info.sock_recv_bufsize > 0)
|
||
{
|
||
setsockopt(sd, SOL_SOCKET, SO_RCVBUF, &g_doris_server_info.sock_recv_bufsize, sizeof(u_int32_t));
|
||
}
|
||
}
|
||
|
||
int doris_create_listen_socket(int bind_port)
|
||
{
|
||
evutil_socket_t listener;
|
||
struct sockaddr_in sin;
|
||
|
||
listener = socket(AF_INET, SOCK_STREAM, 0);
|
||
if(listener < 0)
|
||
{
|
||
MESA_RUNTIME_LOGV3(g_doris_server_info.log_runtime, RLOG_LV_FATAL, "create socket error!\n");
|
||
return -1;
|
||
}
|
||
set_sockopt_keepalive(listener, 300, 10, 2);
|
||
set_listen_sockopt(listener);
|
||
evutil_make_listen_socket_reuseable(listener);
|
||
|
||
sin.sin_family = AF_INET;
|
||
sin.sin_addr.s_addr=htonl(INADDR_ANY);
|
||
sin.sin_port = htons(bind_port);
|
||
|
||
if (bind(listener, (struct sockaddr *)&sin, sizeof(sin)) < 0)
|
||
{
|
||
printf("bind socket to port: %d error: %s!\n", bind_port, strerror(errno));
|
||
MESA_RUNTIME_LOGV3(g_doris_server_info.log_runtime, RLOG_LV_FATAL, "bind socket to port: %d error: %s!\n", bind_port, strerror(errno));
|
||
assert(0);return -2;
|
||
}
|
||
if (listen(listener, 1024)<0)
|
||
{
|
||
MESA_RUNTIME_LOGV3(g_doris_server_info.log_runtime, RLOG_LV_FATAL, "listen socket 1024 error!\n");
|
||
return -3;
|
||
}
|
||
evutil_make_socket_nonblocking(listener);
|
||
return listener;
|
||
}
|
||
|
||
static int doris_chech_name_valid(const char *name)
|
||
{
|
||
size_t i, namelen=strlen(name);
|
||
|
||
for(i=0; i<namelen; i++)
|
||
{
|
||
if(!((name[i]>='a' && name[i]<='z')||(name[i]>='A' && name[i]<='Z') ||
|
||
(name[i]>='0' && name[i]<='9') || name[i]=='_' || name[i]==':'))
|
||
{
|
||
return false;
|
||
}
|
||
}
|
||
return true;
|
||
}
|
||
|
||
int32_t doris_read_profile_configs(const char *config_file)
|
||
{
|
||
char tmp_buf[4096], tmp_dir[512];
|
||
|
||
MESA_load_profile_string_def(config_file, "DORIS_SERVER", "run_log_dir", g_doris_server_info.root_log_dir, sizeof(g_doris_server_info.root_log_dir), "./log");
|
||
MESA_load_profile_uint_def(config_file, "DORIS_SERVER", "run_log_lv", &g_doris_server_info.log_level, 10);
|
||
MESA_load_profile_uint_def(config_file, "DORIS_SERVER", "statistic_log_interval", &g_doris_server_info.statistic_period, 300);
|
||
/*runtimelog*/
|
||
snprintf(tmp_dir, 256, "%s/runtime_log", g_doris_server_info.root_log_dir);
|
||
if(doris_mkdir_according_path(tmp_dir))
|
||
{
|
||
printf("mkdir %s for runtimelog failed: %s\n", tmp_dir, strerror(errno));
|
||
return -1;
|
||
}
|
||
snprintf(tmp_buf, 256, "%s/doris_runtime.log", tmp_dir);
|
||
g_doris_server_info.log_runtime = MESA_create_runtime_log_handle(tmp_buf, g_doris_server_info.log_level);
|
||
if(NULL==g_doris_server_info.log_runtime)
|
||
{
|
||
printf("MESA_create_runtime_log_handle %s failed: %s\n", tmp_buf, strerror(errno));
|
||
return -1;
|
||
}
|
||
|
||
/*System*/
|
||
MESA_load_profile_uint_def(config_file, "DORIS_SERVER", "worker_thread_num", &g_doris_server_info.iothreads, 1);
|
||
if(g_doris_server_info.iothreads > 255)
|
||
{
|
||
printf("%s: [DORIS_SERVER]worker_thread_num=%d should not be bigger than 255!", config_file, g_doris_server_info.iothreads);
|
||
MESA_RUNTIME_LOGV3(g_doris_server_info.log_runtime, RLOG_LV_FATAL, "%s: [DORIS_SERVER]worker_thread_num=%d should not be bigger than 255!", config_file, g_doris_server_info.iothreads);
|
||
assert(0);return -1;
|
||
}
|
||
MESA_load_profile_uint_def(config_file, "DORIS_SERVER", "cache_file_frag_size", &g_doris_server_info.cache_frag_size, 67108864);
|
||
MESA_load_profile_uint_def(config_file, "DORIS_SERVER", "http_post_max_frag_size", &g_doris_server_info.max_http_body_size, 67108864);
|
||
MESA_load_profile_uint_def(config_file, "DORIS_SERVER", "index_file_format_maat", &g_doris_server_info.idx_file_maat, 0);
|
||
MESA_load_profile_uint_def(config_file, "DORIS_SERVER", "http_post_max_concurrence", &g_doris_server_info.max_concurrent_reqs, 100000);
|
||
MESA_load_profile_uint_def(config_file, "DORIS_SERVER", "http_post_cluster_on", &g_doris_server_info.cluster_sync_mode, 0);
|
||
MESA_load_profile_uint_def(config_file, "DORIS_SERVER", "http_post_op_expires", &g_doris_server_info.post_vernode_ttl, 1200);
|
||
|
||
MESA_load_profile_uint_def(config_file, "DORIS_SERVER", "scan_index_file_interval", &g_doris_server_info.scan_idx_interval, 10);
|
||
MESA_load_profile_uint_def(config_file, "DORIS_SERVER", "https_connection_on", &g_doris_server_info.ssl_conn_on, 0);
|
||
if(g_doris_server_info.ssl_conn_on)
|
||
{
|
||
MESA_load_profile_string_def(config_file, "DORIS_SERVER", "ssl_trusted_ca_path", g_doris_server_info.ssl_CA_path, 256, "./conf/ssl_CA_path/");
|
||
MESA_load_profile_string_def(config_file, "DORIS_SERVER", "ssl_certificate_file", g_doris_server_info.ssl_cert_file, 256, "./conf/server_cert.pem");
|
||
MESA_load_profile_string_def(config_file, "DORIS_SERVER", "ssl_private_key_file", g_doris_server_info.ssl_key_file, 256, "./conf/server_key.pem");
|
||
MESA_load_profile_string_def(config_file, "DORIS_SERVER", "ssl_private_key_passwd", g_doris_server_info.ssl_key_passwd, 64, "12345678");
|
||
}
|
||
|
||
/*FiledStat*/
|
||
MESA_load_profile_string_def(config_file, "DORIS_SERVER", "fsstat_log_appname", g_doris_server_info.fsstat_appname, 16, "DORIS_SERVER_S");
|
||
MESA_load_profile_string_def(config_file, "DORIS_SERVER", "fsstat_log_filepath", g_doris_server_info.fsstat_filepath, 256, "./log/doris_server.fs");
|
||
MESA_load_profile_uint_def(config_file, "DORIS_SERVER", "fsstat_log_interval", &g_doris_server_info.fsstat_period, 10);
|
||
MESA_load_profile_int_def(config_file, "DORIS_SERVER", "fsstat_log_print_mode", &g_doris_server_info.fsstat_print_mode, 1);
|
||
MESA_load_profile_string_def(config_file, "DORIS_SERVER", "fsstat_log_dst_ip", g_doris_server_info.fsstat_dst_ip, 64, "127.0.0.1");
|
||
MESA_load_profile_int_def(config_file, "DORIS_SERVER", "fsstat_log_dst_port", &g_doris_server_info.fsstat_dst_port, 8125);
|
||
|
||
//LIBEVENT
|
||
MESA_load_profile_int_def(config_file, "DORIS_SERVER", "consumer_listen_port", &g_doris_server_info.consumer_port, 0);
|
||
MESA_load_profile_int_def(config_file, "DORIS_SERVER", "manager_listen_port", &g_doris_server_info.manager_port, 2233);
|
||
MESA_load_profile_int_def(config_file, "DORIS_SERVER", "socket_recv_bufsize", &g_doris_server_info.sock_recv_bufsize, 524288);
|
||
pthread_mutex_init(&g_doris_server_info.mutex_lock, NULL);
|
||
return 0;
|
||
}
|
||
|
||
static int doris_server_register_field_stat(struct doris_global_info *param)
|
||
{
|
||
const char *field_names[DRS_FSSTAT_FIELD_NUM]={"RecvErrVer", "FileStarts", "FileComplete", "ClientInvReq",
|
||
"ClientMetaReq", "SendNoNewMeta", "ClientFileReq", "SendFiles", "SendBytes", "SendFile404", "VerExpire"};
|
||
const char *status_names[DRS_FSSTAT_STATUS_NUM]={"MemoryUsed"};
|
||
const char *column_names[DRS_FSSTAT_CLUMN_NUM]={"RecvFullVer", "RecvIncVer", "RecvFiles", "PostOnWay", "SendResMeta",
|
||
"CurFullVer", "CurIncVer", "TotalCfgNum"};
|
||
int value;
|
||
|
||
param->fsstat_handle = FS_create_handle();
|
||
FS_set_para(param->fsstat_handle, OUTPUT_DEVICE, param->fsstat_filepath, strlen(param->fsstat_filepath)+1);
|
||
if(param->fsstat_print_mode == 1)
|
||
{
|
||
FS_set_para(param->fsstat_handle, PRINT_MODE, ¶m->fsstat_print_mode, sizeof(param->fsstat_print_mode));
|
||
}
|
||
else
|
||
{
|
||
FS_set_para(param->fsstat_handle, PRINT_MODE, ¶m->fsstat_print_mode, sizeof(param->fsstat_print_mode));
|
||
value = 1;
|
||
FS_set_para(param->fsstat_handle, FLUSH_BY_DATE, &value, sizeof(value));
|
||
}
|
||
value = param->fsstat_period;
|
||
FS_set_para(param->fsstat_handle, STAT_CYCLE, &value, sizeof(value));
|
||
value = 0;
|
||
FS_set_para(param->fsstat_handle, CREATE_THREAD, &value, sizeof(value));
|
||
FS_set_para(param->fsstat_handle, APP_NAME, param->fsstat_appname, strlen(param->fsstat_appname)+1);
|
||
FS_set_para(param->fsstat_handle, STATS_SERVER_IP, param->fsstat_dst_ip, strlen(param->fsstat_dst_ip)+1);
|
||
FS_set_para(param->fsstat_handle, STATS_SERVER_PORT, ¶m->fsstat_dst_port, sizeof(param->fsstat_dst_port));
|
||
|
||
for(int i=0; i<DRS_FSSTAT_FIELD_NUM; i++)
|
||
{
|
||
param->fsstat_field[i] = FS_register(param->fsstat_handle, FS_STYLE_FIELD, FS_CALC_CURRENT, field_names[i]);
|
||
}
|
||
for(int i=0; i<DRS_FSSTAT_STATUS_NUM; i++)
|
||
{
|
||
param->fsstat_status[i] = FS_register(param->fsstat_handle, FS_STYLE_STATUS, FS_CALC_CURRENT, status_names[i]);
|
||
}
|
||
for(int i=0; i<DRS_FSSTAT_CLUMN_NUM; i++)
|
||
{
|
||
param->fsstat_column[i] = FS_register(param->fsstat_handle, FS_STYLE_COLUMN, FS_CALC_CURRENT, column_names[i]);
|
||
}
|
||
FS_start(param->fsstat_handle);
|
||
return 0;
|
||
}
|
||
|
||
static void instance_fsstat_output_timer_cb(int fd, short kind, void *userp)
|
||
{
|
||
struct timeval tv;
|
||
|
||
FS_passive_output(g_doris_server_info.fsstat_handle);
|
||
|
||
tv.tv_sec = g_doris_server_info.fsstat_period;
|
||
tv.tv_usec = 0;
|
||
event_add((struct event*)userp, &tv);
|
||
}
|
||
|
||
static int32_t doris_init_config_for_business(struct doris_global_info *info, struct event_base *manage_evbase, const char *config_file)
|
||
{
|
||
char tmpbuffer[4096], tmp_dir[256], tmp_dir2[256], *bizname, *save=NULL;
|
||
struct doris_business *business;
|
||
map<string, struct doris_csum_param *>::iterator iter_csm;
|
||
map<string, struct doris_prod_param *>::iterator iter_prd;
|
||
|
||
if(0>=MESA_load_profile_string_nodef(config_file, "DORIS_SERVER", "business_system_list", tmpbuffer, sizeof(tmpbuffer)))
|
||
{
|
||
MESA_RUNTIME_LOGV3(info->log_runtime, RLOG_LV_FATAL, "%s: [DORIS_SERVER]business_system_list not found!", config_file);
|
||
assert(0);return -1;
|
||
}
|
||
for(bizname=strtok_r(tmpbuffer, ";", &save); bizname!=NULL; bizname=strtok_r(NULL, ";", &save))
|
||
{
|
||
if(!doris_chech_name_valid(bizname))
|
||
{
|
||
MESA_RUNTIME_LOGV3(info->log_runtime, RLOG_LV_FATAL, "%s: [DORIS_SERVER]business_system_list bizname %s invalid, name must match: [a-zA-Z_:][a-zA-Z0-9_:]", bizname, config_file);
|
||
assert(0);return -1;
|
||
}
|
||
business = &info->business[info->business_num++];
|
||
snprintf(business->bizname, sizeof(business->bizname), "%s", bizname);
|
||
pthread_rwlock_init(&business->rwlock, NULL);
|
||
|
||
business->cfgver_head = config_version_handle_new();
|
||
if(info->name2business->find(string(business->bizname)) != info->name2business->end())
|
||
{
|
||
MESA_RUNTIME_LOGV3(info->log_runtime, RLOG_LV_FATAL, "%s: [%s]business_system_list duplicate system name: %s!", bizname, config_file, business->bizname);
|
||
assert(0);return -1;
|
||
}
|
||
info->name2business->insert(make_pair(string(business->bizname), business));
|
||
|
||
MESA_load_profile_uint_def(config_file, business->bizname, "max_store_full_versions", &business->saves_when_fulldel, 0);
|
||
if(business->saves_when_fulldel > 16)
|
||
{
|
||
MESA_RUNTIME_LOGV3(info->log_runtime, RLOG_LV_FATAL, "\033[1;31;40m[Alert] %s [%s]max_store_full_versions support max 16!!!!\033[0m\n", config_file, business->bizname);
|
||
business->saves_when_fulldel = 16;
|
||
}
|
||
MESA_load_profile_uint_def(config_file, business->bizname, "grafana_monitor_status_id", &business->mmval_status_codeid, 3);
|
||
MESA_load_profile_uint_def(config_file, business->bizname, "persistence_write_on", &business->persistence_write_on, 1);
|
||
MESA_load_profile_int_def(config_file, business->bizname, "mem_cache_max_versions", &business->cache_max_versions, 0);
|
||
if(business->persistence_write_on==0 && business->cache_max_versions==0)
|
||
{
|
||
MESA_RUNTIME_LOGV3(info->log_runtime, RLOG_LV_FATAL, "\033[1;31;40m[Alert] %s [%s], you must set mem_cache_max_versions if you disable persistence_write_on!\033[0m\n", config_file, business->bizname);
|
||
return -2;
|
||
}
|
||
if(0>MESA_load_profile_string_nodef(config_file, business->bizname, "store_config_path", business->store_path_root, sizeof(business->store_path_root)))
|
||
{
|
||
MESA_RUNTIME_LOGV3(info->log_runtime, RLOG_LV_FATAL, "%s: [%s]store_config_path not found!", bizname, config_file);
|
||
assert(0);return -1;
|
||
}
|
||
if(business->persistence_write_on)
|
||
{
|
||
snprintf(tmp_dir, 512, "%s/full/index", business->store_path_root);
|
||
snprintf(tmp_dir2,512, "%s/inc/index", business->store_path_root);
|
||
if(doris_mkdir_according_path(tmp_dir) || doris_mkdir_according_path(tmp_dir2))
|
||
{
|
||
MESA_RUNTIME_LOGV3(info->log_runtime, RLOG_LV_FATAL, "mkdir %s failed: %s\n", tmp_dir, strerror(errno));
|
||
return -1;
|
||
}
|
||
}
|
||
MESA_load_profile_uint_def(config_file, business->bizname, "receive_config_way", &business->recv_way, RECV_WAY_DRS_CLIENT);
|
||
assert(business->recv_way==RECV_WAY_IDX_FILE || business->recv_way==RECV_WAY_DRS_CLIENT || business->recv_way==RECV_WAY_HTTP_POST);
|
||
if(business->recv_way == RECV_WAY_IDX_FILE)
|
||
{
|
||
if(0>MESA_load_profile_string_nodef(config_file, business->bizname, "receive_config_path_full", business->recv_path_full, sizeof(business->recv_path_full)))
|
||
{
|
||
MESA_RUNTIME_LOGV3(info->log_runtime, RLOG_LV_FATAL, "%s: [%s]receive_config_path not found!", bizname, config_file);
|
||
assert(0);return -1;
|
||
}
|
||
if(0>MESA_load_profile_string_nodef(config_file, business->bizname, "receive_config_path_inc", business->recv_path_inc, sizeof(business->recv_path_inc)))
|
||
{
|
||
MESA_RUNTIME_LOGV3(info->log_runtime, RLOG_LV_FATAL, "%s: [%s]receive_config_path not found!", bizname, config_file);
|
||
assert(0);return -1;
|
||
}
|
||
}
|
||
else //<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ͬ<EFBFBD><CDAC>Ҳ<EFBFBD><D2B2>Ҫ
|
||
{
|
||
if(business->recv_way==RECV_WAY_DRS_CLIENT || g_doris_server_info.cluster_sync_mode)
|
||
{
|
||
if(0>=MESA_load_profile_string_nodef(config_file, business->bizname, "doris_client_confile", tmp_dir, sizeof(tmp_dir)))
|
||
{
|
||
MESA_RUNTIME_LOGV3(g_doris_server_info.log_runtime, RLOG_LV_FATAL, "%s: [DORIS_SERVER]doris_client_confile not found!", config_file);
|
||
assert(0);return -1;
|
||
}
|
||
if((iter_csm=info->confile2csmparam->find(string(tmp_dir))) != info->confile2csmparam->end())
|
||
{
|
||
business->param_csum = iter_csm->second;
|
||
}
|
||
else
|
||
{
|
||
business->param_csum = doris_csum_parameter_new(tmp_dir, manage_evbase, info->log_runtime);
|
||
if(business->param_csum == NULL)
|
||
{
|
||
assert(0);return -2;
|
||
}
|
||
info->confile2csmparam->insert(make_pair(string(tmp_dir), business->param_csum));
|
||
}
|
||
}
|
||
if(business->recv_way == RECV_WAY_HTTP_POST) //<2F><>Ҫ<EFBFBD><D2AA><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ͬ<EFBFBD><CDAC>
|
||
{
|
||
MESA_load_profile_uint_def(config_file, business->bizname, "producer_concurrence_allowed", &business->concurrency_allowed, 0);
|
||
if(MESA_load_profile_int_nodef(config_file, business->bizname, "producer_listen_port", &business->producer_port)<0)
|
||
{
|
||
MESA_RUNTIME_LOGV3(g_doris_server_info.log_runtime, RLOG_LV_FATAL, "%s: [%s]producer_listen_port not found!", config_file, business->bizname);
|
||
assert(0);return -1;
|
||
}
|
||
if(g_doris_server_info.cluster_sync_mode &&
|
||
NULL==(business->param_prod = doris_prod_parameter_new(tmp_dir, manage_evbase, info->log_runtime)))
|
||
{
|
||
assert(0);return -2;
|
||
}
|
||
if(!business->persistence_write_on)
|
||
{
|
||
g_doris_server_info.business_post_nopersists++;
|
||
}
|
||
g_doris_server_info.business_post_num++;
|
||
business->token2node = new map<string, struct version_list_node *>;
|
||
}
|
||
}
|
||
business->fs_lineid = FS_register(info->fsstat_handle, FS_STYLE_LINE, FS_CALC_CURRENT, business->bizname);;
|
||
|
||
snprintf(tmp_dir, 512, "latest_cfg_version_%s", business->bizname);
|
||
business->mmid_latest_ver = MESA_Monitor_register(info->monitor, tmp_dir, MONITOR_METRICS_GAUGE, "Latest doris config version.");
|
||
snprintf(tmp_dir, 512, "total_config_num_%s", business->bizname);
|
||
business->mmid_total_cfgnum = MESA_Monitor_register(info->monitor, tmp_dir, MONITOR_METRICS_GAUGE, "Total config num from latest full version till now.");
|
||
}
|
||
return 0;
|
||
}
|
||
|
||
static void manager_statistic_threads_requests_cb(struct evhttp_request *req, void *arg)
|
||
{
|
||
evhttp_send_error(req, HTTP_BADREQUEST, "Not Supported.");
|
||
}
|
||
|
||
static void manager_statistic_status_requests_cb(struct evhttp_request *req, void *arg)
|
||
{
|
||
evhttp_send_error(req, HTTP_BADREQUEST, "Not Supported.");
|
||
}
|
||
|
||
static void manager_statistic_field_requests_cb(struct evhttp_request *req, void *arg)
|
||
{
|
||
evhttp_send_error(req, HTTP_BADREQUEST, "Not Supported.");
|
||
}
|
||
|
||
void manager_generic_requests_cb(struct evhttp_request *req, void *arg)
|
||
{
|
||
evhttp_send_error(req, HTTP_BADREQUEST, "Not Supported.");
|
||
}
|
||
|
||
int main(int argc, char **argv)
|
||
{
|
||
struct event_base *manage_evbase;
|
||
pthread_t thread_desc;
|
||
pthread_attr_t attr;
|
||
struct timeval tv;
|
||
struct event timer_statistic_fsstat;
|
||
struct evhttp *manager_http;
|
||
char netname[32];
|
||
|
||
memset(&g_doris_server_info, 0, sizeof(struct doris_global_info));
|
||
if(doris_read_profile_configs(NIRVANA_CONFIG_FILE) || doris_server_register_field_stat(&g_doris_server_info))
|
||
{
|
||
return -1;
|
||
}
|
||
evthread_use_pthreads();
|
||
g_doris_server_info.name2business = new map<string, struct doris_business*>;
|
||
g_doris_server_info.confile2csmparam = new map<string, struct doris_csum_param *>;
|
||
manage_evbase = event_base_new();
|
||
|
||
/*Doris manager server*/
|
||
g_doris_server_info.manager = doris_create_listen_socket(g_doris_server_info.manager_port);
|
||
if(g_doris_server_info.manager < 0)
|
||
{
|
||
return -6;
|
||
}
|
||
if(NULL == (manager_http = evhttp_new(manage_evbase)))
|
||
{
|
||
printf("evhttp_new error!\n");
|
||
assert(0); return -2;
|
||
}
|
||
evhttp_set_cb(manager_http, "/doris/statistic/field", manager_statistic_field_requests_cb, NULL);
|
||
evhttp_set_cb(manager_http, "/doris/statistic/status", manager_statistic_status_requests_cb, NULL);
|
||
evhttp_set_cb(manager_http, "/doris/statistic/threads", manager_statistic_threads_requests_cb, NULL);
|
||
evhttp_set_gencb(manager_http, manager_generic_requests_cb, NULL);
|
||
g_doris_server_info.monitor = MESA_Monitor_instance_evhttp_new(manager_http, doris_version_20210830);
|
||
if(evhttp_accept_socket(manager_http, g_doris_server_info.manager))
|
||
{
|
||
printf("evhttp_accept_socket %d error!\n", g_doris_server_info.manager);
|
||
MESA_RUNTIME_LOGV3(g_doris_server_info.log_runtime, RLOG_LV_FATAL, "evhttp_accept_socket %d error!\n", g_doris_server_info.manager);
|
||
assert(0); return -3;
|
||
}
|
||
|
||
//Ϊÿ<CEAA><C3BF>ҵ<EFBFBD><D2B5>ϵͳ<CFB5><CDB3>ʼ<EFBFBD><CABC><EFBFBD><EFBFBD>ȡ<EFBFBD><C8A1><EFBFBD>õĽṹ
|
||
if(doris_init_config_for_business(&g_doris_server_info, manage_evbase, NIRVANA_CONFIG_FILE))
|
||
{
|
||
return -4;
|
||
}
|
||
if(g_doris_server_info.business_post_num > 0)
|
||
{
|
||
if((MESA_load_profile_string_nodef(NIRVANA_CONFIG_FILE, "DORIS_SERVER", "local_net_name", netname, sizeof(netname))<0) ||
|
||
(g_doris_server_info.local_ip = get_ip_by_ifname(netname))<0)
|
||
{
|
||
printf("%s: [DORIS_SERVER]local_net_name not valid", NIRVANA_CONFIG_FILE);
|
||
MESA_RUNTIME_LOGV3(g_doris_server_info.log_runtime, RLOG_LV_FATAL, "%s: [DORIS_SERVER]local_net_name not valid", NIRVANA_CONFIG_FILE);
|
||
assert(0);return -11;
|
||
}
|
||
g_doris_server_info.mmid_post_server = MESA_Monitor_register(g_doris_server_info.monitor,
|
||
"http_post_server_status", MONITOR_METRICS_GAUGE, "Running status of doris http post server.");
|
||
MESA_Monitor_operation(g_doris_server_info.monitor, g_doris_server_info.mmid_post_server, MONITOR_VALUE_SET, PROMETHUES_POST_SERVER_DOWN);
|
||
}
|
||
if(g_doris_server_info.business_post_nopersists>0 && NULL==(g_doris_server_info.kvdbhandle=doris_kvdb_hanlde_new("./leveldbdata")))
|
||
{
|
||
MESA_RUNTIME_LOGV3(g_doris_server_info.log_runtime, RLOG_LV_FATAL, "\033[1;31;40mOpen levelDB ./leveldbdata failed.\033[0m");
|
||
assert(0);return -11;
|
||
}
|
||
|
||
if(g_doris_server_info.ssl_conn_on && NULL==(g_doris_server_info.ssl_instance=doris_connections_create_ssl_ctx()))
|
||
{
|
||
assert(0);return -8;
|
||
}
|
||
|
||
pthread_attr_init(&attr);
|
||
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
|
||
for(u_int32_t i=0; i<g_doris_server_info.business_num; i++)
|
||
{
|
||
if(g_doris_server_info.business[i].recv_way == RECV_WAY_DRS_CLIENT)
|
||
{
|
||
if(pthread_create(&thread_desc, &attr, thread_doris_client_recv_cfg, &g_doris_server_info.business[i]))
|
||
{
|
||
MESA_RUNTIME_LOGV3(g_doris_server_info.log_runtime, RLOG_LV_FATAL, "pthread_create(): %s", strerror(errno));
|
||
assert(0);return -5;
|
||
}
|
||
}
|
||
else if(g_doris_server_info.business[i].recv_way == RECV_WAY_HTTP_POST)
|
||
{
|
||
if(pthread_create(&thread_desc, &attr, thread_http_post_recv_cfg, &g_doris_server_info.business[i]))
|
||
{
|
||
MESA_RUNTIME_LOGV3(g_doris_server_info.log_runtime, RLOG_LV_FATAL, "pthread_create(): %s", strerror(errno));
|
||
assert(0);return -6;
|
||
}
|
||
}
|
||
else
|
||
{
|
||
if(pthread_create(&thread_desc, &attr, thread_index_file_recv_cfg, &g_doris_server_info.business[i]))
|
||
{
|
||
MESA_RUNTIME_LOGV3(g_doris_server_info.log_runtime, RLOG_LV_FATAL, "pthread_create(): %s", strerror(errno));
|
||
assert(0);return -6;
|
||
}
|
||
}
|
||
}
|
||
|
||
/*Doris http server*/
|
||
if(g_doris_server_info.consumer_port)
|
||
{
|
||
g_doris_server_info.listener_csum = doris_create_listen_socket(g_doris_server_info.consumer_port);
|
||
if(g_doris_server_info.listener_csum < 0)
|
||
{
|
||
return -7;
|
||
}
|
||
for(u_int32_t i=0; i<g_doris_server_info.iothreads; i++)
|
||
{
|
||
if(pthread_create(&thread_desc, &attr, thread_doris_http_server, NULL))
|
||
{
|
||
MESA_RUNTIME_LOGV3(g_doris_server_info.log_runtime, RLOG_LV_FATAL, "pthread_create(): %s", strerror(errno));
|
||
assert(0);return -9;
|
||
}
|
||
}
|
||
}
|
||
|
||
tv.tv_sec = g_doris_server_info.fsstat_period;
|
||
tv.tv_usec = 0;
|
||
evtimer_assign(&timer_statistic_fsstat, manage_evbase, instance_fsstat_output_timer_cb, &timer_statistic_fsstat);
|
||
evtimer_add(&timer_statistic_fsstat, &tv);
|
||
|
||
event_base_dispatch(manage_evbase);
|
||
printf("Libevent dispath error, should not run here.\n");
|
||
MESA_RUNTIME_LOGV3(g_doris_server_info.log_runtime, RLOG_LV_FATAL, "Libevent dispath error, should not run here.");
|
||
return 0;
|
||
}
|
||
|