Topic

topic 목록 보기

./kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --list 

topic 생성

./kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --create \
  --topic my-topic --partitions 6 --replication-factor 1

min.insync.replicas 와 같은 topic config를 기본값이나 broker의 값을 override 하여 지정하고 싶다면 --config 옵션을 사용한다.

kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --create \
  --topic my-topic --partitions 6 --replication-factor 1 --config min.insync.replicas=1

topic 상세 조회

./kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --topic my-topic --describe

결과는 다음과 같다. partition의 갯수, partition의 leader, replica, in-sync replica broker id 정보를 얻을 수 있다.

Topic:my-topic    PartitionCount:6    ReplicationFactor:1    Configs:
    Topic: my-topic    Partition: 0    Leader: 1    Replicas: 1    Isr: 1
    Topic: my-topic    Partition: 1    Leader: 1    Replicas: 1    Isr: 1
    Topic: my-topic    Partition: 2    Leader: 1    Replicas: 1    Isr: 1
    Topic: my-topic    Partition: 3    Leader: 1    Replicas: 1    Isr: 1
    Topic: my-topic    Partition: 4    Leader: 1    Replicas: 1    Isr: 1
    Topic: my-topic    Partition: 5    Leader: 1    Replicas: 1    Isr: 1

topic 설정 변경

파티션 및 리플리카 수 변경, config 변경등을 할 수 있다. 파티션 수는 현재 수에서 증가만 할 수 있다.

./kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --alter --topic my-topic --partitions 9

결과를 보면 정상적으로 변경된 것을 볼 수 있다.

./kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --describe --topic my-topic
Topic: my-topic    TopicId: gfcRYfJ5QW-pubajGhAelw    PartitionCount: 9    ReplicationFactor: 1    Configs:
    Topic: my-topic    Partition: 0    Leader: 1    Replicas: 1    Isr: 1
    Topic: my-topic    Partition: 1    Leader: 1    Replicas: 1    Isr: 1
    Topic: my-topic    Partition: 2    Leader: 1    Replicas: 1    Isr: 1
    Topic: my-topic    Partition: 3    Leader: 1    Replicas: 1    Isr: 1
    Topic: my-topic    Partition: 4    Leader: 1    Replicas: 1    Isr: 1
    Topic: my-topic    Partition: 5    Leader: 1    Replicas: 1    Isr: 1
    Topic: my-topic    Partition: 6    Leader: 1    Replicas: 1    Isr: 1
    Topic: my-topic    Partition: 7    Leader: 1    Replicas: 1    Isr: 1
    Topic: my-topic    Partition: 8    Leader: 1    Replicas: 1    Isr: 1

파티션수 변경은 키를 사용하는 경우 키별로 저장되는 파티션의 위치가 바뀌면서 ordering 등에 문제가 생길 수 있어 주의해야 한다.

topic 삭제

./kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --delete --topic my-topic

topic 삭제는 broker 설정에서 delete.topic.enable=true 로 되어 있어야 가능하다.

Producer & Consumer console

kafka producer, consumer를 cli로 간단히 사용할 수 있게 해주는 cli 이다. kakfa-console.producer.sh 를 사용하여 메세지를 kafka로 보낼 수 있다.

./kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topic my-topic
>hi
>this is kafka

--producer-property를 사용하면 acks 의 레벨 설정도 가능하다.

./kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topic my-topic acks=all

kafka-console-consumer.sh 를 사용하면 kafka에 저장되어 있는 로그를 받을 수 있다.

./kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic my-topic
hi
this is kafka

--from-beginning 을 사용하면 로그를 처음부터 가져올 수 있다.

key 사용하기

key를 포함하여 로그를 보내고 싶다면 다음과 같이 실행하면 된다.

./kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topic my-topic --property parse.key=true --property key.separator=,
>1,jason
>2,mike

consumer 쪽에서도 key를 보기 위해서는 --property print.key=true 를 사용하면 확인할 수 있다.

./kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic my-topic --property print.key=true --property key.separator=,
1,jason
2,mike

consumer group의 사용

--group 옵션으로 consumer group을 지정할 수 있다. consumer group이 존재하지 않는다면 consumer-group을 새로 생성한다. 이 경우 --from-beginning 옵션을 사용하지 않는다면 새로 들어오는 데이터부터 조회된다.

만약 --group 옵션으로 지정되는 consumer group이 이미 존재한다면, 이전에 커밋한 offset 이후의 메세지만 가져오게 된다. 이 경우 --from-beginning 옵션은 무시된다.

./kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic my-topic --group my-first-group

Consumer Group 관리

kafka-consumer-console.sh 에서는 consumer group 생성을 해주고, 메세지를 받는 역할을 하지만 consumer group을 관리하지는 못한다. consumer group에 대한 미세한 조회 및 관리는 kafka-consumer-groups.sh 로 할 수 있다.

consumer group 리스트 조회

./kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --list
my-first-group

consumer group의 상세 조회

./kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --describe --group my-first-group


GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                                    HOST            CLIENT-ID
my-first-group  my-topic        2          3               3               0               consumer-my-first-group-1-d581edd4-1fd0-4d53-9c8a-eef0af577cd8 /172.18.0.1     consumer-my-first-group-1
my-first-group  my-topic        1          2               2               0               consumer-my-first-group-1-d581edd4-1fd0-4d53-9c8a-eef0af577cd8 /172.18.0.1     consumer-my-first-group-1
my-first-group  my-topic        5          0               0               0               consumer-my-first-group-1-d581edd4-1fd0-4d53-9c8a-eef0af577cd8 /172.18.0.1     consumer-my-first-group-1
my-first-group  my-topic        0          4               4               0               consumer-my-first-group-1-d581edd4-1fd0-4d53-9c8a-eef0af577cd8 /172.18.0.1     consumer-my-first-group-1
my-first-group  my-topic        4          1               1               0               consumer-my-first-group-1-d581edd4-1fd0-4d53-9c8a-eef0af577cd8 /172.18.0.1     consumer-my-first-group-1
my-first-group  my-topic        3          7               7               0               consumer-my-first-group-1-d581edd4-1fd0-4d53-9c8a-eef0af577cd8 /172.18.0.1     consumer-my-first-group-1

consumer group의 파티션별 offset, lagging 정도, 접속된 consumer id, consumer host 정보 등 매우 상세한 정보를 확인할 수 있다.

consumer group의 consumer 정보 조회

현재 어떤 consumer가 conumser group에 포함되어있고, 얼마만큼의 파티션을 담당하고 있는지 간단한 정보를 확인할 수 있다.

./kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --describe --group my-first-group --members

GROUP           CONSUMER-ID                                                    HOST            CLIENT-ID                 #PARTITIONS
my-first-group  consumer-my-first-group-1-f3b03fb1-c65c-4741-a52d-30b701a0787e /172.18.0.1     consumer-my-first-group-1 3
my-first-group  consumer-my-first-group-1-3ad52cf3-4e00-4a29-a549-40506bb5f1ed /172.18.0.1     consumer-my-first-group-1 3

consumer group 설정 변경

만약 consumer-group의 topic에 대한 offset을 초기화 시키고 싶다면 다음 명령어를 사용하면 된다. consumer는 모두 inactive 상태여야 실행이 가능하다.

./kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --group my-first-group --topic my-topic --to-earliest --reset-offsets --execute

GROUP                          TOPIC                          PARTITION  NEW-OFFSET
my-first-group                 my-topic                       2          0
my-first-group                 my-topic                       1          0
my-first-group                 my-topic                       5          0
my-first-group                 my-topic                       0          0
my-first-group                 my-topic                       4          0
my-first-group                 my-topic                       3          0

consumer-group의 current-offset을 확인해보면 모두 0으로 바뀐것을 알 수 있다.

./kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --describe --group my-first-group

Consumer group 'my-first-group' has no active members.

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
my-first-group  my-topic        2          0               3               3               -               -               -
my-first-group  my-topic        1          0               2               2               -               -               -
my-first-group  my-topic        5          0               0               0               -               -               -
my-first-group  my-topic        0          0               4               4               -               -               -
my-first-group  my-topic        4          0               1               1               -               -               -
my-first-group  my-topic        3          0               7               7               -               -

dry-run 옵션도 제공한다. 이 경우 --execute 는 사용하지 않는다.

./kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --group my-first-group --topic my-topic --to-earliest --reset-offsets --dry-run

consumer group의 삭제

consumer 연결이 안된 상태에서 다음 명령어를 실행하면 consumer group이 삭제된다.

./kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --delete --group my-first-group
Deletion of requested consumer groups ('my-first-group') was successful.

이를 위해 ASG에서는 어떤 조건에서 capacity를 조정하도록 하도록 하는 기능을 제공하는 Scaling Policy를 제공한다. 종류로는 Dynamic Scaling policy, Predictive Scaling policy, Scheduled Action policy 등이 있다. 이번 글에서는 이 중에 Predictive Scaling Policy의 작동 원리 추론을 해보고자 한다.

Predictive Scaling Policy는 과거의 CPU 사용량, 네트워크 유입량 등의 로드의 과거 기록을 이용하여 미래에 특정 시점에 적용되어야 할 ASG의 capacity를 기계 학습(Machine Learning)을 통해 예측을 해준다. 가장 중요한 설정값은 Metrics and target utilization이며, 내용을 살펴보면 다음과 같다.

• Metric: 메트릭은 크게 두가지로 나누어진다. 첫번째 메트릭은 Load Metric으로 전체 사용량 추정에 사용하는 메트릭이다. 두번째 메트릭은 Scaling Metric으로 capcity 추측을 위해 사용하는 인스턴스당 사용량의 평균 메트릭이며, Target Utilization 값도 이 메트릭 기준으로 정한다. 선택할 수 있는 메트릭은 다음과 같으며, Load Metric과 Scaling Metric을 다르게 선택하는 Custom metric pair 옵션도 제공한다.
  • CPU Utilization: Load Metric에서는 전체 CPU 사용량, Scaling 메트릭에서는 인스턴스당 CPU 사용량의 평균
  • Network In (bytes): Load Metric에서는 전체 네트워크 유입량, Scaling 메트릭에서는 인스턴스당 메트릭 유입량의 평균
  • Network Out (bytes): Load Metric에서는 전체 네트워크 송출량, Scaling 메트릭에서는 인스턴스당 메트릭 송출량의 평균
  • Application Load Balancer request count: ALB에 연결된 경우만 사용 가능. Load Metric에서는 전체 request count, Scaling 메트릭에서는 인스턴스 당 request count의 평균
• Target Utilization: Scaling Metric의 유닛을 사용하며, 이 수치를 유지하도록 capacity 값이 예측된다. e.g. CPU Utilization의 경우 40%

Predictive Scaling에 대해서 AWS 메뉴얼을 보다보니 Load Metric과 Scaling Metric이 각각 어떤 역할을 하는지 헷갈렸으며, 특히 두 가지 종류를 다르게 설정할 수 있는 Custom Metric Pair 에 대해서는 더욱 이해가 안갔었다. 자료를 찾아보았지만 정확히 원리는 설명해주는 곳은 없어 내가 가진 약간의 ML 지식으로 나라면 이렇게 개발했을 거라 추측해 보았다.

단일 Metric 사용 시

Load Metric의 사용

우선 과거 사용량을 기준으로 미래의 특정 시점의 사용량을 예측해야 한다는 것쯤은 쉽게 추측할 수 있다. 이 때 사용하는 메트릭이 Load Metric이다. 예를 들어 CPU 전체 사용량을 Load Metric으로 선택한 경우, 과거의 특정 시간, 특정 요일의 CPU 전체 사용량을 학습하여, 미래의 특정 시점의 사용량을 예측하도록 한다. 학습하는 데이터를 간단히 적어보면 (학습 시 윈도우는 10분으로 가정한다.)

이런 데이터가 충분히 모여서 학습하면 특정 시점의 CPU 사용량 예측을 할 수 있을 것이다.

이렇게 예측한 Load Metric은 이후에 Scaling Metric 과 함께 capacity를 예측하는데 사용된다.

Scaling Metric의 사용

Load Metric을 통해서 미래의 CPU 전체 사용량까지 마친 시점에서 Scaling Metric은 어떻게 사용할 수 있을까 고민을 해보았다. 다음과 같이 학습 데이터를 준비해 학습한다면 특정 로드에 대한 Target Utilization에 해당하는 capacity 값을 추측할 수 있을 것이다.

이 데이터를 학습을 했다면, 미래의 특정 시점에 Load Metric 데이터와 Target Utilization 값을 통해 capacity 값을 파악할 수 있다. Target Utilization을 40%라 한다면 다음 데이터 입력과 학습된 모델을 통하여 capacity 값을 예측할 수 있을 것이다.

Custom Metric Pair 사용 시

단일 Metric 사용 시, Load Metric, Scaling Metric 설명을 통하여 Custom Metric Pair 가 어떠한 방식으로 작동할 지 예측해 볼 수 있다. Load Metric은 Network In 값을 사용하고, Scaling Metric은 CPU Utilization을 사용한 경우를 살펴보자. Target Utlization 값은 40%이라 가정한다.

Load Metric의 사용

단일 Metric 을 사용할 때와 동일하다.

이러한 데이터의 학습을 통해 생성된 모델을 통해 특점 시점의 Network In 총량을 예측할 수 있다.

Scaling Metric의 사용

단일 Metric과는 다르게 Load Metric과 Scaling Metric의 단위가 다르다.

이를 통해 다음 입력과 학습된 모델을 통해 특정 시간의 Target Utilization에 해당하는 capacity 값을 예측할 수 있다.

주의할 점

단일 Metric은 Load Metric과 Scaling Metric이 동일하기 때문에 두 메트릭의 상관 관계가 비교적 Linear하게 유지되어 예측에 문제가 없을 것이다. 그러나 Custom Metric Pair 를 사용하게 된다면 Load Metric과 Scaling Metric의 상관 관계가 명확하지 않은 케이스가 있을 수 있어 주의가 필요하다.

예를 들어 웹 API 서비스를 제공하는 ASG의 경우, API 별로 네트워크 In 은 비교적 비슷한데, CPU 사용량은 매우 상이하다면, 10000이라는 Network In 양이 사용되는 동안 동일 capacity에 대한 CPU 평균 사용량은 어떤 경우는 10%, 어떤 경우는 50%가 되는 경우가 생긴다고 해보자. 이 경우 Custom Metric Pair를 상기 예제처럼 선택한다면, 제대로된 학습을 기대하기 힘들며 예측도 엉망이 될 가능성이 높다.

아쉬운 점

Predictive Scaling Policy를 알아보면서 아쉬운 점이 있었다. 학습 값으로 Load Metric을 하나밖에 선택할 수 없다는 것이다. 사실 Load Metric으로 제공해주는 모든 종류의 Metric을 학습 시 사용하더라도 모델을 만들기 위한 계산량이나 걸리는 시간은 크게 차이가 없을 것이라 예상되는데, 복수 선택을 할 수 있게 했다면 더 좋았을 것이란 생각이 든다.

Ansible은 인프라 구성 관리 툴로써 puppet과 chef와 비슷한 툴 이다. 회사에서는 chef 정도만 간접적으로 경험해 본게 전부지만 마스터 서버와 에이전트 설치가 필요 없어 관심을 가지게 되었다.

Ansible을 Mac에 설치

mac에서 설치는 역시 homebrew. 다음 명령어로 간단히 설치가 가능하다. 글쓰는 시점을 기준으로 설치 시, 4.5 버전이 설치되었다.

brew install ansible

용어

Ansible에서 사용하는 주요 용어는 다음과 같다.

• Control Node: Ansible이 설치된 머신을 이야기 한다.

• Managed Node: Ansible을 사용하여 관리할 서버 (네트워크)를 이야기 한다. hosts라고도 불린다. 위에서도 이야기 하였지만, managed node에 에이전트를 별도로 설치할 필요가 없다.

• Inventory: manage할 서버에 대한 정보를 모아놓은 파일이다. hostfile이라고도 불린다. ip 주소 등 접근 정보를 저장한다.

• PlayBook: ansible이 실행하는 task의 ordered-list 집합이다.

가장 빠르게 실행해보기

ansible에서 관리할 서버 정보가 있는 inventory file을 작성해보자. 작업하는 파일이 위치하는 디렉토리를 하나 만든 후에 hosts.yml (이름은 상관 없다) 파일을 생성하고 다음과 같이 입력하면 firstserver 라는 alias로 접속할 서버를 정의할 수 있다.

all:
  hosts:
    firstserver:
      ansible_port: 22
      ansible_host: '3.35.0.208'
      ansible_user: 'ubuntu'
      ansible_ssh_private_key_file: '/path/to/ssh_key'

이렇게 입력한 후 다음과 같이 접속 테스트를 해볼 수 있다.

$ ansible all -m ping -i hosts.yml

firstserver | SUCCESS => {
    "ansible_facts": {
        "discovered_interpreter_python": "/usr/bin/python3"
    },
    "changed": false,
    "ping": "pong"
}

자, 이제 접속 되는 것 까지는 확인을 하였으니, 이제 firstserver 에 파일을 하나 생성해보도록 ad-hoc 명령어를 실행해 보자. 아래와 같이 명령어를 실행하면 firstserver에 ansible.txt 파일 하나가 생성된 것을 확인할 수 있다.

$ ansible -i hosts.yml firstserver -a 'touch ansible.txt'

firstserver | CHANGED | rc=0 >>

모듈 변수 지정이 없으면 ansible의 command module을 사용하게 되는데, piping이나 redirecting 같은 작업이 불가능하기 때문에 제한 사항이 많다. redirect를 통해 파일에 내용을 넣고 싶다면ansible.builtin.shell 모듈을 사용하면 된다. 아래 명령어를 실행하면 welcome to ansible 이라는 내용이 적혀 있는 ansible.txt 파일이 생성된다.

$ ansible -i hosts.yml -m ansible.builtin.shell firstserver -a 'echo "welcome to ansible" > ansible.txt'

firstserver | CHANGED | rc=0 >>

IaC 툴처럼 사용해보자

지금까지 ad-hoc 명령어를 직접 실행하도록 해보았다. 그러나 IaC 툴로서의 제대로된 역할을 하려면 결국 이러한 명령어들이 재사용이 가능하도록 코드로 작성되어야 한다. 이때 사용하는 것이 playbook 이다.

playbook 파일(welcome-playbook.yml)을 다음과 같이 만들어보자.

- name: create welcome file
  hosts: firstserver

  tasks:
    - name: create welcome file
      ansible.builtin.shell: 'echo "welcome to ansible" > ansible.txt'

다음 명령어를 실행하면, 이전 ad-hoc 명령어와 동일한 결과를 낸다.

$ ansible-playbook welcome-playbook.yml -i hosts.yml


PLAY [create welcome file] ********************************************************************

TASK [Gathering Facts] ************************************************************************
ok: [firstserver]

TASK [create welcome file] ********************************************************************
changed: [firstserver]

PLAY RECAP ************************************************************************************
firstserver                : ok=2    changed=1    unreachable=0    failed=0    skipped=0    rescued=0    ignored=0

모듈(module)에 대해

현재까지는 ansible.builtin.shell 모듈만을 사용해 봤지만, ansible에는 정말 다양한 목적에 맞는 다양한 모듈이 있다. 파일 복사에 특화된 ansible.builtin.copy 와 같은 모듈부터 aws instance를 생성하거나 상태를 제어할 수 있는 모듈까지 다양한 모듈을 제공한다. 앞으로 반복되는 업무에 지속적으로 사용해보며 유용한 팁이 있다면 공유할 예정이니 기대하시라.

서비스 등장 배경

AWS에서 설명하는 EventBridge 서비스가 생긴 배경에는 EDA를 서버리스 형태로 제공한다기 위해서라 말하면서, EDA가 복잡한 서비스 구조에서는 더 좋은 서비스간 통신 방식이라 이야기 한다.

왜 EDA가 더 좋은가? 항상 그렇지는 않지만...

마이크로서비스 기반의 아키텍처에서 Direct Command 방식의 서비스간 통신을 할 경우, 전체 서비스 규모가 작을 때는 문제가 되지 않지만, 연결된 서비스 종류가 많아지고, 서비스간 종속 관계가 되는 깊이가 깊어지는 경우, 다음 두가지 문제가 발생한다.

• 서비스가 많아질 수록 통신의 복잡성이 기하 급수적으로 높아진다. 단순히 서비스간 연결 구조가 복잡해지는 문제 뿐만 아니라, 서비스 API의 버저닝 문제까지 엮이면 훨씬 복잡해지단.
• 서버 to 서버 통신이 synchronous 하게 이루어질 경우, upstream의 서비스에 문제가 발생하면, downstream의 서비스에도 영향을 주어 더 넓은 장애를 유발할 수 있다.

EDA의 경우는 서비스가 생성하는 이벤트를 다른 서비스가 구독하는 형태이기 때문에, 위에서 거론된 두가지 문제가 발생하지 않는다는 장점이 있다.

EventBridge의 구성 요소

Event

이벤트소스에서 부터오는 이벤트라 보면 된다. AWS 서비스를 예를 들면 EC2의 상태가 변경된다던지, Auto Scaling Group에 인스턴스가 추가될 때, event가 발생한다. 만약 내가 개발하고 있는 커머스 어플리케이션이라 하면, 사용자가 특정 상품을 구매한 경우, 상품을 구매했다는 이벤트를 생성할 수 있다.

Event Rule

Event가 발생한 경우, Event Rule과 매칭되는 이벤트만 Event Rule에 연결된 Event Target으로 라우팅 된다. EDA에서 보면 구독을 할 수 있는 대상이라 보면 된다. 만약 Auto Scaling Group의 인스턴스가 추가되어 이벤트가 발생되었다고 하면, 해당 이벤트의 패턴과 매칭되는 Event Rule은 모두 이 이벤트를 통과시킬 것이다. 이 이벤트의 패턴을 Event Pattern이라 부르고, 이 패턴을 만족하는 이벤트는 이벤트 패턴을 가지는 모든 Rule의 Target에 라우팅 된다.

EC2 인스턴스가 종료 되었을 때 다음과 같은 이벤트가 생성된다.

{
  "version": "0",
  "id": "6a7e8feb-b491-4cf7-a9f1-bf3703467718",
  "detail-type": "EC2 Instance State-change Notification",
  "source": "aws.ec2",
  "account": "111122223333",
  "time": "2017-12-22T18:43:48Z",
  "region": "us-west-1",
  "resources": [
    "arn:aws:ec2:us-west-1:123456789012:instance/i-1234567890abcdef0"
  ],
  "detail": {
    "instance-id": "i-1234567890abcdef0",
    "state": "terminated"
  }
}

만약 Event rule의 이벤트 패턴을 다음과 같이 입력한 경우, 위 이벤트는 이 Event Rule의 필터링을 통과하고, Event Target까지 이벤트가 전달된다.

{
  "source": ["aws.ec2"],
  "detail-type": ["EC2 Instance State-change Notification"],
  "detail": {
    "state": ["terminated"]
  }
}

만약 terminated를 포함한 EC2 status change에 대한 모든 이벤트를 필터링하고 싶다면 다음과 같이 event pattern을 구성하면 된다.

{
  "source": ["aws.ec2"],
  "detail-type": ["EC2 Instance State-change Notification"],
  "detail": {
    "state": ["pending", "running", "shutting-down", "stopped", "stopping", "terminated"]
  }
}

다양한 AWS 서비스의 이벤트 패턴 예제는 이 문서에서 확인 가능하다.

*주의: 처음에는 혼동이 올수도 있는데, 이벤트가 Event Rule을 선택하는 것이 아니다. Event Rule과 연결된 Event Bus를 통과하는 모든 이벤트를 Event Rule이 필터링하여, 매칭되는 이벤트만 Event Target으로 이동시키는 구조임에 유의하자.

Event Target

event rule의 필터링을 통과한 이벤트가 도착하는 대상이다. 하나의 event rule에 여러개의 event target 등록이 가능하다. 즉, 해당 event rule의 이벤트 패턴을 만족하는 이벤트를 구독한다는 개념으로 생각하면 된다. Event Target으로 등록 가능한 AWS 서비스 또는 대상은 다음과 같다.

• API destination
• API Gateway
• AWS Batch job queue
• Amazon CloudWatch Logs group
• AWS CodeBuild project
• AWS CodePipeline
• Amazon Elastic Compute Cloud (Amazon EC2) CreateSnapshot API call
• Amazon EC2 RebootInstances API call
• Amazon EC2 StopInstances API call
• Amazon EC2 TerminateInstances API call
• Amazon ECS task
• Event bus in a different AWS account or AWS Region
• Firehose delivery stream (Amazon Kinesis Data Firehose)
• Incident Manager response plan
• Inspector assessment template (Amazon Kinesis Data Streams)
• Kinesis stream (Kinesis Data Streams)
• AWS Lambda function
• Amazon Redshift cluster (Data API statement execution)
• SageMaker Pipeline
• Amazon SNS topic
• Amazon SQS queue (including a FIFO queue)
• Amazon EC2 Systems Manager (SSM) Automation
• SSM OpsItem
• SSM Run Command
• AWS Step Functions state machine

예를 들어, 인스턴스가 종료 되었을 때 slack 등에 알림을 보내고 싶을 때, 슬랙의 incoming-hook API를 사용해 lambda function에 슬랙 알림을 구현해 놓고, event target으로 이 lambda function을 등록하면 인스턴스 종료 케이스에 대한 모니터링을 쉽게 할 수 있다.

Event Bus

이벤트를 받는 역할을 한다. Event Rule을 생성할 때, 하나의 Event Bus에 연결을 해야 한다. 이 Event Bus로 들어오는 이벤트만 Event Rule에서 필터링을 하게 된다. AWS 어카운트 마다 기본적으로 생성하는 Event Bus가 있는데, Default Event Bus라 부르고, AWS 서비스로 부터 생성된 이벤트는 모두 이 이벤트 버스로 들어온다. 목적에 따라 커스텀한 이벤트버스를 만들어 사용할 수 있다. 예를 들어 커머스 어플리케이션을 만드는데 Orders 라는 이벤트버스를 만들고, 주문 관련 이벤트 소스를 Orders Event Bus로 연결하면, 관련 이벤트가 발생하면 Orders Event Bus로 이벤트가 전달되고, 이 안에 존재하는 Event Rule이 필터링하여 Event Target으로 라우팅 할 것이다.

CloudWatch Events와 관계는?

EventBridge가 나오기 전에는 CloudWatch Events가 비슷한 역할을 담당하고 있었다. EventBridge에서 사용하는 API는 CloudWatch Events의 API와 동일하다. 즉, 기존의 CloudWatch Event에서 생성한 리소스가 EventBridge에서도 100% 호환된다고 보면 된다. 현재 AWS console에도 CloudWatch Events, EventBridge 메뉴가 모두 존재하는데 한쪽에서 Event Rule을 생성을 하면 다른쪽에서도 조회가 가능하다.

EventBridge가 CloudWatch Events에 비해 우수한 점은 CloudWatch Events는 AWS 서비스가 생성한 이벤트만 사용이 가능한 반면, EventBridge는 AWS 서비스 뿐만 아니라 SaaS나 개인 어플리케이션으로부터 생성된 이벤트도 사용이 가능하단 점이다.

그럼 무엇을 사용해야 할까? 개인적으로는 API 상으로는 100% 호환이 되고, EventBridge가 더 넓은 범위의 기능을 제공하므로 AWS EventBridge를 사용하는게 좋을 것 같다는 의견을 내며 글을 마무리 하고자 한다.

다음 명령어를 기준으로 해부를 시작해보겠다.

$ python -m timeit '"-".join(str(n) for n in range(100))'
10000 loops, best of 3: 28.6 usec per loop

위 명령어를 실행하게 되면 (1) "-".join(str(n) for n in range(100)) 코드를 10000번 실행시킨다. (2) 10000번 실행한 시간 / 10000을 계산 하여 평균 값을 낸다. (3) (1) ~ (2)의 과정을 3번 반복한다. (4) 3개의 평균 값 중에 가장 작은 값을 보여준다. best of 3: 28.6 usec per loop

위 과정을 실행하는 python timeit 모듈의 코드는 timeit.main 함수에서 확인할 수 있다

코드 실행 회수와 반복수 지정

코드 실행 횟수 10000과 반복수 3은 기본값이다. 이 값을 조정하고 싶다면 다음과 같이 -n (코드 실행 횟수)과 -r (반복횟수) 옵션을 추가하면 된다.

$ python -m timeit -n 1000 -r 5 '"-".join(str(n) for n in range(100))'
1000 loops, best of 5: 28.3 usec per loop

초기 실행 코드 삽입

반복 실행 별 코드 실행 전에 초기에 한번만 실행하고 싶은 코드가 있다면 -s 옵션을 사용하면 된다. 반복 회수(-r)를 5로 지정되었다면 iterations = 100 코드는 총 5번 실행됨에 유의하자. 이 옵션에 넣은 코드는 시간 측정 시 포함되지 않는다.

$ python -m timeit -n 1000 -r 5 -s "iterations = 100" '"-".join(str(n) for n in range(iterations))'
1000 loops, best of 5: 28.1 usec per loop

파일 내 function 실행

만약 python 파일 내의 function을 실행하고 싶다면 다음과 같이 하면 된다.

loop.py 라는 파일이 있고, 해당 파일의 코드는 다음과 같이 정의 되어 있다 가정하자.

def loop(iterations):
    "-".join(str(n) for n in range(iterations))

이 파일 내의 loop 함수를 실행하고 싶다면 다음과 같이 실행하면 된다.

$ python -m timeit -n 1000 -r 5 -s 'from loop import loop; iterations = 100' 'loop(100)'
1000 loops, best of 5: 21.9 usec per loop

그 외 고려사항

• -v 옵션: 각 반복 횟수의 결과 값을 모두 보여준다.

$ python -m timeit -n 1000 -r 5 -v -s "from loop import loop" 'loop(100)'
raw times: 0.0276 0.0233 0.0229 0.0227 0.0227
1000 loops, best of 5: 22.7 usec per loop

• GC 켜기: timeit 명령어는 각 반복 실행 시작 전에 gc를 disable 하고, 반복 실행이 끝다면 다시 enable 시키도록 되어 있다. (-r이 5이면 gc를 5번 껐다 켰다 함). 만약 gc를 enable한 상태로 코드를 돌리고 싶다면 -s 옵션에 gc.enable()을 추가해주면 된다. 만약 자신의 코드가 python의 GC를 많이 사용하는 코드라면 정확한 성능 측정을 위하여 gc를 enable 하는 것도 고려해보자.

$ python -m timeit -n 1000 -r 5 -s 'from loop import loop; iterations = 100; gc.enable()' 'loop(100)'
1000 loops, best of 5: 22.2 usec per loop

이 정도면 timeit 명령어를 사용할 때 찾아보기에 충분한 것 같다. 끝!

정의

리눅스 메뉴얼의 listen API를 보면, backlog 인자에 대해 다음과 같이 정의되어 있다.

The backlog argument defines the maximum length to which the
queue of pending connections for sockfd may grow.

구글 변역기의 힘을 빌려보자면

무슨 말인지 모르겠다 (아직도 먼 기계학습...). 마음가는대로 해석해보면

socketfd의 커넥션 중, 허용 가능한 pending된 커넥션 큐의 최대 길이

정도가 될 것이다.

그러나 정의 만으로도 정확히 이해하기가 힘들었다. 그래서 무식하게 간단한 프로그램을 만들어 실험을 해보기로 결정했다.

에코 서버, 클라이언트 프로그램

필자는 python을 주로 사용해와서, python이 편하기 때문에 python으로 간단한 프로그램을 짜보았다.

에코 서버 프로그램

클라이언트에서 bytes를 보내면, 그 값을 그대로 돌려주는 간단한 서버 프로그램이다

# echo_server.py
import socket
import sys


if __name__ == '__main__':
    host = sys.argv[1]
    port = int(sys.argv[2])
    num_of_backlog = int(sys.argv[3])

    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        s.bind((host, port))
        s.listen(num_of_backlog)
        print(f'[INFO] listens to {host}:{port} with backlog {num_of_backlog}')
        while True:
            conn, addr = s.accept()
            with conn:
                print(f'[INFO] connected to {addr[0]}:{addr[1]}')
                while True:
                    data = conn.recv(1024)
                    if not data:
                        break
                    print(f'[INFO] received data: {data.decode("utf-8")}')

                    # echo to client
                    conn.sendall(data)
    finally:
        print('[INFO] socket closed')
        s.close()

• python 3.8을 사용. f string을 사용했기 때문에 python 3.6 이상부터 작동할 것으로 예상
• TCP socket을 사용하였다.
• 쓰레드가 1개이기 때문에, 실제로 메세지를 주고 받을 수 있는 클라이언트는 동시에 1개 뿐이다.

실행은 다음과 같이 하면 된다.

python echo_server.py HOST PORT BACKLOG

Backlog를 테스트 해보기 위해서, BACKLOG를 인자로 넣었다.

에코 클라이언트 프로그램

이 프로그램은 입력값을 키보드로 부터 받아, 서버로 넘겨주는 간단한 클라이언트 프로그램이다.

# echo_client.py
import socket
import sys


if __name__ == '__main__':
    host = sys.argv[1]
    port = int(sys.argv[2])

    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        s.connect((host, port))
        print(f'[INFO] connected to {host}:{port}')
        while True:
            # make prompts
            print('>>', end=' ')

            # send input data to server
            input_data = input()
            s.sendall(input_data.encode(encoding='utf-8'))

            # receive data from server
            received_data = s.recv(1024)
            print(f'[INFO] received data: {received_data.decode("utf-8")}')
    finally:
        print('[INFO] socket closed')
        s.close()

• python 3.8을 사용. f string을 사용했기 때문에 python 3.6 이상부터 작동할 것으로 예상
• connection timeout 값은 60초이다. 즉, 60초 안에 TCP 커넥션을 맺지 못하면 timeout error가 발생한다.

실행은 다음과 같이 하면 된다.

실험

• Mac OS Catalina iTerms2 터미널에서 실험하였다.
• 서버는 한개만 실행한다. 실행 명령어는 python echo_server.py 127.0.0.1 10000 2 로 backlog 값은 2로 주었다.
• 클라이언트 프로그램은 총 4개를 실행한다. 실행 명령어는 python echo_client.py 127.0.0.1 10000 이다. 각각 1번, 2번, 3번, 4번 클라이언트라 부르자.

상기 이미지를 보면, 4개의 클라이언트 프로그램을 실행했고, 실행 순서는 위에서 부터 차례로 실행했다. 3개까지는 커넥션이 잘 맺어졌고, 값을 입력할 수 있는 promps로 넘어간걸 볼 수 있다.

netstat 명령어를 사용해, 실제 연결된 커넥션을 봐보자. netstat -anlt | grep 10000 명령어를 치면 결과가 다음과 같이 나온다.

tcp4       0      0  127.0.0.1.60240        127.0.0.1.10000        SYN_SENT
tcp4       0      0  127.0.0.1.10000        127.0.0.1.60237        ESTABLISHED
tcp4       0      0  127.0.0.1.60237        127.0.0.1.10000        ESTABLISHED
tcp4       0      0  127.0.0.1.10000        127.0.0.1.60236        ESTABLISHED
tcp4       0      0  127.0.0.1.60236        127.0.0.1.10000        ESTABLISHED
tcp4       0      0  127.0.0.1.10000        127.0.0.1.60235        ESTABLISHED
tcp4       0      0  127.0.0.1.60235        127.0.0.1.10000        ESTABLISHED
tcp4       0      0  127.0.0.1.10000        *.*                    LISTEN

3개의 클라이언트는 TCP connection이 ESTABLISHED 된 상태이고, 하나의 클라이언트는 SYN_SENT 상태로 대기 중이다. 이 클라이언트가 4번 클라이언트인 것을 추측할 수 있다. 60초가 지난 후, 4번 클라이언트는 TimeoutError와 함께 종료된다. (4번 클라이언트는 이미지 캡처에서 아웃 시키겠다.)

[INFO] socket closed
Traceback (most recent call last):
  File "echo_client.py", line 11, in <module>
    s.connect((host, port))
TimeoutError: [Errno 60] Operation timed out

ESTABLISHED 인 1번, 2번, 3번의 클라이언트에 입력을 해보면, 실제로 작동하는 클라이언트는 1번 클라이언트고, 2번, 3번 클라이언트는 아무런 응답이 없다. 서버의 스레드가 하나여서, 실제로 서버 프로그램과 데이터를 주고 받을 수 있는 클라이언트는 1개인 것이다. 즉, 2번, 3번의 연결 상태는 ESTABLISHED 이미지만, listen 정의에서 언급된 큐에 대기중인 상태인 것을 알 수 있다.

1번 클라이언트를 강제 종료해보자 (Ctrl+C를 주어, KeyboardInterrupt를 발생시켰다). 대기하고 있던 2번 클라이언트가 서버로 부터 응답을 받게 된다.

이를 통해 backlog 정의에서 언급한 queue는 FIFO인것을 확인할 수 있다. (큐가 맞다!)

정리

만약 서버 프로그램이 동시에 응답할 수 있는 수가 3개라 가정하고, backlog 값을 5라 가정하자. 10개의 요청이 차례로 들어왔을 때, 상황을 그림을 그려보면 다음과 같다. (숫자는 요청을 의미하고, 가장 먼저 들어온 요청이 1번이다)

• 1~3 요청을 보낸 클라이언트는 TCP connection이 established되고, 서버와 데이터를 주고 받을 수 있을 것이다.

• listen backlog queue에 있는 48번 요청은 TCP connection은 established 되었지만, 서버와 데이터를 주고 받지는 못한다. 13번 요청 중 커넥션이 끊어지면, 4번 부터 차례로 데이터를 주고 받을 수 있도록 backlog에서 큐를 이동할 것이다.

• tcp max sync backlog라는 곳에 저장된 9, 10번 요청은 SYNC_SENT 상태를 유지하게 된다. 실제로는 유지한다기 보다는 클라이언트에서 지속적으로 SYNC 요청을 시도하고, 서버 쪽에서는 계속 패킷을 drop 시킨다. tcp connection timeout이 값 이상으로 계속 시도하게되면, 클라이언트 쪽에서 커넥션에 대한 timeout을 발생키시고, 연결 시도를 중단한다.

• 이 backlog는 OS 종류에 따라 없을 수도 있고, 다르게 동작할 수도 있다. 예를 들어 어떤 OS에서는 이 backlog가 존재하지 않고 바로 요청을 refuse 할수도 있다. 이 내용은 listen 매뉴얼에도 잘 나와있다.

• tcp max sync backlog를 큐라 부르지 않는 이유는 순서가 보장이 안되기 때문이다. 실험해본 결과, 이 backlog에 대기중인 상태에서 listen의 backlog queue로 넘어갈 때, 요청한 순서대로 listen의 backlog 큐로 넘어가지 않았다. 단, mac OS에서 실험해보았기 때문에, mac OS에 국한된 작동일 수도 있다.

주의

• linux 기준으로 listen의 backlog 값이 net.core.somaxconn 커널 파라미터 값보다 크다면, 실제 적용은 net.core.somaxconn 값이 된다.
• listen의 backlog 인자값과 실제 backlog queue 사이즈는 OS에 따라 다를 수도 있다. backlog 인자값이 1로 지정하더라도, OS에 따라 실제 queue의 사이즈는 4로 적용될 수도 있음을 유의하자.

feat. gunicorn

gunicorn은 python의 WSGI 인터페이스를 지원하는 unix를 위한 http 서버이다. 설정값중에 backlog라는 값이 있다. gunicorn의 코드를 보았을 때, 이 backlog 값은 실제 listen API의 backlog 값과 동일한 것을 알 수 있었다. 다음 코드는 gunicorn 20.1.0의 gunicorn/sock.py 파일의 내용이다.

그렇다. gunicorn의 backlog 설정은, 말그대로 listen API의 backlog 인자와 동일하다.

마무리

사실 이 블로그는 gunicorn의 backlog 값이 어떤 역할을 하는지 궁금해서 시작되었다. 처음에는 gunicorn을 가지고 실험을 해보다, 결국 gunicorn의 코드를 까보게 되었고, listen API를 사용하는 것을 알고, listen API도 무식하게 backlog 인자를 테스트해보면서 여기까지 오게 되었다. 여러 OS 환경(특히 Window)에서 테스트하지 못했다는 점이 아쉽지만, 이 정도 정보면 실무에서 실제로 값을 설정할 때, 내가 뭔짓을 하고 있는지 아는데 부족하지 않을 것이다.

참고 자료

• https://man7.org/linux/man-pages/man2/listen.2.html
• https://meetup.toast.com/posts/55
• https://docs.python.org/3/library/socket.html
• https://realpython.com/python-sockets/
• https://stackoverflow.com/questions/62641621/what-is-the-difference-between-tcp-max-syn-backlog-and-somaxconn
• https://stackoverflow.com/questions/36594400/what-is-backlog-in-tcp-connections