Run DTLS session lifecycle probes asynchronously (#25)

akolosov-n · web-flow · commit 5fc13c8e9fac · 2023-08-01T10:28:13.000+03:00
* Call metrics callbacks asynchronously

* Fixing tests

* Set finish timestamp on handshake failure
diff --git a/kotlin-mbedtls-metrics/src/main/kotlin/org/opencoap/ssl/transport/metrics/micrometer/DtlsServerMetricsCallbacks.kt b/kotlin-mbedtls-metrics/src/main/kotlin/org/opencoap/ssl/transport/metrics/micrometer/DtlsServerMetricsCallbacks.kt
@@ -54,10 +54,10 @@ class DtlsServerMetricsCallbacks(
         handshakesInitiated.increment()
     }
 
-    override fun handshakeFinished(adr: InetSocketAddress, hanshakeStartTimestamp: Long, reason: DtlsSessionLifecycleCallbacks.Reason, throwable: Throwable?) = when {
+    override fun handshakeFinished(adr: InetSocketAddress, hanshakeStartTimestamp: Long, hanshakeFinishTimestamp: Long, reason: DtlsSessionLifecycleCallbacks.Reason, throwable: Throwable?) = when {
         throwable is HelloVerifyRequired -> {} // Skip HelloVerifyRequired handshake states
         reason == DtlsSessionLifecycleCallbacks.Reason.SUCCEEDED ->
-            handshakesSucceeded.record(System.currentTimeMillis() - hanshakeStartTimestamp, TimeUnit.MILLISECONDS)
+            handshakesSucceeded.record(hanshakeFinishTimestamp - hanshakeStartTimestamp, TimeUnit.MILLISECONDS)
         reason == DtlsSessionLifecycleCallbacks.Reason.FAILED ->
             handshakesFailedBuilder.reasonTag(throwable).register(registry).increment()
         reason == DtlsSessionLifecycleCallbacks.Reason.EXPIRED ->
diff --git a/kotlin-mbedtls/src/main/kotlin/org/opencoap/ssl/SslContext.kt b/kotlin-mbedtls/src/main/kotlin/org/opencoap/ssl/SslContext.kt
@@ -68,6 +68,7 @@ class SslHandshakeContext internal constructor(
 ) : SslContext {
     private val logger = LoggerFactory.getLogger(javaClass)
     val startTimestamp: Long = System.currentTimeMillis()
+    var finishTimestamp: Long = 0
     private var stepTimeout: Duration = Duration.ZERO
 
     fun step(send: (ByteBuffer) -> Unit): SslContext = step0(null, send)
@@ -86,10 +87,12 @@ class SslHandshakeContext internal constructor(
             MbedtlsApi.MBEDTLS_ERR_SSL_WANT_READ -> return this
             MbedtlsApi.MBEDTLS_ERR_SSL_HELLO_VERIFY_REQUIRED -> throw HelloVerifyRequired
             0 -> SslSession(conf, sslContext, cid).also {
-                logger.info("[{}] DTLS connected in {}ms {}", peerAdr, System.currentTimeMillis() - startTimestamp, it)
+                finishTimestamp = System.currentTimeMillis()
+                logger.info("[{}] DTLS connected in {}ms {}", peerAdr, finishTimestamp - startTimestamp, it)
             }
 
             else -> throw SslException.from(ret).also {
+                finishTimestamp = System.currentTimeMillis()
                 logger.debug("[{}] DTLS failed handshake: {}", peerAdr, it.message)
             }
         }
diff --git a/kotlin-mbedtls/src/main/kotlin/org/opencoap/ssl/transport/DtlsServer.kt b/kotlin-mbedtls/src/main/kotlin/org/opencoap/ssl/transport/DtlsServer.kt
@@ -172,8 +172,8 @@ class DtlsServer(
                     }
 
                     is SslSession -> {
-                        reportHandshakeFinished(DtlsSessionLifecycleCallbacks.Reason.SUCCEEDED)
                         sessions[peerAddress] = DtlsSession(newCtx, peerAddress)
+                        reportHandshakeFinished(DtlsSessionLifecycleCallbacks.Reason.SUCCEEDED)
                     }
                 }
             } catch (ex: Exception) {
@@ -185,28 +185,28 @@ class DtlsServer(
                     else ->
                         logger.error(ex.toString(), ex)
                 }
-                reportHandshakeFinished(DtlsSessionLifecycleCallbacks.Reason.FAILED, ex)
                 closeAndRemove()
+                reportHandshakeFinished(DtlsSessionLifecycleCallbacks.Reason.FAILED, ex)
             }
             return ReceiveResult.Handled
         }
 
         fun timeout() {
-            reportHandshakeFinished(DtlsSessionLifecycleCallbacks.Reason.EXPIRED)
             closeAndRemove()
             logger.warn("[{}] DTLS handshake expired", peerAddress)
+            reportHandshakeFinished(DtlsSessionLifecycleCallbacks.Reason.EXPIRED)
         }
 
         override fun storeAndClose0() = close()
 
         override fun close() = ctx.close()
 
         private fun reportHandshakeStarted() {
-            lifecycleCallbacks.handshakeStarted(peerAddress)
+            executor.supply { lifecycleCallbacks.handshakeStarted(peerAddress) }
         }
 
         private fun reportHandshakeFinished(reason: DtlsSessionLifecycleCallbacks.Reason, err: Throwable? = null) {
-            lifecycleCallbacks.handshakeFinished(peerAddress, ctx.startTimestamp, reason, err)
+            executor.supply { lifecycleCallbacks.handshakeFinished(peerAddress, ctx.startTimestamp, ctx.finishTimestamp, reason, err) }
         }
     }
 
@@ -272,26 +272,26 @@ class DtlsServer(
             try {
                 return ctx.encrypt(plainPacket)
             } catch (ex: SslException) {
+                closeAndRemove()
                 logger.warn("[{}] DTLS failed: {}", peerAddress, ex.message)
                 reportSessionFinished(DtlsSessionLifecycleCallbacks.Reason.FAILED, ex)
-                closeAndRemove()
                 throw ex
             }
         }
 
         fun timeout() {
-            lifecycleCallbacks.sessionFinished(peerAddress, DtlsSessionLifecycleCallbacks.Reason.EXPIRED)
             sessions.remove(peerAddress, this)
-            logger.info("[{}] DTLS connection expired", peerAddress)
             storeAndClose()
+            logger.info("[{}] DTLS connection expired", peerAddress)
+            lifecycleCallbacks.sessionFinished(peerAddress, DtlsSessionLifecycleCallbacks.Reason.EXPIRED)
         }
 
         private fun reportSessionStarted() {
-            lifecycleCallbacks.sessionStarted(peerAddress, ctx.cipherSuite, ctx.reloaded)
+            executor.supply { lifecycleCallbacks.sessionStarted(peerAddress, ctx.cipherSuite, ctx.reloaded) }
         }
 
         private fun reportSessionFinished(reason: DtlsSessionLifecycleCallbacks.Reason, err: Throwable? = null) {
-            lifecycleCallbacks.sessionFinished(peerAddress, reason, err)
+            executor.supply { lifecycleCallbacks.sessionFinished(peerAddress, reason, err) }
         }
     }
 }
diff --git a/kotlin-mbedtls/src/main/kotlin/org/opencoap/ssl/transport/DtlsSessionLifecycleCallbacks.kt b/kotlin-mbedtls/src/main/kotlin/org/opencoap/ssl/transport/DtlsSessionLifecycleCallbacks.kt
@@ -23,7 +23,7 @@ interface DtlsSessionLifecycleCallbacks {
         SUCCEEDED, FAILED, CLOSED, EXPIRED
     }
     fun handshakeStarted(adr: InetSocketAddress) = Unit
-    fun handshakeFinished(adr: InetSocketAddress, hanshakeStartTimestamp: Long, reason: Reason, throwable: Throwable? = null) = Unit
+    fun handshakeFinished(adr: InetSocketAddress, hanshakeStartTimestamp: Long, hanshakeFinishTimestamp: Long, reason: Reason, throwable: Throwable? = null) = Unit
     fun sessionStarted(adr: InetSocketAddress, cipherSuite: String, reloaded: Boolean) = Unit
     fun sessionFinished(adr: InetSocketAddress, reason: Reason, throwable: Throwable? = null) = Unit
 }
diff --git a/kotlin-mbedtls/src/test/kotlin/org/opencoap/ssl/transport/DtlsServerTransportTest.kt b/kotlin-mbedtls/src/test/kotlin/org/opencoap/ssl/transport/DtlsServerTransportTest.kt
@@ -20,7 +20,6 @@ import io.mockk.clearMocks
 import io.mockk.confirmVerified
 import io.mockk.mockk
 import io.mockk.verify
-import io.mockk.verifyOrder
 import org.awaitility.kotlin.await
 import org.junit.jupiter.api.AfterEach
 import org.junit.jupiter.api.Assertions.assertEquals
@@ -113,11 +112,11 @@ class DtlsServerTransportTest {
         val clientAddress = client.localAddress()
         client.close()
 
-        verifyOrder {
+        verify {
             sslLifecycleCallbacks.handshakeStarted(clientAddress)
-            sslLifecycleCallbacks.handshakeFinished(clientAddress, any(), DtlsSessionLifecycleCallbacks.Reason.FAILED, ofType(HelloVerifyRequired::class))
+            sslLifecycleCallbacks.handshakeFinished(clientAddress, any(), any(), DtlsSessionLifecycleCallbacks.Reason.FAILED, ofType(HelloVerifyRequired::class))
             sslLifecycleCallbacks.handshakeStarted(clientAddress)
-            sslLifecycleCallbacks.handshakeFinished(clientAddress, any(), DtlsSessionLifecycleCallbacks.Reason.SUCCEEDED)
+            sslLifecycleCallbacks.handshakeFinished(clientAddress, any(), any(), DtlsSessionLifecycleCallbacks.Reason.SUCCEEDED)
             sslLifecycleCallbacks.sessionStarted(clientAddress, any(), false)
         }
 
@@ -165,11 +164,11 @@ class DtlsServerTransportTest {
             assertEquals(0, server.numberOfSessions())
         }
         assertFalse(srvReceive.isDone)
-        verifyOrder {
+        verify {
             sslLifecycleCallbacks.handshakeStarted(any())
-            sslLifecycleCallbacks.handshakeFinished(any(), any(), DtlsSessionLifecycleCallbacks.Reason.FAILED, ofType(HelloVerifyRequired::class))
+            sslLifecycleCallbacks.handshakeFinished(any(), any(), any(), DtlsSessionLifecycleCallbacks.Reason.FAILED, ofType(HelloVerifyRequired::class))
             sslLifecycleCallbacks.handshakeStarted(any())
-            sslLifecycleCallbacks.handshakeFinished(any(), any(), DtlsSessionLifecycleCallbacks.Reason.FAILED, ofType(SslException::class))
+            sslLifecycleCallbacks.handshakeFinished(any(), any(), any(), DtlsSessionLifecycleCallbacks.Reason.FAILED, ofType(SslException::class))
         }
 
         verify(exactly = 0) {
@@ -193,9 +192,9 @@ class DtlsServerTransportTest {
 
         client.close()
 
-        verifyOrder {
+        verify {
             sslLifecycleCallbacks.handshakeStarted(any())
-            sslLifecycleCallbacks.handshakeFinished(any(), any(), DtlsSessionLifecycleCallbacks.Reason.SUCCEEDED)
+            sslLifecycleCallbacks.handshakeFinished(any(), any(), any(), DtlsSessionLifecycleCallbacks.Reason.SUCCEEDED)
             sslLifecycleCallbacks.sessionStarted(any(), any(), any())
         }
     }
@@ -237,11 +236,11 @@ class DtlsServerTransportTest {
         cliChannel.close()
 
         verify(atMost = 100) {
-            sslLifecycleCallbacks.handshakeFinished(any(), any(), DtlsSessionLifecycleCallbacks.Reason.FAILED, ofType(SslException::class))
+            sslLifecycleCallbacks.handshakeFinished(any(), any(), any(), DtlsSessionLifecycleCallbacks.Reason.FAILED, ofType(SslException::class))
         }
 
         verify(exactly = 0) {
-            sslLifecycleCallbacks.handshakeFinished(any(), any(), DtlsSessionLifecycleCallbacks.Reason.FAILED, ofType(HelloVerifyRequired::class))
+            sslLifecycleCallbacks.handshakeFinished(any(), any(), any(), DtlsSessionLifecycleCallbacks.Reason.FAILED, ofType(HelloVerifyRequired::class))
         }
     }
 
@@ -286,7 +285,7 @@ class DtlsServerTransportTest {
             assertEquals(0, server.numberOfSessions())
         }
 
-        verifyOrder {
+        verify {
             sslLifecycleCallbacks.sessionStarted(any(), any(), any())
             sslLifecycleCallbacks.sessionFinished(any(), DtlsSessionLifecycleCallbacks.Reason.CLOSED)
         }
@@ -308,12 +307,12 @@ class DtlsServerTransportTest {
 
         // No handshake failures other than HelloVerifyRequired
         verify(exactly = 0) {
-            sslLifecycleCallbacks.handshakeFinished(any(), any(), DtlsSessionLifecycleCallbacks.Reason.FAILED, not(ofType(HelloVerifyRequired::class)))
+            sslLifecycleCallbacks.handshakeFinished(any(), any(), any(), DtlsSessionLifecycleCallbacks.Reason.FAILED, not(ofType(HelloVerifyRequired::class)))
         }
 
         // One successful handshake must happen
         verify(exactly = 1) {
-            sslLifecycleCallbacks.handshakeFinished(any(), any(), DtlsSessionLifecycleCallbacks.Reason.SUCCEEDED)
+            sslLifecycleCallbacks.handshakeFinished(any(), any(), any(), DtlsSessionLifecycleCallbacks.Reason.SUCCEEDED)
         }
     }
 
@@ -335,7 +334,7 @@ class DtlsServerTransportTest {
         cli.close()
 
         verify(exactly = 1) {
-            sslLifecycleCallbacks.handshakeFinished(any(), any(), DtlsSessionLifecycleCallbacks.Reason.FAILED, and(ofType(SslException::class), not(ofType(HelloVerifyRequired::class))))
+            sslLifecycleCallbacks.handshakeFinished(any(), any(), any(), DtlsSessionLifecycleCallbacks.Reason.FAILED, and(ofType(SslException::class), not(ofType(HelloVerifyRequired::class))))
         }
     }
 
@@ -387,11 +386,11 @@ class DtlsServerTransportTest {
         }
         client.close()
 
-        verifyOrder() {
+        verify() {
             sslLifecycleCallbacks.handshakeStarted(any())
-            sslLifecycleCallbacks.handshakeFinished(any(), any(), DtlsSessionLifecycleCallbacks.Reason.FAILED, ofType(HelloVerifyRequired::class))
+            sslLifecycleCallbacks.handshakeFinished(any(), any(), any(), DtlsSessionLifecycleCallbacks.Reason.FAILED, ofType(HelloVerifyRequired::class))
             sslLifecycleCallbacks.handshakeStarted(any())
-            sslLifecycleCallbacks.handshakeFinished(any(), any(), DtlsSessionLifecycleCallbacks.Reason.SUCCEEDED)
+            sslLifecycleCallbacks.handshakeFinished(any(), any(), any(), DtlsSessionLifecycleCallbacks.Reason.SUCCEEDED)
             sslLifecycleCallbacks.sessionStarted(any(), any(), false)
             sslLifecycleCallbacks.sessionFinished(any(), DtlsSessionLifecycleCallbacks.Reason.EXPIRED)
             sslLifecycleCallbacks.sessionStarted(any(), any(), true)

Original file line number	Diff line number	Diff line change
`@@ -68,6 +68,7 @@ class SslHandshakeContext internal constructor(`
`68`	`68`	`) : SslContext {`
`69`	`69`	`private val logger = LoggerFactory.getLogger(javaClass)`
`70`	`70`	`val startTimestamp: Long = System.currentTimeMillis()`
	`71`	`+ var finishTimestamp: Long = 0`
`71`	`72`	`private var stepTimeout: Duration = Duration.ZERO`
`72`	`73`
`73`	`74`	`fun step(send: (ByteBuffer) -> Unit): SslContext = step0(null, send)`
`@@ -86,10 +87,12 @@ class SslHandshakeContext internal constructor(`
`86`	`87`	`MbedtlsApi.MBEDTLS_ERR_SSL_WANT_READ -> return this`
`87`	`88`	`MbedtlsApi.MBEDTLS_ERR_SSL_HELLO_VERIFY_REQUIRED -> throw HelloVerifyRequired`
`88`	`89`	`0 -> SslSession(conf, sslContext, cid).also {`
`89`		`- logger.info("[{}] DTLS connected in {}ms {}", peerAdr, System.currentTimeMillis() - startTimestamp, it)`
	`90`	`+ finishTimestamp = System.currentTimeMillis()`
	`91`	`+ logger.info("[{}] DTLS connected in {}ms {}", peerAdr, finishTimestamp - startTimestamp, it)`
`90`	`92`	`}`
`91`	`93`
`92`	`94`	`else -> throw SslException.from(ret).also {`
	`95`	`+ finishTimestamp = System.currentTimeMillis()`
`93`	`96`	`logger.debug("[{}] DTLS failed handshake: {}", peerAdr, it.message)`
`94`	`97`	`}`
`95`	`98`	`}`
Original file line number	Diff line number	Diff line change
`@@ -172,8 +172,8 @@ class DtlsServer(`
`172`	`172`	`}`
`173`	`173`
`174`	`174`	`is SslSession -> {`
`175`		`- reportHandshakeFinished(DtlsSessionLifecycleCallbacks.Reason.SUCCEEDED)`
`176`	`175`	`sessions[peerAddress] = DtlsSession(newCtx, peerAddress)`
	`176`	`+ reportHandshakeFinished(DtlsSessionLifecycleCallbacks.Reason.SUCCEEDED)`
`177`	`177`	`}`
`178`	`178`	`}`
`179`	`179`	`} catch (ex: Exception) {`
`@@ -185,28 +185,28 @@ class DtlsServer(`
`185`	`185`	`else ->`
`186`	`186`	`logger.error(ex.toString(), ex)`
`187`	`187`	`}`
`188`		`- reportHandshakeFinished(DtlsSessionLifecycleCallbacks.Reason.FAILED, ex)`
`189`	`188`	`closeAndRemove()`
	`189`	`+ reportHandshakeFinished(DtlsSessionLifecycleCallbacks.Reason.FAILED, ex)`
`190`	`190`	`}`
`191`	`191`	`return ReceiveResult.Handled`
`192`	`192`	`}`
`193`	`193`
`194`	`194`	`fun timeout() {`
`195`		`- reportHandshakeFinished(DtlsSessionLifecycleCallbacks.Reason.EXPIRED)`
`196`	`195`	`closeAndRemove()`
`197`	`196`	`logger.warn("[{}] DTLS handshake expired", peerAddress)`
	`197`	`+ reportHandshakeFinished(DtlsSessionLifecycleCallbacks.Reason.EXPIRED)`
`198`	`198`	`}`
`199`	`199`
`200`	`200`	`override fun storeAndClose0() = close()`
`201`	`201`
`202`	`202`	`override fun close() = ctx.close()`
`203`	`203`
`204`	`204`	`private fun reportHandshakeStarted() {`
`205`		`- lifecycleCallbacks.handshakeStarted(peerAddress)`
	`205`	`+ executor.supply { lifecycleCallbacks.handshakeStarted(peerAddress) }`
`206`	`206`	`}`
`207`	`207`
`208`	`208`	`private fun reportHandshakeFinished(reason: DtlsSessionLifecycleCallbacks.Reason, err: Throwable? = null) {`
`209`		`- lifecycleCallbacks.handshakeFinished(peerAddress, ctx.startTimestamp, reason, err)`
	`209`	`+ executor.supply { lifecycleCallbacks.handshakeFinished(peerAddress, ctx.startTimestamp, ctx.finishTimestamp, reason, err) }`
`210`	`210`	`}`
`211`	`211`	`}`
`212`	`212`
`@@ -272,26 +272,26 @@ class DtlsServer(`
`272`	`272`	`try {`
`273`	`273`	`return ctx.encrypt(plainPacket)`
`274`	`274`	`} catch (ex: SslException) {`
	`275`	`+ closeAndRemove()`
`275`	`276`	`logger.warn("[{}] DTLS failed: {}", peerAddress, ex.message)`
`276`	`277`	`reportSessionFinished(DtlsSessionLifecycleCallbacks.Reason.FAILED, ex)`
`277`		`- closeAndRemove()`
`278`	`278`	`throw ex`
`279`	`279`	`}`
`280`	`280`	`}`
`281`	`281`
`282`	`282`	`fun timeout() {`
`283`		`- lifecycleCallbacks.sessionFinished(peerAddress, DtlsSessionLifecycleCallbacks.Reason.EXPIRED)`
`284`	`283`	`sessions.remove(peerAddress, this)`
`285`		`- logger.info("[{}] DTLS connection expired", peerAddress)`
`286`	`284`	`storeAndClose()`
	`285`	`+ logger.info("[{}] DTLS connection expired", peerAddress)`
	`286`	`+ lifecycleCallbacks.sessionFinished(peerAddress, DtlsSessionLifecycleCallbacks.Reason.EXPIRED)`
`287`	`287`	`}`
`288`	`288`
`289`	`289`	`private fun reportSessionStarted() {`
`290`		`- lifecycleCallbacks.sessionStarted(peerAddress, ctx.cipherSuite, ctx.reloaded)`
	`290`	`+ executor.supply { lifecycleCallbacks.sessionStarted(peerAddress, ctx.cipherSuite, ctx.reloaded) }`
`291`	`291`	`}`
`292`	`292`
`293`	`293`	`private fun reportSessionFinished(reason: DtlsSessionLifecycleCallbacks.Reason, err: Throwable? = null) {`
`294`		`- lifecycleCallbacks.sessionFinished(peerAddress, reason, err)`
	`294`	`+ executor.supply { lifecycleCallbacks.sessionFinished(peerAddress, reason, err) }`
`295`	`295`	`}`
`296`	`296`	`}`
`297`	`297`	`}`
Original file line number	Diff line number	Diff line change
`@@ -23,7 +23,7 @@ interface DtlsSessionLifecycleCallbacks {`
`23`	`23`	`SUCCEEDED, FAILED, CLOSED, EXPIRED`
`24`	`24`	`}`
`25`	`25`	`fun handshakeStarted(adr: InetSocketAddress) = Unit`
`26`		`- fun handshakeFinished(adr: InetSocketAddress, hanshakeStartTimestamp: Long, reason: Reason, throwable: Throwable? = null) = Unit`
	`26`	`+ fun handshakeFinished(adr: InetSocketAddress, hanshakeStartTimestamp: Long, hanshakeFinishTimestamp: Long, reason: Reason, throwable: Throwable? = null) = Unit`
`27`	`27`	`fun sessionStarted(adr: InetSocketAddress, cipherSuite: String, reloaded: Boolean) = Unit`
`28`	`28`	`fun sessionFinished(adr: InetSocketAddress, reason: Reason, throwable: Throwable? = null) = Unit`
`29`	`29`	`}`