Added Throttling workload to test native health metrics API

fdbserver/CMakeLists.txt

@@ -158,6 +158,7 @@ set(FDBSERVER_SRCS
   workloads/TargetedKill.actor.cpp
   workloads/TaskBucketCorrectness.actor.cpp
   workloads/ThreadSafety.actor.cpp
+  workloads/Throttling.actor.cpp
   workloads/Throughput.actor.cpp
   workloads/TimeKeeperCorrectness.actor.cpp
   workloads/UnitPerf.actor.cpp

fdbserver/fdbserver.vcxproj

@@ -115,6 +115,7 @@
     <ActorCompiler Include="workloads\SelectorCorrectness.actor.cpp" />
     <ActorCompiler Include="workloads\KVStoreTest.actor.cpp" />
     <ActorCompiler Include="workloads\StreamingRead.actor.cpp" />
+    <ActorCompiler Include="workloads\Throttling.actor.cpp" />
     <ActorCompiler Include="workloads\Throughput.actor.cpp" />
     <ActorCompiler Include="workloads\WriteBandwidth.actor.cpp" />
     <ActorCompiler Include="workloads\QueuePush.actor.cpp" />

fdbserver/fdbserver.vcxproj.filters

@@ -80,6 +80,9 @@
     <ActorCompiler Include="workloads\ConflictRange.actor.cpp">
       <Filter>workloads</Filter>
     </ActorCompiler>
+    <ActorCompiler Include="workloads\Throttling.actor.cpp">
+      <Filter>workloads</Filter>
+    </ActorCompiler>
     <ActorCompiler Include="workloads\Throughput.actor.cpp">
       <Filter>workloads</Filter>
     </ActorCompiler>

fdbserver/workloads/Throttling.actor.cpp

@@ -0,0 +1,234 @@
+/*
+ * Throttling.actor.cpp
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2013-2018 Apple Inc. and the FoundationDB project authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <boost/lexical_cast.hpp>
+
+#include "fdbclient/ReadYourWrites.h"
+#include "fdbserver/workloads/workloads.h"
+
+struct TokenBucket {
+	static constexpr const double addTokensInterval = 0.1;
+	static constexpr const double maxSleepTime = 60.0;
+
+	double transactionRate;
+	double maxBurst;
+	double bucketSize;
+	Future<Void> tokenAdderActor;
+
+	ACTOR static Future<Void> tokenAdder(TokenBucket* self) {
+		loop {
+			self->bucketSize = std::min(self->bucketSize + self->transactionRate * addTokensInterval, self->maxBurst);
+			if (g_random->randomInt(0, 100) == 0)
+				TraceEvent("AddingTokensx100")
+				    .detail("BucketSize", self->bucketSize)
+				    .detail("TransactionRate", self->transactionRate);
+			wait(delay(addTokensInterval));
+		}
+	}
+
+	TokenBucket(double maxBurst = 1000) : transactionRate(0), maxBurst(maxBurst), bucketSize(maxBurst) {
+		tokenAdderActor = tokenAdder(this);
+	}
+
+	ACTOR static Future<Void> startTransaction(TokenBucket* self) {
+		if (self->bucketSize > 1.0) {
+			--self->bucketSize;
+			return Void();
+		}
+		state double sleepTime = addTokensInterval;
+		loop {
+			if (self->bucketSize > 1.0) {
+				--self->bucketSize;
+				return Void();
+			}
+			if (g_random->randomInt(0, 100) == 0)
+				TraceEvent("ThrottlingTransactionx100").detail("SleepTime", sleepTime);
+			wait(delay(sleepTime));
+			sleepTime = std::min(sleepTime * 2, maxSleepTime);
+		}
+	}
+};
+
+struct ThrottlingWorkload : KVWorkload {
+
+	double testDuration;
+	double healthMetricsCheckInterval;
+	int actorsPerClient;
+	int writesPerTransaction;
+	int readsPerTransaction;
+	double throttlingMultiplier;
+	int transactionsCommitted;
+	int64_t worstStorageQueue;
+	int64_t worstStorageNDV;
+	int64_t worstTLogQueue;
+	int64_t detailedWorstStorageQueue;
+	int64_t detailedWorstStorageNDV;
+	int64_t detailedWorstTLogQueue;
+	double detailedWorstCpuUsage;
+	double detailedWorstDiskUsage;
+	bool sendDetailedHealthMetrics;
+	TokenBucket tokenBucket;
+
+	ThrottlingWorkload(WorkloadContext const& wcx)
+	  : KVWorkload(wcx), transactionsCommitted(0), worstStorageQueue(0), worstStorageNDV(0), worstTLogQueue(0),
+	    detailedWorstStorageQueue(0), detailedWorstStorageNDV(0), detailedWorstTLogQueue(0), detailedWorstCpuUsage(0.0),
+	    detailedWorstDiskUsage(0.0) {
+		testDuration = getOption(options, LiteralStringRef("testDuration"), 60.0);
+		healthMetricsCheckInterval = getOption(options, LiteralStringRef("healthMetricsCheckInterval"), 1.0);
+		actorsPerClient = getOption(options, LiteralStringRef("actorsPerClient"), 10);
+		writesPerTransaction = getOption(options, LiteralStringRef("writesPerTransaction"), 10);
+		readsPerTransaction = getOption(options, LiteralStringRef("readsPerTransaction"), 10);
+		throttlingMultiplier = getOption(options, LiteralStringRef("throttlingMultiplier"), 0.5);
+		sendDetailedHealthMetrics = getOption(options, LiteralStringRef("sendDetailedHealthMetrics"), true);
+		int maxBurst = getOption(options, LiteralStringRef("maxBurst"), 1000);
+		tokenBucket.maxBurst = maxBurst;
+	}
+
+	ACTOR static Future<Void> healthMetricsChecker(Database cx, ThrottlingWorkload* self) {
+		loop {
+			wait(delay(self->healthMetricsCheckInterval));
+			HealthMetrics healthMetrics = cx->healthMetrics;
+
+			self->tokenBucket.transactionRate = healthMetrics.tpsLimit * self->throttlingMultiplier / self->clientCount;
+			self->worstStorageQueue = std::max(self->worstStorageQueue, healthMetrics.worstStorageQueue);
+			self->worstStorageNDV = std::max(self->worstStorageNDV, healthMetrics.worstStorageNDV);
+			self->worstTLogQueue = std::max(self->worstTLogQueue, healthMetrics.worstTLogQueue);
+
+			TraceEvent("HealthMetrics")
+			    .detail("WorstStorageQueue", healthMetrics.worstStorageQueue)
+			    .detail("WorstStorageNDV", healthMetrics.worstStorageNDV)
+			    .detail("WorstTLogQueue", healthMetrics.worstTLogQueue)
+			    .detail("TpsLimit", healthMetrics.tpsLimit);
+
+			TraceEvent traceStorageQueue("StorageQueue");
+			for (const auto& ss : healthMetrics.storageQueue) {
+				self->detailedWorstStorageQueue = std::max(self->detailedWorstStorageQueue, ss.second);
+				traceStorageQueue.detail(format("Storage/%s", ss.first.toString().c_str()), ss.second);
+			}
+
+			TraceEvent traceStorageNDV("StorageNDV");
+			for (const auto& ss : healthMetrics.storageNDV) {
+				self->detailedWorstStorageNDV = std::max(self->detailedWorstStorageNDV, ss.second);
+				traceStorageNDV.detail(format("Storage/%s", ss.first.toString().c_str()), ss.second);
+			}
+
+			TraceEvent traceTLogQueue("TLogQueue");
+			for (const auto& ss : healthMetrics.tLogQueue) {
+				self->detailedWorstTLogQueue = std::max(self->detailedWorstTLogQueue, ss.second);
+				traceTLogQueue.detail(format("TLog/%s", ss.first.toString().c_str()), ss.second);
+			}
+
+			TraceEvent traceCpuUsage("CpuUsage");
+			for (const auto& ss : healthMetrics.cpuUsage) {
+				self->detailedWorstCpuUsage = std::max(self->detailedWorstCpuUsage, ss.second);
+				traceCpuUsage.detail(format("Storage/%s", ss.first.toString().c_str()), ss.second);
+			}
+
+			TraceEvent traceDiskUsage("DiskUsage");
+			for (const auto& ss : healthMetrics.diskUsage) {
+				self->detailedWorstDiskUsage = std::max(self->detailedWorstDiskUsage, ss.second);
+				traceDiskUsage.detail(format("Storage/%s", ss.first.toString().c_str()), ss.second);
+			}
+		}
+	}
+
+	static Value getRandomValue() { return Standalone<StringRef>(format("Value/%d", g_random->randomInt(0, 10e6))); }
+
+	ACTOR static Future<Void> clientActor(Database cx, ThrottlingWorkload* self) {
+		state ReadYourWritesTransaction tr(cx);
+
+		loop {
+			wait(TokenBucket::startTransaction(&self->tokenBucket));
+			tr.reset();
+			try {
+				state int i;
+				for (i = 0; i < self->readsPerTransaction; ++i) {
+					state Optional<Value> value = wait(tr.get(self->getRandomKey()));
+				}
+				for (i = 0; i < self->writesPerTransaction; ++i) {
+					tr.set(self->getRandomKey(), getRandomValue());
+				}
+				wait(tr.commit());
+				if (g_random->randomInt(0, 1000) == 0) TraceEvent("TransactionCommittedx1000");
+				++self->transactionsCommitted;
+			} catch (Error& e) {
+				// ignore failing transactions
+			}
+		}
+	}
+
+	ACTOR static Future<Void> _setup(Database cx, ThrottlingWorkload* self) {
+		Standalone<StringRef> value(format("%d", self->sendDetailedHealthMetrics ? 1 : 0));
+		setNetworkOption(FDBNetworkOptions::SEND_DETAILED_HEALTH_METRICS, Optional<StringRef>(value));
+		if (!self->sendDetailedHealthMetrics) {
+			// Clear detailed health metrics that are already populated
+			wait(delay(2 * CLIENT_KNOBS->UPDATE_DETAILED_HEALTH_METRICS_INTERVAL));
+			cx->healthMetrics.cpuUsage.clear();
+			cx->healthMetrics.storageQueue.clear();
+			cx->healthMetrics.storageNDV.clear();
+			cx->healthMetrics.tLogQueue.clear();
+		}
+		return Void();
+	}
+
+	ACTOR static Future<Void> _start(Database cx, ThrottlingWorkload* self) {
+		state Future<Void> hmChecker = timeout(healthMetricsChecker(cx, self), self->testDuration, Void());
+
+		state vector<Future<Void>> clientActors;
+		state int actorId;
+		for (actorId = 0; actorId < self->actorsPerClient; ++actorId) {
+			clientActors.push_back(timeout(clientActor(cx, self), self->testDuration, Void()));
+		}
+		wait(hmChecker);
+		return Void();
+	}
+
+	virtual std::string description() { return "Throttling"; }
+	virtual Future<Void> setup(Database const& cx) { return _setup(cx, this); }
+	virtual Future<Void> start(Database const& cx) { return _start(cx, this); }
+	virtual Future<bool> check(Database const& cx) {
+		if (worstStorageQueue == 0 || worstStorageNDV == 0 || worstTLogQueue == 0 || transactionsCommitted == 0)
+			return false;
+		if (sendDetailedHealthMetrics) {
+			if (detailedWorstStorageQueue == 0 || detailedWorstStorageNDV == 0 || detailedWorstTLogQueue == 0 ||
+			    detailedWorstCpuUsage == 0.0 || detailedWorstDiskUsage == 0.0)
+				return false;
+		} else {
+			if (detailedWorstStorageQueue != 0 || detailedWorstStorageNDV != 0 || detailedWorstTLogQueue != 0 ||
+			    detailedWorstCpuUsage != 0.0 || detailedWorstCpuUsage != 0.0)
+				return false;
+		}
+		return true;
+	}
+
+	virtual void getMetrics(vector<PerfMetric>& m) {
+		m.push_back(PerfMetric("TransactionsCommitted", transactionsCommitted, false));
+		m.push_back(PerfMetric("WorstStorageQueue", worstStorageQueue, true));
+		m.push_back(PerfMetric("DetailedWorstStorageQueue", detailedWorstStorageQueue, true));
+		m.push_back(PerfMetric("WorstStorageNDV", worstStorageNDV, true));
+		m.push_back(PerfMetric("DetailedWorstStorageNDV", detailedWorstStorageNDV, true));
+		m.push_back(PerfMetric("WorstTLogQueue", worstTLogQueue, true));
+		m.push_back(PerfMetric("DetailedWorstTLogQueue", detailedWorstTLogQueue, true));
+		m.push_back(PerfMetric("DetailedWorstCpuUsage", detailedWorstCpuUsage, true));
+		m.push_back(PerfMetric("DetailedWorstDiskUsage", detailedWorstDiskUsage, true));
+	}
+};
+
+WorkloadFactory<ThrottlingWorkload> ThrottlingWorkloadFactory("Throttling");

tests/CMakeLists.txt

@@ -66,6 +66,7 @@ add_fdb_test(TEST_FILES SlowTask.txt IGNORE)
 add_fdb_test(TEST_FILES SpecificUnitTest.txt IGNORE)
 add_fdb_test(TEST_FILES StreamingWrite.txt IGNORE)
 add_fdb_test(TEST_FILES ThreadSafety.txt IGNORE)
+add_fdb_test(TEST_FILES Throttling.txt IGNORE)
 add_fdb_test(TEST_FILES TraceEventMetrics.txt IGNORE)
 add_fdb_test(TEST_FILES default.txt IGNORE)
 add_fdb_test(TEST_FILES errors.txt IGNORE)

tests/Throttling.txt

@@ -0,0 +1,10 @@
+testTitle=Throttling
+testName=Throttling
+testDuration=60.0
+healthMetricsCheckInterval=1.0
+actorsPerClient=10
+readsPerTransaction=10
+writesPerTransaction=10
+throttlingMultiplier=0.5
+sendDetailedHealthMetrics=false
+maxBurst=10000