최저가 항공 요금 수집을 위한 데이터 파이프라인 개발기

aws sqs send-message \
--queue-url https://sqs.ap-northeast-1.amazonaws.com/xxxxxxx/dev-fare-crawler \
--message-body '{ "outputFolder": "2024-01-20"}'

apiVersion: argoproj.io/v1alpha1
kind: EventSource
metadata:
  name: fare-crawler-aws-sqs
  namespace: argo-events
spec:
  sqs:
    sqs-message-received:
      jsonBody: true
      # 여기서 접근하는 시크릿의 경우 argo-events가 설치된 argo-events namespace에 존재해야 읽어올 수 있음
      accessKey:
        name: aws-secret
        key: accesskey
      secretKey:
        name: aws-secret
        key: secretkey
      region: ap-northeast-1
      queue: dev-fare-crawler
      waitTimeSeconds: 20

apiVersion: argoproj.io/v1alpha1
kind: Sensor
metadata:
  name: fare-crawler-aws-sqs
  namespace: argo-events
spec:
  template:
    serviceAccountName: operate-workflow-sa

  dependencies:
    - name: fare-crawler-dep
      eventSourceName: fare-crawler-aws-sqs
      eventName: sqs-message-received

  triggers:
    - template:
        name: trigger-fare-crawler
        k8s:
          group: argoproj.io
          version: v1alpha1
          resource: workflows
          operation: create

          parameters:
            # SQS를 통해 전될된 메시지(Input.body) 내용 중 outputFolder라는 필드 값을 읽어와서 다양한 작업을 수행합니다. 
            # (해당 필드값이 존재하지 않을 경우 빈 스트링을 기본값으로 사용합니다)
            - src:
                dependencyName: fare-crawler-dep
                dataTemplate: ''
              # outputFolder 값을 아래쪽에 인라인 정의된 Workflow의 첫번째 인자로 넣어줍니다.
              dest: spec.arguments.parameters.0.value

            - src:
                dependencyName: fare-crawler-dep
                dataTemplate: '-'
              # outputFolder 값을 아래쪽에 인라인 정의된 Workflow의 metadata.generateName 필드 값에 append 합니다.
              dest: metadata.generateName
              operation: append

          source:
            resource:
              apiVersion: argoproj.io/v1alpha1
              kind: Workflow
              metadata:
                # 쿠버네티스에 생성되는 Pod의 이름을 결정합니다. 
                # 결과적으로 fare-crawler-- 형태로 생성됩니다.
                generateName: fare-crawler-
                # Workflow가 수행될 namespace를 명시합니다. 이부분은 자신이 사용할 쿠버네티스 클러스터의 상황에 맞게 적절히 설정해주세요.
                namespace: dev

              spec:
                imagePullSecrets:
                  - name: squarelabkuber-cred
                # Argo Workflow 설치과정에서 생성한 서비스 어카운트를 넣어주시면 됩니다.
                serviceAccountName: workflow
                entrypoint: main
                arguments:
                  parameters:
                    - name: outputFolder
                    - name: fareUpdaterQueueName
                      value: dev-fare-updater
                activeDeadlineSeconds: 72000 # 20h
                ttlStrategy:
                  # 수행되는 Workflow들의 TTL이 제대로 설정되지 않는 경우 해당 Workflow의 기록이 계속해서 쿠버네티스상에 그대로 남게됩니다.
                  # 이 경우 쿠버네티스 Control Plane이 느려지거나 메모리 부족 현상이 생길 수 있으니 적절한 TTL 설정이 필수입니다.
                  secondsAfterSuccess: 3600 # 성공한 Workflow를 3600초간 유지
                  secondsAfterFailure: 86400 # 실패한 Workflow를 86400초간 유지

                templates:
                  # Workflow 정의
                  - name: main
                    steps:
                      # 2개로 샤딩하여 병렬로 항공권 운임 데이터를 수집합니다.
                      - - name: fare-crawler-1
                          # 아래쪽에 정의된 템플릿 기반으로 작업 단계를 구성합니다.
                          template: launch-fare-crawler
                          arguments:
                            parameters:
                              # 여기 정의된 값을 아래쪽 컨테이너이너 설정에서  형태로 읽어서 사용합니다
                              - name: message
                                value: '{"shardingOptions":{"shards":2,"shardIndex":1,"outputFolder":""}}'
                        - name: fare-crawler-2
                          template: launch-fare-crawler
                          arguments:
                            parameters:
                              - name: message
                                value: '{"shardingOptions":{"shards":2,"shardIndex":2,"outputFolder":""}}'

                      # 위의 fare_crawler 수행이 모두 성공적으로 종료되면 SQS에 메시지를 보내서 fare_updater를 트리거 합니다.
                      - - name: trigger-fare-updater
                          template: launch-aws-cli
                          arguments:
                            parameters:
                              - name: queueName
                                value: ''
                              - name: message
                                value: '{"targetFolder":""}'

                  # Workflow 정의에서 참조해서 사용하는 템플릿
                  - name: launch-fare-crawler
                    metadata:
                      labels:
                        app: fare-crawler
                    inputs:
                      parameters:
                        - name: message
                    nodeSelector:
                      instanceType: general
                    container:
                      image: kyte/fare_crawler:${DOCKER_TAG}
                      imagePullPolicy: Always
                      args: ['']

                  - name: launch-aws-cli
                    inputs:
                      parameters:
                        - name: queueName
                        - name: message
                    nodeSelector:
                      instanceType: general
                    container:
                      image: amazon/aws-cli:2.7.25
                      args:
                        - 'sqs'
                        - 'send-message'
                        - '--queue-url'
                        - 'https://sqs.ap-northeast-1.amazonaws.com/xxxxxxxxxxxxxx/'
                        - '--message-body'
                        - ''
                      env:
                        - name: AWS_ACCESS_KEY_ID
                          value: xxxxxxx
                        - name: AWS_SECRET_ACCESS_KEY
                          value: xxxxxxx

                # Workflow의 Pod들이 동일한 쿠버네티스 노드에 몰려서 수행되지 않도록 설정합니다.
                affinity:
                  podAntiAffinity:
                    preferredDuringSchedulingIgnoredDuringExecution:
                      - weight: 100
                        podAffinityTerm:
                          labelSelector:
                            matchExpressions:
                              - key: app
                                operator: In
                                values:
                                  - fare-crawler
                          topologyKey: kubernetes.io/hostname