Merge pull request #7201 from corda/cc/ENT-6854/node_status_jmx

ENT-6854 - node status published via JMX

node-api/src/main/kotlin/net/corda/nodeapi/internal/NodeStatus.kt

@@ -0,0 +1,8 @@
+package net.corda.nodeapi.internal
+
+enum class NodeStatus {
+    WAITING_TO_START,
+    STARTING,
+    STARTED,
+    STOPPING
+}

node/src/integration-test/kotlin/net/corda/node/jmx/NodeStatusTest.kt

@@ -0,0 +1,42 @@
+package net.corda.node.jmx
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import net.corda.nodeapi.internal.NodeStatus
+import net.corda.testing.driver.DriverParameters
+import net.corda.testing.driver.JmxPolicy
+import net.corda.testing.driver.driver
+import org.junit.Test
+import java.net.HttpURLConnection
+import java.net.HttpURLConnection.HTTP_OK
+import java.net.URL
+import java.util.stream.Collectors
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+class NodeStatusTest {
+
+    @Test(timeout=300_000)
+    fun `node status is published via JMX`() {
+        driver(DriverParameters(notarySpecs = emptyList(), jmxPolicy = JmxPolicy.defaultEnabled())) {
+            val jmxAddress = startNode().get().jmxAddress.toString()
+            val nodeStatusURL = URL("http://$jmxAddress/jolokia/read/net.corda:name=Status,type=Node")
+            val jmxInfo = with(nodeStatusURL.openConnection() as HttpURLConnection) {
+                requestMethod = "GET"
+                inputStream.bufferedReader().use {
+                    it.lines().collect(Collectors.toList()).joinToString()
+                }
+            }
+
+            assertTrue {
+                jmxInfo.isNotEmpty()
+            }
+
+            val jsonTree = ObjectMapper().readTree(jmxInfo)
+            val httpStatus = jsonTree.get("status").asInt()
+            val nodeStatus = jsonTree.get("value").get("Value").asText()
+
+            assertEquals(httpStatus, HTTP_OK)
+            assertEquals(nodeStatus, NodeStatus.STARTED.toString())
+        }
+    }
+}

node/src/integration-test/kotlin/net/corda/node/jmx/PublishTest.kt

@@ -0,0 +1,28 @@
+package net.corda.node.jmx
+
+import net.corda.testing.driver.DriverParameters
+import net.corda.testing.driver.JmxPolicy
+import net.corda.testing.driver.driver
+import org.junit.Test
+import java.net.HttpURLConnection
+import java.net.URL
+import kotlin.test.assertTrue
+
+class PublishTest {
+
+    @Test(timeout=300_000)
+    fun `node publishes node information via JMX when configured to do so`() {
+        driver(DriverParameters(notarySpecs = emptyList(), jmxPolicy = JmxPolicy.defaultEnabled())) {
+            val jmxAddress = startNode().get().jmxAddress.toString()
+            val nodeStatusURL = URL("http://$jmxAddress/jolokia/read/net.corda:*")
+            val httpResponse = with(nodeStatusURL.openConnection() as HttpURLConnection) {
+                requestMethod = "GET"
+                responseCode
+            }
+
+            assertTrue {
+                httpResponse == HttpURLConnection.HTTP_OK
+            }
+        }
+    }
+}

node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt

@@ -1,6 +1,7 @@
 package net.corda.node.internal
 
 import co.paralleluniverse.fibers.instrument.Retransform
+import com.codahale.metrics.Gauge
 import com.codahale.metrics.MetricRegistry
 import com.google.common.collect.MutableClassToInstanceMap
 import com.google.common.util.concurrent.MoreExecutors
@@ -146,6 +147,7 @@ import net.corda.node.utilities.BindableNamedCacheFactory
 import net.corda.node.utilities.NamedThreadFactory
 import net.corda.node.utilities.NotaryLoader
 import net.corda.nodeapi.internal.NodeInfoAndSigned
+import net.corda.nodeapi.internal.NodeStatus
 import net.corda.nodeapi.internal.SignedNodeInfo
 import net.corda.nodeapi.internal.cordapp.CordappLoader
 import net.corda.nodeapi.internal.cryptoservice.CryptoService
@@ -383,6 +385,9 @@ abstract class AbstractNode<S>(val configuration: NodeConfiguration,
 
     protected val keyStoreHandler = KeyStoreHandler(configuration, cryptoService)
 
+    @Volatile
+    private var nodeStatus = NodeStatus.WAITING_TO_START
+
     private fun <T : Any> T.tokenize(): T {
         tokenizableServices?.add(this as? SerializeAsToken ?:
             throw IllegalStateException("${this::class.java} is expected to be extending from SerializeAsToken"))
@@ -524,6 +529,12 @@ abstract class AbstractNode<S>(val configuration: NodeConfiguration,
         Node.printBasicNodeInfo("CorDapp schemas synchronised")
     }
 
+    private fun setNodeStatus(st : NodeStatus) {
+        log.info("Node status update: [$nodeStatus] -> [$st]")
+        nodeStatus = st
+    }
+
+
     @Suppress("ComplexMethod")
     open fun start(): S {
         check(started == null) { "Node has already been started" }
@@ -533,9 +544,12 @@ abstract class AbstractNode<S>(val configuration: NodeConfiguration,
         }
         nodeLifecycleEventsDistributor.distributeEvent(NodeLifecycleEvent.BeforeNodeStart(nodeServicesContext))
         log.info("Node starting up ...")
+        setNodeStatus(NodeStatus.STARTING)
+
+        initialiseJolokia()
+        monitoringService.metrics.register(MetricRegistry.name("Node", "Status"), Gauge { nodeStatus })
 
         val trustRoots = initKeyStores()
-        initialiseJolokia()
 
         schemaService.mappedSchemasWarnings().forEach {
             val warning = it.toWarning()
@@ -658,6 +672,7 @@ abstract class AbstractNode<S>(val configuration: NodeConfiguration,
                 log.warn("Not distributing events as NetworkMap is not ready")
             }
         }
+        setNodeStatus(NodeStatus.STARTED)
         return resultingNodeInfo
     }
 
@@ -1050,6 +1065,8 @@ abstract class AbstractNode<S>(val configuration: NodeConfiguration,
 
     open fun stop() {
 
+        setNodeStatus(NodeStatus.STOPPING)
+
         nodeLifecycleEventsDistributor.distributeEvent(NodeLifecycleEvent.StateMachineStopped(nodeServicesContext))
         nodeLifecycleEventsDistributor.distributeEvent(NodeLifecycleEvent.BeforeNodeStop(nodeServicesContext))
 

node/src/main/kotlin/net/corda/node/internal/Node.kt

@@ -580,14 +580,12 @@ open class Node(configuration: NodeConfiguration,
     }
 
     override fun start(): NodeInfo {
+        registerJmxReporter(services.monitoringService.metrics)
         registerDefaultExceptionHandler()
         initialiseSerialization()
         val nodeInfo: NodeInfo = super.start()
         nodeReadyFuture.thenMatch({
             serverThread.execute {
-
-                registerJmxReporter(services.monitoringService.metrics)
-
                 _startupComplete.set(Unit)
             }
         },