gerrit.onap Code Review - dcaegen2/collectors/hv-ves.git/blob

   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.getOrElse
  23 import arrow.effects.IO
  24 import org.onap.dcae.collectors.veshv.boundary.Collector
  25 import org.onap.dcae.collectors.veshv.boundary.CollectorProvider
  26 import org.onap.dcae.collectors.veshv.boundary.Metrics
  27 import org.onap.dcae.collectors.veshv.boundary.Server
  28 import org.onap.dcae.collectors.veshv.config.api.model.ServerConfiguration
  29 import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.debug
  30 import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.info
  31 import org.onap.dcae.collectors.veshv.model.ClientContext
  32 import org.onap.dcae.collectors.veshv.model.ServiceContext
  33 import org.onap.dcae.collectors.veshv.ssl.boundary.SslContextFactory
  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.time.Duration
  45
  46
  47 /**
  48  * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
  49  * @since May 2018
  50  */
  51 internal class NettyTcpServer(private val serverConfig: ServerConfiguration,
  52                               private val sslContextFactory: SslContextFactory,
  53                               private val collectorProvider: CollectorProvider,
  54                               private val metrics: Metrics) : Server {
  55
  56     override fun start(): IO<ServerHandle> = IO {
  57         TcpServer.create()
  58                 .addressSupplier { serverConfig.serverListenAddress }
  59                 .configureSsl()
  60                 .handle(this::handleConnection)
  61                 .doOnUnbound {
  62                     logger.info(ServiceContext::mdc) { "Netty TCP Server closed" }
  63                     collectorProvider.close().unsafeRunSync()
  64                 }
  65                 .let { NettyServerHandle(it.bindNow()) }
  66     }
  67
  68     private fun TcpServer.configureSsl() =
  69             sslContextFactory
  70                     .createServerContext(serverConfig.securityConfiguration)
  71                     .map { sslContext ->
  72                         logger.info { "Collector configured with SSL enabled" }
  73                         this.secure { b -> b.sslContext(sslContext) }
  74                     }.getOrElse {
  75                         logger.info { "Collector configured with SSL disabled" }
  76                         this
  77                     }
  78
  79     private fun handleConnection(nettyInbound: NettyInbound, nettyOutbound: NettyOutbound): Mono<Void> =
  80             messageHandlingStream(nettyInbound, nettyOutbound).run {
  81                 subscribe()
  82                 nettyOutbound.neverComplete()
  83             }
  84
  85     private fun messageHandlingStream(nettyInbound: NettyInbound, nettyOutbound: NettyOutbound): Mono<Void> =
  86             withNewClientContextFrom(nettyInbound, nettyOutbound)
  87             { clientContext ->
  88                 logger.debug(clientContext::fullMdc, Marker.Entry) { "Client connection request received" }
  89
  90                 clientContext.clientAddress
  91                         .map { acceptIfNotLocalConnection(it, clientContext, nettyInbound) }
  92                         .getOrElse {
  93                             logger.warn(clientContext::fullMdc) {
  94                                 "Client address could not be resolved. Discarding connection"
  95                             }
  96                             nettyInbound.closeConnectionAndReturn(Mono.empty())
  97                         }
  98             }
  99
 100     private fun acceptIfNotLocalConnection(address: InetAddress,
 101                                            clientContext: ClientContext,
 102                                            nettyInbound: NettyInbound): Mono<Void> =
 103             if (address.isLocalClientAddress()) {
 104                 logger.debug(clientContext) {
 105                     "Client address resolved to localhost. Discarding connection as suspected healthcheck"
 106                 }
 107                 nettyInbound.closeConnectionAndReturn(Mono.empty<Void>())
 108             } else {
 109                 acceptClientConnection(clientContext, nettyInbound)
 110             }
 111
 112     private fun acceptClientConnection(clientContext: ClientContext, nettyInbound: NettyInbound): Mono<Void> {
 113         metrics.notifyClientConnected()
 114         logger.info(clientContext::fullMdc) { "Handling new client connection" }
 115         return collectorProvider(clientContext).fold(
 116                 {
 117                     logger.warn(clientContext::fullMdc) { "Collector is not ready. Closing connection" }
 118                     nettyInbound.closeConnectionAndReturn(Mono.empty<Void>())
 119                 },
 120                 handleClient(clientContext, nettyInbound)
 121         )
 122     }
 123
 124     private fun handleClient(clientContext: ClientContext,
 125                              nettyInbound: NettyInbound): (Collector) -> Mono<Void> = { collector ->
 126         withConnectionFrom(nettyInbound) { connection ->
 127             connection
 128                     .configureIdleTimeout(clientContext, serverConfig.idleTimeout)
 129                     .logConnectionClosed(clientContext)
 130         }.run {
 131             collector.handleConnection(nettyInbound.createDataStream())
 132         }
 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 }