다중 소켓 관리하기(1): 트래픽 관리 서버

트래픽 관리 서버가 무엇일까?

• 트랙픽 관리 서버는 클라이언트가 소켓 서버에 연결되기 전에 어떤 소켓 서버로 가면 좋을지 알려주는 서버입니다.

왜 트래픽 관리 서버를 만들었을까?

• 여러 소켓 서버로 나누면 트래픽을 분산은 가능한데 같은 룸에 조인할 수 없는 문제가 발생했습니다.
• 따라서 같은 룸에 조인할 클라이언트들을 같은 소켓 서버로 보내고 이를 중계 해주는 트래픽 관리 서버를 만들었습니다.

트래픽 관리 서버가 각 서버의 상황을 어떻게 공유할까?

• 방향성은 크게 두가지였습니다.
• 트래픽 관리 서버에서 라운드 로빈 방식으로 소켓 URL을 제공해주는 방식입니다.
• 각 소켓 서버에서 자신의 소켓 상태를 주기적으로 업데이트하고 트래픽 관리 서버에서 최종적인 소켓 서버 URL 반환합니다.

라운드 로빈 방식이 별로인 이유..?

• 라운드 로빈 방식은 각 소켓 서버에 대한 정보를 순차적으로 제공하는 것입니다.
• 다만 각 소켓 서버 인스턴스 성능 차이가 있고 어느 정도 부하가 있을지 정확한 지표를 모르기 때문에 트래픽을 관리한다는 측면에 문제가 있습니다.

CPU로 각 소켓 서버의 상황을 측정한 이유!

• 소켓 서버고 메시지를 저장하거나 메모리를 사용하는 로직이 없습니다.
• 즉, 각 소켓 서버에 얼마나 많은 요청이 들어가고 나가는지가 중요하다고 생각했습니다.
• 따라서 CPU가 좀 더 정확한 지표라고 생각했습니다.

메인 스레드에서 CPU를 계산하는 비효율을 어떻게 해결할까?

• 각 소켓 서버는 5초에 한번 서버의 CPU 사용량을 측정하여 이 정보를 종합해 5분에 한번 트래픽 관리 서버로 보냅니다.
• 이 과정에서 CPU를 측정하는 일이 메인 스레드의 작업을 방해 할 수 있었습니다.
• 따라서 CPU 측정 일을 워커 스레드에게 맡긴 후 소켓 서버가 트래픽 관리 서버에 정보를 보낼때만 메인 스레드를 활용했습니다.

클라이언트가 소켓 서버와 연결되는 프로세스

😾 프로세스

• 클라이언트는 소켓 서버에 연결하기전 룸에 대한 정보와 함께 트래픽 관리 서버로 API 요청을 보냅니다.
• 만약 이미 생성된 룸에 대한 정보를 보내온다면 해당 룸과 소켓 서버 URL을 매칭 시켜 놓은 정보를 토대로 URL을 반환합니다.
• 만약 새로운 룸이라면 가장 부하가 적은 소켓 서버의 URL을 반환합니다.
• 이제 클라이언트는 적절한 소켓 URL을 받고 이를 통해 접속하게 됩니다.

그림으로 이해하자!

여러 소켓 서버로 연결될때 어떤 문제가 발생할까?

• 같은 채팅방의 클라이언트 1,2,3,4,5가 채팅을 주고 받기 위해 소켓 서버로 연결합니다.
• 하지만 소켓 서버에 분산되어 배치되었고 서로 채팅을 주고받을 수 없게 되었습니다.

트래픽 관리 서버를 통해 같은 채팅방 유저는 같은 서버로 연결하자!

• 이제 트래픽 관리 서버를 만들어 클라이언트는 룸과 매핑된 소켓 서버 URL을 반환받습니다.

트래픽 관리 서버 코드

시그널링 서버와 트래픽 관리 서버에 대해 순차적으로 정리해보겠습니다.

1. 소켓 서버가 등록될때 Register 이벤트를 PUB을 한다.

private publishSocketInfo() {
  const socketUrl = this.configService.get<string>('SOCKET_URL');

  const message = {
    url: socketUrl,
  };
  this.client.publish('register', JSON.stringify(message));
  this.scheduling();
}

2. Register 이벤트를 SUB한 트래픽 관리 서버가 서버를 등록한다.

private subscribe() {
  this.client.subscribe('register');

  this.client.on('message', async (channel, message) => {
    const data = JSON.parse(message);

    if (channel === 'register') {
      const { url } = data;
      this.handleRegister(url);
    }
  });
}

3. 소켓 서버가 5분에 한번 Signaling 이벤트를 PUB을 한다.

• Signaling 이벤트는 자신의 상태를 업데이트하는 PUB이다.
• Signaling 서버에 대한 이벤트이므로 이름을 Signaling으로 했다.

private scheduling() {
  cron.schedule('*/5 * * * *', () => {
    const connectionCount = this.webRtcGateway.getConnectionCnt();
    const message = {
      url: this.configService.get<string>('SOCKET_URL'),
      connections: connectionCount,
    };
    this.client.publish('signaling', JSON.stringify(message));
  });
}

4. Signaling 이벤트를 SUB한 후 적절한 소켓 서버를 반환한다.

• Signaling 이벤트를 받은 후, 현재 반환하기로 설정되어있던 소켓 서버 상태를 체크한다.
• 소켓 서버 상태 비교후, 직접 정한 기준을 통해 반환하기로 한 소켓 서버를 결정한다.
• 저는 소켓 서버에 연결된 커넥션 수로 결정했다.

private subscribe() {
  this.client.subscribe('signaling');

  this.client.on('message', async (channel, message) => {
    const data = JSON.parse(message);

    if (channel === 'signaling') {
      const { url, connections } = data;
      this.handleSignaling(url, connections);
    }
  });
}

5. 클라이언트의 요청이 들어왔을떄 적절한 URL을 반환합니다.

• 이후 클라이언트가 해당 URL을 통해 배정된 소켓으로 연결할 수 있다.

@Post('signaling/join')
async create(@Body() data: SignalingConnectionDto) {
  const response: ReturnConnectionsDto =
    this.eventService.findSignalingServer(data);
  return response;
}

findSignalingServer(data: SignalingConnectionDto): ReturnConnectionsDto {
  const { roomName } = data;

  const isServer = this.roomToUrl.get(roomName);

  if (isServer) {
    const result: ReturnConnectionsDto = { url: isServer };
    return result;
  }

  const server = this.serverToUrl.get('signaling');
  this.roomToUrl.set(roomName, server);
  const result: ReturnConnectionsDto = { url: server };
  return result;
}

6. 워커스레드를 활용하여 각 소켓 서버는 자기 서버의 부하를 계산하다.

import { parentPort } from 'worker_threads';
import * as os from 'os';

function calculateCpuUsage() {
  const cpus = os.cpus();
  let totalIdleTime = 0;
  let totalWorkTime = 0;

  for (const cpu of cpus) {
    for (const type in cpu.times) {
      totalWorkTime += cpu.times[type];
    }
    totalIdleTime += cpu.times.idle;
  }

  const totalUsage = 100 - (100 * totalIdleTime) / totalWorkTime;
  parentPort?.postMessage(totalUsage);
}

setInterval(calculateCpuUsage, 5000);