chaoyli closed pull request #441: Add HttpClient class
URL: https://github.com/apache/incubator-doris/pull/441
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/be/CMakeLists.txt b/be/CMakeLists.txt
index 3db6acaf..a598d564 100644
--- a/be/CMakeLists.txt
+++ b/be/CMakeLists.txt
@@ -559,6 +559,7 @@ add_subdirectory(${SRC_DIR}/udf)
add_subdirectory(${SRC_DIR}/runtime)
add_subdirectory(${SRC_DIR}/testutil)
add_subdirectory(${SRC_DIR}/tools)
+# add_subdirectory(${SRC_DIR}/udf_samples)
# Utility CMake function to make specifying tests and benchmarks less verbose
FUNCTION(ADD_BE_TEST TEST_NAME)
diff --git a/be/src/agent/CMakeLists.txt b/be/src/agent/CMakeLists.txt
index e523a509..378890ca 100644
--- a/be/src/agent/CMakeLists.txt
+++ b/be/src/agent/CMakeLists.txt
@@ -24,7 +24,6 @@ set(EXECUTABLE_OUTPUT_PATH "${BUILD_DIR}/src/agent")
add_library(Agent STATIC
agent_server.cpp
pusher.cpp
- file_downloader.cpp
heartbeat_server.cpp
task_worker_pool.cpp
utils.cpp
diff --git a/be/src/agent/file_downloader.cpp b/be/src/agent/file_downloader.cpp
deleted file mode 100644
index 0d2f5353..00000000
--- a/be/src/agent/file_downloader.cpp
+++ /dev/null
@@ -1,439 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-#include "agent/file_downloader.h"
-#include <cstdio>
-#include <cstring>
-#include <fstream>
-#include <iostream>
-#include <sstream>
-#include "olap/olap_define.h"
-#include "olap/file_helper.h"
-#include "olap/utils.h"
-
-using std::ofstream;
-using std::ostream;
-using std::string;
-using std::stringstream;
-
-namespace doris {
-
-FileDownloader::FileDownloader(const FileDownloaderParam& param) :
- _downloader_param(param) {
-}
-
-size_t FileDownloader::_write_file_callback(
- void* buffer, size_t size, size_t nmemb, void* param) {
- OLAPStatus status = OLAP_SUCCESS;
- size_t len = size * nmemb;
-
- if (param == NULL) {
- status = OLAP_ERR_OTHER_ERROR;
- OLAP_LOG_WARNING("File downloader output file handler is NULL
pointer.");
- return -1;
- }
-
- if (status == OLAP_SUCCESS) {
- status = static_cast<FileHandler*>(param)->write(buffer, len);
- if (status != OLAP_SUCCESS) {
- OLAP_LOG_WARNING("File downloader callback write failed.");
- }
- }
-
- return len;
-}
-
-size_t FileDownloader::_write_stream_callback(
- void* buffer, size_t size, size_t nmemb, void* param) {
- AgentStatus status = DORIS_SUCCESS;
- size_t len = size * nmemb;
-
- if (param == NULL) {
- status = DORIS_FILE_DOWNLOAD_INVALID_PARAM;
- OLAP_LOG_WARNING("File downloader output stream is NULL pointer.");
- return -1;
- }
-
- if (status == DORIS_SUCCESS) {
- static_cast<ostream*>(param)->write(static_cast<const char*>(buffer),
len);
- }
-
- return len;
-}
-
-AgentStatus FileDownloader::_install_opt(
- OutputType output_type, CURL* curl, char* errbuf,
- stringstream* output_stream, FileHandler* file_handler) {
- AgentStatus status = DORIS_SUCCESS;
- CURLcode curl_ret = CURLE_OK;
-
- // Set request URL
- curl_ret = curl_easy_setopt(curl, CURLOPT_URL,
_downloader_param.remote_file_path.c_str());
-
- if (curl_ret != CURLE_OK) {
- status = DORIS_FILE_DOWNLOAD_INSTALL_OPT_FAILED;
- OLAP_LOG_WARNING("curl setopt URL failed.[error=%s]",
curl_easy_strerror(curl_ret));
- }
-
- // Set username
- if (status == DORIS_SUCCESS) {
- curl_ret = curl_easy_setopt(curl, CURLOPT_USERNAME,
_downloader_param.username.c_str());
-
- if (curl_ret != CURLE_OK) {
- status = DORIS_FILE_DOWNLOAD_INSTALL_OPT_FAILED;
- OLAP_LOG_WARNING("curl setopt USERNAME failed.[error=%s]",
- curl_easy_strerror(curl_ret));
- }
- }
-
- // Set password
- if (status == DORIS_SUCCESS) {
- curl_ret = curl_easy_setopt(curl, CURLOPT_PASSWORD,
_downloader_param.password.c_str());
-
- if (curl_ret != CURLE_OK) {
- status = DORIS_FILE_DOWNLOAD_INSTALL_OPT_FAILED;
- OLAP_LOG_WARNING("curl setopt USERNAME failed.[error=%s]",
- curl_easy_strerror(curl_ret));
- }
- }
-
- // Set process timeout
- if (status == DORIS_SUCCESS) {
- curl_ret = curl_easy_setopt(curl, CURLOPT_TIMEOUT,
_downloader_param.curl_opt_timeout);
-
- if (curl_ret != CURLE_OK) {
- status = DORIS_FILE_DOWNLOAD_INSTALL_OPT_FAILED;
- OLAP_LOG_WARNING("curl setopt TIMEOUT failed.[error=%s]",
curl_easy_strerror(curl_ret));
- }
- }
-
- // Set low speed limit and low speed time
- if (status == DORIS_SUCCESS) {
- curl_ret = curl_easy_setopt(
- curl, CURLOPT_LOW_SPEED_LIMIT,
- config::download_low_speed_limit_kbps * 1024);
-
- if (curl_ret != CURLE_OK) {
- status = DORIS_FILE_DOWNLOAD_INSTALL_OPT_FAILED;
- OLAP_LOG_WARNING(
- "curl setopt CURLOPT_LOW_SPEED_LIMIT failed.[error=%s]",
- curl_easy_strerror(curl_ret));
- }
- }
-
- if (status == DORIS_SUCCESS) {
- curl_ret = curl_easy_setopt(
- curl, CURLOPT_LOW_SPEED_TIME, config::download_low_speed_time);
-
- if (curl_ret != CURLE_OK) {
- status = DORIS_FILE_DOWNLOAD_INSTALL_OPT_FAILED;
- OLAP_LOG_WARNING(
- "curl setopt CURLOPT_LOW_SPEED_TIME failed.[error=%s]",
- curl_easy_strerror(curl_ret));
- }
- }
-
- // Set max recv speed(bytes/s)
- if (status == DORIS_SUCCESS) {
- curl_ret = curl_easy_setopt(
- curl, CURLOPT_MAX_RECV_SPEED_LARGE,
config::max_download_speed_kbps * 1024);
-
- if (curl_ret != CURLE_OK) {
- status = DORIS_FILE_DOWNLOAD_INSTALL_OPT_FAILED;
- OLAP_LOG_WARNING(
- "curl setopt MAX_RECV_SPEED failed.[error=%s]",
curl_easy_strerror(curl_ret));
- }
- }
-
- // Forbid signals
- if (status == DORIS_SUCCESS) {
- curl_ret = curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1L);
-
- if (curl_ret != CURLE_OK) {
- status = DORIS_FILE_DOWNLOAD_INSTALL_OPT_FAILED;
- OLAP_LOG_WARNING("curl setopt nosignal failed.[error=%s]",
- curl_easy_strerror(curl_ret));
- }
- }
-
- if (strncmp(_downloader_param.remote_file_path.c_str(), "http", 4) == 0) {
- if (status == DORIS_SUCCESS) {
- curl_ret = curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L);
-
- if (curl_ret != CURLE_OK) {
- OLAP_LOG_WARNING(
- "curl setopt CURLOPT_FOLLOWLOCATION failed.[error=%s]",
- curl_easy_strerror(curl_ret));
- }
- }
-
- if (status == DORIS_SUCCESS) {
- curl_ret = curl_easy_setopt(curl, CURLOPT_MAXREDIRS, 20);
-
- if (curl_ret != CURLE_OK) {
- OLAP_LOG_WARNING(
- "curl setopt CURLOPT_MAXREDIRS failed.[error=%s]",
- curl_easy_strerror(curl_ret));
- }
- }
- }
-
- // Set nobody
- if (status == DORIS_SUCCESS) {
- if (output_type == OutputType::NONE) {
- curl_ret = curl_easy_setopt(curl, CURLOPT_NOBODY, 1L);
- if (curl_ret != CURLE_OK) {
- OLAP_LOG_WARNING("curl setopt CURLOPT_NOBODY
failed.[error=%s]",
- curl_easy_strerror(curl_ret));
- }
- } else if (output_type == OutputType::STREAM) {
- // Set callback function
- curl_ret = curl_easy_setopt(
- curl,
- CURLOPT_WRITEFUNCTION,
- &FileDownloader::_write_stream_callback);
-
- if (curl_ret != CURLE_OK) {
- status = DORIS_FILE_DOWNLOAD_INSTALL_OPT_FAILED;
- OLAP_LOG_WARNING("curl setopt WRITEDATA failed.[error=%s]",
- curl_easy_strerror(curl_ret));
- }
-
- // Set callback function args
- if (status == DORIS_SUCCESS) {
- curl_ret = curl_easy_setopt(curl, CURLOPT_WRITEDATA,
- static_cast<void*>(output_stream));
-
- if (curl_ret != CURLE_OK) {
- status = DORIS_FILE_DOWNLOAD_INSTALL_OPT_FAILED;
- OLAP_LOG_WARNING("curl setopt WRITEDATA failed.[error=%s]",
- curl_easy_strerror(curl_ret));
- }
- }
- } else if (output_type == OutputType::FILE) {
- // Set callback function
- curl_ret = curl_easy_setopt(
- curl,
- CURLOPT_WRITEFUNCTION,
- &FileDownloader::_write_file_callback);
-
- if (curl_ret != CURLE_OK) {
- status = DORIS_FILE_DOWNLOAD_INSTALL_OPT_FAILED;
- OLAP_LOG_WARNING("curl setopt WRITEDATA failed.[error=%s]",
- curl_easy_strerror(curl_ret));
- }
-
- // Set callback function args
- if (status == DORIS_SUCCESS) {
- curl_ret = curl_easy_setopt(curl, CURLOPT_WRITEDATA,
- static_cast<void*>(file_handler));
-
- if (curl_ret != CURLE_OK) {
- status = DORIS_FILE_DOWNLOAD_INSTALL_OPT_FAILED;
- OLAP_LOG_WARNING("curl setopt WRITEDATA failed.[error=%s]",
- curl_easy_strerror(curl_ret));
- }
- }
- }
- }
-
- // set verbose mode
- /*
- if (status == DORIS_SUCCESS) {
- curl_easy_setopt(curl, CURLOPT_VERBOSE, config::curl_verbose_mode);
- if (curl_ret != CURLE_OK) {
- status = DORIS_FILE_DOWNLOAD_INSTALL_OPT_FAILED;
- OLAP_LOG_WARNING("curl setopt VERBOSE MODE failed.[error=%s]",
- curl_easy_strerror(curl_ret));
- }
- }
- */
-
- // set err buf
- if (status == DORIS_SUCCESS) {
- curl_easy_setopt(curl, CURLOPT_ERRORBUFFER, errbuf);
- if (curl_ret != CURLE_OK) {
- status = DORIS_FILE_DOWNLOAD_INSTALL_OPT_FAILED;
- OLAP_LOG_WARNING("curl setopt ERR BUF failed.[error=%s]",
- curl_easy_strerror(curl_ret));
- }
- errbuf[0] = 0;
- }
-
- return status;
-}
-
-void FileDownloader::_get_err_info(char * errbuf, CURLcode res) {
- if (res != CURLE_OK) {
- size_t len = strlen(errbuf);
- if (len) {
- OLAP_LOG_WARNING("(%d): %s%s", res, errbuf, ((errbuf[len - 1] !=
'\n') ? "\n" : ""));
- } else {
- OLAP_LOG_WARNING("(%d): %s", res, curl_easy_strerror(res));
- }
- }
-}
-
-AgentStatus FileDownloader::get_length(uint64_t* length) {
- AgentStatus status = DORIS_SUCCESS;
- CURL* curl = NULL;
- CURLcode curl_ret = CURLE_OK;
- curl = curl_easy_init();
-
- // Init curl
- if (curl == NULL) {
- status = DORIS_FILE_DOWNLOAD_CURL_INIT_FAILED;
- OLAP_LOG_WARNING("internal error to get NULL curl");
- }
-
- // Set curl opt
- char errbuf[CURL_ERROR_SIZE];
- if (status == DORIS_SUCCESS) {
- status = _install_opt(OutputType::NONE, curl, errbuf, NULL, NULL);
-
- if (DORIS_SUCCESS != status) {
- OLAP_LOG_WARNING("install curl opt failed.");
- }
- }
-
- // Get result
- if (status == DORIS_SUCCESS) {
- curl_ret = curl_easy_perform(curl);
-
- if (curl_ret != CURLE_OK) {
- status = DORIS_FILE_DOWNLOAD_GET_LENGTH_FAILED;
- OLAP_LOG_WARNING("curl get length failed.[path=%s]",
- _downloader_param.remote_file_path.c_str());
- _get_err_info(errbuf, curl_ret);
- } else {
- double content_length = 0.0f;
- curl_easy_getinfo(curl, CURLINFO_CONTENT_LENGTH_DOWNLOAD,
&content_length);
- *length = (uint64_t)content_length;
- }
- }
-
- if (curl != NULL) {
- curl_easy_cleanup(curl);
- }
-
- return status;
-}
-
-AgentStatus FileDownloader::download_file() {
- AgentStatus status = DORIS_SUCCESS;
- CURL* curl = NULL;
- CURLcode curl_ret = CURLE_OK;
- curl = curl_easy_init();
-
- if (curl == NULL) {
- status = DORIS_FILE_DOWNLOAD_CURL_INIT_FAILED;
- OLAP_LOG_WARNING("internal error to get NULL curl");
- }
-
- FileHandler* file_handler = new FileHandler();
- OLAPStatus olap_status = OLAP_SUCCESS;
- // Prepare some infomation
- if (status == DORIS_SUCCESS) {
- olap_status = file_handler->open_with_mode(
- _downloader_param.local_file_path,
- O_CREAT | O_TRUNC | O_WRONLY, S_IRUSR | S_IWUSR);
-
- if (olap_status != OLAP_SUCCESS) {
- status = DORIS_FILE_DOWNLOAD_INVALID_PARAM;
- OLAP_LOG_WARNING("open loacal file failed.[file_path=%s]",
- _downloader_param.local_file_path.c_str());
- }
- }
-
- char errbuf[CURL_ERROR_SIZE];
- if (status == DORIS_SUCCESS) {
- status = _install_opt(OutputType::FILE, curl, errbuf, NULL,
file_handler);
-
- if (DORIS_SUCCESS != status) {
- OLAP_LOG_WARNING("install curl opt failed.");
- }
- }
-
- if (status == DORIS_SUCCESS) {
- curl_ret = curl_easy_perform(curl);
-
- if (curl_ret != CURLE_OK) {
- status = DORIS_FILE_DOWNLOAD_FAILED;
- OLAP_LOG_WARNING(
- "curl easy perform failed.[path=%s]",
- _downloader_param.remote_file_path.c_str());
- _get_err_info(errbuf, curl_ret);
- }
- }
-
- if (file_handler != NULL) {
- file_handler->close();
- delete file_handler;
- file_handler = NULL;
- }
-
- if (curl != NULL) {
- curl_easy_cleanup(curl);
- }
-
- return status;
-}
-
-AgentStatus FileDownloader::list_file_dir(string* file_list_string) {
- AgentStatus status = DORIS_SUCCESS;
- CURL* curl = NULL;
- CURLcode curl_ret = CURLE_OK;
- curl = curl_easy_init();
-
- // Init curl
- if (curl == NULL) {
- status = DORIS_FILE_DOWNLOAD_CURL_INIT_FAILED;
- OLAP_LOG_WARNING("internal error to get NULL curl");
- }
-
- stringstream output_string_stream;
- // Set curl opt
- char errbuf[CURL_ERROR_SIZE];
- if (status == DORIS_SUCCESS) {
- status = _install_opt(OutputType::STREAM, curl, errbuf,
&output_string_stream, NULL);
-
- if (DORIS_SUCCESS != status) {
- OLAP_LOG_WARNING("install curl opt failed.");
- }
- }
-
- // Get result
- if (status == DORIS_SUCCESS) {
- curl_ret = curl_easy_perform(curl);
-
- if (curl_ret != CURLE_OK) {
- status = DORIS_FILE_DOWNLOAD_LIST_DIR_FAIL;
- OLAP_LOG_WARNING(
- "curl list file dir failed.[path=%s]",
- _downloader_param.remote_file_path.c_str());
- _get_err_info(errbuf, curl_ret);
- }
- }
-
- if (curl != NULL) {
- curl_easy_cleanup(curl);
- }
- *file_list_string = output_string_stream.str();
-
- return status;
-}
-} // namespace doris
diff --git a/be/src/agent/file_downloader.h b/be/src/agent/file_downloader.h
deleted file mode 100644
index aaaae6ac..00000000
--- a/be/src/agent/file_downloader.h
+++ /dev/null
@@ -1,83 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-#ifndef DORIS_BE_SRC_AGENT_FILE_DOWNLOADER_H
-#define DORIS_BE_SRC_AGENT_FILE_DOWNLOADER_H
-
-#include <iostream>
-#include <pthread.h>
-#include <cstdint>
-#include <sstream>
-#include <string>
-#include "curl/curl.h"
-#include "agent/status.h"
-#include "olap/olap_define.h"
-#include "olap/file_helper.h"
-
-namespace doris {
-
-const uint32_t GET_LENGTH_TIMEOUT = 10;
-const uint32_t CURL_OPT_CONNECTTIMEOUT = 120;
-// down load file
-class FileDownloader {
-public:
- enum OutputType {
- NONE,
- STREAM,
- FILE
- };
-
- struct FileDownloaderParam {
- std::string username;
- std::string password;
- std::string remote_file_path;
- std::string local_file_path;
- uint32_t curl_opt_timeout;
- };
-
- explicit FileDownloader(const FileDownloaderParam& param);
- virtual ~FileDownloader() {};
-
- // Download file from remote server
- virtual AgentStatus download_file();
-
- // List remote dir file
- virtual AgentStatus list_file_dir(std::string* file_list_string);
-
- // Get file length of remote file
- //
- // Output parameters:
- // * length: The pointer of size of remote file
- virtual AgentStatus get_length(uint64_t* length);
-private:
- static size_t _write_file_callback(
- void* buffer, size_t size, size_t nmemb, void* downloader);
- static size_t _write_stream_callback(
- void* buffer, size_t size, size_t nmemb, void* downloader);
-
- AgentStatus _install_opt(
- OutputType output_type, CURL* curl, char* errbuf,
- std::stringstream* output_stream, FileHandler* file_handler);
-
- void _get_err_info(char * errbuf, CURLcode res);
-
- const FileDownloaderParam& _downloader_param;
-
- DISALLOW_COPY_AND_ASSIGN(FileDownloader);
-}; // class FileDownloader
-} // namespace doris
-#endif // DORIS_BE_SRC_AGENT_FILE_DOWNLOADER_H
diff --git a/be/src/agent/pusher.cpp b/be/src/agent/pusher.cpp
index fc79bfb6..cbf2f70e 100644
--- a/be/src/agent/pusher.cpp
+++ b/be/src/agent/pusher.cpp
@@ -26,12 +26,13 @@
#include "boost/filesystem.hpp"
#include "boost/lexical_cast.hpp"
#include "agent/cgroups_mgr.h"
-#include "agent/file_downloader.h"
#include "gen_cpp/AgentService_types.h"
+#include "http/http_client.h"
#include "olap/olap_common.h"
#include "olap/olap_define.h"
#include "olap/olap_engine.h"
#include "olap/olap_table.h"
+#include "util/stopwatch.hpp"
using std::list;
using std::string;
@@ -39,10 +40,8 @@ using std::vector;
namespace doris {
-
Pusher::Pusher(OLAPEngine* engine, const TPushReq& push_req) :
_push_req(push_req), _engine(engine) {
- _download_status = DORIS_SUCCESS;
}
Pusher::~Pusher() {
@@ -51,11 +50,6 @@ Pusher::~Pusher() {
AgentStatus Pusher::init() {
AgentStatus status = DORIS_SUCCESS;
- if (_is_init) {
- VLOG(3) << "has been inited";
- return status;
- }
-
// Check replica exist
OLAPTablePtr olap_table;
olap_table = _engine->get_table(
@@ -64,44 +58,30 @@ AgentStatus Pusher::init() {
if (olap_table.get() == NULL) {
OLAP_LOG_WARNING("get tables failed. tablet_id: %ld, schema_hash: %ld",
_push_req.tablet_id, _push_req.schema_hash);
- status = DORIS_PUSH_INVALID_TABLE;
+ return DORIS_PUSH_INVALID_TABLE;
}
// Empty remote_path
- if (status == DORIS_SUCCESS && !_push_req.__isset.http_file_path) {
- _is_init = true;
+ if (!_push_req.__isset.http_file_path) {
return status;
}
// Check remote path
- string remote_full_path;
- string tmp_file_dir;
+ _remote_file_path = _push_req.http_file_path;
- if (status == DORIS_SUCCESS) {
- remote_full_path = _push_req.http_file_path;
-
- // Get local download path
- LOG(INFO) << "start get file. remote_full_path: " << remote_full_path;
- string root_path = olap_table->storage_root_path_name();
-
- status = _get_tmp_file_dir(root_path, &tmp_file_dir);
- if (DORIS_SUCCESS != status) {
- LOG(WARNING) << "get local path failed. tmp file dir: " <<
tmp_file_dir;
- }
- }
+ // Get local download path
+ LOG(INFO) << "start get file. remote_file_path: " << _remote_file_path;
// Set download param
- if (status == DORIS_SUCCESS) {
- string tmp_file_name;
- _get_file_name_from_path(_push_req.http_file_path, &tmp_file_name);
-
- _downloader_param.username = "";
- _downloader_param.password = "";
- _downloader_param.remote_file_path = remote_full_path;
- _downloader_param.local_file_path = tmp_file_dir + "/" + tmp_file_name;
-
- _is_init = true;
+ string tmp_file_dir;
+ status = _get_tmp_file_dir(olap_table->storage_root_path_name(),
&tmp_file_dir);
+ if (status != DORIS_SUCCESS) {
+ LOG(WARNING) << "get local path failed. tmp file dir: " <<
tmp_file_dir;
+ return status;
}
+ string tmp_file_name;
+ _get_file_name_from_path(_push_req.http_file_path, &tmp_file_name);
+ _local_file_path = tmp_file_dir + "/" + tmp_file_name;
return status;
}
@@ -129,38 +109,6 @@ AgentStatus Pusher::_get_tmp_file_dir(const string&
root_path, string* download_
return status;
}
-AgentStatus Pusher::_download_file() {
- LOG(INFO) << "begin download file. tablet_id=" << _push_req.tablet_id;
- time_t start = time(NULL);
- AgentStatus status = DORIS_SUCCESS;
-
- status = _file_downloader->download_file();
-
- time_t cost = time(NULL) - start;
- if (cost <= 0) {
- cost = 1;
- }
- // KB/s
- double rate = -1.0;
- if (_push_req.__isset.http_file_size) {
- rate = (double) _push_req.http_file_size / cost / 1024;
- }
- if (status == DORIS_SUCCESS) {
- LOG(INFO) << "down load file success. local_file=" <<
_downloader_param.local_file_path
- << ", remote_file=" << _downloader_param.remote_file_path
- << ", tablet_id" << _push_req.tablet_id
- << ", cost=" << cost << ", file_size" <<
_push_req.http_file_size
- << ", download rage:" << rate << "KB/s";
- } else {
- LOG(WARNING) << "down load file failed. remote_file=" <<
_downloader_param.remote_file_path
- << ", tablet=" << _push_req.tablet_id
- << ", cost=" << cost << " file size: " <<
_push_req.http_file_size << " B";
- }
-
- // todo check data length and mv name tmp
- return status;
-}
-
void Pusher::_get_file_name_from_path(const string& file_path, string*
file_name) {
size_t found = file_path.find_last_of("/\\");
pthread_t tid = pthread_self();
@@ -169,83 +117,80 @@ void Pusher::_get_file_name_from_path(const string&
file_path, string* file_name
AgentStatus Pusher::process(vector<TTabletInfo>* tablet_infos) {
AgentStatus status = DORIS_SUCCESS;
-
- if (!_is_init) {
- OLAP_LOG_WARNING("has not init yet. tablet_id: %d",
_push_req.tablet_id);
- return DORIS_ERROR;
- }
-
// Remote file not empty, need to download
if (_push_req.__isset.http_file_path) {
- // Get file length
+ // Get file length and timeout
uint64_t file_size = 0;
uint64_t estimate_time_out = DEFAULT_DOWNLOAD_TIMEOUT;
if (_push_req.__isset.http_file_size) {
file_size = _push_req.http_file_size;
- estimate_time_out =
- file_size / config::download_low_speed_limit_kbps / 1024;
+ estimate_time_out = file_size /
config::download_low_speed_limit_kbps / 1024;
}
if (estimate_time_out < config::download_low_speed_time) {
estimate_time_out = config::download_low_speed_time;
}
-
- // Download file from hdfs
- for (uint32_t i = 0; i < MAX_RETRY; ++i) {
+ bool is_timeout = false;
+ auto download_cb = [this, estimate_time_out, file_size, &is_timeout]
(HttpClient* client) {
// Check timeout and set timeout
time_t now = time(NULL);
-
- if (_push_req.timeout > 0) {
+ if (_push_req.timeout > 0 && _push_req.timeout < now) {
+ // return status to break this callback
VLOG(3) << "check time out. time_out:" << _push_req.timeout
- << ", now:" << now;
- if (_push_req.timeout < now) {
- OLAP_LOG_WARNING("push time out");
- status = DORIS_PUSH_TIME_OUT;
- break;
- }
+ << ", now:" << now;
+ is_timeout = true;
+ return Status::OK;
}
- _downloader_param.curl_opt_timeout = estimate_time_out;
+ RETURN_IF_ERROR(client->init(_remote_file_path));
+ // sent timeout
uint64_t timeout = _push_req.timeout > 0 ? _push_req.timeout - now
: 0;
if (timeout > 0 && timeout < estimate_time_out) {
- _downloader_param.curl_opt_timeout = timeout;
+ client->set_timeout_ms(timeout * 1000);
+ } else {
+ client->set_timeout_ms(estimate_time_out * 1000);
}
- VLOG(3) << "estimate_time_out: " << estimate_time_out
- << ", download_timeout: " << estimate_time_out
- << ", curl_opt_timeout: " <<
_downloader_param.curl_opt_timeout
- << ", download file, retry time:" << i;
-#ifndef BE_TEST
- _file_downloader = new FileDownloader(_downloader_param);
- _download_status = _download_file();
- if (_file_downloader != NULL) {
- delete _file_downloader;
- _file_downloader = NULL;
- }
-#endif
+ // download remote file
+ RETURN_IF_ERROR(client->download(_local_file_path));
- status = _download_status;
- if (_push_req.__isset.http_file_size && status == DORIS_SUCCESS) {
+ // check file size
+ if (_push_req.__isset.http_file_size) {
// Check file size
- boost::filesystem::path
local_file_path(_downloader_param.local_file_path);
- uint64_t local_file_size =
boost::filesystem::file_size(local_file_path);
- VLOG(3) << "file_size: " << file_size
- << ", local_file_size: " << local_file_size;
-
+ uint64_t local_file_size =
boost::filesystem::file_size(_local_file_path);
if (file_size != local_file_size) {
- OLAP_LOG_WARNING(
- "download_file size error. file_size: %d,
local_file_size: %d",
- file_size, local_file_size);
- status = DORIS_FILE_DOWNLOAD_FAILED;
+ LOG(WARNING) << "download_file size error. file_size=" <<
file_size
+ << ", local_file_size=" << local_file_size;
+ return Status("downloaded file's size isn't right");
}
}
-
- if (status == DORIS_SUCCESS) {
- _push_req.http_file_path = _downloader_param.local_file_path;
- break;
+ // NOTE: change http_file_path is not good design
+ _push_req.http_file_path = _local_file_path;
+ return Status::OK;
+ };
+
+ MonotonicStopWatch stopwatch;
+ stopwatch.start();
+ auto st = HttpClient::execute_with_retry(MAX_RETRY, 1, download_cb);
+ auto cost = stopwatch.elapsed_time();
+ if (cost <= 0) {
+ cost = 1;
+ }
+ if (st.ok() && !is_timeout) {
+ double rate = -1.0;
+ if (_push_req.__isset.http_file_size) {
+ rate = (double) _push_req.http_file_size / (cost / 1000 / 1000
/ 1000) / 1024;
}
-#ifndef BE_TEST
- sleep(config::sleep_one_second);
-#endif
+ LOG(INFO) << "down load file success. local_file=" <<
_local_file_path
+ << ", remote_file=" << _remote_file_path
+ << ", tablet_id" << _push_req.tablet_id
+ << ", cost=" << cost / 1000 << "us, file_size" <<
_push_req.http_file_size
+ << ", download rage:" << rate << "KB/s";
+ } else {
+ LOG(WARNING) << "down load file failed. remote_file=" <<
_remote_file_path
+ << ", tablet=" << _push_req.tablet_id
+ << ", cost=" << cost / 1000
+ << "us, errmsg=" << st.get_error_msg() << ", is_timeout=" <<
is_timeout;
+ status = DORIS_ERROR;
}
}
@@ -263,14 +208,13 @@ AgentStatus Pusher::process(vector<TTabletInfo>*
tablet_infos) {
}
// Delete download file
- boost::filesystem::path
download_file_path(_downloader_param.local_file_path);
- if (boost::filesystem::exists(download_file_path)) {
- if (remove(_downloader_param.local_file_path.c_str()) == -1) {
- OLAP_LOG_WARNING("can not remove file: %s",
- _downloader_param.local_file_path.c_str());
+ if (boost::filesystem::exists(_local_file_path)) {
+ if (remove(_local_file_path.c_str()) == -1) {
+ LOG(WARNING) << "can not remove file=" << _local_file_path;
}
}
return status;
}
+
} // namespace doris
diff --git a/be/src/agent/pusher.h b/be/src/agent/pusher.h
index 9e2979e7..56b9b7d6 100644
--- a/be/src/agent/pusher.h
+++ b/be/src/agent/pusher.h
@@ -20,7 +20,6 @@
#include <utility>
#include <vector>
-#include "agent/file_downloader.h"
#include "agent/status.h"
#include "gen_cpp/AgentService_types.h"
#include "gen_cpp/MasterService_types.h"
@@ -49,15 +48,12 @@ class Pusher {
private:
AgentStatus _get_tmp_file_dir(const std::string& root_path, std::string*
local_path);
- AgentStatus _download_file();
void _get_file_name_from_path(const std::string& file_path, std::string*
file_name);
- bool _is_init = false;
TPushReq _push_req;
- FileDownloader::FileDownloaderParam _downloader_param;
OLAPEngine* _engine;
- FileDownloader* _file_downloader;
- AgentStatus _download_status;
+ std::string _remote_file_path;
+ std::string _local_file_path;
DISALLOW_COPY_AND_ASSIGN(Pusher);
}; // class Pusher
diff --git a/be/src/agent/task_worker_pool.cpp
b/be/src/agent/task_worker_pool.cpp
index 9cb9df17..6d933378 100644
--- a/be/src/agent/task_worker_pool.cpp
+++ b/be/src/agent/task_worker_pool.cpp
@@ -36,6 +36,7 @@
#include "agent/utils.h"
#include "gen_cpp/FrontendService.h"
#include "gen_cpp/Types_types.h"
+#include "http/http_client.h"
#include "olap/olap_common.h"
#include "olap/olap_engine.h"
#include "olap/olap_table.h"
@@ -74,6 +75,9 @@ const std::string HTTP_REQUEST_PREFIX =
"/api/_tablet/_download?";
const std::string HTTP_REQUEST_TOKEN_PARAM = "token=";
const std::string HTTP_REQUEST_FILE_PARAM = "&file=";
+const uint32_t GET_LENGTH_TIMEOUT = 10;
+const uint32_t CURL_OPT_CONNECTTIMEOUT = 120;
+
std::atomic_ulong TaskWorkerPool::_s_report_version(time(NULL) * 10000);
Mutex TaskWorkerPool::_s_task_signatures_lock;
Mutex TaskWorkerPool::_s_running_task_user_count_lock;
@@ -1329,7 +1333,8 @@ AgentStatus TaskWorkerPool::_clone_copy(
}
} else {
LOG(WARNING) << "make snapshot failed. tablet_id: " <<
clone_req.tablet_id
- << ". schema_hash: " << clone_req.schema_hash << ".
backend_ip: " << src_host->host
+ << ". schema_hash: " << clone_req.schema_hash
+ << ". backend_ip: " << src_host->host
<< ". backend_port: " << src_host->be_port << ".
signature: " << signature;
error_msgs->push_back("make snapshot failed. backend_ip: " +
src_host->host);
status = DORIS_ERROR;
@@ -1349,7 +1354,6 @@ AgentStatus TaskWorkerPool::_clone_copy(
string src_file_full_path = src_file_full_path_stream.str();
string local_file_full_path = local_file_full_path_stream.str();
-#ifndef BE_TEST
// Check local path exist, if exist, remove it, then create the dir
if (status == DORIS_SUCCESS) {
boost::filesystem::path local_file_full_dir(local_file_full_path);
@@ -1358,63 +1362,34 @@ AgentStatus TaskWorkerPool::_clone_copy(
}
boost::filesystem::create_directories(local_file_full_dir);
}
-#endif
// Get remove dir file list
- FileDownloader::FileDownloaderParam downloader_param;
- downloader_param.remote_file_path = http_host + HTTP_REQUEST_PREFIX
+ HttpClient client;
+ std::string remote_file_path = http_host + HTTP_REQUEST_PREFIX
+ HTTP_REQUEST_TOKEN_PARAM + token
+ HTTP_REQUEST_FILE_PARAM + src_file_full_path;
- downloader_param.curl_opt_timeout = LIST_REMOTE_FILE_TIMEOUT;
-
-#ifndef BE_TEST
- FileDownloader* file_downloader_ptr = new
FileDownloader(downloader_param);
- if (file_downloader_ptr == NULL) {
- OLAP_LOG_WARNING("clone copy create file downloader failed. try
next backend");
- status = DORIS_ERROR;
- }
-#endif
string file_list_str;
- AgentStatus download_status = DORIS_SUCCESS;
- uint32_t download_retry_time = 0;
- while (status == DORIS_SUCCESS && download_retry_time <
DOWNLOAD_FILE_MAX_RETRY) {
-#ifndef BE_TEST
- download_status =
file_downloader_ptr->list_file_dir(&file_list_str);
-#else
- download_status =
_file_downloader_ptr->list_file_dir(&file_list_str);
-#endif
- if (download_status != DORIS_SUCCESS) {
- OLAP_LOG_WARNING("clone get remote file list failed.
backend_ip: %s, "
- "src_file_path: %s, signature: %ld",
- src_host->host.c_str(),
- downloader_param.remote_file_path.c_str(),
- signature);
- ++download_retry_time;
-#ifndef BE_TEST
- sleep(download_retry_time);
-#endif
- } else {
- break;
- }
- }
-
-#ifndef BE_TEST
- if (file_downloader_ptr != NULL) {
- delete file_downloader_ptr;
- file_downloader_ptr = NULL;
- }
-#endif
-
- vector<string> file_name_list;
- if (download_status != DORIS_SUCCESS) {
+ auto list_files_cb = [&remote_file_path, &file_list_str] (HttpClient*
client) {
+ RETURN_IF_ERROR(client->init(remote_file_path));
+ client->set_timeout_ms(LIST_REMOTE_FILE_TIMEOUT * 1000);
+ RETURN_IF_ERROR(client->execute(&file_list_str));
+ return Status::OK;
+ };
+
+ Status download_status = HttpClient::execute_with_retry(
+ DOWNLOAD_FILE_MAX_RETRY, 1, list_files_cb);
+ if (!download_status.ok()) {
OLAP_LOG_WARNING("clone get remote file list failed over max time.
backend_ip: %s, "
"src_file_path: %s, signature: %ld",
src_host->host.c_str(),
- downloader_param.remote_file_path.c_str(),
+ remote_file_path.c_str(),
signature);
status = DORIS_ERROR;
- } else {
+ }
+
+ vector<string> file_name_list;
+ if (status == DORIS_SUCCESS) {
size_t start_position = 0;
size_t end_position = file_list_str.find("\n");
@@ -1446,127 +1421,64 @@ AgentStatus TaskWorkerPool::_clone_copy(
}
// Get copy from remote
- for (auto file_name : file_name_list) {
- download_retry_time = 0;
- downloader_param.remote_file_path = http_host + HTTP_REQUEST_PREFIX
+ for (auto& file_name : file_name_list) {
+ remote_file_path = http_host + HTTP_REQUEST_PREFIX
+ HTTP_REQUEST_TOKEN_PARAM + token
+ HTTP_REQUEST_FILE_PARAM + src_file_full_path + file_name;
- downloader_param.local_file_path = local_file_full_path +
file_name;
- // Get file length
+ // get file length
uint64_t file_size = 0;
- uint64_t estimate_time_out = 0;
-
- downloader_param.curl_opt_timeout = GET_LENGTH_TIMEOUT;
-#ifndef BE_TEST
- file_downloader_ptr = new FileDownloader(downloader_param);
- if (file_downloader_ptr == NULL) {
- OLAP_LOG_WARNING("clone copy create file downloader failed.
try next backend");
+ auto get_file_size_cb = [&remote_file_path, &file_size]
(HttpClient* client) {
+ RETURN_IF_ERROR(client->init(remote_file_path));
+ client->set_timeout_ms(GET_LENGTH_TIMEOUT * 1000);
+ RETURN_IF_ERROR(client->head());
+ file_size = client->get_content_length();
+ return Status::OK;
+ };
+ download_status = HttpClient::execute_with_retry(
+ DOWNLOAD_FILE_MAX_RETRY, 1, get_file_size_cb);
+ if (!download_status.ok()) {
+ LOG(WARNING) << "clone copy get file length failed over max
time. remote_path="
+ << remote_file_path
+ << ", signature=" << signature;
status = DORIS_ERROR;
break;
}
-#endif
- while (download_retry_time < DOWNLOAD_FILE_MAX_RETRY) {
-#ifndef BE_TEST
- download_status = file_downloader_ptr->get_length(&file_size);
-#else
- download_status = _file_downloader_ptr->get_length(&file_size);
-#endif
- if (download_status != DORIS_SUCCESS) {
- OLAP_LOG_WARNING("clone copy get file length failed.
backend_ip: %s, "
- "src_file_path: %s, signature: %ld",
- src_host->host.c_str(),
- downloader_param.remote_file_path.c_str(),
- signature);
- ++download_retry_time;
-#ifndef BE_TEST
- sleep(download_retry_time);
-#endif
- } else {
- break;
- }
- }
-#ifndef BE_TEST
- if (file_downloader_ptr != NULL) {
- delete file_downloader_ptr;
- file_downloader_ptr = NULL;
- }
-#endif
- if (download_status != DORIS_SUCCESS) {
- OLAP_LOG_WARNING("clone copy get file length failed over max
time. "
- "backend_ip: %s, src_file_path: %s,
signature: %ld",
- src_host->host.c_str(),
- downloader_param.remote_file_path.c_str(),
- signature);
- status = DORIS_ERROR;
- break;
- }
-
- estimate_time_out = file_size /
config::download_low_speed_limit_kbps / 1024;
+ uint64_t estimate_time_out = file_size /
config::download_low_speed_limit_kbps / 1024;
if (estimate_time_out < config::download_low_speed_time) {
estimate_time_out = config::download_low_speed_time;
}
- // Download the file
- download_retry_time = 0;
- downloader_param.curl_opt_timeout = estimate_time_out;
-#ifndef BE_TEST
- file_downloader_ptr = new FileDownloader(downloader_param);
- if (file_downloader_ptr == NULL) {
- OLAP_LOG_WARNING("clone copy create file downloader failed.
try next backend");
- status = DORIS_ERROR;
- break;
- }
-#endif
- while (download_retry_time < DOWNLOAD_FILE_MAX_RETRY) {
-#ifndef BE_TEST
- download_status = file_downloader_ptr->download_file();
-#else
- download_status = _file_downloader_ptr->download_file();
-#endif
- if (download_status != DORIS_SUCCESS) {
- OLAP_LOG_WARNING("download file failed. backend_ip: %s, "
- "src_file_path: %s, signature: %ld",
- src_host->host.c_str(),
- downloader_param.remote_file_path.c_str(),
- signature);
- } else {
- // Check file length
- boost::filesystem::path
local_file_path(downloader_param.local_file_path);
- uint64_t local_file_size =
boost::filesystem::file_size(local_file_path);
- if (local_file_size != file_size) {
- OLAP_LOG_WARNING("download file length error.
backend_ip: %s, "
- "src_file_path: %s, signature: %ld,"
- "remote file size: %d, local file
size: %d",
- src_host->host.c_str(),
-
downloader_param.remote_file_path.c_str(),
- signature, file_size,
local_file_size);
- download_status = DORIS_FILE_DOWNLOAD_FAILED;
- } else {
- chmod(downloader_param.local_file_path.c_str(),
S_IRUSR | S_IWUSR);
- break;
- }
+ std::string local_file_path = local_file_full_path + file_name;
+
+ auto download_cb = [&remote_file_path,
+ estimate_time_out,
+ &local_file_path,
+ file_size] (HttpClient* client) {
+ RETURN_IF_ERROR(client->init(remote_file_path));
+ client->set_timeout_ms(estimate_time_out * 1000);
+ RETURN_IF_ERROR(client->download(local_file_path));
+
+ // Check file length
+ uint64_t local_file_size =
boost::filesystem::file_size(local_file_path);
+ if (local_file_size != file_size) {
+ LOG(WARNING) << "download file length error"
+ << ", remote_path=" << remote_file_path
+ << ", file_size=" << file_size
+ << ", local_file_size=" << local_file_size;
+ return Status("downloaded file size is not equal");
}
- ++download_retry_time;
-#ifndef BE_TEST
- sleep(download_retry_time);
-#endif
- } // Try to download a file from remote backend
-
-#ifndef BE_TEST
- if (file_downloader_ptr != NULL) {
- delete file_downloader_ptr;
- file_downloader_ptr = NULL;
- }
-#endif
-
- if (download_status != DORIS_SUCCESS) {
- OLAP_LOG_WARNING("download file failed over max retry.
backend_ip: %s, "
- "src_file_path: %s, signature: %ld",
- src_host->host.c_str(),
- downloader_param.remote_file_path.c_str(),
- signature);
+ chmod(local_file_path.c_str(), S_IRUSR | S_IWUSR);
+ return Status::OK;
+ };
+ download_status = HttpClient::execute_with_retry(
+ DOWNLOAD_FILE_MAX_RETRY, 1, download_cb);
+ if (!download_status.ok()) {
+ LOG(WARNING) << "download file failed over max retry."
+ << ", remote_path=" << remote_file_path
+ << ", signature=" << signature
+ << ", errormsg=" << download_status.get_error_msg();
status = DORIS_ERROR;
break;
}
diff --git a/be/src/agent/task_worker_pool.h b/be/src/agent/task_worker_pool.h
index 750dbcf2..05b5ae53 100644
--- a/be/src/agent/task_worker_pool.h
+++ b/be/src/agent/task_worker_pool.h
@@ -158,7 +158,6 @@ class TaskWorkerPool {
ExecEnv* _env;
#ifdef BE_TEST
AgentServerClient* _agent_client;
- FileDownloader* _file_downloader_ptr;
Pusher * _pusher;
#endif
diff --git a/be/src/http/CMakeLists.txt b/be/src/http/CMakeLists.txt
index 2e785984..38e63458 100644
--- a/be/src/http/CMakeLists.txt
+++ b/be/src/http/CMakeLists.txt
@@ -35,6 +35,7 @@ add_library(Webserver STATIC
default_path_handlers.cpp
utils.cpp
ev_http_server.cpp
+ http_client.cpp
action/mini_load.cpp
action/health_action.cpp
action/checksum_action.cpp
diff --git a/be/src/http/http_client.cpp b/be/src/http/http_client.cpp
new file mode 100644
index 00000000..e8c4834b
--- /dev/null
+++ b/be/src/http/http_client.cpp
@@ -0,0 +1,204 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "http/http_client.h"
+
+namespace doris {
+
+HttpClient::HttpClient() {
+}
+
+HttpClient::~HttpClient() {
+ if (_curl != nullptr) {
+ curl_easy_cleanup(_curl);
+ _curl = nullptr;
+ }
+}
+
+Status HttpClient::init(const std::string& url) {
+ if (_curl == nullptr) {
+ _curl = curl_easy_init();
+ if (_curl == nullptr) {
+ return Status("fail to initalize curl");
+ }
+ } else {
+ curl_easy_reset(_curl);
+ }
+
+ // set error_buf
+ _error_buf[0] = 0;
+ auto code = curl_easy_setopt(_curl, CURLOPT_ERRORBUFFER, _error_buf);
+ if (code != CURLE_OK) {
+ LOG(WARNING) << "fail to set CURLOPT_ERRORBUFFER, msg=" <<
_to_errmsg(code);
+ return Status("fail to set error buffer");
+ }
+ // forbid signals
+ code = curl_easy_setopt(_curl, CURLOPT_NOSIGNAL, 1L);
+ if (code != CURLE_OK) {
+ LOG(WARNING) << "fail to set CURLOPT_NOSIGNAL, msg=" <<
_to_errmsg(code);
+ return Status("fail to set CURLOPT_NOSIGNAL");
+ }
+ // set fail on error
+ code = curl_easy_setopt(_curl, CURLOPT_FAILONERROR, 1L);
+ if (code != CURLE_OK) {
+ LOG(WARNING) << "fail to set CURLOPT_FAILONERROR, msg=" <<
_to_errmsg(code);
+ return Status("fail to set CURLOPT_FAILONERROR");
+ }
+ // set redirect
+ code = curl_easy_setopt(_curl, CURLOPT_FOLLOWLOCATION, 1L);
+ if (code != CURLE_OK) {
+ LOG(WARNING) << "fail to set CURLOPT_FOLLOWLOCATION, msg=" <<
_to_errmsg(code);
+ return Status("fail to set CURLOPT_FOLLOWLOCATION");
+ }
+ code = curl_easy_setopt(_curl, CURLOPT_MAXREDIRS, 20);
+ if (code != CURLE_OK) {
+ LOG(WARNING) << "fail to set CURLOPT_MAXREDIRS, msg=" <<
_to_errmsg(code);
+ return Status("fail to set CURLOPT_MAXREDIRS");
+ }
+
+ curl_write_callback callback = [] (char* buffer, size_t size, size_t
nmemb, void* param) {
+ HttpClient* client = (HttpClient*)param;
+ return client->on_response_data(buffer, size * nmemb);
+ };
+
+ // set callback function
+ code = curl_easy_setopt(_curl, CURLOPT_WRITEFUNCTION, callback);
+ if (code != CURLE_OK) {
+ LOG(WARNING) << "fail to set CURLOPT_WRITEFUNCTION, msg=" <<
_to_errmsg(code);
+ return Status("fail to set CURLOPT_WRITEFUNCTION");
+ }
+ code = curl_easy_setopt(_curl, CURLOPT_WRITEDATA, (void*) this);
+ if (code != CURLE_OK) {
+ LOG(WARNING) << "fail to set CURLOPT_WRITEDATA, msg=" <<
_to_errmsg(code);
+ return Status("fail to set CURLOPT_WRITEDATA");
+ }
+ // set url
+ code = curl_easy_setopt(_curl, CURLOPT_URL, url.c_str());
+ if (code != CURLE_OK) {
+ LOG(WARNING) << "failed to set CURLOPT_URL, errmsg=" <<
_to_errmsg(code);
+ return Status("fail to set CURLOPT_URL");
+ }
+
+ return Status::OK;
+}
+
+void HttpClient::set_method(HttpMethod method) {
+ switch (method) {
+ case GET:
+ curl_easy_setopt(_curl, CURLOPT_HTTPGET, 1L);
+ return;
+ case PUT:
+ curl_easy_setopt(_curl, CURLOPT_UPLOAD, 1L);
+ return;
+ case POST:
+ curl_easy_setopt(_curl, CURLOPT_POST, 1L);
+ return;
+ case DELETE:
+ curl_easy_setopt(_curl, CURLOPT_CUSTOMREQUEST, "DELETE");
+ return;
+ case HEAD:
+ curl_easy_setopt(_curl, CURLOPT_NOBODY, 1L);
+ return;
+ case OPTIONS:
+ curl_easy_setopt(_curl, CURLOPT_CUSTOMREQUEST, "OPTIONS");
+ return;
+ default:
+ return;
+ }
+}
+
+size_t HttpClient::on_response_data(const void* data, size_t length) {
+ if (*_callback != nullptr) {
+ bool is_continue = (*_callback)(data, length);
+ if (!is_continue) {
+ return -1;
+ }
+ }
+ return length;
+}
+
+Status HttpClient::execute(const std::function<bool(const void* data, size_t
length)>& callback) {
+ _callback = &callback;
+ auto code = curl_easy_perform(_curl);
+ if (code != CURLE_OK) {
+ LOG(WARNING) << "fail to execute HTTP client, errmsg=" <<
_to_errmsg(code);
+ return Status("fail to execute HTTP client");
+ }
+ return Status::OK;
+}
+
+Status HttpClient::download(const std::string& local_path) {
+ // set method to GET
+ set_method(GET);
+
+ // TODO(zc) Move this download speed limit outside to limit download speed
+ // at system level
+ curl_easy_setopt(_curl, CURLOPT_LOW_SPEED_LIMIT,
+ config::download_low_speed_limit_kbps * 1024);
+ curl_easy_setopt(_curl, CURLOPT_LOW_SPEED_TIME,
config::download_low_speed_time);
+ curl_easy_setopt(_curl, CURLOPT_MAX_RECV_SPEED_LARGE,
+ config::max_download_speed_kbps * 1024);
+
+ auto fp_closer = [] (FILE*fp) { fclose(fp); };
+ std::unique_ptr<FILE, decltype(fp_closer)> fp(fopen(local_path.c_str(),
"w"), fp_closer);
+ if (fp == nullptr) {
+ LOG(WARNING) << "open file failed, file=" << local_path;
+ return Status("open file failed");
+ }
+ auto callback = [&fp, &local_path] (const void* data, size_t length) {
+ auto res = fwrite(data, length, 1, fp.get());
+ if (res != 1) {
+ LOG(WARNING) << "fail to write data to file, file=" << local_path
+ << ", error=" << ferror(fp.get());
+ return false;
+ }
+ return true;
+ };
+ return execute(callback);
+}
+
+Status HttpClient::execute(std::string* response) {
+ auto callback = [response] (const void* data, size_t length) {
+ response->append((char*)data, length);
+ return true;
+ };
+ return execute(callback);
+}
+
+const char* HttpClient::_to_errmsg(CURLcode code) {
+ if (_error_buf[0] == 0) {
+ return curl_easy_strerror(code);
+ }
+ return _error_buf;
+}
+
+Status HttpClient::execute_with_retry(int retry_times, int sleep_time,
+ const
std::function<Status(HttpClient*)>& callback) {
+ Status status;
+ for (int i = 0; i < retry_times; ++i) {
+ HttpClient client;
+ status = callback(&client);
+ if (status.ok()) {
+ return status;
+ }
+ sleep(sleep_time);
+ }
+ return status;
+
+}
+
+}
diff --git a/be/src/http/http_client.h b/be/src/http/http_client.h
new file mode 100644
index 00000000..4f8c29cf
--- /dev/null
+++ b/be/src/http/http_client.h
@@ -0,0 +1,116 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <cstdio>
+#include <string>
+
+#include <curl/curl.h>
+
+#include "common/status.h"
+#include "http/http_headers.h"
+#include "http/http_method.h"
+#include "http/utils.h"
+
+namespace doris {
+
+// Helper class to access HTTP resource
+class HttpClient {
+public:
+ HttpClient();
+ ~HttpClient();
+
+ // you can call this function to execute HTTP request with retry,
+ // if callback return OK, this function will end and return OK.
+ // This function will return FAIL if three are more than retry_times
+ // that callback return FAIL.
+ static Status execute_with_retry(int retry_times, int sleep_time,
+ const std::function<Status(HttpClient*)>&
callback);
+
+ // this function must call before other function,
+ // you can call this multiple times to reuse this object
+ Status init(const std::string& url);
+
+ void set_method(HttpMethod method);
+
+ void set_basic_auth(const std::string& user, const std::string& passwd) {
+ curl_easy_setopt(_curl, CURLOPT_HTTPAUTH, CURLAUTH_BASIC);
+ curl_easy_setopt(_curl, CURLOPT_USERNAME, user.c_str());
+ curl_easy_setopt(_curl, CURLOPT_PASSWORD, passwd.c_str());
+ }
+
+ // TODO(zc): support set header
+ // void set_header(const std::string& key, const std::string& value) {
+ // _cntl.http_request().SetHeader(key, value);
+ // }
+
+ std::string get_response_content_type() {
+ char *ct = nullptr;
+ auto code = curl_easy_getinfo(_curl, CURLINFO_CONTENT_TYPE, &ct);
+ if(code == CURLE_OK && ct != nullptr) {
+ return ct;
+ }
+ return std::string();
+ }
+
+ // Set the long gohead parameter to 1L to continue send authentication
(user+password)
+ // credentials when following locations, even when hostname changed.
+ void set_unrestricted_auth(int gohead) {
+ curl_easy_setopt(_curl, CURLOPT_UNRESTRICTED_AUTH, gohead);
+ }
+
+ void set_timeout_ms(int64_t timeout_ms) {
+ curl_easy_setopt(_curl, CURLOPT_TIMEOUT_MS, timeout_ms);
+ }
+
+ // used to get content length
+ int64_t get_content_length() const {
+ double cl = 0.0f;
+ curl_easy_getinfo(_curl, CURLINFO_CONTENT_LENGTH_DOWNLOAD, &cl);
+ return cl;
+ }
+
+ // execute a head method
+ Status head() {
+ set_method(HEAD);
+ return execute();
+ }
+
+ // helper function to download a file, you can call this function to
downlaod
+ // a file to local_path
+ Status download(const std::string& local_path);
+
+ // execute a simple method, and its response is saved in response argument
+ Status execute(std::string* response);
+
+ // execute remote call action
+ Status execute(const std::function<bool(const void* data, size_t length)>&
callback = {});
+
+ size_t on_response_data(const void* data, size_t length);
+
+private:
+ const char* _to_errmsg(CURLcode code);
+
+private:
+ CURL* _curl = nullptr;
+ using HttpCallback = std::function<bool(const void* data, size_t length)>;
+ const HttpCallback* _callback = nullptr;
+ char _error_buf[CURL_ERROR_SIZE];
+};
+
+}
diff --git a/be/src/http/http_request.cpp b/be/src/http/http_request.cpp
index 95149a36..cb2a31f6 100644
--- a/be/src/http/http_request.cpp
+++ b/be/src/http/http_request.cpp
@@ -121,9 +121,9 @@ const std::string& HttpRequest::param(const std::string&
key) const {
}
void HttpRequest::add_output_header(const char* key, const char* value) {
-#ifndef BE_TEST
+// #ifndef BE_TEST
evhttp_add_header(evhttp_request_get_output_headers(_ev_req), key, value);
-#endif
+// #endif
}
std::string HttpRequest::get_request_body() {
diff --git a/be/src/http/http_request.h b/be/src/http/http_request.h
index 547916e5..f464cb8b 100644
--- a/be/src/http/http_request.h
+++ b/be/src/http/http_request.h
@@ -37,8 +37,6 @@ class HttpHandler;
class HttpRequest {
public:
- // Only used for unit test
- HttpRequest() { }
HttpRequest(evhttp_request* ev_req);
diff --git a/be/src/http/utils.cpp b/be/src/http/utils.cpp
index 486173f1..2e8be225 100644
--- a/be/src/http/utils.cpp
+++ b/be/src/http/utils.cpp
@@ -25,6 +25,14 @@
namespace doris {
+std::string encode_basic_auth(const std::string& user, const std::string&
passwd) {
+ std::string auth = user + ":" + passwd;
+ std::string encoded_auth;
+ base64_encode(auth, &encoded_auth);
+ static std::string s_prefix = "Basic ";
+ return s_prefix + encoded_auth;
+}
+
bool parse_basic_auth(const HttpRequest& req, std::string* user, std::string*
passwd) {
const char k_basic[] = "Basic ";
auto& auth = req.header(HttpHeaders::AUTHORIZATION);
diff --git a/be/src/http/utils.h b/be/src/http/utils.h
index 8495ef24..b388e6af 100644
--- a/be/src/http/utils.h
+++ b/be/src/http/utils.h
@@ -26,6 +26,7 @@ namespace doris {
class HttpRequest;
class HttpAuthInfo;
+std::string encode_basic_auth(const std::string& user, const std::string&
passwd);
// parse Basic authorization
// return true, if request contain valid basic authorization.
// Otherwise return fasle
diff --git a/be/src/olap/olap_engine.cpp b/be/src/olap/olap_engine.cpp
index c5f60f1f..4d9df32e 100644
--- a/be/src/olap/olap_engine.cpp
+++ b/be/src/olap/olap_engine.cpp
@@ -33,7 +33,6 @@
#include <rapidjson/document.h>
#include <thrift/protocol/TDebugProtocol.h>
-#include "agent/file_downloader.h"
#include "olap/base_compaction.h"
#include "olap/cumulative_compaction.h"
#include "olap/lru_cache.h"
diff --git a/be/test/agent/CMakeLists.txt b/be/test/agent/CMakeLists.txt
index 98c15246..9f2e8e8a 100644
--- a/be/test/agent/CMakeLists.txt
+++ b/be/test/agent/CMakeLists.txt
@@ -23,8 +23,7 @@ set(EXECUTABLE_OUTPUT_PATH "${BUILD_DIR}/test/agent")
# ADD_BE_TEST(agent_server_test)
ADD_BE_TEST(cgroups_mgr_test)
-ADD_BE_TEST(file_downloader_test)
#ADD_BE_TEST(heartbeat_server_test)
-ADD_BE_TEST(pusher_test)
+# ADD_BE_TEST(pusher_test)
# ADD_BE_TEST(task_worker_pool_test)
ADD_BE_TEST(utils_test)
diff --git a/be/test/agent/file_downloader_test.cpp
b/be/test/agent/file_downloader_test.cpp
deleted file mode 100644
index 32a0530e..00000000
--- a/be/test/agent/file_downloader_test.cpp
+++ /dev/null
@@ -1,130 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-#include <sstream>
-#include "curl/curl.h"
-#include "gtest/gtest.h"
-#include "gmock/gmock.h"
-#include "agent/file_downloader.h"
-#include "olap/file_helper.h"
-#include "util/logging.h"
-
-using ::testing::_;
-using ::testing::Return;
-using ::testing::SetArgPointee;
-using std::string;
-using std::stringstream;
-
-namespace doris {
-
-TEST(FileDownloaderTest, TestWriteFileCallback) {
- FileDownloader::FileDownloaderParam param;
- FileDownloader file_downloader(param);
- char buffer[] = {'x', 'y', 'z'};
-
- // _local_file is NULL
- size_t len = file_downloader._write_file_callback(
- buffer,
- sizeof(char),
- sizeof(buffer),
- NULL);
- EXPECT_EQ(-1, len);
-
- // set _local_file
- FileHandler* file_handler = new FileHandler();
- file_handler->open_with_mode("./test_data/local_file",
- O_CREAT | O_TRUNC | O_WRONLY, S_IRUSR | S_IWUSR);
- len = file_downloader._write_file_callback(
- buffer,
- sizeof(char),
- sizeof(buffer),
- file_handler);
- EXPECT_EQ(3, len);
-
- delete file_handler;
- file_handler = NULL;
-}
-
-TEST(FileDownloaderTest, TestWriteStreamCallback) {
- FileDownloader::FileDownloaderParam param;
- FileDownloader file_downloader(param);
- char buffer[] = {'x', 'y', 'z'};
-
- // _local_file is NULL
- size_t len = file_downloader._write_stream_callback(
- buffer,
- sizeof(char),
- sizeof(buffer),
- NULL);
- EXPECT_EQ(-1, len);
-
- // set _local_file
- stringstream* output_string_stream = new stringstream();
- len = file_downloader._write_stream_callback(
- buffer,
- sizeof(char),
- sizeof(buffer),
- output_string_stream);
- EXPECT_EQ(3, len);
-
- delete output_string_stream;
- output_string_stream = NULL;
-}
-
-TEST(FileDownloaderTest, TestInstallOpt) {
- CURL* curl = curl_easy_init();
- FileDownloader::FileDownloaderParam param;
- param.username = "username";
- param.password = "password";
- param.remote_file_path = "http://xxx";;
- param.local_file_path = "./test_data/download_file";
- param.curl_opt_timeout = 100;
-
- FileDownloader file_downloader(param);
- AgentStatus ret = DORIS_SUCCESS;
-
- // Test for get length
- char errbuf[CURL_ERROR_SIZE];
- ret = file_downloader._install_opt(FileDownloader::OutputType::NONE, curl,
errbuf, NULL, NULL);
- EXPECT_EQ(DORIS_SUCCESS, ret);
-
- // Test for get dir list
- stringstream* output_string_stream = NULL;
- ret = file_downloader._install_opt(
- FileDownloader::OutputType::STREAM, curl, errbuf,
output_string_stream, NULL);
- EXPECT_EQ(DORIS_SUCCESS, ret);
-
- // Test for downlod file
- FileHandler* file_handler = new FileHandler();
- ret = file_downloader._install_opt(
- FileDownloader::OutputType::FILE, curl, errbuf, NULL,
file_handler);
- EXPECT_EQ(DORIS_SUCCESS, ret);
-}
-
-} // namespace doris
-
-int main(int argc, char **argv) {
- std::string conffile = std::string(getenv("DORIS_HOME")) + "/conf/be.conf";
- if (!doris::config::init(conffile.c_str(), false)) {
- fprintf(stderr, "error read config file. \n");
- return -1;
- }
- doris::init_glog("be-test");
- ::testing::InitGoogleTest(&argc, argv);
- return RUN_ALL_TESTS();
-}
-
diff --git a/be/test/agent/mock_file_downloader.h
b/be/test/agent/mock_file_downloader.h
deleted file mode 100644
index 8118f405..00000000
--- a/be/test/agent/mock_file_downloader.h
+++ /dev/null
@@ -1,34 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-#ifndef DORIS_BE_SRC_AGENT_MOCK_MOCK_FILE_DOWNLOADER_H
-#define DORIS_BE_SRC_AGENT_MOCK_MOCK_FILE_DOWNLOADER_H
-
-#include "gmock/gmock.h"
-#include "agent/file_downloader.h"
-
-namespace doris {
-
-class MockFileDownloader : public FileDownloader {
-public:
- explicit MockFileDownloader(const FileDownloaderParam& param);
- MOCK_METHOD0(download_file, AgentStatus());
- MOCK_METHOD1(list_file_dir, AgentStatus(std::string* file_list_string));
- MOCK_METHOD1(get_length, AgentStatus(uint64_t* length));
-}; // class MockFileDownloader
-} // namespace doris
-#endif // DORIS_BE_SRC_AGENT_MOCK_MOCK_FILE_DOWNLOADER_H
diff --git a/be/test/http/CMakeLists.txt b/be/test/http/CMakeLists.txt
index 46bdc646..f81ab025 100644
--- a/be/test/http/CMakeLists.txt
+++ b/be/test/http/CMakeLists.txt
@@ -22,3 +22,4 @@ ADD_BE_TEST(metrics_action_test)
ADD_BE_TEST(message_body_sink_test)
ADD_BE_TEST(http_utils_test)
ADD_BE_TEST(stream_load_test)
+ADD_BE_TEST(http_client_test)
diff --git a/be/test/http/http_client_test.cpp
b/be/test/http/http_client_test.cpp
new file mode 100644
index 00000000..f33c3060
--- /dev/null
+++ b/be/test/http/http_client_test.cpp
@@ -0,0 +1,122 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "http/http_client.h"
+
+#include <gtest/gtest.h>
+
+#include "common/logging.h"
+#include "http/ev_http_server.h"
+#include "http/http_channel.h"
+#include "http/http_handler.h"
+#include "http/http_request.h"
+
+namespace doris {
+
+class HttpClientTestSimpleGetHandler : public HttpHandler {
+public:
+ void handle(HttpRequest* req) override {
+ std::string user;
+ std::string passwd;
+ if (!parse_basic_auth(*req, &user, &passwd) || user != "test1") {
+ HttpChannel::send_basic_challenge(req, "abc");
+ return;
+ }
+ req->add_output_header(HttpHeaders::CONTENT_TYPE, "text/plain;
version=0.0.4");
+ if (req->method() == HttpMethod::HEAD) {
+ req->add_output_header(HttpHeaders::CONTENT_LENGTH,
std::to_string(5).c_str());
+ HttpChannel::send_reply(req);
+ } else {
+ std::string response = "test1";
+ HttpChannel::send_reply(req, response);
+ }
+ }
+};
+
+static HttpClientTestSimpleGetHandler s_simple_get_handler =
HttpClientTestSimpleGetHandler();
+static EvHttpServer* s_server = nullptr;
+
+class HttpClientTest : public testing::Test {
+public:
+ HttpClientTest() { }
+ ~HttpClientTest() override { }
+
+ static void SetUpTestCase() {
+ s_server = new EvHttpServer(29386);
+ s_server->register_handler(GET, "/simple_get", &s_simple_get_handler);
+ s_server->register_handler(HEAD, "/simple_get", &s_simple_get_handler);
+ s_server->start();
+ }
+
+ static void TearDownTestCase() {
+ delete s_server;
+ }
+};
+
+TEST_F(HttpClientTest, get_normal) {
+ HttpClient client;
+ auto st = client.init("http://127.0.0.1:29386/simple_get";);
+ ASSERT_TRUE(st.ok());
+ client.set_method(GET);
+ client.set_basic_auth("test1", "");
+ std::string response;
+ st = client.execute(&response);
+ ASSERT_TRUE(st.ok());
+ ASSERT_STREQ("test1", response.c_str());
+
+ // for head
+ st = client.init("http://127.0.0.1:29386/simple_get";);
+ ASSERT_TRUE(st.ok());
+ client.set_method(HEAD);
+ client.set_basic_auth("test1", "");
+ st = client.execute();
+ ASSERT_TRUE(st.ok());
+ ASSERT_EQ(5, client.get_content_length());
+}
+
+TEST_F(HttpClientTest, download) {
+ HttpClient client;
+ auto st = client.init("http://127.0.0.1:29386/simple_get";);
+ ASSERT_TRUE(st.ok());
+ client.set_basic_auth("test1", "");
+ std::string local_file = ".http_client_test.dat";
+ st = client.download(local_file);
+ ASSERT_TRUE(st.ok());
+ char buf[50];
+ auto fp = fopen(local_file.c_str(), "r");
+ auto size = fread(buf, 1, 50, fp);
+ buf[size] = 0;
+ ASSERT_STREQ("test1", buf);
+ unlink(local_file.c_str());
+}
+
+TEST_F(HttpClientTest, get_failed) {
+ HttpClient client;
+ auto st = client.init("http://127.0.0.1:29386/simple_get";);
+ ASSERT_TRUE(st.ok());
+ client.set_method(GET);
+ std::string response;
+ st = client.execute(&response);
+ ASSERT_FALSE(st.ok());
+}
+
+}
+
+int main(int argc, char* argv[]) {
+ ::testing::InitGoogleTest(&argc, argv);
+ return RUN_ALL_TESTS();
+}
diff --git a/be/test/http/http_utils_test.cpp b/be/test/http/http_utils_test.cpp
index 9a604e39..9beaf543 100644
--- a/be/test/http/http_utils_test.cpp
+++ b/be/test/http/http_utils_test.cpp
@@ -31,15 +31,22 @@ class HttpUtilsTest : public testing::Test {
HttpUtilsTest() { }
virtual ~HttpUtilsTest() {
}
+ void SetUp() override {
+ _evhttp_req = evhttp_request_new(nullptr, nullptr);
+ }
+ void TearDown() override {
+ if (_evhttp_req != nullptr) {
+ evhttp_request_free(_evhttp_req);
+ }
+ }
+private:
+ evhttp_request* _evhttp_req = nullptr;
};
TEST_F(HttpUtilsTest, parse_basic_auth) {
{
- HttpRequest req;
- std::string auth = "Basic ";
- std::string encoded_str;
- base64_encode("doris:passwd", &encoded_str);
- auth += encoded_str;
+ HttpRequest req(_evhttp_req);
+ auto auth = encode_basic_auth("doris", "passwd");
req._headers.emplace(HttpHeaders::AUTHORIZATION, auth);
std::string user;
std::string passwd;
@@ -49,7 +56,7 @@ TEST_F(HttpUtilsTest, parse_basic_auth) {
ASSERT_STREQ("passwd", passwd.data());
}
{
- HttpRequest req;
+ HttpRequest req(_evhttp_req);
std::string auth = "Basic ";
std::string encoded_str = "doris:passwd";
auth += encoded_str;
@@ -60,7 +67,7 @@ TEST_F(HttpUtilsTest, parse_basic_auth) {
ASSERT_FALSE(res);
}
{
- HttpRequest req;
+ HttpRequest req(_evhttp_req);
std::string auth = "Basic ";
std::string encoded_str;
base64_encode("dorispasswd", &encoded_str);
@@ -72,7 +79,7 @@ TEST_F(HttpUtilsTest, parse_basic_auth) {
ASSERT_FALSE(res);
}
{
- HttpRequest req;
+ HttpRequest req(_evhttp_req);
std::string auth = "Basic";
std::string encoded_str;
base64_encode("doris:passwd", &encoded_str);
diff --git a/be/test/http/metrics_action_test.cpp
b/be/test/http/metrics_action_test.cpp
index f1f2948e..4c63761b 100644
--- a/be/test/http/metrics_action_test.cpp
+++ b/be/test/http/metrics_action_test.cpp
@@ -39,6 +39,16 @@ class MetricsActionTest : public testing::Test {
MetricsActionTest() { }
virtual ~MetricsActionTest() {
}
+ void SetUp() override {
+ _evhttp_req = evhttp_request_new(nullptr, nullptr);
+ }
+ void TearDown() override {
+ if (_evhttp_req != nullptr) {
+ evhttp_request_free(_evhttp_req);
+ }
+ }
+private:
+ evhttp_request* _evhttp_req = nullptr;
};
TEST_F(MetricsActionTest, prometheus_output) {
@@ -56,7 +66,7 @@ TEST_F(MetricsActionTest, prometheus_output) {
"test_cpu_idle 50\n"
"# TYPE test_requests_total counter\n"
"test_requests_total{path=\"/sports\",type=\"put\"} 2345\n";
- HttpRequest request(nullptr);
+ HttpRequest request(_evhttp_req);
MetricsAction action(®istry);
action.handle(&request);
}
@@ -69,7 +79,7 @@ TEST_F(MetricsActionTest, prometheus_no_prefix) {
s_expect_response =
"# TYPE cpu_idle gauge\n"
"cpu_idle 50\n";
- HttpRequest request(nullptr);
+ HttpRequest request(_evhttp_req);
MetricsAction action(®istry);
action.handle(&request);
}
@@ -80,7 +90,7 @@ TEST_F(MetricsActionTest, prometheus_no_name) {
cpu_idle.set_value(50);
registry.register_metric("", &cpu_idle);
s_expect_response = "";
- HttpRequest request(nullptr);
+ HttpRequest request(_evhttp_req);
MetricsAction action(®istry);
action.handle(&request);
}
diff --git a/be/test/http/stream_load_test.cpp
b/be/test/http/stream_load_test.cpp
index e5d04313..a4086f29 100644
--- a/be/test/http/stream_load_test.cpp
+++ b/be/test/http/stream_load_test.cpp
@@ -81,6 +81,8 @@ class StreamLoadActionTest : public testing::Test {
_env._master_info = new TMasterInfo();
_env._load_stream_mgr = new LoadStreamMgr();
_env._brpc_stub_cache = new BrpcStubCache();
+
+ _evhttp_req = evhttp_request_new(nullptr, nullptr);
}
void TearDown() override {
delete _env._brpc_stub_cache;
@@ -91,16 +93,21 @@ class StreamLoadActionTest : public testing::Test {
_env._master_info = nullptr;
delete _env._thread_mgr;
_env._thread_mgr = nullptr;
+
+ if (_evhttp_req != nullptr) {
+ evhttp_request_free(_evhttp_req);
+ }
}
private:
ExecEnv _env;
+ evhttp_request* _evhttp_req = nullptr;
};
TEST_F(StreamLoadActionTest, no_auth) {
DorisMetrics::instance()->initialize("StreamLoadActionTest");
StreamLoadAction action(&_env);
- HttpRequest request;
+ HttpRequest request(_evhttp_req);
action.on_header(&request);
action.handle(&request);
@@ -114,7 +121,7 @@ TEST_F(StreamLoadActionTest, no_content_length) {
DorisMetrics::instance()->initialize("StreamLoadActionTest");
StreamLoadAction action(&__env);
- HttpRequest request;
+ HttpRequest request(_evhttp_req);
request._headers.emplace(HttpHeaders::AUTHORIZATION, "Basic cm9vdDo=");
action.on_header(&request);
action.handle(&request);
@@ -128,7 +135,7 @@ TEST_F(StreamLoadActionTest, unknown_encoding) {
DorisMetrics::instance()->initialize("StreamLoadActionTest");
StreamLoadAction action(&_env);
- HttpRequest request;
+ HttpRequest request(_evhttp_req);
request._headers.emplace(HttpHeaders::AUTHORIZATION, "Basic cm9vdDo=");
request._headers.emplace(HttpHeaders::TRANSFER_ENCODING, "chunked111");
action.on_header(&request);
@@ -144,7 +151,7 @@ TEST_F(StreamLoadActionTest, normal) {
DorisMetrics::instance()->initialize("StreamLoadActionTest");
StreamLoadAction action(&_env);
- HttpRequest request;
+ HttpRequest request(_evhttp_req);
struct evhttp_request ev_req;
ev_req.remote_host = nullptr;
@@ -164,7 +171,7 @@ TEST_F(StreamLoadActionTest, put_fail) {
DorisMetrics::instance()->initialize("StreamLoadActionTest");
StreamLoadAction action(&_env);
- HttpRequest request;
+ HttpRequest request(_evhttp_req);
struct evhttp_request ev_req;
ev_req.remote_host = nullptr;
@@ -186,7 +193,7 @@ TEST_F(StreamLoadActionTest, commit_fail) {
DorisMetrics::instance()->initialize("StreamLoadActionTest");
StreamLoadAction action(&_env);
- HttpRequest request;
+ HttpRequest request(_evhttp_req);
struct evhttp_request ev_req;
ev_req.remote_host = nullptr;
request._ev_req = &ev_req;
@@ -206,7 +213,7 @@ TEST_F(StreamLoadActionTest, begin_fail) {
DorisMetrics::instance()->initialize("StreamLoadActionTest");
StreamLoadAction action(&_env);
- HttpRequest request;
+ HttpRequest request(_evhttp_req);
struct evhttp_request ev_req;
ev_req.remote_host = nullptr;
request._ev_req = &ev_req;
@@ -227,7 +234,7 @@ TEST_F(StreamLoadActionTest, receive_failed) {
DorisMetrics::instance()->initialize("StreamLoadActionTest");
StreamLoadAction action(&_env);
- HttpRequest request;
+ HttpRequest request(_evhttp_req);
request._headers.emplace(HttpHeaders::AUTHORIZATION, "Basic cm9vdDo=");
request._headers.emplace(HttpHeaders::TRANSFER_ENCODING, "chunked");
action.on_header(&request);
@@ -243,7 +250,7 @@ TEST_F(StreamLoadActionTest, plan_fail) {
DorisMetrics::instance()->initialize("StreamLoadActionTest");
StreamLoadAction action(&_env);
- HttpRequest request;
+ HttpRequest request(_evhttp_req);
struct evhttp_request ev_req;
ev_req.remote_host = nullptr;
request._ev_req = &ev_req;
diff --git a/run-ut.sh b/run-ut.sh
index ad932c87..cc170707 100755
--- a/run-ut.sh
+++ b/run-ut.sh
@@ -182,6 +182,7 @@ ${DORIS_TEST_BINARY_DIR}/runtime/snapshot_loader_test
${DORIS_TEST_BINARY_DIR}/http/metrics_action_test
${DORIS_TEST_BINARY_DIR}/http/http_utils_test
${DORIS_TEST_BINARY_DIR}/http/stream_load_test
+${DORIS_TEST_BINARY_DIR}/http/http_client_test
# Running OLAPEngine Unittest
${DORIS_TEST_BINARY_DIR}/olap/bit_field_test
@@ -215,7 +216,7 @@ fi
cp -r ${DORIS_HOME}/be/test/agent/test_data ${DORIS_TEST_BINARY_DIR}/agent/
cd ${DORIS_TEST_BINARY_DIR}/agent
# ./agent_server_test
-./file_downloader_test
+# ./file_downloader_test
#./heartbeat_server_test
#./pusher_test
./utils_test
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
