doris-doris-dispatch/server/doris_server_http.cpp

#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 <unistd.h> 
#include <stdio.h> 
#include <stdlib.h>
#include <string.h>
#include <assert.h>
#include <errno.h>
#include <sys/prctl.h>
#include <poll.h>

#include <event2/bufferevent_ssl.h>

#include "doris_server_main.h"
#include "doris_server_http.h"

extern struct doris_global_info g_doris_server_info;

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

void doris_http_server_meta_cb(struct evhttp_request *req, void *arg)
{
	struct worker_statistic_info *statistic=(struct worker_statistic_info *)arg;
	struct evkeyvalq params;
	const char *version;
	int64_t verlong;
	char *endptr=NULL, length[64];
	struct version_list_node *vernode;
	struct evbuffer *evbuf;
	
	statistic->statistic.field[DRS_FSSTAT_CLIENT_META_REQ] += 1;
	if(evhttp_parse_query(evhttp_request_get_uri(req), &params))
	{
		statistic->statistic.field[DRS_FSSTAT_CLIENT_INVALID_REQ] += 1;
		evhttp_send_error(req, HTTP_BADREQUEST, "Parameters invalid");
		return;
	}
	if(NULL == (version = evhttp_find_header(&params, "version")))
	{
		statistic->statistic.field[DRS_FSSTAT_CLIENT_INVALID_REQ] += 1;
		evhttp_clear_headers(&params);
		evhttp_send_error(req, HTTP_BADREQUEST, "Parameters invalid, no version found");
		return;
	}
	if(0==(verlong = strtol(version, &endptr, 10)) || *endptr!='\0')
	{
		statistic->statistic.field[DRS_FSSTAT_CLIENT_INVALID_REQ] += 1;
		evhttp_clear_headers(&params);
		evhttp_send_error(req, HTTP_BADREQUEST, "Parameter version invalid");
		return;
	}
	evhttp_clear_headers(&params);

	pthread_rwlock_rdlock(&g_doris_server_info.rwlock);
	if(verlong > g_doris_server_info.cfgver_head->latest_version)
	{
		pthread_rwlock_unlock(&g_doris_server_info.rwlock);
		statistic->statistic.field[DRS_FSSTAT_SEND_META_NONEW] += 1;
		evhttp_send_error(req, HTTP_NOTMODIFIED, "No new configs found");
		return;
	}

	vernode = TAILQ_FIRST(&g_doris_server_info.cfgver_head->version_head);
	while(vernode->version < verlong)
	{
		vernode = TAILQ_NEXT(vernode, version_node);
	}
	evbuf = evbuffer_new();
	evbuffer_add(evbuf, vernode->metacont, vernode->metalen);
	sprintf(length, "%u", vernode->metalen);
	pthread_rwlock_unlock(&g_doris_server_info.rwlock);

	statistic->statistic.field[DRS_FSSTAT_SEND_META_RES] += 1;
	evhttp_add_header(evhttp_request_get_output_headers(req), "Content-Type", "application/json");
	evhttp_add_header(evhttp_request_get_output_headers(req), "Connection", "keep-alive");
	evhttp_add_header(evhttp_request_get_output_headers(req), "Content-Length", length);
	evhttp_send_reply(req, HTTP_OK, "OK", evbuf);
	evbuffer_free(evbuf);
}

void doris_response_file_range(struct evhttp_request *req, const char *tablename,
	int64_t verlong, size_t start, size_t end, bool range, struct worker_statistic_info *statistic)
{
	struct version_list_node *vernode;
	struct table_list_node *tablenode;
	struct cont_frag_node *fragnode;
	struct evbuffer *evbuf;
	char length[128];
	size_t filesize, res_length=0, copy_len, offset=start;

	pthread_rwlock_rdlock(&g_doris_server_info.rwlock);
	if(verlong > g_doris_server_info.cfgver_head->latest_version)
	{
		pthread_rwlock_unlock(&g_doris_server_info.rwlock);
		statistic->statistic.field[DRS_FSSTAT_SEND_FILE_RES_404] += 1;
		evhttp_send_error(req, HTTP_NOTFOUND, "Version too old");
		return;
	}
	vernode = TAILQ_FIRST(&g_doris_server_info.cfgver_head->version_head);
	while(vernode->version < verlong)
	{
		vernode = TAILQ_NEXT(vernode, version_node);
	}
	tablenode = TAILQ_FIRST(&vernode->table_head);
	while(tablenode!=NULL && strcmp(tablename, tablenode->tablename))
	{
		tablenode = TAILQ_NEXT(tablenode, table_node);
	}
	if(tablenode==NULL || start>tablenode->filesize)
	{
		pthread_rwlock_unlock(&g_doris_server_info.rwlock);
		statistic->statistic.field[DRS_FSSTAT_SEND_FILE_RES_404] += 1;
		evhttp_send_error(req, HTTP_NOTFOUND, "No valid content found");
		return;
	}
	filesize = tablenode->filesize;
	if(end==0 || end >= tablenode->filesize)
	{
		end = tablenode->filesize - 1;
	}
	evbuf = evbuffer_new();
	for(fragnode=TAILQ_FIRST(&tablenode->frag_head); fragnode!=NULL && fragnode->start<=end; fragnode=TAILQ_NEXT(fragnode, frag_node))
	{
		if(offset > fragnode->end)
		{
			continue;
		}
		copy_len = (end>fragnode->end)?(fragnode->end-offset + 1):(end-offset + 1);
		evbuffer_add(evbuf, fragnode->content+(offset-fragnode->start), copy_len);
		offset += copy_len;
		res_length += copy_len;
	}
	pthread_rwlock_unlock(&g_doris_server_info.rwlock);

	assert(res_length == end + 1 - start);
	sprintf(length, "%lu", res_length);
	statistic->statistic.field[DRS_FSSTAT_SEND_FILE_RES] += 1;
	statistic->statistic.field[DRS_FSSTAT_SEND_FILE_BYTES] += res_length;
	evhttp_add_header(evhttp_request_get_output_headers(req), "Content-Length", length);
	if(range)
	{
		sprintf(length, "bytes %lu-%lu/%lu", start, end, filesize);
		evhttp_add_header(evhttp_request_get_output_headers(req), "Content-Range", length);
	}
	evhttp_add_header(evhttp_request_get_output_headers(req), "Content-Type", "application/stream");
	evhttp_add_header(evhttp_request_get_output_headers(req), "Connection", "keep-alive");
	evhttp_send_reply(req, HTTP_OK, "OK", evbuf);
	evbuffer_free(evbuf);
}

void doris_http_server_file_cb(struct evhttp_request *req, void *arg)
{
	struct worker_statistic_info *statistic=(struct worker_statistic_info *)arg;
	struct evkeyvalq params;
	const char *version, *tablename, *content_range;
	int64_t verlong;
	char *endptr=NULL;
	size_t req_start=0, req_end=0;

	statistic->statistic.field[DRS_FSSTAT_CLIENT_FILE_REQ] += 1;
	if(evhttp_parse_query(evhttp_request_get_uri(req), &params))
	{
		statistic->statistic.field[DRS_FSSTAT_CLIENT_INVALID_REQ] += 1;
		evhttp_send_error(req, HTTP_BADREQUEST, "Parameters invalid");
		return;
	}
	if(NULL==(version=evhttp_find_header(&params, "version")) || NULL==(tablename=evhttp_find_header(&params, "tablename")))
	{
		evhttp_clear_headers(&params);
		statistic->statistic.field[DRS_FSSTAT_CLIENT_INVALID_REQ] += 1;
		evhttp_send_error(req, HTTP_BADREQUEST, "Parameters invalid, no version/tablename found");
		return;
	}
	if(0==(verlong = strtol(version, &endptr, 10)) || *endptr!='\0')
	{
		evhttp_clear_headers(&params);
		statistic->statistic.field[DRS_FSSTAT_CLIENT_INVALID_REQ] += 1;
		evhttp_send_error(req, HTTP_BADREQUEST, "Parameter version invalid");
		return;
	}
	if(NULL!=(content_range = evhttp_find_header(evhttp_request_get_input_headers(req), "Range")) &&
		sscanf(content_range, "%*[^0-9]%lu-%lu", &req_start, &req_end)<1)
	{
		evhttp_clear_headers(&params);
		statistic->statistic.field[DRS_FSSTAT_CLIENT_INVALID_REQ] += 1;
		evhttp_send_error(req, HTTP_BADREQUEST, "Header Range invalid");
		return;
	}

	doris_response_file_range(req, tablename, verlong, req_start, req_end, (content_range==NULL)?false:true, statistic);
	evhttp_clear_headers(&params);
}

void doris_http_server_generic_cb(struct evhttp_request *req, void *arg)
{
	evhttp_send_error(req, HTTP_BADREQUEST, "Not Supported.");
}

pthread_t nirvana_pthreads_thread_id(void)
{
	return pthread_self();
}

void nirvana_pthreads_locking_callback(int mode, int type, const char *file, int line)
{
	if(mode & CRYPTO_LOCK)
	{
		pthread_mutex_lock(&g_doris_server_info.lock_cs[type]);
	}
	else
	{
		pthread_mutex_unlock(&g_doris_server_info.lock_cs[type]);
	}
}

int server_verify_callback(int ok, X509_STORE_CTX *ctx)
{
	X509   *client_cert;
	char *subject, *issuer;

	client_cert = X509_STORE_CTX_get_current_cert(ctx);
	
	subject = X509_NAME_oneline(X509_get_subject_name(client_cert), 0, 0);
	issuer = X509_NAME_oneline(X509_get_issuer_name(client_cert), 0, 0);

	MESA_RUNTIME_LOGV3(g_doris_server_info.log_runtime, RLOG_LV_DEBUG, "ClientCert suject: %s, issuer: %s, state: %d.", subject, issuer, ok);
	OPENSSL_free(subject);
	OPENSSL_free(issuer);
	return ok;
}

SSL_CTX *doris_connections_create_ssl_ctx(void)
{
	int crypto_num;
	SSL_CTX *ssl_ctx;
	char session_id_appname[] = "DorisServer";

	SSL_library_init();
	SSLeay_add_ssl_algorithms();
	OpenSSL_add_all_algorithms();
	SSL_load_error_strings();
	ERR_load_BIO_strings();
	
	crypto_num = CRYPTO_num_locks();
	g_doris_server_info.lock_cs = (pthread_mutex_t *)OPENSSL_malloc(crypto_num * sizeof(pthread_mutex_t));
	for(int i=0; i<crypto_num; i++)
	{
		pthread_mutex_init(&g_doris_server_info.lock_cs[i], NULL);
	}
	CRYPTO_set_id_callback(nirvana_pthreads_thread_id);
	CRYPTO_set_locking_callback(nirvana_pthreads_locking_callback);

	ssl_ctx = SSL_CTX_new(SSLv23_server_method());
	//SSL_CTX_set_verify(ssl_ctx, SSL_VERIFY_PEER|SSL_VERIFY_FAIL_IF_NO_PEER_CERT|SSL_VERIFY_CLIENT_ONCE, server_verify_callback);
	//<2F><><EFBFBD><EFBFBD>SESSION Resumption<6F><6E>˫<EFBFBD><CBAB><EFBFBD><EFBFBD><EFBFBD>ã<EFBFBD><C3A3><EFBFBD>Ϊ<EFBFBD><CEAA>֤<EFBFBD><D6A4>˫<EFBFBD><CBAB><EFBFBD>ģ<EFBFBD>
	SSL_CTX_set_session_cache_mode(ssl_ctx, SSL_SESS_CACHE_BOTH);
	//<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>HoldĬ<64><C4AC>SSL_SESSION_CACHE_MAX_SIZE_DEFAULT(1024*20)<29><>SESSION<4F><4E>0-<2D><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
	SSL_CTX_sess_set_cache_size(ssl_ctx, SSL_SESSION_CACHE_MAX_SIZE_DEFAULT);
	SSL_CTX_set_session_id_context(ssl_ctx, (unsigned char*)session_id_appname, strlen(session_id_appname));
	SSL_CTX_set_default_passwd_cb_userdata(ssl_ctx, g_doris_server_info.ssl_key_passwd);
	
	if(!SSL_CTX_load_verify_locations(ssl_ctx, NULL, g_doris_server_info.ssl_CA_path))
	{
		MESA_RUNTIME_LOGV3(g_doris_server_info.log_runtime, RLOG_LV_FATAL, "SSL_CTX_load_verify_locations error: %s.", ERR_reason_error_string(ERR_get_error()));
		SSL_CTX_free(ssl_ctx);
		return NULL;
	}
	if(!SSL_CTX_use_certificate_file(ssl_ctx, g_doris_server_info.ssl_cert_file, SSL_FILETYPE_PEM))
	{
		MESA_RUNTIME_LOGV3(g_doris_server_info.log_runtime, RLOG_LV_FATAL, "SSL_CTX_use_certificate_file error: %s.", ERR_reason_error_string(ERR_get_error()));
		SSL_CTX_free(ssl_ctx);
		return NULL;
	}
	if(SSL_CTX_use_PrivateKey_file(ssl_ctx, g_doris_server_info.ssl_key_file, SSL_FILETYPE_PEM) < 0)
	{
		MESA_RUNTIME_LOGV3(g_doris_server_info.log_runtime, RLOG_LV_FATAL, "SSL_CTX_use_PrivateKey_file_pass error: %s.", ERR_reason_error_string(ERR_get_error()));
		SSL_CTX_free(ssl_ctx);
		return NULL;
	}
	if(!SSL_CTX_check_private_key(ssl_ctx))
	{
		MESA_RUNTIME_LOGV3(g_doris_server_info.log_runtime, RLOG_LV_FATAL, "SSL_CTX_check_private_key error: %s.", ERR_reason_error_string(ERR_get_error()));
		SSL_CTX_free(ssl_ctx);
		return NULL;
	}
	return ssl_ctx;
}

struct bufferevent *doris_https_bufferevent_cb(struct event_base *evabse, void *arg)
{	
	SSL_CTX *ssl_instance = (SSL_CTX *)arg;

	return bufferevent_openssl_socket_new(evabse, -1, SSL_new(ssl_instance), BUFFEREVENT_SSL_ACCEPTING, BEV_OPT_CLOSE_ON_FREE);
}

void* thread_doris_http_server(void *arg)
{
	struct event_base *worker_evbase;
	struct evhttp *worker_http;
	struct worker_statistic_info statistic;
	struct timeval tv;

	prctl(PR_SET_NAME, "http_server");

	memset(&statistic, 0, sizeof(struct worker_statistic_info));
	worker_evbase = event_base_new();

	worker_http = evhttp_new(worker_evbase);

	if(g_doris_server_info.ssl_conn_on)
	{
		g_doris_server_info.ssl_instance = doris_connections_create_ssl_ctx();
		if(g_doris_server_info.ssl_instance == NULL)
		{
			assert(0);return NULL;
		}
		evhttp_set_bevcb(worker_http, doris_https_bufferevent_cb, g_doris_server_info.ssl_instance);
	}

	evhttp_set_cb(worker_http, "/configmeta", doris_http_server_meta_cb, &statistic);
	evhttp_set_cb(worker_http, "/configfile", doris_http_server_file_cb, &statistic);
	evhttp_set_gencb(worker_http, doris_http_server_generic_cb, &statistic);
	evhttp_set_allowed_methods(worker_http, EVHTTP_REQ_GET|EVHTTP_REQ_HEAD);

	if(evhttp_accept_socket(worker_http, g_doris_server_info.listener))
	{
		printf("evhttp_accept_socket %d error!\n", g_doris_server_info.listener);  
		assert(0); return NULL;
	}

	evtimer_assign(&statistic.timer_statistic, worker_evbase, doris_worker_statistic_timer_cb, &statistic);
	tv.tv_sec = g_doris_server_info.fsstat_period;
	tv.tv_usec = 0;
	evtimer_add(&statistic.timer_statistic, &tv);

	event_base_dispatch(worker_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 NULL;
}