| // 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 "cloud/injection_point_action.h" |
| |
| #include <glog/logging.h> |
| |
| #include <chrono> |
| #include <mutex> |
| |
| #include "common/status.h" |
| #include "common/sync_point.h" |
| #include "http/http_channel.h" |
| #include "http/http_request.h" |
| #include "http/http_status.h" |
| #include "olap/rowset/rowset.h" |
| #include "util/stack_util.h" |
| |
| namespace doris { |
| namespace { |
| |
| // TODO(cyx): Provide an object pool |
| // `suite_map` won't be modified after `register_suites` |
| std::map<std::string, std::function<void()>> suite_map; |
| std::once_flag register_suites_once; |
| |
| // only call once |
| void register_suites() { |
| suite_map.emplace("test_compaction", [] { |
| auto sp = SyncPoint::get_instance(); |
| sp->set_call_back("new_cumulative_point", [](auto&& args) { |
| auto output_rowset = try_any_cast<Rowset*>(args[0]); |
| auto last_cumulative_point = try_any_cast<int64_t>(args[1]); |
| auto& [ret_vault, should_ret] = *try_any_cast<std::pair<int64_t, bool>*>(args.back()); |
| ret_vault = output_rowset->start_version() == last_cumulative_point |
| ? output_rowset->end_version() + 1 |
| : last_cumulative_point; |
| should_ret = true; |
| }); |
| }); |
| suite_map.emplace("test_s3_file_writer", [] { |
| auto* sp = SyncPoint::get_instance(); |
| sp->set_call_back("UploadFileBuffer::upload_to_local_file_cache", [](auto&&) { |
| std::srand(static_cast<unsigned int>(std::time(nullptr))); |
| int random_sleep_time_second = std::rand() % 10 + 1; |
| std::this_thread::sleep_for(std::chrono::seconds(random_sleep_time_second)); |
| }); |
| sp->set_call_back("UploadFileBuffer::upload_to_local_file_cache_inject", [](auto&& args) { |
| auto& [ret_status, should_ret] = *try_any_cast<std::pair<Status, bool>*>(args.back()); |
| ret_status = |
| Status::IOError<false>("failed to write into file cache due to inject error"); |
| should_ret = true; |
| }); |
| }); |
| suite_map.emplace("test_storage_vault", [] { |
| auto* sp = SyncPoint::get_instance(); |
| sp->set_call_back("HdfsFileWriter::append_hdfs_file_delay", [](auto&&) { |
| std::srand(static_cast<unsigned int>(std::time(nullptr))); |
| int random_sleep_time_second = std::rand() % 10 + 1; |
| std::this_thread::sleep_for(std::chrono::seconds(random_sleep_time_second)); |
| }); |
| sp->set_call_back("HdfsFileWriter::append_hdfs_file_error", [](auto&& args) { |
| auto& [_, should_ret] = *try_any_cast<std::pair<Status, bool>*>(args.back()); |
| should_ret = true; |
| }); |
| sp->set_call_back("HdfsFileWriter::hdfsFlush", [](auto&& args) { |
| auto& [ret_value, should_ret] = *try_any_cast<std::pair<Status, bool>*>(args.back()); |
| ret_value = Status::InternalError("failed to flush hdfs file"); |
| should_ret = true; |
| }); |
| sp->set_call_back("HdfsFileWriter::hdfsCloseFile", [](auto&& args) { |
| auto& [ret_value, should_ret] = *try_any_cast<std::pair<Status, bool>*>(args.back()); |
| ret_value = Status::InternalError("failed to flush hdfs file"); |
| should_ret = true; |
| }); |
| sp->set_call_back("HdfsFileWriter::hdfeSync", [](auto&& args) { |
| auto& [ret_value, should_ret] = *try_any_cast<std::pair<Status, bool>*>(args.back()); |
| ret_value = Status::InternalError("failed to flush hdfs file"); |
| should_ret = true; |
| }); |
| sp->set_call_back("HdfsFileReader:read_error", [](auto&& args) { |
| auto& [ret_status, should_ret] = *try_any_cast<std::pair<Status, bool>*>(args.back()); |
| ret_status = Status::InternalError("read hdfs error"); |
| should_ret = true; |
| }); |
| }); |
| } |
| |
| void set_sleep(const std::string& point, HttpRequest* req) { |
| int duration = 0; |
| auto& duration_str = req->param("duration"); |
| if (!duration_str.empty()) { |
| try { |
| duration = std::stoi(duration_str); |
| } catch (const std::exception&) { |
| HttpChannel::send_reply(req, HttpStatus::BAD_REQUEST, |
| "invalid duration: " + duration_str); |
| return; |
| } |
| } |
| auto sp = SyncPoint::get_instance(); |
| sp->set_call_back(point, [duration](auto&& args) { |
| std::this_thread::sleep_for(std::chrono::milliseconds(duration)); |
| }); |
| HttpChannel::send_reply(req, HttpStatus::OK, "OK"); |
| } |
| |
| void set_return(const std::string& point, HttpRequest* req) { |
| auto sp = SyncPoint::get_instance(); |
| sp->set_call_back(point, [](auto&& args) { |
| try { |
| auto pred = try_any_cast<bool*>(args.back()); |
| *pred = true; |
| } catch (const std::bad_any_cast&) { |
| LOG_EVERY_N(ERROR, 10) << "failed to process `return` callback\n" << get_stack_trace(); |
| } |
| }); |
| HttpChannel::send_reply(req, HttpStatus::OK, "OK"); |
| } |
| |
| void set_return_ok(const std::string& point, HttpRequest* req) { |
| auto sp = SyncPoint::get_instance(); |
| sp->set_call_back(point, [](auto&& args) { |
| try { |
| auto* pair = try_any_cast_ret<Status>(args); |
| pair->first = Status::OK(); |
| pair->second = true; |
| } catch (const std::bad_any_cast&) { |
| LOG_EVERY_N(ERROR, 10) << "failed to process `return_ok` callback\n" |
| << get_stack_trace(); |
| } |
| }); |
| HttpChannel::send_reply(req, HttpStatus::OK, "OK"); |
| } |
| |
| void set_return_error(const std::string& point, HttpRequest* req) { |
| const std::string CODE_PARAM = "code"; |
| int code = ErrorCode::INTERNAL_ERROR; |
| auto& code_str = req->param(CODE_PARAM); |
| if (!code_str.empty()) { |
| try { |
| code = std::stoi(code_str); |
| } catch (const std::exception& e) { |
| HttpChannel::send_reply(req, HttpStatus::BAD_REQUEST, |
| fmt::format("convert topn failed, {}", e.what())); |
| return; |
| } |
| } |
| |
| auto sp = SyncPoint::get_instance(); |
| sp->set_call_back(point, [code](auto&& args) { |
| try { |
| auto* pair = try_any_cast_ret<Status>(args); |
| pair->first = Status::Error<false>(code, "injected error"); |
| pair->second = true; |
| } catch (const std::bad_any_cast&) { |
| LOG_EVERY_N(ERROR, 10) << "failed to process `return_error` callback\n" |
| << get_stack_trace(); |
| } |
| }); |
| HttpChannel::send_reply(req, HttpStatus::OK, "OK"); |
| } |
| |
| void handle_set(HttpRequest* req) { |
| auto& point = req->param("name"); |
| if (point.empty()) { |
| HttpChannel::send_reply(req, HttpStatus::BAD_REQUEST, "empty point name"); |
| return; |
| } |
| auto& behavior = req->param("behavior"); |
| if (behavior.empty()) { |
| HttpChannel::send_reply(req, HttpStatus::BAD_REQUEST, "empty behavior"); |
| return; |
| } |
| if (behavior == "sleep") { |
| set_sleep(point, req); |
| return; |
| } else if (behavior == "return") { |
| set_return(point, req); |
| return; |
| } else if (behavior == "return_ok") { |
| set_return_ok(point, req); |
| return; |
| } else if (behavior == "return_error") { |
| set_return_error(point, req); |
| return; |
| } |
| HttpChannel::send_reply(req, HttpStatus::BAD_REQUEST, "unknown behavior: " + behavior); |
| } |
| |
| void handle_clear(HttpRequest* req) { |
| auto& point = req->param("name"); |
| if (point.empty()) { |
| HttpChannel::send_reply(req, HttpStatus::BAD_REQUEST, "empty point name"); |
| return; |
| } |
| auto sp = SyncPoint::get_instance(); |
| if (point == "all") { |
| sp->clear_all_call_backs(); |
| HttpChannel::send_reply(req, HttpStatus::OK, "OK"); |
| return; |
| } |
| sp->clear_call_back(point); |
| HttpChannel::send_reply(req, HttpStatus::OK, "OK"); |
| } |
| |
| void handle_suite(HttpRequest* req) { |
| auto& suite = req->param("name"); |
| if (suite.empty()) { |
| HttpChannel::send_reply(req, HttpStatus::BAD_REQUEST, "empty suite name"); |
| return; |
| } |
| |
| std::call_once(register_suites_once, register_suites); |
| if (auto it = suite_map.find(suite); it != suite_map.end()) { |
| it->second(); // set injection callbacks |
| HttpChannel::send_reply(req, HttpStatus::OK, "OK"); |
| return; |
| } |
| HttpChannel::send_reply(req, HttpStatus::INTERNAL_SERVER_ERROR, "unknown suite: " + suite); |
| } |
| |
| } // namespace |
| |
| InjectionPointAction::InjectionPointAction() { |
| SyncPoint::get_instance()->enable_processing(); |
| } |
| |
| void InjectionPointAction::handle(HttpRequest* req) { |
| LOG(INFO) << req->debug_string(); |
| auto& op = req->param("op"); |
| if (op == "set") { |
| handle_set(req); |
| return; |
| } else if (op == "clear") { |
| handle_clear(req); |
| return; |
| } else if (op == "apply_suite") { |
| handle_suite(req); |
| return; |
| } |
| HttpChannel::send_reply(req, HttpStatus::BAD_REQUEST, "unknown op: " + op); |
| } |
| |
| } // namespace doris |