Add metric for rejected clients count

- renamed few counters to be more verbose about what they count
- removed not needed 'total' suffix in metrics name

Change-Id: I6be0201e5f39f1298706c536b12410413d49df19
Issue-ID: DCAEGEN2-1043
Signed-off-by: Filip Krzywka <filip.krzywka@nokia.com>

4 files changed, 31 insertions, 2 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 1334738a..61d28c2b 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,6 +21,7 @@ package org.onap.dcae.collectors.veshv.boundary
 
 import org.onap.dcae.collectors.veshv.domain.WireFrameMessage
 import org.onap.dcae.collectors.veshv.model.ClientContext
+import org.onap.dcae.collectors.veshv.model.ClientRejectionCause
 import org.onap.dcae.collectors.veshv.model.CollectorConfiguration
 import org.onap.dcae.collectors.veshv.model.MessageDropCause
 import org.onap.dcae.collectors.veshv.model.RoutedMessage
@@ -35,6 +36,7 @@ interface Metrics {
     fun notifyMessageReceived(msg: WireFrameMessage)
     fun notifyMessageSent(msg: RoutedMessage)
     fun notifyMessageDropped(cause: MessageDropCause)
+    fun notifyClientRejected(cause: ClientRejectionCause)
 }
 
 @FunctionalInterface
diff --git a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/VesHvCollector.kt b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/VesHvCollector.kt
index 51f894d3..5c3f339c 100644
--- a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/VesHvCollector.kt
+++ b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/VesHvCollector.kt
@@ -27,6 +27,7 @@ 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.wire.WireChunkDecoder
 import org.onap.dcae.collectors.veshv.model.ClientContext
+import org.onap.dcae.collectors.veshv.model.ClientRejectionCause
 import org.onap.dcae.collectors.veshv.model.MessageDropCause.INVALID_MESSAGE
 import org.onap.dcae.collectors.veshv.model.MessageDropCause.ROUTE_NOT_FOUND
 import org.onap.dcae.collectors.veshv.model.RoutedMessage
@@ -60,7 +61,9 @@ internal class VesHvCollector(
                     .transform(::decodeProtobufPayload)
                     .transform(::filterInvalidProtobufMessages)
                     .transform(::routeMessage)
-                    .onErrorResume { logger.handleReactiveStreamError(clientContext, it) }
+                    .onErrorResume {
+                        metrics.notifyClientRejected(ClientRejectionCause.fromThrowable(it))
+                        logger.handleReactiveStreamError(clientContext, it) }
                     .doFinally { releaseBuffersMemory() }
                     .then()
 
diff --git a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireFrameException.kt b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireFrameException.kt
index 83a7cd85..81845400 100644
--- a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireFrameException.kt
+++ b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/wire/WireFrameException.kt
@@ -25,5 +25,5 @@ import org.onap.dcae.collectors.veshv.domain.WireFrameDecodingError
  * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
  * @since June 2018
  */
-class WireFrameException(error: WireFrameDecodingError)
+class WireFrameException(val error: WireFrameDecodingError)
     : Exception("${error::class.simpleName}: ${error.message}")
diff --git a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/model/MessageDropCause.kt b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/model/stream_interruption_cause.kt
index af43ae67..836eab53 100644
--- a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/model/MessageDropCause.kt
+++ b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/model/stream_interruption_cause.kt
@@ -19,6 +19,10 @@
  */
 package org.onap.dcae.collectors.veshv.model
 
+import org.onap.dcae.collectors.veshv.domain.InvalidWireFrameMarker
+import org.onap.dcae.collectors.veshv.domain.PayloadSizeExceeded
+import org.onap.dcae.collectors.veshv.impl.wire.WireFrameException
+
 /**
  * @author Jakub Dudycz <jakub.dudycz@nokia.com>
  * @since December 2018
@@ -27,3 +31,23 @@ enum class MessageDropCause(val tag: String) {
     ROUTE_NOT_FOUND("routing"),
     INVALID_MESSAGE("invalid")
 }
+
+enum class ClientRejectionCause(val tag: String) {
+    INVALID_WIRE_FRAME_MARKER("invalid_marker"),
+    PAYLOAD_SIZE_EXCEEDED_IN_MESSAGE("too_big_payload"),
+    UNEXPECTED_STREAM_ERROR("unexpected");
+
+    companion object {
+        fun fromThrowable(err: Throwable): ClientRejectionCause =
+                when (err) {
+                    is WireFrameException -> fromWireFrameException(err)
+                    else -> UNEXPECTED_STREAM_ERROR
+                }
+
+        private fun fromWireFrameException(err: WireFrameException) = when (err.error) {
+            is InvalidWireFrameMarker -> INVALID_WIRE_FRAME_MARKER
+            is PayloadSizeExceeded -> PAYLOAD_SIZE_EXCEEDED_IN_MESSAGE
+            else -> UNEXPECTED_STREAM_ERROR
+        }
+    }
+}