ExternalVendorCode/corda
1
0
Fork 0
mirror of https://github.com/corda/corda.git synced 2025-01-18 02:39:51 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

ENT-3484 - Added periodic log.warn message to remind that the node has been set into draining mode. (#5151)

Browse Source 
* Added periodic message logged from a timer which is switched on and off by the node drain events.

* Timer also spawns on start() to ensure that if a node is shutdown in draining mode and started with it, it will continue logging messages.
This commit is contained in:
Stefan Iliev 2019-06-03 11:49:29 +01:00 committed by Andrius Dagys
parent d6ea66d0ab
commit 3b7f0aff92
2 changed files with 24 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
node/src/integration-test/kotlin/net/corda/node/modes/draining/P2PFlowsDrainingModeTest.kt
View File

@ -7,7 +7,9 @@ import net.corda.core.internal.concurrent.map
import net.corda.core.messaging.startFlow
import net.corda.core.utilities.contextLogger
import net.corda.core.utilities.getOrThrow
import net.corda.core.utilities.seconds
import net.corda.core.utilities.unwrap
import net.corda.node.logging.logFile
import net.corda.node.services.Permissions
import net.corda.nodeapi.internal.hasCancelledDrainingShutdown
import net.corda.testing.core.ALICE_NAME
@ -19,6 +21,7 @@ import net.corda.testing.driver.internal.incrementalPortAllocation
import net.corda.testing.internal.chooseIdentity
import net.corda.testing.node.User
import net.corda.testing.node.internal.waitForShutdown
import org.assertj.core.api.Assertions
import org.assertj.core.api.AssertionsForInterfaceTypes.assertThat
import org.junit.After
import org.junit.Before

25
node/src/main/kotlin/net/corda/node/services/messaging/P2PMessagingClient.kt
View File

@ -56,6 +56,7 @@ import java.util.*
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.CountDownLatch
import javax.annotation.concurrent.ThreadSafe
import kotlin.concurrent.timer
/**
* This class implements the [MessagingService] API using Apache Artemis, the successor to their ActiveMQ product.
@ -193,7 +194,7 @@ class P2PMessagingClient(val config: NodeConfiguration,
inboxes.forEach { createQueueIfAbsent(it, producerSession!!, exclusive = true) }
p2pConsumer = P2PMessagingConsumer(inboxes, createNewSession, isDrainingModeOn, drainingModeWasChangedEvents)
p2pConsumer = P2PMessagingConsumer(inboxes, createNewSession, isDrainingModeOn, drainingModeWasChangedEvents, metricRegistry)
messagingExecutor = MessagingExecutor(
executorSession!!,
@ -573,10 +574,12 @@ private class P2PMessagingConsumer(
queueNames: Set<String>,
createSession: () -> ClientSession,
private val isDrainingModeOn: () -> Boolean,
private val drainingModeWasChangedEvents: Observable<Pair<Boolean, Boolean>>) : LifecycleSupport {
private val drainingModeWasChangedEvents: Observable<Pair<Boolean, Boolean>>,
private val metricsRegistry : MetricRegistry) : LifecycleSupport {
private companion object {
private const val initialSessionMessages = "${P2PMessagingHeaders.Type.KEY}<>'${P2PMessagingHeaders.Type.SESSION_INIT_VALUE}'"
private val logger by lazy { loggerFor<P2PMessagingClient>() }
}
private var startedFlag = false
@ -587,16 +590,30 @@ private class P2PMessagingConsumer(
private val initialAndExistingConsumer = multiplex(queueNames, createSession)
private val subscriptions = mutableSetOf<Subscription>()
private var notificationTimer : Timer? = null
private fun scheduleDrainNotificationTimer() {
notificationTimer = timer("DrainNotificationTimer", true, 10.seconds.toMillis(), 1.minutes.toMillis()) {
logger.warn("Node is currently in draining mode, new flows will not be processed! Flows in flight: ${metricsRegistry.gauges["Flows.InFlight"]?.value}")
}
}
override fun start() {
synchronized(this) {
require(!startedFlag){"Must not already be started"}
drainingModeWasChangedEvents.filter { change -> change.switchedOn() }.doOnNext { initialAndExistingConsumer.switchTo(existingOnlyConsumer) }.subscribe()
drainingModeWasChangedEvents.filter { change -> change.switchedOff() }.doOnNext { existingOnlyConsumer.switchTo(initialAndExistingConsumer) }.subscribe()
drainingModeWasChangedEvents.filter { change -> change.switchedOn() }.doOnNext {
initialAndExistingConsumer.switchTo(existingOnlyConsumer)
scheduleDrainNotificationTimer()
}.subscribe()
drainingModeWasChangedEvents.filter { change -> change.switchedOff() }.doOnNext {
existingOnlyConsumer.switchTo(initialAndExistingConsumer)
notificationTimer?.cancel()
}.subscribe()
subscriptions += existingOnlyConsumer.messages.doOnNext(messages::onNext).subscribe()
subscriptions += initialAndExistingConsumer.messages.doOnNext(messages::onNext).subscribe()
if (isDrainingModeOn()) {
existingOnlyConsumer.start()
scheduleDrainNotificationTimer()
} else {
initialAndExistingConsumer.start()
}