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 <sourabh.sourabh@est.tech>

3 files changed, 27 insertions, 28 deletions

diff --git a/k6-tests/once-off-test/kafka/produce-avc-event.js b/k6-tests/ncmp/common/produce-avc-event.js
index db222f6a4a..220265c6e7 100644
--- a/k6-tests/once-off-test/kafka/produce-avc-event.js
+++ b/k6-tests/ncmp/common/produce-avc-event.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.
@@ -18,12 +18,14 @@
  *  ============LICENSE_END=========================================================
  */
 
-import { crypto } from 'k6/experimental/webcrypto';
-import { check } from 'k6';
-import { Writer, SchemaRegistry, SCHEMA_TYPE_STRING } from 'k6/x/kafka';
+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',
@@ -33,26 +35,8 @@ const kafkaProducer = new Writer({
     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)];
+    return `neType-${Math.floor(Math.random() * 10) + 1}`;
 };
 
 function getCloudEventHeaders() {
@@ -69,6 +53,7 @@ function getCloudEventHeaders() {
 }
 
 export function sendKafkaMessages() {
+
     const cloudEventHeaders = getCloudEventHeaders();
     const networkElementId = getRandomNetworkElement();
 
@@ -85,8 +70,8 @@ export function sendKafkaMessages() {
     };
 
     try {
-        kafkaProducer.produce({ messages: [avcCloudEvent] });
-
+        kafkaProducer.produce({messages: [avcCloudEvent]});
+        messagesSent++;
         const isMessageSent = check(kafkaProducer, {
             'Message sent successfully': (producer) => producer != null,
         });
@@ -94,9 +79,8 @@ export function sendKafkaMessages() {
         if (!isMessageSent) {
             console.error('Failed to send message:', avcCloudEvent);
         }
-
     } catch (error) {
-        console.error('Error during message production:', error, avcCloudEvent);
+        console.error(`Error during message production: ${error.message}`, avcCloudEvent);
     }
 }
 
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 {