Support adding shard to cluster

RLTest/env.py

@@ -389,6 +389,15 @@ def getClusterConnectionIfNeeded(self):
         else:
             return self.getConnection()
 
+    def addShardToClusterIfExists(self):
+        if isinstance(self.envRunner, ClusterEnv):
+            test_fname = self.testName.replace(':', '_')
+            output_files_format = '%s-' + '%s-oss-cluster' % test_fname
+            kwargs = self.getEnvKwargs()
+            return self.envRunner.addShardToCluster(self.redisBinaryPath, output_files_format, **kwargs)
+        else:
+            raise Exception("env is not an oss-cluster")
+
     def getSlaveConnection(self):
         return self.envRunner.getSlaveConnection()
 

RLTest/redis_cluster.py

@@ -176,6 +176,25 @@ def getConnectionByKey(self, key, command):
         target_node = clusterConn._determine_nodes(command, key) # we will always which will give us the node responsible for the key
         return clusterConn.get_redis_connection(target_node[0])
 
+    def addShardToCluser(self, redisBinaryPath, output_files_format, **kwargs):
+        kwargs.pop('port')
+        port = self.shards[-1].port + 2  # use a fresh port
+        self.shardsCount += 1
+        new_shard = StandardEnv(redisBinaryPath, port, outputFilesFormat=output_files_format,
+                                serverId=self.shardsCount, clusterEnabled=True, **kwargs)
+        try:
+            new_shard.startEnv()
+        except Exception:
+            new_shard.stopEnv()
+            raise
+        self.shards.append(new_shard)
+
+        # Notify other shards that the new shard is available and wait for the topology change to be acknowledged.
+        conn = new_shard.getConnection()
+        for s in self.shards:
+            conn.execute_command('CLUSTER', 'MEET', '127.0.0.1', s.getMasterPort())
+        self.waitCluster()
+
     def flush(self):
         self.getClusterConnection().flushall()
 

tests/unit/test_redis_cluster.py

@@ -139,3 +139,16 @@ def test_connection_by_key(self):
             con = cluster_env.getConnectionByKey(key, "set")
             assert(con.set(key, "1"))
         cluster_env.stopEnv()
+
+    def test_add_shard_to_cluster(self):
+        shardsCount = 3
+        default_args = Defaults().getKwargs()
+        default_args['dbDirPath'] = self.test_dir
+        cluster_env = ClusterEnv(shardsCount=shardsCount, redisBinaryPath=REDIS_BINARY, outputFilesFormat='%s-test',
+                                 randomizePorts=Defaults.randomize_ports, **default_args)
+        cluster_env.startEnv()
+        cluster_env.addShardToCluster(REDIS_BINARY, '%s-test', **default_args)
+        assert cluster_env.shardsCount == shardsCount+1
+        new_shard_conn = cluster_env.getConnection(shardId=4)
+        assert new_shard_conn.ping()
+        assert new_shard_conn.cluster('info')['cluster_state'] == 'ok'

tests/unit/test_redis_std.py

@@ -393,4 +393,4 @@ def test_cluster_node_timeout(self):
         std_env.startEnv()
         con = std_env.getConnection()
         assert(con.execute_command("CONFIG", "GET", "cluster-node-timeout"), "60000")
-        std_env.stopEnv()
+        std_env.stopEnv()