summaryrefslogtreecommitdiffstats
path: root/sources
diff options
context:
space:
mode:
Diffstat (limited to 'sources')
-rw-r--r--sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaSenderOptionsFactory.kt (renamed from sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/kafka.kt)53
-rw-r--r--sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaSinkFactory.kt11
-rw-r--r--sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaSenderOptionsFactoryTest.kt119
-rw-r--r--sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/ProtobufSerializerTest.kt2
-rw-r--r--sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/VesMessageSerializerTest.kt2
-rw-r--r--sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/kotlin.kt27
-rw-r--r--sources/hv-collector-utils/src/test/kotlin/org/onap/dcae/collectors/veshv/utils/kotlin_test.kt43
7 files changed, 225 insertions, 32 deletions
diff --git a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/kafka.kt b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaSenderOptionsFactory.kt
index b16ad109..1c4acf64 100644
--- a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/kafka.kt
+++ b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaSenderOptionsFactory.kt
@@ -17,36 +17,32 @@
* limitations under the License.
* ============LICENSE_END=========================================================
*/
-package org.onap.dcae.collectors.veshv.impl
+package org.onap.dcae.collectors.veshv.impl.adapters.kafka
import org.apache.kafka.clients.CommonClientConfigs
import org.apache.kafka.clients.producer.ProducerConfig
import org.apache.kafka.common.config.SaslConfigs
import org.apache.kafka.common.security.auth.SecurityProtocol
import org.apache.kafka.common.security.plain.internals.PlainSaslServer
+import org.jetbrains.annotations.Nullable
import org.onap.dcae.collectors.veshv.domain.VesMessage
-import org.onap.dcae.collectors.veshv.impl.adapters.kafka.ProtobufSerializer
-import org.onap.dcae.collectors.veshv.impl.adapters.kafka.VesMessageSerializer
-import org.onap.dcaegen2.services.sdk.model.streams.SinkStream
+import org.onap.dcae.collectors.veshv.utils.applyIf
+import org.onap.dcaegen2.services.sdk.model.streams.AafCredentials
import org.onap.dcaegen2.services.sdk.model.streams.dmaap.KafkaSink
import org.onap.ves.VesEventOuterClass.CommonEventHeader
-import reactor.kafka.sender.KafkaSender
import reactor.kafka.sender.SenderOptions
+internal object KafkaSenderOptionsFactory {
-private const val MAXIMUM_REQUEST_SIZE_MULTIPLIER = 1.2f
-private const val BUFFER_MEMORY_MULTIPLIER = 32
-private const val MINIMUM_BUFFER_MEMORY = 32 * 1024 * 1024
+ private const val MAXIMUM_REQUEST_SIZE_MULTIPLIER = 1.2f
+ private const val BUFFER_MEMORY_MULTIPLIER = 32
+ private const val MINIMUM_BUFFER_MEMORY = 32 * 1024 * 1024
-private const val LOGIN_MODULE_CLASS = "org.apache.kafka.common.security.plain.PlainLoginModule"
-private const val USERNAME = "admin"
-private const val PASSWORD = "admin_secret"
-private const val JAAS_CONFIG = "$LOGIN_MODULE_CLASS required username=$USERNAME password=$PASSWORD;"
-private val SASL_PLAINTEXT = (SecurityProtocol.SASL_PLAINTEXT as Enum<SecurityProtocol>).name
+ private const val LOGIN_MODULE_CLASS = "org.apache.kafka.common.security.plain.PlainLoginModule"
+ private val SASL_PLAINTEXT = (SecurityProtocol.SASL_PLAINTEXT as Enum<SecurityProtocol>).name
-internal fun createKafkaSender(sinkStream: SinkStream) =
- (sinkStream as KafkaSink).let { kafkaSink ->
- KafkaSender.create(SenderOptions.create<CommonEventHeader, VesMessage>()
+ fun createSenderOptions(kafkaSink: KafkaSink): SenderOptions<CommonEventHeader, VesMessage> =
+ SenderOptions.create<CommonEventHeader, VesMessage>()
.producerProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaSink.bootstrapServers())
.producerProperty(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, maxRequestSize(kafkaSink))
.producerProperty(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory(kafkaSink))
@@ -55,15 +51,22 @@ internal fun createKafkaSender(sinkStream: SinkStream) =
.producerProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1)
.producerProperty(ProducerConfig.RETRIES_CONFIG, 1)
.producerProperty(ProducerConfig.ACKS_CONFIG, "1")
- .producerProperty(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SASL_PLAINTEXT)
- .producerProperty(SaslConfigs.SASL_MECHANISM, PlainSaslServer.PLAIN_MECHANISM)
- .producerProperty(SaslConfigs.SASL_JAAS_CONFIG, JAAS_CONFIG)
.stopOnError(false)
- )
- }
+ .applyIf(kafkaSink.aafCredentials() != null) {
+ producerProperty(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SASL_PLAINTEXT)
+ .producerProperty(SaslConfigs.SASL_MECHANISM, PlainSaslServer.PLAIN_MECHANISM)
+ .producerProperty(SaslConfigs.SASL_JAAS_CONFIG, jaasConfig(kafkaSink.aafCredentials()!!))
+ }
-private fun maxRequestSize(kafkaSink: KafkaSink) =
- (MAXIMUM_REQUEST_SIZE_MULTIPLIER * kafkaSink.maxPayloadSizeBytes()).toInt()
+ private fun jaasConfig(aafCredentials: AafCredentials) =
+ """$LOGIN_MODULE_CLASS required username="${aafCredentials.username().jaasEscape()}" password="${aafCredentials.password().jaasEscape()}";"""
-private fun bufferMemory(kafkaSink: KafkaSink) =
- Integer.max(MINIMUM_BUFFER_MEMORY, BUFFER_MEMORY_MULTIPLIER * kafkaSink.maxPayloadSizeBytes())
+ private fun String?.jaasEscape() = this?.replace("\"", "\\\"")
+
+ private fun maxRequestSize(kafkaSink: KafkaSink) =
+ (MAXIMUM_REQUEST_SIZE_MULTIPLIER * kafkaSink.maxPayloadSizeBytes()).toInt()
+
+ private fun bufferMemory(kafkaSink: KafkaSink) =
+ Integer.max(MINIMUM_BUFFER_MEMORY, BUFFER_MEMORY_MULTIPLIER * kafkaSink.maxPayloadSizeBytes())
+
+}
diff --git a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaSinkFactory.kt b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaSinkFactory.kt
index 2973fa8d..58363a26 100644
--- a/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaSinkFactory.kt
+++ b/sources/hv-collector-core/src/main/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaSinkFactory.kt
@@ -21,11 +21,11 @@ package org.onap.dcae.collectors.veshv.impl.adapters.kafka
import org.onap.dcae.collectors.veshv.boundary.SinkFactory
import org.onap.dcae.collectors.veshv.domain.VesMessage
-import org.onap.dcae.collectors.veshv.impl.createKafkaSender
import org.onap.dcae.collectors.veshv.domain.logging.ClientContext
import org.onap.dcae.collectors.veshv.domain.logging.ServiceContext
import org.onap.dcae.collectors.veshv.utils.logging.Logger
import org.onap.dcaegen2.services.sdk.model.streams.SinkStream
+import org.onap.dcaegen2.services.sdk.model.streams.dmaap.KafkaSink
import org.onap.ves.VesEventOuterClass.CommonEventHeader
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
@@ -39,13 +39,11 @@ import java.util.Collections.synchronizedMap
*/
internal class KafkaSinkFactory : SinkFactory {
private val messageSinks = synchronizedMap(
- mutableMapOf<SinkStream, KafkaSender<CommonEventHeader, VesMessage>>()
+ mutableMapOf<KafkaSink, KafkaSender<CommonEventHeader, VesMessage>>()
)
override fun invoke(stream: SinkStream, ctx: ClientContext) = lazy {
- messageSinks.computeIfAbsent(stream, ::createKafkaSender).let {
- KafkaPublisher(it, ctx)
- }
+ KafkaPublisher(messageSinks.computeIfAbsent(stream as KafkaSink, this::createKafkaSender), ctx)
}
override fun close(): Mono<Void> =
@@ -57,6 +55,9 @@ internal class KafkaSinkFactory : SinkFactory {
logger.info(ServiceContext::mdc) { "Message sinks flushed and closed" }
}
+ private fun createKafkaSender(stream: KafkaSink) =
+ KafkaSender.create(KafkaSenderOptionsFactory.createSenderOptions(stream))
+
companion object {
private val logger = Logger(KafkaSinkFactory::class)
}
diff --git a/sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaSenderOptionsFactoryTest.kt b/sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaSenderOptionsFactoryTest.kt
new file mode 100644
index 00000000..fec17856
--- /dev/null
+++ b/sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/KafkaSenderOptionsFactoryTest.kt
@@ -0,0 +1,119 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2019 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.kafka
+
+import org.apache.kafka.clients.CommonClientConfigs
+import org.apache.kafka.clients.producer.ProducerConfig
+import org.apache.kafka.common.KafkaException
+import org.apache.kafka.common.config.SaslConfigs
+import org.assertj.core.api.Assertions.assertThat
+import org.jetbrains.spek.api.Spek
+import org.jetbrains.spek.api.dsl.TestContainer
+import org.jetbrains.spek.api.dsl.describe
+import org.jetbrains.spek.api.dsl.given
+import org.jetbrains.spek.api.dsl.it
+import org.jetbrains.spek.api.dsl.on
+import org.onap.dcae.collectors.veshv.domain.VesMessage
+import org.onap.dcaegen2.services.sdk.model.streams.ImmutableAafCredentials
+import org.onap.dcaegen2.services.sdk.model.streams.dmaap.ImmutableKafkaSink
+import org.onap.dcaegen2.services.sdk.model.streams.dmaap.KafkaSink
+import org.onap.ves.VesEventOuterClass
+import reactor.kafka.sender.SenderOptions
+import java.io.IOException
+import java.io.StreamTokenizer
+import java.io.StringReader
+import java.util.*
+import javax.security.auth.login.AppConfigurationEntry
+import javax.security.auth.login.Configuration
+
+/**
+ * @author [Piotr Jaszczyk](mailto:piotr.jaszczyk@nokia.com)
+ * @since April 2019
+ */
+internal class KafkaSenderOptionsFactoryTest : Spek({
+ describe("creation of Kafka Sender options") {
+
+ given("unauthenticated KafkaSink") {
+ val sink = ImmutableKafkaSink.builder()
+ .bootstrapServers("dmaap1,dmaap2")
+ .topicName("PERF_DATA")
+ .build()
+
+ on("calling the CUT method") {
+ val result = KafkaSenderOptionsFactory.createSenderOptions(sink)
+ val itShouldHavePropertySet = propertyChecker(result)
+
+ itShouldHavePropertySet(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, sink.bootstrapServers())
+ itShouldHavePropertySet(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 1_258_291)
+ itShouldHavePropertySet(ProducerConfig.BUFFER_MEMORY_CONFIG, 33_554_432)
+ itShouldHavePropertySet(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ProtobufSerializer::class.java)
+ itShouldHavePropertySet(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, VesMessageSerializer::class.java)
+ itShouldHavePropertySet(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1)
+ itShouldHavePropertySet(ProducerConfig.RETRIES_CONFIG, 1)
+ itShouldHavePropertySet(ProducerConfig.ACKS_CONFIG, "1")
+
+ itShouldHavePropertySet(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, null)
+ itShouldHavePropertySet(SaslConfigs.SASL_MECHANISM, null)
+ itShouldHavePropertySet(SaslConfigs.SASL_JAAS_CONFIG, null)
+ }
+
+ }
+ given("authenticated KafkaSink") {
+ val aafCredentials = ImmutableAafCredentials.builder()
+ .username("user \" with quote")
+ .password("password \" with quote")
+ .build()
+ val sink = ImmutableKafkaSink.builder()
+ .bootstrapServers("dmaap-service")
+ .topicName("OTHER_TOPIC")
+ .aafCredentials(aafCredentials)
+ .build()
+
+ on("calling the CUT method") {
+ val result = KafkaSenderOptionsFactory.createSenderOptions(sink)
+ val itShouldHavePropertySet = propertyChecker(result)
+
+ itShouldHavePropertySet(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, sink.bootstrapServers())
+ itShouldHavePropertySet(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 1_258_291)
+ itShouldHavePropertySet(ProducerConfig.BUFFER_MEMORY_CONFIG, 33_554_432)
+ itShouldHavePropertySet(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ProtobufSerializer::class.java)
+ itShouldHavePropertySet(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, VesMessageSerializer::class.java)
+ itShouldHavePropertySet(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1)
+ itShouldHavePropertySet(ProducerConfig.RETRIES_CONFIG, 1)
+ itShouldHavePropertySet(ProducerConfig.ACKS_CONFIG, "1")
+
+ itShouldHavePropertySet(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT")
+ itShouldHavePropertySet(SaslConfigs.SASL_MECHANISM, "PLAIN")
+ itShouldHavePropertySet(SaslConfigs.SASL_JAAS_CONFIG,
+ "org.apache.kafka.common.security.plain.PlainLoginModule required " +
+ """username="user \" with quote" password="password \" with quote";""")
+ }
+
+ }
+
+ }
+})
+
+private fun TestContainer.propertyChecker(actual: SenderOptions<VesEventOuterClass.CommonEventHeader, VesMessage>) =
+ { property: String, expectedValue: Any? ->
+ it("should have '$property' property set to '$expectedValue'") {
+ assertThat(actual.producerProperty(property)).isEqualTo(expectedValue)
+ }
+ }
diff --git a/sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/ProtobufSerializerTest.kt b/sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/ProtobufSerializerTest.kt
index 63caaf0a..c799a23c 100644
--- a/sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/ProtobufSerializerTest.kt
+++ b/sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/ProtobufSerializerTest.kt
@@ -29,7 +29,7 @@ import org.jetbrains.spek.api.dsl.on
import org.onap.ves.VesEventOuterClass.CommonEventHeader.*
-class ProtobufSerializerTest : Spek({
+internal class ProtobufSerializerTest : Spek({
describe("ProtobufSerializerTest") {
val serializer = ProtobufSerializer()
diff --git a/sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/VesMessageSerializerTest.kt b/sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/VesMessageSerializerTest.kt
index d11e5569..975ed827 100644
--- a/sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/VesMessageSerializerTest.kt
+++ b/sources/hv-collector-core/src/test/kotlin/org/onap/dcae/collectors/veshv/impl/adapters/kafka/VesMessageSerializerTest.kt
@@ -29,7 +29,7 @@ import org.onap.dcae.collectors.veshv.domain.VesMessage
import org.onap.ves.VesEventOuterClass.CommonEventHeader.*
-class VesMessageSerializerTest : Spek({
+internal class VesMessageSerializerTest : Spek({
describe("VesMessageSerializer") {
val serializer = VesMessageSerializer()
diff --git a/sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/kotlin.kt b/sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/kotlin.kt
new file mode 100644
index 00000000..bfa9babe
--- /dev/null
+++ b/sources/hv-collector-utils/src/main/kotlin/org/onap/dcae/collectors/veshv/utils/kotlin.kt
@@ -0,0 +1,27 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2019 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
+
+fun <T> T.applyIf(condition: Boolean, block: T.() -> T): T =
+ if (condition) {
+ block(this)
+ } else {
+ this
+ } \ No newline at end of file
diff --git a/sources/hv-collector-utils/src/test/kotlin/org/onap/dcae/collectors/veshv/utils/kotlin_test.kt b/sources/hv-collector-utils/src/test/kotlin/org/onap/dcae/collectors/veshv/utils/kotlin_test.kt
new file mode 100644
index 00000000..be92757c
--- /dev/null
+++ b/sources/hv-collector-utils/src/test/kotlin/org/onap/dcae/collectors/veshv/utils/kotlin_test.kt
@@ -0,0 +1,43 @@
+/*
+ * ============LICENSE_START=======================================================
+ * dcaegen2-collectors-veshv
+ * ================================================================================
+ * Copyright (C) 2019 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
+
+import org.assertj.core.api.Assertions.assertThat
+import org.jetbrains.spek.api.Spek
+import org.jetbrains.spek.api.dsl.describe
+import org.jetbrains.spek.api.dsl.it
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since April 2019
+ */
+internal class KotlinUtilsTest : Spek({
+ describe("applyIf") {
+ it("should apply the block if condition is true") {
+ val result = 69.applyIf(true) { this * 9 + 45 }
+ assertThat(result).isEqualTo(666)
+ }
+
+ it("should not apply the block if condition is false") {
+ val result = 69.applyIf(false) { this * 9 + 45 }
+ assertThat(result).isEqualTo(69)
+ }
+ }
+})