sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/socket/NettyTcpServer.kt

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * dcaegen2-collectors-veshv
   4  * ================================================================================
   5  * Copyright (C) 2018 NOKIA
   6  * ================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  * ============LICENSE_END=========================================================
  19  */
  20 package org.onap.dcae.collectors.veshv.impl.socket
  21
  22 import arrow.core.Option
  23 import arrow.core.getOrElse
  24 import io.netty.handler.ssl.SslContext
  25 import org.onap.dcae.collectors.veshv.boundary.Collector
  26 import org.onap.dcae.collectors.veshv.boundary.CollectorFactory
  27 import org.onap.dcae.collectors.veshv.boundary.Metrics
  28 import org.onap.dcae.collectors.veshv.boundary.Server
  29 import org.onap.dcae.collectors.veshv.config.api.model.ServerConfiguration
  30 import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.debug
  31 import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.info
  32 import org.onap.dcae.collectors.veshv.model.ClientContext
  33 import org.onap.dcae.collectors.veshv.model.ServiceContext
  34 import org.onap.dcae.collectors.veshv.utils.NettyServerHandle
  35 import org.onap.dcae.collectors.veshv.utils.ServerHandle
  36 import org.onap.dcae.collectors.veshv.utils.logging.Logger
  37 import org.onap.dcae.collectors.veshv.utils.logging.Marker
  38 import reactor.core.publisher.Mono
  39 import reactor.netty.Connection
  40 import reactor.netty.NettyInbound
  41 import reactor.netty.NettyOutbound
  42 import reactor.netty.tcp.TcpServer
  43 import java.net.InetAddress
  44 import java.net.InetSocketAddress
  45 import java.time.Duration
  46
  47
  48 /**
  49  * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
  50  * @since May 2018
  51  */
  52 internal class NettyTcpServer(private val serverConfiguration: ServerConfiguration,
  53                               private val sslContext: Option<SslContext>,
  54                               private val collectorFactory: CollectorFactory,
  55                               private val metrics: Metrics) : Server {
  56
  57     override fun start(): Mono<ServerHandle> =
  58             Mono.defer {
  59                 TcpServer.create()
  60                         .addressSupplier { InetSocketAddress(serverConfiguration.listenPort) }
  61                         .configureSsl()
  62                         .handle(this::handleConnection)
  63                         .bind()
  64                         .map {
  65                             NettyServerHandle(it, closeAction())
  66                         }
  67             }
  68
  69     private fun closeAction(): Mono<Void> =
  70             collectorFactory.close().doOnSuccess {
  71                 logger.info(ServiceContext::mdc) { "Netty TCP Server closed" }
  72             }
  73
  74
  75     private fun TcpServer.configureSsl() =
  76             sslContext
  77                     .map { serverContext ->
  78                         logger.info { "Collector configured with SSL enabled" }
  79                         this.secure { it.sslContext(serverContext) }
  80                     }.getOrElse {
  81                         logger.info { "Collector configured with SSL disabled" }
  82                         this
  83                     }
  84
  85     private fun handleConnection(nettyInbound: NettyInbound, nettyOutbound: NettyOutbound): Mono<Void> =
  86             messageHandlingStream(nettyInbound, nettyOutbound).run {
  87                 subscribe()
  88                 nettyOutbound.neverComplete()
  89             }
  90
  91     private fun messageHandlingStream(nettyInbound: NettyInbound, nettyOutbound: NettyOutbound): Mono<Void> =
  92             withNewClientContextFrom(nettyInbound, nettyOutbound)
  93             { clientContext ->
  94                 logger.debug(clientContext::fullMdc) { "Client connection request received" }
  95
  96                 clientContext.clientAddress
  97                         .map { acceptIfNotLocalConnection(it, clientContext, nettyInbound) }
  98                         .getOrElse {
  99                             logger.warn(clientContext::fullMdc) {
 100                                 "Client address could not be resolved. Discarding connection"
 101                             }
 102                             nettyInbound.closeConnectionAndReturn(Mono.empty())
 103                         }
 104             }
 105
 106     private fun acceptIfNotLocalConnection(address: InetAddress,
 107                                            clientContext: ClientContext,
 108                                            nettyInbound: NettyInbound): Mono<Void> =
 109             if (address.isLocalClientAddress()) {
 110                 logger.debug(clientContext) {
 111                     "Client address resolved to localhost. Discarding connection as suspected healthcheck"
 112                 }
 113                 nettyInbound.closeConnectionAndReturn(Mono.empty<Void>())
 114             } else {
 115                 acceptClientConnection(clientContext, nettyInbound)
 116             }
 117
 118     private fun acceptClientConnection(clientContext: ClientContext, nettyInbound: NettyInbound): Mono<Void> {
 119         metrics.notifyClientConnected()
 120         logger.info(clientContext::fullMdc, Marker.Entry) { "Handling new client connection" }
 121         val collector = collectorFactory(clientContext)
 122         return collector.handleClient(clientContext, nettyInbound)
 123     }
 124
 125     private fun Collector.handleClient(clientContext: ClientContext,
 126                                        nettyInbound: NettyInbound) =
 127             withConnectionFrom(nettyInbound) { connection ->
 128                 connection
 129                         .configureIdleTimeout(clientContext, serverConfiguration.idleTimeout)
 130                         .logConnectionClosed(clientContext)
 131             }.run {
 132                 handleConnection(nettyInbound.createDataStream())
 133             }
 134
 135     private fun Connection.configureIdleTimeout(ctx: ClientContext, timeout: Duration): Connection =
 136             onReadIdle(timeout.toMillis()) {
 137                 logger.info(ctx) {
 138                     "Idle timeout of ${timeout.seconds} s reached. Closing connection from ${address()}..."
 139                 }
 140                 disconnectClient(ctx)
 141             }
 142
 143     private fun Connection.disconnectClient(ctx: ClientContext) =
 144             closeChannelAndThen {
 145                 if (it.isSuccess)
 146                     logger.debug(ctx::fullMdc, Marker.Exit) { "Channel closed successfully." }
 147                 else
 148                     logger.warn(ctx::fullMdc, Marker.Exit, { "Channel close failed" }, it.cause())
 149             }
 150
 151     private fun Connection.logConnectionClosed(ctx: ClientContext): Connection =
 152             onDispose {
 153                 metrics.notifyClientDisconnected()
 154                 logger.info(ctx::fullMdc, Marker.Exit) { "Connection has been closed" }
 155             }
 156
 157     companion object {
 158         private val logger = Logger(NettyTcpServer::class)
 159     }
 160 }