지난 포스팅에서 히트맵 조회 기능 소개, K6, InfluxDB, Grafana 를 연동하여 테스트 모니터링 환경을 구성하였습니다.
https://leehs-dev.tistory.com/3#0.%20%EB%A9%8D%ED%94%8C%20%EC%86%8C%EA%B0%9C-1
이번 포스팅에서는 단일 청크를 이용한 히트맵 조회 기능 구현의 테스트를 해보겠습니다.
1. 히트맵 조회 기능(단일 청크) 흐름 분석
- 사용자가 히트맵을 조회할 때, 사용자 중심 1km 반경의 단일 청크 조회 및 Aggregationing 후 반환
가장 먼저 해당 기능의 흐름을 살펴보면 다음과 같습니다.
- 사용자가 요청 JSON(조회반경의 반지름(side), 위치(lat, lon)) 을 동반하여 히트맵 조회를 요청
- HeatmapQueryListenerService 를 통해 Kafka 히트맵 조회 요청 메시지 발행
- 'heatmap' 토픽을 구독중인 HeatmapQueryExecutorService 가 메시지 처리
- HeatmapChunkPublisher 를 호출하여 ES 데이터를 조회(HeatmapSearcher 처리 위임)한 후
'users:{userId}:{queryType}' 을 key 로 하는 Redis List 에 청크 데이터 publish
- HeatmapChunkPublisher 를 호출하여 ES 데이터를 조회(HeatmapSearcher 처리 위임)한 후
- 일정 시간동안 Redis 의 List 를 polling 하고 있는 HeatmapChunkConsumer 가 전달받은 청크 데이터를 클라이언트에게 반환
@RequiredArgsConstructor
@Component
public class HeatmapChunkConsumer {
private static final int QUEUE_WAIT_TIME = 30;
private final ApplicationLogger logger;
private final RedisTemplate<String, HeatmapChunk> redisTemplate;
private final SimpMessagingTemplate messagingTemplate;
@Async("taskExecutor")
public void consume(String requestId, Long userId, HeatmapQueryType queryType) {
String key = generateQueueKey(userId, queryType);
String destination = generateDestination(queryType);
long endTime = System.currentTimeMillis() + QUEUE_WAIT_TIME * 1000;
while (System.currentTimeMillis() < endTime) {
HeatmapChunk chunk = redisTemplate.opsForList().leftPop(key, 500, TimeUnit.MILLISECONDS);
if (chunk != null) {
logger.log(LogLevel.DEBUG, LogAction.CONSUME, String.format("key: %s, chunk: %s", key, chunk), getClass());
messagingTemplate.convertAndSendToUser(userId.toString(), destination, chunk);
messagingTemplate.convertAndSendToUser(userId.toString(), destination, requestId);
break;
}
}
logger.log(LogLevel.DEBUG, LogAction.CONSUME, String.format("key: %s | 소켓 송신 종료", key), getClass());
}
private String generateQueueKey(Long userId, HeatmapQueryType queryType) {
return String.format("users:%s:%s", userId, queryType.getValue());
}
private String generateDestination(HeatmapQueryType queryType) {
return switch (queryType) {
case USER_BLUEZONE -> "/sub/users/bluezone";
case BLUEZONE -> "/sub/bluezone";
case REDZONE -> "/sub/redzone";
};
}
}
2. K6 스크립트 작성
해당 테스트를 통해 확인해야 하는 사항은, 요청-응답 간의 최소, 최대, 평균 반환 시간, 성공률이 중요하다고 생각하여 k6의 metric 을 다음과 같이 설정하였습니다.
let messageLatency = new Trend('message_latency', true); // 사용자의 응답 반환 시간
let messageCounter = new Counter('message_counter'); // 보낸 요청 개수
let receivedMessageCount = new Counter('received_message_count'); // 보낸 요청의 응답 개수
let requestResponseDifferenceRate = new Rate('request_response_difference_rate'); // 요청 - 응답 성공률
테스트는 가상 사용자 500명이 15초 주기로, 1km 반경의 히트맵을 9분동안 조회하도록 하였습니다.
처음 테스트 작성 시 클라이언트에서 requestId 를 정의하지 않고 테스트를 한 결과,
일정 주기로 요청을 보내고 난 후, 응답을 count 하고 지연율을 반영하는 과정에서 응답이 해당 요청에 관한 응답인지 구별하지 않아 테스트 결과가 정확하지 않게 나왔습니다.
따라서 클라이언트에서 요청을 할 때 요청 DTO 에 requestId 를 추가하여 각 요청을 고유하게 만들어 다음과 같이 스크립트를 작성하였습니다.
사용자가 요청을 보내는 경우
- messageCount 를 증가시키고 Map<String, Date> 를 사용하여 key : 요청 id, value : startTime 를 저장
requestId 에 해당하는 응답이 도착하는 경우
- receivedMessgeCount 를 증가
- requestResponseDifferenceRate 반영
전체 스크립트는 다음과 같습니다.
import http from 'k6/http';
import ws from 'k6/ws';
import { check } from 'k6';
import { Counter, Rate, Trend } from 'k6/metrics';
let requestTimestamps = new Map(); // 요청 id 별 요청 시간 저장
let messageLatency = new Trend('message_latency', true); // 사용자의 응답 반환 시간
let messageCounter = new Counter('message_counter'); // 보낸 요청 개수
let receivedMessageCount = new Counter('received_message_count'); // 보낸 요청의 응답 개수
let requestResponseDifferenceRate = new Rate('request_response_difference_rate'); // 요청 - 응답 성공률
const startTestTime = new Date();
const url = 'ws://host.docker.internal:8080/ws';
const loginUrl = 'http://host.docker.internal:8080/manager/login';
export const options = {
stages: [
{ duration: '20s', target: 500 }, // 100명으로 증가
{ duration: '520s', target: 500 }, // 60초 동안 500명 유지
{ duration: '60s', target: 500 }, // send 종료, 응답을 받는 시간
],
};
// export const options = {
// vus: 100,
// duration: '90s',
// };
function getJwtToken(userId) {
const res = http.post(`${loginUrl}?username=manager${userId}`);
const token = res.json().accessToken;
check(res, {
'login successful': (r) => r.status === 200,
'token exists': (r) => token !== undefined,
});
return token;
}
function sendMessage(socket, userId) {
const currentTime = new Date();
const elapsedTime = (currentTime - startTestTime) / 1000;
if (elapsedTime > 540) return; // 테스트 시간 9분 부터는 요청을 보내지 않음
const requestId = Math.random().toString(36).substr(2, 9);
const startTime = new Date();
const message = JSON.stringify({
requestId: requestId,
side: 500,
point: {
lat: 35.09493885488935,
lon: 128.853454676335,
},
});
const sendFrame = `SEND\ndestination:/pub/bluezone\ncontent-type:application/json\n\n${message}\0`;
socket.send(sendFrame);
requestTimestamps.set(requestId, startTime);
messageCounter.add(1);
requestResponseDifferenceRate.add(0);
}
function handleMessage(msg) {
if(msg.indexOf("cells") !== -1) return
const messageParts = msg.split('\n\n');
const requestId = messageParts[1].trim().slice(0, -1);
// // requestId 가 일치하면
if (requestTimestamps.has(requestId)) {
const startTime = new Date(requestTimestamps.get(requestId));
const endTime = new Date();
const latency = (endTime - startTime);
receivedMessageCount.add(1); // 응답 개수 추가
messageLatency.add(latency); // 요청-응답 시간을 추가
requestResponseDifferenceRate.add(1); // 요청-응답 성공률의 반영
requestTimestamps.delete(requestId);
}
}
export default function () {
const userId = __VU;
const token = getJwtToken(userId);
const response = ws.connect(url + `?Authorization=${token}`, {}, function (socket) {
socket.on('open', function () {
console.log(`User ${userId} connected to WebSocket server`);
const connectFrame = 'CONNECT\naccept-version:1.1,1.0\nhost:localhost\n\n\0';
socket.send(connectFrame);
const subscribeFrame = `SUBSCRIBE\nid:sub-${userId}\ndestination:/user/sub/bluezone\n\n\0`;
socket.send(subscribeFrame);
socket.setInterval(function () {
sendMessage(socket, userId);
}, 15000); // 15초 간격으로 요청
socket.on('message', function (msg) {
handleMessage(msg);
});
});
socket.on('close', function () {
console.log(`User ${userId} disconnected from WebSocket server`);
});
socket.on('error', function (e) {
console.error(`User ${userId} WebSocket error: `, e);
});
});
check(response, {
'WebSocket connection successful': (r) => r && r.status === 101,
});
}
3. Grafana Dashboard 수정
히트맵 조회 요청은 http 요청이 아닌 소켓을 통해 요청 - 응답을 하므로 불필요한 정보인 http 관련 metric 은 제외한 후, socket 관련 metric 을 추가해주었습니다.
응답 성공률
SELECT mean("value") * 2
FROM "request_response_difference_rate"
요청 개수
SELECT sum("value")
FROM "message_counter"
WHERE AND $timeFilter
응답 개수
SELECT sum("value")
FROM "received_message_count"
WHERE $timeFilter
응답 지연시간 그래프
1. 평균
SELECT mean("value")
2. 최대
SELECT max("value")
3. 최소
SELECT min("value")
FROM "message_latency"
WHERE $timeFilter and value > 0
GROUP BY time($__interval) fill(none)
응답 지연시간 표
[Dashboard Settings] - [Variables] - [Custom Option] message_latency 추가
4. 테스트 결과 확인
테스트는 총 3번 수행하였고, 다음과 같은 결과가 나왔습니다.
- 최소 응답 시간 평균 : 228 (ms)
- 평균 응답 시간 평균 : 5006 (ms)
- 최대 응답 시간 평균 : 10103 (ms)
마무리
다음 포스팅에서는 Divide and Conquer 방식을 적용하여 청크의 응답 시간을 테스트 해보겠습니다!
'Test' 카테고리의 다른 글
[멍플] 히트맵 조회 테스트 구현기 - 0 : 기능 소개, k6 테스트 환경 구성 (5) | 2024.10.13 |
---|