loadtest: Fix various issues, update for pre-testnet

tools/loadtest/src/main/kotlin/net/corda/loadtest/ConnectionManager.kt

@@ -72,7 +72,9 @@ class ConnectionManager(private val username: String, private val jSch: JSch) {
             remoteMessagingPort: Int,
             localTunnelAddress: HostAndPort,
             certificatesBaseDirectory: Path,
-            remoteCertificatesDirectory: Path
+            remoteCertificatesDirectory: Path,
+            rpcUsername: String,
+            rpcPassword: String
     ): NodeConnection {
         val session = jSch.getSession(username, nodeHost, 22)
         // We don't check the host fingerprints because they may change often
@@ -97,7 +99,7 @@ class ConnectionManager(private val username: String, private val jSch: JSch) {
         channel.disconnect()
         log.info("Certificates copied!")
 
-        val connection = NodeConnection(nodeHost, session, localTunnelAddress, certificatesDirectory)
+        val connection = NodeConnection(nodeHost, session, localTunnelAddress, certificatesDirectory, rpcUsername, rpcPassword)
         connection.startClient()
         return connection
     }
@@ -121,6 +123,8 @@ fun <A> connectToNodes(
         remoteMessagingPort: Int,
         tunnelPortAllocation: PortAllocation,
         certificatesBaseDirectory: Path,
+        rpcUsername: String,
+        rpcPassword: String,
         withConnections: (List<NodeConnection>) -> A
 ): A {
     val manager = ConnectionManager(username, setupJSchWithSshAgent())
@@ -130,7 +134,9 @@ fun <A> connectToNodes(
                 remoteMessagingPort = remoteMessagingPort,
                 localTunnelAddress = tunnelPortAllocation.nextHostAndPort(),
                 certificatesBaseDirectory = certificatesBaseDirectory,
-                remoteCertificatesDirectory = nodeHostAndCertificatesPath.second
+                remoteCertificatesDirectory = nodeHostAndCertificatesPath.second,
+                rpcUsername = rpcUsername,
+                rpcPassword = rpcPassword
         )
     }.toList()
 
@@ -151,7 +157,9 @@ class NodeConnection(
         val hostName: String,
         private val jSchSession: Session,
         private val localTunnelAddress: HostAndPort,
-        private val certificatesDirectory: Path
+        private val certificatesDirectory: Path,
+        private val rpcUsername: String,
+        private val rpcPassword: String
 ) : Closeable {
 
     private val sslConfig = object : SSLConfiguration {
@@ -187,7 +195,7 @@ class NodeConnection(
     fun <A> doWhileClientStopped(action: () -> A): A {
         val client = client
         val proxy = _proxy
-        check(client == null || proxy == null) { "doWhileClientStopped called with no running client" }
+        require(client != null && proxy != null) { "doWhileClientStopped called with no running client" }
         log.info("Stopping RPC proxy to $hostName, tunnel at $localTunnelAddress")
         client!!.close()
         try {
@@ -196,7 +204,7 @@ class NodeConnection(
             log.info("Starting new RPC proxy to $hostName, tunnel at $localTunnelAddress")
             val newClient = CordaRPCClient(localTunnelAddress, sslConfig)
             // TODO expose these somehow?
-            newClient.start("user1", "test")
+            newClient.start(rpcUsername, rpcPassword)
             val newProxy = newClient.proxy()
             this.client = newClient
             this._proxy = newProxy
@@ -206,7 +214,7 @@ class NodeConnection(
     fun startClient() {
         log.info("Creating RPC proxy to $hostName, tunnel at $localTunnelAddress")
         val client = CordaRPCClient(localTunnelAddress, sslConfig)
-        client.start("user1", "test")
+        client.start(rpcUsername, rpcPassword)
         val proxy = client.proxy()
         log.info("Proxy created")
         this.client = client

tools/loadtest/src/main/kotlin/net/corda/loadtest/LoadTest.kt

@@ -163,7 +163,9 @@ fun runLoadTests(configuration: LoadTestConfiguration, tests: List<Pair<LoadTest
             configuration.nodeHosts.map { it to configuration.remoteNodeDirectory / "certificates" },
             configuration.remoteMessagingPort,
             PortAllocation.Incremental(configuration.localTunnelStartingPort),
-            configuration.localCertificatesBaseDirectory
+            configuration.localCertificatesBaseDirectory,
+            configuration.rpcUsername,
+            configuration.rpcPassword
     ) { connections ->
         log.info("Connected to all nodes!")
         val hostNodeHandleMap = ConcurrentHashMap<String, NodeHandle>()
@@ -175,6 +177,8 @@ fun runLoadTests(configuration: LoadTestConfiguration, tests: List<Pair<LoadTest
             val pubkeysString = otherNodeInfos.map {
                 "    ${it.legalIdentity.name}: ${it.legalIdentity.owningKey.toBase58String()}"
             }.joinToString("\n")
+            log.info("${connection.hostName} waiting for network map")
+            connection.proxy.waitUntilRegisteredWithNetworkMap().get()
             log.info("${connection.hostName} sees\n$pubkeysString")
             val nodeHandle = NodeHandle(configuration, connection, nodeInfo)
             nodeHandle.waitUntilUp()

tools/loadtest/src/main/kotlin/net/corda/loadtest/LoadTestConfiguration.kt

@@ -20,6 +20,8 @@ data class LoadTestConfiguration(
         val localCertificatesBaseDirectory: Path,
         val localTunnelStartingPort: Int,
         val nodeHosts: List<String>,
+        val rpcUsername: String,
+        val rpcPassword: String,
         val remoteNodeDirectory: Path,
         val remoteMessagingPort: Int,
         val remoteSystemdServiceName: String,

tools/loadtest/src/main/kotlin/net/corda/loadtest/Main.kt

@@ -57,9 +57,11 @@ fun main(args: Array<String>) {
             localCertificatesBaseDirectory = Paths.get(resolvedConfig.getString("localCertificatesBaseDirectory")),
             localTunnelStartingPort = resolvedConfig.getInt("localTunnelStartingPort"),
             nodeHosts = resolvedConfig.getStringList("nodeHosts"),
-            remoteNodeDirectory = Paths.get("/opt/r3cev"),
-            remoteMessagingPort = 31337,
-            remoteSystemdServiceName = "r3cev-node",
+            rpcUsername = "corda",
+            rpcPassword = "rgb",
+            remoteNodeDirectory = Paths.get("/opt/corda"),
+            remoteMessagingPort = 10002,
+            remoteSystemdServiceName = "corda",
             seed = if (resolvedConfig.hasPath("seed")) resolvedConfig.getLong("seed") else null
     )
 
@@ -99,7 +101,7 @@ fun main(args: Array<String>) {
             crossCashTest to LoadTest.RunParameters(
                     parallelism = 4,
                     generateCount = 2000,
-                    clearDatabaseBeforeRun = true,
+                    clearDatabaseBeforeRun = false,
                     gatherFrequency = 10,
                     disruptionPatterns = listOf(
                             listOf(),

tools/loadtest/src/main/kotlin/net/corda/loadtest/tests/CrossCashTest.kt

@@ -3,16 +3,15 @@ package net.corda.loadtest.tests
 import net.corda.client.mock.Generator
 import net.corda.client.mock.pickN
 import net.corda.contracts.asset.Cash
+import net.corda.core.*
 import net.corda.core.contracts.Issued
 import net.corda.core.contracts.PartyAndReference
 import net.corda.core.contracts.USD
 import net.corda.core.crypto.AbstractParty
 import net.corda.core.crypto.AnonymousParty
 import net.corda.core.flows.FlowException
-import net.corda.core.getOrThrow
 import net.corda.core.messaging.startFlow
 import net.corda.core.serialization.OpaqueBytes
-import net.corda.core.toFuture
 import net.corda.flows.CashException
 import net.corda.flows.CashFlowCommand
 import net.corda.loadtest.LoadTest
@@ -208,11 +207,12 @@ val crossCashTest = LoadTest<CrossCashCommand, CrossCashState>(
         },
 
         execute = { command ->
-            try {
-                val result = command.command.startFlow(command.node.connection.proxy).returnValue.getOrThrow()
-                log.info("Success: $result")
-            } catch (e: FlowException) {
-                log.error("Failure", e)
+            val result = command.command.startFlow(command.node.connection.proxy).returnValue
+            result.failure {
+                log.error("Failure[$command]", it)
+            }
+            result.success {
+                log.info("Success[$command]: $result")
             }
         },