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

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * dcaegen2-collectors-veshv
   4  * ================================================================================
   5  * Copyright (C) 2018-2019 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
  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.api.Server
  26 import org.onap.dcae.collectors.veshv.boundary.Collector
  27 import org.onap.dcae.collectors.veshv.boundary.CollectorFactory
  28 import org.onap.dcae.collectors.veshv.boundary.Metrics
  29 import org.onap.dcae.collectors.veshv.config.api.model.ServerConfiguration
  30 import org.onap.dcae.collectors.veshv.domain.logging.ClientContext
  31 import org.onap.dcae.collectors.veshv.domain.logging.ServiceContext
  32 import org.onap.dcae.collectors.veshv.utils.NettyServerHandle
  33 import org.onap.dcae.collectors.veshv.utils.ServerHandle
  34 import org.onap.dcae.collectors.veshv.utils.logging.Logger
  35 import org.onap.dcae.collectors.veshv.domain.logging.Marker
  36 import org.onap.dcae.collectors.veshv.domain.logging.MarkerLogging.debug
  37 import org.onap.dcae.collectors.veshv.domain.logging.MarkerLogging.info
  38 import org.onap.dcae.collectors.veshv.domain.logging.MarkerLogging.warn
  39 import org.onap.dcae.collectors.veshv.domain.logging.ClientContextLogging.debug
  40 import org.onap.dcae.collectors.veshv.domain.logging.ClientContextLogging.info
  41 import reactor.core.publisher.Mono
  42 import reactor.netty.Connection
  43 import reactor.netty.NettyInbound
  44 import reactor.netty.NettyOutbound
  45 import reactor.netty.tcp.TcpServer
  46 import java.net.InetAddress
  47 import java.net.InetSocketAddress
  48 import java.time.Duration
  49
  50
  51 /**
  52  * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
  53  * @since May 2018
  54  */
  55 internal class NettyTcpServer(private val serverConfiguration: ServerConfiguration,
  56                               private val sslContext: Option<SslContext>,
  57                               private val collectorFactory: CollectorFactory,
  58                               private val metrics: Metrics) : Server {
  59
  60     override fun start(): Mono<ServerHandle> =
  61             Mono.defer {
  62                 TcpServer.create()
  63                         .addressSupplier { InetSocketAddress(serverConfiguration.listenPort) }
  64                         .configureSsl()
  65                         .handle(this::handleConnection)
  66                         .bind()
  67                         .map {
  68                             NettyServerHandle(it, closeAction())
  69                         }
  70             }
  71
  72     private fun closeAction(): Mono<Void> =
  73             collectorFactory.close().doOnSuccess {
  74                 logger.info(ServiceContext::mdc) { "Netty TCP Server closed" }
  75             }
  76
  77
  78     private fun TcpServer.configureSsl() =
  79             sslContext
  80                     .map { serverContext ->
  81                         logger.info { "Collector configured with SSL enabled" }
  82                         this.secure { it.sslContext(serverContext) }
  83                     }.getOrElse {
  84                         logger.info { "Collector configured with SSL disabled" }
  85                         this
  86                     }
  87
  88     private fun handleConnection(nettyInbound: NettyInbound, nettyOutbound: NettyOutbound): Mono<Void> =
  89             messageHandlingStream(nettyInbound, nettyOutbound).run {
  90                 subscribe()
  91                 nettyOutbound.neverComplete()
  92             }
  93
  94     private fun messageHandlingStream(nettyInbound: NettyInbound, nettyOutbound: NettyOutbound): Mono<Void> =
  95             withNewClientContextFrom(nettyInbound, nettyOutbound)
  96             { clientContext ->
  97                 logger.debug(clientContext::fullMdc) { "Client connection request received" }
  98
  99                 clientContext.clientAddress
 100                         .map { acceptIfNotLocalConnection(it, clientContext, nettyInbound) }
 101                         .getOrElse {
 102                             logger.warn(clientContext::fullMdc) {
 103                                 "Client address could not be resolved. Discarding connection"
 104                             }
 105                             nettyInbound.closeConnectionAndReturn(Mono.empty())
 106                         }
 107             }
 108
 109     private fun acceptIfNotLocalConnection(address: InetAddress,
 110                                            clientContext: ClientContext,
 111                                            nettyInbound: NettyInbound): Mono<Void> =
 112             if (address.isLocalClientAddress()) {
 113                 logger.debug(clientContext) {
 114                     "Client address resolved to localhost. Discarding connection as suspected healthcheck"
 115                 }
 116                 nettyInbound.closeConnectionAndReturn(Mono.empty<Void>())
 117             } else {
 118                 acceptClientConnection(clientContext, nettyInbound)
 119             }
 120
 121     private fun acceptClientConnection(clientContext: ClientContext, nettyInbound: NettyInbound): Mono<Void> {
 122         metrics.notifyClientConnected()
 123         logger.info(clientContext::fullMdc, Marker.Entry) { "Handling new client connection" }
 124         val collector = collectorFactory(clientContext)
 125         return collector.handleClient(clientContext, nettyInbound)
 126     }
 127
 128     private fun Collector.handleClient(clientContext: ClientContext,
 129                                        nettyInbound: NettyInbound) =
 130             withConnectionFrom(nettyInbound) { connection ->
 131                 connection
 132                         .configureIdleTimeout(clientContext, serverConfiguration.idleTimeout)
 133                         .logConnectionClosed(clientContext)
 134             }.run {
 135                 handleConnection(nettyInbound.createDataStream())
 136             }
 137
 138     private fun Connection.configureIdleTimeout(ctx: ClientContext, timeout: Duration): Connection =
 139             onReadIdle(timeout.toMillis()) {
 140                 logger.info(ctx) {
 141                     "Idle timeout of ${timeout.seconds} s reached. Closing connection from ${address()}..."
 142                 }
 143                 disconnectClient(ctx)
 144             }
 145
 146     private fun Connection.disconnectClient(ctx: ClientContext) =
 147             closeChannelAndThen {
 148                 if (it.isSuccess)
 149                     logger.debug(ctx::fullMdc, Marker.Exit) { "Channel closed successfully." }
 150                 else
 151                     logger.warn(ctx::fullMdc, Marker.Exit, { "Channel close failed" }, it.cause())
 152             }
 153
 154     private fun Connection.logConnectionClosed(ctx: ClientContext): Connection =
 155             onDispose {
 156                 metrics.notifyClientDisconnected()
 157                 logger.info(ctx::fullMdc, Marker.Exit) { "Connection has been closed" }
 158             }
 159
 160     companion object {
 161         private val logger = Logger(NettyTcpServer::class)
 162     }
 163 }