company: 'ATT'
id: 'jflucas'
timezone: 'America/New_York'
- - name: 'Przemyslaw Wasala '
- email: 'przemyslaw.wasala@nokia.com'
+ - name: 'Piotr Jaszczyk'
+ email: 'piotr.jaszczyk@nokia.com'
company: 'Nokia'
- id: 'przemyslaw.wasala'
+ id: 'jaszczur'
+ timezone: 'Europe/Warsaw'
+ - name: 'Piotr Wielebski'
+ email: 'piotr.wielebski@nokia.com'
+ company: 'Nokia'
+ id: 'pwielebs'
timezone: 'Europe/Warsaw'
tsc:
approval: 'https://lists.onap.org/pipermail/onap-tsc'
+ changes:
+ - type: 'Removal'
+ name: 'Przemyslaw Wasala'
+ link: 'https://lists.onap.org/g/onap-tsc/message/4248'
+ - type: 'Addition'
+ name: 'Piotr Jaszczyk'
+ name: 'Piotr Wielebski'
+ link: 'https://lists.onap.org/g/onap-tsc/message/4259'
--- /dev/null
+#!/usr/bin/env bash
+# ============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=========================================================
+
+set -euo pipefail
+
+usage() {
+ echo "Put HV-VES configuration into Consul key-value store"
+ echo "Usage: $0 [-h|--help] [-v|--verbose] [domain [topic]]"
+ exit 1
+}
+
+optspec=":vh-:" # catch v, h and -
+while getopts "$optspec" arg; do
+ case "${arg}" in
+ -) # handle longopts
+ case "${OPTARG}" in
+ verbose)
+ VERBOSE=True
+ ;;
+ help)
+ usage
+ ;;
+ *)
+ echo "Unknown option --${OPTARG}" >&2
+ usage
+ ;;
+ esac
+ ;;
+ v)
+ VERBOSE=True
+ ;;
+ h)
+ usage
+ ;;
+ *)
+ echo "Unknown option -${OPTARG}" >&2
+ usage
+ ;;
+ esac
+done
+shift $((OPTIND-1))
+
+DOMAIN=${1:-perf3gpp}
+TOPIC=${2:-HV_VES_PERF3GPP}
+
+CONFIGURATION="
+{
+ \"dmaap.kafkaBootstrapServers\": \"message-router-kafka:9092\",
+ \"collector.routing\":
+ [{
+ \"fromDomain\": \"${DOMAIN}\",
+ \"toTopic\": \"${TOPIC}\"
+ }]
+}"
+CONFIGURATION_ENDPOINT=localhost:8500/v1/kv/veshv-config
+
+
+if [ -n "${VERBOSE+x}" ]; then
+ echo "Configuration: ${CONFIGURATION}"
+ echo "Putting configuration under ${CONFIGURATION_ENDPOINT}."
+fi
+curl --request PUT ${CONFIGURATION_ENDPOINT} -d "${CONFIGURATION}"
+echo
--- /dev/null
+#!/usr/bin/env bash
+# ============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=========================================================
+
+set -euo pipefail
+
+usage() {
+ echo "Return current amount of consumed messages by dcae-app-simulator"
+ echo "Usage: $0 [-h|--help] [-v|--verbose]"
+ exit 1
+}
+
+optspec=":vh-:" # catch v, h and -
+while getopts "$optspec" arg; do
+ case "${arg}" in
+ -) # handle longopts
+ case "${OPTARG}" in
+ verbose)
+ VERBOSE=True
+ ;;
+ help)
+ usage
+ ;;
+ *)
+ echo "Unknown option --${OPTARG}" >&2
+ usage
+ ;;
+ esac
+ ;;
+ v)
+ VERBOSE=True
+ ;;
+ h)
+ usage
+ ;;
+ *)
+ echo "Unknown option -${OPTARG}" >&2
+ usage
+ ;;
+ esac
+done
+shift $((OPTIND-1))
+
+if [ -n "${VERBOSE+x}" ]; then
+ echo "All messages count currently consumed by dcae app simulator: "
+fi
+
+curl --request GET localhost:6063/messages/all/count
+echo
--- /dev/null
+#!/usr/bin/env bash
+# ============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=========================================================
+
+set -euo pipefail
+
+
+usage() {
+ echo "Resets dcae-app-simulator consumed messages count"
+ echo "Usage: $0 [-h|--help] [-v|--verbose]"
+ exit 1
+}
+
+optspec=":vh-:" # catch v, h and -
+while getopts "$optspec" arg; do
+ case "${arg}" in
+ -) # handle longopts
+ case "${OPTARG}" in
+ verbose)
+ VERBOSE=True
+ ;;
+ help)
+ usage
+ ;;
+ *)
+ echo "Unknown option --${OPTARG}" >&2
+ usage
+ ;;
+ esac
+ ;;
+ v)
+ VERBOSE=True
+ ;;
+ h)
+ usage
+ ;;
+ *)
+ echo "Unknown option -${OPTARG}" >&2
+ usage
+ ;;
+ esac
+done
+shift $((OPTIND-1))
+
+if [ -n "${VERBOSE+x}" ]; then
+ echo "Requesting DCAE app running on port 6063 to reset messages count"
+fi
+
+curl --request DELETE localhost:6063/messages
+echo
--- /dev/null
+#!/usr/bin/env bash
+# ============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=========================================================
+
+set -euo pipefail
+
+usage() {
+ echo "Set dcae-app-simulator to start consuming messages from given topic (HV_VES_PERF3GPP by default)"
+ echo "Usage: $0 [-h|--help] [-v|--verbose] [topic]"
+ exit 1
+}
+
+optspec=":vh-:" # catch v, h and -
+while getopts "$optspec" arg; do
+ case "${arg}" in
+ -) # handle longopts
+ case "${OPTARG}" in
+ verbose)
+ VERBOSE=True
+ ;;
+ help)
+ usage
+ ;;
+ *)
+ echo "Unknown option --${OPTARG}" >&2
+ usage
+ ;;
+ esac
+ ;;
+ v)
+ VERBOSE=True
+ ;;
+ h)
+ usage
+ ;;
+ *)
+ echo "Unknown option -${OPTARG}" >&2
+ usage
+ ;;
+ esac
+done
+shift $((OPTIND-1))
+
+TOPIC=${1:-HV_VES_PERF3GPP}
+
+if [ -n "${VERBOSE+x}" ]; then
+ echo "Requesting DCAE app running on port 6063 to consume messages from topic: ${TOPIC}"
+fi
+
+curl --request PUT localhost:6063/configuration/topics -d ${TOPIC}
+echo
\ No newline at end of file
--- /dev/null
+#!/usr/bin/env bash
+# ============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=========================================================
+
+set -euo pipefail
+
+usage() {
+ echo "Start xnf-simulator container on given port and inside of given docker-network (by default 'development_default')"
+ echo "Usage: $0 [-h|--help] [-v|--verbose] <xnf listen port> [<hv ves docker network>]"
+ exit 1
+}
+
+optspec=":vh-:" # catch v, h and -
+while getopts "$optspec" arg; do
+ case "${arg}" in
+ -) # handle longopts
+ case "${OPTARG}" in
+ verbose)
+ VERBOSE=True
+ ;;
+ help)
+ usage
+ ;;
+ *)
+ echo "Unknown option --${OPTARG}" >&2
+ usage
+ ;;
+ esac
+ ;;
+ v)
+ VERBOSE=True
+ ;;
+ h)
+ usage
+ ;;
+ *)
+ echo "Unknown option -${OPTARG}" >&2
+ usage
+ ;;
+ esac
+done
+shift $((OPTIND-1))
+
+[ $# -eq 0 ] && usage
+
+
+LISTEN_PORT=$1
+HV_VES_NETWORK=${2:-development_default}
+
+PORTS="${LISTEN_PORT}:${LISTEN_PORT}/tcp"
+HV_VES_REPO_HOME=`pwd`/..
+
+if [ -n "${VERBOSE+x}" ]; then
+ echo "Starting xnf-simulator with ports configuration: ${PORTS} on network: ${HV_VES_NETWORK}"
+ echo "Container id:"
+fi
+docker run -d \
+ -v ${HV_VES_REPO_HOME}/ssl/:/etc/ves-hv/ \
+ -p ${PORTS} \
+ --network ${HV_VES_NETWORK} \
+ onap/org.onap.dcaegen2.collectors.hv-ves.hv-collector-xnf-simulator \
+ --listen-port ${LISTEN_PORT} \
+ --ves-host ves-hv-collector \
+ --ves-port 6061 \
+ --key-store-password onaponap \
+ --trust-store-password onaponap
\ No newline at end of file
--- /dev/null
+#!/usr/bin/env bash
+# ============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=========================================================
+
+set -euo pipefail
+
+usage() {
+ echo "Send request to xnf-simulator"
+ echo "Usage: $0 [-h|--help] [-v|--verbose] [<xnf listen port> [<messages amount> [<messages type> [<xnf endpoint>]]]]"
+ exit 1
+}
+
+optspec=":vh-:" # catch v, h and -
+while getopts "$optspec" arg; do
+ case "${arg}" in
+ -) # handle longopts
+ case "${OPTARG}" in
+ verbose)
+ VERBOSE=True
+ ;;
+ help)
+ usage
+ ;;
+ *)
+ echo "Unknown option --${OPTARG}" >&2
+ usage
+ ;;
+ esac
+ ;;
+ v)
+ VERBOSE=True
+ ;;
+ h)
+ usage
+ ;;
+ *)
+ echo "Unknown option -${OPTARG}" >&2
+ usage
+ ;;
+ esac
+done
+shift $((OPTIND-1))
+
+XNF_PORT=${1:-6062}
+MESSAGES_AMOUNT=${2:-1}
+MESSAGES_TYPE=${3:-VALID}
+XNF_ENDPOINT=simulator/async
+
+if [ -n "${VERBOSE+x}" ]; then
+ echo "Requesting xnf-simulator on port ${XNF_PORT} to send ${MESSAGES_AMOUNT} messages of type ${MESSAGES_TYPE}"
+fi
+
+REQUEST_ID=$(curl --request POST -s localhost:${XNF_PORT}/${XNF_ENDPOINT} -d "
+[
+ {
+ \"commonEventHeader\": {
+ \"version\": \"sample-version\",
+ \"domain\": \"perf3gpp\",
+ \"sequence\": 1,
+ \"priority\": 1,
+ \"eventId\": \"sample-event-id\",
+ \"eventName\": \"sample-event-name\",
+ \"eventType\": \"sample-event-type\",
+ \"startEpochMicrosec\": 120034455,
+ \"lastEpochMicrosec\": 120034455,
+ \"nfNamingCode\": \"sample-nf-naming-code\",
+ \"nfcNamingCode\": \"sample-nfc-naming-code\",
+ \"reportingEntityId\": \"sample-reporting-entity-id\",
+ \"reportingEntityName\": \"sample-reporting-entity-name\",
+ \"sourceId\": \"sample-source-id\",
+ \"sourceName\": \"sample-source-name\",
+ \"vesEventListenerVersion\": \"7.2.0\"
+ },
+ \"messageType\": \"${MESSAGES_TYPE}\",
+ \"messagesAmount\": ${MESSAGES_AMOUNT}
+ }
+]")
+
+if [ -n "${VERBOSE+x}" ]; then
+ echo -e "Request id: ${REQUEST_ID}\n"
+
+ echo "To check request status execute:"
+ echo "curl --request GET localhost:${XNF_PORT}/simulator/${REQUEST_ID}"
+ echo "To further debug you can try something similiar to:"
+ echo "docker ps -a | grep ${XNF_PORT} | awk '{ print \$1 }' | xargs docker logs"
+else
+ echo "${REQUEST_ID}"
+fi
\ No newline at end of file
--- /dev/null
+version: "3.5"
+services:
+
+ #
+ # DMaaP Message Router
+ #
+
+ message-router-zookeeper:
+ image: wurstmeister/zookeeper
+ ports:
+ - "2181:2181"
+
+ message-router-kafka:
+# image: nexus3.onap.org:10001/onap/dmaap/kafka01101:0.0.1
+ image: wurstmeister/kafka
+ ports:
+ - "9092:9092"
+ environment:
+ KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
+ KAFKA_ZOOKEEPER_CONNECT: "message-router-zookeeper:2181"
+ KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "INTERNAL_PLAINTEXT:PLAINTEXT,EXTERNAL_PLAINTEXT:PLAINTEXT"
+ KAFKA_ADVERTISED_LISTENERS: "INTERNAL_PLAINTEXT://message-router-kafka:9092"
+ KAFKA_LISTENERS: "INTERNAL_PLAINTEXT://0.0.0.0:9092"
+ KAFKA_INTER_BROKER_LISTENER_NAME: "INTERNAL_PLAINTEXT"
+ volumes:
+ - /var/run/docker.sock:/var/run/docker.sock
+ depends_on:
+ - message-router-zookeeper
+
+
+ #
+ # Consul / CBS
+ #
+
+ consul-server:
+ image: docker.io/consul:1.0.6
+ ports:
+ - "8500:8500"
+ command: ["agent","-bootstrap", "-client=0.0.0.0", "-server", "-ui"]
+
+ consul-config:
+ image: consul
+ depends_on:
+ - consul-server
+ restart: on-failure
+ command: ["kv", "put", "-http-addr=http://consul-server:8500", "veshv-config", '{
+ "dmaap.kafkaBootstrapServers": "message-router-kafka:9092",
+ "collector.routing": [
+ {
+ "fromDomain": "perf3gpp",
+ "toTopic": "HV_VES_PERF3GPP"
+ }
+ ]
+ }']
+
+
+ #
+ # DCAE HV VES Collector
+ #
+
+ ves-hv-collector:
+ image: onap/org.onap.dcaegen2.collectors.hv-ves.hv-collector-main:latest
+ ports:
+ - "6060:6060"
+ - "6061:6061/tcp"
+ entrypoint: ["java", "-Dio.netty.leakDetection.level=paranoid",
+ "-cp", "*:", "org.onap.dcae.collectors.veshv.main.MainKt"]
+ command: ["--listen-port", "6061",
+ "--health-check-api-port", "6060",
+ "--config-url", "http://consul-server:8500/v1/kv/veshv-config?raw=true",
+ "--key-store-password", "onaponap",
+ "--trust-store-password", "onaponap"]
+ healthcheck:
+ test: curl -f http://localhost:6060/health/ready || exit 1
+ interval: 10s
+ timeout: 3s
+ retries: 3
+ start_period: 20s
+ depends_on:
+ - message-router-kafka
+ - consul-server
+ volumes:
+ - ./ssl/:/etc/ves-hv/
+
+
+ #
+ # Simulators
+ #
+
+ xnf-simulator:
+ image: onap/org.onap.dcaegen2.collectors.hv-ves.hv-collector-xnf-simulator
+ ports:
+ - "6062:6062/tcp"
+ command: ["--listen-port", "6062",
+ "--ves-host", "ves-hv-collector",
+ "--ves-port", "6061",
+ "--key-store-password", "onaponap",
+ "--trust-store-password", "onaponap"]
+ depends_on:
+ - ves-hv-collector
+ volumes:
+ - ./ssl/:/etc/ves-hv/
+
+ dcae-app-simulator:
+ image: onap/org.onap.dcaegen2.collectors.hv-ves.hv-collector-dcae-app-simulator
+ ports:
+ - "6063:6063/tcp"
+ command: ["--listen-port", "6063",
+ "--kafka-bootstrap-servers", "message-router-kafka:9092",
+ "--kafka-topics", "HV_VES_PERF3GPP"]
+ depends_on:
+ - message-router-kafka
+++ /dev/null
-version: "3.5"
-services:
- zookeeper:
- image: wurstmeister/zookeeper
- ports:
- - "2181:2181"
-
- kafka:
- image: wurstmeister/kafka
- ports:
- - "9092:9092"
- environment:
- KAFKA_ADVERTISED_HOST_NAME: "kafka"
- KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
- KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
- KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://kafka:9092"
- volumes:
- - /var/run/docker.sock:/var/run/docker.sock
- depends_on:
- - zookeeper
-
- consul:
- image: progrium/consul
- ports:
- - "8500:8500"
- environment:
- - CONSUL_BIND_INTERFACE=eth0
- command: ["-server", "-bootstrap", "-ui-dir", "/ui"]
-
- ves-hv-collector:
- image: onap/org.onap.dcaegen2.collectors.hv-ves.hv-collector-main:latest
-# build:
-# context: hv-collector-main
-# dockerfile: Dockerfile
- ports:
- - "6060:6060"
- - "6061:6061/tcp"
- entrypoint: ["java", "-Dio.netty.leakDetection.level=paranoid",
- "-cp", "*:", "org.onap.dcae.collectors.veshv.main.MainKt"]
- command: ["--listen-port", "6061",
- "--health-check-api-port", "6060",
- "--config-url", "http://consul:8500/v1/kv/veshv-config?raw=true",
- "--key-store-password", "onaponap",
- "--trust-store-password", "onaponap"]
- healthcheck:
- test: curl -f http://localhost:6060/health/ready || exit 1
- interval: 10s
- timeout: 3s
- retries: 3
- start_period: 20s
- depends_on:
- - kafka
- - consul
- volumes:
- - ./ssl/:/etc/ves-hv/
-
- xnf-simulator:
- image: onap/org.onap.dcaegen2.collectors.hv-ves.hv-collector-xnf-simulator
-# build:
-# context: hv-collector-xnf-simulator
-# dockerfile: Dockerfile
- ports:
- - "6062:6062/tcp"
- command: ["--listen-port", "6062",
- "--ves-host", "ves-hv-collector",
- "--ves-port", "6061",
- "--key-store-password", "onaponap",
- "--trust-store-password", "onaponap"]
- depends_on:
- - ves-hv-collector
- volumes:
- - ./ssl/:/etc/ves-hv/
-
- dcae-app-simulator:
- image: onap/org.onap.dcaegen2.collectors.hv-ves.hv-collector-dcae-app-simulator
-# build:
-# context: hv-collector-dcae-app-simulator
-# dockerfile: Dockerfile
- ports:
- - "6063:6063/tcp"
- command: ["--listen-port", "6063",
- "--kafka-bootstrap-servers", "kafka:9092",
- "--kafka-topics", "HV_VES_PERF3GPP"]
- depends_on:
- - kafka
*/
package org.onap.dcae.collectors.veshv.boundary
+import org.onap.dcae.collectors.veshv.model.ClientContext
import org.onap.dcae.collectors.veshv.model.CollectorConfiguration
import org.onap.dcae.collectors.veshv.model.RoutedMessage
import reactor.core.publisher.Flux
@FunctionalInterface
interface SinkProvider {
- operator fun invoke(config: CollectorConfiguration): Sink
+ operator fun invoke(config: CollectorConfiguration, ctx: ClientContext): Sink
companion object {
fun just(sink: Sink): SinkProvider =
object : SinkProvider {
- override fun invoke(config: CollectorConfiguration): Sink = sink
+ override fun invoke(config: CollectorConfiguration, ctx: ClientContext): Sink = sink
}
}
}
import arrow.effects.IO
import io.netty.buffer.ByteBuf
import io.netty.buffer.ByteBufAllocator
+import org.onap.dcae.collectors.veshv.model.ClientContext
import org.onap.dcae.collectors.veshv.utils.ServerHandle
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
+import java.util.*
interface Collector {
- fun handleConnection(alloc: ByteBufAllocator, dataStream: Flux<ByteBuf>): Mono<Void>
+ fun handleConnection(dataStream: Flux<ByteBuf>): Mono<Void>
}
-typealias CollectorProvider = () -> Option<Collector>
+typealias CollectorProvider = (ClientContext) -> Option<Collector>
interface Server {
fun start(): IO<ServerHandle>
import org.onap.dcae.collectors.veshv.boundary.Metrics
import org.onap.dcae.collectors.veshv.boundary.SinkProvider
import org.onap.dcae.collectors.veshv.domain.WireFrameDecoder
-import org.onap.dcae.collectors.veshv.healthcheck.api.HealthState
import org.onap.dcae.collectors.veshv.healthcheck.api.HealthDescription
+import org.onap.dcae.collectors.veshv.healthcheck.api.HealthState
import org.onap.dcae.collectors.veshv.impl.Router
import org.onap.dcae.collectors.veshv.impl.VesDecoder
import org.onap.dcae.collectors.veshv.impl.VesHvCollector
import org.onap.dcae.collectors.veshv.impl.wire.WireChunkDecoder
+import org.onap.dcae.collectors.veshv.model.ClientContext
import org.onap.dcae.collectors.veshv.model.CollectorConfiguration
import org.onap.dcae.collectors.veshv.utils.arrow.getOption
import org.onap.dcae.collectors.veshv.utils.logging.Logger
private val healthState: HealthState = HealthState.INSTANCE) {
fun createVesHvCollectorProvider(): CollectorProvider {
- val collector: AtomicReference<Collector> = AtomicReference()
+ val config: AtomicReference<CollectorConfiguration> = AtomicReference()
configuration()
- .map(this::createVesHvCollector)
.doOnNext {
- logger.info("Using updated configuration for new connections")
+ logger.info { "Using updated configuration for new connections" }
healthState.changeState(HealthDescription.HEALTHY)
}
.doOnError {
- logger.error("Failed to acquire configuration from consul")
+ logger.error { "Failed to acquire configuration from consul" }
healthState.changeState(HealthDescription.CONSUL_CONFIGURATION_NOT_FOUND)
}
- .subscribe(collector::set)
- return collector::getOption
+ .subscribe(config::set)
+ return { ctx: ClientContext ->
+ config.getOption().map { config -> createVesHvCollector(config, ctx) }
+ }
}
- private fun createVesHvCollector(config: CollectorConfiguration): Collector {
- return VesHvCollector(
- wireChunkDecoderSupplier = { alloc ->
- WireChunkDecoder(WireFrameDecoder(maximumPayloadSizeBytes), alloc)
- },
- protobufDecoder = VesDecoder(),
- router = Router(config.routing),
- sink = sinkProvider(config),
- metrics = metrics)
- }
+ private fun createVesHvCollector(config: CollectorConfiguration, ctx: ClientContext): Collector = VesHvCollector(
+ clientContext = ctx,
+ wireChunkDecoder = WireChunkDecoder(WireFrameDecoder(maximumPayloadSizeBytes), ctx),
+ protobufDecoder = VesDecoder(),
+ router = Router(config.routing, ctx),
+ sink = sinkProvider(config, ctx),
+ metrics = metrics)
companion object {
private val logger = Logger(CollectorFactory::class)
package org.onap.dcae.collectors.veshv.impl
import arrow.core.Option
+import org.onap.dcae.collectors.veshv.model.ClientContext
+import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.debug
import org.onap.dcae.collectors.veshv.model.RoutedMessage
import org.onap.dcae.collectors.veshv.model.Routing
import org.onap.dcae.collectors.veshv.model.VesMessage
+import org.onap.dcae.collectors.veshv.utils.logging.Logger
-class Router(private val routing: Routing) {
+class Router(private val routing: Routing, private val ctx: ClientContext) {
fun findDestination(message: VesMessage): Option<RoutedMessage> =
- routing.routeFor(message.header).map { it(message) }
+ routing.routeFor(message.header).map { it(message) }.also {
+ if (it.isEmpty()) {
+ logger.debug(ctx) { "No route is defined for domain: ${message.header.domain}" }
+ }
+ }
+
+ companion object {
+ private val logger = Logger(Routing::class)
+ }
}
import arrow.core.Either
import io.netty.buffer.ByteBuf
-import io.netty.buffer.ByteBufAllocator
import org.onap.dcae.collectors.veshv.boundary.Collector
import org.onap.dcae.collectors.veshv.boundary.Metrics
import org.onap.dcae.collectors.veshv.boundary.Sink
import org.onap.dcae.collectors.veshv.domain.ByteData
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.RoutedMessage
import org.onap.dcae.collectors.veshv.model.VesMessage
import org.onap.dcae.collectors.veshv.utils.logging.Logger
-import org.onap.dcae.collectors.veshv.utils.logging.filterFailedWithLog
import org.onap.dcae.collectors.veshv.utils.logging.filterEmptyWithLog
-import org.onap.dcae.collectors.veshv.utils.logging.handleReactiveStreamError
+import org.onap.dcae.collectors.veshv.utils.logging.filterFailedWithLog
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
* @since May 2018
*/
internal class VesHvCollector(
- private val wireChunkDecoderSupplier: (ByteBufAllocator) -> WireChunkDecoder,
+ private val clientContext: ClientContext,
+ private val wireChunkDecoder: WireChunkDecoder,
private val protobufDecoder: VesDecoder,
private val router: Router,
private val sink: Sink,
private val metrics: Metrics) : Collector {
- override fun handleConnection(alloc: ByteBufAllocator, dataStream: Flux<ByteBuf>): Mono<Void> =
- wireChunkDecoderSupplier(alloc).let { wireDecoder ->
- dataStream
- .transform { decodeWireFrame(it, wireDecoder) }
- .transform(::filterInvalidWireFrame)
- .transform(::decodeProtobufPayload)
- .transform(::filterInvalidProtobufMessages)
- .transform(::routeMessage)
- .onErrorResume { logger.handleReactiveStreamError(it) }
- .doFinally { releaseBuffersMemory(wireDecoder) }
- .then()
- }
+ override fun handleConnection(dataStream: Flux<ByteBuf>): Mono<Void> =
+ dataStream
+ .transform { decodeWireFrame(it) }
+ .transform(::filterInvalidWireFrame)
+ .transform(::decodeProtobufPayload)
+ .transform(::filterInvalidProtobufMessages)
+ .transform(::routeMessage)
+ .onErrorResume { logger.handleReactiveStreamError(clientContext, it) }
+ .doFinally { releaseBuffersMemory() }
+ .then()
- private fun decodeWireFrame(flux: Flux<ByteBuf>, decoder: WireChunkDecoder): Flux<WireFrameMessage> = flux
+ private fun decodeWireFrame(flux: Flux<ByteBuf>): Flux<WireFrameMessage> = flux
.doOnNext { metrics.notifyBytesReceived(it.readableBytes()) }
- .concatMap(decoder::decode)
+ .concatMap(wireChunkDecoder::decode)
.doOnNext { metrics.notifyMessageReceived(it.payloadSize) }
private fun filterInvalidWireFrame(flux: Flux<WireFrameMessage>): Flux<WireFrameMessage> = flux
private fun decodePayload(rawPayload: ByteData): Flux<VesMessage> = protobufDecoder
.decode(rawPayload)
- .filterFailedWithLog(logger,
+ .filterFailedWithLog(logger, clientContext::asMap,
{ "Ves event header decoded successfully" },
{ "Failed to decode ves event header, reason: ${it.message}" })
private fun findRoute(msg: VesMessage) = router
.findDestination(msg)
- .filterEmptyWithLog(logger,
+ .filterEmptyWithLog(logger, clientContext::asMap,
{ "Found route for message: ${it.topic}, partition: ${it.partition}" },
{ "Could not find route for message" })
- private fun releaseBuffersMemory(wireChunkDecoder: WireChunkDecoder) = wireChunkDecoder.release()
- .also { logger.debug("Released buffer memory after handling message stream") }
+ private fun releaseBuffersMemory() = wireChunkDecoder.release()
+ .also { logger.debug { "Released buffer memory after handling message stream" } }
fun <T> Flux<T>.filterFailedWithLog(predicate: (T) -> Either<() -> String, () -> String>) =
- filterFailedWithLog(logger, predicate)
+ filterFailedWithLog(logger, clientContext::asMap, predicate)
companion object {
private val logger = Logger(VesHvCollector::class)
--- /dev/null
+/*
+ * ============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::asMap, block)
+ fun Logger.withWarn(ctx: ClientContext, block: AtLevelLogger.() -> Unit) = withWarn(ctx::asMap, block)
+ fun Logger.withInfo(ctx: ClientContext, block: AtLevelLogger.() -> Unit) = withInfo(ctx::asMap, block)
+ fun Logger.withDebug(ctx: ClientContext, block: AtLevelLogger.() -> Unit) = withDebug(ctx::asMap, block)
+ fun Logger.withTrace(ctx: ClientContext, block: AtLevelLogger.() -> Unit) = withTrace(ctx::asMap, block)
+
+ fun Logger.error(ctx: ClientContext, message: () -> String) = error(ctx::asMap, message)
+ fun Logger.warn(ctx: ClientContext, message: () -> String) = warn(ctx::asMap, message)
+ fun Logger.info(ctx: ClientContext, message: () -> String) = info(ctx::asMap, message)
+ fun Logger.debug(ctx: ClientContext, message: () -> String) = debug(ctx::asMap, message)
+ fun Logger.trace(ctx: ClientContext, message: () -> String) = trace(ctx::asMap, message)
+
+ fun <T> Logger.handleReactiveStreamError(context: ClientContext, ex: Throwable,
+ returnFlux: Flux<T> = Flux.empty()): Flux<T> {
+ return this.handleReactiveStreamError({ context.asMap() }, ex, returnFlux)
+ }
+}
+
private val lastConfigurationHash: AtomicReference<Int> = AtomicReference(0)
private val retry = retrySpec
.doOnRetry {
- logger.warn("Could not get fresh configuration", it.exception())
+ logger.withWarn { log("Could not get fresh configuration", it.exception()) }
healthState.changeState(HealthDescription.RETRYING_FOR_CONSUL_CONFIGURATION)
}
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.slf4j.LoggerFactory
import reactor.core.publisher.Mono
import reactor.netty.http.client.HttpClient
*/
open class HttpAdapter(private val httpClient: HttpClient) {
- private val logger = LoggerFactory.getLogger(HttpAdapter::class.java)
-
open fun get(url: String, queryParams: Map<String, Any> = emptyMap()): Mono<String> = httpClient
.get()
.uri(url + createQueryString(queryParams))
}
}
.doOnError {
- logger.error("Failed to get resource on path: $url (${it.localizedMessage})")
- logger.debug("Nested exception:", it)
+ logger.error { "Failed to get resource on path: $url (${it.localizedMessage})" }
+ logger.withDebug { log("Nested exception:", it) }
}
private fun createQueryString(params: Map<String, Any>): String {
return builder.removeSuffix("&").toString()
}
+ companion object {
+
+
+ private val logger = Logger(HttpAdapter::class)
+ }
}
import org.onap.dcae.collectors.veshv.boundary.Sink
import org.onap.dcae.collectors.veshv.boundary.SinkProvider
+import org.onap.dcae.collectors.veshv.model.ClientContext
+import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.info
+import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.trace
import org.onap.dcae.collectors.veshv.model.CollectorConfiguration
import org.onap.dcae.collectors.veshv.model.RoutedMessage
import org.onap.dcae.collectors.veshv.utils.logging.Logger
*/
internal class LoggingSinkProvider : SinkProvider {
- override fun invoke(config: CollectorConfiguration): Sink {
+ override fun invoke(config: CollectorConfiguration, ctx: ClientContext): Sink {
return object : Sink {
private val totalMessages = AtomicLong()
private val totalBytes = AtomicLong()
val bytes = totalBytes.addAndGet(msg.message.rawMessage.size().toLong())
val logMessageSupplier = { "Message routed to ${msg.topic}. Total = $msgs ($bytes B)" }
if (msgs % INFO_LOGGING_FREQ == 0L)
- logger.info(logMessageSupplier)
+ logger.info(ctx, logMessageSupplier)
else
- logger.trace(logMessageSupplier)
+ logger.trace(ctx, logMessageSupplier)
}
}
package org.onap.dcae.collectors.veshv.impl.adapters.kafka
import org.onap.dcae.collectors.veshv.boundary.Sink
+import org.onap.dcae.collectors.veshv.model.ClientContext
+import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.withWarn
+import org.onap.dcae.collectors.veshv.utils.logging.Marker
import org.onap.dcae.collectors.veshv.model.RoutedMessage
import org.onap.dcae.collectors.veshv.model.VesMessage
import org.onap.dcae.collectors.veshv.utils.logging.Logger
* @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
* @since May 2018
*/
-internal class KafkaSink(private val sender: KafkaSender<CommonEventHeader, VesMessage>) : Sink {
+internal class KafkaSink(private val sender: KafkaSender<CommonEventHeader, VesMessage>,
+ private val ctx: ClientContext) : Sink {
private val sentMessages = AtomicLong(0)
override fun send(messages: Flux<RoutedMessage>): Flux<RoutedMessage> {
if (it.isSuccessful()) {
Mono.just(it)
} else {
- logger.warn(it.exception()) { "Failed to send message to Kafka" }
+ logger.withWarn(ctx) { log("Failed to send message to Kafka", it.exception()) }
Mono.empty<SenderResult<RoutedMessage>>()
}
}
.map { it.correlationMetadata() }
- return if (logger.traceEnabled) {
- result.doOnNext(::logSentMessage)
- } else {
- result
- }
+ return result.doOnNext(::logSentMessage)
}
private fun vesToKafkaRecord(msg: RoutedMessage): SenderRecord<CommonEventHeader, VesMessage, RoutedMessage> {
}
private fun logSentMessage(sentMsg: RoutedMessage) {
- logger.trace {
+ logger.trace(ctx::asMap, Marker.INVOKE) {
val msgNum = sentMessages.incrementAndGet()
"Message #$msgNum has been sent to ${sentMsg.topic}:${sentMsg.partition}"
}
import org.apache.kafka.clients.producer.ProducerConfig
import org.onap.dcae.collectors.veshv.boundary.Sink
import org.onap.dcae.collectors.veshv.boundary.SinkProvider
+import org.onap.dcae.collectors.veshv.model.ClientContext
import org.onap.dcae.collectors.veshv.model.CollectorConfiguration
import org.onap.dcae.collectors.veshv.model.VesMessage
import org.onap.ves.VesEventOuterClass.CommonEventHeader
* @since June 2018
*/
internal class KafkaSinkProvider : SinkProvider {
- override fun invoke(config: CollectorConfiguration): Sink {
- return KafkaSink(KafkaSender.create(constructSenderOptions(config)))
+ override fun invoke(config: CollectorConfiguration, ctx: ClientContext): Sink {
+ return KafkaSink(KafkaSender.create(constructSenderOptions(config)), ctx)
}
private fun constructSenderOptions(config: CollectorConfiguration) =
import arrow.effects.IO
import org.onap.dcae.collectors.veshv.boundary.CollectorProvider
import org.onap.dcae.collectors.veshv.boundary.Server
+import org.onap.dcae.collectors.veshv.model.ClientContext
+import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.info
+import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.debug
+import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.withWarn
+import org.onap.dcae.collectors.veshv.utils.logging.Marker
import org.onap.dcae.collectors.veshv.model.ServerConfiguration
import org.onap.dcae.collectors.veshv.ssl.boundary.ServerSslContextFactory
import org.onap.dcae.collectors.veshv.utils.NettyServerHandle
sslContextFactory
.createSslContext(serverConfig.securityConfiguration)
.map { sslContext ->
- logger.info("Collector configured with SSL enabled")
+ logger.info { "Collector configured with SSL enabled" }
this.secure { b -> b.sslContext(sslContext) }
}.getOrElse {
- logger.info("Collector configured with SSL disabled")
+ logger.info { "Collector configured with SSL disabled" }
this
}
- private fun handleConnection(nettyInbound: NettyInbound, nettyOutbound: NettyOutbound): Mono<Void> =
- collectorProvider().fold(
- {
- nettyInbound.withConnection { conn ->
- logger.warn { "Collector not ready. Closing connection from ${conn.address()}..." }
- }
- Mono.empty()
- },
- {
- nettyInbound.withConnection { conn ->
- logger.info { "Handling connection from ${conn.address()}" }
- conn.configureIdleTimeout(serverConfig.idleTimeout)
- .logConnectionClosed()
- }
- it.handleConnection(nettyOutbound.alloc(), createDataStream(nettyInbound))
+ private fun handleConnection(nettyInbound: NettyInbound, nettyOutbound: NettyOutbound): Mono<Void> {
+ val clientContext = ClientContext(nettyOutbound.alloc())
+ nettyInbound.withConnection {
+ clientContext.clientAddress = it.address()
+ }
+
+ logger.debug(clientContext::asMap, Marker.ENTRY) { "Client connection request received" }
+ return collectorProvider(clientContext).fold(
+ {
+ logger.warn(clientContext::asMap) { "Collector not ready. Closing connection..." }
+ Mono.empty()
+ },
+ {
+ logger.info(clientContext::asMap) { "Handling new connection" }
+ nettyInbound.withConnection { conn ->
+ conn.configureIdleTimeout(clientContext, serverConfig.idleTimeout)
+ .logConnectionClosed(clientContext)
}
- )
+ it.handleConnection(createDataStream(nettyInbound))
+ }
+ )
+ }
private fun createDataStream(nettyInbound: NettyInbound): ByteBufFlux = nettyInbound
.receive()
.retain()
- private fun Connection.configureIdleTimeout(timeout: Duration): Connection {
+ private fun Connection.configureIdleTimeout(ctx: ClientContext, timeout: Duration): Connection {
onReadIdle(timeout.toMillis()) {
- logger.info {
+ logger.info(ctx) {
"Idle timeout of ${timeout.seconds} s reached. Closing connection from ${address()}..."
}
- disconnectClient()
+ disconnectClient(ctx)
}
return this
}
- private fun Connection.disconnectClient() {
+ private fun Connection.disconnectClient(ctx: ClientContext) {
channel().close().addListener {
+ logger.debug(ctx::asMap, Marker.EXIT) { "Closing client channel." }
if (it.isSuccess)
- logger.debug { "Channel (${address()}) closed successfully." }
+ logger.debug(ctx) { "Channel closed successfully." }
else
- logger.warn("Channel close failed", it.cause())
+ logger.withWarn(ctx) { log("Channel close failed", it.cause()) }
}
}
- private fun Connection.logConnectionClosed(): Connection {
+ private fun Connection.logConnectionClosed(ctx: ClientContext): Connection {
onTerminate().subscribe {
- logger.info("Connection from ${address()} has been closed")
+ // TODO: this code is never executed (at least with ssl-enabled, did not checked with ssl-disabled)
+ logger.info(ctx::asMap, Marker.EXIT) { "Connection has been closed" }
}
return this
}
import arrow.effects.IO
import io.netty.buffer.ByteBuf
-import io.netty.buffer.ByteBufAllocator
-import org.onap.dcae.collectors.veshv.domain.WireFrameMessage
-import org.onap.dcae.collectors.veshv.domain.WireFrameDecoder
import org.onap.dcae.collectors.veshv.domain.InvalidWireFrame
-import org.onap.dcae.collectors.veshv.domain.WireFrameDecodingError
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.model.ClientContext
import org.onap.dcae.collectors.veshv.utils.logging.Logger
-import org.onap.dcae.collectors.veshv.utils.logging.handleReactiveStreamError
+import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.handleReactiveStreamError
+import org.onap.dcae.collectors.veshv.impl.adapters.ClientContextLogging.trace
import reactor.core.publisher.Flux
+import reactor.core.publisher.Flux.defer
import reactor.core.publisher.SynchronousSink
/**
*/
internal class WireChunkDecoder(
private val decoder: WireFrameDecoder,
- alloc: ByteBufAllocator = ByteBufAllocator.DEFAULT) {
- private val streamBuffer = alloc.compositeBuffer()
+ private val ctx: ClientContext) {
+ private val streamBuffer = ctx.alloc.compositeBuffer()
fun release() {
streamBuffer.release()
}
- fun decode(byteBuf: ByteBuf): Flux<WireFrameMessage> = Flux.defer {
+ fun decode(byteBuf: ByteBuf): Flux<WireFrameMessage> = defer {
logIncomingMessage(byteBuf)
if (byteBuf.readableBytes() == 0) {
byteBuf.release()
} else {
streamBuffer.addComponent(true, byteBuf)
generateFrames()
- .onErrorResume { logger.handleReactiveStreamError(it, Flux.error(it)) }
+ .onErrorResume { logger.handleReactiveStreamError(ctx, it, Flux.error(it)) }
.doFinally { streamBuffer.discardReadComponents() }
}
}
}
private fun logIncomingMessage(wire: ByteBuf) {
- logger.trace { "Got message with total size of ${wire.readableBytes()} B" }
+ logger.trace(ctx) { "Got message with total size of ${wire.readableBytes()} B" }
}
private fun logDecodedWireMessage(wire: WireFrameMessage) {
- logger.trace { "Wire payload size: ${wire.payloadSize} B" }
+ logger.trace(ctx) { "Wire payload size: ${wire.payloadSize} B" }
}
private fun logEndOfData() {
- logger.trace { "End of data in current TCP buffer" }
+ logger.trace(ctx) { "End of data in current TCP buffer" }
}
companion object {
--- /dev/null
+/*
+ * ============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 io.netty.buffer.ByteBufAllocator
+import org.onap.dcae.collectors.veshv.utils.logging.AtLevelLogger
+import org.onap.dcae.collectors.veshv.utils.logging.Logger
+import java.net.InetSocketAddress
+import java.util.*
+
+/**
+ * @author Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
+ * @since December 2018
+ */
+data class ClientContext(
+ val alloc: ByteBufAllocator = ByteBufAllocator.DEFAULT,
+ val clientId: String = UUID.randomUUID().toString(),
+ var clientAddress: InetSocketAddress? = null) {
+ fun asMap(): Map<String, String> {
+ val result = mutableMapOf("clientId" to clientId)
+ if (clientAddress != null) {
+ result["clientAddress"] = clientAddress.toString()
+ }
+ return result
+ }
+}
data class Routing(val routes: List<Route>) {
fun routeFor(commonHeader: CommonEventHeader): Option<Route> =
- Option.fromNullable(routes.find { it.applies(commonHeader) }).also {
- if (it.isEmpty()) {
- logger.debug { "No route is defined for domain: ${commonHeader.domain}" }
- }
- }
-
- companion object {
- private val logger = Logger(Routing::class)
- }
+ Option.fromNullable(routes.find { it.applies(commonHeader) })
}
data class Route(val domain: String, val targetTopic: String, val partitioning: (CommonEventHeader) -> Int) {
import arrow.core.None
import arrow.core.Some
+import io.netty.buffer.ByteBufAllocator
import org.assertj.core.api.Assertions.assertThat
import org.jetbrains.spek.api.Spek
import org.jetbrains.spek.api.dsl.given
import org.onap.dcae.collectors.veshv.domain.VesEventDomain.PERF3GPP
import org.onap.dcae.collectors.veshv.domain.VesEventDomain.HEARTBEAT
import org.onap.dcae.collectors.veshv.domain.VesEventDomain.SYSLOG
+import org.onap.dcae.collectors.veshv.model.ClientContext
import org.onap.dcae.collectors.veshv.model.RoutedMessage
import org.onap.dcae.collectors.veshv.model.VesMessage
import org.onap.dcae.collectors.veshv.model.routing
withFixedPartitioning()
}
}.build()
- val cut = Router(config)
+ val cut = Router(config, ClientContext())
on("message with existing route (rtpm)") {
val message = VesMessage(commonHeader(PERF3GPP), ByteData.EMPTY)
import org.onap.dcae.collectors.veshv.domain.WireFrameEncoder
import org.onap.dcae.collectors.veshv.domain.WireFrameDecoder
import org.onap.dcae.collectors.veshv.domain.WireFrameMessage
+import org.onap.dcae.collectors.veshv.model.ClientContext
import reactor.test.test
/**
fun WireChunkDecoder.decode(frame: WireFrameMessage) = decode(encoder.encode(frame))
- fun createInstance() = WireChunkDecoder(WireFrameDecoder(WireFrameMessage.DEFAULT_MAX_PAYLOAD_SIZE_BYTES), alloc)
+ fun createInstance() = WireChunkDecoder(WireFrameDecoder(WireFrameMessage.DEFAULT_MAX_PAYLOAD_SIZE_BYTES), ClientContext(alloc))
fun verifyMemoryReleased(vararg byteBuffers: ByteBuf) {
for (bb in byteBuffers) {
<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ ============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=========================================================
+-->
<configuration>
- <property name="LOG_FILE"
- value="${LOG_FILE:-${LOG_PATH:-${LOG_TEMP:-${java.io.tmpdir:-/tmp}}/}ves-hv.log}"/>
- <property name="FILE_LOG_PATTERN" value="%d{yyyy-MM-dd'T'HH:mm:ss.SSSXXX,UTC} %-5level [%-40.40logger{10}] - %msg%n"/>
+ <property name="p_tim" value="%date{"yyyy-MM-dd'T'HH:mm:ss.SSSXXX", UTC}"/>
+ <property name="p_lvl" value="%highlight(%-5level)"/>
+ <property name="p_log" value="%50.50logger"/>WireFrameCodecsTest
+ <property name="SIMPLE_LOG_PATTERN" value="
+%nopexception
+| ${p_tim}\t
+| ${p_log}\t
+| ${p_lvl}\t
+| %msg%n"/>
- <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
- <encoder>
- <pattern>
- %d{yyyy-MM-dd'T'HH:mm:ss.SSSXXX,UTC} %highlight(%-5level) [%-40.40logger{10}] - %msg%n
- </pattern>
- </encoder>
- </appender>
-
- <appender name="ROLLING-FILE"
- class="ch.qos.logback.core.rolling.RollingFileAppender">
- <encoder>
- <pattern>${FILE_LOG_PATTERN}</pattern>
- </encoder>
- <file>${LOG_FILE}</file>
- <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
- <fileNamePattern>${LOG_FILE}.%d{yyyy-MM-dd}.log</fileNamePattern>
- <maxFileSize>50MB</maxFileSize>
- <maxHistory>30</maxHistory>
- <totalSizeCap>10GB</totalSizeCap>
- </rollingPolicy>
+ <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
+ <encoder>
+ <pattern>${SIMPLE_LOG_PATTERN}</pattern>
+ </encoder>
</appender>
- <logger name="org.onap.dcae.collectors.veshv" level="TRACE"/>
+ <logger name="org.onap.dcae.collectors.veshv" level="TRACE"/>
- <root level="INFO">
- <appender-ref ref="CONSOLE"/>
- <appender-ref ref="ROLLING-FILE"/>
+ <root level="INFO">
+ <appender-ref ref="CONSOLE"/>
</root>
</configuration>
)
val fluxes = (1.rangeTo(runs)).map {
- sut.collector.handleConnection(sut.alloc, generateDataStream(sut.alloc, params))
+ sut.collector.handleConnection(generateDataStream(sut.alloc, params))
}
val durationMs = measureTimeMillis {
Flux.merge(fluxes).then().block(timeout)
val durationSec = durationMs / 1000.0
val throughput = sink.count / durationSec
- logger.info("Processed $runs connections each containing $numMessages msgs.")
- logger.info("Forwarded ${sink.count / ONE_MILION} Mmsgs in $durationSec seconds, that is $throughput msgs/s")
+ logger.info { "Processed $runs connections each containing $numMessages msgs." }
+ logger.info { "Forwarded ${sink.count / ONE_MILION} Mmsgs in $durationSec seconds, that is $throughput msgs/s" }
assertThat(sink.count)
.describedAs("should send all events")
.isEqualTo(runs * numMessages)
val dataStream = generateDataStream(sut.alloc, params)
.transform(::dropWhenIndex.partially1 { it % 101 == 0L })
- sut.collector.handleConnection(sut.alloc, dataStream)
+ sut.collector.handleConnection(dataStream)
.timeout(timeout)
.block()
- logger.info("Forwarded ${sink.count} msgs")
+ logger.info { "Forwarded ${sink.count} msgs" }
assertThat(sink.count)
.describedAs("should send up to number of events")
.isLessThan(numMessages)
import org.onap.dcae.collectors.veshv.boundary.Sink
import org.onap.dcae.collectors.veshv.boundary.SinkProvider
import org.onap.dcae.collectors.veshv.factory.CollectorFactory
+import org.onap.dcae.collectors.veshv.model.ClientContext
import org.onap.dcae.collectors.veshv.model.RoutedMessage
import org.onap.dcae.collectors.veshv.tests.fakes.FakeConfigurationProvider
import org.onap.dcae.collectors.veshv.tests.fakes.FakeHealthState
private val collectorProvider = collectorFactory.createVesHvCollectorProvider()
val collector: Collector
- get() = collectorProvider().getOrElse{ throw IllegalStateException("Collector not available.") }
+ get() = collectorProvider(ClientContext(alloc)).getOrElse{ throw IllegalStateException("Collector not available.") }
companion object {
const val MAX_PAYLOAD_SIZE_BYTES = 1024
}
fun Sut.handleConnection(sink: StoringSink, vararg packets: ByteBuf): List<RoutedMessage> {
- collector.handleConnection(alloc, Flux.fromArray(packets)).block(Duration.ofSeconds(10))
+ collector.handleConnection(Flux.fromArray(packets)).block(Duration.ofSeconds(10))
return sink.sentMessages
}
.map { vesWireFrameMessage(PERF3GPP) }
- sut.collector.handleConnection(sut.alloc, incomingMessages).block(defaultTimeout)
+ sut.collector.handleConnection(incomingMessages).block(defaultTimeout)
val messages = sink.sentMessages
val firstTopicMessagesCount = messages.count { it.topic == PERF3GPP_TOPIC }
<?xml version="1.0" encoding="UTF-8"?>
<configuration>
- <property name="LOG_FILE"
- value="${LOG_FILE:-${LOG_PATH:-${LOG_TEMP:-${java.io.tmpdir:-/tmp}}/}ves-hv.log}"/>
- <property name="FILE_LOG_PATTERN" value="%d{yyyy-MM-dd'T'HH:mm:ss.SSSXXX,UTC} %-5level [%-40.40logger{10}] - %msg%n"/>
+ <property name="LOG_FILE"
+ value="${LOG_FILE:-${LOG_PATH:-${LOG_TEMP:-${java.io.tmpdir:-/tmp}}/}ves-hv.log}"/>
- <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
- <encoder>
- <pattern>
- %d{yyyy-MM-dd'T'HH:mm:ss.SSSXXX,UTC} %highlight(%-5level) [%-40.40logger{10}] - %msg%n
- </pattern>
- </encoder>
- </appender>
+
+ <property name="p_tim" value="%date{"yyyy-MM-dd'T'HH:mm:ss.SSSXXX", UTC}"/>
+ <property name="p_thr" value="%thread"/>
+ <property name="p_lvl" value="%highlight(%-5level)"/>
+ <property name="p_log" value="%50.50logger"/>
+ <property name="p_mdc" value="%replace(%replace(%mdc){'\t', '\\\\t'}){'\n', '\\\\n'}"/>
+ <property name="p_msg" value="%replace(%replace(%msg){'\t', '\\\\t'}){'\n','\\\\n'}"/>
+ <property name="p_exc" value="%replace(%replace(%rootException){'\t', '\\\\t'}){'\n','\\\\n'}"/>
+ <property name="p_mak" value="%replace(%replace(%marker){'\t', '\\\\t'}){'\n','\\\\n'}"/>
+ <property name="ONAP_LOG_PATTERN" value="
+%nopexception
+| ${p_tim}\t
+| ${p_thr}\t
+| ${p_lvl}\t
+| ${p_log}\t
+| ${p_mdc}\t
+| ${p_msg}\t
+| ${p_exc}\t
+| ${p_mak}%n"/>
+
+ <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
+ <encoder>
+ <pattern>${ONAP_LOG_PATTERN}</pattern>
+ </encoder>
+ </appender>
<appender name="ROLLING-FILE"
- class="ch.qos.logback.core.rolling.RollingFileAppender">
- <encoder>
- <pattern>${FILE_LOG_PATTERN}</pattern>
- </encoder>
- <file>${LOG_FILE}</file>
- <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
- <fileNamePattern>${LOG_FILE}.%d{yyyy-MM-dd}.log</fileNamePattern>
- <maxFileSize>50MB</maxFileSize>
- <maxHistory>30</maxHistory>
- <totalSizeCap>10GB</totalSizeCap>
- </rollingPolicy>
+ class="ch.qos.logback.core.rolling.RollingFileAppender">
+ <encoder>
+ <pattern>${ONAP_LOG_PATTERN}</pattern>
+ </encoder>
+ <file>${LOG_FILE}</file>
+ <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
+ <fileNamePattern>${LOG_FILE}.%d{yyyy-MM-dd}.log</fileNamePattern>
+ <maxFileSize>50MB</maxFileSize>
+ <maxHistory>30</maxHistory>
+ <totalSizeCap>10GB</totalSizeCap>
+ </rollingPolicy>
</appender>
- <logger name="org.onap.dcae.collectors.veshv" level="DEBUG"/>
+ <logger name="org.onap.dcae.collectors.veshv" level="DEBUG"/>
- <root level="INFO">
- <appender-ref ref="CONSOLE"/>
- <appender-ref ref="ROLLING-FILE"/>
+ <root level="INFO">
+
<appender-ref ref="CONSOLE"/>
+
<appender-ref ref="ROLLING-FILE"/>
</root>
</configuration>
throw IllegalArgumentException(message)
}
- logger.info("Received new configuration. Creating consumer for topics: $topics")
+ logger.info { "Received new configuration. Creating consumer for topics: $topics" }
consumerState.set(consumerFactory.createConsumerForTopics(topics).bind())
}.fix()
return messageParams.fold(
{
logger.warn { "Error while parsing message parameters: ${it::class.qualifiedName} : ${it.message}" }
- logger.debug { "Detailed stack trace: ${it}" }
+ logger.debug { "Detailed stack trace: $it" }
throw IllegalArgumentException("Parsing error: " + it.message)
},
{
if (it.isEmpty()) {
val message = "Message param list cannot be empty"
- logger.warn(message)
+ logger.warn { message }
throw IllegalArgumentException(message)
}
it
}
.delete("messages") { ctx ->
ctx.response.contentType(CONTENT_TEXT)
- logger.info("Resetting simulator state")
+ logger.info { "Resetting simulator state" }
ctx.response.sendOrError(simulator.resetState())
}
.get("messages/all/count") { ctx ->
- logger.info("Processing request for count of received messages")
+ logger.info { "Processing request for count of received messages" }
simulator.state().fold(
{
ctx.response.status(HttpConstants.STATUS_NOT_FOUND)
- logger.warn("Error - number of messages could not be specified")
+ logger.warn { "Error - number of messages could not be specified" }
},
{
logger.info { "Returned number of received messages: ${it.messagesCount}" }
}
.post("messages/all/validate") { ctx ->
ctx.request.body.then { body ->
- logger.info("Processing request for message validation")
+ logger.info { "Processing request for message validation" }
val response = simulator.validate(body.inputStream)
.map { isValid ->
if (isValid) {
.map(::startApp)
.unsafeRunEitherSync(
{ ex ->
- logger.error("Failed to start a server", ex)
+ logger.withError { log("Failed to start a server", ex) }
ExitFailure(1)
},
{
- logger.info("Started DCAE-APP Simulator API server")
+ logger.info { "Started DCAE-APP Simulator API server" }
}
)
private fun startApp(config: DcaeAppSimConfiguration): IO<Unit> {
- logger.info("Using configuration: $config")
+ logger.info { "Using configuration: $config" }
val consumerFactory = ConsumerFactory(config.kafkaBootstrapServers)
val messageStreamValidation = MessageStreamValidation(MessageGeneratorFactory.create(config.maxPayloadSizeBytes))
return DcaeAppApiServer(DcaeAppSimulator(consumerFactory, messageStreamValidation))
<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ ============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=========================================================
+-->
<configuration>
- <property name="LOG_FILE"
- value="${LOG_FILE:-${LOG_PATH:-${LOG_TEMP:-${java.io.tmpdir:-/tmp}}/}ves-hv.log}"/>
- <property name="FILE_LOG_PATTERN" value="%d{yyyy-MM-dd'T'HH:mm:ss.SSSXXX,UTC} %-5level [%-40.40logger{10}] - %msg%n"/>
+ <property name="COMPONENT_NAME"
+ value="hv-ves-dcae-app-simulator"/>
+ <property name="COMPONENT_SHORT_NAME"
+ value="dcae-app-simulator"/>
- <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
- <encoder>
- <pattern>
- %d{yyyy-MM-dd'T'HH:mm:ss.SSSXXX,UTC} %highlight(%-5level) [%-40.40logger{10}] - %msg%n
- </pattern>
- </encoder>
- </appender>
+ <property name="LOG_FILENAME" value="${COMPONENT_SHORT_NAME}"/>
+ <property name="LOG_PATH" value="/var/log/ONAP/${COMPONENT_NAME}"/>
+ <property name="ARCHIVE" value="${LOG_PATH}/archive"/>
+
+ <property name="p_tim" value="%date{"yyyy-MM-dd'T'HH:mm:ss.SSSXXX", UTC}"/>
+ <property name="p_thr" value="%thread"/>
+ <property name="p_lvl" value="%highlight(%-5level)"/>
+ <property name="p_log" value="%50.50logger"/>
+ <property name="p_mdc" value="%replace(%replace(%mdc){'\t', '\\\\t'}){'\n', '\\\\n'}"/>
+ <property name="p_msg" value="%replace(%replace(%msg){'\t', '\\\\t'}){'\n','\\\\n'}"/>
+ <property name="p_exc" value="%replace(%replace(%rootException){'\t', '\\\\t'}){'\n','\\\\n'}"/>
+ <property name="p_mak" value="%replace(%replace(%marker){'\t', '\\\\t'}){'\n','\\\\n'}"/>
+ <property name="SIMPLE_LOG_PATTERN" value="
+%nopexception
+| ${p_tim}\t
+| ${p_log}\t
+| ${p_lvl}\t
+| %msg\t
+| %rootException%n"/>
+ <property name="READABLE_LOG_PATTERN" value="
+%nopexception
+| ${p_tim}\t
+| ${p_log}\t
+| ${p_lvl}\t
+| %msg\t
+| ${p_mak}\t
+| %rootException\t
+| ${p_mdc}\t
+| ${p_thr}%n"/>
+ <property name="ONAP_LOG_PATTERN" value="
+%nopexception
+| ${p_tim}\t
+| ${p_thr}\t
+| ${p_lvl}\t
+| ${p_log}\t
+| ${p_mdc}\t
+| ${p_msg}\t
+| ${p_exc}\t
+| ${p_mak}%n"/>
+ <property name="LOG_PATTERN_IN_USE" value="${SIMPLE_LOG_PATTERN}"/>
+
+ <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
+ <encoder>
+ <pattern>${LOG_PATTERN_IN_USE}</pattern>
+ </encoder>
+ </appender>
<appender name="ROLLING-FILE"
- class="ch.qos.logback.core.rolling.RollingFileAppender">
- <encoder>
- <pattern>${FILE_LOG_PATTERN}</pattern>
- </encoder>
- <file>${LOG_FILE}</file>
- <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
- <fileNamePattern>${LOG_FILE}.%d{yyyy-MM-dd}.log</fileNamePattern>
- <maxFileSize>50MB</maxFileSize>
- <maxHistory>30</maxHistory>
- <totalSizeCap>10GB</totalSizeCap>
- </rollingPolicy>
+ class="ch.qos.logback.core.rolling.RollingFileAppender">
+ <encoder>
+ <pattern>${LOG_PATTERN_IN_USE}</pattern>
+ </encoder>
+ <file>${LOG_PATH}/${LOG_FILENAME}.log</file>
+ <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
+ <FileNamePattern>${ARCHIVE}/${LOG_FILENAME}.%d{yyyy-MM-dd}.%i.log.gz</FileNamePattern>
+ <maxFileSize>50MB</maxFileSize>
+ <maxHistory>30</maxHistory>
+ <totalSizeCap>10GB</totalSizeCap>
+ </rollingPolicy>
</appender>
- <logger name="org.onap.dcae.collectors.veshv" level="DEBUG"/>
- <!--<logger name="reactor.netty" level="DEBUG"/>-->
+ <logger name="org.onap.dcae.collectors.veshv" level="DEBUG"/>
+ <!--<logger name="reactor.netty" level="DEBUG"/>-->
- <root level="INFO">
- <appender-ref ref="CONSOLE"/>
- <appender-ref ref="ROLLING-FILE"/>
+ <root level="INFO">
+
<appender-ref ref="CONSOLE"/>
+
<appender-ref ref="ROLLING-FILE"/>
</root>
</configuration>
\ No newline at end of file
+++ /dev/null
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
- ~ ============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=========================================================
- -->
-<configuration>
- <property name="LOG_FILE"
- value="${LOG_FILE:-${LOG_PATH:-${LOG_TEMP:-${java.io.tmpdir:-/tmp}}/}ves-hv.log}"/>
- <property name="FILE_LOG_PATTERN" value="%d{yyyy-MM-dd'T'HH:mm:ss.SSSXXX,UTC} %-5level [%-40.40logger{10}] - %msg%n"/>
-
- <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
- <encoder>
- <pattern>
- %d{yyyy-MM-dd'T'HH:mm:ss.SSSXXX,UTC} %highlight(%-5level) [%-40.40logger{10}] - %msg%n
- </pattern>
- </encoder>
- </appender>
-
- <appender name="ROLLING-FILE"
- class="ch.qos.logback.core.rolling.RollingFileAppender">
- <encoder>
- <pattern>${FILE_LOG_PATTERN}</pattern>
- </encoder>
- <file>${LOG_FILE}</file>
- <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
- <fileNamePattern>${LOG_FILE}.%d{yyyy-MM-dd}.log</fileNamePattern>
- <maxFileSize>50MB</maxFileSize>
- <maxHistory>30</maxHistory>
- <totalSizeCap>10GB</totalSizeCap>
- </rollingPolicy>
- </appender>
-
- <logger name="org.onap.dcae.collectors.veshv" level="DEBUG"/>
-
- <root level="INFO">
- <appender-ref ref="CONSOLE"/>
- <appender-ref ref="ROLLING-FILE"/>
- </root>
-</configuration>
\ No newline at end of file
<groupId>io.arrow-kt</groupId>
<artifactId>arrow-effects</artifactId>
</dependency>
+ <dependency>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-classic</artifactId>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>org.jetbrains.kotlin</groupId>
<artifactId>kotlin-test</artifactId>
.map(::startAndAwaitServers)
.unsafeRunEitherSync(
{ ex ->
- logger.error("Failed to start a server", ex)
+ logger.withError { log("Failed to start a server", ex) }
ExitFailure(1)
},
- { logger.info("Gentle shutdown") }
+ { logger.info { "Gentle shutdown" } }
)
private fun startAndAwaitServers(config: ServerConfiguration) =
IO.monad().binding {
- logger.info("Using configuration: $config")
+ logger.info { "Using configuration: $config" }
HealthCheckServer.start(config).bind()
VesServer.start(config).bind()
.await().bind()
abstract class ServerStarter {
fun start(config: ServerConfiguration): IO<ServerHandle> =
startServer(config)
- .map { logger.info(serverStartedMessage(it)); it }
+ .map { logger.info { serverStartedMessage(it) }; it }
protected abstract fun startServer(config: ServerConfiguration): IO<ServerHandle>
protected abstract fun serverStartedMessage(handle: ServerHandle): String
<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ ============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=========================================================
+-->
<configuration>
<property name="COMPONENT_NAME"
value="dcae-hv-ves-collector"/>
<property name="LOG_FILENAME" value="${COMPONENT_SHORT_NAME}"/>
<property name="LOG_PATH" value="/var/log/ONAP/${COMPONENT_NAME}"/>
<property name="ARCHIVE" value="${LOG_PATH}/archive"/>
- <property name="FILE_LOG_PATTERN" value="
-%nopexception%50.50logger
-| %date{yyyy-MM-dd'T'HH:mm:ss.SSSXXX,UTC}
-| %highlight(%-5level)
-| %msg
-| %rootException
-| %thread%n"/>
+
+ <property name="p_tim" value="%date{"yyyy-MM-dd'T'HH:mm:ss.SSSXXX", UTC}"/>
+ <property name="p_thr" value="%thread"/>
+ <property name="p_lvl" value="%highlight(%-5level)"/>
+ <property name="p_log" value="%50.50logger"/>
+ <property name="p_mdc" value="%replace(%replace(%mdc){'\t', '\\\\t'}){'\n', '\\\\n'}"/>
+ <property name="p_msg" value="%replace(%replace(%msg){'\t', '\\\\t'}){'\n','\\\\n'}"/>
+ <property name="p_exc" value="%replace(%replace(%rootException){'\t', '\\\\t'}){'\n','\\\\n'}"/>
+ <property name="p_mak" value="%replace(%replace(%marker){'\t', '\\\\t'}){'\n','\\\\n'}"/>
+ <property name="SIMPLE_LOG_PATTERN" value="
+%nopexception
+| ${p_tim}\t
+| ${p_log}\t
+| ${p_lvl}\t
+| %msg\t
+| %rootException%n"/>
+ <property name="READABLE_LOG_PATTERN" value="
+%nopexception
+| ${p_tim}\t
+| ${p_log}\t
+| ${p_lvl}\t
+| %msg\t
+| ${p_mak}\t
+| %rootException\t
+| ${p_mdc}\t
+| ${p_thr}%n"/>
+ <property name="ONAP_LOG_PATTERN" value="
+%nopexception
+| ${p_tim}\t
+| ${p_thr}\t
+| ${p_lvl}\t
+| ${p_log}\t
+| ${p_mdc}\t
+| ${p_msg}\t
+| ${p_exc}\t
+| ${p_mak}%n"/>
<appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
- <pattern>${FILE_LOG_PATTERN}</pattern>
+ <pattern>${READABLE_LOG_PATTERN}</pattern>
</encoder>
</appender>
<appender name="ROLLING-FILE"
class="ch.qos.logback.core.rolling.RollingFileAppender">
<encoder>
- <pattern>${FILE_LOG_PATTERN}</pattern>
+ <pattern>${ONAP_LOG_PATTERN}</pattern>
</encoder>
<file>${LOG_PATH}/${LOG_FILENAME}.log</file>
<rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ ============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=========================================================
+-->
<configuration>
- <property name="LOG_FILE"
- value="${LOG_FILE:-${LOG_PATH:-${LOG_TEMP:-${java.io.tmpdir:-/tmp}}/}ves-hv.log}"/>
- <property name="FILE_LOG_PATTERN" value="%d{yyyy-MM-dd'T'HH:mm:ss.SSSXXX,UTC} %-5level [%-40.40logger{10}] - %msg%n"/>
+ <property name="p_tim" value="%date{"yyyy-MM-dd'T'HH:mm:ss.SSSXXX", UTC}"/>
+ <property name="p_lvl" value="%highlight(%-5level)"/>
+ <property name="p_log" value="%50.50logger"/>
+ <property name="SIMPLE_LOG_PATTERN" value="
+%nopexception
+| ${p_tim}\t
+| ${p_log}\t
+| ${p_lvl}\t
+| %msg%n"/>
<appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
- <pattern>
- %d{yyyy-MM-dd'T'HH:mm:ss.SSSXXX,UTC} %highlight(%-5level) [%-40.40logger{10}] - %msg%n
- </pattern>
+ <pattern>${SIMPLE_LOG_PATTERN}</pattern>
</encoder>
</appender>
- <appender name="ROLLING-FILE"
- class="ch.qos.logback.core.rolling.RollingFileAppender">
- <encoder>
- <pattern>${FILE_LOG_PATTERN}</pattern>
- </encoder>
- <file>${LOG_FILE}</file>
- <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
- <fileNamePattern>${LOG_FILE}.%d{yyyy-MM-dd}.log</fileNamePattern>
- <maxFileSize>50MB</maxFileSize>
- <maxHistory>30</maxHistory>
- <totalSizeCap>10GB</totalSizeCap>
- </rollingPolicy>
- </appender>
-
<logger name="org.onap.dcae.collectors.veshv" level="TRACE"/>
<root level="INFO">
- <appender-ref ref="CONSOLE"/>
- <appender-ref ref="ROLLING-FILE"/>
- </root>
+ <appender-ref ref="CONSOLE"/>
+ </root>
</configuration>
private val logger = Logger("org.onap.dcae.collectors.veshv.tests.utils")
object Assertions : org.assertj.core.api.Assertions() {
- fun <A,B> assertThat(actual: Either<A, B>) = EitherAssert(actual)
+ fun <A,
B> assertThat(actual: Either<A, B>) = EitherAssert(actual)
}
while (tryNum <= retries) {
tryNum++
try {
- logger.debug("Try number $tryNum")
+ logger.debug { "Try number $tryNum" }
action()
break
} catch (ex: Throwable) {
fun ratpack.http.Response.sendAndHandleErrors(response: IO<Response>) {
response.attempt().unsafeRunSync().fold(
{ err ->
- logger.warn("Error occurred. Sending .", err)
+ logger.withWarn { log("Error occurred. Sending .", err) }
val message = err.message
send(errorResponse(message))
},
import kotlin.reflect.KClass
import org.slf4j.LoggerFactory
+import org.slf4j.MDC
+
+typealias MappedDiagnosticContext = () -> Map<String, String>
@Suppress("TooManyFunctions", "SuboptimalLoggerUsage")
-class Logger(val logger: org.slf4j.Logger) {
+class Logger(logger: org.slf4j.Logger) {
constructor(clazz: KClass<out Any>) : this(LoggerFactory.getLogger(clazz.java))
constructor(name: String) : this(LoggerFactory.getLogger(name))
- //
- // TRACE
- //
+ private val errorLogger = if (logger.isErrorEnabled) ErrorLevelLogger(logger) else OffLevelLogger
+ private val warnLogger = if (logger.isWarnEnabled) WarnLevelLogger(logger) else OffLevelLogger
+ private val infoLogger = if (logger.isInfoEnabled) InfoLevelLogger(logger) else OffLevelLogger
+ private val debugLogger = if (logger.isDebugEnabled) DebugLevelLogger(logger) else OffLevelLogger
+ private val traceLogger = if (logger.isTraceEnabled) TraceLevelLogger(logger) else OffLevelLogger
- val traceEnabled: Boolean
- get() = logger.isTraceEnabled
+ // ERROR
- fun trace(messageProvider: () -> String) {
- if (logger.isTraceEnabled) {
- logger.trace(messageProvider())
- }
+ fun withError(block: AtLevelLogger.() -> Unit) = errorLogger.block()
+
+ fun withError(mdc: MappedDiagnosticContext, block: AtLevelLogger.() -> Unit) =
+ errorLogger.withMdc(mdc, block)
+
+ fun error(message: () -> String) = errorLogger.run {
+ log(message())
}
- //
+ fun error(mdc: MappedDiagnosticContext, message: () -> String) =
+ errorLogger.withMdc(mdc) { log(message()) }
+
+ fun error(mdc: MappedDiagnosticContext, marker: Marker, message: () -> String) =
+ errorLogger.withMdc(mdc) { log(marker, message()) }
+
+ // WARN
+
+ fun withWarn(block: AtLevelLogger.() -> Unit) = warnLogger.block()
+
+ fun withWarn(mdc: MappedDiagnosticContext, block: AtLevelLogger.() -> Unit) =
+ warnLogger.withMdc(mdc, block)
+
+ fun warn(message: () -> String) = warnLogger.run {
+ log(message())
+ }
+
+ fun warn(mdc: MappedDiagnosticContext, message: () -> String) =
+ warnLogger.withMdc(mdc) { log(message()) }
+
+ fun warn(mdc: MappedDiagnosticContext, marker: Marker, message: () -> String) =
+ warnLogger.withMdc(mdc) { log(marker, message()) }
+
+ // INFO
+
+ fun withInfo(block: AtLevelLogger.() -> Unit) = infoLogger.block()
+
+ fun withInfo(mdc: MappedDiagnosticContext, block: AtLevelLogger.() -> Unit) =
+ infoLogger.withMdc(mdc, block)
+
+ fun info(message: () -> String) = infoLogger.run {
+ log(message())
+ }
+
+ fun info(mdc: MappedDiagnosticContext, message: () -> String) =
+ infoLogger.withMdc(mdc) { log(message()) }
+
+ fun info(mdc: MappedDiagnosticContext, marker: Marker, message: () -> String) =
+ infoLogger.withMdc(mdc) { log(marker, message()) }
+
// DEBUG
- //
- fun debug(message: String) {
- logger.debug(message)
+ fun withDebug(block: AtLevelLogger.() -> Unit) = debugLogger.block()
+
+ fun withDebug(mdc: MappedDiagnosticContext, block: AtLevelLogger.() -> Unit) =
+ debugLogger.withMdc(mdc, block)
+
+ fun debug(message: () -> String) = debugLogger.run {
+ log(message())
}
- fun debug(message: String, t: Throwable) {
- logger.debug(message, t)
+ fun debug(mdc: MappedDiagnosticContext, message: () -> String) =
+ debugLogger.withMdc(mdc) { log(message()) }
+
+ fun debug(mdc: MappedDiagnosticContext, marker: Marker, message: () -> String) =
+ debugLogger.withMdc(mdc) { log(marker, message()) }
+
+ // TRACE
+
+ fun withTrace(block: AtLevelLogger.() -> Unit) = traceLogger.block()
+
+ fun withTrace(mdc: MappedDiagnosticContext, block: AtLevelLogger.() -> Unit) =
+ traceLogger.withMdc(mdc, block)
+
+ fun trace(message: () -> String) = traceLogger.run {
+ log(message())
}
- fun debug(messageProvider: () -> String) {
- if (logger.isDebugEnabled) {
- logger.debug(messageProvider())
+ fun trace(mdc: MappedDiagnosticContext, message: () -> String) =
+ traceLogger.withMdc(mdc) { log(message()) }
+
+ fun trace(mdc: MappedDiagnosticContext, marker: Marker, message: () -> String) =
+ traceLogger.withMdc(mdc) { log(marker, message()) }
+
+}
+
+abstract class AtLevelLogger {
+ abstract fun log(message: String)
+ abstract fun log(message: String, t: Throwable)
+ abstract fun log(marker: Marker, message: String)
+ open val enabled: Boolean
+ get() = true
+
+ inline fun withMdc(mdc: MappedDiagnosticContext, block: AtLevelLogger.() -> Unit) {
+ if (enabled) {
+ try {
+ MDC.setContextMap(mdc())
+ block()
+ } finally {
+ MDC.clear()
+ }
}
}
+}
- fun debug(t: Throwable, messageProvider: () -> String) {
- if (logger.isDebugEnabled) {
- logger.debug(messageProvider(), t)
- }
+object OffLevelLogger : AtLevelLogger() {
+ override val enabled = false
+
+ override fun log(message: String) {
+ // do not log anything
}
- //
- // INFO
- //
- fun info(message: String) {
- logger.info(message)
+ override fun log(message: String, t: Throwable) {
+ // do not log anything
}
- fun info(messageProvider: () -> String) {
- if (logger.isInfoEnabled) {
- logger.info(messageProvider())
- }
+ override fun log(marker: Marker, message: String) {
+ // do not log anything
}
+}
- fun info(message: String, t: Throwable) {
- logger.info(message, t)
+@Suppress("SuboptimalLoggerUsage")
+class ErrorLevelLogger(private val logger: org.slf4j.Logger) : AtLevelLogger() {
+ override fun log(message: String) {
+ logger.error(message)
}
- fun info(t: Throwable, messageProvider: () -> String) {
- if (logger.isInfoEnabled) {
- logger.info(messageProvider(), t)
- }
+ override fun log(message: String, t: Throwable) {
+ logger.error(message, t)
}
- //
- // WARN
- //
+ override fun log(marker: Marker, message: String) {
+ logger.error(marker(), message)
+ }
+}
- fun warn(message: String) {
+@Suppress("SuboptimalLoggerUsage")
+class WarnLevelLogger(private val logger: org.slf4j.Logger) : AtLevelLogger() {
+ override fun log(message: String) {
logger.warn(message)
}
- fun warn(message: String, t: Throwable) {
+ override fun log(message: String, t: Throwable) {
logger.warn(message, t)
}
- fun warn(messageProvider: () -> String) {
- if (logger.isWarnEnabled) {
- logger.warn(messageProvider())
- }
+ override fun log(marker: Marker, message: String) {
+ logger.warn(marker(), message)
}
+}
- fun warn(t: Throwable, messageProvider: () -> String) {
- if (logger.isWarnEnabled) {
- logger.warn(messageProvider(), t)
- }
+@Suppress("SuboptimalLoggerUsage")
+class InfoLevelLogger(private val logger: org.slf4j.Logger) : AtLevelLogger() {
+ override fun log(message: String) {
+ logger.info(message)
}
- //
- // ERROR
- //
+ override fun log(message: String, t: Throwable) {
+ logger.info(message, t)
+ }
- fun error(message: String) {
- logger.error(message)
+ override fun log(marker: Marker, message: String) {
+ logger.info(marker(), message)
}
+}
- fun error(message: String, t: Throwable) {
- logger.error(message, t)
+@Suppress("SuboptimalLoggerUsage")
+class DebugLevelLogger(private val logger: org.slf4j.Logger) : AtLevelLogger() {
+ override fun log(message: String) {
+ logger.debug(message)
}
- fun error(messageProvider: () -> String) {
- if (logger.isErrorEnabled) {
- logger.error(messageProvider())
- }
+ override fun log(message: String, t: Throwable) {
+ logger.debug(message, t)
}
- fun error(t: Throwable, messageProvider: () -> String) {
- if (logger.isErrorEnabled) {
- logger.error(messageProvider(), t)
- }
+ override fun log(marker: Marker, message: String) {
+ logger.debug(marker(), message)
+ }
+}
+
+@Suppress("SuboptimalLoggerUsage")
+class TraceLevelLogger(private val logger: org.slf4j.Logger) : AtLevelLogger() {
+ override fun log(message: String) {
+ logger.trace(message)
+ }
+
+ override fun log(message: String, t: Throwable) {
+ logger.trace(message, t)
+ }
+
+ override fun log(marker: Marker, message: String) {
+ logger.trace(marker(), message)
}
}
--- /dev/null
+/*
+ * ============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.utils.logging
+
+import org.slf4j.MarkerFactory
+
+enum class Marker(private val marker: org.slf4j.Marker) {
+ ENTRY(MarkerFactory.getMarker("ENTRY")),
+ EXIT(MarkerFactory.getMarker("EXIT")),
+ INVOKE(MarkerFactory.getMarker("INVOKE"));
+
+ operator fun invoke() = marker
+}
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
-fun <T> Logger.handleReactiveStreamError(ex: Throwable, returnFlux: Flux<T> = Flux.empty()): Flux<T> {
- logger.warn("Error while handling message stream: ${ex::class.qualifiedName} (${ex.localizedMessage})")
- logger.debug("Detailed stack trace", ex)
+fun <T> Logger.handleReactiveStreamError(
+ context: MappedDiagnosticContext,
+ ex: Throwable,
+ returnFlux: Flux<T> = Flux.empty()): Flux<T> {
+ warn(context) { "Error while handling message stream: ${ex::class.qualifiedName} (${ex.localizedMessage})" }
+ withDebug(context) { log("Detailed stack trace", ex) }
return returnFlux
}
-
fun <T> Try<T>.filterFailedWithLog(logger: Logger,
+ context: MappedDiagnosticContext,
acceptedMsg: (T) -> String,
rejectedMsg: (Throwable) -> String): Flux<T> =
- fold({
- logger.warn(rejectedMsg(it))
+ fold({ ex ->
+ logger.withWarn(context) { log(rejectedMsg(ex)) }
Flux.empty<T>()
- }, {
- logger.trace { acceptedMsg(it) }
- Flux.just(it)
+ }, { obj ->
+ logger.trace(context) { acceptedMsg(obj) }
+ Flux.just(obj)
})
fun <T> Option<T>.filterEmptyWithLog(logger: Logger,
+ context: MappedDiagnosticContext,
acceptedMsg: (T) -> String,
rejectedMsg: () -> String): Flux<T> =
fold({
- logger.warn(rejectedMsg)
+ logger.warn(context, rejectedMsg)
Flux.empty<T>()
}, {
- logger.trace { acceptedMsg(it) }
+ logger.trace(context) { acceptedMsg(it) }
Flux.just(it)
})
-fun <T> Flux<T>.filterFailedWithLog(logger: Logger, predicate: (T) -> Either<() -> String, () -> String>) =
+fun <T> Flux<T>.filterFailedWithLog(logger: Logger,
+ context: MappedDiagnosticContext,
+ predicate: (T) -> Either<() -> String, () -> String>) =
flatMap { t ->
predicate(t).fold({
- logger.warn(it)
+ logger.warn(context, it)
Mono.empty<T>()
}, {
- logger.trace(it)
+ logger.trace(context, it)
Mono.just<T>(t)
})
}
object LoggerTest : Spek({
lateinit var slf4jLogger: org.slf4j.Logger
- lateinit var cut: Logger
+ fun cut() = Logger(slf4jLogger).also {
+ verify(slf4jLogger).isTraceEnabled
+ verify(slf4jLogger).isDebugEnabled
+ verify(slf4jLogger).isInfoEnabled
+ verify(slf4jLogger).isWarnEnabled
+ verify(slf4jLogger).isErrorEnabled
+ }
beforeEachTest {
slf4jLogger = mock()
- cut = Logger(slf4jLogger)
}
afterEachTest {
val exception = Exception("fail")
describe("debug levels") {
- it("should log message") {
- cut.debug(message)
- verify(slf4jLogger).debug(message)
- }
-
- it("should log message with exception") {
- cut.debug(message, exception)
- verify(slf4jLogger).debug(message, exception)
- }
describe("lazy logging message") {
it("should log when debug is ON") {
whenever(slf4jLogger.isDebugEnabled).thenReturn(true)
- cut.debug { message }
+ cut().debug { message }
verify(slf4jLogger).isDebugEnabled
verify(slf4jLogger).debug(message)
}
it("should not log when debug is OFF") {
whenever(slf4jLogger.isDebugEnabled).thenReturn(false)
- cut.debug { message }
+ cut().debug { message }
verify(slf4jLogger).isDebugEnabled
}
}
it("should log when debug is ON") {
whenever(slf4jLogger.isDebugEnabled).thenReturn(true)
- cut.debug(exception) { message }
+ cut().withDebug { log(message, exception) }
verify(slf4jLogger).isDebugEnabled
verify(slf4jLogger).debug(message, exception)
}
it("should not log when debug is OFF") {
whenever(slf4jLogger.isDebugEnabled).thenReturn(false)
- cut.debug(exception) { message }
+ cut().withDebug { log(message, exception) }
verify(slf4jLogger).isDebugEnabled
}
}
}
describe("info levels") {
- it("should log message") {
- cut.info(message)
- verify(slf4jLogger).info(message)
- }
-
- it("should log message with exception") {
- cut.info(message, exception)
- verify(slf4jLogger).info(message, exception)
- }
describe("lazy logging message") {
it("should log when debug is ON") {
whenever(slf4jLogger.isInfoEnabled).thenReturn(true)
- cut.info { message }
+ cut().info { message }
verify(slf4jLogger).isInfoEnabled
verify(slf4jLogger).info(message)
}
it("should not log when debug is OFF") {
whenever(slf4jLogger.isInfoEnabled).thenReturn(false)
- cut.info { message }
+ cut().info { message }
verify(slf4jLogger).isInfoEnabled
}
}
it("should log when debug is ON") {
whenever(slf4jLogger.isInfoEnabled).thenReturn(true)
- cut.info(exception) { message }
+ cut().withInfo { log(message, exception) }
verify(slf4jLogger).isInfoEnabled
verify(slf4jLogger).info(message, exception)
}
it("should not log when debug is OFF") {
whenever(slf4jLogger.isInfoEnabled).thenReturn(false)
- cut.info(exception) { message }
+ cut().withInfo { log(message, exception) }
verify(slf4jLogger).isInfoEnabled
}
}
}
describe("warning levels") {
- it("should log message") {
- cut.warn(message)
- verify(slf4jLogger).warn(message)
- }
-
- it("should log message with exception") {
- cut.warn(message, exception)
- verify(slf4jLogger).warn(message, exception)
- }
-
describe("lazy logging message") {
it("should log when debug is ON") {
whenever(slf4jLogger.isWarnEnabled).thenReturn(true)
- cut.warn { message }
+ cut().warn { message }
verify(slf4jLogger).isWarnEnabled
verify(slf4jLogger).warn(message)
}
it("should not log when debug is OFF") {
whenever(slf4jLogger.isWarnEnabled).thenReturn(false)
- cut.warn { message }
+ cut().warn { message }
verify(slf4jLogger).isWarnEnabled
}
}
it("should log when debug is ON") {
whenever(slf4jLogger.isWarnEnabled).thenReturn(true)
- cut.warn(exception) { message }
+ cut().withWarn { log(message, exception) }
verify(slf4jLogger).isWarnEnabled
verify(slf4jLogger).warn(message, exception)
}
it("should not log when debug is OFF") {
whenever(slf4jLogger.isWarnEnabled).thenReturn(false)
- cut.warn(exception) { message }
+ cut().withWarn { log(message, exception) }
verify(slf4jLogger).isWarnEnabled
}
}
}
describe("error levels") {
- it("should log message") {
- cut.error(message)
- verify(slf4jLogger).error(message)
- }
-
- it("should log message with exception") {
- cut.error(message, exception)
- verify(slf4jLogger).error(message, exception)
- }
describe("lazy logging message") {
it("should log when debug is ON") {
whenever(slf4jLogger.isErrorEnabled).thenReturn(true)
- cut.error { message }
+ cut().error { message }
verify(slf4jLogger).isErrorEnabled
verify(slf4jLogger).error(message)
}
it("should not log when debug is OFF") {
whenever(slf4jLogger.isErrorEnabled).thenReturn(false)
- cut.error { message }
+ cut().error { message }
verify(slf4jLogger).isErrorEnabled
}
}
it("should log when debug is ON") {
whenever(slf4jLogger.isErrorEnabled).thenReturn(true)
- cut.error(exception) { message }
+ cut().withError { log(message, exception) }
verify(slf4jLogger).isErrorEnabled
verify(slf4jLogger).error(message, exception)
}
it("should not log when debug is OFF") {
whenever(slf4jLogger.isErrorEnabled).thenReturn(false)
- cut.error(exception) { message }
+ cut().withError { log(message, exception) }
verify(slf4jLogger).isErrorEnabled
}
}
val cut = Try.just(event)
it("should not filter stream event and log accepted message") {
- cut.filterFailedWithLog(logger, ACCEPTED_MESSAGE, FAILED_WITH_EXCEPTION_MESSAGE)
+ cut.filterFailedWithLog(logger, ::emptyMap, ACCEPTED_MESSAGE, FAILED_WITH_EXCEPTION_MESSAGE)
.test()
.expectNext(event)
.verifyComplete()
val e = Exception()
val cut = Failure(e)
it("should filter stream event and log rejected message") {
- cut.filterFailedWithLog(logger, ACCEPTED_MESSAGE, FAILED_WITH_EXCEPTION_MESSAGE)
+ cut.filterFailedWithLog(logger, ::emptyMap, ACCEPTED_MESSAGE, FAILED_WITH_EXCEPTION_MESSAGE)
.test()
.verifyComplete()
}
val cut = Option.just(event)
it("should not filter stream event and log accepted message") {
- cut.filterEmptyWithLog(logger, ACCEPTED_MESSAGE, FAILED_MESSAGE)
+ cut.filterEmptyWithLog(logger, ::emptyMap, ACCEPTED_MESSAGE, FAILED_MESSAGE)
.test()
.expectNext(event)
.verifyComplete()
given("empty Option") {
val cut = Option.empty<Int>()
it("should filter stream event and log rejected message") {
- cut.filterEmptyWithLog(logger, ACCEPTED_MESSAGE, FAILED_MESSAGE)
+ cut.filterEmptyWithLog(logger,::emptyMap, ACCEPTED_MESSAGE, FAILED_MESSAGE)
.test()
.verifyComplete()
}
val cut = Flux.just(event)
it("should not filter stream event and log accepted message") {
- cut.filterFailedWithLog(logger, right())
+ cut.filterFailedWithLog(logger,::emptyMap, right())
.test()
.expectNext(event)
.verifyComplete()
val cut = Flux.just(event)
it("should filter stream event and log rejected message") {
- cut.filterFailedWithLog(logger, left())
+ cut.filterFailedWithLog(logger,::emptyMap, left())
.test()
.verifyComplete()
}
<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ ============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=========================================================
+-->
<configuration>
- <property name="LOG_FILE"
- value="${LOG_FILE:-${LOG_PATH:-${LOG_TEMP:-${java.io.tmpdir:-/tmp}}/}ves-hv.log}"/>
- <property name="FILE_LOG_PATTERN" value="%d{yyyy-MM-dd'T'HH:mm:ss.SSSXXX,UTC} %-5level [%-40.40logger{10}] - %msg%n"/>
+ <property name="p_tim" value="%date{"yyyy-MM-dd'T'HH:mm:ss.SSSXXX", UTC}"/>
+ <property name="p_lvl" value="%highlight(%-5level)"/>
+ <property name="p_log" value="%50.50logger"/>
+ <property name="SIMPLE_LOG_PATTERN" value="
+%nopexception
+| ${p_tim}\t
+| ${p_log}\t
+| ${p_lvl}\t
+| %msg%n"/>
<appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
- <pattern>
- %d{yyyy-MM-dd'T'HH:mm:ss.SSSXXX,UTC} %highlight(%-5level) [%-40.40logger{10}] - %msg%n
- </pattern>
+ <pattern>${SIMPLE_LOG_PATTERN}</pattern>
</encoder>
</appender>
- <appender name="ROLLING-FILE"
- class="ch.qos.logback.core.rolling.RollingFileAppender">
- <encoder>
- <pattern>${FILE_LOG_PATTERN}</pattern>
- </encoder>
- <file>${LOG_FILE}</file>
- <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
- <fileNamePattern>${LOG_FILE}.%d{yyyy-MM-dd}.log</fileNamePattern>
- <maxFileSize>50MB</maxFileSize>
- <maxHistory>30</maxHistory>
- <totalSizeCap>10GB</totalSizeCap>
- </rollingPolicy>
- </appender>
-
<logger name="org.onap.dcae.collectors.veshv" level="TRACE"/>
<root level="INFO">
- <appender-ref ref="CONSOLE"/>
- <appender-ref ref="ROLLING-FILE"/>
- </root>
+ <appender-ref ref="CONSOLE"/>
+ </root>
</configuration>
.handle { _, output -> handler(complete, messages, output) }
.connect()
.doOnError {
- logger.info("Failed to connect to VesHvCollector on " +
- "${configuration.vesHost}:${configuration.vesPort}")
+ logger.info {
+ "Failed to connect to VesHvCollector on ${configuration.vesHost}:${configuration.vesPort}"
+ }
}
.subscribe {
- logger.info("Connected to VesHvCollector on " +
- "${configuration.vesHost}:${configuration.vesPort}")
+ logger.info {
+ "Connected to VesHvCollector on ${configuration.vesHost}:${configuration.vesPort}"
+ }
}
return complete.then()
}
.options { it.flushOnBoundary() }
.sendGroups(frames)
.then {
- logger.info("Messages have been sent")
+ logger.info { "Messages have been sent" }
complete.onComplete()
}
.then()
.post("simulator/async", ::startSimulationHandler)
.get("simulator/:id", ::simulatorStatusHandler)
.get("healthcheck") { ctx ->
- logger.info("Checking health")
+ logger.info { "Checking health" }
ctx.response.status(HttpConstants.STATUS_OK).send()
}
}
private fun startSimulationHandler(ctx: Context) {
- logger.info("Attempting to start asynchronous scenario")
+ logger.info { "Attempting to start asynchronous scenario" }
ctx.request.body.then { body ->
val id = startSimulation(body)
when (id) {
- is Either.Left -> logger.warn { "Failed to start scenario, ${id.a}"}
+ is Either.Left -> logger.warn { "Failed to start scenario, ${id.a}" }
is Either.Right -> logger.info { "Scenario started, details: ${id.b}" }
}
ctx.response.sendEitherErrorOrResponse(id)
}
private fun simulatorStatusHandler(ctx: Context) {
- logger.debug("Checking task status")
+ logger.debug { "Checking task status" }
val id = UUID.fromString(ctx.pathTokens["id"])
logger.debug { "Checking status for id: $id" }
val status = ongoingSimulations.status(id)
simulationIo.continueOn(asyncSimulationContext).unsafeRunAsync { result ->
result.fold(
{ err ->
- logger.warn("Error", err)
+ logger.withWarn { log("Error", err) }
simulations[id] = StatusFailure(err)
},
{
- logger.info("Finished sending messages")
+ logger.info { "Finished sending messages" }
simulations[id] = StatusSuccess
}
)
fun main(args: Array<String>) = ArgXnfSimulatorConfiguration().parse(args)
.mapLeft(handleWrongArgumentErrorCurried(PROGRAM_NAME))
.map { config ->
- logger.info("Using configuration: $config")
+ logger.info { "Using configuration: $config" }
val xnfSimulator = XnfSimulator(
VesHvClient(config),
MessageGeneratorFactory.create(config.maxPayloadSizeBytes))
}
.unsafeRunEitherSync(
{ ex ->
- logger.error("Failed to start a server", ex)
+ logger.withError { log("Failed to start a server", ex) }
ExitFailure(1)
},
{
- logger.info("Started xNF Simulator API server")
+ logger.info { "Started xNF Simulator API server" }
}
)
<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ ============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=========================================================
+-->
<configuration>
- <property name="LOG_FILE"
- value="${LOG_FILE:-${LOG_PATH:-${LOG_TEMP:-${java.io.tmpdir:-/tmp}}/}ves-hv.log}"/>
- <property name="FILE_LOG_PATTERN" value="%d{yyyy-MM-dd'T'HH:mm:ss.SSSXXX,UTC} %-5level [%-40.40logger{10}] - %msg%n"/>
+ <property name="COMPONENT_NAME"
+ value="hv-ves-xnf-simulator"/>
+ <property name="COMPONENT_SHORT_NAME"
+ value="xnf-simulatr"/>
- <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
- <encoder>
- <pattern>
- %d{yyyy-MM-dd'T'HH:mm:ss.SSSXXX,UTC} %highlight(%-5level) [%-40.40logger{10}] - %msg%n
- </pattern>
- </encoder>
- </appender>
+ <property name="LOG_FILENAME" value="${COMPONENT_SHORT_NAME}"/>
+ <property name="LOG_PATH" value="/var/log/ONAP/${COMPONENT_NAME}"/>
+ <property name="ARCHIVE" value="${LOG_PATH}/archive"/>
+
+ <property name="p_tim" value="%date{"yyyy-MM-dd'T'HH:mm:ss.SSSXXX", UTC}"/>
+ <property name="p_thr" value="%thread"/>
+ <property name="p_lvl" value="%highlight(%-5level)"/>
+ <property name="p_log" value="%50.50logger"/>
+ <property name="p_mdc" value="%replace(%replace(%mdc){'\t', '\\\\t'}){'\n', '\\\\n'}"/>
+ <property name="p_msg" value="%replace(%replace(%msg){'\t', '\\\\t'}){'\n','\\\\n'}"/>
+ <property name="p_exc" value="%replace(%replace(%rootException){'\t', '\\\\t'}){'\n','\\\\n'}"/>
+ <property name="p_mak" value="%replace(%replace(%marker){'\t', '\\\\t'}){'\n','\\\\n'}"/>
+ <property name="SIMPLE_LOG_PATTERN" value="
+%nopexception
+| ${p_tim}\t
+| ${p_log}\t
+| ${p_lvl}\t
+| %msg\t
+| %rootException%n"/>
+ <property name="READABLE_LOG_PATTERN" value="
+%nopexception
+| ${p_tim}\t
+| ${p_log}\t
+| ${p_lvl}\t
+| %msg\t
+| ${p_mak}\t
+| %rootException\t
+| ${p_mdc}\t
+| ${p_thr}%n"/>
+ <property name="ONAP_LOG_PATTERN" value="
+%nopexception
+| ${p_tim}\t
+| ${p_thr}\t
+| ${p_lvl}\t
+| ${p_log}\t
+| ${p_mdc}\t
+| ${p_msg}\t
+| ${p_exc}\t
+| ${p_mak}%n"/>
+ <property name="LOG_PATTERN_IN_USE" value="${SIMPLE_LOG_PATTERN}"/>
+
+ <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
+ <encoder>
+ <pattern>${LOG_PATTERN_IN_USE}</pattern>
+ </encoder>
+ </appender>
<appender name="ROLLING-FILE"
- class="ch.qos.logback.core.rolling.RollingFileAppender">
- <encoder>
- <pattern>${FILE_LOG_PATTERN}</pattern>
- </encoder>
- <file>${LOG_FILE}</file>
- <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
- <fileNamePattern>${LOG_FILE}.%d{yyyy-MM-dd}.log</fileNamePattern>
- <maxFileSize>50MB</maxFileSize>
- <maxHistory>30</maxHistory>
- <totalSizeCap>10GB</totalSizeCap>
- </rollingPolicy>
+ class="ch.qos.logback.core.rolling.RollingFileAppender">
+ <encoder>
+ <pattern>${LOG_PATTERN_IN_USE}</pattern>
+ </encoder>
+ <file>${LOG_PATH}/${LOG_FILENAME}.log</file>
+ <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
+ <FileNamePattern>${ARCHIVE}/${LOG_FILENAME}.%d{yyyy-MM-dd}.%i.log.gz</FileNamePattern>
+ <maxFileSize>50MB</maxFileSize>
+ <maxHistory>30</maxHistory>
+ <totalSizeCap>10GB</totalSizeCap>
+ </rollingPolicy>
</appender>
- <logger name="org.onap.dcae.collectors.veshv" level="DEBUG"/>
+ <logger name="org.onap.dcae.collectors.veshv" level="DEBUG"/>
- <root level="INFO">
- <appender-ref ref="CONSOLE"/>
- <appender-ref ref="ROLLING-FILE"/>
+ <root level="INFO">
+
<appender-ref ref="CONSOLE"/>
+
<appender-ref ref="ROLLING-FILE"/>
</root>
</configuration>
\ No newline at end of file
Code Review / dcaegen2 / collectors / hv-ves.git / commitdiff