Streams: a new general purpose data structure in Redis.

2017-10-02

스트림: Redis의 새로운 범용 자료구조

몇 달 전까지만 해도 저에게 스트림은 메시징 분야에서 흥미롭고 비교적 단순한 개념 그 이상이 아니었습니다. Kafka가 이 개념을 널리 알린 뒤에는 주로 Disque라는 메시지 큐에서 스트림의 활용 가능성을 검토했습니다. Disque는 이제 Redis 4.2 모듈로 전환될 예정인 프로젝트입니다. 이후 Disque의 본질이 AP 메시징, 즉 클라이언트가 별다른 수고를 들이지 않고도 내결함성과 전달 보장을 얻는 데 있다는 결론에 도달했고, 그런 관점에서는 스트림 개념이 적합하지 않다고 판단했습니다.

하지만 같은 시기에 Redis에는 기본으로 제공하는 자료구조에 대해 마음이 편치 않은 문제가 있었습니다. Redis 리스트, 정렬 집합(sorted set), Pub/Sub 기능 사이에는 일종의 간극이 존재합니다. 이 도구들을 잘 조합하면 메시지나 이벤트의 연속을 모델링할 수 있지만, 각각 분명한 트레이드오프가 있습니다. 정렬 집합은 메모리를 많이 차지하고, 같은 메시지가 반복해서 전달되는 상황을 자연스럽게 표현할 수 없으며, 클라이언트가 새 메시지를 기다리며 블로킹할 수도 없습니다. 정렬 집합은 순차적인 자료구조가 아니라 점수를 바꿔 원소를 이리저리 옮길 수 있는 집합이기 때문에, 시계열 같은 용도에는 맞지 않는 것은 당연합니다. 리스트도 특정 사용 사례에서 비슷한 적용상의 제약을 만듭니다. 리스트 중간에 무엇이 있는지 탐색할 수 없는 것이 대표적인데, 이 경우 접근 시간이 선형으로 늘어나기 때문입니다. 게다가 팬아웃도 불가능하고, 리스트에 대한 블로킹 연산은 하나의 원소를 하나의 클라이언트에게만 전달합니다. 리스트에는 “이 원소부터 시작하는 데이터를 달라”고 지정할 수 있는 고정된 원소 식별자도 없었습니다. 일대다 작업에는 Pub/Sub이 있고 많은 경우에 훌륭하게 동작하지만, 어떤 경우에는 fire-and-forget 방식을 원하지 않습니다. 이력을 유지하는 것이 중요합니다. 단순히 연결이 끊긴 뒤 메시지를 다시 가져오기 위해서만이 아니라, 시계열처럼 메시지 목록 자체를 범위 질의로 탐색해야 하는 경우도 많기 때문입니다. 예컨대 “지난 10초 동안의 온도 측정값은 어땠는가?” 같은 질문에 답해야 합니다.

저는 위 문제들을 정렬 집합과 리스트를 하나의 더 유연한 자료구조로 일반화하는 방식으로 해결해 보려 했습니다. 하지만 설계 시도는 거의 매번 결과물이 기존 자료구조보다 훨씬 더 인위적인 형태로 귀결되었습니다. Redis의 장점 중 하나는 외부에 노출하는 자료구조가 “살바토레가 만든 API”라기보다 자연스러운 컴퓨터 과학 자료구조에 가깝다는 점입니다. 결국 시도를 멈추고 “지금 제공할 수 있는 건 여기까지다. 나중에 Pub/Sub에 이력을 추가하거나 리스트 접근 패턴을 좀 더 유연하게 만들면 되겠지”라고 생각했습니다. 하지만 컨퍼런스에서 사용자가 다가와 “Redis에서 시계열을 어떻게 모델링하면 좋을까요?” 같은 질문을 할 때마다 얼굴이 굳어지곤 했습니다.

탄생

Redis 4.0에 모듈 기능이 도입된 뒤, 사용자들이 이 문제를 직접 해결하려는 시도를 하기 시작했습니다. 그중 한 명인 Timothy Downs가 IRC로 다음과 같은 메시지를 보내왔습니다.

<forkfork> 제가 만들려는 모듈은 트랜잭션 로그 스타일의 자료형을 추가하는 것입니다. 즉, 아주 많은 구독자가 Redis 메모리 사용량을 크게 늘리지 않고도 Pub/Sub과 유사한 동작을 할 수 있게 하는 것이죠
<forkfork> Redis가 각 컨슈머의 위치를 추적하고 구독자마다 메시지를 복제하는 대신, 구독자가 메시지 큐에서 자신의 위치를 직접 유지하는 방식입니다

이 아이디어가 제 상상력을 자극했습니다. 며칠 동안 곰곰이 생각하다가, 이 기회에 위의 모든 문제를 한 번에 해결할 수 있겠다는 생각이 들었습니다. 제가 필요했던 것은 “로그”라는 개념을 다시 상상하는 것이었습니다. 로그는 기본적인 프로그래밍 요소이고 누구나 익숙합니다. 파일을 추가 모드로 열어 어떤 형식으로든 데이터를 쓰는 것만큼 단순하기 때문입니다. 하지만 Redis의 자료구조는 추상적이어야 합니다. 자료구조가 메모리에 상주하고, 우리가 RAM을 쓰는 이유는 게으르기 때문이 아니라 몇 개의 포인터를 활용해 자료구조를 개념화하고 추상화함으로써 명백한 한계에서 벗어나게 하기 위해서입니다. 예컨대 일반적인 로그에는 몇 가지 문제가 있습니다. 오프셋이 논리적인 값이 아니라 실제 바이트 오프셋이라는 점이 그렇습니다. 만약 항목이 삽입된 시각과 연관된 논리적 오프셋을 원한다면 어떨까요? 그러면 범위 질의를 공짜로 얻을 수 있습니다. 마찬가지로 로그는 가비지 컬렉션이 어려운 경우가 많습니다. 추가만 가능한 자료구조에서 오래된 원소를 어떻게 제거할까요? 우리가 이상적으로 상상하는 로그에서는 단순히 “최대 이 개수만큼만 유지하고 싶다”고 말하면 오래된 항목이 사라지도록 하면 됩니다.

Timothy의 초기 아이디어에서 출발해 명세를 작성하는 동안, 저는 Redis Cluster 내부의 일부를 최적화하기 위해 사용하던 래딕스 트리(radix tree) 구현을 작업하고 있었습니다. 이는 공간 효율적이면서도 범위 조회를 로그 시간 안에 수행할 수 있는 로그를 구현할 토대가 되었습니다. 동시에 제 설계에 잘 어울릴 만한 다른 아이디어를 얻기 위해 Kafka 스트림에 대해 읽기 시작했고, 그 과정에서 Kafka 컨슈머 그룹 개념을 차용해 Redis와 인메모리 사용 사례에 맞게 다시 이상화했습니다. 하지만 명세는 수개월 동안 명세로만 남아 있었고, 한참 뒤에는 그동안 사람들과 이야기를 나누며 얻은 많은 힌트를 반영해 거의 처음부터 다시 작성했습니다. Redis 스트림이 특히 시계열을 위한 훌륭한 사례가 되기를 바랐고, 다른 종류의 이벤트나 메시징 애플리케이션을 위한 용도로도 마찬가지였습니다.

코드를 작성해 봅시다

Redis Conf에서 돌아온 뒤 여름 동안 저는 “listpack”이라는 라이브러리를 구현하고 있었습니다. 이 라이브러리는 ziplist.c의 후계자에 해당하며, 단일 메모리 할당 안에 문자열 원소들의 리스트를 표현할 수 있는 자료구조입니다. 매우 특화된 직렬화 형식일 뿐이지만, 오른쪽에서 왼쪽으로 역방향 파싱도 가능하다는 특징이 있습니다. 이는 모든 사용 사례에서 ziplist를 대체하기 위해 필요한 특성이었습니다.

래딕스 트리와 listpack을 결합하면 공간 효율적이면서도 인덱싱된, 즉 ID와 시간으로 랜덤 액세스가 가능한 로그를 쉽게 만들 수 있습니다. 이 준비가 끝나자 스트림 자료구조를 구현하는 코드를 작성하기 시작했습니다. 아직 구현을 마무리하는 중이지만, 현재 Github의 Redis “streams” 브랜치에는 이미 충분히 많은 기능이 들어가 있어 직접 실행해 보고 즐길 수 있습니다. API가 100% 확정되었다고 주장하려는 것은 아니지만, 흥미로운 사실 두 가지가 있습니다. 하나는 이 시점에서 컨슈머 그룹과 스트림을 조작하는 몇몇 덜 중요한 명령어를 제외하고는 큰 기능들은 이미 구현되었다는 점입니다. 다른 하나는 약 두 달 뒤 모든 것이 안정적으로 보이면 스트림 관련 작업을 4.0 브랜치로 백포트하기로 결정했다는 점입니다. 이는 Redis 사용자가 스트림을 사용하기 위해 Redis 4.2까지 기다릴 필요가 없으며, 가능한 한 빨리 프로덕션에서 사용할 수 있게 된다는 뜻입니다. 새로운 자료구조이기 때문에 코드 변경의 거의 대부분이 새로운 코드 안에 자체적으로 포함되어 있어 이런 일이 가능합니다. 단, 리스트 블로킹 연산은 예외였는데, 스트림과 리스트의 블로킹 연산이 같은 코드를 공유하도록 리팩터링하면서 오히려 Redis 내부가 크게 단순해졌습니다.

튜토리얼: Redis 스트림에 오신 것을 환영합니다

어떤 면에서 스트림은 Redis 리스트의 강화판이라고 생각할 수 있습니다. 스트림의 원소는 단순한 문자열 하나가 아니라 필드와 값으로 이루어진 객체에 가깝습니다. 범위 질의가 가능하고 빠릅니다. 스트림의 각 항목은 논리적 오프셋인 ID를 가집니다. 서로 다른 클라이언트가 지정된 ID보다 큰 ID를 가진 원소를 기다리며 블로킹할 수 있습니다. Redis 스트림의 핵심 명령어는 XADD입니다. 그렇습니다. 모든 Redis 스트림 명령어는 “X”로 시작합니다.

> XADD mystream * sensor-id 1234 temperature 10.5
1506871964177.0

XADD 명령어는 지정된 스트림 “mystream”에 지정된 항목을 새로운 원소로 추가합니다. 위 예제에서 항목은 sensor-id와 temperature라는 두 필드를 가지고 있지만, 같은 스트림 안의 각 항목은 서로 다른 필드를 가질 수 있습니다. 같은 필드 이름을 사용하면 메모리 사용량이 더 효율해진다는 점만 기억하면 됩니다. 흥미로운 점은 필드 순서가 보장된다는 것입니다. XADD는 방금 삽입된 항목의 ID를 반환하는데, 세 번째 인자에 별표(*)를 넣어 ID 자동 생성을 요청했기 때문입니다. 거의 항상 이렇게 사용하는 것이 바람직하지만, 명령어를 슬레이브와 AOF 파일로 복제하기 위해 특정 ID를 강제 지정하는 것도 가능합니다.

ID는 두 부분으로 구성됩니다. 밀리초 단위 시각과 일련번호입니다. 1506871964177이 밀리초 시각이며, 이는 밀리초 해상도를 가진 Unix 시간입니다. 점 뒤의 숫자 0은 일련번호로, 같은 밀리초 안에 추가된 항목을 구분하기 위해 사용됩니다. 두 숫자 모두 64비트 부호 없는 정수입니다. 따라서 같은 밀리초 안이라도 원하는 만큼 많은 항목을 스트림에 추가할 수 있습니다. ID의 밀리초 부분은 ID를 생성하는 Redis 서버의 현재 로컬 시각과 스트림 안의 마지막 항목 중 더 큰 값을 사용해 얻습니다. 따라서 예컨대 컴퓨터 시계가 뒤로 돌아가더라도 ID는 계속 증가합니다. 어떤 면에서는 스트림 항목 ID를 128비트 정수 전체로 생각할 수도 있습니다. 하지만 ID가 항목이 추가된 인스턴스의 로컬 시각과 연관되어 있다는 사실 덕분에 밀리초 정밀도의 범위 질의를 공짜로 얻을 수 있습니다.

짐작하셨겠지만, 매우 빠른 속도로 두 항목을 추가하면 일련번호만 증가하게 됩니다. MULTI/EXEC 블록으로 “빠른 삽입”을 간단히 시뮬레이션해 볼 수 있습니다.

> MULTI
OK
> XADD mystream * foo 10
QUEUED
> XADD mystream * bar 20
QUEUED
> EXEC
1) 1506872463535.0
2) 1506872463535.1

위 예제는 스키마를 미리 지정하지 않고도 서로 다른 항목에 대해 서로 다른 필드를 사용할 수 있음을 보여줍니다. 실제로 일어나는 일은 각 블록(보통 50~150개의 메시지를 포함)의 첫 번째 메시지가 기준점으로 사용되고, 같은 필드를 가진 이후 항목들은 “이 블록의 첫 항목과 필드가 같다”는 하나의 플래그로 압축된다는 것입니다. 따라서 연속된 메시지에 같은 필드를 사용하면 시간이 지나면서 필드 집합이 서서히 바뀌더라도 메모리를 크게 절약할 수 있습니다.

스트림에서 데이터를 가져오는 방법은 두 가지입니다. XRANGE 명령어로 구현된 범위 질의와 XREAD 명령어로 구현된 스트리밍입니다. XRANGE는 시작부터 끝까지의 항목을 양 끝을 포함해 가져옵니다. 예컨대 ID를 알고 있다면 단일 항목을 다음과 같이 가져올 수 있습니다.

> XRANGE mystream 1506871964177.0 1506871964177.0
1) 1) 1506871964177.0
   2) 1) "sensor-id"
      2) "1234"
      3) "temperature"
      4) "10.5"

하지만 최소 ID와 최대 ID를 의미하는 특별한 시작 기호 “-”와 종료 기호 “+”를 사용할 수도 있습니다. 반환되는 항목 수를 제한하기 위해 COUNT 옵션을 사용하는 것도 가능합니다. 좀 더 복잡한 XRANGE 예제는 다음과 같습니다.

> XRANGE mystream - + COUNT 2
1) 1) 1506871964177.0
   2) 1) "sensor-id"
      2) "1234"
      3) "temperature"
      4) "10.5"
2) 1) 1506872463535.0
   2) 1) "foo"
      2) "10"

여기서는 ID 범위를 기준으로 생각하고 있지만, ID의 “일련번호” 부분을 생략할 수 있기 때문에 XRANGE를 이용해 특정 시간 범위에 해당하는 원소들을 가져올 수도 있습니다. 따라서 밀리초 단위 시각만 지정하면 됩니다. 다음 명령은 “Unix 시각 1506872463부터 시작하는 10개의 항목을 달라”는 의미입니다.

127.0.0.1:6379> XRANGE mystream 1506872463000 + COUNT 10
1) 1) 1506872463535.0
   2) 1) "foo"
      2) "10"
2) 1) 1506872463535.1
   2) 1) "bar"
      2) "20"

XRANGE에 대해 마지막으로 짚어야 할 중요한 점은, 응답으로 ID를 함께 받게 되고 바로 다음 ID는 ID의 일련번호 부분만 증가시키면 간단히 얻을 수 있으므로, XRANGE를 이용해 전체 스트림을 점진적으로 순회할 수 있다는 것입니다. 호출마다 지정된 개수의 원소를 받으면 됩니다. Redis의 *SCAN 계열 명령어가 원래 순회를 염두에 두고 설계되지 않았던 자료구조를 *그럼에도* 순회할 수 있게 해 준 뒤로, 같은 실수를 반복하지 않으려 했습니다.

XREAD로 스트리밍하기: 새 데이터를 기다리며 블로킹하기

XRANGE는 ID나 시간으로 범위를 조회하거나 ID로 단일 원소에 접근하려 할 때 완벽합니다. 하지만 데이터가 도착하는 대로 스트림을 소비해야 하는 서로 다른 클라이언트들의 경우에는 충분하지 않으며, 일종의 폴링을 필요로 하게 됩니다. 이는 가끔씩 연결해 데이터를 가져오는 *특정* 애플리케이션에는 괜찮은 방법일 수 있습니다.

XREAD 명령어는 여러 스트림에서 동시에 읽어오도록 설계되었으며, 스트림에서 마지막으로 받은 항목의 ID만 지정하면 됩니다. 또한 데이터가 없을 경우 데이터가 도착할 때까지 블로킹되도록 요청할 수 있습니다. 리스트의 블로킹 연산과 유사하게 동작하지만, 여기서는 데이터가 스트림에서 소비되지 않으며 여러 클라이언트가 동시에 같은 데이터에 접근할 수 있습니다.

다음은 정석적인 XREAD 호출 예제입니다.

> XREAD BLOCK 5000 STREAMS mystream otherstream $ $

이 명령은 “mystream”과 “otherstream”에서 데이터를 가져오라는 뜻입니다. 사용할 수 있는 데이터가 없으면 5000밀리초 타임아웃으로 클라이언트를 블로킹합니다. STREAMS 옵션 뒤에는 구독하려는 키와 마지막으로 보유한 ID를 지정합니다. 다만 특별한 ID인 “$”는 “현재 스트림에 있는 모든 원소를 이미 가지고 있다고 가정하고, 다음에 도착하는 원소부터 달라”는 의미입니다.

다른 클라이언트에서 다음 명령을 보낸다면,

> XADD otherstream * message "Hi There"

XREAD 쪽에서는 다음과 같은 결과가 나타납니다.

1) 1) "otherstream"
   2) 1) 1) 1506935385635.0
         2) 1) "message"
            2) "Hi There"

데이터를 받은 키와 함께 수신한 데이터를 얻게 됩니다. 다음 호출에서는 아마도 마지막으로 받은 메시지의 ID를 사용할 것입니다.

> XREAD BLOCK 5000 STREAMS mystream otherstream $ 1506935385635.0

이후에도 같은 방식으로 진행됩니다. 다만 이 사용 패턴에서는 클라이언트가 메시지를 처리하는 데 시간이 걸리거나 다른 이유로 인해 매우 오랜 지연 뒤에 다시 연결될 수도 있다는 점에 유의해야 합니다. 그런 경우 그동안 많은 메시지가 쌓여 있을 수 있으므로, XREAD에서 항상 COUNT 옵션을 사용해 클라이언트가 메시지에 압도되지 않고 서버도 단일 클라이언트에 엄청난 양의 메시지를 제공하는 데 과도한 시간을 쓰지 않도록 하는 것이 현명합니다.

크기가 제한된 스트림

지금까지는 좋습니다… 하지만 스트림은 언젠가 오래된 메시지를 제거해야 합니다. 다행히 XADD 명령어의 MAXLEN 옵션을 이용하면 가능합니다.

> XADD mystream MAXLEN 1000000 * field1 value1 field2 value2

이는 스트림에 새 원소를 추가한 뒤 길이가 100만 개를 초과하면, 길이가 다시 100만 개가 되도록 오래된 메시지를 제거하라는 의미입니다. 리스트에서 RPUSH + LTRIM을 사용하는 것과 비슷하지만, 이번에는 이를 위한 메커니즘이 내장되어 있습니다. 다만 위 방식은 새 메시지를 추가할 때마다 스트림 반대편에서 메시지 하나를 제거하는 데 필요한 작업도 함께 수행해야 하므로 어느 정도 CPU 비용이 듭니다. 따라서 MAXLEN의 개수 앞에 “~” 기호를 사용해 *정확히* 100만 개가 아니어도 괜찮고 조금 더 많아도 큰 문제가 되지 않는다는 점을 지정할 수 있습니다.

> XADD mystream MAXLEN ~ 1000000 * foo bar

이렇게 하면 XADD는 전체 노드를 한 번에 제거할 수 있을 때만 메시지를 삭제합니다. 이를 통해 크기가 제한된 스트림을 사용하는 비용이 일반적인 XADD와 비교해 거의 추가 비용이 들지 않게 됩니다.

컨슈머 그룹 (작업 중)

이 기능은 아직 Redis에 구현되지 않았지만 현재 작업이 진행 중인 첫 번째 기능입니다. 또한 Kafka에서 가장 명확하게 영감을 받은 아이디어이기도 하며, 비록 여기서는 상당히 다른 방식으로 구현되었습니다. 핵심은 XREAD에서 클라이언트가 “GROUP <이름>” 옵션을 추가할 수 있다는 점입니다. 그러면 같은 그룹에 속한 모든 클라이언트는 *서로 다른* 메시지를 받게 됩니다. 물론 같은 스트림을 읽는 여러 그룹이 존재할 수도 있으며, 이 경우 모든 그룹은 스트림에 도착하는 동일한 메시지의 복사본을 받지만, 각 그룹 내에서는 메시지가 중복되지 않습니다.

그룹에 대한 확장으로, 그룹을 지정할 때 “RETRY <밀리초>” 옵션을 지정할 수 있게 될 예정입니다. 이 경우 XACK로 처리 확인을 받지 못한 메시지는 지정된 시간이 지난 뒤 다시 전달됩니다. 이는 클라이언트가 메시지를 처리 완료로 표시할 별도의 수단이 없을 때 메시지 전달에 어느 정도 최선을 다하는 신뢰성을 제공합니다. 이 부분 역시 현재 작업 중입니다.

메모리 사용량과 저장/로드 시간

Redis 스트림을 모델링하는 데 사용된 설계 덕분에 메모리 사용량이 현저히 낮습니다. 필드 수와 값, 그리고 그 길이에 따라 달라지지만, 단순한 메시지 기준으로 100MB당 수백만 개의 메시지를 저장할 수 있습니다. 게다가 이 형식은 매우 최소한의 직렬화만 필요하도록 고안되었습니다. 래딕스 트리 노드로 저장되는 listpack 블록은 디스크와 메모리에서 동일한 표현을 가지므로 저장과 읽기가 매우 간단합니다. 예를 들어 Redis는 RDB 파일에서 500만 개 항목을 0.3초 만에 읽어들일 수 있습니다. 이는 스트림의 복제와 영속화를 매우 효율적으로 만듭니다.

가운데에 있는 항목의 삭제도 허용할 계획이며, 이는 부분적으로만 구현되어 있습니다. 전략은 항목 플래그에 삭제 표시를 하고, 전체 항목 수 대비 삭제된 항목 수의 비율이 특정 값에 도달하면 블록을 다시 작성해 가비지를 정리하고, 필요하면 단편화를 피하기 위해 인접한 다른 블록과 합치는 것입니다.

결론 및 출시 예정

Redis 스트림은 올해 말 이전에 Redis 4.0 안정 버전에 포함될 예정입니다. 이 범용 자료구조가 Redis가 그동안 다루기 어려웠던 많은 사용 사례를 포괄하는 데 큰 도움이 될 것이라고 생각합니다. 즉, 기존 자료구조를 창의적으로 남용해 특정 문제를 해결해야 했던 경우를 말합니다. 매우 중요한 사용 사례 중 하나는 시계열이지만, TREAD를 통한 다른 용도의 메시지 스트리밍 역시 fire-and-forget보다 더 높은 신뢰성이 필요한 Pub/Sub 애플리케이션을 대체하거나 완전히 새로운 사용 사례를 위해서도 매우 흥미로울 것이라는 느낌이 듭니다. 지금 당장 여러분의 문제 영역에서 새로운 기능을 평가해 보고 싶다면, Github에서 “streams” 브랜치를 가져와 직접 실행해 보시기 바랍니다. 결국 버그 리포트는 언제나 환영입니다 :-)

영상을 선호하신다면, 스트림을 실시간으로 시연하는 세션은 여기에서 보실 수 있습니다: https://www.youtube.com/watch?v=ELDzy9lCFHQ