summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--build/hv-collector-analysis/src/main/resources/onap-detekt-config.yml2
-rw-r--r--development/docker-compose.yml2
-rw-r--r--sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/socket/NettyTcpServer.kt123
-rw-r--r--sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/socket/networking.kt81
-rw-r--r--sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/logging/Logger.kt36
5 files changed, 175 insertions, 69 deletions
diff --git a/build/hv-collector-analysis/src/main/resources/onap-detekt-config.yml b/build/hv-collector-analysis/src/main/resources/onap-detekt-config.yml
index 36f092dd..f8d8b1ee 100644
--- a/build/hv-collector-analysis/src/main/resources/onap-detekt-config.yml
+++ b/build/hv-collector-analysis/src/main/resources/onap-detekt-config.yml
@@ -116,7 +116,7 @@ complexity:
thresholdInObjects: 11
thresholdInEnums: 11
ignoreDeprecated: false
- ignorePrivate: true
+ ignorePrivate: false
empty-blocks:
active: true
diff --git a/development/docker-compose.yml b/development/docker-compose.yml
index b3257b3d..c93100ef 100644
--- a/development/docker-compose.yml
+++ b/development/docker-compose.yml
@@ -73,7 +73,7 @@ services:
environment:
JAVA_OPTS: "-Dio.netty.leakDetection.level=paranoid -Dlogback.configurationFile=/etc/ONAP/dcae-hv-ves-collector/logback.xml"
healthcheck:
- test: curl -f http://localhost:6060/health/ready || exit 1
+ test: ./healthcheck.sh || exit 1
interval: 10s
timeout: 3s
retries: 3
diff --git a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/socket/NettyTcpServer.kt b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/socket/NettyTcpServer.kt
index adc629bc..0d07d167 100644
--- a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/socket/NettyTcpServer.kt
+++ b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/socket/NettyTcpServer.kt
@@ -19,18 +19,15 @@
*/
package org.onap.dcae.collectors.veshv.impl.socket
-import arrow.core.None
-import arrow.core.Option
+import arrow.core.Try
import arrow.core.getOrElse
import arrow.effects.IO
-import arrow.syntax.collections.firstOption
-import io.netty.handler.ssl.SslHandler
+import org.onap.dcae.collectors.veshv.boundary.Collector
import org.onap.dcae.collectors.veshv.boundary.CollectorProvider
import org.onap.dcae.collectors.veshv.boundary.Metrics
import org.onap.dcae.collectors.veshv.boundary.Server
import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.debug
import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.info
-import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.withWarn
import org.onap.dcae.collectors.veshv.model.ClientContext
import org.onap.dcae.collectors.veshv.model.ServerConfiguration
import org.onap.dcae.collectors.veshv.model.ServiceContext
@@ -40,14 +37,12 @@ import org.onap.dcae.collectors.veshv.utils.ServerHandle
import org.onap.dcae.collectors.veshv.utils.logging.Logger
import org.onap.dcae.collectors.veshv.utils.logging.Marker
import reactor.core.publisher.Mono
-import reactor.netty.ByteBufFlux
import reactor.netty.Connection
import reactor.netty.NettyInbound
import reactor.netty.NettyOutbound
import reactor.netty.tcp.TcpServer
-import java.security.cert.X509Certificate
+import java.net.InetAddress
import java.time.Duration
-import javax.net.ssl.SSLSession
/**
@@ -82,64 +77,61 @@ internal class NettyTcpServer(private val serverConfig: ServerConfiguration,
this
}
- private fun handleConnection(nettyInbound: NettyInbound, nettyOutbound: NettyOutbound): Mono<Void> {
- metrics.notifyClientConnected()
- val clientContext = ClientContext(nettyOutbound.alloc())
- nettyInbound.withConnection {
- populateClientContext(clientContext, it)
- it.channel().pipeline().get(SslHandler::class.java)?.engine()?.session?.let { sslSession ->
- sslSession.peerCertificates.firstOption().map { it as X509Certificate }.map { it.subjectDN.name }
+ private fun handleConnection(nettyInbound: NettyInbound, nettyOutbound: NettyOutbound): Mono<Void> =
+ messageHandlingStream(nettyInbound, nettyOutbound).run {
+ subscribe()
+ nettyOutbound.neverComplete()
}
- }
- logger.debug(clientContext::fullMdc, Marker.Entry) { "Client connection request received" }
- messageHandlingStream(clientContext, nettyInbound).subscribe()
- return nettyOutbound.neverComplete()
- }
+ private fun messageHandlingStream(nettyInbound: NettyInbound, nettyOutbound: NettyOutbound): Mono<Void> =
+ withNewClientContextFrom(nettyInbound, nettyOutbound)
+ { clientContext ->
+ logger.debug(clientContext::fullMdc, Marker.Entry) { "Client connection request received" }
+
+ clientContext.clientAddress
+ .map { acceptIfNotLocalConnection(it, clientContext, nettyInbound) }
+ .getOrElse {
+ logger.warn(clientContext::fullMdc) {
+ "Client address could not be resolved. Discarding connection"
+ }
+ nettyInbound.closeConnectionAndReturn(Mono.empty())
+ }
+ }
- private fun messageHandlingStream(clientContext: ClientContext, nettyInbound: NettyInbound): Mono<Void> =
- collectorProvider(clientContext).fold(
+ private fun acceptIfNotLocalConnection(address: InetAddress,
+ clientContext: ClientContext,
+ nettyInbound: NettyInbound): Mono<Void> =
+ if (address.isLocalClientAddress()) {
+ logger.debug(clientContext) {
+ "Client address resolved to localhost. Discarding connection as suspected healthcheck"
+ }
+ nettyInbound.closeConnectionAndReturn(Mono.empty<Void>())
+ } else {
+ acceptClientConnection(clientContext, nettyInbound)
+ }
+
+ private fun acceptClientConnection(clientContext: ClientContext, nettyInbound: NettyInbound): Mono<Void> {
+ metrics.notifyClientConnected()
+ logger.info(clientContext::fullMdc) { "Handling new client connection" }
+ return collectorProvider(clientContext).fold(
{
- logger.warn(clientContext::fullMdc) { "Collector not ready. Closing connection..." }
- Mono.empty()
+ logger.warn(clientContext::fullMdc) { "Collector is not ready. Closing connection" }
+ nettyInbound.closeConnectionAndReturn(Mono.empty<Void>())
},
- {
- logger.info(clientContext::fullMdc) { "Handling new connection" }
- nettyInbound.withConnection { conn ->
- conn
- .configureIdleTimeout(clientContext, serverConfig.idleTimeout)
- .logConnectionClosed(clientContext)
- }
- it.handleConnection(createDataStream(nettyInbound))
- }
+ handleClient(clientContext, nettyInbound)
)
-
- private fun populateClientContext(clientContext: ClientContext, connection: Connection) {
- clientContext.clientAddress = try {
- Option.fromNullable(connection.address().address)
- } catch (ex: Exception) {
- None
- }
- clientContext.clientCert = getSslSession(connection).flatMap(::findClientCert)
}
- private fun getSslSession(connection: Connection) = Option.fromNullable(
+ private fun handleClient(clientContext: ClientContext,
+ nettyInbound: NettyInbound): (Collector) -> Mono<Void> = { collector ->
+ withConnectionFrom(nettyInbound) { connection ->
connection
- .channel()
- .pipeline()
- .get(SslHandler::class.java)
- ?.engine()
- ?.session)
-
- private fun findClientCert(sslSession: SSLSession): Option<X509Certificate> =
- sslSession
- .peerCertificates
- .firstOption()
- .flatMap { Option.fromNullable(it as? X509Certificate) }
-
- private fun createDataStream(nettyInbound: NettyInbound): ByteBufFlux = nettyInbound
- .receive()
- .retain()
+ .configureIdleTimeout(clientContext, serverConfig.idleTimeout)
+ .logConnectionClosed(clientContext)
+ }.run {
+ collector.handleConnection(nettyInbound.createDataStream())
+ }
+ }
private fun Connection.configureIdleTimeout(ctx: ClientContext, timeout: Duration): Connection =
onReadIdle(timeout.toMillis()) {
@@ -149,16 +141,13 @@ internal class NettyTcpServer(private val serverConfig: ServerConfiguration,
disconnectClient(ctx)
}
-
- private fun Connection.disconnectClient(ctx: ClientContext) {
- channel().close().addListener {
- logger.debug(ctx::fullMdc, Marker.Exit) { "Closing client channel." }
- if (it.isSuccess)
- logger.debug(ctx) { "Channel closed successfully." }
- else
- logger.withWarn(ctx) { log("Channel close failed", it.cause()) }
- }
- }
+ private fun Connection.disconnectClient(ctx: ClientContext) =
+ closeChannelAndThen {
+ if (it.isSuccess)
+ logger.debug(ctx::fullMdc, Marker.Exit) { "Channel closed successfully." }
+ else
+ logger.warn(ctx::fullMdc, Marker.Exit, { "Channel close failed" }, it.cause())
+ }
private fun Connection.logConnectionClosed(ctx: ClientContext): Connection =
onDispose {
diff --git a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/socket/networking.kt b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/socket/networking.kt
new file mode 100644
index 00000000..91f502e6
--- /dev/null
+++ b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/socket/networking.kt
@@ -0,0 +1,81 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2019 NOKIA
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.dcae.collectors.veshv.impl.socket
+
+import arrow.core.Option
+import arrow.core.Try
+import arrow.syntax.collections.firstOption
+import io.netty.handler.ssl.SslHandler
+import io.netty.util.concurrent.Future
+import org.onap.dcae.collectors.veshv.model.ClientContext
+import reactor.core.publisher.Mono
+import reactor.netty.ByteBufFlux
+import reactor.netty.Connection
+import reactor.netty.NettyInbound
+import reactor.netty.NettyOutbound
+import java.net.InetAddress
+import java.security.cert.X509Certificate
+import javax.net.ssl.SSLSession
+
+internal fun InetAddress.isLocalClientAddress() = hostAddress == "127.0.0.1" || hostName == "localhost"
+
+internal fun Connection.getSslSession(): Option<SSLSession> =
+ Option.fromNullable(
+ channel()
+ .pipeline()
+ .get(SslHandler::class.java)
+ ?.engine()
+ ?.session
+ )
+
+internal fun SSLSession.findClientCert(): Option<X509Certificate> =
+ peerCertificates
+ .firstOption()
+ .flatMap { Option.fromNullable(it as? X509Certificate) }
+
+internal fun withConnectionFrom(nettyInboud: NettyInbound, task: (Connection) -> Unit) =
+ nettyInboud.withConnection(task)
+
+internal fun Connection.closeChannel() = channel().close()
+
+internal fun Connection.closeChannelAndThen(task: (Future<in Void>) -> Unit) =
+ closeChannel().addListener { task(it) }
+
+internal fun <T> NettyInbound.closeConnectionAndReturn(returnValue: T): T =
+ withConnectionFrom(this) { it.closeChannel() }.let { returnValue }
+
+internal fun NettyInbound.createDataStream(): ByteBufFlux = receive().retain()
+
+//
+// ClientContext related
+//
+
+internal inline fun withNewClientContextFrom(nettyInbound: NettyInbound,
+ nettyOutbound: NettyOutbound,
+ reactiveTask: (ClientContext) -> Mono<Void>) =
+ ClientContext(nettyOutbound.alloc())
+ .also { populateClientContextFromInbound(it, nettyInbound) }
+ .run(reactiveTask)
+
+internal fun populateClientContextFromInbound(clientContext: ClientContext, nettyInbound: NettyInbound) =
+ withConnectionFrom(nettyInbound) { connection ->
+ clientContext.clientAddress = Try { connection.address().address }.toOption()
+ clientContext.clientCert = connection.getSslSession().flatMap { it.findClientCert() }
+ } \ No newline at end of file
diff --git a/sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/logging/Logger.kt b/sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/logging/Logger.kt
index ade9b480..7fcc73a0 100644
--- a/sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/logging/Logger.kt
+++ b/sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/logging/Logger.kt
@@ -64,6 +64,9 @@ class Logger(logger: org.slf4j.Logger) {
fun error(mdc: MappedDiagnosticContext, marker: Marker, message: () -> String) =
errorLogger.withMdc(mdc) { log(marker, message()) }
+ fun error(mdc: MappedDiagnosticContext, marker: Marker, message: () -> String, t: Throwable) =
+ errorLogger.withMdc(mdc) { log(marker, message(), t) }
+
// WARN
fun withWarn(block: AtLevelLogger.() -> Unit) = warnLogger.block()
@@ -81,6 +84,9 @@ class Logger(logger: org.slf4j.Logger) {
fun warn(mdc: MappedDiagnosticContext, marker: Marker, message: () -> String) =
warnLogger.withMdc(mdc) { log(marker, message()) }
+ fun warn(mdc: MappedDiagnosticContext, marker: Marker, message: () -> String, t: Throwable) =
+ warnLogger.withMdc(mdc) { log(marker, message(), t) }
+
// INFO
fun withInfo(block: AtLevelLogger.() -> Unit) = infoLogger.block()
@@ -144,6 +150,7 @@ abstract class AtLevelLogger {
abstract fun log(message: String)
abstract fun log(message: String, t: Throwable)
abstract fun log(marker: Marker, message: String)
+ abstract fun log(marker: Marker, message: String, t: Throwable)
open val enabled: Boolean
get() = true
@@ -187,6 +194,10 @@ object OffLevelLogger : AtLevelLogger() {
override fun log(marker: Marker, message: String) {
// do not log anything
}
+
+ override fun log(marker: Marker, message: String, t: Throwable) {
+ // do no log anything
+ }
}
@Suppress("SuboptimalLoggerUsage")
@@ -203,6 +214,11 @@ class ErrorLevelLogger(private val logger: org.slf4j.Logger) : AtLevelLogger() {
withAdditionalMdc(marker.mdc) {
logger.error(marker.slf4jMarker, message)
}
+
+ override fun log(marker: Marker, message: String, t: Throwable) =
+ withAdditionalMdc(marker.mdc) {
+ logger.error(marker.slf4jMarker, message, t)
+ }
}
@Suppress("SuboptimalLoggerUsage")
@@ -219,6 +235,11 @@ class WarnLevelLogger(private val logger: org.slf4j.Logger) : AtLevelLogger() {
withAdditionalMdc(marker.mdc) {
logger.warn(marker.slf4jMarker, message)
}
+
+ override fun log(marker: Marker, message: String, t: Throwable) =
+ withAdditionalMdc(marker.mdc) {
+ logger.warn(marker.slf4jMarker, message, t)
+ }
}
@Suppress("SuboptimalLoggerUsage")
@@ -235,6 +256,11 @@ class InfoLevelLogger(private val logger: org.slf4j.Logger) : AtLevelLogger() {
withAdditionalMdc(marker.mdc) {
logger.info(marker.slf4jMarker, message)
}
+
+ override fun log(marker: Marker, message: String, t: Throwable) =
+ withAdditionalMdc(marker.mdc) {
+ logger.info(marker.slf4jMarker, message, t)
+ }
}
@Suppress("SuboptimalLoggerUsage")
@@ -251,6 +277,11 @@ class DebugLevelLogger(private val logger: org.slf4j.Logger) : AtLevelLogger() {
withAdditionalMdc(marker.mdc) {
logger.debug(marker.slf4jMarker, message)
}
+
+ override fun log(marker: Marker, message: String, t: Throwable) =
+ withAdditionalMdc(marker.mdc) {
+ logger.debug(marker.slf4jMarker, message, t)
+ }
}
@Suppress("SuboptimalLoggerUsage")
@@ -267,4 +298,9 @@ class TraceLevelLogger(private val logger: org.slf4j.Logger) : AtLevelLogger() {
withAdditionalMdc(marker.mdc) {
logger.trace(marker.slf4jMarker, message)
}
+
+ override fun log(marker: Marker, message: String, t: Throwable) =
+ withAdditionalMdc(marker.mdc) {
+ logger.trace(marker.slf4jMarker, message, t)
+ }
}