From d2e5a379840ceac8ae798546b3abf40f678d9f3a Mon Sep 17 00:00:00 2001 From: sourabh_sourabh Date: Thu, 13 Feb 2025 10:26:19 +0000 Subject: Add CM Data Notification background Load to K6 - Implement Kafka message producer with controlled throughput - Configured message rate to 1000 (default: 500). - Implemented constant-arrival-rate scenario to maintain controlled throughput. - Adjusted maxVUs from 1000 to prevent resource overload. - Added graceful shutdown with a 10s stop period. - Added it into main K6 suite Issue-ID:CPS-2587 Change-Id: I75fd1fa83892fd4775b27132059f73e9c0ea4d75 Signed-off-by: sourabh_sourabh --- k6-tests/ncmp/common/produce-avc-event.js | 89 ++++++++++++++++++ k6-tests/ncmp/config/kpi.json | 10 +++ k6-tests/ncmp/ncmp-test-runner.js | 7 +- k6-tests/once-off-test/kafka/produce-avc-event.js | 105 ---------------------- 4 files changed, 105 insertions(+), 106 deletions(-) create mode 100644 k6-tests/ncmp/common/produce-avc-event.js delete mode 100644 k6-tests/once-off-test/kafka/produce-avc-event.js (limited to 'k6-tests') diff --git a/k6-tests/ncmp/common/produce-avc-event.js b/k6-tests/ncmp/common/produce-avc-event.js new file mode 100644 index 0000000000..220265c6e7 --- /dev/null +++ b/k6-tests/ncmp/common/produce-avc-event.js @@ -0,0 +1,89 @@ +/* + * ============LICENSE_START======================================================= + * Copyright (C) 2024-2025 Nordix Foundation + * ================================================================================ + * 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. + * + * SPDX-License-Identifier: Apache-2.0 + * ============LICENSE_END========================================================= + */ + +import {crypto} from 'k6/experimental/webcrypto'; +import {check} from 'k6'; +import {Writer, SchemaRegistry, SCHEMA_TYPE_STRING} from 'k6/x/kafka'; + +const testEventPayload = JSON.stringify(JSON.parse(open('../../resources/sampleAvcInputEvent.json'))); +const schemaRegistry = new SchemaRegistry(); +let messagesSent = 0; + +const kafkaProducer = new Writer({ + brokers: ['localhost:9092'], + topic: 'dmi-cm-events', + autoCreateTopic: true, + batchSize: 5000, + compression: 'gzip', + requestTimeout: 30000 +}); + +const getRandomNetworkElement = () => { + return `neType-${Math.floor(Math.random() * 10) + 1}`; +}; + +function getCloudEventHeaders() { + return { + ce_type: 'org.onap.cps.ncmp.events.avc1_0_0.AvcEvent', + ce_source: 'DMI', + ce_destination: 'dmi-cm-events', + ce_specversion: '1.0', + ce_time: new Date().toISOString(), + ce_id: crypto.randomUUID(), + ce_dataschema: 'urn:cps:org.onap.cps.ncmp.events.avc1_0_0.AvcEvent:1.0.0', + ce_correlationid: crypto.randomUUID() + }; +} + +export function sendKafkaMessages() { + + const cloudEventHeaders = getCloudEventHeaders(); + const networkElementId = getRandomNetworkElement(); + + const avcCloudEvent = { + key: schemaRegistry.serialize({ + data: networkElementId, + schemaType: SCHEMA_TYPE_STRING, + }), + value: schemaRegistry.serialize({ + data: testEventPayload, + schemaType: SCHEMA_TYPE_STRING + }), + headers: cloudEventHeaders + }; + + try { + kafkaProducer.produce({messages: [avcCloudEvent]}); + messagesSent++; + const isMessageSent = check(kafkaProducer, { + 'Message sent successfully': (producer) => producer != null, + }); + + if (!isMessageSent) { + console.error('Failed to send message:', avcCloudEvent); + } + } catch (error) { + console.error(`Error during message production: ${error.message}`, avcCloudEvent); + } +} + +export function teardown() { + kafkaProducer.close(); +} diff --git a/k6-tests/ncmp/config/kpi.json b/k6-tests/ncmp/config/kpi.json index eed041de85..b691ed5d7f 100644 --- a/k6-tests/ncmp/config/kpi.json +++ b/k6-tests/ncmp/config/kpi.json @@ -147,6 +147,16 @@ "exec": "legacyBatchConsumeScenario", "vus": 1, "iterations": 1 + }, + "produceKafkaMessages": { + "executor": "constant-arrival-rate", + "rate": 500, + "timeUnit": "1s", + "duration": "15m", + "preAllocatedVUs": 10, + "maxVUs": 10, + "exec": "produceAvcEventsScenario", + "gracefulStop": "10s" } }, "thresholds": { diff --git a/k6-tests/ncmp/ncmp-test-runner.js b/k6-tests/ncmp/ncmp-test-runner.js index 9ab326c44c..3c87057e8c 100644 --- a/k6-tests/ncmp/ncmp-test-runner.js +++ b/k6-tests/ncmp/ncmp-test-runner.js @@ -1,6 +1,6 @@ /* * ============LICENSE_START======================================================= - * Copyright (C) 2024 Nordix Foundation + * Copyright (C) 2024-2025 Nordix Foundation * ================================================================================ * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -30,6 +30,7 @@ import { import { createCmHandles, deleteCmHandles, waitForAllCmHandlesToBeReady } from './common/cmhandle-crud.js'; import { executeCmHandleSearch, executeCmHandleIdSearch } from './common/search-base.js'; import { passthroughRead, passthroughWrite, legacyBatchRead } from './common/passthrough-crud.js'; +import { sendKafkaMessages } from './common/produce-avc-event.js'; let cmHandlesCreatedPerSecondTrend = new Trend('cmhandles_created_per_second', false); let cmHandlesDeletedPerSecondTrend = new Trend('cmhandles_deleted_per_second', false); @@ -219,6 +220,10 @@ export function legacyBatchProduceScenario() { check(response, { 'data operation batch read status equals 200': (r) => r.status === 200 }); } +export function produceAvcEventsScenario() { + sendKafkaMessages(); +} + export function legacyBatchConsumeScenario() { const TOTAL_MESSAGES_TO_CONSUME = LEGACY_BATCH_THROUGHPUT_TEST_NUMBER_OF_REQUESTS * LEGACY_BATCH_THROUGHPUT_TEST_BATCH_SIZE; try { diff --git a/k6-tests/once-off-test/kafka/produce-avc-event.js b/k6-tests/once-off-test/kafka/produce-avc-event.js deleted file mode 100644 index db222f6a4a..0000000000 --- a/k6-tests/once-off-test/kafka/produce-avc-event.js +++ /dev/null @@ -1,105 +0,0 @@ -/* - * ============LICENSE_START======================================================= - * Copyright (C) 2024 Nordix Foundation - * ================================================================================ - * 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. - * - * SPDX-License-Identifier: Apache-2.0 - * ============LICENSE_END========================================================= - */ - -import { crypto } from 'k6/experimental/webcrypto'; -import { check } from 'k6'; -import { Writer, SchemaRegistry, SCHEMA_TYPE_STRING } from 'k6/x/kafka'; - -const testEventPayload = JSON.stringify(JSON.parse(open('../../resources/sampleAvcInputEvent.json'))); -const schemaRegistry = new SchemaRegistry(); -const kafkaProducer = new Writer({ - brokers: ['localhost:9092'], - topic: 'dmi-cm-events', - autoCreateTopic: true, - batchSize: 5000, - compression: 'gzip', - requestTimeout: 30000 -}); - -const TOTAL_MESSAGES = 100000; -const VIRTUAL_USERS = 1000; - -export const options = { - setupTimeout: '1m', - teardownTimeout: '1m', - scenarios: { - produceKafkaMessages: { - executor: 'shared-iterations', - exec: 'sendKafkaMessages', - vus: VIRTUAL_USERS, - iterations: TOTAL_MESSAGES, - maxDuration: '10m', - } - } -}; - -const getRandomNetworkElement = () => { - const networkElementIds = Array.from({ length: 10 }, (_, i) => `neType-${i + 1}`); - return networkElementIds[Math.floor(Math.random() * networkElementIds.length)]; -}; - -function getCloudEventHeaders() { - return { - ce_type: 'org.onap.cps.ncmp.events.avc1_0_0.AvcEvent', - ce_source: 'DMI', - ce_destination: 'dmi-cm-events', - ce_specversion: '1.0', - ce_time: new Date().toISOString(), - ce_id: crypto.randomUUID(), - ce_dataschema: 'urn:cps:org.onap.cps.ncmp.events.avc1_0_0.AvcEvent:1.0.0', - ce_correlationid: crypto.randomUUID() - }; -} - -export function sendKafkaMessages() { - const cloudEventHeaders = getCloudEventHeaders(); - const networkElementId = getRandomNetworkElement(); - - const avcCloudEvent = { - key: schemaRegistry.serialize({ - data: networkElementId, - schemaType: SCHEMA_TYPE_STRING, - }), - value: schemaRegistry.serialize({ - data: testEventPayload, - schemaType: SCHEMA_TYPE_STRING - }), - headers: cloudEventHeaders - }; - - try { - kafkaProducer.produce({ messages: [avcCloudEvent] }); - - const isMessageSent = check(kafkaProducer, { - 'Message sent successfully': (producer) => producer != null, - }); - - if (!isMessageSent) { - console.error('Failed to send message:', avcCloudEvent); - } - - } catch (error) { - console.error('Error during message production:', error, avcCloudEvent); - } -} - -export function teardown() { - kafkaProducer.close(); -} -- cgit