Locality filter support to perpetual storage wiggler feature.

documentation/sphinx/source/command-line-interface.rst

@@ -64,7 +64,7 @@ The ``commit`` command commits the current transaction. Any sets or clears execu
 configure
 ---------
 
-The ``configure`` command changes the database configuration. Its syntax is ``configure [new|tss] [single|double|triple|three_data_hall|three_datacenter] [ssd|memory] [grv_proxies=<N>] [commit_proxies=<N>] [resolvers=<N>] [logs=<N>] [count=<TSS_COUNT>] [perpetual_storage_wiggle=<WIGGLE_SPEED>] [storage_migration_type={disabled|aggressive|gradual}]``.
+The ``configure`` command changes the database configuration. Its syntax is ``configure [new|tss] [single|double|triple|three_data_hall|three_datacenter] [ssd|memory] [grv_proxies=<N>] [commit_proxies=<N>] [resolvers=<N>] [logs=<N>] [count=<TSS_COUNT>] [perpetual_storage_wiggle=<WIGGLE_SPEED>] [perpetual_storage_wiggle_locality=<<LOCALITY_KEY>:<LOCALITY_VALUE>|0>] [storage_migration_type={disabled|aggressive|gradual}]``.
 
 The ``new`` option, if present, initializes a new database with the given configuration rather than changing the configuration of an existing one. When ``new`` is used, both a redundancy mode and a storage engine must be specified.
 
@@ -112,7 +112,10 @@ For recommendations on appropriate values for process types in large clusters, s
 perpetual storage wiggle
 ^^^^^^^^^^^^^^^^^^^^^^^^
 
-Set the value speed (a.k.a., the number of processes that the Data Distributor should wiggle at a time). Currently, only 0 and 1 are supported. The value 0 means to disable the perpetual storage wiggle. For more details, see :ref:`perpetual-storage-wiggle`.
+``perpetual_storage_wiggle`` sets the value speed (a.k.a., the number of processes that the Data Distributor should wiggle at a time). Currently, only 0 and 1 are supported. The value 0 means to disable the perpetual storage wiggle.
+``perpetual_storage_wiggle_locality`` sets the process filter for wiggling. The processes that match the given locality key and locality value are only wiggled. The value 0 will disable the locality filter and matches all the processes for wiggling.
+
+For more details, see :ref:`perpetual-storage-wiggle`.
 
 storage migration type
 ^^^^^^^^^^^^^^^^^^^^^^

documentation/sphinx/source/mr-status-json-schemas.rst.inc

@@ -736,6 +736,7 @@
          "proxies":6, // this field will be absent if a value has not been explicitly set
          "backup_worker_enabled":1,
          "perpetual_storage_wiggle": 0,
+         "perpetual_storage_wiggle_locality":"0"
          "storage_migration_type":{
          "$enum":[
              "disabled",

documentation/sphinx/source/perpetual-storage-wiggle.rst

@@ -40,6 +40,10 @@ Open perpetual storage wiggle: ``configure perpetual_storage_wiggle=1``.
 
 Disable perpetual storage wiggle on the cluster: ``configure perpetual_storage_wiggle=0``.
 
+Open perpetual storage wiggle for only processes matching the given locality key and value: ``configure perpetual_storage_wiggle=1 perpetual_storage_wiggle_locality=<LOCALITY_KEY>:<LOCALITY_VALUE>``.
+
+Disable perpetual storage wiggle locality matching filter, which wiggles all the processes: ``configure perpetual_storage_wiggle_locality=0``.
+
 Monitor
 =======
 

fdbcli/ConfigureCommand.actor.cpp

@@ -262,9 +262,9 @@ CommandFactory configureFactory(
     CommandHelp(
         "configure [new|tss]"
         "<single|double|triple|three_data_hall|three_datacenter|ssd|memory|memory-radixtree-beta|proxies=<PROXIES>|"
-        "commit_proxies=<COMMIT_PROXIES>|grv_proxies=<GRV_PROXIES>|logs=<LOGS>|resolvers=<RESOLVERS>>*|"
+        "commit_proxies=<COMMIT_PROXIES>|grv_proxies=<GRV_PROXIES>|logs=<LOGS>|resolvers=<RESOLVERS>>*|count=<TSS_COUNT>|"
-        "count=<TSS_COUNT>|perpetual_storage_wiggle=<WIGGLE_SPEED>|storage_migration_type={disabled|gradual|"
+        "perpetual_storage_wiggle=<WIGGLE_SPEED>|perpetual_storage_wiggle_locality=<<LOCALITY_KEY>:<LOCALITY_VALUE>|0>|"
-        "aggressive}",
+		"storage_migration_type={disabled|gradual|aggressive}",
         "change the database configuration",
         "The `new' option, if present, initializes a new database with the given configuration rather than changing "
         "the configuration of an existing one. When used, both a redundancy mode and a storage engine must be "
@@ -292,6 +292,9 @@ CommandFactory configureFactory(
         "perpetual_storage_wiggle=<WIGGLE_SPEED>: Set the value speed (a.k.a., the number of processes that the Data "
         "Distributor should wiggle at a time). Currently, only 0 and 1 are supported. The value 0 means to disable the "
         "perpetual storage wiggle.\n\n"
+        "perpetual_storage_wiggle_locality=<<LOCALITY_KEY>:<LOCALITY_VALUE>|0>: Set the process filter for wiggling. "
+        "The processes that match the given locality key and locality value are only wiggled. The value 0 will disable "
+        "the locality filter and matches all the processes for wiggling.\n\n"
         "See the FoundationDB Administration Guide for more information."));
 
 } // namespace fdb_cli

fdbcli/fdbcli.actor.cpp

@@ -1171,6 +1171,7 @@ void configureGenerator(const char* text, const char* line, std::vector<std::str
 		                   "logs=",
 		                   "resolvers=",
 		                   "perpetual_storage_wiggle=",
+		                   "perpetual_storage_wiggle_locality=",
 		                   "storage_migration_type=",
 		                   nullptr };
 	arrayGenerator(text, line, opts, lc);

fdbclient/DatabaseConfiguration.cpp

@@ -45,6 +45,7 @@ void DatabaseConfiguration::resetInternal() {
 	remoteTLogReplicationFactor = repopulateRegionAntiQuorum = 0;
 	backupWorkerEnabled = false;
 	perpetualStorageWiggleSpeed = 0;
+	perpetualStorageWiggleLocality = "0";
 	storageMigrationType = StorageMigrationType::DEFAULT;
 }
 
@@ -200,6 +201,8 @@ bool DatabaseConfiguration::isValid() const {
 	      (regions.size() == 0 || tLogPolicy->info() != "dcid^2 x zoneid^2 x 1") &&
 	      // We cannot specify regions with three_datacenter replication
 	      (perpetualStorageWiggleSpeed == 0 || perpetualStorageWiggleSpeed == 1) &&
+	      (perpetualStorageWiggleLocality.find(':') != std::string::npos ||
+	       !perpetualStorageWiggleLocality.compare("0")) &&
 	      storageMigrationType != StorageMigrationType::UNSET)) {
 		return false;
 	}
@@ -391,6 +394,7 @@ StatusObject DatabaseConfiguration::toJSON(bool noPolicies) const {
 
 	result["backup_worker_enabled"] = (int32_t)backupWorkerEnabled;
 	result["perpetual_storage_wiggle"] = perpetualStorageWiggleSpeed;
+	result["perpetual_storage_wiggle_locality"] = perpetualStorageWiggleLocality;
 	result["storage_migration_type"] = storageMigrationType.toString();
 	return result;
 }
@@ -545,6 +549,11 @@ bool DatabaseConfiguration::setInternal(KeyRef key, ValueRef value) {
 		parse(&regions, value);
 	} else if (ck == LiteralStringRef("perpetual_storage_wiggle")) {
 		parse(&perpetualStorageWiggleSpeed, value);
+	} else if (ck == LiteralStringRef("perpetual_storage_wiggle_locality")) {
+		if (value.toString().find(':') == std::string::npos && value.toString().compare("0")) {
+			return false;
+		}
+		perpetualStorageWiggleLocality = value.toString();
 	} else if (ck == LiteralStringRef("storage_migration_type")) {
 		parse((&type), value);
 		storageMigrationType = (StorageMigrationType::MigrationType)type;

fdbclient/DatabaseConfiguration.h

@@ -245,6 +245,7 @@ struct DatabaseConfiguration {
 
 	// Perpetual Storage Setting
 	int32_t perpetualStorageWiggleSpeed;
+	std::string perpetualStorageWiggleLocality;
 
 	// Storage Migration Type
 	StorageMigrationType storageMigrationType;

fdbclient/ManagementAPI.actor.cpp

@@ -150,6 +150,14 @@ std::map<std::string, std::string> configForToken(std::string const& mode) {
 			}
 			out[p + key] = value;
 		}
+		if (key == "perpetual_storage_wiggle_locality") {
+			if (value.find(':') == std::string::npos && value.compare("0")) {
+				printf("Error: perpetual_storage_wiggle_locality should be in <locality_key>:<locality_value> "
+				       "format or enter 0 to disable the locality match for wiggling.\n");
+				return out;
+			}
+			out[p + key] = value;
+		}
 		if (key == "storage_migration_type") {
 			StorageMigrationType type;
 			if (value == "disabled") {

fdbclient/Schemas.cpp

@@ -760,6 +760,7 @@ const KeyRef JSONSchemas::statusSchema = LiteralStringRef(R"statusSchema(
          "proxies":6,
          "backup_worker_enabled":1,
          "perpetual_storage_wiggle":0,
+         "perpetual_storage_wiggle_locality":"0",
          "storage_migration_type": {
              "$enum":[
              "disabled",

fdbclient/SystemData.cpp

@@ -630,6 +630,7 @@ const KeyRangeRef configKeys(LiteralStringRef("\xff/conf/"), LiteralStringRef("\
 const KeyRef configKeysPrefix = configKeys.begin;
 
 const KeyRef perpetualStorageWiggleKey(LiteralStringRef("\xff/conf/perpetual_storage_wiggle"));
+const KeyRef perpetualStorageWiggleLocalityKey(LiteralStringRef("\xff/conf/perpetual_storage_wiggle_locality"));
 const KeyRef wigglingStorageServerKey(LiteralStringRef("\xff/storageWigglePID"));
 
 const KeyRef triggerDDTeamInfoPrintKey(LiteralStringRef("\xff/triggerDDTeamInfoPrint"));

fdbclient/SystemData.h

@@ -210,6 +210,7 @@ extern const KeyRangeRef configKeys;
 extern const KeyRef configKeysPrefix;
 
 extern const KeyRef perpetualStorageWiggleKey;
+extern const KeyRef perpetualStorageWiggleLocalityKey;
 extern const KeyRef wigglingStorageServerKey;
 // Change the value of this key to anything and that will trigger detailed data distribution team info log.
 extern const KeyRef triggerDDTeamInfoPrintKey;

fdbserver/DataDistribution.actor.cpp

@@ -4009,16 +4009,69 @@ ACTOR Future<std::vector<std::pair<StorageServerInterface, ProcessClass>>> getSe
 // to a sorted PID set maintained by the data distributor. If now no storage server exists, the new Process ID is 0.
 ACTOR Future<Void> updateNextWigglingStoragePID(DDTeamCollection* teamCollection) {
 	state ReadYourWritesTransaction tr(teamCollection->cx);
-	state Value writeValue;
+	state Value writeValue = LiteralStringRef("");
 	state const Key writeKey =
 	    wigglingStorageServerKey.withSuffix(teamCollection->primary ? "/primary"_sr : "/remote"_sr);
 	loop {
 		try {
 			tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
-			Optional<Value> value = wait(tr.get(writeKey));
+			Optional<Value> locality = wait(tr.get(perpetualStorageWiggleLocalityKey));
+
 			if (teamCollection->pid2server_info.empty()) {
 				writeValue = LiteralStringRef("");
+			} else if (locality.present() && locality.get().toString().compare("0")) {
+				// if perpetual_storage_wiggle_locality has value and not 0(disabled).
+				state std::string localityKeyValue = locality.get().toString();
+				int split = localityKeyValue.find(':');
+				ASSERT(split != std::string::npos);
+
+				// get key and value from perpetual_storage_wiggle_locality.
+				state std::string localityKey = localityKeyValue.substr(0, split);
+				state std::string localityValue = localityKeyValue.substr(split + 1);
+				state Value prevValue;
+				state int serverInfoSize = teamCollection->pid2server_info.size();
+
+				Optional<Value> value = wait(tr.get(writeKey));
+				if (value.present()) {
+					prevValue = value.get();
+				} else {
+					// if value not present, check for locality match of the first entry in pid2server_info.
+					auto& info_vec = teamCollection->pid2server_info.begin()->second;
+					if (info_vec.size() && info_vec[0]->lastKnownInterface.locality.get(localityKey) == localityValue) {
+						writeValue = teamCollection->pid2server_info.begin()->first; // first entry locality matched.
+					} else {
+						prevValue = teamCollection->pid2server_info.begin()->first;
+						serverInfoSize--;
+					}
+				}
+
+				// If first entry of pid2server_info, did not match the locality.
+				if (!(writeValue.compare(LiteralStringRef("")))) {
+					while (true) {
+						auto nextIt = teamCollection->pid2server_info.upper_bound(prevValue);
+						if (nextIt == teamCollection->pid2server_info.end()) {
+							nextIt = teamCollection->pid2server_info.begin();
+						}
+
+						if (nextIt->second.size() &&
+						    nextIt->second[0]->lastKnownInterface.locality.get(localityKey) == localityValue) {
+							writeValue = nextIt->first; // locality matched
+							break;
+						}
+						serverInfoSize--;
+						if (!serverInfoSize) {
+							// None of the entries in pid2server_info matched the given locality.
+							writeValue = LiteralStringRef("");
+							TraceEvent("PerpetualNextWigglingStoragePIDNotFound", teamCollection->distributorId)
+							    .detail("WriteValue", "No process matched the given perpetualStorageWiggleLocality")
+							    .detail("PerpetualStorageWiggleLocality", localityKeyValue);
+							break;
+						}
+						prevValue = nextIt->first;
+					}
+				}
 			} else {
+				Optional<Value> value = wait(tr.get(writeKey));
 				Value pid = teamCollection->pid2server_info.begin()->first;
 				if (value.present()) {
 					auto nextIt = teamCollection->pid2server_info.upper_bound(value.get());
@@ -4031,6 +4084,7 @@ ACTOR Future<Void> updateNextWigglingStoragePID(DDTeamCollection* teamCollection
 					writeValue = pid;
 				}
 			}
+
 			tr.set(writeKey, writeValue);
 			wait(tr.commit());
 			break;

fdbserver/SimulatedCluster.actor.cpp

@@ -1781,7 +1781,7 @@ void setupSimulatedSystem(std::vector<Future<Void>>* systemActors,
 	}
 	auto configDBType = testConfig.getConfigDBType();
 	for (auto kv : startingConfigJSON) {
-		if ("tss_storage_engine" == kv.first) {
+		if ("tss_storage_engine" == kv.first || "perpetual_storage_wiggle_locality" == kv.first) {
 			continue;
 		}
 		startingConfigString += " ";