Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Summary of Changes
Hello @youyeon11, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!
이 Pull Request는 S3 이미지 처리의 효율성을 높이고 Redis Stream 기반의 비동기 작업 처리 시스템을 더욱 견고하게 만드는 데 중점을 둡니다. 이미지 데이터를 파일 시스템에 저장하는 대신 메모리 스트림으로 직접 처리하여 성능을 개선했으며, Redis Stream 메시지 처리에 대한 오류 처리 및 완료 로직을 강화했습니다. 또한, 블루/그린 배포를 위한 인프라 설정을 추가하여 서비스의 안정적인 배포 및 운영을 지원합니다.
Highlights
- S3 이미지 처리 로직 개선: S3에서 이미지를 다운로드할 때, 기존에는 임시 파일로 저장하는 방식이었으나, 이제는
BytesIO스트림을 사용하여 메모리 내에서 직접 처리하도록 변경하여 디스크 I/O를 줄이고 효율성을 높였습니다. - Redis Stream 비동기 처리 전환 및 기능 강화: Redis Stream 작업을 비동기 기반으로 전면 전환하고,
RedisStreamClient클래스를 도입하여xadd,xreadgroup,xack,xack_and_del,xclaim등 Redis Stream 관련 기능을 통합 관리하도록 구현했습니다. 특히, 메시지 처리 완료 후XACK와XDEL을 통해 스트림에서 메시지를 제거하고, 처리 실패 시 Dead Letter Queue(DLQ)로 메시지를 전송하는 로직을 추가하여 견고성을 확보했습니다. - 배포 자동화 및 블루/그린 배포 환경 구축: 블루/그린 배포 전략을 지원하는
deploy.sh스크립트와docker-compose.yml,nginx.conf파일을 추가하여 무중단 배포 환경을 구축했습니다. 이를 통해 서비스 업데이트 시 다운타임 없이 안정적인 전환이 가능해졌습니다. - S3 서비스 및 API 엔드포인트 테스트 코드 추가: S3 이미지 다운로드 서비스에 대한 단위 테스트를 추가하여 핵심 로직의 신뢰성을 확보하고, 헬스 체크 API 엔드포인트에 대한 테스트도 추가하여 서비스의 기본 동작을 검증합니다.
Using Gemini Code Assist
The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.
Invoking Gemini
You can request assistance from Gemini at any point in your pull request via creating an issue comment (i.e. comment on the pull request page) using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands.
| Feature | Command | Description |
|---|---|---|
| Code Review | /gemini review |
Performs a code review for the current pull request in its current state. |
| Pull Request Summary | /gemini summary |
Provides a summary of the current pull request in its current state. |
| Comment | @gemini-code-assist | Responds in comments when explicitly tagged, both in issue comments and review comments. |
| Help | /gemini help |
Displays a list of available commands. |
Customization
To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.
Limitations & Feedback
Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here.
You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension.
Footnotes
-
Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩
🤖 Gemini 코드 리뷰 결과1. 주요 변경 사항 요약 및 의도 파악이 PR은 S3 이미지 처리 로직의 개선과 Redis 스트림 처리 방식을 비동기로 전환하는 것을 목표로 합니다. S3에서 이미지를 다운로드하여 로컬에 저장하는 대신, 바이트 스트림으로 직접 처리하여 I/O 오버헤드를 줄이고 예측 작업을 수행합니다. Redis 스트림 작업을 비동기로 전환하여 동시성을 높이고 처리량을 개선하려는 의도로 보입니다. 추가적으로 S3 관련 설정을 정리하고, Redis 메시지 처리 후 삭제 로직을 구현하여 스트림의 크기를 관리하고자 합니다. 2. 코드 품질 및 가독성
3. 잠재적 버그 및 엣지 케이스
4. 성능 및 효율성
5. 보안 및 아키텍처
|
There was a problem hiding this comment.
Code Review
이번 PR은 S3 이미지 다운로드와 Redis 스트림 처리를 비동기 방식으로 전환하여 성능을 개선하려는 좋은 시도입니다. 하지만 비동기 코드 구현과 관련하여 몇 가지 심각한 문제점들이 발견되었습니다. asyncio.to_thread의 잘못된 사용, 코루틴에 대한 await 누락, 함수 호출 시의 타입 불일치 등이 있으며, 이는 런타임 오류를 유발할 가능성이 높습니다. 또한, 중복 코드 리팩토링, 적절한 로깅 프레임워크 사용, Redis 작업의 원자성 보장 등을 통해 코드 품질을 개선할 여지가 있습니다. 추가된 테스트 코드 또한 올바르게 동작하기 위해 수정이 필요합니다.
app/worker/worker.py
Outdated
| streams={settings.STREAM_JOB: ">"}, | ||
| group_name=settings.GROUP_NAME, | ||
| consumer_name=settings.CONSUMER_NAME, | ||
| stream_name={settings.STREAM_JOB: ">"}, |
There was a problem hiding this comment.
xreadgroup 메서드는 stream_name 인자로 스트림의 이름(문자열)을 기대하지만, 딕셔너리({settings.STREAM_JOB: ">"})가 전달되고 있습니다. xreadgroup 내부에서 id를 처리하므로 스트림 이름만 전달해야 합니다.
| stream_name={settings.STREAM_JOB: ">"}, | |
| stream_name=settings.STREAM_JOB, |
| task.add_done_callback(lambda t: asyncio.create_task( | ||
| self.redis_client.xack_and_del(settings.STREAM_JOB, settings.GROUP_NAME, msg_id) | ||
| if not t.exception() else | ||
| self.redis_client.xadd(f"{settings.STREAM_JOB}:DLQ", | ||
| {"id": msg_id, "error": str(t.exception()), **fields}) | ||
| )) |
There was a problem hiding this comment.
self.redis_client.xack_and_del 메서드는 message_ids 인자로 문자열 리스트(List[str])를 기대하지만, 단일 문자열 msg_id가 전달되고 있습니다. [msg_id]와 같이 리스트로 감싸서 전달해야 합니다.
| task.add_done_callback(lambda t: asyncio.create_task( | |
| self.redis_client.xack_and_del(settings.STREAM_JOB, settings.GROUP_NAME, msg_id) | |
| if not t.exception() else | |
| self.redis_client.xadd(f"{settings.STREAM_JOB}:DLQ", | |
| {"id": msg_id, "error": str(t.exception()), **fields}) | |
| )) | |
| task.add_done_callback(lambda t: asyncio.create_task( | |
| self.redis_client.xack_and_del(settings.STREAM_JOB, settings.GROUP_NAME, [msg_id]) | |
| if not t.exception() else | |
| self.redis_client.xadd(f"{settings.STREAM_JOB}:DLQ", | |
| {"id": msg_id, "error": str(t.exception()), **fields}) | |
| )) |
app/services/predictor_service.py
Outdated
|
|
||
| def predict(self, image_path: Path) -> tuple[str, str, float]: | ||
| image = Image.open(image_path).convert('RGB') | ||
| async def predict(self, stream_file: BytesIO) -> tuple[str, str, float]: |
There was a problem hiding this comment.
predict 메서드는 내부적으로 동기적인(blocking) 코드(Pillow, PyTorch)를 포함하고 있으므로 async def로 선언해서는 안 됩니다. app.worker.tasks에서 asyncio.to_thread를 사용하여 이 함수를 호출하려고 하는데, async def 함수는 to_thread로 실행할 수 없어 TypeError가 발생합니다. 이 메서드를 일반 동기 함수로 변경해야 합니다.
| async def predict(self, stream_file: BytesIO) -> tuple[str, str, float]: | |
| def predict(self, stream_file: BytesIO) -> tuple[str, str, float]: |
There was a problem hiding this comment.
anyio.to_thread와 async는 실행 방식에서 근본적인 차이가 있습니다. async 키워드는 비동기 함수를 선언하는 문법(코드 스타일)이고, anyio.to_thread는 이러한 비동기 함수 내에서 CPU 바운드 작업을 별도의 스레드로 옮겨 실행하여 메인 스레드의 블로킹을 방지하는 함수입니다. 즉, async는 비동기 코드의 작성 방식이고, anyio.to_thread는 비동기 코드에서 스레드를 활용해 동시성을 구현하는 기능입니다.
- anyio.to_thread 의 블로킹 작업 실행: anyio.to_thread는 CPU 집약적인(블로킹) 함수를 별도의 스레드 풀에 할당하여 실행합니다.
- 이벤트 루프 보호: async 함수는 기본적으로 이벤트 루프를 사용하는데, CPU를 많이 사용하는 작업은 이벤트 루프를 멈추게 합니다. anyio.to_thread는 이러한 블로킹 작업을 별도의 스레드에서 처리함으로써 메인 이벤트 루프가 다른 비동기 작업들을 계속 처리할 수 있도록 합니다.
- 동시성 구현: async 코드가 I/O 바운드 작업에 효율적인 반면, anyio.to_thread는 비동기 코드에서 스레드를 활용하여 CPU 바운드 작업과의 동시성을 효과적으로 관리하는 도구입니다.
| task.add_done_callback(lambda t: asyncio.create_task( | ||
| self.redis_client.xack_and_del(settings.STREAM_JOB, settings.GROUP_NAME, msg_id) | ||
| if not t.exception() else | ||
| self.redis_client.xadd(f"{settings.STREAM_JOB}:DLQ", | ||
| {"id": msg_id, "error": str(t.exception()), **fields}) | ||
| )) |
There was a problem hiding this comment.
여기에서도 self.redis_client.xack_and_del 메서드에 msg_id를 리스트([msg_id])로 전달해야 합니다. 또한, 28-48행의 로직과 이 부분이 거의 동일하므로, 중복을 제거하기 위해 별도의 헬퍼 메서드로 추출하는 것을 고려해보세요.
| task.add_done_callback(lambda t: asyncio.create_task( | |
| self.redis_client.xack_and_del(settings.STREAM_JOB, settings.GROUP_NAME, msg_id) | |
| if not t.exception() else | |
| self.redis_client.xadd(f"{settings.STREAM_JOB}:DLQ", | |
| {"id": msg_id, "error": str(t.exception()), **fields}) | |
| )) | |
| task.add_done_callback(lambda t: asyncio.create_task( | |
| self.redis_client.xack_and_del(settings.STREAM_JOB, settings.GROUP_NAME, [msg_id]) | |
| if not t.exception() else | |
| self.redis_client.xadd(f"{settings.STREAM_JOB}:DLQ", | |
| {"id": msg_id, "error": str(t.exception()), **fields}) | |
| )) |
|
|
||
| try: | ||
| s3_service.download_file_from_presigned_url(job.presignedUrl, temp_image_path) | ||
| stream_file = await asyncio.to_thread(s3_service.download_file_from_presigned_url(job.presignedUrl)) |
There was a problem hiding this comment.
asyncio.to_thread의 사용법이 잘못되었습니다. to_thread는 첫 번째 인자로 실행할 동기 함수를 받고, 나머지 인자들은 그 함수에 전달될 인자들입니다. 현재 코드는 s3_service.download_file_from_presigned_url 함수를 먼저 호출하여 이벤트 루프를 블로킹하고, 그 반환 값을 to_thread로 넘기고 있습니다.
| stream_file = await asyncio.to_thread(s3_service.download_file_from_presigned_url(job.presignedUrl)) | |
| stream_file = await asyncio.to_thread(s3_service.download_file_from_presigned_url, job.presignedUrl) |
app/worker/redis_client.py
Outdated
| async def xack_and_del(self, stream_name: str, group_name: str, message_ids: List[str]) -> int: | ||
|
|
||
| acked_count = await self.redis_client.xack(stream_name, group_name, *message_ids) | ||
|
|
||
| # XACK가 성공하면 스트림에서 해당 메시지를 삭제 (XDEL) | ||
| if acked_count > 0: | ||
| await self.redis_client.xdel(stream_name, *message_ids) | ||
|
|
||
| return acked_count |
There was a problem hiding this comment.
XACK와 XDEL은 별개의 명령으로 실행됩니다. 두 명령이 원자적으로 실행되도록 보장하려면 Redis의 pipeline을 사용하는 것이 좋습니다. 이렇게 하면 XACK가 성공한 후에 XDEL이 실패하는 경우를 방지할 수 있습니다.
|
|
||
| class JobWorker: | ||
| def __init__(self, redis_client: redis.Redis): | ||
| def __init__(self, redis_client: redis_client): |
There was a problem hiding this comment.
redis_client의 타입 힌트가 redis_client 인스턴스 자체로 되어 있습니다. RedisStreamClient 클래스로 지정해야 올바른 타입 힌트가 됩니다. 파일 상단에 from app.worker.redis_client import RedisStreamClient를 추가하고 타입 힌트를 수정해주세요.
| def __init__(self, redis_client: redis_client): | |
| def __init__(self, redis_client: "RedisStreamClient"): |
| async def lifespan(app: FastAPI): | ||
| redis_client = redis.from_url(settings.REDIS_URL, decode_responses=True) | ||
| limiter = anyio.to_thread.current_default_thread_limiter() | ||
| limiter.total_tokens = 100 |
There was a problem hiding this comment.
스레드 제한 수(100)가 하드코딩되어 있습니다. 이러한 값은 애플리케이션 설정(app/core/config.py)으로 옮겨 관리하는 것이 유지보수 측면에서 더 좋습니다.
| finally: | ||
| if temp_image_path.exists(): | ||
| temp_image_path.unlink() | ||
| print(f"[task] Image scan finished for job_id={correlation_id}") |
There was a problem hiding this comment.
이 로그 메시지는 36행에 이미 존재하여 중복됩니다. finally 블록은 항상 실행되므로 36행의 로그를 제거하고 이 로그만 남기는 것이 더 깔끔합니다. 또한 print 대신 logging 모듈 사용을 고려해보세요.
| def test_download_file_http_error(self, mock_get): | ||
| # arrange | ||
| mock_get.return_value = MagicMock() | ||
| mock_get.return_value.raise_for_status.side_effect = Exception("HTTP Error") |
There was a problem hiding this comment.
HTTP 오류를 테스트할 때 requests.exceptions.HTTPError와 같이 더 구체적인 예외 타입을 확인하는 것이 좋습니다. Exception은 너무 광범위하여 의도치 않은 다른 예외 상황에서도 테스트가 통과될 수 있습니다.
1 participant
📌 작업 목적
🗂 작업 유형
🔨 주요 작업 내용
**
s3_service.py**predict가 사용하게 됨lifespan.py에서 설정한 비동기 컨텍스트 설정📎 관련 이슈
💬 논의 및 고민한 점