pub/sub 로직 개선하기

pub/sub 로직 개선했던 서비스에서 간단하게 배경 설명을 한다면,,

 

수강신청이라는 서비스가 있다.

 

b2e 화면에서 수강생들이 수강신청을 진행을 하면 backoffice 에서 관리자가 해당 상품에 대해서 수강신청 확정 버튼을 클릭해 수강생들이 신청한 강의들을 확정처리하여 강의를 들을 수 있게 하는 서비스가 있다.

 

이 서비스는 처음에 batch 로 작업이 진행이 되었다.

batch 로 1시간에 한번씩 작업이 진행이 되었다.

 

하지만 이 서비스의 확정 처리 기능은 빈번하게 발생하는 작업이 아니였고 비정기적으로 진행이 되는 작업이였다.

비정기적으로 작업이 되는 서비스에 batch 는 과한 스펙이 되었다.

 

그래서 이 확정처리 기능은 batch 에서 pub/sub 으로 변경이 되었다.

pub/sub 은 이렇게 구현을 하였다.

 

pub/sub 을 통해서 수강신청 확정 버튼을 클릭할때마다 확정 처리 기능이 진행이 되었다.

근데 여기에도 비효율적인 로직이 있었다.

pub/sub 에서 Run 으로 message 로 확정 처리 대상이 되는 id 값을 던지지만 run 에서 run job 을 호출할때 해당 메시지를 전달하지 않고 있다라는 점이다.

run job 에서는 전체 수강신청 상품 대상으로 조회한뒤 해당 되는 상품을 찾아 확정 처리 작업을 진행하고 있었다.

run 에서 run job 을 호출할때 pub/sub 에서 받은 id 를 그대로 전달해주면 되지 않을까??? 라는 의문이 들었다.

 

그래서 run 의 소스 코드를 한번 확인을 해보았다.

        # Pub/Sub 메시지 추출
        envelope = request.get_json()
        if not envelope or 'message' not in envelope:
            return "Invalid message", 200

        message = envelope['message']
        data = ''
        if 'data' in message:
            data = base64.b64decode(message['data']).decode('utf-8')

        print(f"Received Pub/Sub message: {data}")

        # Cloud Run Job 실행
        client = run_v2.JobsClient()
        job_path = client.job_path(project=project_id, location=location, job=job_name)
        run_request = run_v2.RunJobRequest(name=job_path)

        print(f"Triggering Cloud Run Job: {job_path}")
        operation = client.run_job(request=run_request)

 

message 만 추출하고 바로 run-job 을 호출하는 로직이였다.

 

여기서 message 를 추출한 다음 run-job 호출할때 해당 값을 넣어 주면 되지 않을까? 라는 생각이 들었다.

 

어떤 방법이 있는지 GPT 에게 물어봤다. GPT 는 환경변수 방식, 커맨드 인자 방식 두가지 방법을 알려주었다.

 

기준	환경변수 사용	커맨드 인자 사용
간단한 설정값 전달	✅ 권장	가능하지만 과함
보안 정보 (e.g. API 키)	❌ 비추천 (Secret Manager 권장)	❌ 비추천
실행 시간마다 바뀌는 값 (동적 파라미터)	✅ 좋음	✅ 좋음
스크립트나 CLI 스타일에 익숙함	❌ 적합하지 않음	✅ 적합
가독성 / 로깅이 중요한 경우	덜 직관적 (env)	✅ 명확하게 로그로 노출
컨테이너 재사용 가능성	✅ 환경에 따라 분기 쉬움	✅ 인자에 따라 다양화 가능

 

 

💡권장 방식 -> 일반적으로는 환경변수 방식이 더 유연하고 안전

1. Cloud Run Job의 실행 오버헤드가 적고 설정이 간단합니다
2. Dockerfile이나 ENTRYPOINT 수정 없이도 작동 가능
3. 컨테이너 내부에서 표준 환경변수 접근 (process.env, os.environ) 방식으로 활용 가능
4. Cloud Console이나 gcloud UI에서도 시각적으로 관리하기 쉬움

 

이렇게 두가지 방식에 대해 특징과 권장방식에 대해서 확인을 해보고 실 사용하는 서비스를 각각의 특징에 대입해서 봤을때 환경변수 사용하는 것이 더 적합한 방식이라고 판단이 되어 환경변수 사용하는 방식을 선택한 뒤 run 의 소스 코드를 수정하였다.

 

        # 3. 메시지에서 필요한 값 추출
        try:
            payload = json.loads(data)
        except json.JSONDecodeError:
            logger.warning("Message is not valid JSON")
            payload = {}

        registration_product_id = str(payload.get("registrationProductId", '')).strip() or None
        product_id = str(payload.get("productId", '')).strip() or None

        # 4. Cloud Run Job 실행을 위한 클라이언트 및 job path
        client = run_v2.JobsClient()
        job_path = client.job_path(project=project_id, location=location, job=job_name)

        # 5. 환경변수 오버라이드 설정
        env_vars = []
        if registration_product_id: # message 로 받은 key 값
            env_vars.append(run_v2.EnvVar(name="REGISTRATION_PRODUCT_ID", value=registration_product_id))
        if product_id: # message 로 받은 key 값
            env_vars.append(run_v2.EnvVar(name="PRODUCT_ID", value=product_id))

        overrides = None
        if env_vars:
            container_override = run_v2.RunJobRequest.Overrides.ContainerOverride(env=env_vars)
            overrides = run_v2.RunJobRequest.Overrides(container_overrides=[container_override])

        # 6. Job 실행 요청 생성
        run_request = run_v2.RunJobRequest(
            name=job_path,
            overrides=overrides
        )

 

message 에서 필요한 값을 추출한 뒤 해당 값을 환경변수 오버라이드로 설정을 하여 job 실행시에 해당 환경변수를 넣어 주었다.

 

이렇게 run 소스 코드를 수정하고 run-job 을 실행하여 환경 변수가 잘 들어오는지 테스트를 해보았다.

 

 

run 에서는 로그가 잘 나오고 있어 run-job 에서 확인해보았다.

 

 

input 을 넣어 로그를 확인해보니 잘 노출 되고 있어 환경 변수가 run-job 에도 잘 전달 되고 있는 것을 확인했다.

 

 

이렇게 환경변수를 활용하여 run 에서 run-job 을 호출할때 해당 값을 전달 할 수 있었다!!