You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by ad...@apache.org on 2017/10/10 20:23:38 UTC
kudu git commit: itest: allow use of verifiers with EMCs that specify
non-default block managers
Repository: kudu
Updated Branches:
refs/heads/master 2a802f9f3 -> 6523a773e
itest: allow use of verifiers with EMCs that specify non-default block managers
ClusterVerifier and LogVerifier both rely on opening a new FsManager,
using the current FLAGS_block_manager of the caller process.
ExternalMiniClusters, on the other hand, grant the flexibility to
specify any flags, and create the cluster in a separate processes.
This precluded the Verifiers from operating on ExternalMiniClusters that
are specified with different block manager types than that of the
Verifier's caller (e.g. in parameterized EMC tests).
To remediate this, block manager type has been added to various
constructor Options (i.e. ExternalMiniClusterOptions,
ExternalDaemonOptions, FsManagerOpts, DataDirManagerOptions).
This field is used by the Verifiers when opening a FsManager.
There is also some cleanup of ExternalDaemon to store and use
ExternalDaemonOptions instead of the individual members.
To test, I parameterized disk_failure-itest, which exercises using a
ClusterVerifier and LogVerifier, and restarts an ExternalTabletServer.
Change-Id: Iaa44eb33d0c025830f97a2ed7583c8186f915e94
Reviewed-on: http://gerrit.cloudera.org:8080/8208
Reviewed-by: Adar Dembo <ad...@cloudera.com>
Tested-by: Kudu Jenkins
Reviewed-by: Mike Percy <mp...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/6523a773
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/6523a773
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/6523a773
Branch: refs/heads/master
Commit: 6523a773eda899fd376dd0b68871c3dcd5737b8e
Parents: 2a802f9
Author: Andrew Wong <aw...@cloudera.com>
Authored: Tue Oct 3 14:12:38 2017 -0700
Committer: Adar Dembo <ad...@cloudera.com>
Committed: Tue Oct 10 20:23:21 2017 +0000
----------------------------------------------------------------------
src/kudu/fs/block_manager.h | 9 ++
src/kudu/fs/data_dirs.cc | 5 +-
src/kudu/fs/data_dirs.h | 6 +
src/kudu/fs/fs_manager.cc | 16 ++-
src/kudu/fs/fs_manager.h | 7 +
.../integration-tests/disk_failure-itest.cc | 17 ++-
.../integration-tests/disk_reservation-itest.cc | 7 +-
.../external_mini_cluster-itest-base.cc | 14 +-
.../external_mini_cluster-itest-base.h | 7 +-
src/kudu/integration-tests/log_verifier.cc | 2 +
.../integration-tests/multidir_cluster-itest.cc | 8 +-
src/kudu/mini-cluster/external_mini_cluster.cc | 140 ++++++++++---------
src/kudu/mini-cluster/external_mini_cluster.h | 42 +++---
src/kudu/server/server_base.cc | 1 +
14 files changed, 165 insertions(+), 116 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/6523a773/src/kudu/fs/block_manager.h
----------------------------------------------------------------------
diff --git a/src/kudu/fs/block_manager.h b/src/kudu/fs/block_manager.h
index d682546..991b4ed 100644
--- a/src/kudu/fs/block_manager.h
+++ b/src/kudu/fs/block_manager.h
@@ -199,6 +199,15 @@ struct BlockManagerOptions {
// thread-safe.
class BlockManager {
public:
+ // Lists the available block manager types.
+ static std::vector<std::string> block_manager_types() {
+#if defined(__linux__)
+ return { "file", "log" };
+#else
+ return { "file" };
+#endif
+ }
+
virtual ~BlockManager() {}
// Opens an existing on-disk representation of this block manager and
http://git-wip-us.apache.org/repos/asf/kudu/blob/6523a773/src/kudu/fs/data_dirs.cc
----------------------------------------------------------------------
diff --git a/src/kudu/fs/data_dirs.cc b/src/kudu/fs/data_dirs.cc
index 42fa7f3..ac839b0 100644
--- a/src/kudu/fs/data_dirs.cc
+++ b/src/kudu/fs/data_dirs.cc
@@ -279,7 +279,8 @@ Status DataDir::RefreshIsFull(RefreshMode mode) {
const char* DataDirManager::kDataDirName = "data";
DataDirManagerOptions::DataDirManagerOptions()
- : read_only(false) {
+ : block_manager_type(FLAGS_block_manager),
+ read_only(false) {
}
vector<string> DataDirManager::GetRootNames(const CanonicalizedRootsList& root_list) {
@@ -293,7 +294,7 @@ DataDirManager::DataDirManager(Env* env,
DataDirManagerOptions opts,
CanonicalizedRootsList canonicalized_data_roots)
: env_(env),
- block_manager_type_(FLAGS_block_manager),
+ block_manager_type_(std::move(opts.block_manager_type)),
read_only_(opts.read_only),
canonicalized_data_fs_roots_(std::move(canonicalized_data_roots)),
rng_(GetRandomSeed32()) {
http://git-wip-us.apache.org/repos/asf/kudu/blob/6523a773/src/kudu/fs/data_dirs.h
----------------------------------------------------------------------
diff --git a/src/kudu/fs/data_dirs.h b/src/kudu/fs/data_dirs.h
index b1de012..0ac01da 100644
--- a/src/kudu/fs/data_dirs.h
+++ b/src/kudu/fs/data_dirs.h
@@ -200,6 +200,12 @@ class DataDir {
struct DataDirManagerOptions {
DataDirManagerOptions();
+ // The block manager type the directory manager should support.
+ // Must be either "file" or "log".
+ //
+ // Defaults to the value of FLAGS_block_manager.
+ std::string block_manager_type;
+
// The entity under which all metrics should be grouped. If null, metrics
// will not be produced.
//
http://git-wip-us.apache.org/repos/asf/kudu/blob/6523a773/src/kudu/fs/fs_manager.cc
----------------------------------------------------------------------
diff --git a/src/kudu/fs/fs_manager.cc b/src/kudu/fs/fs_manager.cc
index d85f3f7..a0c78c6 100644
--- a/src/kudu/fs/fs_manager.cc
+++ b/src/kudu/fs/fs_manager.cc
@@ -69,16 +69,16 @@ TAG_FLAG(enable_data_block_fsync, unsafe);
#if defined(__linux__)
DEFINE_string(block_manager, "log", "Which block manager to use for storage. "
"Valid options are 'file' and 'log'.");
-static bool ValidateBlockManagerType(const char* /*flagname*/, const std::string& value) {
- return value == "log" || value == "file";
-}
#else
DEFINE_string(block_manager, "file", "Which block manager to use for storage. "
"Only the file block manager is supported for non-Linux systems.");
+#endif
static bool ValidateBlockManagerType(const char* /*flagname*/, const std::string& value) {
- return value == "file";
+ for (const std::string& type : kudu::fs::BlockManager::block_manager_types()) {
+ if (type == value) return true;
+ }
+ return false;
}
-#endif
DEFINE_validator(block_manager, &ValidateBlockManagerType);
TAG_FLAG(block_manager, advanced);
@@ -129,6 +129,7 @@ const char *FsManager::kConsensusMetadataDirName = "consensus-meta";
FsManagerOpts::FsManagerOpts()
: wal_path(FLAGS_fs_wal_dir),
+ block_manager_type(FLAGS_block_manager),
read_only(false) {
data_paths = strings::Split(FLAGS_fs_data_dirs, ",", strings::SkipEmpty());
}
@@ -139,6 +140,7 @@ FsManagerOpts::~FsManagerOpts() {
FsManager::FsManager(Env* env, const string& root_path)
: env_(DCHECK_NOTNULL(env)),
read_only_(false),
+ block_manager_type_(FLAGS_block_manager),
wal_fs_root_(root_path),
data_fs_roots_({ root_path }),
metric_entity_(nullptr),
@@ -150,6 +152,7 @@ FsManager::FsManager(Env* env,
const FsManagerOpts& opts)
: env_(DCHECK_NOTNULL(env)),
read_only_(opts.read_only),
+ block_manager_type_(opts.block_manager_type),
wal_fs_root_(opts.wal_path),
data_fs_roots_(opts.data_paths),
metric_entity_(opts.metric_entity),
@@ -269,7 +272,7 @@ void FsManager::InitBlockManager() {
opts.metric_entity = metric_entity_;
opts.parent_mem_tracker = parent_mem_tracker_;
opts.read_only = read_only_;
- if (FLAGS_block_manager == "file") {
+ if (block_manager_type_ == "file") {
block_manager_.reset(new FileBlockManager(env_, dd_manager_.get(), error_manager_.get(), opts));
} else {
block_manager_.reset(new LogBlockManager(env_, dd_manager_.get(), error_manager_.get(), opts));
@@ -320,6 +323,7 @@ Status FsManager::Open(FsReport* report) {
if (!dd_manager_) {
DataDirManagerOptions dm_opts;
dm_opts.metric_entity = metric_entity_;
+ dm_opts.block_manager_type = block_manager_type_;
dm_opts.read_only = read_only_;
LOG_TIMING(INFO, "opening directory manager") {
RETURN_NOT_OK(DataDirManager::OpenExisting(env_,
http://git-wip-us.apache.org/repos/asf/kudu/blob/6523a773/src/kudu/fs/fs_manager.h
----------------------------------------------------------------------
diff --git a/src/kudu/fs/fs_manager.h b/src/kudu/fs/fs_manager.h
index 2ff8b91..f5ec089 100644
--- a/src/kudu/fs/fs_manager.h
+++ b/src/kudu/fs/fs_manager.h
@@ -80,6 +80,10 @@ struct FsManagerOpts {
// The paths where data blocks will be stored. Cannot be empty.
std::vector<std::string> data_paths;
+ // The block manager type. Must be either "file" or "log".
+ // Defaults to the value of FLAGS_block_manager.
+ std::string block_manager_type;
+
// Whether or not read-write operations should be allowed. Defaults to false.
bool read_only;
};
@@ -289,6 +293,9 @@ class FsManager {
// If false, operations that mutate on-disk state are prohibited.
const bool read_only_;
+ // The block manager type.
+ const std::string block_manager_type_;
+
// These roots are the constructor input verbatim. None of them are used
// as-is; they are first canonicalized during Init().
const std::string wal_fs_root_;
http://git-wip-us.apache.org/repos/asf/kudu/blob/6523a773/src/kudu/integration-tests/disk_failure-itest.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/disk_failure-itest.cc b/src/kudu/integration-tests/disk_failure-itest.cc
index ff240f0..4bb888d 100644
--- a/src/kudu/integration-tests/disk_failure-itest.cc
+++ b/src/kudu/integration-tests/disk_failure-itest.cc
@@ -26,6 +26,7 @@
#include "kudu/integration-tests/cluster_verifier.h"
#include "kudu/integration-tests/external_mini_cluster-itest-base.h"
#include "kudu/integration-tests/test_workload.h"
+#include "kudu/fs/block_manager.h"
#include "kudu/mini-cluster/external_mini_cluster.h"
#include "kudu/util/metrics.h"
#include "kudu/util/monotime.h"
@@ -37,14 +38,17 @@ METRIC_DECLARE_gauge_uint64(data_dirs_failed);
namespace kudu {
+using cluster::ExternalMiniClusterOptions;
using cluster::ExternalTabletServer;
+using fs::BlockManager;
using std::string;
using std::vector;
using strings::Substitute;
const MonoDelta kAgreementTimeout = MonoDelta::FromSeconds(30);
-class DiskFailureITest : public ExternalMiniClusterITestBase {
+class DiskFailureITest : public ExternalMiniClusterITestBase,
+ public ::testing::WithParamInterface<string> {
public:
// Waits for 'ext_tserver' to experience 'target_failed_disks' disk failures.
@@ -63,9 +67,13 @@ class DiskFailureITest : public ExternalMiniClusterITestBase {
// cluster is started and loaded with some tablets. The tablet server is then
// shut down and restarted. Errors are injected to one of the directories while
// it is shut down.
-TEST_F(DiskFailureITest, TestFailDuringServerStartup) {
+TEST_P(DiskFailureITest, TestFailDuringServerStartup) {
// Set up a cluster with three servers with five disks each.
- NO_FATALS(StartCluster({}, {}, 3, 5));
+ ExternalMiniClusterOptions opts;
+ opts.num_tablet_servers = 3;
+ opts.num_data_dirs = 5;
+ opts.block_manager_type = GetParam();
+ NO_FATALS(StartClusterWithOpts(opts));
const int kNumTablets = 5;
const int kNumRows = 100;
@@ -106,4 +114,7 @@ TEST_F(DiskFailureITest, TestFailDuringServerStartup) {
write_workload.batches_completed()));
}
+INSTANTIATE_TEST_CASE_P(DiskFailure, DiskFailureITest,
+ ::testing::ValuesIn(BlockManager::block_manager_types()));
+
} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/6523a773/src/kudu/integration-tests/disk_reservation-itest.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/disk_reservation-itest.cc b/src/kudu/integration-tests/disk_reservation-itest.cc
index e25582f..5b5d26b 100644
--- a/src/kudu/integration-tests/disk_reservation-itest.cc
+++ b/src/kudu/integration-tests/disk_reservation-itest.cc
@@ -19,6 +19,7 @@
#include <memory>
#include <ostream>
#include <string>
+#include <utility>
#include <vector>
#include <glog/logging.h>
@@ -42,6 +43,7 @@ METRIC_DECLARE_gauge_uint64(data_dirs_full);
namespace kudu {
+using cluster::ExternalMiniClusterOptions;
using cluster::ExternalTabletServer;
namespace {
@@ -73,7 +75,10 @@ TEST_F(DiskReservationITest, TestFillMultipleDisks) {
// failing requests.
ts_flags.emplace_back("--fs_data_dirs_reserved_bytes=1");
- NO_FATALS(StartCluster(ts_flags, {}, /* num_tablet_servers= */ 1, /* num_data_dirs= */ 2));
+ ExternalMiniClusterOptions opts;
+ opts.extra_tserver_flags = std::move(ts_flags);
+ opts.num_data_dirs = 2;
+ NO_FATALS(StartClusterWithOpts(opts));
ASSERT_OK(cluster_->SetFlag(cluster_->tablet_server(0),
"disk_reserved_override_prefix_1_path_for_testing",
http://git-wip-us.apache.org/repos/asf/kudu/blob/6523a773/src/kudu/integration-tests/external_mini_cluster-itest-base.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/external_mini_cluster-itest-base.cc b/src/kudu/integration-tests/external_mini_cluster-itest-base.cc
index 9fa12ff..db7d83b 100644
--- a/src/kudu/integration-tests/external_mini_cluster-itest-base.cc
+++ b/src/kudu/integration-tests/external_mini_cluster-itest-base.cc
@@ -41,6 +41,8 @@ namespace kudu {
using cluster::ExternalMiniCluster;
using cluster::ExternalMiniClusterOptions;
+using std::string;
+using std::vector;
void ExternalMiniClusterITestBase::TearDown() {
StopCluster();
@@ -48,15 +50,13 @@ void ExternalMiniClusterITestBase::TearDown() {
}
void ExternalMiniClusterITestBase::StartCluster(
- const std::vector<std::string>& extra_ts_flags,
- const std::vector<std::string>& extra_master_flags,
- int num_tablet_servers,
- int num_data_dirs) {
+ vector<string> extra_ts_flags,
+ vector<string> extra_master_flags,
+ int num_tablet_servers) {
ExternalMiniClusterOptions opts;
- opts.num_data_dirs = num_data_dirs;
opts.num_tablet_servers = num_tablet_servers;
- opts.extra_master_flags = extra_master_flags;
- opts.extra_tserver_flags = extra_ts_flags;
+ opts.extra_master_flags = std::move(extra_master_flags);
+ opts.extra_tserver_flags = std::move(extra_ts_flags);
StartClusterWithOpts(std::move(opts));
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/6523a773/src/kudu/integration-tests/external_mini_cluster-itest-base.h
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/external_mini_cluster-itest-base.h b/src/kudu/integration-tests/external_mini_cluster-itest-base.h
index 963930e..eba956c 100644
--- a/src/kudu/integration-tests/external_mini_cluster-itest-base.h
+++ b/src/kudu/integration-tests/external_mini_cluster-itest-base.h
@@ -44,10 +44,9 @@ class ExternalMiniClusterITestBase : public KuduTest {
void TearDown() override;
protected:
- void StartCluster(const std::vector<std::string>& extra_ts_flags = {},
- const std::vector<std::string>& extra_master_flags = {},
- int num_tablet_servers = 3,
- int num_data_dirs = 1);
+ void StartCluster(std::vector<std::string> extra_ts_flags = {},
+ std::vector<std::string> extra_master_flags = {},
+ int num_tablet_servers = 3);
void StartClusterWithOpts(cluster::ExternalMiniClusterOptions opts);
http://git-wip-us.apache.org/repos/asf/kudu/blob/6523a773/src/kudu/integration-tests/log_verifier.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/log_verifier.cc b/src/kudu/integration-tests/log_verifier.cc
index 58f960e..fdd321e 100644
--- a/src/kudu/integration-tests/log_verifier.cc
+++ b/src/kudu/integration-tests/log_verifier.cc
@@ -75,6 +75,8 @@ Status LogVerifier::OpenFsManager(ExternalTabletServer* ets,
fs_opts.read_only = true;
fs_opts.wal_path = ets->wal_dir();
fs_opts.data_paths = ets->data_dirs();
+ fs_opts.block_manager_type = cluster_->block_manager_type();
+
unique_ptr<FsManager> ret(new FsManager(Env::Default(), fs_opts));
RETURN_NOT_OK_PREPEND(ret->Open(),
Substitute("Couldn't initialize FS Manager for $0", ets->wal_dir()));
http://git-wip-us.apache.org/repos/asf/kudu/blob/6523a773/src/kudu/integration-tests/multidir_cluster-itest.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/multidir_cluster-itest.cc b/src/kudu/integration-tests/multidir_cluster-itest.cc
index 41cbd3c..b9d6cb4 100644
--- a/src/kudu/integration-tests/multidir_cluster-itest.cc
+++ b/src/kudu/integration-tests/multidir_cluster-itest.cc
@@ -19,6 +19,7 @@
#include <map>
#include <memory>
#include <string>
+#include <utility>
#include <vector>
#include <gtest/gtest.h>
@@ -35,6 +36,7 @@
namespace kudu {
+using cluster::ExternalMiniClusterOptions;
using cluster::ExternalTabletServer;
using std::map;
using std::string;
@@ -53,7 +55,11 @@ TEST_F(MultiDirClusterITest, TestBasicMultiDirCluster) {
"--fs_target_data_dirs_per_tablet=0"
};
- NO_FATALS(StartCluster(ts_flags, {}, /* num_tablet_servers= */ 1, kNumDataDirs));
+ ExternalMiniClusterOptions opts;
+ opts.extra_tserver_flags = std::move(ts_flags);
+ opts.num_tablet_servers = 1;
+ opts.num_data_dirs = kNumDataDirs;
+ NO_FATALS(StartClusterWithOpts(opts));
ExternalTabletServer* ts = cluster_->tablet_server(0);
TestWorkload work(cluster_.get());
work.set_num_replicas(1);
http://git-wip-us.apache.org/repos/asf/kudu/blob/6523a773/src/kudu/mini-cluster/external_mini_cluster.cc
----------------------------------------------------------------------
diff --git a/src/kudu/mini-cluster/external_mini_cluster.cc b/src/kudu/mini-cluster/external_mini_cluster.cc
index 5284275..7609376 100644
--- a/src/kudu/mini-cluster/external_mini_cluster.cc
+++ b/src/kudu/mini-cluster/external_mini_cluster.cc
@@ -28,6 +28,7 @@
#include <utility>
#include <gflags/gflags.h>
+#include <gflags/gflags_declare.h>
#include <gtest/gtest.h>
#include <rapidjson/document.h>
@@ -88,6 +89,8 @@ using strings::Substitute;
typedef ListTabletsResponsePB::StatusAndSchemaPB StatusAndSchemaPB;
+DECLARE_string(block_manager);
+
DEFINE_bool(perf_record, false,
"Whether to run \"perf record --call-graph fp\" on each daemon in the cluster");
@@ -129,15 +132,17 @@ Status ExternalMiniCluster::DeduceBinRoot(std::string* ret) {
}
Status ExternalMiniCluster::HandleOptions() {
- daemon_bin_path_ = opts_.daemon_bin_path;
- if (daemon_bin_path_.empty()) {
- RETURN_NOT_OK(DeduceBinRoot(&daemon_bin_path_));
+ if (opts_.daemon_bin_path.empty()) {
+ RETURN_NOT_OK(DeduceBinRoot(&opts_.daemon_bin_path));
}
- data_root_ = opts_.data_root;
- if (data_root_.empty()) {
+ if (opts_.data_root.empty()) {
// If they don't specify a data root, use the current gtest directory.
- data_root_ = JoinPathSegments(GetTestDataDirectory(), "minicluster-data");
+ opts_.data_root = JoinPathSegments(GetTestDataDirectory(), "minicluster-data");
+ }
+
+ if (opts_.block_manager_type.empty()) {
+ opts_.block_manager_type = FLAGS_block_manager;
}
return Status::OK();
@@ -156,9 +161,9 @@ Status ExternalMiniCluster::Start() {
.Build(&messenger_),
"Failed to start Messenger for minicluster");
- Status s = Env::Default()->CreateDir(data_root_);
+ Status s = Env::Default()->CreateDir(opts_.data_root);
if (!s.ok() && !s.IsAlreadyPresent()) {
- RETURN_NOT_OK_PREPEND(s, "Could not create root dir " + data_root_);
+ RETURN_NOT_OK_PREPEND(s, "Could not create root dir " + opts_.data_root);
}
if (opts_.enable_kerberos) {
@@ -235,7 +240,7 @@ Status ExternalMiniCluster::Restart() {
}
void ExternalMiniCluster::SetDaemonBinPath(string daemon_bin_path) {
- daemon_bin_path_ = std::move(daemon_bin_path);
+ opts_.daemon_bin_path = std::move(daemon_bin_path);
for (auto& master : masters_) {
master->SetExePath(GetBinaryPath(kMasterBinaryName));
}
@@ -245,18 +250,18 @@ void ExternalMiniCluster::SetDaemonBinPath(string daemon_bin_path) {
}
string ExternalMiniCluster::GetBinaryPath(const string& binary) const {
- CHECK(!daemon_bin_path_.empty());
- return JoinPathSegments(daemon_bin_path_, binary);
+ CHECK(!opts_.daemon_bin_path.empty());
+ return JoinPathSegments(opts_.daemon_bin_path, binary);
}
string ExternalMiniCluster::GetLogPath(const string& daemon_id) const {
- CHECK(!data_root_.empty());
- return JoinPathSegments(JoinPathSegments(data_root_, daemon_id), "logs");
+ CHECK(!opts_.data_root.empty());
+ return JoinPathSegments(JoinPathSegments(opts_.data_root, daemon_id), "logs");
}
string ExternalMiniCluster::GetDataPath(const string& daemon_id,
boost::optional<uint32_t> dir_index) const {
- CHECK(!data_root_.empty());
+ CHECK(!opts_.data_root.empty());
string data_path = "data";
if (dir_index) {
CHECK_LT(*dir_index, opts_.num_data_dirs);
@@ -264,7 +269,7 @@ string ExternalMiniCluster::GetDataPath(const string& daemon_id,
} else {
CHECK_EQ(1, opts_.num_data_dirs);
}
- return JoinPathSegments(JoinPathSegments(data_root_, daemon_id), data_path);
+ return JoinPathSegments(JoinPathSegments(opts_.data_root, daemon_id), data_path);
}
vector<string> ExternalMiniCluster::GetDataPaths(const string& daemon_id) const {
@@ -279,8 +284,8 @@ vector<string> ExternalMiniCluster::GetDataPaths(const string& daemon_id) const
}
string ExternalMiniCluster::GetWalPath(const string& daemon_id) const {
- CHECK(!data_root_.empty());
- return JoinPathSegments(JoinPathSegments(data_root_, daemon_id), "wal");
+ CHECK(!opts_.data_root.empty());
+ return JoinPathSegments(JoinPathSegments(opts_.data_root, daemon_id), "wal");
}
namespace {
@@ -305,6 +310,7 @@ Status ExternalMiniCluster::StartSingleMaster() {
opts.wal_dir = GetWalPath(daemon_id);
opts.data_dirs = GetDataPaths(daemon_id);
opts.log_dir = GetLogPath(daemon_id);
+ opts.block_manager_type = opts_.block_manager_type;
if (FLAGS_perf_record) {
opts.perf_record_filename =
Substitute("$0/perf-$1.data", opts.log_dir, daemon_id);
@@ -348,6 +354,7 @@ Status ExternalMiniCluster::StartDistributedMasters() {
ExternalDaemonOptions opts;
opts.messenger = messenger_;
+ opts.block_manager_type = opts_.block_manager_type;
opts.exe = exe;
opts.wal_dir = GetWalPath(daemon_id);
opts.data_dirs = GetDataPaths(daemon_id);
@@ -397,6 +404,7 @@ Status ExternalMiniCluster::AddTabletServer() {
ExternalDaemonOptions opts;
opts.messenger = messenger_;
+ opts.block_manager_type = opts_.block_manager_type;
opts.exe = GetBinaryPath(kTabletServerBinaryName);
opts.wal_dir = GetWalPath(daemon_id);
opts.data_dirs = GetDataPaths(daemon_id);
@@ -663,17 +671,8 @@ Status ExternalMiniCluster::SetFlag(ExternalDaemon* daemon,
//------------------------------------------------------------
ExternalDaemon::ExternalDaemon(ExternalDaemonOptions opts)
- : messenger_(std::move(opts.messenger)),
- wal_dir_(std::move(opts.wal_dir)),
- data_dirs_(std::move(opts.data_dirs)),
- log_dir_(std::move(opts.log_dir)),
- perf_record_filename_(std::move(opts.perf_record_filename)),
- start_process_timeout_(opts.start_process_timeout),
- logtostderr_(opts.logtostderr),
- rpc_bind_address_(std::move(opts.rpc_bind_address)),
- exe_(std::move(opts.exe)),
- extra_flags_(std::move(opts.extra_flags)) {
- CHECK(rpc_bind_address_.Initialized());
+ : opts_(std::move(opts)) {
+ CHECK(rpc_bind_address().Initialized());
}
ExternalDaemon::~ExternalDaemon() {
@@ -685,11 +684,11 @@ Status ExternalDaemon::EnableKerberos(MiniKdc* kdc, const string& bind_host) {
RETURN_NOT_OK_PREPEND(kdc->CreateServiceKeytab(spn, &ktpath),
"could not create keytab");
extra_env_ = kdc->GetEnvVars();
- extra_flags_.push_back(Substitute("--keytab_file=$0", ktpath));
- extra_flags_.push_back(Substitute("--principal=$0", spn));
- extra_flags_.emplace_back("--rpc_authentication=required");
- extra_flags_.emplace_back("--superuser_acl=test-admin");
- extra_flags_.emplace_back("--user_acl=test-user");
+ opts_.extra_flags.push_back(Substitute("--keytab_file=$0", ktpath));
+ opts_.extra_flags.push_back(Substitute("--principal=$0", spn));
+ opts_.extra_flags.emplace_back("--rpc_authentication=required");
+ opts_.extra_flags.emplace_back("--superuser_acl=test-admin");
+ opts_.extra_flags.emplace_back("--user_acl=test-user");
return Status::OK();
}
@@ -699,7 +698,7 @@ Status ExternalDaemon::StartProcess(const vector<string>& user_flags) {
vector<string> argv;
// First the exe for argv[0].
- argv.push_back(exe_);
+ argv.push_back(opts_.exe);
// Then all the flags coming from the minicluster framework.
argv.insert(argv.end(), user_flags.begin(), user_flags.end());
@@ -728,7 +727,7 @@ Status ExternalDaemon::StartProcess(const vector<string>& user_flags) {
// Enable metrics logging.
argv.emplace_back("--metrics_log_interval_ms=1000");
- if (logtostderr_) {
+ if (opts_.logtostderr) {
// Ensure that logging goes to the test output and doesn't get buffered.
argv.emplace_back("--logtostderr");
argv.emplace_back("--logbuflevel=-1");
@@ -736,11 +735,11 @@ Status ExternalDaemon::StartProcess(const vector<string>& user_flags) {
// Even if we are logging to stderr, metrics logs and minidumps end up being
// written based on -log_dir. So, we have to set that too.
- argv.push_back("--log_dir=" + log_dir_);
- RETURN_NOT_OK(env_util::CreateDirsRecursively(Env::Default(), log_dir_));
+ argv.push_back("--log_dir=" + log_dir());
+ RETURN_NOT_OK(env_util::CreateDirsRecursively(Env::Default(), log_dir()));
// Tell the server to dump its port information so we can pick it up.
- string info_path = JoinPathSegments(data_dirs_[0], "info.pb");
+ string info_path = JoinPathSegments(data_dirs()[0], "info.pb");
argv.push_back("--server_dump_info_path=" + info_path);
argv.emplace_back("--server_dump_info_format=pb");
@@ -756,7 +755,7 @@ Status ExternalDaemon::StartProcess(const vector<string>& user_flags) {
// Then the "extra flags" passed into the ctor (from the ExternalMiniCluster
// options struct). These come at the end so they can override things like
// web port or RPC bind address if necessary.
- argv.insert(argv.end(), extra_flags_.begin(), extra_flags_.end());
+ argv.insert(argv.end(), opts_.extra_flags.begin(), opts_.extra_flags.end());
// A previous instance of the daemon may have run in the same directory. So, remove
// the previous info file if it's there.
@@ -767,21 +766,21 @@ Status ExternalDaemon::StartProcess(const vector<string>& user_flags) {
p->SetEnvVars(extra_env_);
string env_str;
JoinMapKeysAndValues(extra_env_, "=", ",", &env_str);
- LOG(INFO) << "Running " << exe_ << "\n" << JoinStrings(argv, "\n")
+ LOG(INFO) << "Running " << opts_.exe << "\n" << JoinStrings(argv, "\n")
<< " with env {" << env_str << "}";
RETURN_NOT_OK_PREPEND(p->Start(),
- Substitute("Failed to start subprocess $0", exe_));
+ Substitute("Failed to start subprocess $0", opts_.exe));
// If requested, start a monitoring subprocess.
unique_ptr<Subprocess> perf_record;
- if (!perf_record_filename_.empty()) {
+ if (!opts_.perf_record_filename.empty()) {
perf_record.reset(new Subprocess({
"perf",
"record",
"--call-graph",
"fp",
"-o",
- perf_record_filename_,
+ opts_.perf_record_filename,
Substitute("--pid=$0", p->pid())
}, SIGINT));
RETURN_NOT_OK_PREPEND(perf_record->Start(),
@@ -792,7 +791,7 @@ Status ExternalDaemon::StartProcess(const vector<string>& user_flags) {
Stopwatch sw;
sw.start();
bool success = false;
- while (sw.elapsed().wall_seconds() < start_process_timeout_.ToSeconds()) {
+ while (sw.elapsed().wall_seconds() < opts_.start_process_timeout.ToSeconds()) {
if (Env::Default()->FileExists(info_path)) {
success = true;
break;
@@ -813,7 +812,7 @@ Status ExternalDaemon::StartProcess(const vector<string>& user_flags) {
return Status::OK();
}
- RETURN_NOT_OK_PREPEND(s, Substitute("Failed waiting on $0", exe_));
+ RETURN_NOT_OK_PREPEND(s, Substitute("Failed waiting on $0", opts_.exe));
string exit_info;
RETURN_NOT_OK(p->GetExitStatus(nullptr, &exit_info));
return Status::RuntimeError(exit_info);
@@ -823,14 +822,14 @@ Status ExternalDaemon::StartProcess(const vector<string>& user_flags) {
ignore_result(p->Kill(SIGKILL));
return Status::TimedOut(
Substitute("Timed out after $0s waiting for process ($1) to write info file ($2)",
- start_process_timeout_.ToString(), exe_, info_path));
+ opts_.start_process_timeout.ToString(), opts_.exe, info_path));
}
status_.reset(new ServerStatusPB());
RETURN_NOT_OK_PREPEND(pb_util::ReadPBFromPath(Env::Default(), info_path, status_.get()),
"Failed to read info file from " + info_path);
- LOG(INFO) << "Started " << exe_ << " as pid " << p->pid();
- VLOG(1) << exe_ << " instance information:\n" << SecureDebugString(*status_);
+ LOG(INFO) << "Started " << opts_.exe << " as pid " << p->pid();
+ VLOG(1) << opts_.exe << " instance information:\n" << SecureDebugString(*status_);
process_.swap(p);
perf_record_process_.swap(perf_record);
@@ -839,15 +838,15 @@ Status ExternalDaemon::StartProcess(const vector<string>& user_flags) {
void ExternalDaemon::SetExePath(string exe) {
CHECK(IsShutdown()) << "Call Shutdown() before changing the executable path";
- exe_ = std::move(exe);
+ opts_.exe = std::move(exe);
}
Status ExternalDaemon::Pause() {
if (!process_) {
return Status::IllegalState(Substitute(
- "Request to pause '$0' but the process is not there", exe_));
+ "Request to pause '$0' but the process is not there", opts_.exe));
}
- VLOG(1) << "Pausing " << exe_ << " with pid " << process_->pid();
+ VLOG(1) << "Pausing " << opts_.exe << " with pid " << process_->pid();
const Status s = process_->Kill(SIGSTOP);
RETURN_NOT_OK(s);
paused_ = true;
@@ -857,9 +856,9 @@ Status ExternalDaemon::Pause() {
Status ExternalDaemon::Resume() {
if (!process_) {
return Status::IllegalState(Substitute(
- "Request to resume '$0' but the process is not there", exe_));
+ "Request to resume '$0' but the process is not there", opts_.exe));
}
- VLOG(1) << "Resuming " << exe_ << " with pid " << process_->pid();
+ VLOG(1) << "Resuming " << opts_.exe << " with pid " << process_->pid();
const Status s = process_->Kill(SIGCONT);
RETURN_NOT_OK(s);
paused_ = false;
@@ -959,10 +958,10 @@ void ExternalDaemon::Shutdown() {
CheckForLeaks();
}
- LOG(INFO) << "Killing " << exe_ << " with pid " << process_->pid();
+ LOG(INFO) << "Killing " << opts_.exe << " with pid " << process_->pid();
ignore_result(process_->Kill(SIGKILL));
}
- WARN_NOT_OK(process_->Wait(), "Waiting on " + exe_);
+ WARN_NOT_OK(process_->Wait(), "Waiting on " + opts_.exe);
paused_ = false;
process_.reset();
perf_record_process_.reset();
@@ -980,9 +979,9 @@ void ExternalDaemon::FlushCoverage() {
#ifndef COVERAGE_BUILD
return; // NOLINT(*)
#else
- LOG(INFO) << "Attempting to flush coverage for " << exe_ << " pid " << process_->pid();
+ LOG(INFO) << "Attempting to flush coverage for " << opts_.exe << " pid " << process_->pid();
server::GenericServiceProxy proxy(
- messenger_, bound_rpc_addr(), bound_rpc_addr().host());
+ opts_.messenger, bound_rpc_addr(), bound_rpc_addr().host());
server::FlushCoverageRequestPB req;
server::FlushCoverageResponsePB resp;
@@ -993,15 +992,15 @@ void ExternalDaemon::FlushCoverage() {
if (s.ok() && !resp.success()) {
s = Status::RemoteError("Server does not appear to be running a coverage build");
}
- WARN_NOT_OK(s, Substitute("Unable to flush coverage on $0 pid $1", exe_, process_->pid()));
+ WARN_NOT_OK(s, Substitute("Unable to flush coverage on $0 pid $1", opts_.exe, process_->pid()));
#endif
}
void ExternalDaemon::CheckForLeaks() {
#if defined(__has_feature)
# if __has_feature(address_sanitizer)
- LOG(INFO) << "Attempting to check leaks for " << exe_ << " pid " << process_->pid();
- server::GenericServiceProxy proxy(messenger_, bound_rpc_addr(), bound_rpc_addr().host());
+ LOG(INFO) << "Attempting to check leaks for " << opts_.exe << " pid " << process_->pid();
+ server::GenericServiceProxy proxy(opts_.messenger, bound_rpc_addr(), bound_rpc_addr().host());
server::CheckLeaksRequestPB req;
server::CheckLeaksResponsePB resp;
@@ -1013,10 +1012,10 @@ void ExternalDaemon::CheckForLeaks() {
if (!resp.success()) {
s = Status::RemoteError("Server does not appear to be running an LSAN build");
} else {
- CHECK(!resp.found_leaks()) << "Found leaks in " << exe_ << " pid " << process_->pid();
+ CHECK(!resp.found_leaks()) << "Found leaks in " << opts_.exe << " pid " << process_->pid();
}
}
- WARN_NOT_OK(s, Substitute("Unable to check leaks on $0 pid $1", exe_, process_->pid()));
+ WARN_NOT_OK(s, Substitute("Unable to check leaks on $0 pid $1", opts_.exe, process_->pid()));
# endif
#endif
}
@@ -1170,7 +1169,7 @@ Status ExternalMaster::Restart() {
Status ExternalMaster::WaitForCatalogManager() {
unique_ptr<MasterServiceProxy> proxy(new MasterServiceProxy(
- messenger_, bound_rpc_addr(), bound_rpc_addr().host()));
+ opts_.messenger, bound_rpc_addr(), bound_rpc_addr().host()));
Stopwatch sw;
sw.start();
while (sw.elapsed().wall_seconds() < kMasterCatalogManagerTimeoutSeconds) {
@@ -1212,8 +1211,9 @@ Status ExternalMaster::WaitForCatalogManager() {
vector<string> ExternalMaster::GetCommonFlags() const {
return {
- "--fs_wal_dir=" + wal_dir_,
- "--fs_data_dirs=" + JoinStrings(data_dirs_, ","),
+ "--fs_wal_dir=" + wal_dir(),
+ "--fs_data_dirs=" + JoinStrings(data_dirs(), ","),
+ "--block_manager=" + opts_.block_manager_type,
"--webserver_interface=localhost",
// See the in-line comment for "--ipki_server_key_size" flag in
@@ -1243,8 +1243,9 @@ ExternalTabletServer::~ExternalTabletServer() {
Status ExternalTabletServer::Start() {
vector<string> flags;
- flags.push_back("--fs_wal_dir=" + wal_dir_);
- flags.push_back("--fs_data_dirs=" + JoinStrings(data_dirs_, ","));
+ flags.push_back("--fs_wal_dir=" + wal_dir());
+ flags.push_back("--fs_data_dirs=" + JoinStrings(data_dirs(), ","));
+ flags.push_back("--block_manager=" + opts_.block_manager_type);
flags.push_back(Substitute("--rpc_bind_addresses=$0",
rpc_bind_address().ToString()));
flags.push_back(Substitute("--local_ip_for_outbound_sockets=$0",
@@ -1264,8 +1265,9 @@ Status ExternalTabletServer::Restart() {
return Status::IllegalState("Tablet server cannot be restarted. Must call Shutdown() first.");
}
vector<string> flags;
- flags.push_back("--fs_wal_dir=" + wal_dir_);
- flags.push_back("--fs_data_dirs=" + JoinStrings(data_dirs_, ","));
+ flags.push_back("--fs_wal_dir=" + wal_dir());
+ flags.push_back("--fs_data_dirs=" + JoinStrings(data_dirs(), ","));
+ flags.push_back("--block_manager=" + opts_.block_manager_type);
flags.push_back(Substitute("--rpc_bind_addresses=$0", bound_rpc_.ToString()));
flags.push_back(Substitute("--local_ip_for_outbound_sockets=$0",
rpc_bind_address().host()));
http://git-wip-us.apache.org/repos/asf/kudu/blob/6523a773/src/kudu/mini-cluster/external_mini_cluster.h
----------------------------------------------------------------------
diff --git a/src/kudu/mini-cluster/external_mini_cluster.h b/src/kudu/mini-cluster/external_mini_cluster.h
index 29c209e..8d5e9c3 100644
--- a/src/kudu/mini-cluster/external_mini_cluster.h
+++ b/src/kudu/mini-cluster/external_mini_cluster.h
@@ -90,6 +90,11 @@ struct ExternalMiniClusterOptions {
// Default: "", which auto-generates a unique path for this cluster.
std::string data_root;
+ // Block manager type. Must be either "file" or "log".
+ //
+ // Default: "", which uses the current value of FLAGS_block_manager.
+ std::string block_manager_type;
+
MiniCluster::BindMode bind_mode;
// The path where the kudu daemons should be run from.
@@ -257,6 +262,10 @@ class ExternalMiniCluster : public MiniCluster {
std::shared_ptr<master::MasterServiceProxy> master_proxy() const override;
std::shared_ptr<master::MasterServiceProxy> master_proxy(int idx) const override;
+ std::string block_manager_type() const {
+ return opts_.block_manager_type;
+ }
+
// Wait until the number of registered tablet servers reaches the given count
// on all of the running masters. Returns Status::TimedOut if the desired
// count is not achieved with the given timeout.
@@ -331,12 +340,7 @@ class ExternalMiniCluster : public MiniCluster {
Status DeduceBinRoot(std::string* ret);
Status HandleOptions();
- const ExternalMiniClusterOptions opts_;
-
- // The root for binaries.
- std::string daemon_bin_path_;
-
- std::string data_root_;
+ ExternalMiniClusterOptions opts_;
std::vector<scoped_refptr<ExternalMaster> > masters_;
std::vector<scoped_refptr<ExternalTabletServer> > tablet_servers_;
@@ -354,6 +358,7 @@ struct ExternalDaemonOptions {
bool logtostderr;
std::shared_ptr<rpc::Messenger> messenger;
+ std::string block_manager_type;
std::string exe;
HostPort rpc_bind_address;
std::string wal_dir;
@@ -435,21 +440,21 @@ class ExternalDaemon : public RefCountedThreadSafe<ExternalDaemon> {
// Delete files specified by 'wal_dir_' and 'data_dirs_'.
Status DeleteFromDisk() const WARN_UNUSED_RESULT;
- const std::string& wal_dir() const { return wal_dir_; }
+ const std::string& wal_dir() const { return opts_.wal_dir; }
const std::string& data_dir() const {
- CHECK_EQ(1, data_dirs_.size());
- return data_dirs_[0];
+ CHECK_EQ(1, opts_.data_dirs.size());
+ return opts_.data_dirs[0];
}
- const std::vector<std::string>& data_dirs() const { return data_dirs_; }
+ const std::vector<std::string>& data_dirs() const { return opts_.data_dirs; }
// Returns the log dir of the external daemon.
- const std::string& log_dir() const { return log_dir_; }
+ const std::string& log_dir() const { return opts_.log_dir; }
// Return a pointer to the flags used for this server on restart.
// Modifying these flags will only take effect on the next restart.
- std::vector<std::string>* mutable_flags() { return &extra_flags_; }
+ std::vector<std::string>* mutable_flags() { return &opts_.extra_flags; }
// Retrieve the value of a given metric from this server. The metric must
// be of int64_t type.
@@ -494,19 +499,10 @@ class ExternalDaemon : public RefCountedThreadSafe<ExternalDaemon> {
// Get RPC bind address for daemon.
const HostPort& rpc_bind_address() const {
- return rpc_bind_address_;
+ return opts_.rpc_bind_address;
}
- const std::shared_ptr<rpc::Messenger> messenger_;
- const std::string wal_dir_;
- std::vector<std::string> data_dirs_;
- const std::string log_dir_;
- const std::string perf_record_filename_;
- const MonoDelta start_process_timeout_;
- const bool logtostderr_;
- const HostPort rpc_bind_address_;
- std::string exe_;
- std::vector<std::string> extra_flags_;
+ ExternalDaemonOptions opts_;
std::map<std::string, std::string> extra_env_;
std::unique_ptr<Subprocess> process_;
http://git-wip-us.apache.org/repos/asf/kudu/blob/6523a773/src/kudu/server/server_base.cc
----------------------------------------------------------------------
diff --git a/src/kudu/server/server_base.cc b/src/kudu/server/server_base.cc
index 556cdca..e17358f 100644
--- a/src/kudu/server/server_base.cc
+++ b/src/kudu/server/server_base.cc
@@ -156,6 +156,7 @@ ServerBase::ServerBase(string name, const ServerBaseOptions& options,
FsManagerOpts fs_opts;
fs_opts.metric_entity = metric_entity_;
fs_opts.parent_mem_tracker = mem_tracker_;
+ fs_opts.block_manager_type = options.fs_opts.block_manager_type;
fs_opts.wal_path = options.fs_opts.wal_path;
fs_opts.data_paths = options.fs_opts.data_paths;
fs_manager_.reset(new FsManager(options.env, fs_opts));