From 49f43c856c8ca793bc6972d9d4b47c2d0d4c0816 Mon Sep 17 00:00:00 2001 From: kjaniak Date: Wed, 3 Apr 2019 15:48:28 +0200 Subject: Creation of server module Issue-ID: DCAEGEN2-1390 Change-Id: I07410b16ed6566b933d5f1efa35bddb965225794 Signed-off-by: kjaniak Signed-off-by: Filip Krzywka --- sources/hv-collector-core/pom.xml | 28 ++-- .../dcae/collectors/veshv/boundary/adapters.kt | 2 +- .../org/onap/dcae/collectors/veshv/boundary/api.kt | 6 +- .../veshv/factory/HvVesCollectorFactory.kt | 2 +- .../dcae/collectors/veshv/factory/ServerFactory.kt | 48 ------- .../dcae/collectors/veshv/impl/HvVesCollector.kt | 4 +- .../org/onap/dcae/collectors/veshv/impl/Router.kt | 2 +- .../veshv/impl/adapters/ClientContextLogging.kt | 47 ------ .../collectors/veshv/impl/adapters/HttpAdapter.kt | 2 +- .../veshv/impl/adapters/kafka/KafkaPublisher.kt | 8 +- .../veshv/impl/adapters/kafka/KafkaSinkFactory.kt | 4 +- .../collectors/veshv/impl/socket/NettyTcpServer.kt | 160 --------------------- .../collectors/veshv/impl/socket/networking.kt | 81 ----------- .../collectors/veshv/impl/wire/WireChunkDecoder.kt | 6 +- .../dcae/collectors/veshv/model/ClientContext.kt | 61 -------- .../dcae/collectors/veshv/model/ServiceContext.kt | 45 ------ .../onap/dcae/collectors/veshv/impl/RouterTest.kt | 2 +- .../veshv/impl/wire/WireChunkDecoderTest.kt | 2 +- .../collectors/veshv/model/ClientContextTest.kt | 98 ------------- .../collectors/veshv/model/ServiceContextTest.kt | 67 --------- 20 files changed, 33 insertions(+), 642 deletions(-) delete mode 100644 sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/factory/ServerFactory.kt delete mode 100644 sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/ClientContextLogging.kt delete mode 100644 sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/socket/NettyTcpServer.kt delete mode 100644 sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/socket/networking.kt delete mode 100644 sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/model/ClientContext.kt delete mode 100644 sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/model/ServiceContext.kt delete mode 100644 sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/model/ClientContextTest.kt delete mode 100644 sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/model/ServiceContextTest.kt (limited to 'sources/hv-collector-core') diff --git a/sources/hv-collector-core/pom.xml b/sources/hv-collector-core/pom.xml index e1e35d8b..122de173 100644 --- a/sources/hv-collector-core/pom.xml +++ b/sources/hv-collector-core/pom.xml @@ -59,31 +59,25 @@ - ${project.parent.groupId} + ${project.groupId} hv-collector-configuration - ${project.parent.version} + ${project.version} - ${project.parent.groupId} + ${project.groupId} hv-collector-domain - ${project.parent.version} + ${project.version} compile - ${project.parent.groupId} + ${project.groupId} hv-collector-utils - ${project.parent.version} + ${project.version} - ${project.parent.groupId} + ${project.groupId} hv-collector-ssl - ${project.parent.version} - - - ${project.parent.groupId} - hv-collector-test-utils - ${project.parent.version} - test + ${project.version} io.arrow-kt @@ -105,6 +99,12 @@ io.projectreactor.kafka reactor-kafka + + ${project.groupId} + hv-collector-test-utils + ${project.version} + test + diff --git a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/boundary/adapters.kt b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/boundary/adapters.kt index 48f335a1..28b28203 100644 --- a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/boundary/adapters.kt +++ b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/boundary/adapters.kt @@ -21,7 +21,7 @@ package org.onap.dcae.collectors.veshv.boundary import org.onap.dcae.collectors.veshv.domain.RoutedMessage import org.onap.dcae.collectors.veshv.domain.WireFrameMessage -import org.onap.dcae.collectors.veshv.model.ClientContext +import org.onap.dcae.collectors.veshv.domain.logging.ClientContext import org.onap.dcae.collectors.veshv.model.ClientRejectionCause import org.onap.dcae.collectors.veshv.model.ConsumedMessage import org.onap.dcae.collectors.veshv.model.MessageDropCause diff --git a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/boundary/api.kt b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/boundary/api.kt index 4c54d7d2..23a5d376 100644 --- a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/boundary/api.kt +++ b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/boundary/api.kt @@ -20,9 +20,8 @@ package org.onap.dcae.collectors.veshv.boundary import io.netty.buffer.ByteBuf -import org.onap.dcae.collectors.veshv.model.ClientContext +import org.onap.dcae.collectors.veshv.domain.logging.ClientContext import org.onap.dcae.collectors.veshv.utils.Closeable -import org.onap.dcae.collectors.veshv.utils.ServerHandle import reactor.core.publisher.Flux import reactor.core.publisher.Mono @@ -34,6 +33,3 @@ interface CollectorFactory : Closeable { operator fun invoke(ctx: ClientContext): Collector } -interface Server { - fun start(): Mono -} diff --git a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/factory/HvVesCollectorFactory.kt b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/factory/HvVesCollectorFactory.kt index c3c5d733..1f221c60 100644 --- a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/factory/HvVesCollectorFactory.kt +++ b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/factory/HvVesCollectorFactory.kt @@ -29,7 +29,7 @@ import org.onap.dcae.collectors.veshv.impl.HvVesCollector import org.onap.dcae.collectors.veshv.impl.Router import org.onap.dcae.collectors.veshv.impl.VesDecoder import org.onap.dcae.collectors.veshv.impl.wire.WireChunkDecoder -import org.onap.dcae.collectors.veshv.model.ClientContext +import org.onap.dcae.collectors.veshv.domain.logging.ClientContext /** * @author Piotr Jaszczyk diff --git a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/factory/ServerFactory.kt b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/factory/ServerFactory.kt deleted file mode 100644 index e0f611b6..00000000 --- a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/factory/ServerFactory.kt +++ /dev/null @@ -1,48 +0,0 @@ -/* - * ============LICENSE_START======================================================= - * dcaegen2-collectors-veshv - * ================================================================================ - * Copyright (C) 2018 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.factory - -import org.onap.dcae.collectors.veshv.boundary.CollectorFactory -import org.onap.dcae.collectors.veshv.boundary.Metrics -import org.onap.dcae.collectors.veshv.boundary.Server -import org.onap.dcae.collectors.veshv.config.api.model.ServerConfiguration -import org.onap.dcae.collectors.veshv.impl.socket.NettyTcpServer -import org.onap.dcae.collectors.veshv.ssl.boundary.SecurityConfiguration -import org.onap.dcae.collectors.veshv.ssl.boundary.SslContextFactory - -/** - * @author Piotr Jaszczyk - * @since May 2018 - */ -object ServerFactory { - - private val sslFactory = SslContextFactory() - - fun createNettyTcpServer(serverConfig: ServerConfiguration, - securityConfig: SecurityConfiguration, - collectorFactory: CollectorFactory, - metrics: Metrics - ): Server = NettyTcpServer( - serverConfig, - sslFactory.createServerContext(securityConfig), - collectorFactory, - metrics - ) -} diff --git a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/HvVesCollector.kt b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/HvVesCollector.kt index 7d8f0cb1..ac7c3917 100644 --- a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/HvVesCollector.kt +++ b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/HvVesCollector.kt @@ -23,9 +23,9 @@ import io.netty.buffer.ByteBuf import org.onap.dcae.collectors.veshv.boundary.Collector import org.onap.dcae.collectors.veshv.boundary.Metrics import org.onap.dcae.collectors.veshv.domain.WireFrameMessage -import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.handleReactiveStreamError +import org.onap.dcae.collectors.veshv.domain.logging.ClientContextLogging.handleReactiveStreamError import org.onap.dcae.collectors.veshv.impl.wire.WireChunkDecoder -import org.onap.dcae.collectors.veshv.model.ClientContext +import org.onap.dcae.collectors.veshv.domain.logging.ClientContext import org.onap.dcae.collectors.veshv.model.ClientRejectionCause import org.onap.dcae.collectors.veshv.model.ConsumedMessage import org.onap.dcae.collectors.veshv.model.FailedToConsumeMessage diff --git a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/Router.kt b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/Router.kt index fec713ad..2190eba3 100644 --- a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/Router.kt +++ b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/Router.kt @@ -26,9 +26,9 @@ import org.onap.dcae.collectors.veshv.boundary.Sink import org.onap.dcae.collectors.veshv.boundary.SinkFactory import org.onap.dcae.collectors.veshv.config.api.model.Route import org.onap.dcae.collectors.veshv.config.api.model.Routing +import org.onap.dcae.collectors.veshv.domain.logging.ClientContext import org.onap.dcae.collectors.veshv.domain.RoutedMessage import org.onap.dcae.collectors.veshv.domain.VesMessage -import org.onap.dcae.collectors.veshv.model.ClientContext import org.onap.dcae.collectors.veshv.model.ConsumedMessage import org.onap.dcae.collectors.veshv.model.MessageDropCause import org.onap.dcae.collectors.veshv.utils.logging.Logger diff --git a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/ClientContextLogging.kt b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/ClientContextLogging.kt deleted file mode 100644 index 954de978..00000000 --- a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/ClientContextLogging.kt +++ /dev/null @@ -1,47 +0,0 @@ -/* - * ============LICENSE_START======================================================= - * dcaegen2-collectors-veshv - * ================================================================================ - * Copyright (C) 2018 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.adapters - -import org.onap.dcae.collectors.veshv.model.ClientContext -import org.onap.dcae.collectors.veshv.utils.logging.AtLevelLogger -import org.onap.dcae.collectors.veshv.utils.logging.Logger -import org.onap.dcae.collectors.veshv.utils.logging.handleReactiveStreamError -import reactor.core.publisher.Flux - -@Suppress("TooManyFunctions") -internal object ClientContextLogging { - fun Logger.withError(ctx: ClientContext, block: AtLevelLogger.() -> Unit) = withError(ctx::fullMdc, block) - fun Logger.withWarn(ctx: ClientContext, block: AtLevelLogger.() -> Unit) = withWarn(ctx::fullMdc, block) - fun Logger.withInfo(ctx: ClientContext, block: AtLevelLogger.() -> Unit) = withInfo(ctx::fullMdc, block) - fun Logger.withDebug(ctx: ClientContext, block: AtLevelLogger.() -> Unit) = withDebug(ctx::fullMdc, block) - fun Logger.withTrace(ctx: ClientContext, block: AtLevelLogger.() -> Unit) = withTrace(ctx::fullMdc, block) - - fun Logger.error(ctx: ClientContext, message: () -> String) = error(ctx::fullMdc, message) - fun Logger.warn(ctx: ClientContext, message: () -> String) = warn(ctx::fullMdc, message) - fun Logger.info(ctx: ClientContext, message: () -> String) = info(ctx::fullMdc, message) - fun Logger.debug(ctx: ClientContext, message: () -> String) = debug(ctx::fullMdc, message) - fun Logger.trace(ctx: ClientContext, message: () -> String) = trace(ctx::fullMdc, message) - - fun Logger.handleReactiveStreamError(context: ClientContext, ex: Throwable, - returnFlux: Flux = Flux.empty()): Flux { - return this.handleReactiveStreamError({ context.fullMdc }, ex, returnFlux) - } -} - diff --git a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/HttpAdapter.kt b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/HttpAdapter.kt index 8d154091..8f66de2b 100644 --- a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/HttpAdapter.kt +++ b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/HttpAdapter.kt @@ -21,7 +21,7 @@ package org.onap.dcae.collectors.veshv.impl.adapters import io.netty.handler.codec.http.HttpStatusClass import org.onap.dcae.collectors.veshv.utils.logging.Logger -import org.onap.dcae.collectors.veshv.utils.logging.OnapMdc +import org.onap.dcae.collectors.veshv.domain.logging.OnapMdc import reactor.core.publisher.Mono import reactor.netty.http.client.HttpClient import java.util.* diff --git a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaPublisher.kt b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaPublisher.kt index 7b726ab4..91e6fde5 100644 --- a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaPublisher.kt +++ b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaPublisher.kt @@ -22,14 +22,16 @@ package org.onap.dcae.collectors.veshv.impl.adapters.kafka import org.onap.dcae.collectors.veshv.boundary.Sink import org.onap.dcae.collectors.veshv.domain.RoutedMessage import org.onap.dcae.collectors.veshv.domain.VesMessage -import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.withDebug -import org.onap.dcae.collectors.veshv.model.ClientContext +import org.onap.dcae.collectors.veshv.domain.logging.ClientContextLogging.withDebug +import org.onap.dcae.collectors.veshv.domain.logging.ClientContext import org.onap.dcae.collectors.veshv.model.ConsumedMessage import org.onap.dcae.collectors.veshv.model.FailedToConsumeMessage import org.onap.dcae.collectors.veshv.model.MessageDropCause import org.onap.dcae.collectors.veshv.model.SuccessfullyConsumedMessage import org.onap.dcae.collectors.veshv.utils.logging.Logger -import org.onap.dcae.collectors.veshv.utils.logging.Marker +import org.onap.dcae.collectors.veshv.domain.logging.Marker +import org.onap.dcae.collectors.veshv.domain.logging.MarkerLogging.trace +import org.onap.dcae.collectors.veshv.domain.logging.MarkerLogging.warn import org.onap.ves.VesEventOuterClass.CommonEventHeader import reactor.core.publisher.Flux import reactor.kafka.sender.KafkaSender diff --git a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaSinkFactory.kt b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaSinkFactory.kt index 9df1af31..2973fa8d 100644 --- a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaSinkFactory.kt +++ b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaSinkFactory.kt @@ -22,8 +22,8 @@ package org.onap.dcae.collectors.veshv.impl.adapters.kafka import org.onap.dcae.collectors.veshv.boundary.SinkFactory import org.onap.dcae.collectors.veshv.domain.VesMessage import org.onap.dcae.collectors.veshv.impl.createKafkaSender -import org.onap.dcae.collectors.veshv.model.ClientContext -import org.onap.dcae.collectors.veshv.model.ServiceContext +import org.onap.dcae.collectors.veshv.domain.logging.ClientContext +import org.onap.dcae.collectors.veshv.domain.logging.ServiceContext import org.onap.dcae.collectors.veshv.utils.logging.Logger import org.onap.dcaegen2.services.sdk.model.streams.SinkStream import org.onap.ves.VesEventOuterClass.CommonEventHeader 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 deleted file mode 100644 index 7ce86f98..00000000 --- a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/socket/NettyTcpServer.kt +++ /dev/null @@ -1,160 +0,0 @@ -/* - * ============LICENSE_START======================================================= - * dcaegen2-collectors-veshv - * ================================================================================ - * Copyright (C) 2018 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.getOrElse -import io.netty.handler.ssl.SslContext -import org.onap.dcae.collectors.veshv.boundary.Collector -import org.onap.dcae.collectors.veshv.boundary.CollectorFactory -import org.onap.dcae.collectors.veshv.boundary.Metrics -import org.onap.dcae.collectors.veshv.boundary.Server -import org.onap.dcae.collectors.veshv.config.api.model.ServerConfiguration -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.model.ClientContext -import org.onap.dcae.collectors.veshv.model.ServiceContext -import org.onap.dcae.collectors.veshv.utils.NettyServerHandle -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.Connection -import reactor.netty.NettyInbound -import reactor.netty.NettyOutbound -import reactor.netty.tcp.TcpServer -import java.net.InetAddress -import java.net.InetSocketAddress -import java.time.Duration - - -/** - * @author Piotr Jaszczyk - * @since May 2018 - */ -internal class NettyTcpServer(private val serverConfiguration: ServerConfiguration, - private val sslContext: Option, - private val collectorFactory: CollectorFactory, - private val metrics: Metrics) : Server { - - override fun start(): Mono = - Mono.defer { - TcpServer.create() - .addressSupplier { InetSocketAddress(serverConfiguration.listenPort) } - .configureSsl() - .handle(this::handleConnection) - .bind() - .map { - NettyServerHandle(it, closeAction()) - } - } - - private fun closeAction(): Mono = - collectorFactory.close().doOnSuccess { - logger.info(ServiceContext::mdc) { "Netty TCP Server closed" } - } - - - private fun TcpServer.configureSsl() = - sslContext - .map { serverContext -> - logger.info { "Collector configured with SSL enabled" } - this.secure { it.sslContext(serverContext) } - }.getOrElse { - logger.info { "Collector configured with SSL disabled" } - this - } - - private fun handleConnection(nettyInbound: NettyInbound, nettyOutbound: NettyOutbound): Mono = - messageHandlingStream(nettyInbound, nettyOutbound).run { - subscribe() - nettyOutbound.neverComplete() - } - - private fun messageHandlingStream(nettyInbound: NettyInbound, nettyOutbound: NettyOutbound): Mono = - withNewClientContextFrom(nettyInbound, nettyOutbound) - { clientContext -> - logger.debug(clientContext::fullMdc) { "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 acceptIfNotLocalConnection(address: InetAddress, - clientContext: ClientContext, - nettyInbound: NettyInbound): Mono = - if (address.isLocalClientAddress()) { - logger.debug(clientContext) { - "Client address resolved to localhost. Discarding connection as suspected healthcheck" - } - nettyInbound.closeConnectionAndReturn(Mono.empty()) - } else { - acceptClientConnection(clientContext, nettyInbound) - } - - private fun acceptClientConnection(clientContext: ClientContext, nettyInbound: NettyInbound): Mono { - metrics.notifyClientConnected() - logger.info(clientContext::fullMdc, Marker.Entry) { "Handling new client connection" } - val collector = collectorFactory(clientContext) - return collector.handleClient(clientContext, nettyInbound) - } - - private fun Collector.handleClient(clientContext: ClientContext, - nettyInbound: NettyInbound) = - withConnectionFrom(nettyInbound) { connection -> - connection - .configureIdleTimeout(clientContext, serverConfiguration.idleTimeout) - .logConnectionClosed(clientContext) - }.run { - handleConnection(nettyInbound.createDataStream()) - } - - private fun Connection.configureIdleTimeout(ctx: ClientContext, timeout: Duration): Connection = - onReadIdle(timeout.toMillis()) { - logger.info(ctx) { - "Idle timeout of ${timeout.seconds} s reached. Closing connection from ${address()}..." - } - disconnectClient(ctx) - } - - 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 { - metrics.notifyClientDisconnected() - logger.info(ctx::fullMdc, Marker.Exit) { "Connection has been closed" } - } - - companion object { - private val logger = Logger(NettyTcpServer::class) - } -} 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 deleted file mode 100644 index a1e5b8fd..00000000 --- a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/socket/networking.kt +++ /dev/null @@ -1,81 +0,0 @@ -/* - * ============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 = - Option.fromNullable( - channel() - .pipeline() - .get(SslHandler::class.java) - ?.engine() - ?.session - ) - -internal fun SSLSession.findClientCert(): Option = - 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) -> Unit) = - closeChannel().addListener { task(it) } - -internal fun 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) = - 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() } - } diff --git a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireChunkDecoder.kt b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireChunkDecoder.kt index ca9d28ae..0d0f8ea7 100644 --- a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireChunkDecoder.kt +++ b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireChunkDecoder.kt @@ -25,9 +25,9 @@ import org.onap.dcae.collectors.veshv.domain.MissingWireFrameBytes import org.onap.dcae.collectors.veshv.domain.WireFrameDecoder import org.onap.dcae.collectors.veshv.domain.WireFrameDecodingError import org.onap.dcae.collectors.veshv.domain.WireFrameMessage -import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.handleReactiveStreamError -import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.trace -import org.onap.dcae.collectors.veshv.model.ClientContext +import org.onap.dcae.collectors.veshv.domain.logging.ClientContextLogging.handleReactiveStreamError +import org.onap.dcae.collectors.veshv.domain.logging.ClientContextLogging.trace +import org.onap.dcae.collectors.veshv.domain.logging.ClientContext import org.onap.dcae.collectors.veshv.utils.logging.Logger import reactor.core.publisher.Flux import reactor.core.publisher.Flux.defer diff --git a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/model/ClientContext.kt b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/model/ClientContext.kt deleted file mode 100644 index 7b082e64..00000000 --- a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/model/ClientContext.kt +++ /dev/null @@ -1,61 +0,0 @@ -/* - * ============LICENSE_START======================================================= - * dcaegen2-collectors-veshv - * ================================================================================ - * Copyright (C) 2018 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.model - -import arrow.core.None -import arrow.core.Option -import arrow.core.getOrElse -import io.netty.buffer.ByteBufAllocator -import org.onap.dcae.collectors.veshv.utils.logging.OnapMdc -import java.net.InetAddress -import java.security.cert.X509Certificate -import java.util.* - -/** - * @author Piotr Jaszczyk - * @since December 2018 - */ -data class ClientContext( - val alloc: ByteBufAllocator = ByteBufAllocator.DEFAULT, - var clientAddress: Option = None, - var clientCert: Option = None, - val requestId: String = UUID.randomUUID().toString(), // Should be somehow propagated to DMAAP - val invocationId: String = UUID.randomUUID().toString()) { - - val mdc: Map - get() = mapOf( - OnapMdc.REQUEST_ID to requestId, - OnapMdc.INVOCATION_ID to invocationId, - OnapMdc.STATUS_CODE to DEFAULT_STATUS_CODE, - OnapMdc.CLIENT_NAME to clientDn().getOrElse { DEFAULT_VALUE }, - OnapMdc.CLIENT_IP to clientIp().getOrElse { DEFAULT_VALUE } - ) - - val fullMdc: Map - get() = mdc + ServiceContext.mdc - - private fun clientDn(): Option = clientCert.map { it.subjectX500Principal.toString() } - private fun clientIp(): Option = clientAddress.map(InetAddress::getHostAddress) - - companion object { - const val DEFAULT_STATUS_CODE = "INPROGRESS" - const val DEFAULT_VALUE = "" - } -} diff --git a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/model/ServiceContext.kt b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/model/ServiceContext.kt deleted file mode 100644 index a72ec034..00000000 --- a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/model/ServiceContext.kt +++ /dev/null @@ -1,45 +0,0 @@ -/* - * ============LICENSE_START======================================================= - * dcaegen2-collectors-veshv - * ================================================================================ - * Copyright (C) 2018 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.model - -import org.onap.dcae.collectors.veshv.utils.logging.OnapMdc -import java.net.InetAddress -import java.net.UnknownHostException -import java.util.* - -/** - * @author Piotr Jaszczyk - * @since December 2018 - */ -object ServiceContext { - val instanceId = UUID.randomUUID().toString() - val serverFqdn = getHost().hostName!! - - val mdc = mapOf( - OnapMdc.INSTANCE_ID to instanceId, - OnapMdc.SERVER_FQDN to serverFqdn - ) - - private fun getHost() = try { - InetAddress.getLocalHost() - } catch (ex: UnknownHostException) { - InetAddress.getLoopbackAddress() - } -} diff --git a/sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/RouterTest.kt b/sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/RouterTest.kt index 6b9c6803..533581d5 100644 --- a/sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/RouterTest.kt +++ b/sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/RouterTest.kt @@ -36,7 +36,7 @@ import org.onap.dcae.collectors.veshv.domain.RoutedMessage import org.onap.dcae.collectors.veshv.domain.VesEventDomain.PERF3GPP import org.onap.dcae.collectors.veshv.domain.VesEventDomain.HEARTBEAT import org.onap.dcae.collectors.veshv.domain.VesEventDomain.SYSLOG -import org.onap.dcae.collectors.veshv.model.ClientContext +import org.onap.dcae.collectors.veshv.domain.logging.ClientContext import org.onap.dcae.collectors.veshv.domain.VesMessage import org.onap.dcae.collectors.veshv.model.SuccessfullyConsumedMessage import org.onap.dcae.collectors.veshv.tests.utils.commonHeader diff --git a/sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireChunkDecoderTest.kt b/sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireChunkDecoderTest.kt index e0092cf9..10dea82d 100644 --- a/sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireChunkDecoderTest.kt +++ b/sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireChunkDecoderTest.kt @@ -30,7 +30,7 @@ import org.jetbrains.spek.api.dsl.it import org.onap.dcae.collectors.veshv.domain.WireFrameEncoder import org.onap.dcae.collectors.veshv.domain.WireFrameDecoder import org.onap.dcae.collectors.veshv.domain.WireFrameMessage -import org.onap.dcae.collectors.veshv.model.ClientContext +import org.onap.dcae.collectors.veshv.domain.logging.ClientContext import reactor.test.test /** diff --git a/sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/model/ClientContextTest.kt b/sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/model/ClientContextTest.kt deleted file mode 100644 index a49428a7..00000000 --- a/sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/model/ClientContextTest.kt +++ /dev/null @@ -1,98 +0,0 @@ -/* - * ============LICENSE_START======================================================= - * dcaegen2-collectors-veshv - * ================================================================================ - * Copyright (C) 2018 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.model - -import arrow.core.Some -import com.nhaarman.mockitokotlin2.mock -import com.nhaarman.mockitokotlin2.whenever -import org.assertj.core.api.Assertions.assertThat -import org.jetbrains.spek.api.Spek -import org.jetbrains.spek.api.dsl.describe -import org.jetbrains.spek.api.dsl.given -import org.jetbrains.spek.api.dsl.it -import org.jetbrains.spek.api.dsl.on -import org.onap.dcae.collectors.veshv.utils.logging.OnapMdc -import java.net.Inet4Address -import java.net.InetAddress -import java.net.InetSocketAddress -import java.security.cert.X509Certificate -import java.util.* -import javax.security.auth.x500.X500Principal - -/** - * @author Piotr Jaszczyk - * @since December 2018 - */ -internal object ClientContextTest : Spek({ - describe("ClientContext") { - given("default instance") { - val cut = ClientContext() - - on("mapped diagnostic context") { - val mdc = cut.mdc - - it("should contain ${OnapMdc.REQUEST_ID}") { - assertThat(mdc[OnapMdc.REQUEST_ID]).isEqualTo(cut.requestId) - } - - it("should contain ${OnapMdc.INVOCATION_ID}") { - assertThat(mdc[OnapMdc.INVOCATION_ID]).isEqualTo(cut.invocationId) - } - - it("should contain ${OnapMdc.STATUS_CODE}") { - assertThat(mdc[OnapMdc.STATUS_CODE]).isEqualTo("INPROGRESS") - } - - it("should contain ${OnapMdc.CLIENT_NAME}") { - assertThat(mdc[OnapMdc.CLIENT_NAME]).isBlank() - } - - it("should contain ${OnapMdc.CLIENT_IP}") { - assertThat(mdc[OnapMdc.CLIENT_IP]).isBlank() - } - } - } - - given("instance with client data") { - val clientDn = "C=PL, O=Nokia, CN=NokiaBTS" - val clientIp = "192.168.52.34" - val cert: X509Certificate = mock() - val principal: X500Principal = mock() - val cut = ClientContext( - clientAddress = Some(InetAddress.getByName(clientIp)), - clientCert = Some(cert)) - - whenever(cert.subjectX500Principal).thenReturn(principal) - whenever(principal.toString()).thenReturn(clientDn) - - on("mapped diagnostic context") { - val mdc = cut.mdc - - it("should contain ${OnapMdc.CLIENT_NAME}") { - assertThat(mdc[OnapMdc.CLIENT_NAME]).isEqualTo(clientDn) - } - - it("should contain ${OnapMdc.CLIENT_IP}") { - assertThat(mdc[OnapMdc.CLIENT_IP]).isEqualTo(clientIp) - } - } - } - } -}) diff --git a/sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/model/ServiceContextTest.kt b/sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/model/ServiceContextTest.kt deleted file mode 100644 index 5b6e4526..00000000 --- a/sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/model/ServiceContextTest.kt +++ /dev/null @@ -1,67 +0,0 @@ -/* - * ============LICENSE_START======================================================= - * dcaegen2-collectors-veshv - * ================================================================================ - * Copyright (C) 2018 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.model - -import org.assertj.core.api.Assertions.assertThat -import org.jetbrains.spek.api.Spek -import org.jetbrains.spek.api.dsl.describe -import org.jetbrains.spek.api.dsl.given -import org.jetbrains.spek.api.dsl.it -import org.jetbrains.spek.api.dsl.on -import org.onap.dcae.collectors.veshv.utils.logging.OnapMdc -import java.util.* - -/** - * @author Piotr Jaszczyk - * @since December 2018 - */ -internal object ServiceContextTest : Spek({ - describe("ServiceContext") { - given("singleton instance") { - val cut = ServiceContext - - on("instanceId") { - val instanceId = cut.instanceId - it("should be valid UUID") { - UUID.fromString(instanceId) // should not throw - } - } - - on("serverFqdn") { - val serverFqdn = cut.serverFqdn - it("should be non empty") { - assertThat(serverFqdn).isNotBlank() - } - } - - on("mapped diagnostic context") { - val mdc = cut.mdc - - it("should contain ${OnapMdc.INSTANCE_ID}") { - assertThat(mdc[OnapMdc.INSTANCE_ID]).isEqualTo(cut.instanceId) - } - - it("should contain ${OnapMdc.SERVER_FQDN}") { - assertThat(mdc[OnapMdc.SERVER_FQDN]).isEqualTo(cut.serverFqdn) - } - } - } - } -}) -- cgit 1.2.3-korg