/root/doris/be/src/http/http_client.cpp
Line | Count | Source (jump to first uncovered line) |
1 | | // Licensed to the Apache Software Foundation (ASF) under one |
2 | | // or more contributor license agreements. See the NOTICE file |
3 | | // distributed with this work for additional information |
4 | | // regarding copyright ownership. The ASF licenses this file |
5 | | // to you under the Apache License, Version 2.0 (the |
6 | | // "License"); you may not use this file except in compliance |
7 | | // with the License. You may obtain a copy of the License at |
8 | | // |
9 | | // http://www.apache.org/licenses/LICENSE-2.0 |
10 | | // |
11 | | // Unless required by applicable law or agreed to in writing, |
12 | | // software distributed under the License is distributed on an |
13 | | // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
14 | | // KIND, either express or implied. See the License for the |
15 | | // specific language governing permissions and limitations |
16 | | // under the License. |
17 | | |
18 | | #include "http/http_client.h" |
19 | | |
20 | | #include <glog/logging.h> |
21 | | #include <unistd.h> |
22 | | |
23 | | #include <memory> |
24 | | #include <ostream> |
25 | | |
26 | | #include "common/config.h" |
27 | | #include "common/status.h" |
28 | | #include "http/http_headers.h" |
29 | | #include "http/http_status.h" |
30 | | #include "runtime/exec_env.h" |
31 | | #include "util/security.h" |
32 | | #include "util/stack_util.h" |
33 | | |
34 | | namespace doris { |
35 | | |
36 | | class MultiFileSplitter { |
37 | | public: |
38 | | MultiFileSplitter(std::string local_dir, std::unordered_set<std::string> expected_files) |
39 | 1 | : _local_dir_path(std::move(local_dir)), _expected_files(std::move(expected_files)) {} |
40 | 1 | ~MultiFileSplitter() { |
41 | 1 | if (_fd >= 0) { |
42 | 0 | close(_fd); |
43 | 0 | } |
44 | | |
45 | 1 | if (!_status.ok() && !downloaded_files.empty()) { |
46 | 0 | LOG(WARNING) << "download files to " << _local_dir_path << " failed, try remove the " |
47 | 0 | << downloaded_files.size() << " downloaded files"; |
48 | 0 | for (const auto& file : downloaded_files) { |
49 | 0 | remove(file.c_str()); |
50 | 0 | } |
51 | 0 | } |
52 | 1 | } |
53 | | |
54 | 661 | bool append(const char* data, size_t length) { |
55 | | // Already failed. |
56 | 661 | if (!_status.ok()) { |
57 | 0 | return false; |
58 | 0 | } |
59 | | |
60 | 661 | std::string buf; |
61 | 661 | if (!_buffer.empty()) { |
62 | 0 | buf.swap(_buffer); |
63 | 0 | buf.append(data, length); |
64 | 0 | data = buf.data(); |
65 | 0 | length = buf.size(); |
66 | 0 | } |
67 | 661 | return append_inner(data, length); |
68 | 661 | } |
69 | | |
70 | 1 | Status finish() { |
71 | 1 | if (_status.ok()) { |
72 | 1 | _status = finish_inner(); |
73 | 1 | } |
74 | | |
75 | 1 | return _status; |
76 | 1 | } |
77 | | |
78 | | private: |
79 | 661 | bool append_inner(const char* data, size_t length) { |
80 | 1.39k | while (length > 0) { |
81 | 729 | int consumed = 0; |
82 | 729 | if (_is_reading_header) { |
83 | 35 | consumed = parse_header(data, length); |
84 | 694 | } else { |
85 | 694 | consumed = append_file(data, length); |
86 | 694 | } |
87 | | |
88 | 729 | if (consumed < 0) { |
89 | 0 | return false; |
90 | 0 | } |
91 | | |
92 | 729 | DCHECK(consumed <= length); |
93 | 729 | data += consumed; |
94 | 729 | length -= consumed; |
95 | 729 | } |
96 | 661 | return true; |
97 | 661 | } |
98 | | |
99 | 35 | int parse_header(const char* data, size_t length) { |
100 | 35 | DCHECK(_fd < 0); |
101 | | |
102 | 35 | std::string_view buf(data, length); |
103 | 35 | size_t pos = buf.find("\r\n\r\n"); |
104 | 35 | if (pos == std::string::npos) { |
105 | 0 | _buffer.append(data, length); |
106 | 0 | return static_cast<int>(length); |
107 | 0 | } |
108 | | |
109 | | // header already read. |
110 | 35 | _is_reading_header = false; |
111 | | |
112 | 35 | bool has_file_name = false; |
113 | 35 | bool has_file_size = false; |
114 | 35 | std::string_view header = buf.substr(0, pos); |
115 | 35 | std::vector<std::string> headers = |
116 | 35 | strings::Split(header, "\r\n", strings::SkipWhitespace()); |
117 | 70 | for (auto& s : headers) { |
118 | 70 | size_t header_pos = s.find(':'); |
119 | 70 | if (header_pos == std::string::npos) { |
120 | 0 | continue; |
121 | 0 | } |
122 | 70 | std::string_view header_view(s); |
123 | 70 | std::string_view key = header_view.substr(0, header_pos); |
124 | 70 | std::string_view value = header_view.substr(header_pos + 1); |
125 | 70 | if (value.starts_with(' ')) { |
126 | 70 | value.remove_prefix(std::min(value.find_first_not_of(' '), value.size())); |
127 | 70 | } |
128 | 70 | if (key == "File-Name") { |
129 | 35 | _file_name = value; |
130 | 35 | has_file_name = true; |
131 | 35 | } else if (key == "Content-Length") { |
132 | 35 | auto res = std::from_chars(value.data(), value.data() + value.size(), _file_size); |
133 | 35 | if (res.ec != std::errc()) { |
134 | 0 | std::string error_msg = fmt::format("invalid content length: {}", value); |
135 | 0 | LOG(WARNING) << "download files to " << _local_dir_path |
136 | 0 | << "failed, err=" << error_msg; |
137 | 0 | _status = Status::HttpError(std::move(error_msg)); |
138 | 0 | return -1; |
139 | 0 | } |
140 | 35 | has_file_size = true; |
141 | 35 | } |
142 | 70 | } |
143 | | |
144 | 35 | if (!has_file_name || !has_file_size) { |
145 | 0 | std::string error_msg = |
146 | 0 | fmt::format("invalid multi part header, has file name: {}, has file size: {}", |
147 | 0 | has_file_name, has_file_size); |
148 | 0 | LOG(WARNING) << "download files to " << _local_dir_path << "failed, err=" << error_msg; |
149 | 0 | _status = Status::HttpError(std::move(error_msg)); |
150 | 0 | return -1; |
151 | 0 | } |
152 | | |
153 | 35 | if (!_expected_files.contains(_file_name)) { |
154 | 0 | std::string error_msg = fmt::format("unexpected file: {}", _file_name); |
155 | 0 | LOG(WARNING) << "download files to " << _local_dir_path << "failed, err=" << error_msg; |
156 | 0 | _status = Status::HttpError(std::move(error_msg)); |
157 | 0 | return -1; |
158 | 0 | } |
159 | | |
160 | 35 | VLOG_DEBUG << "receive file " << _file_name << ", size " << _file_size; |
161 | | |
162 | 35 | _written_size = 0; |
163 | 35 | _local_file_path = fmt::format("{}/{}", _local_dir_path, _file_name); |
164 | 35 | _fd = open(_local_file_path.c_str(), O_WRONLY | O_CREAT | O_TRUNC, 0644); |
165 | 35 | if (_fd < 0) { |
166 | 0 | std::string error_msg = "fail to open file to write: " + _local_file_path; |
167 | 0 | LOG(WARNING) << "download files to " << _local_dir_path << "failed, err=" << error_msg; |
168 | 0 | _status = Status::IOError(std::move(error_msg)); |
169 | 0 | return -1; |
170 | 0 | } |
171 | 35 | downloaded_files.push_back(_local_file_path); |
172 | | |
173 | 35 | return static_cast<int>(pos + 4); |
174 | 35 | } |
175 | | |
176 | 694 | int append_file(const char* data, size_t length) { |
177 | 694 | DCHECK(_fd >= 0); |
178 | 694 | DCHECK(_file_size >= _written_size); |
179 | | |
180 | 694 | size_t write_size = std::min(length, _file_size - _written_size); |
181 | 694 | if (write_size > 0 && write(_fd, data, write_size) < 0) { |
182 | 0 | auto msg = fmt::format("write file failed, file={}, error={}", _local_file_path, |
183 | 0 | strerror(errno)); |
184 | 0 | LOG(WARNING) << "download files to " << _local_dir_path << "failed, err=" << msg; |
185 | 0 | _status = Status::HttpError(std::move(msg)); |
186 | 0 | return -1; |
187 | 0 | } |
188 | | |
189 | 694 | _written_size += write_size; |
190 | 694 | if (_written_size == _file_size) { |
191 | | // This file has been downloaded, switch to the next one. |
192 | 34 | switch_to_next_file(); |
193 | 34 | } |
194 | | |
195 | 694 | return write_size; |
196 | 694 | } |
197 | | |
198 | 1 | Status finish_inner() { |
199 | 1 | if (!_is_reading_header && _written_size == _file_size) { |
200 | 1 | switch_to_next_file(); |
201 | 1 | } |
202 | | |
203 | 1 | if (_fd >= 0) { |
204 | | // This file is not completely downloaded. |
205 | 0 | close(_fd); |
206 | 0 | _fd = -1; |
207 | 0 | auto error_msg = fmt::format("file {} is not completely downloaded", _local_file_path); |
208 | 0 | LOG(WARNING) << "download files to " << _local_dir_path << "failed, err=" << error_msg; |
209 | 0 | return Status::HttpError(std::move(error_msg)); |
210 | 0 | } |
211 | | |
212 | 1 | if (!_expected_files.empty()) { |
213 | 0 | auto error_msg = fmt::format("not all files are downloaded, {} missing files", |
214 | 0 | _expected_files.size()); |
215 | 0 | LOG(WARNING) << "download files to " << _local_dir_path << "failed, err=" << error_msg; |
216 | 0 | return Status::HttpError(std::move(error_msg)); |
217 | 0 | } |
218 | | |
219 | 1 | downloaded_files.clear(); |
220 | 1 | return Status::OK(); |
221 | 1 | } |
222 | | |
223 | 35 | void switch_to_next_file() { |
224 | 35 | DCHECK(_fd >= 0); |
225 | 35 | DCHECK(_written_size == _file_size); |
226 | | |
227 | 35 | close(_fd); |
228 | 35 | _fd = -1; |
229 | 35 | _expected_files.erase(_file_name); |
230 | 35 | _is_reading_header = true; |
231 | 35 | } |
232 | | |
233 | | const std::string _local_dir_path; |
234 | | std::string _buffer; |
235 | | std::unordered_set<std::string> _expected_files; |
236 | | Status _status; |
237 | | |
238 | | bool _is_reading_header = true; |
239 | | int _fd = -1; |
240 | | std::string _local_file_path; |
241 | | std::string _file_name; |
242 | | size_t _file_size = 0; |
243 | | size_t _written_size = 0; |
244 | | std::vector<std::string> downloaded_files; |
245 | | }; |
246 | | |
247 | 0 | static const char* header_error_msg(CURLHcode code) { |
248 | 0 | switch (code) { |
249 | 0 | case CURLHE_OK: |
250 | 0 | return "OK"; |
251 | 0 | case CURLHE_BADINDEX: |
252 | 0 | return "header exists but not with this index "; |
253 | 0 | case CURLHE_MISSING: |
254 | 0 | return "no such header exists"; |
255 | 0 | case CURLHE_NOHEADERS: |
256 | 0 | return "no headers at all exist (yet)"; |
257 | 0 | case CURLHE_NOREQUEST: |
258 | 0 | return "no request with this number was used"; |
259 | 0 | case CURLHE_OUT_OF_MEMORY: |
260 | 0 | return "out of memory while processing"; |
261 | 0 | case CURLHE_BAD_ARGUMENT: |
262 | 0 | return "a function argument was not okay"; |
263 | 0 | case CURLHE_NOT_BUILT_IN: |
264 | 0 | return "curl_easy_header() was disabled in the build"; |
265 | 0 | default: |
266 | 0 | return "unknown"; |
267 | 0 | } |
268 | 0 | } |
269 | | |
270 | 24 | HttpClient::HttpClient() = default; |
271 | | |
272 | 24 | HttpClient::~HttpClient() { |
273 | 24 | if (_curl != nullptr) { |
274 | 23 | curl_easy_cleanup(_curl); |
275 | 23 | _curl = nullptr; |
276 | 23 | } |
277 | 24 | if (_header_list != nullptr) { |
278 | 0 | curl_slist_free_all(_header_list); |
279 | 0 | _header_list = nullptr; |
280 | 0 | } |
281 | 24 | } |
282 | | |
283 | 23 | Status HttpClient::init(const std::string& url, bool set_fail_on_error) { |
284 | 23 | if (_curl == nullptr) { |
285 | 22 | _curl = curl_easy_init(); |
286 | 22 | if (_curl == nullptr) { |
287 | 0 | return Status::InternalError("fail to initialize curl"); |
288 | 0 | } |
289 | 22 | } else { |
290 | 1 | curl_easy_reset(_curl); |
291 | 1 | } |
292 | | |
293 | 23 | if (_header_list != nullptr) { |
294 | 0 | curl_slist_free_all(_header_list); |
295 | 0 | _header_list = nullptr; |
296 | 0 | } |
297 | | // set error_buf |
298 | 23 | _error_buf[0] = 0; |
299 | 23 | auto code = curl_easy_setopt(_curl, CURLOPT_ERRORBUFFER, _error_buf); |
300 | 23 | if (code != CURLE_OK) { |
301 | 0 | LOG(WARNING) << "fail to set CURLOPT_ERRORBUFFER, msg=" << _to_errmsg(code); |
302 | 0 | return Status::InternalError("fail to set error buffer"); |
303 | 0 | } |
304 | | // forbid signals |
305 | 23 | code = curl_easy_setopt(_curl, CURLOPT_NOSIGNAL, 1L); |
306 | 23 | if (code != CURLE_OK) { |
307 | 0 | LOG(WARNING) << "fail to set CURLOPT_NOSIGNAL, msg=" << _to_errmsg(code); |
308 | 0 | return Status::InternalError("fail to set CURLOPT_NOSIGNAL"); |
309 | 0 | } |
310 | | // set fail on error |
311 | | // When this option is set to `1L` (enabled), libcurl will return an error directly |
312 | | // when encountering HTTP error codes (>= 400), without reading the body of the error response. |
313 | 23 | if (set_fail_on_error) { |
314 | 21 | code = curl_easy_setopt(_curl, CURLOPT_FAILONERROR, 1L); |
315 | 21 | if (code != CURLE_OK) { |
316 | 0 | LOG(WARNING) << "fail to set CURLOPT_FAILONERROR, msg=" << _to_errmsg(code); |
317 | 0 | return Status::InternalError("fail to set CURLOPT_FAILONERROR"); |
318 | 0 | } |
319 | 21 | } |
320 | | // set redirect |
321 | 23 | code = curl_easy_setopt(_curl, CURLOPT_FOLLOWLOCATION, 1L); |
322 | 23 | if (code != CURLE_OK) { |
323 | 0 | LOG(WARNING) << "fail to set CURLOPT_FOLLOWLOCATION, msg=" << _to_errmsg(code); |
324 | 0 | return Status::InternalError("fail to set CURLOPT_FOLLOWLOCATION"); |
325 | 0 | } |
326 | 23 | code = curl_easy_setopt(_curl, CURLOPT_MAXREDIRS, 20); |
327 | 23 | if (code != CURLE_OK) { |
328 | 0 | LOG(WARNING) << "fail to set CURLOPT_MAXREDIRS, msg=" << _to_errmsg(code); |
329 | 0 | return Status::InternalError("fail to set CURLOPT_MAXREDIRS"); |
330 | 0 | } |
331 | | |
332 | 673 | curl_write_callback callback = [](char* buffer, size_t size, size_t nmemb, void* param) { |
333 | 673 | auto* client = (HttpClient*)param; |
334 | 673 | return client->on_response_data(buffer, size * nmemb); |
335 | 673 | }; |
336 | | |
337 | | // set callback function |
338 | 23 | code = curl_easy_setopt(_curl, CURLOPT_WRITEFUNCTION, callback); |
339 | 23 | if (code != CURLE_OK) { |
340 | 0 | LOG(WARNING) << "fail to set CURLOPT_WRITEFUNCTION, msg=" << _to_errmsg(code); |
341 | 0 | return Status::InternalError("fail to set CURLOPT_WRITEFUNCTION"); |
342 | 0 | } |
343 | 23 | code = curl_easy_setopt(_curl, CURLOPT_WRITEDATA, (void*)this); |
344 | 23 | if (code != CURLE_OK) { |
345 | 0 | LOG(WARNING) << "fail to set CURLOPT_WRITEDATA, msg=" << _to_errmsg(code); |
346 | 0 | return Status::InternalError("fail to set CURLOPT_WRITEDATA"); |
347 | 0 | } |
348 | | |
349 | 23 | std::string escaped_url; |
350 | 23 | RETURN_IF_ERROR(_escape_url(url, &escaped_url)); |
351 | | // set url |
352 | 23 | code = curl_easy_setopt(_curl, CURLOPT_URL, escaped_url.c_str()); |
353 | 23 | if (code != CURLE_OK) { |
354 | 0 | LOG(WARNING) << "failed to set CURLOPT_URL, errmsg=" << _to_errmsg(code); |
355 | 0 | return Status::InternalError("fail to set CURLOPT_URL"); |
356 | 0 | } |
357 | | |
358 | 23 | return Status::OK(); |
359 | 23 | } |
360 | | |
361 | 22 | void HttpClient::set_method(HttpMethod method) { |
362 | 22 | _method = method; |
363 | 22 | switch (method) { |
364 | 4 | case GET: |
365 | 4 | curl_easy_setopt(_curl, CURLOPT_HTTPGET, 1L); |
366 | 4 | return; |
367 | 0 | case PUT: |
368 | 0 | curl_easy_setopt(_curl, CURLOPT_UPLOAD, 1L); |
369 | 0 | return; |
370 | 13 | case POST: |
371 | 13 | curl_easy_setopt(_curl, CURLOPT_POST, 1L); |
372 | 13 | return; |
373 | 0 | case DELETE: |
374 | 0 | curl_easy_setopt(_curl, CURLOPT_CUSTOMREQUEST, "DELETE"); |
375 | 0 | return; |
376 | 5 | case HEAD: |
377 | 5 | curl_easy_setopt(_curl, CURLOPT_NOBODY, 1L); |
378 | 5 | return; |
379 | 0 | case OPTIONS: |
380 | 0 | curl_easy_setopt(_curl, CURLOPT_CUSTOMREQUEST, "OPTIONS"); |
381 | 0 | return; |
382 | 0 | default: |
383 | 0 | return; |
384 | 22 | } |
385 | 22 | } |
386 | | |
387 | 2 | void HttpClient::set_speed_limit() { |
388 | 2 | curl_easy_setopt(_curl, CURLOPT_LOW_SPEED_LIMIT, config::download_low_speed_limit_kbps * 1024); |
389 | 2 | curl_easy_setopt(_curl, CURLOPT_LOW_SPEED_TIME, config::download_low_speed_time); |
390 | 2 | curl_easy_setopt(_curl, CURLOPT_MAX_RECV_SPEED_LARGE, config::max_download_speed_kbps * 1024); |
391 | 2 | } |
392 | | |
393 | 673 | size_t HttpClient::on_response_data(const void* data, size_t length) { |
394 | 673 | if (*_callback != nullptr) { |
395 | 673 | bool is_continue = (*_callback)(data, length); |
396 | 673 | if (!is_continue) { |
397 | 0 | return -1; |
398 | 0 | } |
399 | 673 | } |
400 | 673 | return length; |
401 | 673 | } |
402 | | |
403 | 10 | Status HttpClient::execute_post_request(const std::string& payload, std::string* response) { |
404 | 10 | set_method(POST); |
405 | 10 | set_payload(payload); |
406 | 10 | return execute(response); |
407 | 10 | } |
408 | | |
409 | 0 | Status HttpClient::execute_delete_request(const std::string& payload, std::string* response) { |
410 | 0 | set_method(DELETE); |
411 | 0 | set_payload(payload); |
412 | 0 | return execute(response); |
413 | 0 | } |
414 | | |
415 | 23 | Status HttpClient::execute(const std::function<bool(const void* data, size_t length)>& callback) { |
416 | 23 | if (VLOG_DEBUG_IS_ON) { |
417 | 0 | VLOG_DEBUG << "execute http " << to_method_desc(_method) << " request, url " << _get_url(); |
418 | 0 | } |
419 | 23 | _callback = &callback; |
420 | 23 | auto code = curl_easy_perform(_curl); |
421 | 23 | if (code != CURLE_OK) { |
422 | 5 | std::string url = mask_token(_get_url()); |
423 | 5 | LOG(WARNING) << "fail to execute HTTP client, errmsg=" << _to_errmsg(code) |
424 | 5 | << ", trace=" << get_stack_trace() << ", url=" << url; |
425 | 5 | std::string errmsg = fmt::format("{}, url={}", _to_errmsg(code), url); |
426 | 5 | return Status::HttpError(std::move(errmsg)); |
427 | 5 | } |
428 | 18 | if (VLOG_DEBUG_IS_ON) { |
429 | 0 | VLOG_DEBUG << "execute http " << to_method_desc(_method) << " request, url " << _get_url() |
430 | 0 | << " done"; |
431 | 0 | } |
432 | 18 | return Status::OK(); |
433 | 23 | } |
434 | | |
435 | 3 | Status HttpClient::get_content_md5(std::string* md5) const { |
436 | 3 | struct curl_header* header_ptr; |
437 | 3 | auto code = curl_easy_header(_curl, HttpHeaders::CONTENT_MD5, 0, CURLH_HEADER, 0, &header_ptr); |
438 | 3 | if (code == CURLHE_MISSING || code == CURLHE_NOHEADERS) { |
439 | | // no such headers exists |
440 | 1 | md5->clear(); |
441 | 1 | return Status::OK(); |
442 | 2 | } else if (code != CURLHE_OK) { |
443 | 0 | auto msg = fmt::format("failed to get http header {}: {} ({})", HttpHeaders::CONTENT_MD5, |
444 | 0 | header_error_msg(code), code); |
445 | 0 | LOG(WARNING) << msg << ", trace=" << get_stack_trace(); |
446 | 0 | return Status::HttpError(std::move(msg)); |
447 | 0 | } |
448 | | |
449 | 2 | *md5 = header_ptr->value; |
450 | 2 | return Status::OK(); |
451 | 3 | } |
452 | | |
453 | 1 | Status HttpClient::download(const std::string& local_path) { |
454 | 1 | set_method(GET); |
455 | 1 | set_speed_limit(); |
456 | | |
457 | 1 | auto fp_closer = [](FILE* fp) { fclose(fp); }; |
458 | 1 | std::unique_ptr<FILE, decltype(fp_closer)> fp(fopen(local_path.c_str(), "w"), fp_closer); |
459 | 1 | if (fp == nullptr) { |
460 | 0 | LOG(WARNING) << "open file failed, file=" << local_path; |
461 | 0 | return Status::InternalError("open file failed"); |
462 | 0 | } |
463 | 1 | Status status; |
464 | 1 | auto callback = [&status, &fp, &local_path](const void* data, size_t length) { |
465 | 1 | auto res = fwrite(data, length, 1, fp.get()); |
466 | 1 | if (res != 1) { |
467 | 0 | LOG(WARNING) << "fail to write data to file, file=" << local_path |
468 | 0 | << ", error=" << ferror(fp.get()); |
469 | 0 | status = Status::InternalError("fail to write data when download"); |
470 | 0 | return false; |
471 | 0 | } |
472 | 1 | return true; |
473 | 1 | }; |
474 | | |
475 | 1 | if (auto s = execute(callback); !s.ok()) { |
476 | 0 | status = s; |
477 | 0 | } |
478 | 1 | if (!status.ok()) { |
479 | 0 | remove(local_path.c_str()); |
480 | 0 | } |
481 | 1 | return status; |
482 | 1 | } |
483 | | |
484 | | Status HttpClient::download_multi_files(const std::string& local_dir, |
485 | 1 | const std::unordered_set<std::string>& expected_files) { |
486 | 1 | set_speed_limit(); |
487 | | |
488 | 1 | MultiFileSplitter splitter(local_dir, expected_files); |
489 | 661 | auto callback = [&](const void* data, size_t length) { |
490 | 661 | return splitter.append(reinterpret_cast<const char*>(data), length); |
491 | 661 | }; |
492 | 1 | if (auto s = execute(callback); !s.ok()) { |
493 | 0 | return s; |
494 | 0 | } |
495 | 1 | return splitter.finish(); |
496 | 1 | } |
497 | | |
498 | 17 | Status HttpClient::execute(std::string* response) { |
499 | 17 | auto callback = [response](const void* data, size_t length) { |
500 | 11 | response->append((char*)data, length); |
501 | 11 | return true; |
502 | 11 | }; |
503 | 17 | return execute(callback); |
504 | 17 | } |
505 | | |
506 | 10 | const char* HttpClient::_to_errmsg(CURLcode code) const { |
507 | 10 | if (_error_buf[0] == 0) { |
508 | 0 | return curl_easy_strerror(code); |
509 | 0 | } |
510 | 10 | return _error_buf; |
511 | 10 | } |
512 | | |
513 | 5 | const char* HttpClient::_get_url() const { |
514 | 5 | const char* url = nullptr; |
515 | 5 | curl_easy_getinfo(_curl, CURLINFO_EFFECTIVE_URL, &url); |
516 | 5 | if (!url) { |
517 | 0 | url = "<unknown>"; |
518 | 0 | } |
519 | 5 | return url; |
520 | 5 | } |
521 | | |
522 | | // execute remote call action with retry |
523 | | Status HttpClient::execute(int retry_times, int sleep_time, |
524 | 0 | const std::function<Status(HttpClient*)>& callback) { |
525 | 0 | Status status; |
526 | 0 | for (int i = 0; i < retry_times; ++i) { |
527 | 0 | status = callback(this); |
528 | 0 | if (status.ok()) { |
529 | 0 | auto http_status = get_http_status(); |
530 | 0 | if (http_status == 200) { |
531 | 0 | return status; |
532 | 0 | } else { |
533 | 0 | std::string url = mask_token(_get_url()); |
534 | 0 | auto error_msg = fmt::format("http status code is not 200, code={}, url={}", |
535 | 0 | http_status, url); |
536 | 0 | LOG(WARNING) << error_msg; |
537 | 0 | return Status::HttpError(error_msg); |
538 | 0 | } |
539 | 0 | } |
540 | 0 | sleep(sleep_time); |
541 | 0 | } |
542 | 0 | return status; |
543 | 0 | } |
544 | | |
545 | | Status HttpClient::execute_with_retry(int retry_times, int sleep_time, |
546 | 4 | const std::function<Status(HttpClient*)>& callback) { |
547 | 4 | Status status; |
548 | 7 | for (int i = 0; i < retry_times; ++i) { |
549 | 6 | HttpClient client; |
550 | 6 | status = callback(&client); |
551 | 6 | if (status.ok()) { |
552 | 3 | auto http_status = client.get_http_status(); |
553 | 3 | if (http_status == 200) { |
554 | 3 | return status; |
555 | 3 | } else { |
556 | 0 | std::string url = mask_token(client._get_url()); |
557 | 0 | auto error_msg = fmt::format("http status code is not 200, code={}, url={}", |
558 | 0 | http_status, url); |
559 | 0 | LOG(WARNING) << error_msg; |
560 | 0 | return Status::HttpError(error_msg); |
561 | 0 | } |
562 | 3 | } |
563 | 3 | sleep(sleep_time); |
564 | 3 | } |
565 | 1 | return status; |
566 | 4 | } |
567 | | |
568 | | // http://example.com/page?param1=value1¶m2=value+with+spaces#section |
569 | 30 | Status HttpClient::_escape_url(const std::string& url, std::string* escaped_url) { |
570 | 30 | size_t query_pos = url.find('?'); |
571 | 30 | if (query_pos == std::string::npos) { |
572 | 16 | *escaped_url = url; |
573 | 16 | return Status::OK(); |
574 | 16 | } |
575 | 14 | size_t fragment_pos = url.find('#'); |
576 | 14 | std::string query; |
577 | 14 | std::string fragment; |
578 | | |
579 | 14 | if (fragment_pos == std::string::npos) { |
580 | 13 | query = url.substr(query_pos + 1, url.length() - query_pos - 1); |
581 | 13 | } else { |
582 | 1 | query = url.substr(query_pos + 1, fragment_pos - query_pos - 1); |
583 | 1 | fragment = url.substr(fragment_pos, url.length() - fragment_pos); |
584 | 1 | } |
585 | | |
586 | 14 | std::string encoded_query; |
587 | 14 | size_t ampersand_pos = query.find('&'); |
588 | 14 | size_t equal_pos; |
589 | | |
590 | 14 | if (ampersand_pos == std::string::npos) { |
591 | 7 | ampersand_pos = query.length(); |
592 | 7 | } |
593 | | |
594 | 25 | while (true) { |
595 | 25 | equal_pos = query.find('='); |
596 | 25 | if (equal_pos != std::string::npos) { |
597 | 22 | std::string key = query.substr(0, equal_pos); |
598 | 22 | std::string value = query.substr(equal_pos + 1, ampersand_pos - equal_pos - 1); |
599 | | |
600 | 22 | auto encoded_value = std::unique_ptr<char, decltype(&curl_free)>( |
601 | 22 | curl_easy_escape(_curl, value.c_str(), value.length()), &curl_free); |
602 | 22 | if (encoded_value) { |
603 | 22 | encoded_query += key + "=" + std::string(encoded_value.get()); |
604 | 22 | } else { |
605 | 0 | return Status::InternalError("escape url failed, url={}", url); |
606 | 0 | } |
607 | 22 | } else { |
608 | 3 | encoded_query += query.substr(0, ampersand_pos); |
609 | 3 | } |
610 | | |
611 | 25 | if (ampersand_pos == query.length() || ampersand_pos == std::string::npos) { |
612 | 14 | break; |
613 | 14 | } |
614 | | |
615 | 11 | encoded_query += "&"; |
616 | 11 | query = query.substr(ampersand_pos + 1); |
617 | 11 | ampersand_pos = query.find('&'); |
618 | 11 | } |
619 | 14 | *escaped_url = url.substr(0, query_pos + 1) + encoded_query + fragment; |
620 | 14 | return Status::OK(); |
621 | 14 | } |
622 | | |
623 | | } // namespace doris |