#include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #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; } struct version_list_node *lookup_vernode_accord_version(struct version_list_handle *handle, int64_t version) { struct version_list_node *vernode; map::iterator iter; vernode = TAILQ_FIRST(&handle->version_head); if(vernode!=NULL && versionversion) //当前全量里最小的版本号 { version = vernode->version; } for(; version <= handle->latest_version; version++) { if((iter = handle->version2node->find(version)) != handle->version2node->end()) { vernode = iter->second; return vernode; } } return NULL; } void doris_http_server_meta_cb(struct evhttp_request *req, void *arg) { struct evkeyvalq params; const char *version, *bizname; int64_t verlong; char *endptr=NULL, length[64]; struct version_list_node *vernode; struct evbuffer *evbuf; struct doris_business *business; map::iterator iter; FS_operate(g_doris_server_info.fsstat_handle, g_doris_server_info.fsstat_field[DRS_FSSTAT_CLIENT_META_REQ], 0, FS_OP_ADD, 1); if(evhttp_parse_query(evhttp_request_get_uri(req), ¶ms)) { FS_operate(g_doris_server_info.fsstat_handle, g_doris_server_info.fsstat_field[DRS_FSSTAT_CLIENT_INVALID_REQ], 0, FS_OP_ADD, 1); evhttp_send_error(req, HTTP_BADREQUEST, "Parameters invalid"); return; } if(NULL == (version = evhttp_find_header(¶ms, "version"))) { FS_operate(g_doris_server_info.fsstat_handle, g_doris_server_info.fsstat_field[DRS_FSSTAT_CLIENT_INVALID_REQ], 0, FS_OP_ADD, 1); evhttp_clear_headers(¶ms); evhttp_send_error(req, HTTP_BADREQUEST, "Parameters invalid, no version found"); return; } if(0==(verlong = strtol(version, &endptr, 10)) || *endptr!='\0') { FS_operate(g_doris_server_info.fsstat_handle, g_doris_server_info.fsstat_field[DRS_FSSTAT_CLIENT_INVALID_REQ], 0, FS_OP_ADD, 1); evhttp_clear_headers(¶ms); evhttp_send_error(req, HTTP_BADREQUEST, "Parameter version invalid"); return; } if(NULL == (bizname = evhttp_find_header(¶ms, "business"))) { FS_operate(g_doris_server_info.fsstat_handle, g_doris_server_info.fsstat_field[DRS_FSSTAT_CLIENT_INVALID_REQ], 0, FS_OP_ADD, 1); evhttp_clear_headers(¶ms); evhttp_send_error(req, HTTP_BADREQUEST, "Parameters invalid, no business found"); return; } if((iter = g_doris_server_info.name2business->find(string(bizname)))==g_doris_server_info.name2business->end()) { FS_operate(g_doris_server_info.fsstat_handle, g_doris_server_info.fsstat_field[DRS_FSSTAT_CLIENT_INVALID_REQ], 0, FS_OP_ADD, 1); evhttp_clear_headers(¶ms); evhttp_send_error(req, HTTP_BADREQUEST, "Parameters invalid, business invalid"); return; } evhttp_clear_headers(¶ms); business = iter->second; pthread_rwlock_rdlock(&business->rwlock); if(NULL == (vernode = lookup_vernode_accord_version(business->cfgver_head, verlong))) { pthread_rwlock_unlock(&business->rwlock); FS_operate(g_doris_server_info.fsstat_handle, g_doris_server_info.fsstat_field[DRS_FSSTAT_SEND_META_NONEW], 0, FS_OP_ADD, 1); evhttp_send_error(req, HTTP_NOTMODIFIED, "No new configs found"); return; } evbuf = evbuffer_new(); evbuffer_add(evbuf, vernode->metacont, vernode->metalen); sprintf(length, "%u", vernode->metalen); pthread_rwlock_unlock(&business->rwlock); FS_operate(g_doris_server_info.fsstat_handle, business->fs_lineid, g_doris_server_info.fsstat_column[DRS_FSCLM_SEND_META_RES], FS_OP_ADD, 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); } struct evbuffer *evbuf_content_from_memory(struct table_list_node *tablenode, size_t start, size_t end, size_t *length) { struct evbuffer *evbuf; struct cont_frag_node *fragnode; size_t copy_len, offset=start, res_length=0; 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; } *length = res_length; return evbuf; } struct evbuffer *evbuf_content_from_disk(struct table_list_node *tablenode, size_t start, size_t end, size_t *length) { struct evbuffer *evbuf; int32_t fd; if((fd = open(tablenode->localpath, O_RDONLY, 0)) < 0) { MESA_RUNTIME_LOGV3(g_doris_server_info.log_runtime, RLOG_LV_FATAL, "Send response, open %s failed: %s", strerror(errno)); return NULL; } evbuf = evbuffer_new(); if(evbuffer_add_file(evbuf, fd, start, end-start+1)) { evbuffer_free(evbuf); MESA_RUNTIME_LOGV3(g_doris_server_info.log_runtime, RLOG_LV_FATAL, "Send response, evbuffer_add_file %s failed"); close(fd); return NULL; } *length = evbuffer_get_length(evbuf); return evbuf; } void doris_response_file_range(struct evhttp_request *req, const char *bizname, const char *tablename, int64_t verlong, size_t start, size_t end, bool range) { struct version_list_node *vernode; struct table_list_node *tablenode; struct evbuffer *evbuf; char length[128]; size_t filesize, res_length; struct doris_business *business; map::iterator iter; if((iter = g_doris_server_info.name2business->find(string(bizname)))==g_doris_server_info.name2business->end()) { FS_operate(g_doris_server_info.fsstat_handle, g_doris_server_info.fsstat_field[DRS_FSSTAT_CLIENT_INVALID_REQ], 0, FS_OP_ADD, 1); evhttp_send_error(req, HTTP_BADREQUEST, "Parameters invalid, business invalid"); return; } business = iter->second; pthread_rwlock_rdlock(&business->rwlock); if(NULL == (vernode = lookup_vernode_accord_version(business->cfgver_head, verlong))) { pthread_rwlock_unlock(&business->rwlock); FS_operate(g_doris_server_info.fsstat_handle, g_doris_server_info.fsstat_field[DRS_FSSTAT_SEND_FILE_RES_404], 0, FS_OP_ADD, 1); evhttp_send_error(req, HTTP_NOTFOUND, "Version too old"); return; } 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(&business->rwlock); FS_operate(g_doris_server_info.fsstat_handle, g_doris_server_info.fsstat_field[DRS_FSSTAT_SEND_FILE_RES_404], 0, FS_OP_ADD, 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; } if(vernode->cont_in_disk) { evbuf = evbuf_content_from_disk(tablenode, start, end, &res_length); } else { evbuf = evbuf_content_from_memory(tablenode, start, end, &res_length); } pthread_rwlock_unlock(&business->rwlock); assert(res_length == end + 1 - start); sprintf(length, "%lu", res_length); FS_operate(g_doris_server_info.fsstat_handle, business->fs_lineid, g_doris_server_info.fsstat_column[DRS_FSCLM_SEND_FILE_RES], FS_OP_ADD, 1); FS_operate(g_doris_server_info.fsstat_handle, g_doris_server_info.fsstat_field[DRS_FSSTAT_SEND_FILE_BYTES], 0, FS_OP_ADD, 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 evkeyvalq params; const char *version, *tablename, *content_range, *bizname; int64_t verlong; char *endptr=NULL; size_t req_start=0, req_end=0; FS_operate(g_doris_server_info.fsstat_handle, g_doris_server_info.fsstat_field[DRS_FSSTAT_CLIENT_FILE_REQ], 0, FS_OP_ADD, 1); if(evhttp_parse_query(evhttp_request_get_uri(req), ¶ms)) { FS_operate(g_doris_server_info.fsstat_handle, g_doris_server_info.fsstat_field[DRS_FSSTAT_CLIENT_INVALID_REQ], 0, FS_OP_ADD, 1); evhttp_send_error(req, HTTP_BADREQUEST, "Parameters invalid"); return; } if(NULL==(version=evhttp_find_header(¶ms, "version")) || NULL==(tablename=evhttp_find_header(¶ms, "tablename"))) { evhttp_clear_headers(¶ms); FS_operate(g_doris_server_info.fsstat_handle, g_doris_server_info.fsstat_field[DRS_FSSTAT_CLIENT_INVALID_REQ], 0, FS_OP_ADD, 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(¶ms); FS_operate(g_doris_server_info.fsstat_handle, g_doris_server_info.fsstat_field[DRS_FSSTAT_CLIENT_INVALID_REQ], 0, FS_OP_ADD, 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(¶ms); FS_operate(g_doris_server_info.fsstat_handle, g_doris_server_info.fsstat_field[DRS_FSSTAT_CLIENT_INVALID_REQ], 0, FS_OP_ADD, 1); evhttp_send_error(req, HTTP_BADREQUEST, "Header Range invalid"); return; } if(NULL == (bizname = evhttp_find_header(¶ms, "business"))) { FS_operate(g_doris_server_info.fsstat_handle, g_doris_server_info.fsstat_field[DRS_FSSTAT_CLIENT_INVALID_REQ], 0, FS_OP_ADD, 1); evhttp_clear_headers(¶ms); evhttp_send_error(req, HTTP_BADREQUEST, "Parameters invalid, no business found"); return; } doris_response_file_range(req, bizname, tablename, verlong, req_start, req_end, (content_range==NULL)?false:true); evhttp_clear_headers(¶ms); } 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