숨고 커뮤니티 FastAPI gRPC 적용기

들어가며

안녕하세요. 숨고 Back-end Engineer 제임스입니다. 숨고 백엔드 아키텍처는 MSA 환경을 기반으로 구축되어 있으며, 서비스 간 통신에 다양한 프로토콜을 활용해 운영되고 있습니다. 도메인별로 독립적인 서비스 (예: 견적, 채팅, 고수, 유저, 커뮤니티 등)가 존재하며 서비스 간 데이터 교환 시 주로 REST API를 통해 JSON 형식으로 통신을 수행합니다.

그러나 최근에는 챕터내에서 신규 개발시 내부 서비스 간 통신에는 gRPC를 점점 더 많이 채택하고 있습니다. gRPC를 활용한 방식이 기존 대비 서비스 간에 효율적으로 바이너리 데이터를 전송할 수 있고, 성능 최적화가 용이하며 무엇보다 확장성 측면에서 유리하기 때문입니다.

이런 배경으로 MSA 환경에서 gRPC를 활용하여 고수 서비스와 FastAPI를 사용하는 커뮤니티 서비스 간 통신을 구현했던 과정을 공유하고자 합니다.

요구사항 분석 및 검토

스쿼드 내에서 수행했던 주요 작업은 고수의 리뷰 데이터를 작성할 때 커뮤니티 내에서 조회할 수 있도록 구현하는 것이었습니다. 이는 A/B 테스트 실험을 통해 리뷰 콘텐츠를 커뮤니티 기능에 제공했을 때, 사용자들이 긍정적인 반응을 보였다는 결과를 바탕으로 설계되었습니다.

이를 위해 MSA(Microservice Architecture) 환경에서 독립적으로 운영되는 두 개의 서비스인 고수 서비스와 커뮤니티 서비스 간의 데이터 동기화 문제를 해결하는 것을 목표로 삼았습니다. 각 서비스는 별도의 데이터베이스를 사용하고 있었으며, 고수 서비스는 관계형 데이터베이스를, 커뮤니티 서비스는 문서 지향 데이터베이스 (MongoDB)를 사용하고 있었습니다.

이에 따라 스키마가 서로 상이하기 때문에 데이터 모델 간의 차이를 해소함과 동시에 요청과 응답 네트워크 사이에 일관된 데이터를 교환할 수 있도록 해야 했습니다. 또한, 고수 서비스와 커뮤니티 서비스 간의 프레임워크와 구조적 차이도 고려하여 통신 방식과 데이터 처리 방식을 설계하는 것이 필요했습니다.

최적의 기술 스택 선정

상황 진단

이번 일감에서 중점적으로 고려했던 것은 한정된 자원하에 최적의 기술 스택을 선택하는 것이었습니다.

서비스 간 통신 방식은 프로젝트의 성능과 안정성에 큰 영향을 미치기 때문에, MSA 기반의 서비스 간 통신에서 가장 적합한 방식을 신중히 결정해야 했습니다.

백엔드 챕터내에 MSA 기반의 서비스 간 통신 방식으로는 크게 세 가지입니다.

1. REST API를 통한 서버 간 통신
2. gRPC를 활용한 서버 간 통신
3. 메시지 큐(RabbitMQ, Kafka 등)를 통한 비동기 데이터 전송

방식마다 장단점이 존재하지만, 이번 작업에 주요 고려 사항은 데이터 일관성과 동기화였습니다. 특히, 리뷰 작성과 동시에 이벤트 기반으로 커뮤니티에 데이터를 전달하고 이를 기반으로 게시글을 생성하는 구조였기 때문에, 데이터가 항상 일관되게 동기화되는 것이 매우 주요한 과제였습니다. 대표적으로 이중 쓰기가 있습니다.

예를 들어, 고수 서비스에서 리뷰 데이터를 성공적으로 저장했지만, 네트워크 장애나 시스템 오류로 인해 커뮤니티 서비스에서의 데이터 저장은 실패할 수 있습니다. 이 경우 두 서비스 간의 데이터가 데이터 불일치(Data Inconsistency)가 발생합니다. 이 밖에도 네트워크 장애에 대비한 재시도 로직, 보안성 강화, 로깅 및 모니터링 등도 고려 사항이었습니다.

이 문제들을 해결하기 위해서는 통신 방식이 안정적이고 효율적이어야 했으며, 이러한 요소들을 종합적으로 고려한 결과 gRPC가 최적의 선택이라고 결론지었습니다.

gRPC란 무엇인가?

gRPC는 Google에서 개발한 오픈소스 RPC(Remote Procedure Call) 프레임워크입니다. gRPC는 클라이언트와 서버 간의 통신을 간편하게 처리할 수 있도록 설계되었기 때문에 전통적인 HTTP 기반 API 통신 방식보다 더 효율적이고, 다양한 프로그래밍 언어를 지원하는 강력한 원격 호출 시스템을 제공한다는 점이 있습니다.

그리고 Protocol Buffers (ProtoBuf)라는 직렬화 방식을 활용하여 데이터를 주고받으며, HTTP/2 기반의 통신 프로토콜을 사용하여 높은 성능과 확장성을 보장할 수 있습니다.

gRPC의 클라이언트는 서버의 메소드를 원격 서버에서 호출할 수 있는 환경이 제공됩니다. 이를 가능하게 하는 핵심 요소가 채널과 스텁입니다. 이 두 가지는 클라이언트와 서버 간의 통신을 효율적으로 관리하기 위해 설계된 개념입니다.

채널 (Channel)

채널은 클라이언트와 서버 간의 연결을 설정하고 이를 유지하는 역할을 합니다. gRPC는 HTTP/2 프로토콜을 기반으로 하며, 이를 통해 연결을 효율적으로 재사용할 수 있습니다.

• 클라이언트는 채널을 통해 서버로 요청을 보내고 응답을 받습니다
• 한 채널에서 여러 gRPC 호출을 처리할 수 있어 성능과 자원 효율성을 극대화합니다

스텁 (Stub)

스텁은 클라이언트 측에서 서버의 gRPC 메소드를 호출할 수 있도록 제공되는 프록시 객체입니다.

• 클라이언트는 서버와 직접 통신하지 않고, 스텁을 통해 메소드를 호출합니다
• 이는 클라이언트가 원격 서버의 구현을 알 필요 없이, 로컬 환경에서 메소드를 호출하듯 간편하게 사용할 수 있게 합니다

채널은 서버와의 연결을 관리하고, 스텁은 이 연결을 통해 메소드를 호출할 수 있는 인터페이스를 제공합니다.

이 밖에도 제가 gRPC를 선택한 가장 특장점은 다음과 같습니다.

• 고성능: gRPC는 HTTP/2 기반의 통신을 사용하며, 빠르고 효율적인 데이터 전송이 가능합니다. 특히, 바이너리 프로토콜을 사용해 전송 속도를 크게 개선할 수 있었고, REST API 대비 대역폭 절감 효과도 있습니다.
• 강력한 타입 안전성: gRPC는 Protocol Buffers (ProtoBuf)를 사용하여 컴파일 타임 타입 검사를 제공합니다. 이를 통해 서비스 간 통신에서 발생할 수 있는 데이터 불일치를 사전에 방지할 수 있으며, 컴파일 타임에 오류를 발견할 수 있는 장점이 있습니다.
• 양방향 스트리밍: gRPC는 양방향 스트리밍을 지원하여, 클라이언트와 서버 간에 실시간으로 데이터를 주고받을 수 있습니다. 이를 통해 이벤트 기반의 데이터 동기화와 동시에 데이터 불일치를 예방할 수 있으며 실시간 통신을 효율적으로 구현할 수 있습니다.
• 레퍼런스 확보: gRPC는 이미 백엔드 챕터 내에서 다른 서비스들에서도 사용되고 있었기 때문에, 레퍼런스가 풍부합니다. 덕분에 개발에 필요한 시간과 리소스를 절감할 수 있었고, 기존 시스템과의 호환성도 확보할 수 있습니다.

FastAPI와 gRPC 통합

gRPC는 FastAPI와 함께 사용 시 매우 뛰어난 성능을 발휘합니다. FastAPI는 비동기 처리를 기본으로 지원하기 때문에, gRPC와의 통합이 자연스럽게 이루어집니다.

특히 FastAPI는 asyncio를 활용한 비동기 처리에서 뛰어난 성능을 보여주기 때문에, gRPC와 결합하면 높은 성능의 비동기 서버를 구축할 수 있습니다. FastAPI의 async 및 await 키워드를 활용하면 I/O 작업이 완료될 때까지 다른 요청을 처리할 수 있어, 서버가 더욱 효율적으로 동작합니다. 또한 gRPC는 HTTP/2 기반의 비동기 프로토콜을 지원하므로, FastAPI와의 통합 시 동시성을 극대화된다는 점도 기술 검토 과정에 있어 중요한 결정 요인이 되었습니다.

로직 구현

고수 서비스는 이미 gRPC 통신을 기반으로 구현된 로직이 있었기 때문에 필요한 IDL(Interface definition language)을 추가한 후 비즈니스 로직만 구현하면 되었습니다. 반면에, 커뮤니티 서비스는 gRPC를 처음 도입하는 상황이었기 때문에 gRPC 환경 설정부터 시작하여 구현을 진행하였습니다.

1. gRPC 라이브러리 설치

gRPC는 .proto 파일을 기반으로 클라이언트와 서버 간의 통신이 필요한 코드를 자동으로 생성합니다.

이를 위해 grpcio-tools라는 패키지를 활용하여 컴파일을 진행하게 됩니다.

커뮤니티 서비스는 poetry로 패키지 관리를 하고 있어 poetry를 활용해 라이브러리를 설치합니다.

poetry add grpcio grpcio-tools

2. .proto 파일 작성

gRPC 구현을 위해 첫 단계는 protocol buffers 정의입니다.

protocol buffers는 구글이 개발한 직렬화 포맷으로 gRPC는 이 포맷을 활용해 데이터를 주고받습니다.

이때 클라이언트와 서버가 주고받을 메시시지 형식 및 서비스 정의 등을 .proto 파일에 작성해야 합니다.

이번 일감에서는 리뷰 데이터들의 많은 양 그리고 커뮤니티 서비스가 받아야 할 데이터들의 스키마 차이로 인한 인터페이스 정의가 복잡해졌습니다.

촉박한 데드라인으로 인하여 효율적 방법을 고민한 끝에 클라이언트는 protobuf를 정의할 때 dict 객체를 문자열로 변환하여 전송하고 서버 측에서는 JSON으로 변환 후 유효성 검증을 수행하는 방식으로 구현하기로 했습니다.

syntax = "proto3";
package Review.v1;
import "google/protobuf/struct.proto";
service Review {
  rpc UpdateReviewPost (UpdateReviewRequest) returns (BaseResponse) {}
  rpc DeleteReviewPost (DeleteReviewRequest) returns (BaseResponse) {}
  ...
}
message UpdateReviewPost {
  string data = 1;
}

message DeleteReviewPost {
  string data = 1;
}

이 방식을 선택한 이유 중 하나는 커뮤니티 서비스의 기존 게시글 등록, 수정, 삭제 등 서비스 레이어가 잘 모듈화되어 있었기 때문입니다. 이에 따라, 기존 로직을 재사용하면서 FastAPI 프레임워크의 Pydantic을 이용한 유효성 검증 및 데이터 관리가 용이했습니다.

3. gRPC 코드 생성

.proto 파일을 정의한 후, 이를 기반으로 애플리케이션에서 사용할 gRPC 코드를 생성해야 합니다. Python에서는 grpc_tools.protoc 명령어를 사용해 필요한 코드들을 자동으로 생성할 수 있습니다.

python -m grpc_tools.protoc -I.{{path}} --python_out=. --grpc_python_out=. {{path}}/review.proto

위 명령을 실행하면 아래와 같이 gRPC 통신에 필요한 메세지 정의 클래스 및 인터페이스 파일이 생성되는걸 확인할 수 있습니다.

.
├── __init__.py
├── review.proto # 프로토콜 정의
├── review_pb2.py # 메세지 클래스
└── review_pb2_grpc.py # gRPC 서비스 Stub

4. gRPC server 설정

gRPC 서버는 FastAPI 기반의 비동기 서버로 설정하였습니다. 비동기 서버의 장점 중 하나는 스레드 풀을 관리하여 여러 클라이언트의 요청을 동시에 처리할 수 있다는 점입니다. 예를 들어, 하나의 요청이 지연되더라도 다른 요청에 영향을 주지 않고 처리할 수 있습니다.

gRPC는 grpc.aio.server를 제공하며, Python의 asyncio 모듈과 호환됩니다. 이를 통해 효율적인 I/O 작업 처리가 가능하여 비동기 서버로 구현하는 데 적합했습니다.

@app.command(help="GRPC Async Server Execute")
@async_run
@init_middlewares
async def serve():
    server = grpc.aio.server(futures.ThreadPoolExecutor(max_workers=grpc_workers))
    add_ReviewServicer_to_server(ReviewServicerImpl(), server)
    server.add_insecure_port(f'[::]:{grpc_port}')
    await server.start()
    typer.echo(f'Async gRPC Started...\n' f'Workers: {grpc_workers}\n' f'Port:{grpc_port}')
    await server.wait_for_termination()

위 코드에서는 async_run(이벤트 루프 비동기 코루틴)과 init_middlewares(버그스낵, 뉴렐릭 설정)를 데코레이터로 구현 후 사용하여 버그 추적과 서버 실행을 위한 로직을 재사용할 수 있도록 구현하였습니다.

5. Pydantic을 활용한 유효성 검사

서버 사이드에서 요청받은 데이터는 Pydantic을 사용하여 유효성 검증을 거칩니다. 모델의 각 필드에 대해 유효성 검사를 정의하고, 특정 필드에 대한 추가적인 검증이 필요할 때는 @field_validator 및 @model_validator를 사용하여 검사 로직을 추가할 수 있습니다.

class GrpcUpdateReviewPostRequest(UpdatePostContentBlockValidatorMixin):
    author_user_id: Annotated[int, Field(title='작성자 유저 ID')]
    review_id: Annotated[int, Field(title='리뷰 ID')]
    title: Annotated[str, Field(title='제목')] = ''
    rating: Annotated[float, Field(title='평점')]
    surveys: Annotated[list[str], Field(title='설문 정보')]
    @field_validator('author_user_id')
    @classmethod
    def author_user_id_validator(cls, v: int) -> int:
        if v < 1:
            raise ValueError('유저 ID가 유효하지 않습니다.')
        return v
    @field_validator('review_id')
    @classmethod
    def review_id_validator(cls, v: int) -> int:
        if v < 1:
            raise ValueError('리뷰 ID가 유효하지 않습니다.')
        return v

이때 유효성 검사를 실패하게 되면 ValidationError가 발생하도록 구현하였습니다.

6. 비지니스 로직 구현

gRPC 코드로 생성된 gRPC 서비스 Stub의 ReviewServicer 인터페이스를 활용하고 해당 클래스내에 필요한 메소드를 선언하여 구현합니다. 구현시 커뮤니티 서비스는 facade 패턴으로 서비스 로직을 분리하고 있어 이에 맞게 클래스를 구현하였습니다.

class ReviewServicerImpl(ReviewServicer, metaclass=ServiceDecoratorMeta):
    post_facade: PostFacadeV1 = inject.attr(PostFacadeV1)
    async def UpdateReviewPost(self, request, context: grpc.aio.ServicerContext):
        try:
            request_dict = MessageToDict(request, preserving_proto_field_name=True)
            request = GrpcUpdateReviewPostRequest.model_validate_json(request_dict['data'])
            review_post = await self.post_facade.grpc_update_review_post(request)
            return self._make_response(GrpcResponseCode.OK, review_post)
        except (KeyError, ValueError) as e:
            bugsnag.notify(e, context=f"update review post request: {request}")
            return self._make_response(GrpcResponseCode.INVALID_PARAM, None)
        except Exception as e:
            bugsnag.notify(e, context=f"update review post request: {request}")
            return self._make_response(GrpcResponseCode.INTERNAL_SERVER_ERROR, None)

위 로직에서 Pydantic의 model_validate_json을 사용하여 요청 데이터의 유효성을 검증하고, JSON으로 파싱하여 모델의 필드와 일치시키는 작업을 수행했습니다. post_facade는 서비스 로직을 처리하는 역할을 하며, 비즈니스 로직의 구현은 이 부분에서 처리됩니다.

또한, 서비스를 안정적으로 운영하기 위해서는 모니터링과 로깅이 필수적이므로, 기존 챕터에서 사용하던 gRPC 프로세스 로깅 함수를 데코레이터로 구현하고 비동기 환경에서도 사용할 수 있도록 적용했습니다.

UpdateReviewPost처럼 메소드가 추가될 때마다 중복된 데코레이터를 적용해야하는 불편함을 피하고자 메타클래스를 활용하여 클래스가 생성될때 사전에 작업해둔 로깅 및 모니터링 데코레이터 함수를 자동으로, 메소드별로 적용되도록 하는 방식을 채택하였습니다.

7. 테스트 코드

프로덕션에서의 안정적인 서비스 운영을 위해, pytest를 활용하여 비동기 테스트 코드를 작성했습니다. 테스트는 gRPC Stub을 사용하여 실제 gRPC 요청과 응답을 대신 처리하며, pytest의 fixture를 통해 모의 데이터를 생성하고 필요한 의존성을 주입하여 테스트를 진행했습니다.

class TestGrpcUpdateReviewPost:
    REVIEW_ID = 1
    pos_service: PostServiceV3 = inject.attr(PostServiceV3)
    @pytest.fixture(scope='function')
    def mock_request_data(self, faker_ko: Faker, provider_user: User):
        text = faker_ko.text()
        return GrpcUpdateReviewPostRequest(
            user_id=1,
            title=review_text,
            service_id=1,
            content_blocks=[...],
            review_info=GrpcRequestReviewInfoData(review_id=self.REVIEW_ID),
        )
    async def test_update_review_post(
        self,
        grpc_stub: ReviewStub,
        mock_request_data: GrpcUpdateReviewPostRequest,
    ):
        request = review_pb2.UpdateReviewRequest(data=mock_request_data.model_dump_json())  # type: ignore
        response = await grpc_stub.UpdateReviewPost(request)
        assert isinstance(response, review_pb2.BaseResponse)  # type: ignore
        with does_not_raise():
            post = await self.pos_service.get_review_post(
                user_id=mock_request_data.user_id, review_id=self.REVIEW_ID
            )
            assert post.review is not None
            assert post.review.id == mock_request_data.review.id

위 테스트에서는 mock_request_data라는 fixture를 사용하여 요청 데이터를 동적으로 생성하고, 테스트 환경에서 일관되게 사용할 수 있도록 설정하였습니다. 특히, 실제 데이터 모델에 기반한 요청 객체를 사용하여 신뢰성을 높였습니다.

테스트의 핵심은 응답 데이터 검증으로 응답이 기대하는 메시지 클래스의 형태로 반환되는지 확인과 더불어 생성된 리뷰 ID와 요청한 리뷰 ID가 일치하는지 비교하여 데이터 일관성까지 검증하였습니다.

8. 작업 후 모니터링

서비스를 배포한 후 안정적인 운영을 위해 모니터링은 필수적인 단계입니다. 숨고는 커뮤니티를 포함해 모든 서비스를 뉴렐릭(New Relic)을 활용하여 서비스의 응답 속도를 확인하고, 잠재적인 병목 현상이나 성능 저하를 조기에 감지할 수 있습니다.

CUD 요청의 응답 시간을 뉴렐릭 대시보드에서 확인한 결과입니다. gRPC 이벤트마다 다르지만 response time이 평균 70ms로 유지하고 있으며, 이를 통해 서비스가 안정적으로 관리되고 있음을 확인할 수 있었습니다.

마무리하며

이번 글에서는 FastAPI에 gRPC를 활용하여 비동기 서버를 구현하고 비지니스 로직을 적용했던 경험을 소개해 드렸습니다.

gRPC에 관한 자료는 많지만, Python을 이용한 실제 서비스 구현에 대한 정보는 상대적으로 부족하기에 이 부분은 팀 동료와 함께 다양한 레퍼런스를 참고하며 경험을 통해 문제를 해결해 나갈 수 있었습니다.

또한 작업을 마무리한 뒤에 모니터링을 통해 서비스 운영 시 안정적인 API 응답 속도를 유지할 수 있음을 확인하였습니다. 이는 gRPC 작업을 통해 이룬 API 성능 최적화와 안정적인 서비스를 운영하는 데 긍정적인 결과였습니다.

다만, 작업을 진행하며 아쉬운 점은 스키마의 타입 안정성을 확보하는 부분에서 미흡한 점이 있었다는 것입니다. 이 문제를 방지하기 위해 ProtoBuf를 더 철저히 활용하여 데이터 타입과 스키마를 명확히 정의해야 한다는 중요성을 깨달았습니다. 추후 이러한 부분을 보완하여 안정적인 서비스를 제공할 수 있도록 준비해 나가고자 합니다.

Python을 활용하여 gRPC를 도입하려는 분들에게 조금이나마 도움이 되었기를 바랍니다.