Don't allow to switch off after enabling

git-hulk · git-hulk · commit 4391ba44bc10 · 2023-10-01T18:39:55.000+08:00
diff --git a/kvrocks.conf b/kvrocks.conf
@@ -55,7 +55,8 @@ cluster-enabled no
 # to replicas. This option allows to change this behavior, so that namespaces are also
 # propagated to slaves. Note that:
 # 1) it won't replicate the 'masterauth' to prevent breaking master/replica replication
-# 2) it will overwrite replica's namespace with master's namespace, so be careful of in-using namespaces.
+# 2) it will overwrite replica's namespace with master's namespace, so be careful of in-using namespaces
+# 3) cannot switch off the namespace replication once it's enabled
 #
 # Default: no
 repl-namespace-enabled no
diff --git a/src/cluster/replication.cc b/src/cluster/replication.cc
@@ -976,7 +976,7 @@ Status ReplicationThread::parseWriteBatch(const std::string &batch_string) {
           }
         }
       } else if (write_batch_handler.Key() == kNamespaceDBKey) {
-        auto s = srv_->GetNamespace()->Load();
+        auto s = srv_->GetNamespace()->LoadAndRewrite();
         if (!s.IsOK()) {
           return s.Prefixed("failed to load namespaces");
         }
diff --git a/src/config/config.cc b/src/config/config.cc
@@ -517,15 +517,11 @@ void Config::initFieldCallback() {
          return Status::OK();
        }},
       {"repl-namespace-enabled",
-       [this](Server *srv, const std::string &k, const std::string &v) -> Status {
-         if (!srv || !cluster_enabled) return Status::OK();
-         auto new_value = util::ToLower(v);
-         // If the namespace replication is enabled/disabled, need to reload.
-         if ((repl_namespace_enabled && new_value == "no") || (!repl_namespace_enabled && new_value == "yes")) {
-           return srv->GetNamespace()->Load();
-         }
-         return Status::OK();
+       [](Server *srv, const std::string &k, const std::string &v) -> Status {
+         if (!srv) return Status::OK();
+         return srv->GetNamespace()->LoadAndRewrite();
        }},
+
       {"rocksdb.target_file_size_base",
        [this](Server *srv, const std::string &k, const std::string &v) -> Status {
          if (!srv) return Status::OK();
diff --git a/src/server/namespace.cc b/src/server/namespace.cc
@@ -43,7 +43,7 @@ Status IsNamespaceLegal(const std::string& ns) {
   return Status::OK();
 }
 
-Status Namespace::Load() {
+Status Namespace::LoadAndRewrite() {
   auto config = storage_->GetConfig();
   // Load from the configuration file first
   tokens_ = config->load_tokens;
@@ -52,18 +52,25 @@ Status Namespace::Load() {
   // this can avoid missing some namespaces when turn on/off repl_namespace_enabled.
   std::string value;
   auto s = storage_->Get(rocksdb::ReadOptions(), cf_, kNamespaceDBKey, &value);
-  if (!s.ok()) {
-    if (s.IsNotFound()) return Status::OK();
+  if (!s.ok() && !s.IsNotFound()) {
     return {Status::NotOK, s.ToString()};
   }
-  jsoncons::json j = jsoncons::json::parse(value);
-  for (const auto& iter : j.object_range()) {
-    if (tokens_.find(iter.key()) == tokens_.end()) {
-      // merge the namespace from db
-      tokens_[iter.key()] = iter.value().as<std::string>();
+  if (s.ok()) {
+    // The namespace db key is existed, so it doesn't allow to switch off repl_namespace_enabled
+    if (!config->repl_namespace_enabled) {
+      return {Status::NotOK, "cannot switch off repl_namespace_enabled when namespaces exist in db"};
+    }
+
+    jsoncons::json j = jsoncons::json::parse(value);
+    for (const auto& iter : j.object_range()) {
+      if (tokens_.find(iter.key()) == tokens_.end()) {
+        // merge the namespace from db
+        tokens_[iter.key()] = iter.value().as<std::string>();
+      }
     }
   }
-  return rewriteOrWriteDB();
+
+  return Rewrite();
 }
 
 StatusOr<std::string> Namespace::Get(const std::string& ns) const {
@@ -108,7 +115,7 @@ Status Namespace::Set(const std::string& ns, const std::string& token) {
   }
   tokens_[token] = ns;
 
-  s = rewriteOrWriteDB();
+  s = Rewrite();
   if (!s.IsOK()) {
     tokens_.erase(token);
     return s;
@@ -139,7 +146,7 @@ Status Namespace::Del(const std::string& ns) {
   for (const auto& iter : tokens_) {
     if (iter.second == ns) {
       tokens_.erase(iter.first);
-      auto s = rewriteOrWriteDB();
+      auto s = Rewrite();
       if (!s.IsOK()) {
         tokens_[iter.first] = iter.second;
         return s;
@@ -150,7 +157,7 @@ Status Namespace::Del(const std::string& ns) {
   return {Status::NotOK, kErrNamespaceNotFound};
 }
 
-Status Namespace::rewriteOrWriteDB() {
+Status Namespace::Rewrite() {
   auto config = storage_->GetConfig();
   auto s = config->Rewrite(tokens_);
   if (!s.IsOK()) {
diff --git a/src/server/namespace.h b/src/server/namespace.h
@@ -34,18 +34,17 @@ class Namespace {
   Namespace(const Namespace &) = delete;
   Namespace &operator=(const Namespace &) = delete;
 
-  Status Load();
+  Status LoadAndRewrite();
   StatusOr<std::string> Get(const std::string &ns) const;
   StatusOr<std::string> GetByToken(const std::string &token) const;
   Status Set(const std::string &ns, const std::string &token);
   Status Add(const std::string &ns, const std::string &token);
   Status Del(const std::string &ns);
   const std::map<std::string, std::string> &List() const { return tokens_; }
+  Status Rewrite();
 
  private:
   engine::Storage *storage_;
   rocksdb::ColumnFamilyHandle *cf_ = nullptr;
   std::map<std::string, std::string> tokens_;
-
-  Status rewriteOrWriteDB();
 };
diff --git a/src/server/server.cc b/src/server/server.cc
@@ -132,7 +132,7 @@ Server::~Server() {
 //     - feed-replica-data-info: generate checkpoint and send files list when full sync
 //     - feed-replica-file: send SST files when slaves ask for full sync
 Status Server::Start() {
-  auto s = namespace_.Load();
+  auto s = namespace_.LoadAndRewrite();
   if (!s.IsOK()) {
     return s;
   }
diff --git a/tests/gocase/unit/namespace/namespace_test.go b/tests/gocase/unit/namespace/namespace_test.go
@@ -89,6 +89,11 @@ func TestNamespace(t *testing.T) {
 				require.NoError(t, r.Err())
 				require.Equal(t, "OK", r.Val())
 			}
+			for ns, token := range nsTokens {
+				r := rdb.Do(ctx, "NAMESPACE", "GET", ns)
+				require.NoError(t, r.Err())
+				require.Equal(t, token, r.Val())
+			}
 
 			srv.Restart()
 			for ns, token := range nsTokens {
@@ -167,9 +172,9 @@ func TestNamespaceReplicate(t *testing.T) {
 		"n4": "t4",
 	}
 
-	t.Run("Replicate namespaces", func(t *testing.T) {
-		require.NoError(t, masterRdb.ConfigSet(ctx, "repl-namespace-enabled", "yes").Err())
-		require.NoError(t, slaveRdb.ConfigSet(ctx, "repl-namespace-enabled", "yes").Err())
+	t.Run("Disable Replicate namespces", func(t *testing.T) {
+		require.NoError(t, masterRdb.ConfigSet(ctx, "repl-namespace-enabled", "no").Err())
+		require.NoError(t, slaveRdb.ConfigSet(ctx, "repl-namespace-enabled", "no").Err())
 
 		for ns, token := range nsTokens {
 			r := masterRdb.Do(ctx, "NAMESPACE", "ADD", ns, token)
@@ -178,53 +183,61 @@ func TestNamespaceReplicate(t *testing.T) {
 		}
 		util.WaitForOffsetSync(t, slaveRdb, masterRdb)
 
+		// Can read namespaces on master
 		for ns, token := range nsTokens {
-			r := slaveRdb.Do(ctx, "NAMESPACE", "GET", ns)
+			r := masterRdb.Do(ctx, "NAMESPACE", "GET", ns)
 			require.NoError(t, r.Err())
 			require.Equal(t, token, r.Val())
 		}
+		// Can't read namespaces on slave
+		for ns := range nsTokens {
+			r := slaveRdb.Do(ctx, "NAMESPACE", "GET", ns)
+			require.EqualError(t, r.Err(), redis.Nil.Error())
+		}
 
 		for ns := range nsTokens {
 			r := masterRdb.Do(ctx, "NAMESPACE", "DEL", ns)
 			require.NoError(t, r.Err())
 			require.Equal(t, "OK", r.Val())
 		}
-		util.WaitForOffsetSync(t, slaveRdb, masterRdb)
-
-		for ns := range nsTokens {
-			r := slaveRdb.Do(ctx, "NAMESPACE", "GET", ns)
-			require.EqualError(t, r.Err(), redis.Nil.Error())
-		}
 	})
 
-	t.Run("Don't allow to operate slave's namespace if replication is enabled", func(t *testing.T) {
-		r := slaveRdb.Do(ctx, "NAMESPACE", "ADD", "ns_xxxx", "token_xxxx")
-		util.ErrorRegexp(t, r.Err(), ".*ERR namespace is read-only for slave.*")
-	})
+	t.Run("Enable Replicate namespaces", func(t *testing.T) {
+		require.NoError(t, masterRdb.ConfigSet(ctx, "repl-namespace-enabled", "yes").Err())
+		require.NoError(t, slaveRdb.ConfigSet(ctx, "repl-namespace-enabled", "yes").Err())
 
-	t.Run("Turn off namespace replication", func(t *testing.T) {
-		require.NoError(t, masterRdb.ConfigSet(ctx, "repl-namespace-enabled", "no").Err())
-		require.NoError(t, slaveRdb.ConfigSet(ctx, "repl-namespace-enabled", "no").Err())
 		for ns, token := range nsTokens {
 			r := masterRdb.Do(ctx, "NAMESPACE", "ADD", ns, token)
 			require.NoError(t, r.Err())
 			require.Equal(t, "OK", r.Val())
 		}
+		util.WaitForOffsetSync(t, slaveRdb, masterRdb)
+
 		for ns, token := range nsTokens {
-			r := masterRdb.Do(ctx, "NAMESPACE", "GET", ns)
+			r := slaveRdb.Do(ctx, "NAMESPACE", "GET", ns)
 			require.NoError(t, r.Err())
 			require.Equal(t, token, r.Val())
 		}
+
+		for ns := range nsTokens {
+			r := masterRdb.Do(ctx, "NAMESPACE", "DEL", ns)
+			require.NoError(t, r.Err())
+			require.Equal(t, "OK", r.Val())
+		}
 		util.WaitForOffsetSync(t, slaveRdb, masterRdb)
+
 		for ns := range nsTokens {
 			r := slaveRdb.Do(ctx, "NAMESPACE", "GET", ns)
 			require.EqualError(t, r.Err(), redis.Nil.Error())
 		}
 	})
 
-	t.Run("Allow to operate slave's namespace if replication is disabled", func(t *testing.T) {
+	t.Run("Don't allow to operate slave's namespace if replication is enabled", func(t *testing.T) {
 		r := slaveRdb.Do(ctx, "NAMESPACE", "ADD", "ns_xxxx", "token_xxxx")
-		require.NoError(t, r.Err())
-		require.Equal(t, "OK", r.Val())
+		util.ErrorRegexp(t, r.Err(), ".*ERR namespace is read-only for slave.*")
+	})
+
+	t.Run("Turn off namespace replication is not allowed", func(t *testing.T) {
+		util.ErrorRegexp(t, masterRdb.ConfigSet(ctx, "repl-namespace-enabled", "no").Err(), ".*cannot switch off repl_namespace_enabled when namespaces exist in db.*")
 	})
 }

Original file line number	Diff line number	Diff line change
`@@ -976,7 +976,7 @@ Status ReplicationThread::parseWriteBatch(const std::string &batch_string) {`
`976`	`976`	`}`
`977`	`977`	`}`
`978`	`978`	`} else if (write_batch_handler.Key() == kNamespaceDBKey) {`
`979`		`- auto s = srv_->GetNamespace()->Load();`
	`979`	`+ auto s = srv_->GetNamespace()->LoadAndRewrite();`
`980`	`980`	`if (!s.IsOK()) {`
`981`	`981`	`return s.Prefixed("failed to load namespaces");`
`982`	`982`	`}`
Original file line number	Diff line number	Diff line change
`@@ -132,7 +132,7 @@ Server::~Server() {`
`132`	`132`	`// - feed-replica-data-info: generate checkpoint and send files list when full sync`
`133`	`133`	`// - feed-replica-file: send SST files when slaves ask for full sync`
`134`	`134`	`Status Server::Start() {`
`135`		`- auto s = namespace_.Load();`
	`135`	`+ auto s = namespace_.LoadAndRewrite();`
`136`	`136`	`if (!s.IsOK()) {`
`137`	`137`	`return s;`
`138`	`138`	`}`