diff options
Diffstat (limited to 'sources/hv-collector-core/src/main/kotlin/org/onap')
15 files changed, 19 insertions, 467 deletions
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<ServerHandle> -} 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 <piotr.jaszczyk@nokia.com> 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 <piotr.jaszczyk@nokia.com> - * @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 fe34a9c7..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,7 +26,7 @@ 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.model.ClientContext +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.ConsumedMessage @@ -43,12 +43,8 @@ class Router internal constructor(private val routing: Routing, sinkFactory: SinkFactory, ctx: ClientContext, metrics: Metrics) : - this(routing, - constructMessageSinks(routing, sinkFactory, ctx), - ctx, - metrics) { - logger.debug(ctx::mdc) { "Routing for client: $routing" } - logger.trace(ctx::mdc) { "Message sinks configured for client: $messageSinks" } + this(routing, constructMessageSinks(routing, sinkFactory, ctx), ctx, metrics) { + logger.debug(ctx::mdc) { "Routing configuration for client: $routing" } } fun route(message: VesMessage): Flux<ConsumedMessage> = 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 <T> Logger.handleReactiveStreamError(context: ClientContext, ex: Throwable, - returnFlux: Flux<T> = Flux.empty()): Flux<T> { - 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 <piotr.jaszczyk@nokia.com> - * @since May 2018 - */ -internal class NettyTcpServer(private val serverConfiguration: ServerConfiguration, - private val sslContext: Option<SslContext>, - private val collectorFactory: CollectorFactory, - private val metrics: Metrics) : Server { - - override fun start(): Mono<ServerHandle> = - Mono.defer { - TcpServer.create() - .addressSupplier { InetSocketAddress(serverConfiguration.listenPort) } - .configureSsl() - .handle(this::handleConnection) - .bind() - .map { - NettyServerHandle(it, closeAction()) - } - } - - private fun closeAction(): Mono<Void> = - 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<Void> = - messageHandlingStream(nettyInbound, nettyOutbound).run { - subscribe() - nettyOutbound.neverComplete() - } - - private fun messageHandlingStream(nettyInbound: NettyInbound, nettyOutbound: NettyOutbound): Mono<Void> = - 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<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, 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<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() } - } 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 <piotr.jaszczyk@nokia.com> - * @since December 2018 - */ -data class ClientContext( - val alloc: ByteBufAllocator = ByteBufAllocator.DEFAULT, - var clientAddress: Option<InetAddress> = None, - var clientCert: Option<X509Certificate> = None, - val requestId: String = UUID.randomUUID().toString(), // Should be somehow propagated to DMAAP - val invocationId: String = UUID.randomUUID().toString()) { - - val mdc: Map<String, String> - 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<String, String> - get() = mdc + ServiceContext.mdc - - private fun clientDn(): Option<String> = clientCert.map { it.subjectX500Principal.toString() } - private fun clientIp(): Option<String> = 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 <piotr.jaszczyk@nokia.com> - * @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() - } -} |