Merge pull request #18 from sfc-gh-sgupta/gate-blob-related-work

Add a knob to gate blob-related work.

fdbclient/ClientKnobs.cpp

@@ -251,6 +251,9 @@ void ClientKnobs::initialize(Randomize randomize) {
 	init( BUSYNESS_SPIKE_START_THRESHOLD,         0.100 );
 	init( BUSYNESS_SPIKE_SATURATED_THRESHOLD,     0.500 );
 
+	// blob granules 
+	init( ENABLE_BLOB_GRANULES,                    true );
+
 	// clang-format on
 }
 

fdbclient/ClientKnobs.h

@@ -242,6 +242,9 @@ public:
 	double BUSYNESS_SPIKE_START_THRESHOLD;
 	double BUSYNESS_SPIKE_SATURATED_THRESHOLD;
 
+	// blob granules
+	bool ENABLE_BLOB_GRANULES;
+
 	ClientKnobs(Randomize randomize);
 	void initialize(Randomize randomize);
 };

fdbclient/NativeAPI.actor.cpp

@@ -7096,6 +7096,9 @@ ACTOR Future<Void> getBlobGranuleRangesStreamActor(Reference<DatabaseContext> db
 }
 
 Future<Void> DatabaseContext::getBlobGranuleRangesStream(const PromiseStream<KeyRange>& results, KeyRange range) {
+	if (!CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
+		throw client_invalid_operation();
+	}
 	return getBlobGranuleRangesStreamActor(Reference<DatabaseContext>::addRef(this), results, range);
 }
 
@@ -7296,6 +7299,9 @@ Future<Void> DatabaseContext::readBlobGranulesStream(const PromiseStream<Standal
                                                      KeyRange range,
                                                      Version begin,
                                                      Version end) {
+	if (!CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
+		throw client_invalid_operation();
+	}
 	return readBlobGranulesStreamActor(Reference<DatabaseContext>::addRef(this), results, range, begin, end);
 }
 

fdbclient/ServerKnobs.cpp

@@ -766,7 +766,7 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi
 	init( BG_DELTA_FILE_TARGET_BYTES,   BG_DELTA_BYTES_BEFORE_COMPACT/10 );
 
 	// TODO should discuss proper value for this
-	init( BLOB_WORKER_TIMEOUT,                                   10.0 ); if( randomize && BUGGIFY ) BLOB_WORKER_TIMEOUT = 1.0;
+	init( BLOB_WORKER_TIMEOUT,                                  10.0 ); if( randomize && BUGGIFY ) BLOB_WORKER_TIMEOUT = 1.0;
 
 	// clang-format on
 

fdbserver/ClusterController.actor.cpp

@@ -5241,7 +5241,9 @@ ACTOR Future<Void> clusterControllerCore(ClusterControllerFullInterface interf,
 	self.addActor.send(handleForcedRecoveries(&self, interf));
 	self.addActor.send(monitorDataDistributor(&self));
 	self.addActor.send(monitorRatekeeper(&self));
-	self.addActor.send(monitorBlobManager(&self));
+	if (CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
+		self.addActor.send(monitorBlobManager(&self));
+	}
 	// self.addActor.send(monitorTSSMapping(&self));
 	self.addActor.send(dbInfoUpdater(&self));
 	self.addActor.send(traceCounters("ClusterControllerMetrics",

fdbserver/SimulatedCluster.actor.cpp

@@ -2002,7 +2002,10 @@ void setupSimulatedSystem(vector<Future<Void>>* systemActors,
 		// TODO: caching disabled for this merge
 		// FIXME: we hardcode some machines to specifically test storage cache and blob workers
 		int storageCacheMachines = dc == 0 ? 1 : 0;
-		int blobWorkerMachines = 2 + deterministicRandom()->randomInt(0, NUM_EXTRA_BW_MACHINES + 1);
+		int blobWorkerMachines = 0;
+		if (CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
+			blobWorkerMachines = 2 + deterministicRandom()->randomInt(0, NUM_EXTRA_BW_MACHINES + 1);
+		}
 
 		int totalMachines = machines + storageCacheMachines + blobWorkerMachines;
 		int useSeedForMachine = deterministicRandom()->randomInt(0, totalMachines);

fdbserver/workloads/BlobGranuleVerifier.actor.cpp

@@ -123,6 +123,10 @@ struct BlobGranuleVerifierWorkload : TestWorkload {
 
 	std::string description() const override { return "BlobGranuleVerifier"; }
 	Future<Void> setup(Database const& cx) override {
+		if (!CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
+			return Void();
+		}
+
 		if (doSetup) {
 			double initialDelay = deterministicRandom()->random01() * (maxDelay - minDelay) + minDelay;
 			if (BGV_DEBUG) {
@@ -380,6 +384,10 @@ struct BlobGranuleVerifierWorkload : TestWorkload {
 	}
 
 	Future<Void> start(Database const& cx) override {
+		if (!CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
+			return Void();
+		}
+
 		clients.reserve(threads + 1);
 		clients.push_back(timeout(findGranules(cx, this), testDuration, Void()));
 		for (int i = 0; i < threads; i++) {
@@ -445,7 +453,13 @@ struct BlobGranuleVerifierWorkload : TestWorkload {
 		return self->mismatches == 0 && checks > 0;
 	}
 
-	Future<bool> check(Database const& cx) override { return _check(cx, this); }
+	Future<bool> check(Database const& cx) override {
+		if (!CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
+			return true;
+		}
+
+		return _check(cx, this);
+	}
 	void getMetrics(vector<PerfMetric>& m) override {}
 };
 

fdbserver/workloads/ConsistencyCheck.actor.cpp

@@ -296,9 +296,11 @@ struct ConsistencyCheckWorkload : TestWorkload {
 					wait(::success(self->checkForExtraDataStores(cx, self)));
 
 					// Check blob workers are operating as expected
-					bool blobWorkersCorrect = wait(self->checkBlobWorkers(cx, configuration, self));
-					if (!blobWorkersCorrect)
-						self->testFailure("Blob workers incorrect");
+					if (CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
+						bool blobWorkersCorrect = wait(self->checkBlobWorkers(cx, configuration, self));
+						if (!blobWorkersCorrect)
+							self->testFailure("Blob workers incorrect");
+					}
 
 					// Check that each machine is operating as its desired class
 					bool usingDesiredClasses = wait(self->checkUsingDesiredClasses(cx, self));
@@ -2335,7 +2337,7 @@ struct ConsistencyCheckWorkload : TestWorkload {
 		}
 
 		// Check BlobManager
-		if (db.blobManager.present() &&
+		if (CLIENT_KNOBS->ENABLE_BLOB_GRANULES && db.blobManager.present() &&
 		    (!nonExcludedWorkerProcessMap.count(db.blobManager.get().address()) ||
 		     nonExcludedWorkerProcessMap[db.blobManager.get().address()].processClass.machineClassFitness(
 		         ProcessClass::BlobManager) > fitnessLowerBound)) {