nestjs 에서 class 를 인스턴스화 시켜서 언제 어디서 사용해도 동일한 값을 이용하기 위해 constructor() 안에 사용하고 싶은 객체를 주입해서 주로 사용한다.

여러 예시가 있지만, 이 글에서는 db 에 접근하는 layer 인 repository 에서 어떻게 주입해서 사용하는지 코드를 통해 확인해보자.

code

// user.repository.ts
/**
 * @description default db connection : main db
 */
export class UserFunctions extends ReadRepo implements IUsersReadRepo {
  constructor(
    @InjectEntityManager() private readonly EntityManager: EntityManager,
    @InjectEntityManager('bo') private readonly boEntityManager: EntityManager,
  ) {
    super(EntityManager);
  }

우리는 읽는 repository 와 쓰는 repository 를 구분해서 사용하는데, 각 repository 에서 상속받는 클래스가 다르다. read repository 에서는 ReadRepo 를 상속받는데, ReadRepo 는 다음과 같이 생겼다.

export abstract class ReadRepo extends Repo {
  constructor(entityManager: EntityManager) {
    super(entityManager);
  }

  protected async queryMany<T extends object>(
    query: string,
    parameters?: any[],
    classConstructor?: ClassConstructor<T>,
  ): Promise<T[]> {
    const queryResult = columnToCamel(await this.query(query, parameters));
    if (!classConstructor) {
      return queryResult;
    }

    return Promise.all(
      queryResult.map((r) => this.validateReturnType(r, classConstructor)),
    );
  }

UserFunctions 의 생성자에서

super(EntityManager);

를 통해서 ReadRepo 의 생성자로 entityManager 를 주입시켜 주고,

export class Repo {
  constructor(protected entityManager: EntityManager) {}

  protected query<T>(query: string, parameters?: any[]): Promise<T> {
    return this.entityManager.query(query, parameters);
  }
}

주입받는 entityManager 를 다시

super(EntityManager);

를 통해서 상속받는 Repo 로 주입시켜서 쿼리를 실행하는 구조이다.

problem

하나의 repository 에서 하나의 entityManager 만 사용한다면 위 코드는 정말 잘 짜여진 코드라고 생각된다. 모듈화도 잘 시켜놨고, 코드의 재사용성 및 유지보수에도 좋다고 생각이 들지만... 문제는 여러 entityManager 를 사용하는 경우에 발생했다.

// user.repository.ts
/**
 * @description default db connection : main db
 */
export class UserFunctions extends ReadRepo implements IUsersReadRepo {
  constructor(
    @InjectEntityManager() private readonly EntityManager: EntityManager,
    @InjectEntityManager('bo') private readonly boEntityManager: EntityManager,
  ) {
    super(EntityManager);
  }

  async func1(): Promise<any> {
    this.entityManager = this.EntityManager
    this.queryMany('select ~'); // 1
    this.queryMany('select ~'); // 2
    this.queryMany('select ~'); // 3
  }

  async func2(): Promise<any> {
    this.entityManager = this.boEntityManager
    return this.queryMany('select ~');
  }

상황설명

func1 함수에서는 일반 디비 entityManager 를 사용하고, func2 함수에서는 bo 디비 entityManager 를 사용하고 있었고, 다음과 같이 각 함수에서 생성자에 주입된 entityManager 를 다른 entityManager 로 재할당 해줘서 사용하고 있었다. 그리고 func1 은 실행되는데 5초가 걸린다고 하고 호출 순서는 func1 -> func2 라고 해보자.

그렇게 된다면 일반 디비 entityManager 를 주입해줘서 func1 을 실행하는 도중에 func2 가 실행된다면 주입되는 entityManager 는 bo entityManager 로 재할당이 되어 버려서 원하는 테이블 혹은 sp 를 찾을 수 없다라는 에러 가 발생한다.

func1 에서 사용하는 entityManager 가 function scope 안에서 선언되어서 사용된 객체가 아니기 때문에, func2 가 실행되는 도중에 재할당을 시켜버리면 func1 에서도 bo entityManager 를 사용하게 되는 것이었다.

쿼리 결과에 대한 validator 체크를 해주는 queryMany() 함수를 사용하고자 하면, 주입된 entityManager 를 바꾸지 말고 하나만 사용을 해야 한다.

해결방법

하나의, 수정되지 않는 entityManager 를 사용하기 위해서는 기존에 사용하고 있던 queryMany() 를 수정하는 것 보다는 새로 만드는 것이 더 효율적이라고 판단했다.

애초에 기존에 사용하고 있는 queryMany() 를 수정하면 해당 함수를 사용하고 있는 모든 부분을 찾아서 수정을 해양하고, 하나의 repository 에서 두개 이상의 entityManager 를 주입받아서 사용하는 경우도 흔치 않기 때문이다.

그래서 새로 만든 함수는 다음과 같다.

protected async queryManyForPickDB<T extends object>(
    entityManager: EntityManager, // 인스턴스의 생성자에 종속된 DB 말고 다른 DB 를 사용하고자 하는 함수가 있다면 해당 entityManager 를 주입시킴. ex) getDashboard
    query: string,
    parameters?: any[],
    classConstructor?: ClassConstructor<T>,
  ): Promise<T[]> {
    const queryResult = columnToCamel(
      await entityManager.query(query, parameters),
    );
    if (!classConstructor) {
      return queryResult;
    }

    return Promise.all(
      queryResult.map((r: object) =>
        this.validateReturnType(r, classConstructor),
      ),
    );
  }

특정 원하는 디비에 접근하기 위해서 인자값으로 entityManager 를 받았고, 이것으로 쿼리를 실행하는 함수이다. queryManyForPickDB 호출부를 보면

// user.repository.ts
/**
 * @description default db connection : main db
 */
export class UserFunctions extends ReadRepo implements IUsersReadRepo {
  constructor(
    @InjectEntityManager() private readonly EntityManager: EntityManager,
    @InjectEntityManager('bo') private readonly boEntityManager: EntityManager,
  ) {
    super(EntityManager);
  }

  async func1(): Promise<any> {
    return this.queryManyForPickDB(
      this.boEntityManager,
      'select ~',
    );
  }

다음과 같이 super(EntityManager); 를 통해 기존 default 로 연결된 디비 말고 다른 디비 entityManager 를 인자값으로 넘겨줘서 쿼리를 실행시킬 수 있다.

생성자에 주입된 객체를 바꾸려고 하지 말자!

how to install

1. wget https://artifacts.elastic.co/downloads/kibana/kibana-8.6.2-x86_64.rpm
2. rpm -Uvh kibana-8.6.2-x86_64.rpm
3. sudo systemctl daemon-reload sudo systemctl enable logstash.service sudo systemctl start logstash.service sudo systemctl status logstash.service

config file

/etc/kibana/kibana.yml

## monitoring 
monitoring.ui.container.elasticsearch.enabled: true
monitoring.ui.container.logstash.enabled: true

# Specifies whether Kibana should rewrite requests that are prefixed with
# `server.basePath` or require that they are rewritten by your reverse proxy.
# Defaults to `false`.
#server.rewriteBasePath: false

# Specifies the public URL at which Kibana is available for end users. If
# `server.basePath` is configured this URL should end with the same basePath.
#server.publicBaseUrl: ""

# The maximum payload size in bytes for incoming server requests.
#server.maxPayload: 1048576

# The Kibana server's name. This is used for display purposes.
server.name: "kibana"

# =================== System: Kibana Server (Optional) ===================
# Enables SSL and paths to the PEM-format SSL certificate and SSL key files, respectively.
# These settings enable SSL for outgoing requests from the Kibana server to the browser.
#server.ssl.enabled: false
#server.ssl.certificate: /path/to/your/server.crt
#server.ssl.key: /path/to/your/server.key

# =================== System: Elasticsearch ===================
# The URLs of the Elasticsearch instances to use for all your queries.
elasticsearch.hosts: ["http://localhost:9200"]

kibana 와 elasticsearch 는 같은 ec2 에 설치되어 있기 때문에, ["http://localhost:9200"] 을 통해 elasticsearch 에 접근할 수 있다.

자 이제 localhost:5601 에 접속해서 확인해보면, 다음과 같이 필터도 걸고 검색을 통해서 원하는 로그를 확인할 수 있다.!

REF

https://www.elastic.co/kr/kibana 사진: Unsplash의Luke Chesser

elasticsearch 는 apahce lucene 기반의 java 로 만든 오픈소스 검색엔진이다. 엄청나게 빠른 검색 기능과 효율을 자랑하기 때문에, 흔히 사용하는 rdbms 에서 full-text-search 검색을 하는 것 보다 더 좋은 결과를 보인다. 그렇기 때문에, 일반적인 검색 기능은 웬만하면 elasticsearch 를 이용한다고 보면 된다.

elasticsearch vs opensearch

elasticsearch 와 용호상박을 이루는 opensearch 가 있다. opensearch 는 aws에서 만든 elasticsearch 와 아주 유사한 검색엔진인데, 한때 elasticsearch 와 분쟁이 있었다.

결국은 elasticsearch 가 이긴것으로 알고 있다.

아래는 간단하게 두 검색엔진을 비교한 표이다. 우리는 비용 및 여러가지 이유로 인해 elasticsearch 를 이용하기로 했다.

how to install

1. wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-8.6.2-x86_64.rpm
2. rpm -Uvh elasticsearch-8.6.2-x86_64.rpm
3. sudo systemctl daemon-reload sudo systemctl enable logstash.service sudo systemctl start logstash.service sudo systemctl status logstash.service

config file

elasticsearch 도 filebeat 와 마찬가지로 java 로 만들었기 때문에 heap 메모리 관리를 해줘야 한다. 기본적으로 아주 낮게 설정이 되어 있기 때문에, 이를 늘려주는 것이 좋다. Xms4g -> Xmx512m

/etc/elasticsearch/jvm.options

################################################################
## IMPORTANT: JVM heap size
################################################################
##
## The heap size is automatically configured by Elasticsearch
## based on the available memory in your system and the roles
## each node is configured to fulfill. If specifying heap is
## required, it should be done through a file in jvm.options.d,
## which should be named with .options suffix, and the min and
## max should be set to the same value. For example, to set the
## heap to 4 GB, create a new file in the jvm.options.d
## directory containing these lines:
##
## -Xms4g
## -Xmx4g
-Xms512m
-Xmx512m

/etc/elasticsearch/elasticsearch.yml

Xpack.security.transport.ssl:
  enabled: true
  verification_mode: certificate
  keystore.path: certs/transport.p12
  truststore.path: certs/transport.p12
# Create a new cluster with the current node only
# Additional nodes can still join the cluster later
cluster.initial_master_nodes: ["[ec2 ip where elasticsearch installed]"]

다음글은 마지막으로 kibana 에 대해서 알아보자.

REF

https://www.elastic.co/kr/elasticsearch 사진: Unsplash의Marten Newhall

그렇다면 실제로 어떻게 로그를 쌓고, 어떻게 연결이 되어서 긁어가는지 알아보자.

우리는 어디서든 에러가 발생하면 ExceptionFilter 를 통해서 에러를 캐치한다. 이를 실제로 구현한 코드는 아래와 같다.

// all-exception.filter.ts
import { LoggerService } from '@common/utility/logger.service';
import { ResponseObj } from '@core/class/response/response-obj';
import {
  ArgumentsHost,
  Catch,
  ExceptionFilter,
  HttpAdapterHost,
  HttpException,
  HttpStatus,
} from '@nestjs/common';

@Catch()
export class AllExceptionFilter implements ExceptionFilter {
  constructor(
    private readonly httpAdapterHost: HttpAdapterHost,
    private readonly logger: LoggerService,
  ) {}


  catch(exception: any, host: ArgumentsHost) {
    const { httpAdapter } = this.httpAdapterHost;
    const ctx = host.switchToHttp();

    const hostArgs = host.getArgs();
    const req = host.getArgByIndex(0);
    const incommingMessage = hostArgs[0];
    const serverIp = req.connection.localAddress.split(':')[3];
    // host port 
    const serverPort = req.connection.localPort;
    const hostUrl = `${req.headers.host}::${serverPort}`;

    const clientIp =
      incommingMessage.headers['x-forwarded-for']?.split(',')[0] ||
      req.connection.remoteAddress.split(':')[3];

    const { url, method } = incommingMessage;
    const appName = url.split('/')[2];
    let httpStatus = HttpStatus.INTERNAL_SERVER_ERROR;
    const errorCode = Number(exception.errorCode)
      ? exception.errorCode
      : Number(exception.message)
      ? exception.message
      : '0000';

    exception.url = url;
    exception.method = method;
    exception.appName = appName;
    exception.serverIp = serverIp;
    exception.clientIp = clientIp;
    exception.body = JSON.stringify(req.body);
    exception.hostUrl = hostUrl;

    httpStatus = exception.status
      ? exception.status
      : exception.statusCode
      ? exception.statusCode
      : httpStatus;

    switch(errorCode){
      case ...:
        this.logger.info(exception);
        this.logger.warn(exception);
        this.logger.error(exception):
    }

    let exceptionInfo: {
      statusCode?: number;
      error?: string;
      message?: string;
    } = {
      statusCode: exception?.status,
      message: exception?.message,
      error: exception?.message,
    };
   /**
    exceptionInfo 에 어떻게 값을 할당하는지는 생략하겠다.
   */

    const responseObj: ResponseObj<any> = new ResponseObj<any>(
      false,
      null,
      exceptionInfo,
      httpAdapter.getRequestUrl(ctx.getRequest()),
    );

    httpAdapter.reply(ctx.getResponse(), responseObj, httpStatus);
  }
}

// logger.service.ts
import { LoggerService, LogLevel } from '@nestjs/common';
import * as winston from 'winston';
import * as winstonDaily from 'winston-daily-rotate-file';

const logDir = `${process.cwd()}/logs`;
const { errors, combine, timestamp, printf, colorize, label } = winston.format;

export class LoggerService implements LoggerService {
  private logger: winston.Logger;

  constructor(service: string) {
    this.logger = winston.createLogger({
      exitOnError: true,
      format: combine(
        timestamp({ format: 'YYYY-MM-DD HH:mm:ss.SSSZ' }),
        errors({ stack: true }),
        label({ label: service }),
        printf((msg) => {
          return `#${msg.timestamp}#${msg.level}#${msg.message.appName}#${msg.message.method}#${msg.message.hostUrl}#${msg.message.url}#${msg.message.error}#${msg.message.body}#${msg.message.clientIp}#${msg.message.serverIp}#Exception ${msg.message.stackTrace}`;
        }),
      ),
      transports: [new winstonDaily(this.dailyOptions('info', service))],
      // handleExceptions: true,
    });

  }
  public log(message: any, ...optionalParams: any[]) {
    throw new Error('Method not implemented.');
  }
  public setLogLevels?(levels: LogLevel[]) {
    throw new Error('Method not implemented.');
  }

  console(): object {
    return this.logger;
  }

  public error(message: string, meta?) {
    this.logger.error(message, meta);
  }

  public warn(message: string) {
    this.logger.warn(message);
  }

  public info(message: string) {
    this.logger.info(message);
  }

  public debug(message: string) {
    this.logger.debug(message);
  }

  private dailyOptions(level: string, service: string) {
    return {
      level,
      datePattern: 'YYYY-MM-DD',
      dirname: logDir + `/${service}`,
      filename: `%DATE%.log`,
      maxFiles: '14d',
      maxSize: '256m',
      zippedArchive: false,
    };
  }
}

생략한 부분이 많긴 하지만, 중요하게 볼 부분은 아래 부분이다.

switch(errorCode){
  case ...:
    this.logger.info(exception);
    this.logger.warn(exception);
    this.logger.error(exception):
}

LoggerService 를 주입받아서 사용한 this.logger. 를 통해서 #${msg.timestamp}#${msg.level}#${msg.message.appName}#${msg.message.method}#${msg.message.hostUrl}#${msg.message.url}#${msg.message.error}#${msg.message.body}#${msg.message.clientIp}#${msg.message.serverIp}#Exception ${msg.message.stackTrace} 형태로 ${process.cwd()}/logs 파일에 저장할 수 있다.

이렇게 로컬 파일에 저장을 하게 되면 우리는 도커 이미지를 통해 배포를 하고 서비스를 운영하기 때문에 실제로는 ec2 instance 에 로그파일이 쌓이는 것이 아니고, 도커 이미지 안에 위치하게 된다. 그렇기 때문에, 도커 volume 을 통해서 파일을 공유해야 하는데, 아래와 같이 공유할 수 있다.

version: '3.7'
services:
  gateway:
    image: gateway
    container_name: gateway
    restart: always
    ports:
      - 3000:3000
      - 3051:3051
    working_dir: /app
    volumes:
      - /var/elk-log:/app/logs
    entrypoint:
      - node
      - ./dist/apps/gateway/main.js

도커 이미지 안에 위치한 /app/logs 이 부분을 실제 ec2 instance 의 /var/elk-log 로 이동시켜줄 수 있다. 이를 volumne mounting 이라고 한다.

이렇게 공유된 로그 파일은 /var/elk-log 에서 확인할 수 있다.

이렇게 공유가 되면 filebeat 에서는 input -> paths 를 통해서 파일을 긁어올 수 있게 된다.

filebeat.inputs:

# filestream is an input for collecting log messages from files.
- type: log

  # Unique ID among all inputs, an ID is required.
  #id: my-filestream-id

  # Change to true to enable this input configuration.
  enabled: true

  # Paths that should be crawled and fetched. Glob based paths.
  paths:
    - /var/elk-log/*/*.log

다음에는 이렇게 긁어온 파일을 받은 logstash 에서 어떻게 처리하는지 확인해보자.

REF

사진: Unsplash의Anis Rahman

모든 기능이 정상적으로 돌아가고 있는 가운데,

다음과 같은 에러가 발생했다. 의역하면, 요청이 너무 많아서 가용가능한 worker 가 정상적으로 일을 할 수 없다. 라는 뜻 인 것 같았다. 요청은 기존 그대로 보낼것 같은데, 혹시 디스크가 부족한건가 싶어서 디스크를 확인해봤다.

디스크 용량 확인하기

$ df -h

/dev/xvda1 용량을 보면 거의 다 사용한것을 알 수 있다.

가장 많이 사용된 용량 확인하기

$ cd / $ sudo du -ckx | sort -n > /tmp/duck-root $ sudo tail /tmp/duck-root

/var/log 와 /var/lib 에서 엄청나게 사용하는 것을 확인할 수 있다. /var/log/elasticsearch 와 /var/lib/promethus 에 가서 필요없는 파일들을 삭제하여 용량을 늘려줄 수 있다.

특정 디렉토리에서 많이 쓰이는 용량 확인하기

$ du -ah | sort -n -r | head -n 10 용량을 꽤나 차지 하고 있는 /var/log/elasticsearch 에 들어가서 어떤 파일이 실제로 용량을 차지하고 있는지 확인하려면, 위 명령어를 통해서 확인할 수 있다.

logstash

무료 개방형 서버의 데이터 처리 파이프라인인 Logstash는 다양한 소스에서 데이터를 수집하여 변환한 후 자주 사용하는 저장소로 전달합니다. logstash 공식문서

데이터를 수집해서 원하는 stash 로 전송한다. 데이터를 가공할 필요가 없다면, logstash 를 사용하지 않고, 바로 elasticsearch 혹은 다른 stash 로 전송하면 되지만, 비즈니스 상황 및 로직에 따라 데이터를 가공해야 할 경우 logstash 를 사용하면 효용성을 높일 수 있다.

how to install

1. wget https://artifacts.elastic.co/downloads/logstash/logstash-8.6.2-x86_64.rpm
2. rpm -Uvh logstash-8.6.2-x86_64.rpm
3. sudo systemctl daemon-reload sudo systemctl enable logstash.service sudo systemctl start logstash.service sudo systemctl status logstash.service

config file

logstash 는 filebeat 가 설치된 ec2(서버가 돌아가고 있는) 에 같이 설치해도 되지만, 관리의 편의성과 logstash 때문에 서버에 문제가 발생하면 안되기 때문에 다른 ec2 에 설치를 해주었다.

logstash 도 filebeat 와 마찬가지로 기본적으로 /etc/logstash 에 설정파일이 있다.

jvm.options

## JVM configuration

# Xms represents the initial size of total heap space
# Xmx represents the maximum size of total heap space

#-Xms1g
#-Xmx1g
-Xms256m
-Xmx256m

기본적으로 logstash 는 자바로 만들어져있어 따로 메모리 관리를 해줘야 한다. *처음에는 Xms1g 로 되어 있을텐데, 이건 용량이 너무 작아서 늘려주는 것(Xms256m)이 좋다. *

pipelines.yml

# This file is where you define your pipelines. You can define multiple.
# For more information on multiple pipelines, see the documentation:
#   https://www.elastic.co/guide/en/logstash/current/multiple-pipelines.html

- pipeline.id: main
  path.config: "/etc/logstash/pipeline.conf"

pipeline.yml 파일에는 어떤 설정파일을 가지고 logstash 를 구동할지에 대한 정의가 들어있다. 만약에 본인이 설정파일의 이름을 바꾸던지 위치를 바꿨다면 여기도 같이 바꿔줘야 한다.

pipeline.conf

# Beats -> Logstash -> Elasticsearch pipeline.


input {
  beats {
    port => 5044
  }
}

filter {
    grok {
        match => [ "message", "#(?<current_time>[^#]*)#(?<log_level>[^#]*)#(?<app_name>[^#]*)#(?<http_method>[^#]*)#(?<host_url>[^#]*)#(?<url>[^#]*)#(?<error_message>[^#]*)#(?<body>[^#]*)#(?<client_ip>[^#]*)#(?<server_ip>[^#]*)#(?<stack_trace>[^#]*)" ]
    }
    #current_time에서 년월일 추출
    grok {
        match => [ "current_time", "(?<py>[^-]*)-(?<pm>[^-]*)-(?<pd>[^-]*) (?<ph>[^:]*)" ]
    }

    mutate {
        convert => {
            # convert 구문에서 integer는 정수로 long, integer 같이 사용함
            "server_type" => "integer"
            "server_id" => "integer"
            "thread_id" => "integer"
        }
        rename => {
            "[fields][instance_name]" => "instance_name"
            "[py]" => "[@metadata][py]"
            "[pm]" => "[@metadata][pm]"
            "[pd]" => "[@metadata][pd]"
            "[ph]" => "[@metadata][ph]"
        }
        remove_field => ["_id", "_score", "_type", "beat", "source", "offset", "prospector", "message", "tags", "@version", "fields", "host", "input", "log", "ecs", "agent"]
    }
    if [app_name] =~ "undefined" {
        mutate {
          replace => { "log_level" => "info" }
        }
    }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "log-%{+YYYY.MM.dd}"
    user => "elastic"
    password => "[password]"
  }
  s3 {
    access_key_id => "[access_key_id]"
    secret_access_key => "[secret_access_key]"
    bucket => "[bucket_name]"
    codec => rubydebug
  }
}

여기가 가장 중요하다.

input

• beats : filebeat 로 부터 데이터를 받는다. elk 는 prometheus 와 다르게 데이터를 pull 해서 받는것이 아니라 push 해서 받는것이기 때문에 filebeat 가 설치되어 있는 ec2 의 ip를 따로 설정을 해줄 필요가 없다. 대신 port 번호(5044)는 있어야 한다.

filter

• grok : 쿼리를 짤 수 있도록 raw data 를 변형시켜주는 플러그인이다. match 를 통해 특정 field를 정형화된 형태로 매칭시켜줄 수 있다. 첫번째 match 를 보면 message field를 두번쨰 인자값인 #(?<current_time>[^#]*)#(?<log_level>[^#]*)#(?<app_name>[^#]*)#(?<http_method>[^#]*)#(?<host_url>[^#]*)#(?<url>[^#]*)#(?<error_message>[^#]*)#(?<body>[^#]*)#(?<client_ip>[^#]*)#(?<server_ip>[^#]*)#(?<stack_trace>[^#]*)" 형태로 매칭시켜줬다.
  잘 보면 (?<...>[^#]*) 와 같은 형태와 filebeat 설정파일에서 세팅한 multiline pattern 을 구분자로 나누는 것을 확인할 수 있다.

  (# 를 기준으로 (?<current_time>[^#]*) 과 같이 원하는 값을 구분지을 수 있다)

  current_time, log_level, app_name 이런 값들이 나중에 kibana 에서 쿼리를 짤때 쓰이는 field들이다.
  

• 대신 중요한 점이 있다. filebeat 에서 수집하는 로그 파일에도 저 형태와 동일하게 데이터가 쌓여야 한다. 즉, 반드시 # 를 구분자로 로그를 쌓아야 하다는 뜻이다.**
  두번째 match 를 보면 첫번째 match 에서 사용한 current_time field를 두번째 형태 `"(?[^-])-(?[^-])-(?[^-]) (?[^:])"` 로 매칭시켜줬다. *년월일시간**으로 형태화 시키기 위한 정규식을 통해서 kibana 에서 정형화된 날짜 형태 데이터로 확인할 수 있다.

• mutate : mutate 는 형태를 정형화 시켜주는 match 와는 다르게 값을 바꿔주는 역할을 한다 mutate 에서 쓰이는 기능들로는 coerce, rename, update, replace, convert, gsub, uppercase, capitalize, lowercase, strip, split, join, merge, copy 들이 있다. convert는 사용하는 field 타입을 변경해주는 type casting 을 할 수 있다. rename은 field 의 이름을 바꿔주는 역할을 해준다. filebeat 의 설정파일에서 세팅한 fields 키 값을 보면 instance_name 을 설정했었다.

  fields:
   instance_name: "instance_1"

  여기서 쓰이는 instance_1 을 instance_name field 로 사용하겠다라는 의미이다. 아래 있는 [@metadata][py] 는 kibana 에서 쓰이는 metadata 로 사용하겠다라는 의미이다. remove_field는 kibana 에서 특정 field 들을 보이지 않게 할 수 있다.
  마지막으로 에러 로그중에서 app_name 이 undefined 는 우리 서비스 페이지에서 보낸 요청이 아닌, 외부에서 의도적으로 공격성을 띈 요청이라고 판단되어, log_level 을 info 로 전환시킬 수도 있다.

  if [app_name] =~ "undefined" {
        mutate {
          replace => { "log_level" => "info" }
        }
    }

output

• elasticsearch : logstash 를 통해서 데이터를 가공하고 나면 해당 데이터들을 elasticsearch 로 전송하는 역할이다.

  logstash 와 elasticsearch 를 같은 ec2 에 설치를 하였기 때문에 localhost:9200 임을 알 수 있다.

• s3 : 가공된 데이터를 elasticsearch 뿐만 아니라, s3에도 저장을 하여, 추후에 분쟁이 있거나 혹시 모를 문제에 대응하기 위해 s3에 저장을 하게끔 설정한다.

다음에는 elasticsearch 에서 어떻게 kibana 로 보내는 지 알아보자.

REF

https://www.elastic.co/kr/logstash https://www.elastic.co/guide/en/logstash/current/plugins-filters-grok.html https://www.elastic.co/guide/en/logstash/current/plugins-filters-mutate.html 사진: Unsplash의Crystal Mapes

디비 설계를 할때, 무결성을 지켜야 함

무결성

데이터의 정확성, 일관성, 유효성을 유지하는 성질

무결성 종류

1. 개체 무결성 (PK의 성질입니다)

   • 기본키는 null을 가질 수 없다.
   • 기본키는 유일한 값이어야 한다.

2. 참조 무결성 (FK의 성질입니다)

   • 외래키는 null 이거나 참조테이블의 기본키여야 한다.

3. 도메인 무결성 (도메인: 칼럼 값이 가질 수 있는 데이터의 성질)

   • 데이터는 속해있는 필드의 속성에 부합해야 한다.

     

→ s3_file_name 은 uuid 성질이어야 한다. 

4. null 무결성

   • 특정 속성(필드)값이 null 이 될 수 없다

5. key 무결성

   • 한 테이블에는 적어도 하나 이상의 key 가 있어야 한다.

6. 고유 무결성

   • 특정 필드가 identity 하면 같은 값이 있을 수 없다

     

무결성을 지키지 않으면 이상현상이 발생할 수 있음

이상현상 종류

1. 삽입 이상

| 학번 | 학생 이름 | 학과 명 | 학과 코드 |
| --- | --- | --- | --- |
| 1 | 에디 | 수학과 | 001 |
| 2 | 제이크 | 경영학과 | 002 |
|  |  | 물리학과 | 003 |

물리학과를 추가하고자 하는데 물리학과에는 학생이 한명도 없다 → 물리학과 추가 불가능

→ 학번, 학생이름 / 학과 명, 학과 코드 분리 필요

2. 삭제 이상

| 학번 | 학생 이름 | 학과 명 | 학과 코드 |
| --- | --- | --- | --- |
| 1 | 에디 | 수학과 | 001 |
| 2 | 제이크 | 경영학과 | 002 |
| 3 | 올리 | 물리학과 | 003 |

올리가 자퇴를 해서 3번 row 를 지워야 한다 → 학과명과 학과 코드까지 한번에 지워지기 때문에 학과명, 학과 코드 정보가 사라진다

→ 학번, 학생이름 / 학과 명, 학과 코드 분리 필요

3. 수정 이상

| 학번 | 학생 이름 | 학과 명 | 학과 코드 |
| --- | --- | --- | --- |
| 1 | 에디 | 수학과 | 001 |
| 2 | 제이크 | 경영학과 | 002 |
| 3 | 올리 | 물리학과 | 003 |
| 4 | 파이 | 물리학과 | 003 |
| 5 | 랑 | 물리학과 | 003 |
| 6 | 조 | 물리학과 | 003 |

물리학과의 학과 코드가 004 로 바뀐다면 4개의 행을 수정해야 한다 → 여기서 실수로 3개만 수정한다면 데이터가 상이한 문제가 발생

→ 학번, 학생이름 / 학과 명, 학과 코드 분리 필요

다음과 같은 이상현상을 해결하려면 정규화가 필요하다 (정규화를 해야하는 이유 → 이상현상을 제거 하기 위해)

함수 종속성

하나의 값 A에 의해 다른 값 B가 하나의 값으로 결정되는 속성 (A가 B를 결정한다)

학번은 학생이름을 결정한다 학번 → 학생이름 (종속성), 학번 은 종속 결정자이다

정규화

1. 제 1 정규화

   • 각 칼럼이 하나의 속성만을 가져야 한다 → 학과명에서 불만족

   • 하나의 칼럼은 같은 타입의 값을 가져야 한다 → 학과명에서 불만족

   • 각 칼럼이 유일한 이름을 가져야 한다 → 만족

   • 칼럼의 순서가 상관없어야 한다 → 만족

     수정전

     학번	학생 이름	학과 명	학과 코드
     1	에디	수학과, 정치학과	001
     2	제이크	경영학과	002
     3	올리	물리학과	003

     수정후

     학번	학생 이름	학과 명	학과 코드
     1	에디	수학과	001
     2	제이크	경영학과	002
     3	올리	물리학과	003
     1	에디	정치학과	004

2. 제 2 정규화

   • 1 정규화를 만족해야 한다 → 만족

   • 모든 칼럼이 부분 종속성(PK가 아닌 다른 키에 종속됨)이 없어야 한다,

   • 완전 함수 종속(PK에만 종속될 수 있다)*을 만족해야 한다. 해당 테이블의 PK는 (학번+학과명) 이라고 했을 떄, 점수는 PK에 종속되지만, 학과 코드는 학과명에 종속된다.

     수정전

     학번	학과 명	학과 코드	점수
     1	수학과	001	100
     2	경영학과	002	90
     3	물리학과	003	80

     수정후

     학번	학과 코드	점수
     1	001	100
     2	002	90
     3	003	80

     학과 코드	학과 명
     001	수학과
     002	경영학과
     003	물리학과

3. 제 3 정규화

   • 2 정규화를 만족해야 한다
   • 기본키를 제외하고 모든 칼럼은 이행 종속성(X→Y, Y→Z ⇒ X→Z)이 없어야 한다.

수정전

| 학번 | 학생 이름 | 전화번호 |
| --- | --- | --- |
| 1 | 에디 | 01055558888 |
| 2 | 제이크 | 01044447777 |

학번을 알면 학생이름을 알 수 있고, 학생이름을 알면 전화번호를 알 수 있다.

학번을 알면 전화번호를 알 수 있게 됨

수정후

| 학번 | 학생 이름 |
| --- | --- |
| 1 | 에디 |
| 2 | 제이크 |

| 학생 이름 | 전화번호 |
| --- | --- |
| 에디 | 01055558888 |
| 제이크 | 01044447777 |

4. BCNF 정규화

   • 3 정규화를 만족해야 한다.

   • 후보키(PK가 될 수 있는 칼럼들)에 속하지 않은 칼럼이 종속 결정자가 될 수 없다.

     수정전

     payment_id	rfq_id	project_name
     1	4	test-1
     2	4	test-1
     3	6	test-3

     PK는 payment_id 이다

     rfq_id 는 payment_id 에 의해 결정 되지만, project_name 은 rfq_id 에 의해 결정된다.

     rfq_id 는 후보키가 아니기 때문에 만족하지 않는다.

     수정후

     payment_id	rfq_id
     1	4
     2	4
     3	6

     
     

     rfq_id	project_name
     4	test-1
     6	test-3

모니터링 툴에는 여러가지가 있다.

• sentry
• datadog
• grafana
• etc...

이번에는 grafana 에 대해서 얘기를 해보려고 한다.

회사에서 아직 모니터링 대시보드 역할을 할 수 있는 것이 없었기 때문에, 이를 구축하기 위해서 prometheus + grafana stack 으로 대시보드를 구성한 경험에 대해서 기록하자.

어떤 것이 필요할까?

일단 모니터링이 필요한 경우 어떻게 해야할지에 대해서 생각을 해보자.

모니터링을 한다는 것은, 모니터링 대상에 대한 정보를 실시간으로 긁어오거나, 모니터링 대상이 본인의 정보를 실시간으로 제공을 해주거나 하는 방법으로 지속적으로 데이터를 긁어와야 한다. 그렇다면, 그 긁어오거나 제공을 해주는 툴이 있어야 한다. 긁어오는 툴을 찾아보니, prometheus 라는 툴이 있었다. 또한, 그런 데이터들을 수집하는 collector 중에 우리가 사용하는 것은 node_exporter, postgres_exporter, cadvisor, nestjs-prometheus 가 있다.

✅ collector 1 - node exporter

node exporter 는 하드웨어의 상태 및 정보와 커널관련 메트릭을 수집하는 collector 이다. node exporter 를 통해 현재 하드웨어의 cpu 및 메모리 등 하드웨어의 정보를 수집할 수 있다.

how to install node exporter

1. download archive file $ wget https://github.com/prometheus/node_exporter/releases/download/v0.15.2/node_exporter-0.15.2.linux-amd64.tar.gz

2. extract archive file $ tar -xf node_exporter-0.15.2.linux-amd64.tar.gz

3. move binary file to /usr/local/bin $ sudo mv node_exporter-0.15.2.linux-amd64/node_exporter /usr/local/bin

4. create user only for node exporter $ sudo useradd -rs /bin/false node_exporter

5. make file that node exporter can be started at boot $ sudo vim /etc/systemd/system/node_exporter.service

   [Unit]
   Description=Node Exporter
   After=network.target
   
   [Service]
   User=node_exporter
   Group=node_exporter
   Type=simple
   ExecStart=/usr/local/bin/node_exporter
   
   [Install]
   WantedBy=multi-user.target

how to start node exporter

$ sudo systemctl daemon-reload $ sudo systemctl enable node_exporter $ sudo systemctl start node_exporter

✅ collector 2 - postgres exporter

how to install postgres exporter

1. make directory for postgres exporter $ mkdir /opt/postgres_exporter && cd /opt/postgres_exporter
2. download archive file % wget https://github.com/wrouesnel/postgres_exporter/releases/download/v0.5.1/postgres_exporter_v0.5.1_linux-amd64.tar.gz
3. extract archive file tar -xzvf postgres_exporter_v0.5.1_linux-amd64.tar.gz
4. copy to /usr/local/bin $ cd postgres_exporter_v0.5.1_linux-amd64 $ sudo cp postgres_exporter /usr/local/bin

env file for postgres exporter

$ cd /opt/postgres_exporter $ sudo vim postgres_exporter.env

특정 데이터베이스만 원하는 경우. database-name 치환

DATA_SOURCE_NAME="postgresql://username:password@localhost:5432/database-name?sslmode=disable"

모든 데이터베이스

DATA_SOURCE_NAME="postgresql://postgres:postgres@localhost:5432/?sslmode=disable"

setup postgres exporter user

1. create user $ sudo useradd -rs /bin/false postgres
2. make service file $ sudo vim /etc/systemd/system/postgres_exporter.service

   [Unit]
   Description=Prometheus exporter for Postgresql
   Wants=network-online.target
   After=network-online.target
   [Service]
   User=postgres
   Group=postgres
   WorkingDirectory=/opt/postgres_exporter
   EnvironmentFile=/opt/postgres_exporter/postgres_exporter.env
   ExecStart=/usr/local/bin/postgres_exporter --web.listen-address=:9187 --web.telemetry-path=/metrics
   Restart=always
   [Install]
   WantedBy=multi-user.target

   how to start prometheus exporter

   $ sudo systemctl daemon-reload $ sudo systemctl start postgres_exporter $ sudo systemctl enable postgres_exporter $ sudo systemctl status postgres_exporter

✅ collector 3 - cadvisor

cadvisor 는 docker, kubernetes 의 시계열 데이터를 수집하는 collector 이다. 각 container 들의 cpu, memory, network 사용량 등 리소스의 사용량 및 잔여량을 알 수 있다.

how to install cadvisor

$ sudo apt-get update $ sudo apt-get -y install cadvisor $ systemctl status cadvisor

✅ collector 4 - nestjs prometheus

node_exporter 에서는 실제 어플리케이션의 http request 및 response time 등을 핸들링하거나 추적하기에 적합하지 않은 데이터를 주고 있다보니, 실제 http request 및 response 를 다루기 위해 새로운 라이브러리가 필요했다.

이를 해결하기 위해서 node 라이브러리인, nestjs-prometheus 를 이용했다. https://github.com/willsoto/nestjs-prometheus 위 링크를 타고 들어가면 어떻게 설치하고 어떻게 사용하는지 나와있는데...

이게 생각보다 어려웠다. 기본적으로 제공하는 metric 들에 대한 개념도 있어야 하고, 이를 우리가 운영하고 있는 코드에 녹이는 것도 어느정도 시행착오가 있었다. 그냥 단순하게 연동하는 것 보다 최대한 현재 운영되고 있는 서비스에 영향을 가지 않도록 개발하는것이 포인트다.

구현방법

우리는 모든 요청에 대해서 기존 request flow 를 방해하지 않기 위해, interceptor 를 이용해서 데이터를 가져오도록 구현했다.

1. 우선 metric 모듈을 만들어주고, 모든 요청에 대해서 데이터를 수집하기 위해 gateway 모듈에 주입을 해준다.

   // metrics.module.ts
   import { Global, Module } from '@nestjs/common';
   import {
    makeCounterProvider,
    makeHistogramProvider,
    PrometheusModule as Prometheus,
   } from '@willsoto/nestjs-prometheus';
   
   @Module({
    imports: [
      Prometheus.register({
        path: '/metrics',
        defaultMetrics: {
          enabled: true,
        },
      }),
    ],
   })
   export class MetricsModule {}

   // gateway.module.ts
   @Module({
    imports: [
      MetricsModule,
    ],
    controllers: [GatewayController],
    providers: [GatewayService],
   })
   export class GatewayModule {
   ... 이하 생략
   }

2. 그 다음에 interceptor를 구현한다.

   import {
    CallHandler,
    ExecutionContext,
    Injectable,
    NestInterceptor,
    OnModuleInit,
   } from '@nestjs/common';
   import { Counter, Gauge, Histogram } from 'prom-client';
   import { Observable, tap } from 'rxjs';
   
   @Injectable()
   export class PrometheusInterceptor implements NestInterceptor, OnModuleInit {
    onModuleInit() {
      this.requestSuccessHistogram.reset();
      this.requestFailHistogram.reset();
      this.failureCounter.reset();
    }
    // status code 2XX
    private readonly requestSuccessHistogram = new Histogram({
      name: 'nestjs_success_requests',
      help: 'NestJs success requests - duration in seconds',
      labelNames: ['handler', 'controller', 'method'],
      buckets: [
        0.0001, 0.001, 0.005, 0.01, 0.025, 0.05, 0.075, 0.09, 0.1, 0.25, 0.5, 1,
        2.5, 5, 10,
      ],
    });
   
    // status code != 2XX
    private readonly requestFailHistogram = new Histogram({
      name: 'nestjs_fail_requests',
      help: 'NestJs fail requests - duration in seconds',
      labelNames: ['handler', 'controller', 'method'],
      buckets: [
        0.0001, 0.001, 0.005, 0.01, 0.025, 0.05, 0.075, 0.09, 0.1, 0.25, 0.5, 1,
        2.5, 5, 10,
      ],
    });
   
    private readonly failureCounter = new Counter({
      name: 'nestjs_requests_failed_count',
      help: 'NestJs requests that failed',
      labelNames: ['handler', 'controller', 'error', 'method'],
    });
   
    static registerServiceInfo(serviceInfo: {
      domain: string;
      name: string;
      version: string;
    }): PrometheusInterceptor {
      new Gauge({
        name: 'nestjs_info',
        help: 'NestJs service version info',
        labelNames: ['domain', 'name', 'version'],
      }).set(
        {
          domain: serviceInfo.domain,
          name: `${serviceInfo.domain}.${serviceInfo.name}`,
          version: serviceInfo.version,
        },
        1,
      );
   
      return new PrometheusInterceptor();
    }
   
      // metrics url 요청은 트래킹 필요 x
    private isAvailableMetricsUrl(url: string): boolean {
      const excludePaths = 'metrics';
      if (url.includes(excludePaths)) {
        return false;
      }
      return true;
    }
   
    intercept(context: ExecutionContext, next: CallHandler): Observable<any> {
      const originUrl = context.switchToHttp().getRequest().url.toString();
   
      const method = context.switchToHttp().getRequest().method.toString();
      const labels = {
        controller: context.getClass().name,
        handler: context.getHandler().name,
        method: method,
      };
   
      try {
        const requestSuccessTimer =
          this.requestSuccessHistogram.startTimer(labels);
        const requestFailTimer = this.requestFailHistogram.startTimer(labels);
        return next.handle().pipe(
          tap({
            next: () => {
              if (this.isAvailableMetricsUrl(originUrl)) {
                requestSuccessTimer();
              }
              // Handle the next event here
            },
            error: () => {
              if (this.isAvailableMetricsUrl(originUrl)) {
                requestFailTimer();
                this.failureCounter.labels({ ...labels }).inc(1);
              }
   
              // Handle the error event here
            },
          }),
        );
      } catch (error) {}
    }
   }

   위 코드에서 보듯이, 우선적으로 요청데이터를 저장할 histogram을 선언한다. Histogram class 에서 제공하는 startTimer() 함수가 모든 요청에 있어서 정보를 긁어올 수 있다.

• 요청의 response 가 성공하면 requestSuccessTimer() 를 통해 requestSuccessHistogram Histogram 에 정보를 저장하고,
• 요청의 response 가 실패하면 requestFailTimer() 를 통해 requestFailHistogram Histogram 에 정보를 저장한다. 또한, failureCounter.labels({ ...labels }).inc(1); 를 통해 실패 횟수를 1씩 증가시킬 수 있다.
• 해당 모듈이 실행될 때 마다 histogram 을 리셋해줌으로서(onModuleInit()), 메모리 용량을 관리해준다.

3. main.ts 에 interceptor 를 등록해준다.

   // main.ts
   async function bootstrap(): Promise<void> {
    const app = await NestFactory.create(GatewayModule);
    app.enableCors({
      origin: '*',
      methods: '*',
      allowedHeaders: '*',
    });
   
    app.useGlobalInterceptors(new ResponseInterceptor());
    app.useGlobalInterceptors(new PrometheusInterceptor());
   
    await app.listen(3000);
   
   }
   bootstrap();

4. response interceptor 에서 메트릭 정보를 가공하지 않고 그대로 반환해주기 위해 조건을 추가한다. 서비스내에서 클라이언트와 약속한 표준 응답으로 메트릭을 제공하면 데이터를 바인딩하는데 있어서 에러가 났다. 그래서 이를 약속한 response format 이 아닌 메트릭을 수집할 수 있도록 raw 형태 그대로 반환을 해줘야 한다.

혹시 response format 을 사용하고 있다면 메트릭 수집을 위해서는 꼭 약속된 format 을 이용하지 않고, 반환할 수 있도록!!!

  // response.interceptor.ts
@Injectable()
export class ResponseInterceptor implements NestInterceptor {
  intercept(_context: ExecutionContext, next: CallHandler) {
    const req: Request = _context.switchToHttp().getRequest();

    const excludePaths = ['/api/capa-metrics', '/capa-metrics'];

    return next
      .handle()
      .pipe(defaultIfEmpty(null))
      .pipe(
        map((result) => {
          if (excludePaths.includes(req.url)) {
            return result;
          }
          // CAPA4.0 표준 응답으로 변환하여 반환
          return new ResponseObj(true, result);
        }),
      );
  }
}

결과 확인

다음과 같이 어떤 요청이 어떻게 들어오고 있는지 확인할 수 있다.

자 이제 이 collector 들을 한군데 모으기 위해 prometheus 를 이용해보자.

✅ prometheus

prometheus 는 SoundCloud 에서 만든 오픈소스 모니터링 툴이다. prometheus 의 구조는 다음과 같다.

jobs/exporter 가 실제로 메트릭을 수집하고 이를 prometheus server 가 해당 메트릭 정보를 pull 을 통해서 가져온다. 해당 메트릭 정보는 각 모니터링 대상의 수집기로 부터 긁어오는데, 그 때 이제 http endpoint port 를 통해서 데이터를 가져온다.

모니터링 수집기의 예로는 node_exporter, cadvisor, postgres_exporter 이 있다.

또한, prometheus 에서 취급하는 메트릭 데이터는 전부 time series data 들로, 이러한 시계열 매트릭 데이터를 수집하고 저장하는 데이터베이스 역할도 한다.

그리고 오른쪽에 보면 Alertmanager 와 Grafana 가 있다.

Alertmanager 는 수집한 메트릭 정보가 어떠한 alert rule 에 위배가 되는 경우, email, teams, slack 으로 알림을 보낼 수 있는 기능이다.

Grafana 는 수집한 메트릭 정보를 시각화 해줄 수 있는 visual dashboard 역할을 한다.

우리는 prometheus 에서 수집한 메트릭 정보를 grafana 에서 시각화하는 방법으로 모니터링을 구축했다. 약간 elasticSearch & Kibana 같은 관계로 생각하면 된다.

prometheus install

1. download prometheus archive file $ wget https://github.com/prometheus/prometheus/releases/download/v2.1.0/prometheus-2.1.0.linux-amd64.tar.gz

2. extract from archieve file $ tar -xf prometheus-2.1.0.linux-amd64.tar.gz

3. move binary file to /usr/local/bin $ sudo mv prometheus-2.1.0.linux-amd64/prometheus prometheus-2.1.0.linux-amd64/promtool /usr/local/bin

4. make configuration file $ sudo mkdir /etc/prometheus /var/lib/prometheus $ sudo mv prometheus-2.1.0.linux-amd64/consoles prometheus-2.1.0.linux-amd64/console_libraries /etc/prometheus

5. delete useless file $ rm -r prometheus-2.1.0.linux-amd64*

prometheus configuration file

global:
  scrape_interval: 10s

scrape_configs:
  - job_name: 'prometheus_metrics'
    scrape_interval: 5s
    static_configs:
      - targets: ['localhost:9090']
  - job_name: 'node_exporter_B/E'
    scrape_interval: 5s
    static_configs:
      - targets: ['xx:9100', 'xx:9100', 'xx:9100']
  - job_name: 'node_exporter_F/E'
    scrape_interval: 5s
    static_configs:
      - targets: ['xx:9100', 'xx:9100']
  - job_name: 'postgres_exporter'
    static_configs:
      - targets: ['localhost:9187']
  - job_name: 'container_B/E'
    scrape_interval: 10s
    static_configs:
      - targets: ['xx:8080', 'xx:8080', 'xx:8080']
  - job_name: 'container_F/E'
    scrape_interval: 10s
    static_configs:
      - targets: ['xx:8080', 'xx:8080']
  - job_name: 'nestjs-app'
    scrape_interval: 10s
    metrics_path: /api/capa-metrics
    static_configs:
      - targets: ['xx:3000', 'xx:3000']

prometheus 설정파일을 보면, 각 collector 들의 할당된 port를 통해 긁어오게 세팅을 해줘야 한다.

* XX 로 되어 있는 host 는 실제 Frontend, Backend ec2 instance IP 이다

✅ grafana dashboard

grafana의 대시보드 datasource 를 prometheus 혹은 postgresql 와 연결하면 수집한 메트익 정보를 다음과 같이 시각화하여 대시보드를 구성할 수 있다.

collector 들로 수집한 메트릭 데이터를 prometheus 를 이용해서 grafana 에서 시각화 하려면 promql(prometheus query) 를 알아야 한다. 일반적으로 rdbms 혹은 nosql 에서 사용하는 언어와는 결이 다르기 때문에 따로 공부를 해야한다. promql 공식 문서 대시보드를 구성하는 방법은 여기서는 따로 언급하지는 않겠다. 대시보드 구성방법

Architecture

대시보드 파이프라인을 구축한 전체적인 아키텍처이다.

Ref

https://medium.com/devops-dudes/install-prometheus-on-ubuntu-18-04-a51602c6256b https://schh.medium.com/monitoring-postgresql-databases-using-postgres-exporter-along-with-prometheus-and-grafana-1d68209ca687 https://github.com/willsoto/nestjs-prometheus https://github.com/raphaabreu/nestjs-prometheus-requests/blob/main/src/prometheus.interceptor.ts

3.0 이상의 typeorm 을 가지고 트랜잭션 관리를 해보자. 레퍼런스가 너무 많아서 그냥 기록용으로 작성하려고 한다.

// room.repository.ts
import { Injectable } from '@nestjs/common';
import { RoomEntity } from 'lib/database/entity';
import { DataSource, Repository } from 'typeorm';

@Injectable()
export class RoomRepository extends Repository<RoomEntity> {
  constructor(private dataSource: DataSource) {
    super(
      RoomEntity, 
      dataSource.createEntityManager(),
      dataSource.createQueryRunner(),
    );
  }

  async createRoom(name: string): Promise<RoomEntity> {
    const queryRunner = this.dataSource.createQueryRunner();
    await queryRunner.connect();
    await queryRunner.startTransaction();
    try {
      const result = await queryRunner.manager.save(RoomEntity, { name: name });
      await queryRunner.commitTransaction();
      return result;
    } catch (error) {
      await queryRunner.rollbackTransaction();
    } finally {
      await queryRunner.release();
    }
  }
}

자세히 보면 생성자 부분을 잘 봐야 한다. 상속받은 Repository class 에

• 접근하고자 하는 엔티티 (RoomEntity)
• entityManager (dataSource.createEntityManager())
• queryRunner (dataSource.createQueryRunner())

를 주입시켜줘서 사용한다.

위와 같이 작성하는것이 일반적인데, repo 단에서 로직 구현할떄 마다 저렇게 할 수는 없으니, 코드를 줄여보자.

상속받은 Repository class 에는 EntityManager 에 접근할 수 있는 manager 라는 멤버변수가 있다.

export declare class Repository<Entity extends ObjectLiteral> {
    /**
     * Entity target that is managed by this repository.
     * If this repository manages entity from schema,
     * then it returns a name of that schema instead.
     */
    readonly target: EntityTarget<Entity>;
    /**
     * Entity Manager used by this repository.
     */
    readonly manager: EntityManager;
    /**
     * Query runner provider used for this repository.
     */
    readonly queryRunner?: QueryRunner;
    /**
     * Entity metadata of the entity current repository manages.
     */
    get metadata(): import("..").EntityMetadata;
    constructor(target: EntityTarget<Entity>, manager: EntityManager, queryRunner?: QueryRunner);

manager 를 통해서 EntityManager class 에 접근해서 보면 transaction 함수가 보인다.

export declare class EntityManager {
    /**
     * Wraps given function execution (and all operations made there) in a transaction.
     * All database operations must be executed using provided entity manager.
     */
    transaction<T>(runInTransaction: (entityManager: EntityManager) => Promise<T>): Promise<T>;
    /**
     * Wraps given function execution (and all operations made there) in a transaction.
     * All database operations must be executed using provided entity manager.
     */
    transaction<T>(isolationLevel: IsolationLevel, runInTransaction: (entityManager: EntityManager) => Promise<T>): Promise<T>;

설명을 잘 읽어보면, 주어진 함수를 하나의 transaction 으로 감싸서 실행할 수 있다. 그리고 반드시 주어진 entityManager 를 이용해야 한다. 라고 나와있다.

이거를 사용하면 되겠다 생각해서, 다음과 같이 transaction 안에 실행시킬 함수의 구현체를 넣어주면 구현할 수 있다. 대신, transaction 안에서는 에러가 발생하면 unhandled error 로 처리되기 때문에 반드시 안에서 try ~ catch 구문으로 에러를 핸들링해줘야 한다.

try {
      await this.manager.transaction(async (entityManager) => {
        try {
          await entityManager.save(RoomEntity, { name: name });
        } catch (error) {
          throw new Error('duplicated key');
        }
      });
    } catch (error) {
      throw error;
    }

그래서 다음과 같이 구현해냈다.

// room.repository.ts
@Injectable()
export class RoomRepository extends Repository<RoomEntity> {
  constructor(private dataSource: DataSource) {
    super(
      RoomEntity,
      dataSource.createEntityManager(),
      dataSource.createQueryRunner(),
    );
  }

  async getAllRoomIds(): Promise<RoomEntity[]> {
    return await this.find();
  }

  async getRoom(id: string): Promise<RoomEntity> {
    return await this.findOne({
      where: {
        id: id,
      },
    });
  }

  async createRoom(name: any): Promise<void> {
    try {
      await this.manager.transaction(async (entityManager) => {
        try {
          await entityManager.save(RoomEntity, { name: name });
        } catch (error) {
          throw new Error('duplicated key');
        }
      });
    } catch (error) {
      throw error;
    }
  }
}

REF

사진: Unsplash의Yeik CD

저번에는 실제 디비를 연결하지 않고, 고정 값으로 dataloader 를 구현해봤다면, 이번에는 실제로 디비에 연결해서 dataloader 가 어떻게 돌아가는지 확인해보자.

전체적인 플로우를 먼저 살펴보자 시나리오는 다음과 같다.

" 전체 채팅방에 대해서 각 채팅방에 속해있는 유저들의 정보를 긁어온다. " chat resolver → chat service → chat repository → user resolver

chat resolver

@Resolver(() => RoomModel)
// 이 resolver 는 chatModel 을 뽑아내기 위한 리졸버이다
export class ChatResolver {
  constructor(
    private readonly chatService: ChatService,
    private readonly chatLoader: ChatLoader,
  ) {}

  @Query(() => [UserModel], { name: 'user' })
  init() {}

  /**
   *
   * @description 전체 채팅방과 그 안에 포함된 유저들의 정보를 불러오자
   */

  @Query(() => [RoomModel])
  async getAllRoomInfo(): Promise<RoomModel[]> {
    return await this.chatService.getAllRoomIds();
  }

  /**
   * @description user resolver 로 필요한 정보를 얻기 위한 쿼리 전송
   * roomId 로 그 안에 속해있는 유저들의 정보를 긁어와야 하는데,
   * resolveField 호출은 여러번 해도 실제 디비에서 조회하는 로직은 한번만 할 수 있도록
   */
  @ResolveField('users', () => [UserModel])
  async getUsers(@Parent() room: RoomModel): Promise<UserModel[]> {
    try {
      const result = await this.chatLoader.findByUserId.load(room.roomId);
      return result;
    } catch (error) {
      this.chatLoader.findByUserId.clear(room.roomId);
    }
  }
}

chat service

@Injectable()
export class ChatService {
  constructor(private readonly roomRepo: RoomRepository) {}
  findAll() {
    return `This action returns all chat`;
  }

  async getAllRoomIds(): Promise<RoomModel[]> {
    const rooms = await this.roomRepo.getAllRoomIds();
    const result: RoomModel[] = [];
    for (const r of rooms) {
      result.push({
        roomId: r.id.toString(),
      });
    }
    return result;
  }
}

chat repository

@Injectable()
export class RoomRepository extends Repository<RoomEntity> {
  constructor(private dataSource: DataSource) {
    super(RoomEntity, dataSource.createEntityManager());
  }

  async getAllRoomIds(): Promise<RoomEntity[]> {
    return await this.createQueryBuilder().getMany();
  }
}

user resolver

@Resolver(() => UserModel)
export class UserResolver {
  constructor(private readonly userService: UserService) {}

  @Query(() => [UserModel], { name: 'user' })
  init() {}

  /**
   *
   * @description chat resolver 에서 처리하지 못하는 쿼리를 여기서 처리함 (find user)
   * resolverField 를 여기서 수신해서 처리한다.
   */
  @ResolveReference()
  async resolveReference(reference: {
    __typename: string;
    id: string;
  }): Promise<UserModel> {
    return await this.userService.getUser(reference.id);
  }
}

user service

@Injectable()
export class UserService {
  constructor(private readonly userRepo: UserRepository) {}
  async getUser(userId: string): Promise<UserModel> {
    const user = await this.userRepo.getUser(userId);
    const result: UserModel = {
      id: user.id,
      name: user.name,
    };
    return result;
  }
}

user repository

@Injectable()
export class UserRepository extends Repository<UserEntity> {
  constructor(private dataSource: DataSource) {
    super(UserEntity, dataSource.createEntityManager());
  }

  async getUser(userId: string): Promise<UserEntity> {
    return await this.createQueryBuilder().where({ id: userId }).getOne();
  }
}

db schema

play ground result

{
  "data": {
    "getAllRoomInfo": [
      {
        "roomId": "23bfddea-ef11-413f-940d-0101fc8ab23c",
        "users": [
          {
            "id": "ab5e6db1-2054-4fd2-9066-4ad1e407e903",
            "name": "tom"
          },
          {
            "id": "268311cb-66ce-486c-85c4-1bc6e0adbd6a",
            "name": "ciara"
          },
          {
            "id": "323d22e7-0edd-4d4f-ae3b-cb8b7ce42a97",
            "name": "json"
          },
          {
            "id": "ba176fc3-b860-4cae-9703-c9030212f5e7",
            "name": "daniel"
          }
        ]
      },
      {
        "roomId": "68e75332-c884-4f20-b05a-2795bf03945a",
        "users": [
          {
            "id": "9220a7f9-c958-48b7-80ff-15d1b462ae5a",
            "name": "jonny"
          },
          {
            "id": "7c7c2b9d-337c-4da7-b0b1-3f88b1adc486",
            "name": "hardy"
          }
        ]
      },
      {
        "roomId": "1ec747ab-b761-4889-89bb-6c6516f4874d",
        "users": []
      }
    ]
  }
}

repository ???

여기서 주의 깊게 봐야할 부분은 repository 이다 현재 나는 typeorm 버전이 3.0.0 이상이다 대부분의 사람들이 알겠지만, typeOrm 버전이 올라가면서 @EntityRepository() 가 없어졌다. 구글링을 해보면 다들 custom repository 를 만들어서, @InjectRepository() 를 이용해서 각 모듈마다 주입을 해주는 방법을 구현하고 있는데, 생각보다 너무 복잡해서 굳이 이렇게까지 어렵게 구현하라고 nest 가 @EntityRepository() 를 없애진 않았을 것이다.

찾아보다가 좋은 방법을 찾았다. https://stackoverflow.com/questions/72549668/how-to-do-custom-repository-using-typeorm-mongodb-in-nestjs

위에 나와있는 repository 코드가 다음 레퍼런스를 참고해서 만들었다. 생성자 부분을 잘 보면, 필요한 엔티티를 부모 클래스인 Repository 로 넘겨주면 provider 로 사용할 수 있게 된다.

여기서 transaction 은 어떻게 처리하는지는 다음에 알아보겠다.

REF

사진: Unsplash의Barn Images

예를 들어서, 결제가 일어났다 그러면 '결제완료' 혹은 대출을 신청했다 그러면 '대출신청' 이라는 이벤트를 웹훅으로 전송해준다.

우리 서비스에 녹여져있는 webhook flow 를 보기 전에 rabbitmq 에 대해서 간단하게 알아보자.

✅ What is RabbitMQ

비동기 작업을 하기 위해서 주로 사용하는 메세지큐의 종류 중 하나이다. ex) rabbitmq, kafka, etc...

            ------------ broker ------------
producer -> [exchange -> queue(binding rule)] -> comsumer

producer 가 메세지를 발송하면, 무조건 모든 메세지는 broker의 exchange 로 전달된다. exchange 는 어떤 queue 로 보내줄지 binding rule 에 의거해서 queue 로 메세지를 복사하고 이를 consumer 가 수신한다.

consumer 가 message 를 처리하는 방법

1. consumer 가 메세지를 받기만 하면 큐에서 메세지를 삭제한다.

   • msa 안에서 MQ 를 사용해서 [gateway -> microservivce] 같이 내부 api request 를 처리할 때 주로 사용할 수 있다. 내부 api에서 일어나기 때문에 언제든 다시 발송하게 할 수 있기 때문

2. consumer 가 메세지를 받고 잘 받았다는 신호를 보내면 큐에서 삭제한다. 이 신호를 consumer acknowledge 라고 한다.

   • 외부 모듈 혹은 외부 api 와 통신할 때 주로 사용한다. 정상적으로 정해진 로직으로 처리를 하지 못하거나 시스템 적으로 문제가 발생하면 다시 수신을 해서 처리를 해야 하는 경우가 생기기 때문.

exchange 종류

binding

각 exchange 에는 어떤 큐에 어떤 방식으로 메세지를 보낼지에 대한 규칙을 지정할 수 있다. 이 규칙은 binding 이라고 한다.

✅ webhook flow

해당 웹훅을 우리 서비스(api server)가 점검중이거나 혹은 예기치 못한 상황에 전송을 할 수도 있기 때문에, API Gateway 가 트리거로 걸린 람다 함수를 사용하고 있다. 아래는 해당 AWS lambda diagam 이다

1.결제 및 모듈에서 우리가 제공한 API Gateway 의 엔드포인트로 웹훅을 전송하면 2.그에 할당된 람다함수가 작동하여 RabbitMQ 로 메세지를 날린다. 3.그러면 서비스 API 서버가 해당 MQ에서 메세지를 가져간다. 4.메세지를 수신하고 로직을 처리하고 나서 Consumer Acknowledge 를 MQ 쪽으로 반환해준다.

✅ subscribe & publish message

@MessagePattern({ cmd: 'webHook' })
  async webHook(@Ctx() context: RmqContext): Promise<void> {
    const content = JSON.parse(context.getMessage().content);
    const data = content?.data;
    const jsonData = data?.body;

    const mqChannel = context.getChannelRef();
    try {
      await this.service.webHook(jsonData);
      mqChannel.ack(context.getMessage());
    } catch (error) {
      mqChannel.ack(context.getMessage());
    }
  }

mqChannel.ack() 를 통해서 "나 메세지 잘 수신했고, 잘 처리했어~" 라고 MQ 에 다시 알려준다. 그러면 해당 메세지를 MQ 는 다시 publish 하지 않는다. 이 말을 반대로 생각하면, *mqChannel.ack() 를 하지 않는다면, "잘 못 받았나? 무슨 문제가 있나?" 라고 간주해서 반복적으로 Publish 한다. *

✅ publish noAck message

테스트로나 서비스로나 한번도 발송하지 않은 메세지들이 계속 위 코드를 통해서 수신하고 있던 것을 확인할 수 있었다.

[Nest] 1  - 12/07/2023, 4:58:48 AM   ERROR [RpcExceptionsHandler] Unexpected token o in JSON at position 1
SyntaxError: Unexpected token o in JSON at position 1 at JSON.parse (<anonymous>)

원치 않은 형식의 message 가 날라와서 파싱하는데 있어서 계속 문제가 발생했다. 근데 이게 30분 마다 로그에 반복적으로 찍혔다. 혹시 noAck message 를 계속 전송하는건가? 생각해보니, 개발환경에 올리기 전에 로컬에서 테스트 했었는데, 이때 람다를 자체적으로 실행하기 위해서 input message 형태를 임의로 수정해서 테스트를 했었던 메세지가 아직도 수신되고 있었던 것이었다. 2023.11.29 일에 publish 된 메세지가 2023.12.07 에도 수신하고 있었다. aws console 창을 확인해보니 다음과 같았다.

• 첫번째 이미지를 보면, Redelivered 메세지가 반복적으로 날라오고 있었다. consumer가 ack 처리를 하지 않아 재전송한 것으로 보인다.
• 두번째 이미지를 보면 해당 메세지를 두번째 채널에서 Unacked 처리한 메세지가 하나 있다는 것을 알 수 있다.

리소스 낭비를 막기 위해 해당 메세지를 큐에서 지우기 위해 찾아보니 아래 화면에서 가장 하단부에 위치한 Purge messages 를 통해서 특정 큐에 존재하는 메세지를 지울 수 있다고 한다.

아래는 aws rabbitmq console 이다.

하지만 그래도 계속 위 에러가 발생했다. 에러가 발생하고 나서 콘솔창을 확인해보니 계속해서 Unack 처리된 메세지가 있고, Redelivered 를 하고 있었다.

✅ how to solve

unAck message 가 할당된 channel 을 먼저 close 하고 나서, 그 후에 purge message 를 하라고 나와있다. 개발환경에서 해당 channel에 연결중이고, 결제 docker container 만 잠시 내리고 purge 했다.

다른 방법으로 JSON.parse 가 실패해도 무조건 ack 처리를 하는것도 나쁘지 않을듯 하다. 애초에 정해진 메세지가 아닌 이상한 형태의 메세지가 왔다는 것 자체가 해당 메세지로 로직을 처리할 필요가 없으니까.

해결.

Ref

https://stackoverflow.com/questions/25114230/rabbitmq-purge-a-queue-from-all-of-its-unacked-messages https://jonnung.dev/rabbitmq/2019/02/06/about-amqp-implementtation-of-rabbitmq/#gsc.tab=0 https://www.rabbitmq.com/tutorials/amqp-concepts.html

graphql 에서 그냥 무작정으로 resolveField 를 사용하면 N+1 문제가 발생할 수 있는 경우가 있다. 여기서 N+1 문제란,

예를 들어, 채팅방에 여러 사람이 있는데 여러 사람의 정보를 긁어오기 위해 채팅방은 1개의 정보만 조회하지만, 그 후에 해당 채팅방의 사람 수(N) 만큼 정보를 조회해야 하는 경우를 뜻 한다.

그래서 @ResolveField 를 잘~ 써야 하는데,

만약에 한 쿼리에서 join을 통해서 데이터를 가져오면 되지 않을까? 싶지만, 그렇게 되면 client 의 요청에서는 over fetching 은 일어나지 않지만, back단에서 over fetching 이 일어난다. client에서는 원하지 않을 수 있는 데이터를 back 단에서 미리 조인을 걸어서 데이터를 가져오기 때문에 그렇다.

즉, join 을 사용하지 않고, @ResolveField 를 영리하게 사용을 하려면 dataloader 를 사용해야 하는데, dataloader 에 대해서 좀 자세하게 알아보자

dataloader

dataloader 는 데이터를 조회하는데 있어서 발생하는 N+1 문제를 batching, caching 을 통해서 1+1 로 해결해주는 모듈이다.

✅ batching

N번 실행하지 않기 위해서 여러 개의 key값을 배열로 받아서 해당 key 값의 value 를 매핑한 promise array object 을 생성한다. 그리고 각각의 value 들을 반환한다. 즉, promise array object 전체를 반환하는 것이 아닌, 각 value 들을 각각 반환한다. 아래는 dataloader 공식 git repository 에서 가져온 글이다.

A batch loading function accepts an Array of keys, and returns a Promise which resolves to an Array of values.

Then load individual values from the loader. DataLoader will coalesce all individual loads which occur within a single frame of execution (a single tick of the event loop) and then call your batch function with all requested keys.

여기서 지켜야 할 점들이 있다.

• value 들의 총 길이는 key 들의 총 길이와 같아야 한다.
• value 들의 인덱스 순서는 key 들의 인덱스 순서와 같아야 한다.

이 부분에 대해서는 아래 예시에서 확인해보겠다.

✅ caching

dataloader 는 key 를 바탕으로 캐싱처리를 해준다. .load() 를 통해서 캐싱처리를 해준다고 생각해준다.

주어진 key 값을 기반으로 캐싱으로 하기 때문에, 반복된 요청으로 부터 빠른 반환 시간을 보일 수 있지만, 다른 요청으로 부터 캐싱된 데이터의 원본이 수정이 되더라도, 반복된 요청으로 부터 동일한 값이 반환될 수 밖에 없다. 그렇기 때문에 요청이 올때 마다 dataloader instance 를 생성해주는 것을 권장한다.

const result = await this.chatLoader.findByUserId.load(room.roomId);
console.log(this.chatLoader.findByUserId);

DataLoader {
  _batchLoadFn: [AsyncFunction (anonymous)],
  _maxBatchSize: Infinity,
  _batchScheduleFn: [Function (anonymous)],
  _cacheKeyFn: [Function (anonymous)],
  _cacheMap: Map(3) {
    4 => Promise { [Array] },
    5 => Promise { [Array] },
    6 => Promise { [Array] }
  },
  _batch: {
    hasDispatched: true,
    keys: [ 4, 5, 6 ],
    callbacks: [ [Object], [Object], [Object] ]
  },
  name: null
}
DataLoader {
  _batchLoadFn: [AsyncFunction (anonymous)],
  _maxBatchSize: Infinity,
  _batchScheduleFn: [Function (anonymous)],
  _cacheKeyFn: [Function (anonymous)],
  _cacheMap: Map(3) {
    4 => Promise { [Array] },
    5 => Promise { [Array] },
    6 => Promise { [Array] }
  },
  _batch: {
    hasDispatched: true,
    keys: [ 4, 5, 6 ],
    callbacks: [ [Object], [Object], [Object] ]
  },
  name: null
}
DataLoader {
  _batchLoadFn: [AsyncFunction (anonymous)],
  _maxBatchSize: Infinity,
  _batchScheduleFn: [Function (anonymous)],
  _cacheKeyFn: [Function (anonymous)],
  _cacheMap: Map(3) {
    4 => Promise { [Array] },
    5 => Promise { [Array] },
    6 => Promise { [Array] }
  },
  _batch: {
    hasDispatched: true,
    keys: [ 4, 5, 6 ],
    callbacks: [ [Object], [Object], [Object] ]
  },
  name: null
}

위 결과를 보면, dataloder 는 자체적으로 cacheMap 을 가지고 있어 key 값을 바탕으로 캐시를 적용하는 것을 알 수 있다. room.roomId 가 4,5,6이 들어갈 수 있는데, 이를 키값으로 가지고 있고, 또한 value 값을 promise 배열 객체로 가지고 있는 것을 알 수 있다. 즉, 4,5,6 의 value 를 모두 resolve 하고 나서의 결과가 콘솔로 찍히는 것을 볼 수 있다.

dataloader 는 결국, 들어오는 모든 값들을 resolve 하고 난 결과를 뱉어내고, 이를 호출하는 부분에서 pending 상태로 대기하다가 resolve 되고 나서 결과를 받아볼 수 있다.

캐시 사용 안하기 - { cache: false}

@Injectable({ scope: Scope.REQUEST })
export class ChatLoader {
  findByUserId = new DataLoader<number, UserModel[]>(
    async (roomIds: number[]) => {
      // 여기 있는 roomIds 에 있는 아이템들이 key 가 된다.
      ... context
    { cache: false },
  );
}

캐시 전부 삭제하기 - clear & clearAll

@ResolveField('users', () => [UserModel])
  async getUsers(@Parent() room: RoomModel): Promise<UserModel[]> {
    try {
      await this.chatLoader.findByUserId.clear(room.roomId);
      await this.chatLoader.findByUserId.clearAll(room.roomId);
    } catch (error) {
      this.chatLoader.findByUserId.clear(room.roomId);
    }
  }

✅ 소스코드

// chat.resolver.ts
@Query(() => [RoomModel])
  async getRoomInfo(
    @Args('roomId', { type: () => Int }) roomId: number,
  ): Promise<RoomModel[]> {
    return [
      {
        roomId: roomId + 1,
      },
      {
        roomId: roomId + 2,
      },
      {
        roomId: roomId + 3,
      },
    ];
  }

@ResolveField('users', () => [UserModel])
  async getUsers(@Parent() room: RoomModel): Promise<UserModel[]> {
    try {
      const result = await this.chatLoader.findByUserId.load(room.roomId);
      console.log(result);
      return result;
    } catch (error) {
      this.chatLoader.findByUserId.clear(room.roomId);
    }
  }


// chat.loader.ts
@Injectable({ scope: Scope.REQUEST })
export class ChatLoader {
  findByUserId = new DataLoader<number, UserModel[]>(
    async (roomIds: number[]) => {
      // 여기 있는 roomIds 에 있는 아이템들이 key 가 된다.
      const userModels = {
        4: [{ id: '4' }, { id: '8' }, { id: '12' }],
        5: [{ id: '5' }, { id: '10' }, { id: '15' }],
        6: [{ id: '6' }, { id: '12' }, { id: '18' }],
      };
      const userModelGroup: { [key: number]: UserModel[] } = {};
      for (const roomId of roomIds) {
        if (!userModelGroup[roomId]) {
          userModelGroup[roomId] = [];
        }
        userModelGroup[roomId] = userModels[roomId] ?? [];
      }
      const result = roomIds.map((roomId: number) => userModelGroup[roomId]);

      console.log(result);
      return result;
    },
    { cache: true },
  );
}

// user.resolver.ts
@ResolveReference()
  resolveReference(reference: { __typename: string; id: string }): UserModel {
    return {
      id: reference.id,
      name: `${reference.id} + name`,
    };
  }

채팅방 4,5,6 에 속해있는 유저의 정보를 조회한다고 가정을 해보자. resolveField 를 통해서 채팅방 하나에 속해있는 유저들의 정보를 가져와야 하는데, dataloader 를 사용하지 않는다고 하면, resolveField에서 roomId:4 에 속한 유저들의 정보를 가져오고, roomId:5 에 속한 유저들의 정보를 가져오고, roomId:6 에 속한 유저들의 정보를 가져오고... N+1 문제가 발생할 수 있다.

그래서 @ResolveField 쿼리가 3번 호출되더라도, 실제 디비에서 데이터를 조회하는 로직은 한번만 할 수 있도록 dataloader 를 사용해보자.

await this.chatLoader.findByUserId.load(room.roomId)

여기서 실제 인자값으로는 roomId 가 한개씩 들어가지만, dataloader 의 batching 특징으로 인해

async (roomIds: number[])

으로 [4,5,6] 의 채팅방id 들의 배열이 들어간다.

그로인해, 채팅방에 속한 유저들의 정보가 다음과 같다고 가정을 해보자 채팅방 4에 속한 유저들은 4, 8, 12 채팅방 5에 속한 유저들은 5, 10, 15 채팅방 6에 속한 유저들은 6, 12, 18

const userModels = {
  4: [{ id: '4' }, { id: '8' }, { id: '12' }],
  5: [{ id: '5' }, { id: '10' }, { id: '15' }],
  6: [{ id: '6' }, { id: '12' }, { id: '18' }],
};

dataloader 를 통한 결과 값은 아래와 같다.

const result = roomIds.map((roomId: number) => userModelGroup[roomId]);
console.log(result);
return result;

[
  [ { id: '4' }, { id: '8' }, { id: '12' } ],
  [ { id: '5' }, { id: '10' }, { id: '15' } ],
  [ { id: '6' }, { id: '12' }, { id: '18' } ]
]

하지만 해당 dataloader 를 호출한 부분에서 console을 찍어보면 다음과 같다.

const result = await this.chatLoader.findByUserId.load(room.roomId);
console.log(result);

[ { id: '4' }, { id: '8' }, { id: '12' } ]
[ { id: '5' }, { id: '10' }, { id: '15' } ]
[ { id: '6' }, { id: '12' }, { id: '18' } ]

🔥🔥🔥 이걸 보면 결국 dataloader 에서 디비에서 조회는 한번만 해서 배열의 결과를 반환하지만, 호출부에서는 인자값으로 room.roomId 처럼 한개씩만 보냈기 때문에, 결과도 채팅방 하나의 속해있는 유저 정보를 조회하고 이를 @ResolveReference() 에서 수신해서 쿼리를 실행하는 것을 알 수 있다.

그래서 dataloader 의 키값이랑 반환값의 사이즈가 같아야 하고, 순서가 보장되어야 한다는 것을 여기서 알 수 있다 🔥🔥🔥

✅ playground 결과

{
  "data": {
    "getRoomInfo": [
      {
        "roomId": 4,
        "users": [
          {
            "id": "4",
            "name": "4 + name"
          },
          {
            "id": "8",
            "name": "8 + name"
          },
          {
            "id": "12",
            "name": "12 + name"
          }
        ]
      },
      {
        "roomId": 5,
        "users": [
          {
            "id": "5",
            "name": "5 + name"
          },
          {
            "id": "10",
            "name": "10 + name"
          },
          {
            "id": "15",
            "name": "15 + name"
          }
        ]
      },
      {
        "roomId": 6,
        "users": [
          {
            "id": "6",
            "name": "6 + name"
          },
          {
            "id": "12",
            "name": "12 + name"
          },
          {
            "id": "18",
            "name": "18 + name"
          }
        ]
      }
    ]
  }
}

REF

https://github.com/graphql/dataloader https://y0c.github.io/2019/11/24/graphql-query-optimize-with-dataloader/ 사진: Unsplash의Zac Edmonds

저번 글에서 @ResolveField() 를 통해서 추가 필드값을 받고, 해당 필드를 구현하기 위해서 @ResolveReference() 로 받아서 로직을 처리하는 방법을 구현했었다.

해당 내용을 공부해보니, "이렇게 하면 N+1 문제가 발생해서 DataLoader 를 사용하면 문제를 해결할 수 있다~" 라는 글을 많이 봤는데, N+1 문제로 인해 쿼리 지연이 발생하려면, resolveField 가 배열, 즉 여러 값이 나와야 그러한 문제가 발생할 수 있겠다~ 라고 생각해서, 저번에는 필드가 배열이 아닌 하나의 값으로 구현을 했는데 이번에는 배열로 한번 받아보자 라는 생각에 배열로 구현을 해보기로 했다.

그런데, 아무리 해도 계속 배열로 뭔가 받질 못하는 것 같아서 수 많은 구글링과 phind 및 gpt 를 찾아봐도 내가 원하는 결과를 도출해낼 수 없었다.

구현하고자 하는 내용을 말하자면, 하나의 채팅방에 여러명의 유저가 있을 수 있고, 그러한 유저들의 정보를 채팅방 정보와 같이 조회를 하기 위한 기능이었다.

그래서 근본적으로 생각을 해봤는데,

1. 일단 ! 필드 자체는 배열이긴 하다.(하나의 채팅방에 유저는 여러명이니까) 그러면 이 resolveReference 가 쿼리를 여러번 수신해야 하는건가? 👉 이건 아무리 생각해도 말이 안된다. 그래프큐엘을 만든 팀에서 과연 그렇게 비효율적으로 만들었을까
2. 그렇다면 일단 resolveField 는 배열을 반환하는 것은 맞다.
3. 그리고 이거를 resolveReference 도 결국에는 한번만 받고, 그 배열을 풀어내기 위한 로직을 여러번 하는건가
4. 이거다 !

결국에 구현해냈다. 물론 디비를 연결해서 실제로 데이터 커넥션을 일으킨건 아니지만, 전체적인 흐름을 파악할 수 있게 되었다.

/// chat.resolver.ts
@ResolveField('users', () => [AccountModel])
  async getUsers(
    @Parent() chat: RoomModel,
  ): Promise<{ __typename: string; id: number }[]> {
    console.log(chat.roomId);
    const userIds: number[] = [1, 2, 3];
    return [
      { __typename: 'RoomModel', id: userIds[0] },
      { __typename: 'RoomModel', id: userIds[1] },
      { __typename: 'RoomModel', id: userIds[2] },
    ];
  }

특정 채팅방의 roomId 를 이용해서 디비에서 조회를 한 후, 해당 채팅방에 존재 하는 유저들의 id 가 1,2,3 이라고 하자. 이 유저들의 정보를 chat.resolver 에서 조회하는 것이 아닌, account.resolver 에서 조회하기 때문에, account.resolver 쪽으로 array object 을 보내준다.

@ResolveReference()
  resolveReference(reference: {
    __typename: string;
    id: string;
  }): AccountModel {
    return {
      id: reference.id,
      name: `${reference.id} + name`,
    };
  }

그러면 여기서 받아서 처리를 해주는데, 배열안에 아이템이 3개가 있기 때문에, @ResolveReference 를 3번 처리해줘야 한다. 이렇게 하나를 처리하기 위해 여러번의 쿼리를 실행할 경우 N+1 문제가 발생할 수 있다.

REF

https://unsplash.com/ko/%EC%82%AC%EC%A7%84/%ED%9A%8C%EC%83%89-%ED%83%9C%EC%96%91-%EC%A0%84%EC%A7%80%ED%8C%90-%EB%A1%9C%ED%8A%B8-dCx2xFuPWks

procedure 버전관리를 어떻게 해야하지

개발환경마다 디비를 다르게 해서 관리를 해서 배포를 하는 방법으로 진행했지만, 소스코드는 github로 버전관리를 하면 되는데, procedure, function 은 버전관리를 깃허브와 완전 비슷하게 우리 입맛에 구현해주는 툴이 없었다. datagrip 에서 github repository 를 연동해서 하면 된다고 했는데, 예상보다 쉽지 않았고, 뭔가 맘에 들지 않았다.

그래서 팀원 모두들 찾아보다가 툴 말고 그냥 디비안에서 관리를 해야겠다는 생각을 했다. 그 결과 trigger를 사용하기로 했다. trigger 는 어떠한 액션에 대해서 자체적으로 이벤트를 캐치해서 원하는 동작을 하게 해주는 기술이다. postgresql에만 있는건 아니고 다른 rdbms, nosql 에도 다~ 있다. 일반적인 트리거는 특정 테이블에서 발생하는 DML 을 캐치해서 그에 따른 처리를 해주는 것이 일반적이다. 하지만 우리는 테이블의 상태 변화가 아닌, procedure, function 의 상태 변화를 캐치했어야 했다. 그래서 구현한 방법이 일반적인 트리거가 아닌 event trigger 이다.

Event trigger

특정 하나의 테이블에서 발생하는 DML 만을 캐치하는것이 아닌, event trigger 는 특정 한 테이블에 국한되는 것이 아닌, 전역적으로 모든 DDL 이벤트를 캐치하는 트리거이다.

postgres 16 버전에서는

• ddl_command_start
• ddl_command_end
• table_rewrite
• sql_drop

위 4가지 이벤트에 대해서 이벤트 트리거가 발동한다고 한다. 그 외의 기능들은 추후에 추가될 예정이라고 한다.

1. ddl_command_start CREATE, ALTER, DROP, SECURITY LABEL, COMMENT, GRANT, REVOKE, SELECT INTO(CREATE TABLE AS) 위 명령어들이 시작 하기 전에 발동한다.

2. ddl_command_end ddl_command_start 와 같은 이벤트들에 대해서 명령어 후에 발동한다. 해당 명령어들이 끝났을 때 어떤 명령어가 끝났고, 그에 대한 추가 정보가 필요하면 pg_event_trigger_ddl_commands() 함수를 이용하면 정보를 알 수 있다. 예를들어, create procedure 후에 pg_event_trigger_ddl_commands() 함수를 실행하면, 어떤 프로시저가 생성되고, 생성시점, 추가 정보 등을 알 수 있다. 대신에 해당 이벤트는 트랜잭션이 커밋되기전에 발동한다.

3. sql_drop sql_drop 은 ddl_command_end 바로 직전에 어떠한 객체가 drop 될 때 발동한다. 어떤 객체들이 drop 되었는지 알기 위해서는 pg_event_trigger_dropped_objects() 함수를 이용하면 알 수 있다. 이 이벤트는 시스템 카탈로그에서 객체가 사라지면 확인할 수 있다.

system catalog: 데이터베이스의 모든 객체에 대한 정의 와 정보를 저장하고 있는 시스템 테이블이다. 성능 평가를 위한 모든 통계정보도 저장하고, 카탈로그에 저장된 데이터를 meta data라고 한다.

4. table_rewrite 테이블의 상태가 변하는 ALTER TABLE, ALTER TYPE 명령어 바로 직전에 실행되는 이벤트이다.

🔥 주의할 점 🔥

*이벤트 트리거는 트랜잭션이 롤백되면 실행되지 않는다. *

그렇기 때문에,

1. ddl_command_start 실패하면 event trigger 또한 실행되지 않는다.
2. ddl 이 실패하면 ddl_command_end 또한 실행되지 않는다.

3. ddl_command_end 이 실패하면 ddl 또한 롤백된다.

ddl_command_end 이 commit 되기 전에 실행되기 때문에 3번 같은 문제가 발생하는데, 그렇기 때문에 ddl_command_end 이벤트에 따른 이벤트트리거는 신중하게 작성해야겠다.

그래서 어떻게 설계를 했는가

procedure, function 이 CREATE, ALTER, DROP 중 하나의 상태가 되면 해당 이벤트를 잡아서 액션을 취하도록 하였다.

1) 로그를 저장할 테이블을 만들자

create table tb_sp_fn_ddl_log
(
    name        text, // procedure, function 이름
    obj_id      integer, // trigger action id
    ddl         text, // procedure, function DDL
    create_date timestamp with time zone default now() not null
);

2) 이벤트 트리거가 걸리면 작동할 function을 만들어준다.

create function cr_sp_fn_ddl_log() returns event_trigger
    language plpgsql
as
$$

DECLARE
    r   RECORD;
    ddl text;
BEGIN
    IF tg_tag in
       ('CREATE PROCEDURE', 'CREATE FUNCTION')
    THEN
        r := pg_event_trigger_ddl_commands();
        select * into ddl from pg_get_functiondef(r.objid);
        insert into tb_sp_fn_ddl_log (name, obj_id, ddl) values (r.object_identity, r.objid, ddl);
    END IF;
END;
$$;

이벤트 트리거로 작동할 함수는 반드시 event_trigger 를 반환하는 형식의 함수여야 한다.

• tg_tag 는 postgresql event_trigger 를 반환하는 함수에서 자체적으로 수집할 수 있는 객체라고 생각된다.
• tg_tag 가 create procedure 혹은 create function 일 경우에 이 함수가 실행되는데, pg_event_trigger_ddl_commands() 를 통해서 ddl_command_end 이벤트의 정보를 조회할 수 있다.
• pg_get_functiondef() 함수는 procedure, function 의 정의 DDL 를 조회할 수 있다.
• 조회할 결과를 토대로 사전에 생성한 테이블에 저장한다.

3) 마지막으로 해당 function을 일으킬 트리거를 생성해준다.

create event trigger tr_sp_fn_ddl_log
    on ddl_command_end
execute procedure public.cr_sp_fn_ddl_log();

이렇게 하면 테이블에 그동안 특정 procedure 혹은 function 이 어떻게 변화했는가 모든 로그를 트래킹할 수 있다. 저장 시간까지 저장하기 때문에, 원하는 시점에 어떤 DDL이었는지 알 수 있기 때문에, 롤백 및 버전관리를 하기 수월해졌다.

Reference https://www.postgresql.org/docs/current/event-triggers.html https://www.postgresql.org/docs/current/event-trigger-definition.html https://code-lab1.tistory.com/133

🔥 MSA 구조를 graphql 로 만들어보자 🔥

그전에 생각해보자... grapqhl 은 엔드포인트가 하나인데, MSA 를 어떻게 구현해야 하지...? 뭔가 msa 로 할 수 있을 것 같으면서도... 이게 restapi 처럼 msa 를 구성 가능할까...? 여러가지 생각이 든다. 음...각 app 별로 tcp 서버를 열어야 하나? tcp 통신이 필요한가? 어차피 엔드포인트 하나인데..? 한번 해보자.

1. 각 도메인에 따른 app 을 만든다 nest g app gateway, nest g app account, nest g app chat

2. 생성한 앱은 rest 기반으로 만들기 때문에 graphql 을 위한 resolver 생성이 필요하다. nest g resource

   gateway 를 제외한 다른 app 들을 각자 선택해서 resolver 를 생성하면 기본 구조(resolver, service, dto, entity...) 및 로직(CRUD)을 만들어준다.

3. 이제 gateway 에서 각 앱들을 하나로 연결을 시켜줘야 한다.

   여기서 중요한 개념이 있다.

Federation

Federation(직역하면 연방)의 의미는 graphql 서버를 각 microservice 로 주입을 해준다라는 의미이다.

요청한 쿼리를 하나의 리졸버에서 모두 해결하는 것이 아닌, 각 subgraph 에서 스키마를 처리해서 처리된 schema 를 superGraph 에서 취합해서 프론트로 던져준다.

Federation 을 구현하기 위해서는 하나의 gateway 와 한개 이상의 microservice(federated microservice) 가 필요하다. 이때 gateway 를 supergraph 라고 하고, 각각의 service 들을 subgraph 라고 이해하면 편하다.

federated microservice(subgraph) 는 각각의 schema 를 가지고 있고, gateway(supergraph) 는 그 schema 들을 한 곳으로 모을 수 있다.

🔥 물론 하나의 resolver 에서 모든 로직을 처리할 수는 있지만, 그렇게 하면 여러 테이블에 join 을 걸어야 하고 그렇게 되면 graphql 을 사용하고도 overFetching 을 할 수 밖에 없다. 그렇기 때문에 원하는 값만 조회하기 위해 각 entity를 담당하는 resolver 에서 각각 처리하고 superGraph 에서 해당 entity 들을 모아서 반환하는 것이다.

그러면 각각의 subgraph 를 어떻게 한번에 모으는지 아래 코드를 통해 확인해보자.

// gateway.module.ts
import { Module } from '@nestjs/common';
import { GatewayService } from './gateway.service';
import { GraphQLModule } from '@nestjs/graphql';
import { ApolloGatewayDriver, ApolloGatewayDriverConfig } from '@nestjs/apollo';
import { IntrospectAndCompose } from '@apollo/gateway';
@Module({
  imports: [
    GraphQLModule.forRoot<ApolloGatewayDriverConfig>({
      driver: ApolloGatewayDriver,
      gateway: {
        supergraphSdl: new IntrospectAndCompose({
          // 각각의 subGraph 를 하나로 연결 시켜준다.
          subgraphs: [
            {
              name: 'account',
              url: 'http://localhost:3001/graphql',
            },
            {
              name: 'talk',
              url: 'http://localhost:3002/graphql',
            },
          ],
        }),
      },
    }),
  ],
  providers: [GatewayService],
})
export class GatewayModule {}

코드는 간단하다. ApolloGatewayDriver, IntrospectAndCompose class 를 이용해서 각각의 subgrapqh 를 통합해줄 수 있다.

각각의 federated microservice 의 schema 중 @ObjectType 간의 relation 을 활용하면 하나의 query 에서 데이터를 가져오는 것이 가능하다. 그러기 위해서는 directive 지시어 가 필요하다.

Directive

graphql 에서는 동일한 이름의 @ObjectType 이 존재 하면 안된다

하지만 여러 앱에서 동일한 모델을 쓰고 싶을 수 있다. 예를 들어, 계정 페이지에서 고객의 정보를 보여줄 수도 있고, 채팅방에서 고객 정보를 보여줄 수도 있다. 채팅방에서는 채팅 정보와 고객의 정보를 같이 보여줘야 하는 경우가 생긴다. MSA 구조에 따라서 각 앱에서 각 모델을 교차해서 종속시키는건 옳지 않은 방법이기 때문에, 각 앱에서 각 모델을 둘다 선언을 해준다.

// chat > chat.model.ts
import { Directive, Field, ID, ObjectType } from '@nestjs/graphql';
@ObjectType()
@Directive('@key(fields: "id")')
export class AccountModel {
  @Field(() => ID)
  id: string;
}



// account > account.model.ts
import { Directive, Field, ID, ObjectType } from '@nestjs/graphql';
@ObjectType()
@Directive('@key(fields: "id")')
// 동일 이름의 objectType 이 있으면 안된다
// -> 동일한 objectType 이 있다면 반드시 @Directive('@key(fields: "id")') 로 obejct 의 pk 를 설정해줘야 함
export class AccountModel {
  @Field(() => ID)
  id: string;

  @Field(() => String)
  name: string;
}

이렇게 될 경우 schema 에 같은 이름의 모델이 생겨버리기 떄문에 graphql 철학에 위배된다. 이때, @Directive('@key(fields: "id")') 로 문제를 풀어준다. 저 데코레이터는 중복을 허용해준다는 의미를 포함하고 있다. 또한, 안에 있는 @Key 는 모델에도 테이블처럼 PK 를 지정해주는 것이다. 저 PK로 정한 id 가 같은 모델이 있으면 두개는 같은것으로 판단해서 하나로 묶어준다. 결국

@ObjectType()
export class AccountModel {
  @Field(() => ID)
  id: string;

  @Field(() => String)
  name: string;
}

// schema.gql
type AccountModel {
  id: ID!
  name: String!
}

다음과 같은 하나의 모델이 되는 것이다.

ResolveField

처음에 이거 떄문에 애를 썩었다. 아무리 구글링해서 나오는 설명을 봐도 이해가 안되고, 대신에 이거를 쓰면 dataLoader 뭐시기 뭐시기... 그래서 유튜브 영상 및 지인들을 통해서 알아냈다.

ResolveField... 일단 Field 다. resolve... 해결하다?

의역하면 필드를 하나 열어주는 것이다. 즉 ResolveField 가 속해있는 리졸버에서 반환하는 모델에 필드를 하나 더 추가하는 행위다.

// chat.resolver.ts
import {
  Resolver,
  Query,
  Args,
  Int,
  ResolveField,
  Parent,
} from '@nestjs/graphql';
import { ChatService } from './chat.service';
import { AccountModel, ChatModel } from './entities';

@Resolver(() => ChatModel)
// 이 resolver 는 chatModel 을 뽑아내기 위한 리졸버이다
export class ChatResolver {
  constructor(private readonly chatService: ChatService) {}

  @Query(() => ChatModel)
  async getChatInfo(
    @Args('roomId', { type: () => Int }) roomId: number,
  ): Promise<ChatModel> {
    return {
      chatId: roomId,
    };
  }

  @ResolveField(() => AccountModel)
  user(@Parent() chat: ChatModel) {
    return { __typename: 'UserEntity', id: chat.chatId };
  }
}



// chat.model.ts
@ObjectType()
export class ChatModel {
  @Field(() => Number)
  chatId: number;
}

위 리졸버는 ChatModel을 반환하는 리졸버이다. @ResolveField 를 통해서 ChatModel 에 user 라는 필드를 추가적으로 반환하겠다라는 뜻이다.

근데 User 는 AccountModel 이다. AccountModel 은 Chat 앱이 아니라, Account 앱에 종속되어있다. 어떻게 가져오지? 이때 @Directive 가 쓰인다!!!

@Query() getChatInfo 쿼리는 ChatModel 을 반환한다. 그러면 @Parent() 가 ChatModel 인 user @ResolveField() 가 이를 캐치한다. user @ResolveField() 가 { __typename: 'UserEntity', id: chat.chatId } 를 반환하는데, 이를 누구한테 반환하는가..?

✅ 앞서 말했듯이 graphql 에서는 동일한 ObjectType이 있을 수 없다. chat app 에 종속되어 있는 Account Model 와 Account app 에 종속되어 있는 Account Model 이 @Directive key 로 연결되어 있기 때문에, AccountModel 을 종속하고 있는 Account app 에서 이를 수신한다.

만약에 또다른 app 에서도 Account Model 을 사용하고 있다면? 어느쪽에서 수신할까? 이런 고민은 잘못됐다. 애초에 Account Model 은 Account app 에서 다루는것이 옳은 방법이다. (관심사의 분리가 적용된다) 그렇기 때문에 애초에 또 다른 app 에서는 @ResolveReference() 이 아니라, @ResolveField() user 를 사용하는 것이 맞다. 즉, Account Model 은 Account app 에서만 로직을 풀어가자!

// account.resolver.ts
/**
   *
   * @description chat resolver 에서 처리하지 못하는 쿼리를 여기서 처리함 (find account)
   * resolverField 를 여기서 수신해서 처리한다.
   */
  @ResolveReference()
  resolveReference(reference: {
    __typename: string;
    id: string;
  }): AccountModel {
    return {
      id: reference.id,
      name: `${reference.id} + name`,
    };
  }

이렇게 ResolveField -> ResolveReference -> ResolveField -> ResolveReference -> ... 계속 하면 가장 마지막에 수신하는 리졸버를 수만가지 요청을 처리해야 하는 경우가 생긴다. 이를 N+1 문제 라고 하는데, 다음에 다뤄보자.

Ref

https://blog.doctor-cha.com/integrating-graphql-services-with-graphql-federation https://www.apollographql.com/docs/federation/

https://unsplash.com/ko/%EC%82%AC%EC%A7%84/%ED%8C%8C%EB%9E%80%EC%83%89-%ED%8E%98%EC%9D%B8%ED%8A%B8%EA%B0%80-%EC%B9%A0%ED%95%B4%EC%A7%84-%EC%82%AC%EB%9E%8C-%EB%B0%9C-EIyAz8blaAk

singleton 객체의 인스턴스가 오직 1개만 생성되는 패턴 메모리에 인스턴스를 한번만 만들어서 올려놓고 필요할 때 마다 생성하는 패턴이다.

IOC(inversion of control) 제어의 권한을 바꾼다 인스턴스의 생성 및 할당과 해제를 개발자가 하는 것이 아니고, 프레임워크가 이를 맡아주는 것을 말한다.

DI(dependency of injection) 의존성 주입 A가 B를 사용하기 위해 직접 B를 생성하는 것이 아니라, 외부로부터 가져와서 사용한다는 것을 말한다.

❗️ 의존성 주입을 사용하지 않을 경우 단점 ❗️

1. A를 B에서도 사용하고 C에서도 사용을 하는데 둘다 의존성 주입이 아니라 직접 인스턴스를 생성해서 사용한다고 가정을 해보자. (A->B / A->C) 이때 A가 변하면 B, C에서도 수정된 A를 다시 생성해야 한다. 이를 강한 결합 이라고 한다.
2. 또한, A의 인스턴스들의 생명주기를 B, C 가 직접 관리를 해주면 메모리 관리를 해야한다. 이때 memory leak 가 발생할 수 있다.

여기서 잠깐! 근본적으로 생각해보자. 이런 의존성이니 주입이니 이런말이 왜 나온걸까? 아키텍처를 구성할때, 도메인에 따라서 서비스 레이어를 구분하도록 구성하면 구현체의 로직이 변경되도 호출부에서는 해당 내용을 알 필요가 없다. 이로써, 생산성과 재사용성이 좋아지게 된다.

nest는 내부적으로 IOC container 를 이용해서 DI를 관리해준다 ioc container는 @Injectable(), @Module() 데코레이터가 달린 클래스들을 DI대상으로 관리한다. nest 는 typescript 이기 때문에, 타입기준으로 DI를 관리하기 수월하다.

✅ nestjs 싱글톤 패턴의 특징

• 공유된 자원을 사용하기 위해.
  • 하나의 인스턴스를 여러 곳에 사용했을 경우 동시성 이슈와 데드락 같은 문제를 방지할 수 있다.
• 인스턴스를 필요할 때 마다 생성하고 필요없을 때, 제거 하면서 까지 메모리 관리를 신경 쓰지 않아도 된다.
• 하나의 인스턴스를 global 하게 사용할 수도 집약적으로 사용할 수 있다.
  • global 하게 사용한다면 A클래스에서 B클래스에서 사용하는 인스턴스에 접근을 방지 하기 위해 private 을 선언해주는 방법도 있다.

✅ 구현방법

nest는 애초에 싱글스레드이기 때문에, 인스턴스간의 동시성이 발생하지 않는다고 한다. nestjs 에서는 docs 에서 알 수 있듯이, singleton 을 지향하고 있다. 하지만 무조건 싱글톤으로 구현되는 건 아니다. 아래서 다뤄보겠다.

싱글톤을 구현하기 위해서는 간단하게 설명을 하자면, @Injectable(), @Module() 데코레이터를 사용하면 된다. @Injectable(), @Module() 데코레이터를 선언하면, 해당 인스턴스를 nest 내장 IOC container 가 관리한다 (데코레이터가 달린 클래스는 타입스크립트가 컴파일시 메타데이터로 어떤 서비스에 의존하고 있는지 명시를 해준다. 그러면 nestjs 가 어떤 의존관계가 있는지 알 수 있다.)
👉 nest 내부의 IOC container 가 @Injectable(), @Module() 데코레이터 클래스를 싱글톤으로 생성하여 DI의 대상으로 관리한다.

✅ 뭐야 두개 생기는데?

socket adapter 를 공부하다가 소켓에 연결하면 즉각 실행하는 afterInit() 이라는 함수를 구현하다가 뭔가 이상함을 봤다.

// ws.gateway.ts
@WebSocketGateway(80)
export class WsKorGateway implements OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect {
  constructor(private readonly logger: Logger){
  }
  @WebSocketServer()
  wsServer: Server;
  /**
     * 
     * @param server 해당 gateway 가 실행되면 가장 먼저 실행되는 함수 -> handleConnection 보다 먼저 실행된다
     */
  afterInit(server: any) {
    this.logger.log('kor ws gateway started!') 
    // 터미널을 보면 이게 두번 실행됨을 알 수 있다.
  }
}



// ws.module.ts
@Module({
    providers: [WsKorGateway, Logger]
})
export class WsKorModule {
}



// app.module.ts
@Module({
  imports: [WsKorModule, WsUsaModule],
  controllers: [AppController],
  providers: [AppService, WsKorGateway, Logger],
})
export class AppModule {}

위와 같이 소스코드를 구현해 놓고, ws server 를 구동하면 다음과 같은 결과가 나온다.

위 사진을 보면 kor connected! 라는 로그가 두번찍혔는데 이로써 afterInit() 함수가 두번 실행됐음을 알 수 있다.

app module 에서 import 에서 한번 providers 에서 한번, 총 두번 주입을 해줬기 때문에, WsKorGateway 가 두번 실행되서 로그가 두번찍히는건 알겠는데

app module 에서 import 에서 한번, providers 에서 한번, 총 두번 주입을 하지만 한개의 인스턴스를 그냥 두번 주입하는게 아닌가?

nest 는 인스턴스를 싱글톤으로 만든다고 했는데..?

여기서 그냥 아무 decorator 로 싱글톤을 구현하는것이 아닌것을 알 수 있다. @injectable() @Module() 이 두개로만 싱글톤이 적용되는 것으로 추측된다.

✅ 주입된 놈이 수정되면 나도 컴파일 해야 하나?

@Controller('payment')
export class PaymentController extends CommonController {
  constructor(
    @Inject('PAYMENT_SERVICE')
    private readonly paymentService: PaymentService,
  ) {
    super();
  }

위 같은 경우는 PaymentService 가 변하면 PaymentController 도 같이 컴파일을 다시 해야한다.

@Controller('talk')
export class ChatController extends BaseController {
  constructor(
    @Inject(CHAT_SERVICE)
    private readonly chatService: IChatService,
  ) {
    super();
  }

하지만 다음과 같은 경우는 Interface 를 주입받았기 때문에, ChatService의 구현체가 변해도 ChatController 는 컴파일 하지 않아도 된다. 이게 interface 를 사용하는 중요한 이유다 (일종의 추상화)

References reference 1 reference 2

입사했을 당시, 회사 백엔드 코드의 스택은 nestjs + graphql 이었다.

그 당시 graphql 에 대해서 잘 알지 못했고, 코드 구조를 보고 graphql 은 원래 이렇게 사용하는구나 싶었다.

새로 오신 시니어 개발자 분이 현재 코드 구조가 graphql 의 철학과 맞지 않는 것 같다 라는 의견과 함께, 대규모 개편에 앞서 graphql 을 철학에 맞게 사용하거나 혹은 restapi 를 새롭게 도입하자 라는 의견을 제시해줘 고민을 하다가 시간이 촉박한 탓에 그래도 좀 더 익숙하고 레퍼런스가 많은 restapi 를 사용하자라는 결론이 났었다.

결과적으로 대규모 개편은 restapi 로 개발을 하였다. 배포를 하고 나니, 그렇다면 정말 graphql 의 철학은 무엇인가? 라는 의문이 들기 시작한다.

다시 초심의 마인드로 겸손하게 하나하나 차근차근 공부해보자.

graphql 은 ql 에서 유추할 수 있듯이, 데이터를 가져오는 쿼리 언어이자, 서버측 런타임이다.

restapi 는 하나의 api 요청에 하나의 데이터를 가져올 수 있지만, graphql 은 하나의 api 요청에 여러 데이터 소스에서 데이터를 가져올 수 있다.

✅ schema

gql 에서 사용하는 모든 resolver(query, mutation, subscription), model 등을 설명 및 정리한 것이다.

# ------------------------------------------------------
# THIS FILE WAS AUTOMATICALLY GENERATED (DO NOT MODIFY)
# ------------------------------------------------------

type Message {
  id: String!
  message: String!
}

type Query {
  getMessages: Message!
}

graphql 의 가장 큰 장점은 프레임워크와 상관없이 프론트에서 schema generate 를 통해서 schema.gql 전체를 한번에 긁어올 수 있다는 것 같다. 이는 정말 엄청난 생산성을 높일 수 있다. rest api 는 이것이 불가능해서 생각보다 불편했다.

✅ resolver

schema 에 정리된 데이터를 호출하기 위한 일종의 controller 같은 것이다. 실제 데이터를 가져오는 로직이 들어가 있다.

✅ schema first vs code fisrt

1. 설계 방법 중에 스키마를 먼저 설계 하고 이를 바탕으로 코드를 짜는 방법이 있고,
2. 코드를 먼저 생성하고 graphql module 의 autoSchemaFile 를 이용하여 스키마를 자동 생성하는 방법이 있다.

검색을 해보니 말들이 많다. 각자의 장점과 단점이 있다. schema fisrt 가 레퍼런스가 더 많고, 이를 우선적으로 공부를 하라고 많이들 나와있다. 하지만, 이 방법이 현업에서 정말 더 좋을까?

schema 를 먼저 설계를 하고 배포를 하면 시간을 더 줄일 수 있는건 확실하다. schema 를 기반으로 프론트와 백엔드가 동시에 개발을 할 수 있기 때문이다.

하지만 결정적으로 resolver 와 schema 간의 동기화가 맹점이다.

nestjs 를 restapi 로 개발을 하다 보니, 동기화 작업이 여간 귀찮고 까탈스러운 일이 아니였다. swagger 를 도입한다고 해도, 불편한 점이 더러 있었다.

동기화를 위해서는 code-first 방식이 더 좋을 것 같다. (개발을 하면서 장단점을 좀 더 비교해보겠다)

✅ graphql 장점

GraphQL 호출은 단일 왕복으로 처리되며 클라이언트는 오버페칭 없이 요청한 결과만 얻습니다 라는 말은 너무 매력적으로 다가온다. fetching 이 하나라면, 네트워크 비용을 줄일 수 있다는 말로 들리는데, 대규모 배포 전의 graphql 백엔드 코드는 단일 왕복이 아니였다. 이를 중점적으로 공부를 해보려고 한다.

REST API의 단점으로 많이 거론되는 바로 ‘오버패칭(필요한 정보 외에도 불필요한 정보까지 받게 되는 것)’과 ‘언더패칭(필요한 정보를 서버에서 가져오지 못해 추가 요청을 해야 하는 경우)’ 문제도 있다.

한 예로, 현재 운영중인 서비스에는 제조 파트너 정보를 보여주는 화면이 있다. 해당 화면에는 파트너의 각 종 정보를 다 보여줘야 한다. (프로필, 포트폴리오, 리뷰, 소개서, 인증서, 통계, 장비, etc..) 이것들을 호출하기 위해 너무 많은 api 를 호출해야 한다고 생각했다. 이러한 문제도 graphql 로 어느정도 해결할 수 있지 않을까?

✅ 코드

/// app.module.ts
@Module({
  imports: [
    GraphQLModule.forRoot<ApolloDriverConfig>({
      driver: ApolloDriver,
      playground: true,
      autoSchemaFile: 'src/schema.gql',
      path: 'v1/gql',
    }),
    ChatModule,
  ],
  controllers: [AppController],
  providers: [AppService],
})
export class AppModule {}

/// chat.resolver.ts
import { Resolver, Query } from '@nestjs/graphql';
import { ChatService } from './chat.service';
import { Message } from './model';

@Resolver('Chat')
export class ChatResolver {
  constructor(private readonly chatService: ChatService) {}

  @Query(() => Message)
  async getMessages(): Promise<Message> {
    return await this.chatService.getMessages();
  }
}

/// model/message.ts
import { Field, Int, ObjectType } from '@nestjs/graphql';
import { randomUUID } from 'crypto';

@ObjectType()
export class Message {
  @Field(() => String)
  id: string;

  @Field(() => String)
  message: string;
}

ApolloDriver 이 @Resolver(), @ObjectType(), @InputType() decorator 를 다 긁어와서 schema 를 자동 생성해주는 것 같다.

다음과 같이 프론트는 백엔드에서 던져주는 값들 중에 원하는 값만 받을 수 있다. (message 필드는 무시하고, id 만 받은 것을 볼 수 있음)

다음에는 orm 및 mutation 에 대해서 알아보겠다.

REF

https://unsplash.com/ko/%EC%82%AC%EC%A7%84/%ED%8C%8C%EB%9E%80-%EA%B3%84%EB%8B%A8%EC%9D%84-%EB%B0%9F%EB%8A%94-%EC%82%AC%EB%9E%8C-7_kRuX1hSXM

sokcet 은 기본적으로, http 와 같이 프로토콜의 한 종류이며, http는 단방향 통신에 한번 통신을 하고 나면 연결을 끊어버리는 성격과 다르게, socket 은 양방향 통신에 한번 연결을 해도 끊어버리지 않고, 연결을 유지하기 때문에, 실시간으로 소통해야 하는 경우에 자주 사용된다. 예를 들면, 채팅 같이 실시간 통신을 하기 위한 기능에 자주 사용된다.

socket 을 사용되는 기술에는 websocket 과 socketio 가 있는데 차이점은 아래와 같다.

WebSocket 과 socketIO 의 차이점

ws 는 프로토콜이다(웹소켓이 등장하기 전에는 http 로만 통신을 했기 때문에 주로 polling 을 사용함)
io 는 node로 만든 websocket 기반 양방향 통신 모듈(라이브러리)이다.
ws 를 지원하지 못하는 브라우저가 많아서 io가 생김.

그러면 어떻게 구현할 수 있는지 알아보자.

install package

$ npm i --save @nestjs/websockets @nestjs/platform-socket.io $ npm i --save-dev @types/socket.io

architecture

socket 서버를 열기 위해서는 일단, nest 서버가 있어야 한다. 그리고 nest 서버에서 socket 서버를 주입받아서 사용한다. 어떻게 되어 있는지 코드를 통해 알아보자.

// ws.gateway.ts
@WebSocketGateway(80)
export class WsKorGateway implements OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect {
    constructor(private readonly logger: Logger){}
  ... 이하 생략
}

// ws.module.ts
@Module({
    providers: [WsKorGateway, Logger]
})
export class WsKorModule {
}

socket 서버만을 여는 socket gateway 모듈화 시켜준다.

// app.module.ts
@Module({
  imports: [WsKorModule, WsUsaModule],
  controllers: [AppController],
  providers: [AppService],
})
export class AppModule {}

socket 모듈(WsKorModule)을 app module 에 주입해준다.

app module 에서는 단순하게 socket 서버를 여는 방법으로 연결을 할 수도 있지만, socket 서버만 달랑 하나 여는 것 보다, adapter 를 이용하면, 여러 socket server 를 통합하여 관리할 수 있다. 아래는 1. socket io adapter 또는 2. web socket adapter 로 관리하는 방법이다.

// main.ts
async function bootstrap() {
  const app = await NestFactory.create(AppModule);

  app.enableCors({
    origin: true,
    credentials: true,
  });

  // 1. io adapter init using socketIO
  const redisIoAdapter = new RedisIoAdapter(app);
  await redisIoAdapter.connectToRedis();
  app.useWebSocketAdapter(redisIoAdapter);

  // 2. ws adapter init using WS (faster then socketIO, lower skill then socketIO)
  app.useWebSocketAdapter(new WsAdapter(app));


  await app.listen(3000);
}
bootstrap();

코드에 대한 설명은 아래에서 더 자세하게 다뤄보겠다.

socket server gateway

Gateway lifeCycle

ws gateway 가 연결이 되고 끊어때까지의 생명주기는

implements OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect

를 통해서 알 수 있다.

export interface OnGatewayInit<T = any> { 
  // ws gateway 시작되는 순간 가장먼저 실행됨
    afterInit(server: T): any;
}

export interface OnGatewayConnection<T = any> { 
  // 소켓 연결되면 실행됨
    handleConnection(client: T, ...args: any[]): any;
}

export interface OnGatewayDisconnect<T = any> { 
  // 소켓 연결 끊어지면 실행됨
    handleDisconnect(client: T): any;
}

OnGatewayInit

ws gateway 에 namespace 속성을 추가하면 해당 namespace 로 방을 만들었음을 알 수 있다. (nsps 는 namespaces 의 축약이다 ^^)

web socket 에서 말하는 namespace 는 하나의 방이라고 생각하면 된다. (http 통신에서의 방은 엔드포인트라고 볼 수 있듯이) namespace 이름에 따라 요청을 다르게 처리할 수 있다.

OnGatewayConnection & OnGatewayDisconnect

동일 클라이언트가 연결이 됐다가 끊겼다는 것을 id 로 알 수 있다.

** socket adapter**

adapter는 쉽게 말해, 여러 소켓 서버들을 연결해주는 모듈이라고 생각하면 된다. 여러 서버들간의 데이터를 주고 받는 다거나 상태 값을 공유할 때 쓰인다.

nestjs 에서는 기본적으로 socket을 socketIO 로 연결한다. ws로 연결시켜주기 위해서는 따로 세팅이 필요하다.

1. socketio with not Adapter

Namespace {
  _events: [Object: null prototype] {},
  _eventsCount: 0,
  _maxListeners: undefined,
  sockets: Map(0) {},
  _fns: [],
  _ids: 0,
  server: Server {
    _events: [Object: null prototype] {},
    _eventsCount: 0,
    _maxListeners: undefined,
    _nsps: Map(1) { '/' => [Circular *1] },
    parentNsps: Map(0) {},
    parentNamespacesFromRegExp: Map(0) {},
    _path: '/socket.io',
    clientPathRegex: /^\/socket\.io\/socket\.io(\.msgpack|\.esm)?(\.min)?\.js(\.map)?(?:\?|$)/,
    _connectTimeout: 45000,
    _serveClient: true,
    _parser: {
      protocol: 5,
      PacketType: [Object],
      Encoder: [class Encoder],
      Decoder: [class Decoder extends Emitter]
    },
    ... 이하 생략
  },
  name: '/',
  adapter: Adapter {
    _events: [Object: null prototype] {},
    _eventsCount: 0,
    _maxListeners: undefined,
    nsp: [Circular *1],
    rooms: Map(0) {},
    sids: Map(0) {},
    encoder: Encoder { replacer: undefined },
    [Symbol(kCapture)]: false
  },
  [Symbol(kCapture)]: false
}

그림과 같이 socketIO 로 연결하고 Adapter 또한 사용하고 있지 않는 것을 볼 수 있다. ** 하지만 redis 를 이용해 socketIO Adapter 를 이용한다면 ** (아래 코드가 있음)

2. socketio with Adapter

Namespace {
  _events: [Object: null prototype] {},
  _eventsCount: 0,
  _maxListeners: undefined,
  sockets: Map(0) {},
  _fns: [],
  _ids: 0,
  server: Server {
    _events: [Object: null prototype] {},
    _eventsCount: 0,
    _maxListeners: undefined,
    _nsps: Map(1) { '/' => [Circular *1] },
    parentNsps: Map(0) {},
    parentNamespacesFromRegExp: Map(0) {},
    _path: '/socket.io',
    clientPathRegex: /^\/socket\.io\/socket\.io(\.msgpack|\.esm)?(\.min)?\.js(\.map)?(?:\?|$)/,
   ... 이하 생략
  },
  name: '/',
  adapter: RedisAdapter {
    _events: [Object: null prototype] {},
    _eventsCount: 0,
    _maxListeners: undefined,
    nsp: [Circular *1],
    rooms: Map(0) {},
    sids: Map(0) {},
    encoder: Encoder { replacer: undefined },
    pubClient: Commander {
      _events: [Object: null prototype],
      _eventsCount: 1,
      _maxListeners: undefined,
      commandOptions: [Function: commandOptions],
      select: [AsyncFunction: SELECT],
      subscribe: [Function: SUBSCRIBE],
      unsubscribe: [Function: UNSUBSCRIBE],
      pSubscribe: [Function: PSUBSCRIBE],
      pUnsubscribe: [Function: PUNSUBSCRIBE],
      sSubscribe: [Function: SSUBSCRIBE],
      sUnsubscribe: [Function: SUNSUBSCRIBE],
      ... 이하생략
    },
    subClient: Commander {
      _events: [Object: null prototype],
      _eventsCount: 1,
      _maxListeners: undefined,
      commandOptions: [Function: commandOptions],
      select: [AsyncFunction: SELECT],
      subscribe: [Function: SUBSCRIBE],
      unsubscribe: [Function: UNSUBSCRIBE],
      pSubscribe: [Function: PSUBSCRIBE],
      pUnsubscribe: [Function: PUNSUBSCRIBE],
      sSubscribe: [Function: SSUBSCRIBE],
      sUnsubscribe: [Function: SUNSUBSCRIBE],
      ... 이하생략
    },
    ... 이하생략
    channel: 'socket.io#/#',
    requestChannel: 'socket.io-request#/#',
    responseChannel: 'socket.io-response#/#',
    specificResponseChannel: 'socket.io-response#/#aVBDBs#',
    friendlyErrorHandler: [Function (anonymous)],
    [Symbol(kCapture)]: false
  },
  [Symbol(kCapture)]: false
}

다음과 같이 socketIO 를 사용하면서 Adapter 또한 사용함을 알 수 있다.

코드

// io-adapter.ts
import { IoAdapter } from '@nestjs/platform-socket.io';
import { ServerOptions } from 'socket.io';
import { createAdapter } from '@socket.io/redis-adapter';
import { createClient } from 'redis';

export class RedisIoAdapter extends IoAdapter {
  private adapterConstructor: ReturnType<typeof createAdapter>;

  async connectToRedis(): Promise<void> {
    const pubClient = createClient({ url: `redis://localhost:6379` });
    const subClient = pubClient.duplicate();

    await Promise.all([pubClient.connect(), subClient.connect()]);

    this.adapterConstructor = createAdapter(pubClient, subClient); // publish client 와 subscribe client 끼리 연결?
  }

  createIOServer(port: number, options?: ServerOptions): any {
    const server = super.createIOServer(port, options);
    server.adapter(this.adapterConstructor);
    return server;
  }
}

3. web socket with Adapter

다음으로 ** ws adpater ** 를 사용하도록 설정하면 (아래 코드 있음)

WebSocketServer {
  _events: [Object: null prototype] {
    connection: [Function (anonymous)],
    error: [Function (anonymous)]
  },
  _eventsCount: 2,
  _maxListeners: undefined,
  _server: <ref *1> Server {
    maxHeaderSize: undefined,
    insecureHTTPParser: undefined,
    requestTimeout: 300000,
    headersTimeout: 60000,
    keepAliveTimeout: 5000,
    connectionsCheckingInterval: 30000,
    joinDuplicateHeaders: undefined,
    ... 이하 생략
  _removeListeners: [Function: removeListeners],
  clients: Set(0) {},
  _shouldEmitClose: false,
  options: {
    ... 이하생략
    port: 80,
    WebSocket: <ref *2> [class WebSocket extends EventEmitter] {
      CONNECTING: 0,
      OPEN: 1,
      CLOSING: 2,
      CLOSED: 3,
      createWebSocketStream: [Function: createWebSocketStream],
      Server: [class WebSocketServer extends EventEmitter],
      Receiver: [class Receiver extends Writable],
      Sender: [class Sender],
      WebSocket: [Circular *2],
      WebSocketServer: [class WebSocketServer extends EventEmitter]
    }
  },
  _state: 0,
  [Symbol(kCapture)]: false
}

코드

// ws-adapter.ts
import * as WebSocket from 'ws';
import { WebSocketAdapter, INestApplicationContext } from '@nestjs/common';
import { MessageMappingProperties } from '@nestjs/websockets';
import { Observable, fromEvent, EMPTY } from 'rxjs';
import { mergeMap, filter } from 'rxjs/operators';

export class WsAdapter implements WebSocketAdapter {
  constructor(private app: INestApplicationContext) {}

  create(port: number, options: any = {}): any {
    return new WebSocket.Server({ port, ...options });
  }


  bindClientConnect(server, callback: Function) {
    server.on('connection', callback);
  }

  bindMessageHandlers(
    client: WebSocket,
    handlers: MessageMappingProperties[],
    process: (data: any) => Observable<any>,
  ) {
    fromEvent(client, 'message')
      .pipe(
        mergeMap(data => this.bindMessageHandler(data, handlers, process)),
        filter(result => result),
      )
      .subscribe(response => client.send(JSON.stringify(response)));
  }

  bindMessageHandler(
    buffer,
    handlers: MessageMappingProperties[],
    process: (data: any) => Observable<any>,
  ): Observable<any> {
    const message = JSON.parse(buffer.data);
    const messageHandler = handlers.find(
      handler => handler.message === message.event,
    );
    if (!messageHandler) {
      return EMPTY;
    }
    return process(messageHandler.callback(message.data));
  }

  close(server) {
    server.close();
  }
}

얼추 어떻게 다르고 역할을 하는지 알았으니, 이제 adapter (redis io apater) 를 이용해서 테스트를 해보자. 두개의 소켓 서버를 생성한다

port : 80(kor), 81(usa)

테스트는 postman 으로 했다.

postman test video

영상을 보면 한국에서 보낸 메세지를 미국에서도 수신할 수 있는 것으로 보인다.

이 처럼 다른 서버간의 소켓 통신을 위해서는 adapter 로 통신할 수 있음을 확인할 수 있다.

github

https://github.com/jjmmll0727/potential-nestjs-memory/tree/main/socket-kafka

이번에는 기존에 존재하는 결제 기능에 대출서비스, 마켓 기능을 추가로 붙이다 보니 추가해야 할 결제 관련 테이블들이 늘어났다. ex) 세금계산서, 거래명세서, 주문이력, 대출이력...

물론 현재 있는 결제 테이블에 계속 더하거나 비정규화를 통해서 만들 수 있었지만, 이렇게 계속 개발하면 발전도 없고, 점점 시간이 지나서 다른 pg 사를 사용하던가 다른 방법으로 결제를 하게 되는 기능이 생긴다면 무결성은 위배될 것이고, 스키마 확장성은 줄어들 것으로 보였다.

그래서 이번에 기존에 있던 결제 관련 스키마를 재설계 하면서 대출 기능을 추가하기로 하였다. 대출 기능 배포 기간도 맞춰야 하는 문제가 있고, 재설계를 한다면 기존에 구현되어 있던 기능들 테스트도 물론 해야하는 문제도 있다.

🔥어떻게 진행했나🔥

두번째 문제를 잘 해결하기 위해 우리는 이런 방법으로 진행했다.

1) 기존 테이블에 insert 하는 로직은 그대로 두고, 새롭게 설계된 테이블에도 insert 를 한다. 기존 테이블이 payment 라는 테이블이라면 우리는 payment_root 와 payment_rfq, payment_strategy_nicepay 와 같이 분할했다. insert into payment 하는 로직에서 insert into payment_root, insert into payment_rfq, insert into payment_strategy_nicepay 를 같이 하는거다.

2) 조회하는 로직에서는 대출 관련 데이터는 새롭게 설계된 테이블을 대상으로 새롭게 로직을 짜고, 대출 아닌 데이터에 대해서는 기존에 사용하던 로직 그대로 이용 대신 추후에 데이터를 이관했을 때에도 정상적으로 조회가 될 수 있게끔 쿼리를 짠다. 👉 이 부분이 쿼리를 짜는데 까다로웠다.

3) 수정하는 로직은 다행히도 대출관련 결제 기능에서는 수정하는 부분이 많지 않아서 2번과 비슷하게 기존에 사용하던 수정 로직은 사용하고, 대출관련 데이터들은 새롭게 만들었다. 수정하는 로직은 배포하고 나서 천천히 같이 바꿔줘도 되기 때문에 급하지 않았다.

물론 회사가 운영하고 있는 서비스의 상황이나 디비의 상황에 따라 방법은 다르겠지만, 기존에 잘 쓰고 있는 로직을 한꺼번에 다 바꾸는 것은 리스크가 따른다. 그렇기 때문에 우리는 조회부터 바꿀꺼야 우리는 삽입부터 바꿀꺼야 이런것들을 결정하는 건 회사 상황에 따라 결정하면 되지만, 이거 하나는 확실한 것 같다. 새로 설계된 테이블에 삽입할때는 새로 설계한 테이블에도 같이 넣어주는 것이다.

그리고 나중에 배포하기 전에 데이터 마이그레이션 했을 경우 새로 만든 조회 로직으로도 잘 나오면 땡큐고 아니면 일단 마이그레이션 롤백하고 배포 후에 다시 조회 로직 수정하면 되는 것이다.

📦 new schema

현재는 단건결제만 있지만, 대출이 들어오면서 분할결제도 생기고 추후에 있을 복합결제까지 고려해서 설계 하였다.

<단건 결제> 👉🏻 APP : 통합결제 : 결제서비스 : 수단 = 1 : 1 : 1 : 1 ex) 발주서 1 : 통합결제 1 : RFQ 결제 1 : 나이스페이 1

<분할 결제> 👉🏻 APP : 통합결제 : 결제 서비스 : 수단 = 1 : N : N : N ex) 대출 결제 (선결제 10%, 대출 90%) 발주서 1 : 통합결제 2 : 결제서비스 2 : 수단 2 (나이스페이 1 + 대출 1)

<복합 결제> 👉🏻 APP : 통합결제 : 결제서비스 : 수단 = 1 : 1 : N : N ex) 포인트로 일부 결제 발주서 1 : 통합결제 1 : 결제서비스 2 : 수단 2 (나이스페이 1 + 포인트 1)

☠️ 이거 생각보다 할일이 많은데...?

개발하면서 잘못 건들였다는 생각이 한두 번이 아니였다. 기존에 구현되어 있는 로직 수정하는 것도 일이었고, 기존에 비정규화되어 있던 테이블에 있던 칼럼들이 뿔뿔이 흩어지는 것 이기 때문에 체크할 것들도 너무 많았다.

괜히 뭐 하나 빠트리거나 중복되는 로직이 생긴다면 이대로 또 골치였다.

그래도 이렇게 기존에 잘 쓰고 있는 테이블을 재설계 하는 경험은 어디서도 할 수 없는 값진 경험이었다.

아직 대출 기능 개발중이지만, 이미 엎질러진 물이다. 문제가 생기더라도 어떻게 해서든지 반드시 해결해야 한다. 2023.10.01

🔥 개발 계기

CX 팀에서 VOC 를 듣고 핸들링할 때 BO사이트에서 대고객 사이트를 미러링 할 수 있는 기능이 있으면 좋겠다 라는 애로사항이 있어서 미러링 기능을 만들게 되었다.

실제 완벽한 미러링이라면, 원격지원처럼 실시간으로 마우스 포인터도 움직여지고 클릭 하나하나 전부 동시 확인 가능한 기능이겠지만, 우리는 그렇게 까지는 아니고, 고객의 동의하에 고객의 계정으로 로그인을 해서 접속하는 방식을 채택했다.

🔥 시나리오

시나리오는 다음과 같다.

1. BO 프론트에서 BO 백으로 미러링 요청을 한다.
2. BO 백에서 일련의 검증 후에 hmac signiture 을 만들어 반환한다.
3. BO 프론트에서 해당 hmac signiture 을 가지고 메인 사이트 프론트로 끌고간다.
4. 메인 사이트 프론트에서 메인 사이트 백으로 hmac signiture 를 전달한다.
5. hmac signiture 를 다시 BO 백으로 전달후, hmac signiture 를 검증한다.
6. 검증 후 이상 없으면 메인 사이트 토큰(authentication token)을 발급한다.

🔥 프로세스

🔥 질문

1. 왜 굳이 bo api 에서 발급한 hmac signiture 를 bo api -> bo -> capa -> capa api -> bo api 와 같이 복잡하게 돌아가는가 ? 👉 bo api 로 토큰 발급 요청이 우리 서비스 capa api 로 부터 온 것인지 아닌지를 확인할 수 있다.

2. timeStamp(5s)는 무슨 역할인가 ? 👉 혹여나 누군가 hmac signiture 를 약탈해서 요청을 했을 경우를 대비해서 5초 라는 약간의 방어책을 만든 것이다. 동일한 환경에서 동일한 시그니처를 약탈해도 5초가 넘어가면 이상 요청이라고 판단하기 위함이다.

3. authCode 는 무슨 역할인가 ? 👉 timeStamp 와 마찬가지로, 방어책 역할이다. 동일 환경에서 동일 시그니처로 요청을 보냈는데 5초 안에 요청을 보냈다면 정상 요청이라고 판단할 수 있고, 혹시 이미 보냈던 요청을 그대로 복사해서 다시 요청을 보낼 수도 있다. 또한, timeStamp 는 사실 요청 시간을 변조해서 보낼 수 있기 때문에 완벽한 방어책이라고는 볼 수 없다. 그래서 authCode(uuid)를 만들어 한번 요청을 보내면 디비에 체크를 하고 동일 authCode 로 요청이 오면 이상 요청이라고 판단하기 위함이다.

🔥 코드

const hmac = crypto.createHmac('sha-256', accessKey);
const signiture = hmac.update(message.join('')).digest('base64');

hmac 만드는 코드는 많이 나와있기 때문에 간단하게 넘어가겠다. 주고 받는 쪽에서 그들만의 규칙을 정해서 만들면 되기 때문에, 중요한 건 저 코드 두줄이 될 것으로 보인다.