Start of system test framework.

Scripts and runs system tests across the cluster in accordance with
the user's localconfig.py.
Includes a few very small tests for now. One tests a simple recovery
of one object. Others test recovery when recovery masters fail.

This patch also includes a way to push configuration options to
the coordinator. For now it is used just to set testing options
(that is, to ask the coordinator to crash some recovery masters).

Author: rstutsman

bindings/python/ramcloud.py

@@ -159,6 +159,11 @@ def from_param(param):
                                    ctypes.c_uint32, ctypes.c_uint32]
     so.rc_testing_fill.restype  = status
 
+    so.rc_testing_set_runtime_option.argtypes = [client,
+                                                 ctypes.c_char_p,
+                                                 ctypes.c_char_p]
+    so.rc_testing_set_runtime_option.restype = status
+
     so.rc_testing_wait_for_all_tablets_normal.argtypes = [client]
     so.rc_testing_wait_for_all_tablets_normal.restype = None
 
@@ -319,6 +324,9 @@ def testing_fill(self, table_id, id, object_count, object_size):
                                object_count, object_size)
         self.handle_error(s)
 
+    def testing_set_runtime_option(self, option, value):
+        so.rc_testing_set_runtime_option(self.client, option, value)
+
     def testing_wait_for_all_tablets_normal(self):
         so.rc_testing_wait_for_all_tablets_normal(self.client)
 

scripts/cluster.py

@@ -345,8 +345,9 @@ def __enter__(self):
         self.sandbox.__enter__()
         return self
 
-    def __exit__(self, exc_type, exc_value, exc_tb):
+    def __exit__(self, exc_type=None, exc_value=None, exc_tb=None):
         self.sandbox.__exit__(exc_type, exc_value, exc_tb)
+        return False # rethrow exception, if any
 
 def run(
         num_servers=4,             # Number of hosts on which to start

scripts/rawmetrics.py

@@ -286,7 +286,6 @@ def dump_metric_info_code(self, out, path, counter):
 rpc.metric('enlistServerCount', 'number of invocations of ENLIST_SERVER RPC')
 rpc.metric('getServerListCount', 'number of invocations of GET_SERVER_LIST RPC')
 rpc.metric('getTabletMapCount', 'number of invocations of GET_TABLET_MAP RPC')
-rpc.metric('getServerStatistics', 'number of invocations of GET_SERVER_STATISTICS RPC')
 rpc.metric('recoverCount', 'number of invocations of RECOVER RPC')
 rpc.metric('hintServerDownCount', 'number of invocations of HINT_SERVER_DOWN RPC')
 rpc.metric('recoveryMasterFinishedCount', 'number of invocations of RECOVERY_MASTER_FINISHED_RPC')
@@ -310,14 +309,16 @@ def dump_metric_info_code(self, out, path, counter):
 rpc.metric('dropTabletOwnershipCount', 'number of invocations of DROP_TABLET_OWNERSHIP RPC')
 rpc.metric('takeTabletOwnershipCount', 'number of invocations of TAKE_TABLET_OWNERSHIP RPC')
 rpc.metric('backupAssignGroupCount', 'number of invocations of BACKUP_ASSIGN_GROUP RPC')
-rpc.metric('incrementRpcCount', 'number of invocations of INCREMENT RPC')
-rpc.metric('splitTabletRpcCount', 'number of invocations of SPLIT_TABLET RPC')
 rpc.metric('getHeadOfLogCount', 'number of invocations of GET_HEAD_OF_LOG RPC')
+rpc.metric('incrementRpcCount', 'number of invocations of INCREMENT RPC')
 rpc.metric('prepForMigrationCount', 'number of invocations of PREP_FOR_MIGRATION RPC')
 rpc.metric('receiveMigrationDataCount', 'number of invocations of RECEIVE_MIGRATION_DATA RPC')
 rpc.metric('reassignTabletOwnershipCount', 'number of invocations of REASSIGN_TABLET_OWNERSHIP RPC')
 rpc.metric('migrateTabletCount', 'number of invocations of MIGRATE_TABLET RPC')
 rpc.metric('isReplicaNeededCount', 'number of invocations of IS_REPLICA_NEEDED_RPC')
+rpc.metric('splitTabletCount', 'number of invocations of SPLIT_TABLET')
+rpc.metric('getServerStatisticsCount', 'number of invocations of GET_SERVER_STATISTICS')
+rpc.metric('setRuntimeOptionCount', 'number of invocations of SET_RUNTIME_OPTION')
 rpc.metric('illegalRpcCount', 'number of invocations of RPCs with illegal opcodes')
 
 rpc.metric('rpc0Ticks', 'time spent executing RPC 0 (undefined)')
@@ -339,8 +340,6 @@ def dump_metric_info_code(self, out, path, counter):
 rpc.metric('enlistServerTicks', 'time spent executing ENLIST_SERVER RPC')
 rpc.metric('getServerListTicks', 'time spent executing GET_SERVER_LIST RPC')
 rpc.metric('getTabletMapTicks', 'time spent executing GET_TABLET_MAP RPC')
-rpc.metric('getServerStatisticsTicks', 'time spent executing GET_SERVER_STATISTICS RPC')
-rpc.metric('setTabletsTicks', 'time spent executing SET_TABLETS RPC')
 rpc.metric('recoverTicks', 'time spent executing RECOVER RPC')
 rpc.metric('hintServerDownTicks', 'time spent executing HINT_SERVER_DOWN RPC')
 rpc.metric('recoveryMasterFinishedTicks', 'time spent executing RECOVERY_MASTER_FINISHED RPC')
@@ -364,14 +363,16 @@ def dump_metric_info_code(self, out, path, counter):
 rpc.metric('dropTabletOwnershipTicks', 'number of invocations of DROP_TABLET_OWNERSHIP RPC')
 rpc.metric('takeTabletOwnershipTicks', 'number of invocations of TAKE_TABLET_OWNERSHIP RPC')
 rpc.metric('backupAssignGroupTicks', 'time spent executing BACKUP_ASSIGN_GROUP RPC')
-rpc.metric('incrementTicks', 'time spent executing INCREMENT RPC')
-rpc.metric('splitTabletTicks', 'time spent executing SPLIT_TABLET RPC')
 rpc.metric('getHeadOfLogTicks', 'time spent executing GET_HEAD_OF_LOG RPC')
+rpc.metric('incrementTicks', 'time spent executing INCREMENT RPC')
 rpc.metric('prepForMigrationTicks', 'time spent executing PREP_FOR_MIGRATION RPC')
 rpc.metric('receiveMigrationDataTicks', 'time spent executing RECEIVE_MIGRATION_DATA RPC')
 rpc.metric('reassignTabletOwnershipTicks', 'time spent executing REASSIGN_TABLET_OWNERSHIP RPC')
 rpc.metric('migrateTabletTicks', 'time spent executing MIGRATE_TABLET RPC')
 rpc.metric('isReplicaNeededTicks', 'time spent executing IS_REPLICA_NEEDED_RPC')
+rpc.metric('splitTabletTicks', 'time spent executing SPLIT_TABLET RPC')
+rpc.metric('getServerStatisticsTicks', 'time spent executing GET_SERVER_STATISTICS RPC')
+rpc.metric('setRuntimeOptionTicks', 'time spent executing SET_RUNTIME_OPTION RPC')
 rpc.metric('illegalRpcTicks', 'time spent executing RPCs with illegal opcodes')
 
 transmit = Group('Transmit', 'metrics related to transmitting messages')

src/CRamCloud.cc

@@ -273,6 +273,19 @@ rc_testing_fill(struct rc_client* client, uint64_t tableId,
     return STATUS_OK;
 }
 
+Status
+rc_testing_set_runtime_option(struct rc_client* client,
+                              const char* option,
+                              const char* value)
+{
+    try {
+        client->client->testingSetRuntimeOption(option, value);
+    } catch (const ClientException& e) {
+        return e.status;
+    }
+    return STATUS_OK;
+}
+
 void
 rc_testing_wait_for_all_tablets_normal(struct rc_client* client)
 {

src/CRamCloud.h

@@ -66,7 +66,9 @@ RAMCloud::Status    rc_testing_kill(struct rc_client* client, uint64_t tableId,
 RAMCloud::Status    rc_testing_fill(struct rc_client* client, uint64_t tableId,
                                     const char* key, uint16_t keyLength,
                                     uint32_t objectCount, uint32_t objectSize);
-
+RAMCloud::Status    rc_testing_set_runtime_option(struct rc_client* client,
+                                                  const char* option,
+                                                  const char* value);
 void rc_testing_wait_for_all_tablets_normal(struct rc_client* client);
 
 #ifdef __cplusplus

src/CoordinatorClient.cc

@@ -446,6 +446,34 @@ CoordinatorClient::sendServerList(ServerId destination)
     checkStatus(HERE);
 }
 
+/**
+ * Sets a runtime option field on the coordinator to the indicated value.
+ *
+ * \param option
+ *      String name which corresponds to a member field in the RuntimeOptions
+ *      class (e.g.  "failRecoveryMasters") whose value should be replaced with
+ *      the given value.
+ * \param value
+ *      String which can be parsed into the type of the field indicated by
+ *      \a option. The format is specific to the type of each field but is
+ *      generally either a single value (e.g. "10", "word") or a collection
+ *      separated by spaces (e.g. "1 2 3", "first second"). See RuntimeOptions
+ *      for more information.
+ */
+void
+CoordinatorClient::setRuntimeOption(const char* option, const char* value)
+{
+    Buffer req, resp;
+    SetRuntimeOptionRpc::Request& reqHdr(
+        allocHeader<SetRuntimeOptionRpc>(req));
+    reqHdr.optionLength = downCast<uint32_t>(strlen(option) + 1);
+    reqHdr.valueLength = downCast<uint32_t>(strlen(value) + 1);
+    Buffer::Chunk::appendToBuffer(&req, option, reqHdr.optionLength);
+    Buffer::Chunk::appendToBuffer(&req, value, reqHdr.valueLength);
+    sendRecv<SetRuntimeOptionRpc>(session, req, resp);
+    checkStatus(HERE);
+}
+
 CoordinatorClient::SetMinOpenSegmentId::SetMinOpenSegmentId(
         CoordinatorClient& client,
         ServerId serverId,

src/CoordinatorClient.h

@@ -66,6 +66,7 @@ class CoordinatorClient : public Client {
                                 bool successful);
     void setWill(uint64_t masterId, const ProtoBuf::Tablets& will);
     void sendServerList(ServerId destination);
+    void setRuntimeOption(const char* option, const char* value);
 
     class SetMinOpenSegmentId {
       public:

src/CoordinatorService.cc

@@ -35,7 +35,8 @@ CoordinatorService::CoordinatorService(Context& context)
     , nextTableId(0)
     , nextTableMasterIdx(0)
     , nextReplicationId(1)
-    , recoveryManager(context, serverList, tabletMap)
+    , runtimeOptions()
+    , recoveryManager(context, serverList, tabletMap, &runtimeOptions)
     , forceServerDownForTesting(false)
 {
     recoveryManager.start();
@@ -95,6 +96,10 @@ CoordinatorService::dispatch(RpcOpcode opcode,
             callHandler<SendServerListRpc, CoordinatorService,
                         &CoordinatorService::sendServerList>(rpc);
             break;
+        case SetRuntimeOptionRpc::opcode:
+            callHandler<SetRuntimeOptionRpc, CoordinatorService,
+                        &CoordinatorService::setRuntimeOption>(rpc);
+            break;
         case ReassignTabletOwnershipRpc::opcode:
             callHandler<ReassignTabletOwnershipRpc, CoordinatorService,
                         &CoordinatorService::reassignTabletOwnership>(rpc);
@@ -635,6 +640,30 @@ CoordinatorService::sendServerList(
     sendServerList(id);
 }
 
+/**
+ * Sets a runtime option field on the coordinator to the indicated value.
+ * See CoordinatorClient::setRuntimeOption() for details.
+ *
+ * \copydetails Service::ping
+ */
+void
+CoordinatorService::setRuntimeOption(const SetRuntimeOptionRpc::Request& reqHdr,
+                                     SetRuntimeOptionRpc::Response& respHdr,
+                                     Rpc& rpc)
+{
+    const char* option = getString(rpc.requestPayload, sizeof(reqHdr),
+                                   reqHdr.optionLength);
+    const char* value = getString(rpc.requestPayload,
+                                  downCast<uint32_t>(sizeof(reqHdr) +
+                                                     reqHdr.optionLength),
+                                  reqHdr.valueLength);
+    try {
+        runtimeOptions.set(option, value);
+    } catch (const std::out_of_range& e) {
+        respHdr.common.status = STATUS_OBJECT_DOESNT_EXIST;
+    }
+}
+
 /**
  * Assign a new replicationId to a backup, and inform the backup which nodes
  * are in its replication group.

src/CoordinatorService.h

@@ -25,6 +25,7 @@
 #include "RawMetrics.h"
 #include "Recovery.h"
 #include "Rpc.h"
+#include "RuntimeOptions.h"
 #include "ServerId.h"
 #include "Service.h"
 #include "TabletMap.h"
@@ -92,6 +93,9 @@ class CoordinatorService : public Service {
     void sendServerList(const SendServerListRpc::Request& reqHdr,
                         SendServerListRpc::Response& respHdr,
                         Rpc& rpc);
+    void setRuntimeOption(const SetRuntimeOptionRpc::Request& reqHdr,
+                          SetRuntimeOptionRpc::Response& respHdr,
+                          Rpc& rpc);
     // - helper methods -
     bool assignReplicationGroup(uint64_t replicationId,
                                 const vector<ServerId>& replicationGroupIds);
@@ -151,6 +155,13 @@ class CoordinatorService : public Service {
      */
     uint64_t nextReplicationId;
 
+    /**
+     * Contains coordinator configuration options which can be modified while
+     * the cluster is running. Currently mostly used for setting debugging
+     * or testing parameters.
+     */
+    RuntimeOptions runtimeOptions;
+
     /**
      * Handles all master recovery details on behalf of the coordinator.
      */

src/CoordinatorServiceTest.cc

@@ -14,6 +14,7 @@
  */
 
 #include "TestUtil.h"
+#include "ClientException.h"
 #include "CoordinatorClient.h"
 #include "CoordinatorService.h"
 #include "MasterService.h"
@@ -614,6 +615,17 @@ TEST_F(CoordinatorServiceTest, sendServerList_service) {
         "entries (version number 2)"));
 }
 
+TEST_F(CoordinatorServiceTest, setRuntimeOption) {
+    client->setRuntimeOption("failRecoveryMasters", "1 2 3");
+    ASSERT_EQ(3u, service->runtimeOptions.failRecoveryMasters.size());
+    EXPECT_EQ(1u, service->runtimeOptions.popFailRecoveryMasters());
+    EXPECT_EQ(2u, service->runtimeOptions.popFailRecoveryMasters());
+    EXPECT_EQ(3u, service->runtimeOptions.popFailRecoveryMasters());
+    EXPECT_EQ(0u, service->runtimeOptions.popFailRecoveryMasters());
+    EXPECT_THROW(client->setRuntimeOption("BAD", "1 2 3"),
+                 ObjectDoesntExistException);
+}
+
 TEST_F(CoordinatorServiceTest, setMinOpenSegmentId) {
     EXPECT_THROW(client->setMinOpenSegmentId(ServerId(2, 2), 100),
                  ClientException);

src/MakefragCoordinator

@@ -1,7 +1,8 @@
 COORDINATOR_SRCFILES := \
 			src/CoordinatorService.cc \
-			src/TabletMap.cc \
 			src/MasterRecoveryManager.cc \
+			src/TabletMap.cc \
+			src/RuntimeOptions.cc \
 			$(NULL)
 
 COORDINATOR_OBJFILES := $(COORDINATOR_SRCFILES)

src/MakefragTest

@@ -59,6 +59,7 @@ TESTS_SRCFILES := \
 		  src/ReplicaManagerTest.cc \
 		  src/ReplicatedSegmentTest.cc \
 		  src/RpcTest.cc \
+		  src/RuntimeOptionsTest.cc \
 		  src/SegmentTest.cc \
 		  src/SegmentIteratorTest.cc \
 		  src/ServerTest.cc \

src/MasterClient.cc

@@ -29,9 +29,21 @@ RejectRules defaultRejectRules;
 /**
  * Fill a master server with the given number of objects, each of the
  * same given size. Objects are added to all tables in the master in
- * a round-robin fashion.
+ * a round-robin fashion. This method exists simply to quickly fill a
+ * master for experiments.
  *
- * This method exists simply to quickly fill a master for experiments.
+ * See MasterClient::fillWithTestData() for more information.
+ *
+ * \bug Will return an error if the master only owns part of a table
+ * (because the hash of the fabricated keys may land in a region it
+ * doesn't own).
+ *
+ * \param numObjects
+ *      Total number of objects to add to the server.
+ * \param objectSize
+ *      Bytes of garbage data to place in each object not including the
+ *      key (the keys are ASCII strings starting with "0" and increasing
+ *      numerically in each table).
  */
 void
 MasterClient::fillWithTestData(uint32_t numObjects, uint32_t objectSize)

src/MasterRecoveryManager.cc

@@ -28,12 +28,17 @@ namespace RAMCloud {
  *      Authoritative list of all servers in the system and their details.
  * \param  tabletMap
  *      Authoritative information about tablets and their mapping to servers.
+ * \param runtimeOptions
+ *      Configuration options which are stored by the coordinator.
+ *      May be NULL for testing.
  */
 MasterRecoveryManager::MasterRecoveryManager(Context& context,
                                              CoordinatorServerList& serverList,
-                                             TabletMap& tabletMap)
+                                             TabletMap& tabletMap,
+                                             RuntimeOptions* runtimeOptions)
     : serverList(serverList)
     , tabletMap(tabletMap)
+    , runtimeOptions(runtimeOptions)
     , thread()
     , waitingRecoveries()
     , activeRecoveries()
@@ -226,6 +231,9 @@ class MaybeStartRecoveryTask : public Task {
                     "another recovery is active for the same ServerId",
                     recovery->crashedServerId.getId());
             } else {
+                if (mgr.runtimeOptions)
+                    recovery->testingFailRecoveryMasters =
+                        mgr.runtimeOptions->popFailRecoveryMasters();
                 recovery->schedule();
                 mgr.activeRecoveries[recovery->getRecoveryId()] = recovery;
                 mgr.waitingRecoveries.pop();

src/MasterRecoveryManager.h

@@ -21,6 +21,7 @@
 #include "CoordinatorServerList.h"
 #include "ProtoBuf.h"
 #include "Recovery.h"
+#include "RuntimeOptions.h"
 #include "ServerTracker.h"
 #include "TabletMap.h"
 #include "Tub.h"
@@ -51,7 +52,8 @@ class MasterRecoveryManager : public Recovery::Owner
   PUBLIC:
     MasterRecoveryManager(Context& context,
                           CoordinatorServerList& serverList,
-                          TabletMap& tabletMap);
+                          TabletMap& tabletMap,
+                          RuntimeOptions* runtimeOptions);
     ~MasterRecoveryManager();
 
     void start();
@@ -78,6 +80,13 @@ class MasterRecoveryManager : public Recovery::Owner
     /// Authoritative information about tablets and their mapping to servers.
     TabletMap& tabletMap;
 
+    /**
+     * Contains coordinator configuration options which can be modified while
+     * the cluster is running. Currently mostly used for setting debugging
+     * or testing parameters.
+     */
+    RuntimeOptions* runtimeOptions;
+
     /**
      * Drives recoveries; wakes up whenever new recoveries are waiting
      * or active recoveries have new work to complete.

src/MasterRecoveryManagerTest.cc

@@ -34,7 +34,7 @@ struct MasterRecoveryManagerTest : public ::testing::Test {
         : context()
         , serverList(context)
         , tabletMap()
-        , mgr(context, serverList, tabletMap)
+        , mgr(context, serverList, tabletMap, NULL)
     {
         Logger::get().setLogLevels(RAMCloud::SILENT_LOG_LEVEL);
     }