You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by gr...@apache.org on 2020/09/03 19:46:16 UTC
[kudu] branch master updated: KUDU-2574: Add a unique cluster ID
This is an automated email from the ASF dual-hosted git repository.
granthenke pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
The following commit(s) were added to refs/heads/master by this push:
new 6689ba3 KUDU-2574: Add a unique cluster ID
6689ba3 is described below
commit 6689ba3c81be193309afd27de016dd35c0db71de
Author: Grant Henke <gr...@apache.org>
AuthorDate: Tue Sep 1 15:00:43 2020 -0500
KUDU-2574: Add a unique cluster ID
This patch adds a unique cluster ID to the cluster. The ID is
a UUID that is automatically generated and stored in the
sys_catalog if missing (on fresh startup or upgrade).
This patch also exposes the cluster ID via the masters web-ui
and the `kudu master list` tool.
I plan to use this cluster ID to enhance the HMS sync
functionality, but it can be exposed further as
needed for other uses as well.
Change-Id: I4df1d3f7d100336f52563f3008cacf6d9e328fae
Reviewed-on: http://gerrit.cloudera.org:8080/16403
Reviewed-by: Alexey Serbin <as...@cloudera.com>
Reviewed-by: Andrew Wong <aw...@cloudera.com>
Tested-by: Kudu Jenkins
---
src/kudu/common/wire_protocol.proto | 4 ++
src/kudu/integration-tests/cluster_itest_util.cc | 13 ++++
src/kudu/integration-tests/cluster_itest_util.h | 5 ++
.../integration-tests/master_failover-itest.cc | 31 ++++++++
src/kudu/master/catalog_manager.cc | 82 ++++++++++++++++++++++
src/kudu/master/catalog_manager.h | 15 ++++
src/kudu/master/master.cc | 4 ++
src/kudu/master/master.h | 8 +++
src/kudu/master/master.proto | 9 +++
src/kudu/master/master_path_handlers.cc | 1 +
src/kudu/master/master_service.cc | 1 +
src/kudu/master/sys_catalog-test.cc | 30 ++++++++
src/kudu/master/sys_catalog.cc | 49 ++++++++++++-
src/kudu/master/sys_catalog.h | 13 ++++
src/kudu/tools/tool_action_master.cc | 6 +-
www/masters.mustache | 2 +
16 files changed, 271 insertions(+), 2 deletions(-)
diff --git a/src/kudu/common/wire_protocol.proto b/src/kudu/common/wire_protocol.proto
index 3dce5a0..59682ba 100644
--- a/src/kudu/common/wire_protocol.proto
+++ b/src/kudu/common/wire_protocol.proto
@@ -118,6 +118,10 @@ message ServerEntryPB {
// If an error has occured earlier in the RPC call, the role
// may be not be set.
optional consensus.RaftPeerPB.Role role = 4;
+
+ // The unique cluster ID of the cluster this server belongs too.
+ // TODO(ghenke): Currently only set for masters.
+ optional string cluster_id = 5;
}
// A row block in which each row is stored contiguously.
diff --git a/src/kudu/integration-tests/cluster_itest_util.cc b/src/kudu/integration-tests/cluster_itest_util.cc
index 26d27c1..75a411d 100644
--- a/src/kudu/integration-tests/cluster_itest_util.cc
+++ b/src/kudu/integration-tests/cluster_itest_util.cc
@@ -1316,6 +1316,19 @@ Status GetInt64Metric(const HostPort& http_hp,
return Status::NotFound(msg);
}
+Status GetMasterRegistration(const shared_ptr<MasterServiceProxy>& master_proxy,
+ const MonoDelta& timeout,
+ master::GetMasterRegistrationResponsePB* registration) {
+ master::GetMasterRegistrationRequestPB req;
+ RpcController rpc;
+ rpc.set_timeout(timeout);
+ RETURN_NOT_OK(master_proxy->GetMasterRegistration(req, registration, &rpc));
+ if (registration->has_error()) {
+ return StatusFromPB(registration->error().status());
+ }
+ return Status::OK();
+}
+
Status GetTsCounterValue(ExternalTabletServer* ets,
MetricPrototype* metric,
int64_t* value) {
diff --git a/src/kudu/integration-tests/cluster_itest_util.h b/src/kudu/integration-tests/cluster_itest_util.h
index 404af90..dc89e1c 100644
--- a/src/kudu/integration-tests/cluster_itest_util.h
+++ b/src/kudu/integration-tests/cluster_itest_util.h
@@ -455,6 +455,11 @@ Status GetTsCounterValue(cluster::ExternalTabletServer* ets,
MetricPrototype* metric,
int64_t* value);
+// Get the registration from the Master.
+Status GetMasterRegistration(const std::shared_ptr<master::MasterServiceProxy>& master_proxy,
+ const MonoDelta& timeout,
+ master::GetMasterRegistrationResponsePB* registration);
+
// Alter the table name.
Status AlterTableName(const std::shared_ptr<master::MasterServiceProxy>& master_proxy,
const std::string& table_id,
diff --git a/src/kudu/integration-tests/master_failover-itest.cc b/src/kudu/integration-tests/master_failover-itest.cc
index 467079c..04d581c 100644
--- a/src/kudu/integration-tests/master_failover-itest.cc
+++ b/src/kudu/integration-tests/master_failover-itest.cc
@@ -36,6 +36,7 @@
#include "kudu/gutil/strings/strip.h" // IWYU pragma: keep
#include "kudu/gutil/strings/substitute.h"
#include "kudu/integration-tests/cluster_itest_util.h"
+#include "kudu/master/master.pb.h"
#include "kudu/master/sys_catalog.h" // IWYU pragma: keep
#include "kudu/mini-cluster/external_mini_cluster.h"
#include "kudu/util/metrics.h"
@@ -57,6 +58,7 @@ using kudu::cluster::ExternalMiniCluster;
using kudu::cluster::ExternalMiniClusterOptions;
using kudu::cluster::ScopedResumeExternalDaemon;
using kudu::itest::GetInt64Metric;
+using kudu::itest::GetMasterRegistration;
using std::set;
using std::string;
using std::unique_ptr;
@@ -72,6 +74,8 @@ namespace client {
const int kNumTabletServerReplicas = 3;
+const MonoDelta kOperationTimeout = MonoDelta::FromSeconds(30);
+
// Parameterized based on HmsMode.
class MasterFailoverTest : public KuduTest,
public ::testing::WithParamInterface<HmsMode> {
@@ -153,6 +157,14 @@ class MasterFailoverTest : public KuduTest,
->Alter();
}
+ std::string GetClusterId(const int master_idx) {
+ master::GetMasterRegistrationResponsePB registration;
+ CHECK_OK(GetMasterRegistration(cluster_->master_proxy(master_idx),
+ kOperationTimeout, ®istration));
+ CHECK(registration.has_cluster_id());
+ return registration.cluster_id();
+ }
+
protected:
ExternalMiniClusterOptions opts_;
unique_ptr<ExternalMiniCluster> cluster_;
@@ -487,5 +499,24 @@ TEST_P(MasterFailoverTest, TestMasterPermanentFailure) {
}
}
+TEST_P(MasterFailoverTest, TestClusterIdOnFailover) {
+ // Validate and store the initial cluster ID.
+ int leader_idx;
+ ASSERT_OK(cluster_->GetLeaderMasterIndex(&leader_idx));
+ string original_cluster_id = GetClusterId(leader_idx);
+ ASSERT_TRUE(!original_cluster_id.empty());
+
+ LOG(INFO) << "Shutdown the leader master";
+ cluster_->master(leader_idx)->Shutdown();
+
+ // Validate the cluster ID of the new leader matches.
+ int new_leader_idx;
+ ASSERT_OK(cluster_->GetLeaderMasterIndex(&new_leader_idx));
+ ASSERT_NE(leader_idx, new_leader_idx);
+ string new_cluster_id = GetClusterId(new_leader_idx);
+ ASSERT_TRUE(!new_cluster_id.empty());
+ ASSERT_EQ(original_cluster_id, new_cluster_id);
+}
+
} // namespace client
} // namespace kudu
diff --git a/src/kudu/master/catalog_manager.cc b/src/kudu/master/catalog_manager.cc
index 9194a09..d834af2 100644
--- a/src/kudu/master/catalog_manager.cc
+++ b/src/kudu/master/catalog_manager.cc
@@ -886,6 +886,34 @@ Status CatalogManager::WaitUntilCaughtUpAsLeader(const MonoDelta& timeout) {
return Status::OK();
}
+Status CatalogManager::InitClusterId() {
+ leader_lock_.AssertAcquiredForWriting();
+
+ string cluster_id;
+ Status s = LoadClusterId(&cluster_id);
+ if (s.IsNotFound()) {
+ // Status::NotFound is returned if no cluster ID record is
+ // found in the system catalog table. It can happen on the very first run
+ // of a Kudu cluster or on upgrade from an older version that did not have
+ // cluster IDs. If so, it's necessary to create and persist
+ // a new cluster ID record which, if persisted, will be used for this and next runs.
+
+ // Generate new cluster ID.
+ cluster_id = GenerateId();
+
+ // If the leadership was lost, writing into the system table fails.
+ s = StoreClusterId(cluster_id);
+ }
+
+ // Once the cluster ID is loaded or stored, store it in a variable for
+ // fast lookup.
+ if (s.ok()) {
+ master_->set_cluster_id(cluster_id);
+ }
+
+ return s;
+}
+
Status CatalogManager::InitCertAuthority() {
leader_lock_.AssertAcquiredForWriting();
@@ -1005,6 +1033,17 @@ Status CatalogManager::InitCertAuthorityWith(
return Status::OK();
}
+Status CatalogManager::LoadClusterId(string* cluster_id) {
+ leader_lock_.AssertAcquired();
+
+ SysClusterIdEntryPB entry;
+ RETURN_NOT_OK(sys_catalog_->GetClusterIdEntry(&entry));
+ *cluster_id = entry.cluster_id();
+ LOG(INFO) << "Loaded cluster ID: " << *cluster_id;
+
+ return Status::OK();
+}
+
Status CatalogManager::LoadCertAuthorityInfo(unique_ptr<PrivateKey>* key,
unique_ptr<Cert>* cert) {
leader_lock_.AssertAcquired();
@@ -1029,6 +1068,18 @@ Status CatalogManager::LoadCertAuthorityInfo(unique_ptr<PrivateKey>* key,
return Status::OK();
}
+// Store cluster ID into the system table.
+Status CatalogManager::StoreClusterId(const string& cluster_id) {
+ leader_lock_.AssertAcquiredForWriting();
+
+ SysClusterIdEntryPB entry;
+ entry.set_cluster_id(cluster_id);
+ RETURN_NOT_OK(sys_catalog_->AddClusterIdEntry(entry));
+ LOG(INFO) << "Generated new cluster ID: " << cluster_id;
+
+ return Status::OK();
+}
+
// Store internal Kudu CA cert authority information into the system table.
Status CatalogManager::StoreCertAuthorityInfo(const PrivateKey& key,
const Cert& cert) {
@@ -1151,6 +1202,15 @@ void CatalogManager::PrepareForLeadershipTask() {
}
}
+ static const char* const kClustIdInitOpDescription = "Initializing Kudu cluster ID";
+ LOG(INFO) << kClustIdInitOpDescription << "...";
+ LOG_SLOW_EXECUTION(WARNING, 1000, LogPrefix() + kClustIdInitOpDescription) {
+ if (!check([this]() { return this->InitClusterId(); },
+ *consensus, term, kClustIdInitOpDescription).ok()) {
+ return;
+ }
+ }
+
// TODO(KUDU-1920): update this once "BYO PKI" feature is supported.
static const char* const kCaInitOpDescription =
"Initializing Kudu internal certificate authority";
@@ -1203,6 +1263,24 @@ void CatalogManager::PrepareForLeadershipTask() {
leader_ready_term_ = term;
}
+Status CatalogManager::PrepareFollowerClusterId() {
+ static const char* const kDescription =
+ "loading cluster ID for follower catalog manager";
+
+ // Load the cluster ID.
+ string cluster_id;
+ Status s = LoadClusterId(&cluster_id);
+ if (s.ok()) {
+ LOG_WITH_PREFIX(INFO) << kDescription << ": success";
+ // Once the cluster ID is loaded or stored, store it in a variable for
+ // fast lookup.
+ master_->set_cluster_id(cluster_id);
+ } else {
+ LOG_WITH_PREFIX(WARNING) << kDescription << ": " << s.ToString();
+ }
+ return s;
+}
+
Status CatalogManager::PrepareFollowerCaInfo() {
static const char* const kDescription =
"acquiring CA information for follower catalog manager";
@@ -1250,6 +1328,10 @@ Status CatalogManager::PrepareFollowerTokenVerifier() {
Status CatalogManager::PrepareFollower(MonoTime* last_tspk_run) {
leader_lock_.AssertAcquiredForReading();
+ // Load the cluster ID.
+ if (master_->cluster_id().empty()) {
+ RETURN_NOT_OK(PrepareFollowerClusterId());
+ }
// Load the CA certificate and CA private key.
if (!master_->tls_context().has_signed_cert()) {
RETURN_NOT_OK(PrepareFollowerCaInfo());
diff --git a/src/kudu/master/catalog_manager.h b/src/kudu/master/catalog_manager.h
index 0836b05..091aa70 100644
--- a/src/kudu/master/catalog_manager.h
+++ b/src/kudu/master/catalog_manager.h
@@ -872,6 +872,9 @@ class CatalogManager : public tserver::TabletReplicaLookupIf {
// Currently, it's about having a means to authenticate clients by authn tokens.
Status PrepareFollower(MonoTime* last_tspk_run);
+ // Load the cluster ID for the follower catalog manager.
+ Status PrepareFollowerClusterId();
+
// Prepare CA-related information for the follower catalog manager. Currently,
// this includes reading the CA information from the system table, creating
// TLS server certificate request, signing it with the CA key, and installing
@@ -897,6 +900,11 @@ class CatalogManager : public tserver::TabletReplicaLookupIf {
// This method is thread-safe.
Status InitSysCatalogAsync(bool is_first_run);
+ // Initialize the cluster ID: load the cluster ID record
+ // from the system table. If a cluster ID record is not present in the
+ // table, generate and store a new one.
+ Status InitClusterId();
+
// Initialize the IPKI certificate authority: load the CA information record
// from the system table. If the CA information record is not present in the
// table, generate and store a new one.
@@ -907,12 +915,19 @@ class CatalogManager : public tserver::TabletReplicaLookupIf {
Status InitCertAuthorityWith(std::unique_ptr<security::PrivateKey> key,
std::unique_ptr<security::Cert> cert);
+ // Load the cluster ID from the system table.
+ // If the cluster ID is not found in the table, return Status::NotFound.
+ Status LoadClusterId(std::string* cluster_id);
+
// Load the IPKI certficate authority information from the system
// table: the private key and the certificate. If the CA info entry is not
// found in the table, return Status::NotFound.
Status LoadCertAuthorityInfo(std::unique_ptr<security::PrivateKey>* key,
std::unique_ptr<security::Cert>* cert);
+ // Store cluster ID into the system table.
+ Status StoreClusterId(const std::string& cluster_id);
+
// Store the IPKI certificate authority information into the system table.
Status StoreCertAuthorityInfo(const security::PrivateKey& key,
const security::Cert& cert);
diff --git a/src/kudu/master/master.cc b/src/kudu/master/master.cc
index d0a7136..f86776b 100644
--- a/src/kudu/master/master.cc
+++ b/src/kudu/master/master.cc
@@ -320,6 +320,9 @@ Status GetMasterEntryForHost(const shared_ptr<rpc::Messenger>& messenger,
}
e->mutable_registration()->CopyFrom(resp.registration());
e->set_role(resp.role());
+ if (resp.has_cluster_id()) {
+ e->set_cluster_id(resp.cluster_id());
+ }
return Status::OK();
}
@@ -331,6 +334,7 @@ Status Master::ListMasters(vector<ServerEntryPB>* masters) const {
local_entry.mutable_instance_id()->CopyFrom(catalog_manager_->NodeInstance());
RETURN_NOT_OK(GetMasterRegistration(local_entry.mutable_registration()));
local_entry.set_role(RaftPeerPB::LEADER);
+ local_entry.set_cluster_id(cluster_id_);
masters->emplace_back(std::move(local_entry));
return Status::OK();
}
diff --git a/src/kudu/master/master.h b/src/kudu/master/master.h
index 28670fe..8b9d4e8 100644
--- a/src/kudu/master/master.h
+++ b/src/kudu/master/master.h
@@ -75,6 +75,8 @@ class Master : public kserver::KuduServer {
Status WaitUntilCatalogManagerIsLeaderAndReadyForTests(const MonoDelta& timeout)
WARN_UNUSED_RESULT;
+ const std::string& cluster_id() const { return cluster_id_; }
+
MasterCertAuthority* cert_authority() { return cert_authority_.get(); }
security::TokenSigner* token_signer() { return token_signer_.get(); }
@@ -123,6 +125,7 @@ class Master : public kserver::KuduServer {
private:
friend class MasterTest;
+ friend class CatalogManager;
void InitCatalogManagerTask();
Status InitCatalogManager();
@@ -136,6 +139,9 @@ class Master : public kserver::KuduServer {
// safe in a particular case.
void ShutdownImpl();
+ // Set the cluster ID on this master for fast lookup.
+ void set_cluster_id(const std::string& cluster_id) { cluster_id_ = cluster_id; }
+
enum MasterState {
kStopped,
kInitialized,
@@ -144,6 +150,8 @@ class Master : public kserver::KuduServer {
MasterState state_;
+ std::string cluster_id_ = "";
+
std::unique_ptr<MasterCertAuthority> cert_authority_;
std::unique_ptr<security::TokenSigner> token_signer_;
std::unique_ptr<CatalogManager> catalog_manager_;
diff --git a/src/kudu/master/master.proto b/src/kudu/master/master.proto
index 98e8d16..edd36d1 100644
--- a/src/kudu/master/master.proto
+++ b/src/kudu/master/master.proto
@@ -231,6 +231,12 @@ message SysNotificationLogEventIdPB {
optional int64 latest_notification_log_event_id = 1;
}
+// The on-disk entry in the sys.catalog table ("metadata" column) to represent
+// the cluster ID.
+message SysClusterIdEntryPB {
+ optional string cluster_id = 1;
+}
+
// The on-disk entry in the sys.catalog table to represent the existence of
// on-going tserver state (e.g. maintenance mode).
message SysTServerStateEntryPB {
@@ -921,6 +927,9 @@ message GetMasterRegistrationResponsePB {
// Set if there an error retrieving the registration information.
optional MasterErrorPB error = 4;
+
+ // The unique cluster ID of the cluster this server belongs too.
+ optional string cluster_id = 5;
}
// ListMastersRequest/Response: get information about all of the known
diff --git a/src/kudu/master/master_path_handlers.cc b/src/kudu/master/master_path_handlers.cc
index 671b4e4..fd281aa 100644
--- a/src/kudu/master/master_path_handlers.cc
+++ b/src/kudu/master/master_path_handlers.cc
@@ -590,6 +590,7 @@ void MasterPathHandlers::HandleMasters(const Webserver::WebRequest& /*req*/,
master_json["start_time"] = StartTimeToString(reg);
reg.clear_start_time(); // Clear 'start_time' before dumping to string.
master_json["registration"] = pb_util::SecureShortDebugString(reg);
+ master_json["cluster_id"] = master.has_cluster_id() ? master.cluster_id() : "";
}
}
diff --git a/src/kudu/master/master_service.cc b/src/kudu/master/master_service.cc
index ef25efc..c4c31c9 100644
--- a/src/kudu/master/master_service.cc
+++ b/src/kudu/master/master_service.cc
@@ -625,6 +625,7 @@ void MasterServiceImpl::GetMasterRegistration(const GetMasterRegistrationRequest
Status s = server_->GetMasterRegistration(resp->mutable_registration());
CheckRespErrorOrSetUnknown(s, resp);
resp->set_role(server_->catalog_manager()->Role());
+ resp->set_cluster_id(server_->cluster_id());
rpc->RespondSuccess();
}
diff --git a/src/kudu/master/sys_catalog-test.cc b/src/kudu/master/sys_catalog-test.cc
index a10349c..bbcc027 100644
--- a/src/kudu/master/sys_catalog-test.cc
+++ b/src/kudu/master/sys_catalog-test.cc
@@ -427,5 +427,35 @@ TEST_F(SysCatalogTest, AttemptOverwriteCertAuthorityInfo) {
ASSERT_EQ("Corruption: failed to write one or more rows", s.ToString());
}
+// Check loading the auto-generated cluster ID upon startup.
+TEST_F(SysCatalogTest, LoadClusterID) {
+ // The system catalog should already contain a generated cluster ID:
+ // The SetUp() method awaits for the catalog manager becoming leader master,
+ // and by that time the cluster ID should be loaded.
+ SysClusterIdEntryPB cluster_id_entry;
+ ASSERT_OK(master_->catalog_manager()->sys_catalog()->
+ GetClusterIdEntry(&cluster_id_entry));
+
+ ASSERT_TRUE(cluster_id_entry.has_cluster_id());
+ ASSERT_TRUE(!cluster_id_entry.cluster_id().empty());
+
+ ASSERT_EQ(cluster_id_entry.cluster_id(), master_->cluster_id());
+ const string init_id = master_->cluster_id();
+
+ // Check that if a cluster ID is already present,
+ // it cannot be overwritten using SysCatalogTable::AddClusterIdEntry().
+ const Status s = master_->catalog_manager()->sys_catalog()->
+ AddClusterIdEntry(cluster_id_entry);
+ ASSERT_TRUE(s.IsCorruption()) << s.ToString();
+ ASSERT_EQ("Corruption: failed to write one or more rows", s.ToString());
+
+ // Restart the master and ensure the generated cluster ID is the same.
+ mini_master_->Shutdown();
+ mini_master_->Restart();
+ ASSERT_OK(mini_master_->master()->catalog_manager()->sys_catalog()->
+ GetClusterIdEntry(&cluster_id_entry));
+ ASSERT_EQ(init_id, cluster_id_entry.cluster_id());
+}
+
} // namespace master
} // namespace kudu
diff --git a/src/kudu/master/sys_catalog.cc b/src/kudu/master/sys_catalog.cc
index e92399f..1ee8b20 100644
--- a/src/kudu/master/sys_catalog.cc
+++ b/src/kudu/master/sys_catalog.cc
@@ -155,6 +155,8 @@ const char* const SysCatalogTable::kInjectedFailureStatusMsg =
"INJECTED FAILURE";
const char* const SysCatalogTable::kLatestNotificationLogEntryIdRowId =
"latest_notification_log_entry_id";
+const char* const SysCatalogTable::kClusterIdRowId =
+ "cluster_id";
namespace {
@@ -790,6 +792,29 @@ Status SysCatalogTable::GetLatestNotificationLogEventId(int64_t* event_id) {
return ProcessRows<SysNotificationLogEventIdPB, HMS_NOTIFICATION_LOG>(processor);
}
+Status SysCatalogTable::GetClusterIdEntry(SysClusterIdEntryPB* entry) {
+ CHECK(entry);
+ vector<SysClusterIdEntryPB> entries;
+ auto processor = [&](
+ const string& entry_id,
+ const SysClusterIdEntryPB& entry_data) {
+ CHECK_EQ(entry_id, kClusterIdRowId);
+ DCHECK(entry_data.has_cluster_id());
+ DCHECK(!entry_data.cluster_id().empty());
+ entries.push_back(entry_data);
+ return Status::OK();
+ };
+ RETURN_NOT_OK((ProcessRows<SysClusterIdEntryPB, CLUSTER_ID>(processor)));
+ // There should be no more than one cluster ID entry in the system table.
+ CHECK_LE(entries.size(), 1);
+ if (entries.empty()) {
+ return Status::NotFound("cluster ID entry not found");
+ }
+ *entry = std::move(entries.front());
+
+ return Status::OK();
+}
+
Status SysCatalogTable::GetCertAuthorityEntry(SysCertAuthorityEntryPB* entry) {
CHECK(entry);
vector<SysCertAuthorityEntryPB> entries;
@@ -807,11 +832,33 @@ Status SysCatalogTable::GetCertAuthorityEntry(SysCertAuthorityEntryPB* entry) {
if (entries.empty()) {
return Status::NotFound("root CA entry not found");
}
- entries.front().Swap(entry);
+ *entry = std::move(entries.front());
return Status::OK();
}
+Status SysCatalogTable::AddClusterIdEntry(
+ const SysClusterIdEntryPB& entry) {
+ WriteRequestPB req;
+ req.set_tablet_id(kSysCatalogTabletId);
+ RETURN_NOT_OK(SchemaToPB(schema_, req.mutable_schema()));
+
+ CHECK(entry.has_cluster_id());
+ CHECK(!entry.cluster_id().empty());
+
+ faststring metadata_buf;
+ pb_util::SerializeToString(entry, &metadata_buf);
+
+ KuduPartialRow row(&schema_);
+ CHECK_OK(row.SetInt8(kSysCatalogTableColType, CLUSTER_ID));
+ CHECK_OK(row.SetStringNoCopy(kSysCatalogTableColId, kClusterIdRowId));
+ CHECK_OK(row.SetStringNoCopy(kSysCatalogTableColMetadata, metadata_buf));
+ RowOperationsPBEncoder enc(req.mutable_row_operations());
+ enc.Add(RowOperationsPB::INSERT, row);
+
+ return SyncWrite(req);
+}
+
Status SysCatalogTable::AddCertAuthorityEntry(
const SysCertAuthorityEntryPB& entry) {
WriteRequestPB req;
diff --git a/src/kudu/master/sys_catalog.h b/src/kudu/master/sys_catalog.h
index 68ae695..ac08a3a 100644
--- a/src/kudu/master/sys_catalog.h
+++ b/src/kudu/master/sys_catalog.h
@@ -59,6 +59,7 @@ namespace master {
class Master;
class SysCertAuthorityEntryPB;
+class SysClusterIdEntryPB;
class SysTServerStateEntryPB;
class SysTablesEntryPB;
class SysTabletsEntryPB;
@@ -102,6 +103,7 @@ class TServerStateVisitor {
// SysCatalogTable is a Kudu table that keeps track of the following
// system information:
+// * cluster id
// * table metadata
// * tablet metadata
// * root CA (certificate authority) certificate of the Kudu IPKI
@@ -140,6 +142,9 @@ class SysCatalogTable {
// The row ID of the latest notification log entry in the sys catalog table.
static const char* const kLatestNotificationLogEntryIdRowId;
+ // The row ID of the cluster ID entry in the sys catalog table.
+ static const char* const kClusterIdRowId;
+
typedef std::function<Status()> ElectedLeaderCallback;
enum CatalogEntryType {
@@ -149,6 +154,7 @@ class SysCatalogTable {
TSK_ENTRY = 4, // Token Signing Key entry.
HMS_NOTIFICATION_LOG = 5, // HMS notification log latest event ID.
TSERVER_STATE = 6, // TServer state.
+ CLUSTER_ID = 7 // Unique Cluster ID.
};
// 'leader_cb_' is invoked whenever this node is elected as a leader
@@ -220,9 +226,16 @@ class SysCatalogTable {
// Get the latest processed HMS notification log event ID.
Status GetLatestNotificationLogEventId(int64_t* event_id) WARN_UNUSED_RESULT;
+ // Get the cluster ID from the system table.
+ Status GetClusterIdEntry(SysClusterIdEntryPB* entry) WARN_UNUSED_RESULT;
+
// Retrive the CA entry (private key and certificate) from the system table.
Status GetCertAuthorityEntry(SysCertAuthorityEntryPB* entry);
+ // Add cluster ID entry into the system table.
+ // There should be no more than one cluster ID in the system table.
+ Status AddClusterIdEntry(const SysClusterIdEntryPB& entry);
+
// Add root CA entry (private key and certificate) into the system table.
// There should be no more than one CA entry in the system table.
Status AddCertAuthorityEntry(const SysCertAuthorityEntryPB& entry);
diff --git a/src/kudu/tools/tool_action_master.cc b/src/kudu/tools/tool_action_master.cc
index f0464cc..3a0a24f 100644
--- a/src/kudu/tools/tool_action_master.cc
+++ b/src/kudu/tools/tool_action_master.cc
@@ -157,6 +157,10 @@ Status ListMasters(const RunnerContext& context) {
for (const auto& master : masters) {
values.push_back(master.instance_id().permanent_uuid());
}
+ } else if (boost::iequals(column, "cluster_id")) {
+ for (const auto& master : masters) {
+ values.emplace_back(master.has_cluster_id() ? master.cluster_id() : "");
+ }
} else if (boost::iequals(column, "seqno")) {
for (const auto& master : masters) {
values.push_back(std::to_string(master.instance_id().instance_seqno()));
@@ -418,7 +422,7 @@ unique_ptr<Mode> BuildMasterMode() {
"columns",
string("uuid,rpc-addresses,role"),
string("Comma-separated list of master info fields to "
- "include in output.\nPossible values: uuid, "
+ "include in output.\nPossible values: uuid, cluster_id"
"rpc-addresses, http-addresses, version, seqno, "
"start_time and role"))
.AddOptionalParameter("format")
diff --git a/www/masters.mustache b/www/masters.mustache
index 268533e..d2a2786 100644
--- a/www/masters.mustache
+++ b/www/masters.mustache
@@ -31,6 +31,7 @@ under the License.
<table class='table table-striped'>
<thead><tr>
<th>UUID</th>
+ <th>Cluster ID</th>
<th>Role</th>
<th>Start time</th>
<th>Registration</th>
@@ -39,6 +40,7 @@ under the License.
{{#masters}}
<tr>
<td>{{#target}}<a href="{{.}}">{{/target}}{{uuid}}{{#target}}</a>{{/target}}</td>
+ <td>{{cluster_id}}</td>
<td>{{role}}</td>
<td>{{start_time}}</td>
<td><pre><code>{{registration}}</code></pre></td>