| // 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 <brpc/controller.h> |
| #include <brpc/server.h> |
| #include <foundationdb/fdb_c_options.g.h> |
| #include <foundationdb/fdb_c_types.h> |
| #include <gen_cpp/cloud.pb.h> |
| #include <gtest/gtest.h> |
| |
| #include <chrono> |
| #include <cstdint> |
| #include <cstdlib> |
| #include <iostream> |
| #include <memory> |
| |
| #include "common/bvars.h" |
| #include "common/config.h" |
| #include "common/logging.h" |
| #include "cpp/sync_point.h" |
| #include "meta-service/meta_service.h" |
| #include "meta-service/txn_kv.h" |
| |
| using namespace doris; |
| |
| static std::unique_ptr<cloud::MetaServiceProxy> create_meta_service() { |
| std::shared_ptr<cloud::TxnKv> txn_kv = std::make_unique<cloud::FdbTxnKv>(); |
| [&]() { ASSERT_EQ(txn_kv->init(), 0); }(); |
| |
| auto rate_limiter = std::make_shared<cloud::RateLimiter>(); |
| auto rc_mgr = std::make_shared<cloud::ResourceManager>(txn_kv); |
| [&]() { ASSERT_EQ(rc_mgr->init(), 0); }(); |
| |
| auto meta_service_impl = std::make_unique<cloud::MetaServiceImpl>(txn_kv, rc_mgr, rate_limiter); |
| return std::make_unique<cloud::MetaServiceProxy>(std::move(meta_service_impl)); |
| } |
| |
| static std::unique_ptr<cloud::MetaServiceProxy> meta_service; |
| |
| int main(int argc, char** argv) { |
| const std::string conf_file = "doris_cloud.conf"; |
| if (!cloud::config::init(conf_file.c_str(), true)) { |
| std::cerr << "failed to init config file, conf=" << conf_file << std::endl; |
| return -1; |
| } |
| |
| if (!cloud::init_glog("fdb_injection_test")) { |
| std::cerr << "failed to init glog" << std::endl; |
| return -1; |
| } |
| |
| ::testing::InitGoogleTest(&argc, argv); |
| |
| cloud::config::enable_txn_store_retry = true; |
| cloud::config::txn_store_retry_times = 100; |
| cloud::config::txn_store_retry_base_intervals_ms = 1; |
| cloud::config::fdb_cluster_file_path = "fdb.cluster"; |
| cloud::config::write_schema_kv = true; |
| cloud::config::enable_check_instance_id = false; |
| cloud::config::enable_loopback_address_for_ms = true; |
| |
| auto sp = SyncPoint::get_instance(); |
| sp->enable_processing(); |
| sp->set_call_back("encrypt_ak_sk:get_encryption_key", [](auto&& args) { |
| auto* ret = try_any_cast<int*>(args[0]); |
| *ret = 0; |
| auto* key = try_any_cast<std::string*>(args[1]); |
| *key = "test"; |
| auto* key_id = try_any_cast<int64_t*>(args[2]); |
| *key_id = 1; |
| }); |
| sp->set_call_back("decrypt_ak_sk:get_encryption_key", [](auto&& args) { |
| auto* key = try_any_cast<std::string*>(args[0]); |
| *key = "test"; |
| auto* ret = try_any_cast<int*>(args[1]); |
| *ret = 0; |
| }); |
| sp->set_call_back("MetaServiceProxy::call_impl_duration_ms", |
| [](auto&& args) { *try_any_cast<uint64_t*>(args[0]) = 0; }); |
| sp->set_call_back("put_schema_kv:schema_key_exists_return", |
| [](auto&& args) { *try_any_cast<bool*>(args.back()) = true; }); |
| sp->set_call_back("resource_manager::set_safe_drop_time", |
| [](auto&& args) { *try_any_cast<int64_t*>(args[0]) = -1; }); |
| |
| meta_service = create_meta_service(); |
| |
| int ret = RUN_ALL_TESTS(); |
| if (ret != 0) { |
| std::cerr << "run first round of tests failed" << std::endl; |
| return ret; |
| } |
| |
| std::vector<std::string> sync_points { |
| "transaction:init:create_transaction_err", |
| "transaction:commit:get_err", |
| "transaction:get:get_err", |
| "transaction:get_range:get_err", |
| "transaction:get_read_version:get_err", |
| "range_get_iterator:init:get_keyvalue_array_err", |
| }; |
| |
| // See |
| // 1. https://apple.github.io/foundationdb/api-error-codes.html#developer-guide-error-codes |
| // 2. FDB source code: flow/include/flow/error_definitions.h |
| std::vector<fdb_error_t> retryable_not_committed { |
| // future version |
| 1009, |
| // process_behind |
| 1037, |
| // database locked |
| 1038, |
| // commit_proxy_memory_limit_exceeded |
| 1042, |
| // batch_transaction_throttled |
| 1051, |
| // grv_proxy_memory_limit_exceeded |
| 1078, |
| // tag_throttled |
| 1213, |
| // proxy_tag_throttled |
| 1223, |
| }; |
| |
| for (auto err : retryable_not_committed) { |
| if (!fdb_error_predicate(FDB_ERROR_PREDICATE_RETRYABLE_NOT_COMMITTED, err)) { |
| LOG_WARNING("skip unknown err").tag("err", err).tag("msg", fdb_get_error(err)); |
| continue; |
| } |
| for (auto&& name : sync_points) { |
| for (auto&& clear_name : sync_points) { |
| sp->clear_call_back(clear_name); |
| } |
| |
| auto count = std::make_shared<std::atomic<uint64_t>>(0); |
| auto inject_at = std::make_shared<std::atomic<uint64_t>>(0); |
| sp->set_call_back(name, [=](auto&& args) mutable { |
| size_t n = count->fetch_add(1); |
| if (n == *inject_at) { |
| *try_any_cast<fdb_error_t*>(args[0]) = err; |
| } |
| }); |
| sp->set_call_back("MetaServiceProxy::call_impl:1", [=](auto&&) { |
| // For each RPC invoking, inject every fdb txn kv call. |
| count->store(0); |
| inject_at->store(0); |
| }); |
| sp->set_call_back("MetaServiceProxy::call_impl:2", [=](auto&&) { |
| count->store(0); |
| inject_at->fetch_add(1); |
| }); |
| |
| ret = RUN_ALL_TESTS(); |
| if (ret != 0) { |
| std::cerr << "run test failed, sync_point=" << name << ", err=" << err |
| << ", msg=" << fdb_get_error(err) << std::endl; |
| return ret; |
| } |
| } |
| } |
| |
| meta_service.reset(); |
| |
| return 0; |
| } |
| |
| namespace doris::cloud { |
| |
| using Status = MetaServiceResponseStatus; |
| |
| static std::string cloud_unique_id(const std::string& instance_id) { |
| // degraded format |
| return fmt::format("1:{}:unique_id", instance_id); |
| } |
| |
| static int create_instance(MetaService* service, const std::string& instance_id) { |
| CreateInstanceRequest req; |
| CreateInstanceResponse resp; |
| req.set_instance_id(instance_id); |
| req.set_user_id("user_id"); |
| req.set_name(fmt::format("instance-{}", instance_id)); |
| req.set_sse_enabled(false); |
| |
| auto* obj_info = req.mutable_obj_info(); |
| obj_info->set_ak("access-key"); |
| obj_info->set_sk("secret-key"); |
| obj_info->set_bucket("cloud-test-bucket"); |
| obj_info->set_prefix("cloud-test"); |
| obj_info->set_endpoint("endpoint"); |
| obj_info->set_external_endpoint("endpoint"); |
| obj_info->set_region("region"); |
| obj_info->set_provider(ObjectStoreInfoPB_Provider_COS); |
| |
| brpc::Controller ctrl; |
| service->create_instance(&ctrl, &req, &resp, nullptr); |
| if (ctrl.Failed()) { |
| LOG_ERROR("create instance") |
| .tag("instance_id", instance_id) |
| .tag("code", ctrl.ErrorCode()) |
| .tag("msg", ctrl.ErrorText()); |
| return -1; |
| } |
| |
| auto code = resp.status().code(); |
| if (code != cloud::MetaServiceCode::OK && code != cloud::MetaServiceCode::ALREADY_EXISTED) { |
| LOG_ERROR("create instance") |
| .tag("instance_id", instance_id) |
| .tag("code", code) |
| .tag("msg", resp.status().msg()); |
| return -1; |
| } |
| |
| return 0; |
| } |
| |
| static int remove_instance(MetaService* service, const std::string& instance_id) { |
| AlterInstanceRequest req; |
| AlterInstanceResponse resp; |
| req.set_instance_id(instance_id); |
| req.set_op(AlterInstanceRequest_Operation_DROP); |
| |
| brpc::Controller ctrl; |
| service->alter_instance(&ctrl, &req, &resp, nullptr); |
| if (ctrl.Failed()) { |
| LOG_ERROR("alter_instance") |
| .tag("instance_id", instance_id) |
| .tag("code", ctrl.ErrorCode()) |
| .tag("msg", ctrl.ErrorText()); |
| return -1; |
| } |
| |
| auto code = resp.status().code(); |
| if (code != cloud::MetaServiceCode::OK) { |
| LOG_ERROR("create instance") |
| .tag("instance_id", instance_id) |
| .tag("code", code) |
| .tag("msg", resp.status().msg()); |
| return -1; |
| } |
| |
| return 0; |
| } |
| |
| static int add_cluster(MetaService* service, const std::string& instance_id) { |
| bool retry = false; |
| while (true) { |
| brpc::Controller ctrl; |
| cloud::AlterClusterRequest req; |
| cloud::AlterClusterResponse resp; |
| |
| req.set_instance_id(instance_id); |
| req.set_op(cloud::AlterClusterRequest_Operation::AlterClusterRequest_Operation_ADD_CLUSTER); |
| auto* cluster = req.mutable_cluster(); |
| auto name = fmt::format("instance_{}_cluster", instance_id); |
| cluster->set_cluster_id(name); |
| cluster->set_cluster_name(name); |
| cluster->set_type(cloud::ClusterPB_Type::ClusterPB_Type_SQL); |
| cluster->set_desc("cluster description"); |
| auto* node = cluster->add_nodes(); |
| node->set_ip("0.0.0.0"); |
| node->set_node_type(cloud::NodeInfoPB_NodeType::NodeInfoPB_NodeType_FE_MASTER); |
| node->set_cloud_unique_id(cloud_unique_id(instance_id)); |
| node->set_edit_log_port(123); |
| node->set_heartbeat_port(456); |
| node->set_name("default_node"); |
| |
| service->alter_cluster(&ctrl, &req, &resp, nullptr); |
| if (ctrl.Failed()) { |
| LOG_ERROR("alter cluster") |
| .tag("instance_id", instance_id) |
| .tag("code", ctrl.ErrorCode()) |
| .tag("msg", ctrl.ErrorText()); |
| return -1; |
| } |
| |
| auto code = resp.status().code(); |
| if (code == cloud::MetaServiceCode::OK) { |
| return 0; |
| } else if (code == cloud::MetaServiceCode::ALREADY_EXISTED) { |
| if (!retry) { |
| retry = true; |
| req.set_op(cloud::AlterClusterRequest_Operation:: |
| AlterClusterRequest_Operation_DROP_CLUSTER); |
| ctrl.Reset(); |
| service->alter_cluster(&ctrl, &req, &resp, nullptr); |
| } |
| return 0; |
| } else { |
| LOG_ERROR("add default cluster") |
| .tag("instance_id", instance_id) |
| .tag("code", code) |
| .tag("msg", resp.status().msg()); |
| return -1; |
| } |
| } |
| } |
| |
| static int drop_cluster(MetaService* service, const std::string& instance_id) { |
| AlterClusterRequest req; |
| AlterClusterResponse resp; |
| req.set_instance_id(instance_id); |
| req.set_op(AlterClusterRequest_Operation_DROP_CLUSTER); |
| auto cluster = req.mutable_cluster(); |
| cluster->set_cluster_id(fmt::format("instance_{}_cluster", instance_id)); |
| |
| brpc::Controller ctrl; |
| service->alter_cluster(&ctrl, &req, &resp, nullptr); |
| if (ctrl.Failed()) { |
| LOG_ERROR("drop cluster") |
| .tag("instance_id", instance_id) |
| .tag("code", ctrl.ErrorCode()) |
| .tag("msg", ctrl.ErrorText()); |
| return -1; |
| } |
| if (resp.status().code() != cloud::MetaServiceCode::OK) { |
| LOG_ERROR("drop cluster") |
| .tag("instance_id", instance_id) |
| .tag("code", resp.status().code()) |
| .tag("msg", resp.status().msg()); |
| return -1; |
| } |
| return 0; |
| } |
| |
| static doris::TabletMetaCloudPB add_tablet(int64_t table_id, int64_t index_id, int64_t partition_id, |
| int64_t tablet_id) { |
| doris::TabletMetaCloudPB tablet; |
| tablet.set_table_id(table_id); |
| tablet.set_index_id(index_id); |
| tablet.set_partition_id(partition_id); |
| tablet.set_tablet_id(tablet_id); |
| auto schema = tablet.mutable_schema(); |
| schema->set_schema_version(1); |
| auto first_rowset = tablet.add_rs_metas(); |
| first_rowset->set_rowset_id(0); // required |
| first_rowset->set_rowset_id_v2(std::to_string(1)); |
| first_rowset->set_start_version(0); |
| first_rowset->set_end_version(1); |
| first_rowset->mutable_tablet_schema()->CopyFrom(*schema); |
| return tablet; |
| } |
| |
| static int create_tablet(MetaService* meta_service, const std::string& instance_id, |
| int64_t table_id, int64_t index_id, int64_t partition_id, |
| int64_t tablet_id) { |
| if (tablet_id < 0) { |
| LOG_ERROR("invalid tablet id").tag("id", tablet_id); |
| return -1; |
| } |
| |
| brpc::Controller ctrl; |
| cloud::CreateTabletsRequest req; |
| cloud::CreateTabletsResponse resp; |
| req.set_cloud_unique_id(cloud_unique_id(instance_id)); |
| req.add_tablet_metas()->CopyFrom(add_tablet(table_id, index_id, partition_id, tablet_id)); |
| meta_service->create_tablets(&ctrl, &req, &resp, nullptr); |
| if (ctrl.Failed()) { |
| LOG_ERROR("create_tablets") |
| .tag("instance_id", instance_id) |
| .tag("tablet_id", tablet_id) |
| .tag("code", ctrl.ErrorCode()) |
| .tag("msg", ctrl.ErrorText()); |
| return -1; |
| } |
| if (resp.status().code() != cloud::MetaServiceCode::OK) { |
| LOG_ERROR("create tablet") |
| .tag("instance_id", instance_id) |
| .tag("tablet_id", tablet_id) |
| .tag("code", resp.status().code()) |
| .tag("msg", resp.status().msg()); |
| return -1; |
| } |
| return 0; |
| } |
| |
| static Status begin_txn(MetaService* service, const std::string& instance_id, std::string label, |
| int64_t db_id, const std::vector<uint64_t>& tablet_ids, int64_t* txn_id) { |
| brpc::Controller ctrl; |
| cloud::BeginTxnRequest req; |
| cloud::BeginTxnResponse resp; |
| req.set_cloud_unique_id(cloud_unique_id(instance_id)); |
| auto* txn_info = req.mutable_txn_info(); |
| txn_info->set_label(label); |
| txn_info->set_db_id(db_id); |
| txn_info->mutable_table_ids()->Add(tablet_ids.begin(), tablet_ids.end()); |
| txn_info->set_timeout_ms(1000 * 60 * 60); |
| |
| service->begin_txn(&ctrl, &req, &resp, nullptr); |
| if (ctrl.Failed()) { |
| LOG_ERROR("begin_txn") |
| .tag("instance_id", instance_id) |
| .tag("table_ids", fmt::format("{}", fmt::join(tablet_ids, ","))) |
| .tag("label", label) |
| .tag("code", ctrl.ErrorCode()) |
| .tag("msg", ctrl.ErrorText()); |
| Status status; |
| status.set_code(MetaServiceCode::UNDEFINED_ERR); |
| status.set_msg(ctrl.ErrorText()); |
| return status; |
| } |
| |
| *txn_id = resp.txn_id(); |
| return resp.status(); |
| } |
| |
| static Status commit_txn(MetaService* service, const std::string& instance_id, int64_t txn_id, |
| int64_t db_id) { |
| brpc::Controller ctrl; |
| cloud::CommitTxnRequest req; |
| cloud::CommitTxnResponse resp; |
| req.set_cloud_unique_id(cloud_unique_id(instance_id)); |
| req.set_db_id(db_id); |
| req.set_txn_id(txn_id); |
| |
| service->commit_txn(&ctrl, &req, &resp, nullptr); |
| if (ctrl.Failed()) { |
| LOG_ERROR("commit_txn") |
| .tag("instance_id", instance_id) |
| .tag("txn_id", txn_id) |
| .tag("code", ctrl.ErrorCode()) |
| .tag("msg", ctrl.ErrorText()); |
| Status status; |
| status.set_code(MetaServiceCode::UNDEFINED_ERR); |
| status.set_msg(ctrl.ErrorText()); |
| return status; |
| } |
| |
| return resp.status(); |
| } |
| |
| static doris::RowsetMetaCloudPB create_rowset(int64_t tablet_id, std::string rowset_id, |
| int64_t txn_id, int64_t index_id, |
| int64_t partition_id, int64_t version = -1, |
| int num_rows = 100) { |
| doris::RowsetMetaCloudPB rowset; |
| rowset.set_rowset_id(0); // required |
| rowset.set_rowset_id_v2(rowset_id); |
| rowset.set_index_id(index_id); |
| rowset.set_partition_id(partition_id); |
| rowset.set_tablet_id(tablet_id); |
| rowset.set_txn_id(txn_id); |
| if (version >= 0) { |
| rowset.set_start_version(version); |
| rowset.set_end_version(version); |
| } |
| rowset.set_num_segments(1); |
| rowset.set_num_rows(num_rows); |
| rowset.set_data_disk_size(num_rows * 100); |
| rowset.set_txn_expiration(10000); |
| rowset.set_schema_version(1); |
| |
| // auto* schema = rowset.mutable_tablet_schema(); |
| // schema->set_schema_version(1); |
| // schema->set_disable_auto_compaction(true); |
| |
| auto* bound = rowset.add_segments_key_bounds(); |
| bound->set_min_key("min_key"); |
| bound->set_max_key("max_key"); |
| |
| return rowset; |
| } |
| |
| static Status prepare_rowset(MetaService* service, const std::string& instance_id, int64_t txn_id, |
| int64_t tablet_id, std::string rowset_id, int64_t index_id, |
| int64_t partition_id) { |
| brpc::Controller ctrl; |
| cloud::CreateRowsetRequest req; |
| cloud::CreateRowsetResponse resp; |
| req.set_cloud_unique_id(cloud_unique_id(instance_id)); |
| req.mutable_rowset_meta()->CopyFrom( |
| create_rowset(tablet_id, rowset_id, txn_id, index_id, partition_id)); |
| |
| service->prepare_rowset(&ctrl, &req, &resp, nullptr); |
| if (ctrl.Failed()) { |
| LOG_ERROR("prepare_rowset") |
| .tag("instance_id", instance_id) |
| .tag("txn_id", txn_id) |
| .tag("tablet_id", tablet_id) |
| .tag("rowset_id", rowset_id) |
| .tag("code", ctrl.ErrorCode()) |
| .tag("msg", ctrl.ErrorText()); |
| Status status; |
| status.set_code(MetaServiceCode::UNDEFINED_ERR); |
| status.set_msg(ctrl.ErrorText()); |
| return status; |
| } |
| return resp.status(); |
| } |
| |
| static Status commit_rowset(MetaService* service, const std::string& instance_id, int64_t txn_id, |
| int64_t tablet_id, std::string rowset_id, int64_t index_id, |
| int64_t partition_id) { |
| brpc::Controller ctrl; |
| cloud::CreateRowsetRequest req; |
| cloud::CreateRowsetResponse resp; |
| req.set_cloud_unique_id(cloud_unique_id(instance_id)); |
| req.mutable_rowset_meta()->CopyFrom( |
| create_rowset(tablet_id, rowset_id, txn_id, index_id, partition_id)); |
| LOG_INFO("send commit rowset request"); |
| |
| service->commit_rowset(&ctrl, &req, &resp, nullptr); |
| if (ctrl.Failed()) { |
| LOG_ERROR("commit_rowset") |
| .tag("instance_id", instance_id) |
| .tag("txn_id", txn_id) |
| .tag("tablet_id", tablet_id) |
| .tag("rowset_id", rowset_id) |
| .tag("code", ctrl.ErrorCode()) |
| .tag("msg", ctrl.ErrorText()); |
| |
| Status status; |
| status.set_code(MetaServiceCode::UNDEFINED_ERR); |
| status.set_msg(ctrl.ErrorText()); |
| return status; |
| } |
| return resp.status(); |
| } |
| |
| static Status insert_rowset(MetaService* service, const std::string& instance_id, int64_t txn_id, |
| int64_t tablet_id, std::string rowset_id, int64_t index_id, |
| int64_t partition_id) { |
| auto status = prepare_rowset(service, instance_id, txn_id, tablet_id, rowset_id, index_id, |
| partition_id); |
| if (status.code() != cloud::MetaServiceCode::OK) { |
| LOG_ERROR("prepare_rowset") |
| .tag("instance_id", instance_id) |
| .tag("txn_id", txn_id) |
| .tag("tablet_id", tablet_id) |
| .tag("rowset_id", rowset_id) |
| .tag("code", status.code()) |
| .tag("msg", status.msg()); |
| if (status.code() != cloud::MetaServiceCode::ALREADY_EXISTED) { |
| return status; |
| } |
| } |
| |
| LOG_INFO("prepare rowset").tag("code", status.code()).tag("msg", status.msg()); |
| status = commit_rowset(service, instance_id, txn_id, tablet_id, rowset_id, index_id, |
| partition_id); |
| if (status.code() != cloud::MetaServiceCode::OK) { |
| LOG_ERROR("commit_rowset") |
| .tag("instance_id", instance_id) |
| .tag("txn_id", txn_id) |
| .tag("tablet_id", tablet_id) |
| .tag("rowset_id", rowset_id) |
| .tag("code", status.code()) |
| .tag("msg", status.msg()); |
| return status; |
| } |
| |
| return Status(); |
| } |
| |
| static Status copy_into(MetaService* service, const std::string& instance_id, std::string label, |
| int64_t db_id, int64_t index_id, int64_t partition_id, |
| const std::vector<uint64_t>& tablet_ids) { |
| int64_t txn_id = 0; |
| auto status = begin_txn(service, instance_id, label, db_id, tablet_ids, &txn_id); |
| if (status.code() != cloud::MetaServiceCode::OK) { |
| LOG_ERROR("begin_txn") |
| .tag("instance_id", instance_id) |
| .tag("label", label) |
| .tag("code", status.code()) |
| .tag("msg", status.msg()); |
| return status; |
| } |
| |
| for (uint64_t tablet_id : tablet_ids) { |
| auto rowset_id = fmt::format("rowset_{}_{}", label, tablet_id); |
| status = insert_rowset(service, instance_id, txn_id, tablet_id, rowset_id, index_id, |
| partition_id); |
| if (status.code() != cloud::MetaServiceCode::OK) { |
| return status; |
| } |
| } |
| |
| status = commit_txn(service, instance_id, txn_id, db_id); |
| if (status.code() != cloud::MetaServiceCode::OK) { |
| LOG_ERROR("commit_txn") |
| .tag("instance_id", instance_id) |
| .tag("label", label) |
| .tag("code", status.code()) |
| .tag("msg", status.msg()); |
| return status; |
| } |
| |
| return Status(); |
| } |
| |
| static Status get_read_version(MetaService* service, const std::string& instance_id, |
| int64_t* version) { |
| brpc::Controller ctrl; |
| cloud::GetCurrentMaxTxnRequest req; |
| cloud::GetCurrentMaxTxnResponse resp; |
| req.set_cloud_unique_id(cloud_unique_id(instance_id)); |
| service->get_current_max_txn_id(&ctrl, &req, &resp, nullptr); |
| |
| if (ctrl.Failed()) { |
| LOG_ERROR("get read version") |
| .tag("instance_id", instance_id) |
| .tag("code", ctrl.ErrorCode()) |
| .tag("msg", ctrl.ErrorText()); |
| Status status; |
| status.set_code(MetaServiceCode::UNDEFINED_ERR); |
| status.set_msg(ctrl.ErrorText()); |
| return status; |
| } |
| |
| auto status = resp.status(); |
| if (status.code() != cloud::MetaServiceCode::OK) { |
| LOG_ERROR("get read version") |
| .tag("instance_id", instance_id) |
| .tag("code", status.code()) |
| .tag("msg", status.msg()); |
| return status; |
| } |
| |
| *version = resp.current_max_txn_id(); |
| |
| return Status(); |
| } |
| |
| static Status get_version(MetaService* service, const std::string& instance_id, int64_t db_id, |
| int64_t partition_id, int64_t table_id, int64_t* version) { |
| brpc::Controller ctrl; |
| cloud::GetVersionRequest req; |
| cloud::GetVersionResponse resp; |
| req.set_cloud_unique_id(cloud_unique_id(instance_id)); |
| req.set_db_id(db_id); |
| req.set_partition_id(partition_id); |
| req.set_table_id(table_id); |
| |
| service->get_version(&ctrl, &req, &resp, nullptr); |
| if (ctrl.Failed()) { |
| LOG_ERROR("get version") |
| .tag("instance_id", instance_id) |
| .tag("db_id", db_id) |
| .tag("partition_id", partition_id) |
| .tag("table_id", table_id) |
| .tag("code", ctrl.ErrorCode()) |
| .tag("msg", ctrl.ErrorText()); |
| Status status; |
| status.set_code(MetaServiceCode::UNDEFINED_ERR); |
| status.set_msg(ctrl.ErrorText()); |
| return status; |
| } |
| |
| auto status = resp.status(); |
| if (status.code() != cloud::MetaServiceCode::OK) { |
| LOG_ERROR("get version") |
| .tag("instance_id", instance_id) |
| .tag("db_id", db_id) |
| .tag("partition_id", partition_id) |
| .tag("table_id", table_id) |
| .tag("code", status.code()) |
| .tag("msg", status.msg()); |
| return status; |
| } |
| |
| *version = resp.version(); |
| |
| return Status(); |
| } |
| |
| static Status get_tablet_meta(MetaService* service, const std::string& instance_id, |
| int64_t tablet_id) { |
| brpc::Controller ctrl; |
| cloud::GetTabletRequest req; |
| cloud::GetTabletResponse resp; |
| req.set_cloud_unique_id(cloud_unique_id(instance_id)); |
| req.set_tablet_id(tablet_id); |
| service->get_tablet(&ctrl, &req, &resp, nullptr); |
| if (ctrl.Failed()) { |
| LOG_ERROR("get_tablet") |
| .tag("instance_id", instance_id) |
| .tag("tablet_id", tablet_id) |
| .tag("code", ctrl.ErrorCode()) |
| .tag("msg", ctrl.ErrorText()); |
| Status status; |
| status.set_code(MetaServiceCode::UNDEFINED_ERR); |
| status.set_msg(ctrl.ErrorText()); |
| return status; |
| } |
| |
| auto status = resp.status(); |
| if (status.code() != cloud::MetaServiceCode::OK) { |
| LOG_ERROR("get_tablet") |
| .tag("instance_id", instance_id) |
| .tag("tablet_id", tablet_id) |
| .tag("code", status.code()) |
| .tag("msg", status.msg()); |
| return status; |
| } |
| |
| return Status(); |
| } |
| |
| static Status get_tablet_stats(MetaService* service, const std::string& instance_id, |
| int64_t tablet_id, cloud::TabletStatsPB* stats) { |
| brpc::Controller ctrl; |
| cloud::GetTabletStatsRequest req; |
| cloud::GetTabletStatsResponse resp; |
| req.set_cloud_unique_id(cloud_unique_id(instance_id)); |
| auto* tablet_idx = req.mutable_tablet_idx()->Add(); |
| tablet_idx->set_tablet_id(tablet_id); |
| service->get_tablet_stats(&ctrl, &req, &resp, nullptr); |
| if (ctrl.Failed()) { |
| LOG_ERROR("get tablet stats") |
| .tag("instance_id", instance_id) |
| .tag("tablet_id", tablet_id) |
| .tag("code", ctrl.ErrorCode()) |
| .tag("msg", ctrl.ErrorText()); |
| Status status; |
| status.set_code(MetaServiceCode::UNDEFINED_ERR); |
| status.set_msg(ctrl.ErrorText()); |
| return status; |
| } |
| |
| auto status = resp.status(); |
| if (status.code() != cloud::MetaServiceCode::OK) { |
| LOG_ERROR("get_tablet_stats") |
| .tag("instance_id", instance_id) |
| .tag("tablet_id", tablet_id) |
| .tag("code", status.code()) |
| .tag("msg", status.msg()); |
| return status; |
| } |
| |
| stats->CopyFrom(resp.tablet_stats().at(0)); |
| |
| return Status(); |
| } |
| |
| static Status get_rowset_meta(MetaService* service, const std::string& instance_id, |
| int64_t tablet_id, int64_t version, |
| const cloud::TabletStatsPB& stats) { |
| brpc::Controller ctrl; |
| cloud::GetRowsetRequest req; |
| cloud::GetRowsetResponse resp; |
| req.set_cloud_unique_id(cloud_unique_id(instance_id)); |
| req.mutable_idx()->set_tablet_id(tablet_id); |
| req.set_start_version(0); |
| req.set_end_version(version); |
| req.set_base_compaction_cnt(stats.base_compaction_cnt()); |
| req.set_cumulative_compaction_cnt(stats.cumulative_compaction_cnt()); |
| req.set_cumulative_point(stats.cumulative_point()); |
| service->get_rowset(&ctrl, &req, &resp, nullptr); |
| if (ctrl.Failed()) { |
| LOG_ERROR("get_rowset") |
| .tag("instance_id", instance_id) |
| .tag("tablet_id", tablet_id) |
| .tag("code", ctrl.ErrorCode()) |
| .tag("msg", ctrl.ErrorText()); |
| Status status; |
| status.set_code(MetaServiceCode::UNDEFINED_ERR); |
| status.set_msg(ctrl.ErrorText()); |
| return status; |
| } |
| |
| auto status = resp.status(); |
| if (status.code() != cloud::MetaServiceCode::OK) { |
| LOG_ERROR("get_rowset") |
| .tag("instance_id", instance_id) |
| .tag("tablet_id", tablet_id) |
| .tag("code", status.code()) |
| .tag("msg", status.msg()); |
| return status; |
| } |
| |
| return Status(); |
| } |
| |
| TEST(FdbInjectionTest, AllInOne) { |
| auto now = std::chrono::high_resolution_clock::now(); |
| auto ns = std::chrono::duration_cast<std::chrono::nanoseconds>(now.time_since_epoch()); |
| std::string instance_id = fmt::format("fdb_injection_test_{}", ns.count()); |
| int ret = create_instance(meta_service.get(), instance_id); |
| ASSERT_EQ(ret, 0); |
| |
| ret = add_cluster(meta_service.get(), instance_id); |
| ASSERT_EQ(ret, 0); |
| |
| int64_t db_id = 1; |
| int64_t table_id = 1; |
| int64_t index_id = 1; |
| int64_t partition_id = 1; |
| std::vector<uint64_t> tablet_ids; |
| for (int64_t tablet_id = 1; tablet_id <= 10; ++tablet_id) { |
| ret = create_tablet(meta_service.get(), instance_id, table_id, index_id, partition_id, |
| tablet_id); |
| ASSERT_EQ(ret, 0) << tablet_id; |
| tablet_ids.push_back(tablet_id); |
| } |
| |
| // Run copy into. |
| std::string label = fmt::format("{}-label", instance_id); |
| auto status = copy_into(meta_service.get(), instance_id, label, db_id, index_id, partition_id, |
| tablet_ids); |
| ASSERT_EQ(status.code(), MetaServiceCode::OK) << status.msg(); |
| |
| // Get version |
| int64_t version = 0; |
| status = get_version(meta_service.get(), instance_id, db_id, partition_id, table_id, &version); |
| ASSERT_EQ(status.code(), MetaServiceCode::OK) << status.msg(); |
| |
| // Get tablet meta & stats |
| TabletStatsPB stats; |
| status = get_tablet_stats(meta_service.get(), instance_id, tablet_ids.back(), &stats); |
| ASSERT_EQ(status.code(), MetaServiceCode::OK) << status.msg(); |
| |
| status = get_tablet_meta(meta_service.get(), instance_id, tablet_ids.back()); |
| ASSERT_EQ(status.code(), MetaServiceCode::OK) << status.msg(); |
| |
| // Get rowset metas. |
| status = get_rowset_meta(meta_service.get(), instance_id, tablet_ids.back(), version, stats); |
| ASSERT_EQ(status.code(), MetaServiceCode::OK) << status.msg(); |
| |
| int64_t max_txn_id = 0; |
| status = get_read_version(meta_service.get(), instance_id, &max_txn_id); |
| ASSERT_EQ(status.code(), MetaServiceCode::OK) << status.msg(); |
| ASSERT_GE(max_txn_id, version); |
| |
| ret = drop_cluster(meta_service.get(), instance_id); |
| ASSERT_EQ(ret, 0); |
| |
| ret = remove_instance(meta_service.get(), instance_id); |
| ASSERT_EQ(ret, 0); |
| } |
| |
| } // namespace doris::cloud |
