Update notary health check tool to test each replica individually

tools/notaryhealthcheck/build.gradle

@@ -45,17 +45,17 @@ publishing {
 
 task runTest(type: JavaExec) {
     classpath = sourceSets.main.runtimeClasspath
-    main = 'net.corda.notarytest.MainKt'
+    main = 'net.corda.notaryhealthcheck.MainKt'
 }
 
-task deployHealthCheck(type: Cordform, dependsOn: 'jar') {
-    definitionClass = 'net.corda.notarytest.HealthCheckCordform'
+task deployNodes(type: Cordform, dependsOn: 'jar') {
+    definitionClass = 'net.corda.notaryhealthcheck.HealthCheckCordform'
 }
 
 jar {
     manifest {
         attributes(
-                'Automatic-Module-Name': 'net.corda.notarytest'
+                'Automatic-Module-Name': 'net.corda.notaryhealthcheck'
         )
     }
 }

tools/notaryhealthcheck/src/main/kotlin/net/corda/notaryhealthcheck/HealthCheckCordform.kt

@@ -0,0 +1,58 @@
+package net.corda.notaryhealthcheck
+
+import net.corda.cordform.CordformContext
+import net.corda.cordform.CordformDefinition
+import net.corda.cordform.CordformNode
+import net.corda.core.identity.CordaX500Name
+import net.corda.core.utilities.NetworkHostAndPort
+import net.corda.node.services.Permissions.Companion.all
+import net.corda.node.services.config.NotaryConfig
+import net.corda.node.services.config.RaftConfig
+import net.corda.nodeapi.internal.DevIdentityGenerator
+import net.corda.testing.node.User
+import net.corda.testing.node.internal.demorun.*
+import java.nio.file.Paths
+
+fun main(args: Array<String>) = HealthCheckCordform().deployNodes()
+
+class HealthCheckCordform : CordformDefinition() {
+    private fun createNotaryNames(clusterSize: Int) = (0 until clusterSize).map { CordaX500Name("Notary Service $it", "Zurich", "CH") }
+    private val notaryDemoUser = User("demou", "demop", setOf(all()))
+    private val notaryNames = createNotaryNames(3)
+    private val clusterName = CordaX500Name("Raft", "Zurich", "CH")
+
+    init {
+        nodesDirectory = Paths.get("build", "nodes")
+        fun notaryNode(index: Int, nodePort: Int, clusterPort: Int? = null, configure: CordformNode.() -> Unit) = node {
+            name(notaryNames[index])
+            val clusterAddresses = if (clusterPort != null) listOf(NetworkHostAndPort("localhost", clusterPort)) else emptyList()
+            notary(NotaryConfig(validating = true, raft = RaftConfig(NetworkHostAndPort("localhost", nodePort), clusterAddresses)))
+            configure()
+        }
+        notaryNode(0, 10008) {
+            p2pPort(10009)
+            rpcPort(10010)
+        }
+        notaryNode(1, 10012, 10008) {
+            p2pPort(10013)
+            rpcPort(10014)
+        }
+        notaryNode(2, 10016, 10008) {
+            p2pPort(10017)
+            rpcPort(10018)
+        }
+        node {
+            name(CordaX500Name("R3 Notary Health Check", "London", "GB"))
+            p2pPort(10002)
+            rpcPort(10003)
+            rpcUsers(notaryDemoUser)
+        }
+    }
+
+    override fun setup(context: CordformContext) {
+        DevIdentityGenerator.generateDistributedNotarySingularIdentity(
+                notaryNames.map { context.baseDirectory(it.toString()) },
+                clusterName
+        )
+    }
+}

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

@@ -1,4 +1,4 @@
-package net.corda.notarytest
+package net.corda.notaryhealthcheck
 
 import net.corda.client.rpc.CordaRPCClient
 import net.corda.core.messaging.CordaRPCOps
@@ -6,21 +6,22 @@ import net.corda.core.messaging.startFlow
 import net.corda.core.utilities.NetworkHostAndPort
 import net.corda.nodeapi.internal.config.User
 import net.corda.node.services.Permissions
-import net.corda.notarytest.flows.HealthCheckFlow
+import net.corda.notaryhealthcheck.flows.HealthCheckFlow
 
 fun main(args: Array<String>) {
     val addresses = listOf(NetworkHostAndPort("localhost", 10003))
-
     val notaryDemoUser = User("demou", "demop", setOf(Permissions.all()))
 
     addresses.parallelStream().forEach {
        val c = CordaRPCClient(it).start(notaryDemoUser.username, notaryDemoUser.password)
        healthCheck(c.proxy)
     }
-    println("ok")
+    println("Health check complete.")
 }
 
 fun healthCheck(rpc: CordaRPCOps) {
     val notary = rpc.notaryIdentities().first()
-    rpc.startFlow(::HealthCheckFlow, notary).returnValue.get()
+    print("Running health check for notary cluster ${notary.name}... ")
+    rpc.startFlow(::HealthCheckFlow, notary, true).returnValue.get()
+    println("Done.")
 }

tools/notaryhealthcheck/src/main/kotlin/net/corda/notaryhealthcheck/flows/HealthCheckFlow.kt

@@ -1,11 +1,10 @@
-package net.corda.notarytest.flows
+package net.corda.notaryhealthcheck.flows
 
 import co.paralleluniverse.fibers.Suspendable
 import net.corda.core.contracts.CommandData
 import net.corda.core.contracts.Contract
 import net.corda.core.contracts.ContractState
 import net.corda.core.flows.FlowLogic
-import net.corda.core.flows.NotaryFlow
 import net.corda.core.flows.StartableByRPC
 import net.corda.core.identity.AbstractParty
 import net.corda.core.identity.Party
@@ -13,26 +12,21 @@ import net.corda.core.transactions.LedgerTransaction
 import net.corda.core.transactions.TransactionBuilder
 
 @StartableByRPC
-open class HealthCheckFlow(val notaryParty: Party) : FlowLogic<Unit>() {
-
+open class HealthCheckFlow(val notaryParty: Party, val checkEntireCluster: Boolean = false) : FlowLogic<Unit>() {
     class DoNothingContract : Contract {
         override fun verify(tx: LedgerTransaction) {}
     }
 
     data class DummyCommand(val dummy: Int = 0) : CommandData
+    data class State(override val participants: List<AbstractParty>) : ContractState
 
     @Suspendable
-    override fun call(): Unit {
-
-        data class State(override val participants: List<AbstractParty>) : ContractState
-
+    override fun call() {
         val state = State(listOf(ourIdentity))
-
-         val stx = serviceHub.signInitialTransaction(TransactionBuilder(notaryParty).apply {
-            addOutputState(state, "net.corda.notarytest.flows.HealthCheckFlow\$DoNothingContract")
+        val stx = serviceHub.signInitialTransaction(TransactionBuilder(notaryParty).apply {
+            addOutputState(state, "net.corda.notaryhealthcheck.flows.HealthCheckFlow\$DoNothingContract")
             addCommand(DummyCommand(), listOf(ourIdentity.owningKey))
         })
-
-        subFlow(NotaryFlow.Client(stx))
+        subFlow(HealthCheckNotaryClientFlow(stx, checkEntireCluster = checkEntireCluster))
     }
 }

tools/notaryhealthcheck/src/main/kotlin/net/corda/notaryhealthcheck/flows/HealthCheckNotaryClientFlow.kt

@@ -0,0 +1,56 @@
+package net.corda.notaryhealthcheck.flows
+
+import co.paralleluniverse.fibers.Suspendable
+import net.corda.core.crypto.TransactionSignature
+import net.corda.core.flows.NotaryError
+import net.corda.core.flows.NotaryException
+import net.corda.core.flows.NotaryFlow
+import net.corda.core.transactions.SignedTransaction
+
+/**
+ * Notarises the provided transaction. If [checkEntireCluster] is set to *true*, will repeat the notarisation request
+ * to each member of the notary cluster.
+ */
+class HealthCheckNotaryClientFlow(
+        stx: SignedTransaction,
+        /**
+         * If set to *true*, will issue a notarisation request to each replica in the notary cluster,
+         * rather than sending the request to one replica only.
+         */
+        private val checkEntireCluster: Boolean = false
+) : NotaryFlow.Client(stx) {
+    @Suspendable
+    @Throws(NotaryException::class)
+    override fun call(): List<TransactionSignature> {
+        progressTracker.currentStep = REQUESTING
+        val notaryParty = checkTransaction()
+
+        val parties = if (checkEntireCluster) {
+            serviceHub.networkMapCache
+                    .getNodesByLegalIdentityKey(notaryParty.owningKey)
+                    .map { it.legalIdentities.first() }
+        } else {
+            listOf(notaryParty)
+        }
+        var signatures: List<TransactionSignature> = emptyList()
+        parties.forEach { nodeLegalIdentity ->
+            logger.info("Sending notarisation request to: $nodeLegalIdentity")
+            val response = try {
+                val session = initiateFlow(nodeLegalIdentity)
+                if (serviceHub.networkMapCache.isValidatingNotary(notaryParty)) {
+                    sendAndReceiveValidating(session)
+                } else {
+                    sendAndReceiveNonValidating(nodeLegalIdentity, session)
+                }
+            } catch (e: NotaryException) {
+                if (e.error is NotaryError.Conflict) {
+                    (e.error as NotaryError.Conflict).conflict.verified()
+                }
+                throw e
+            }
+            signatures = validateResponse(response, notaryParty)
+            logger.info("Received a valid signature from $nodeLegalIdentity, signed by $notaryParty")
+        }
+        return signatures
+    }
+}

tools/notaryhealthcheck/src/main/kotlin/net/corda/notarytest/HealthCheckCordform.kt

@@ -1,44 +0,0 @@
-package net.corda.notarytest
-
-import net.corda.cordform.CordformContext
-import net.corda.cordform.CordformDefinition
-import net.corda.core.identity.CordaX500Name
-import net.corda.node.services.Permissions.Companion.all
-import net.corda.node.services.config.NotaryConfig
-import net.corda.testing.node.User
-import net.corda.testing.node.internal.demorun.*
-import java.nio.file.Paths
-
-fun main(args: Array<String>) = HealthCheckCordform().deployNodes()
-
-val notaryDemoUser = User("demou", "demop", setOf(all()))
-
-// This is not the intended final design for how to use CordformDefinition, please treat this as experimental and DO
-// NOT use this as a design to copy.
-class HealthCheckCordform : CordformDefinition() {
-    init {
-        nodesDirectory = Paths.get("build", "nodes")
-        node {
-            name(CordaX500Name("R3 Notary Health Check", "London", "GB"))
-            p2pPort(10002)
-            rpcPort(10003)
-            rpcUsers(notaryDemoUser)
-        }
-        node {
-            name(CordaX500Name("Notary Service 0", "London", "GB"))
-            p2pPort(10009)
-            rpcPort(10010)
-            notary(NotaryConfig(validating = false))
-            extraConfig = mapOf(
-                    "mysql" to mapOf(
-                            "jdbcUrl" to "jdbc:mysql://notary-10.northeurope.cloudapp.azure.com:3306/corda?rewriteBatchedStatements=true&useSSL=false",
-                            "username" to "",
-                            "password" to "",
-                            "autoCommit" to "false"
-                    )
-            )
-        }
-    }
-
-    override fun setup(context: CordformContext) {}
-}