IoT 동적 스케일 프로젝트

import { check } from 'k6' import tcp from 'k6/x/tcp' import { Counter, Trend } from 'k6/metrics' import { produce, writer } from 'k6/x/kafka' import exec from 'k6/execution' const kafkaSendCounter = new Counter('kafka_send_counter') const relayDuration = new Trend('relay_duration', true) const randomNumber = getRandomArbitrary(0, 10000000) const iotIdString = `${ randomNumber }000000000000` const iotId = iotIdString.slice(0, 15) function getRandomArbitrary(min, max) { return Math.floor(Math.random() * (max - min) + min) } const topicList = [ 'request-a', 'request-b', 'request-c', ] const ipMap = { 'request-a': 'xxx.xxx.xxx.xxx', 'request-b': 'xxx.xxx.xxx.xxx', 'request-c': 'xxx.xxx.xxx.xxx', } const counterMap = { 'request-a': new Counter('request-a'), 'request-b': new Counter('request-b'), 'request-c': new Counter('request-c'), } const tcpMap = { 'request-a': createConnection('request-a'), 'request-b': createConnection('request-b'), 'request-c': createConnection('request-c'), } /* 여러 개의 토픽을 사용하는 경우 const writerMap = { 'request-a': createProducer('request-a'), 'request-b': createProducer('request-b'), 'request-c': createProducer('request-c'), } */ // 1개의 토픽을 사용하는 경우 const writerMap = { 'request-a': createProducer('request'), 'request-b': createProducer('request'), 'request-c': createProducer('trequest'), } function createConnection(kafkaTopic) { // kafkaTopic을 가지고 IP를 찾아서 TCP Send } function createProducer(kafkaTopic) { // kafka Topic에 Produce하는 Producer 생성 } function getTopicListFromIndex(index) { // index를 가지고 topic 중 하나를 가져옵니다 return topicList[index % topicList.length] } let conn export default function () { // k6의 경우 병렬적으로 실행되는 프로세스에서 고유한 id를 가져올 수 있습니다 // 이를 통해 프로세스마다 topic을 나눠가질 수 있습니다 const kafkaTopic = getTopicListFromIndex(Number(exec.vu.idInTest)) // topic을 가지고 TCP 커넥션을 가져옵니다 conn = tcpMap[kafkaTopic] // topic을 가지고 Kafka Producer를 가져옵니다 const producer = writerMap[kafkaTopic] counterMap[kafkaTopic].add(1) const message = { key: JSON.stringify(null), value: JSON.stringify({ // Kafka 메시지 }) } const startTime = Date.now() // kafka에 메시지를 보냅니다 let error = produce(producer, [message]) // Kafka에 메시지를 보냈으니 kafkaSendCounter를 1 올립니다 kafkaSendCounter.add(1, {iotId}) check(error, { 'Messages are sent': (err) => err == null, }) // TCP 연결로부터 메시지를 읽습니다 let read = tcp.read(conn) const endTime = Date.now() // Kafka에 메시지를 보낸 시점부터 TCP로 응답을 받은 시점까지의 시간을 측정합니다 relayDuration.add(endTime - startTime, {iotId}) // TCP 응답이 제대로 왔는지 확인합니다 let response = String.fromCharCode(...(read.slice(2))).slice(0, 15) check(response, { 'Received Data is correct': (r) => r == iotId }) } export function teardown() { if (conn) { tcp.close(conn) } }