[Redis] All keys in the pipeline should belong to the same slots allocation group 에러 해결하기

상황

ioredis 클라이언트를 사용하여, pipeline 기능을 사용하여 Redis Cluster에 Key와 Value를 저장하려고 했는데, 다음과 같은 에러가 발생했다.

 

Error: All keys in the pipeline should belong to the same slots allocation group

 

Redis Cluster를 사용하는 경우, MSET 및 MGET과 같은 멀티 키 명령을 사용할 수 없으며, ioredis 클라이언트에서 제공하는 pipeline의 경우에도 사용할 수 없다. 그래서 위와 같은 에러가 발생하게 되었다.

해결

pipeline을 사용하려면, 같은 슬롯인 경우만 가능하다는 것이다. 그래서 생각한 방법이 10개의 Key를 저장한다고 가정한 상황에서 3개의 슬롯이 존재하고, 1번 슬롯에는 3개, 2번 슬롯에는 5개, 3번 슬롯에는 2개의 데이터가 저장되어야 하는 경우, 3개의 슬롯에 분산되어 저장할 수 있는 로직을 Api 단에서 직접 구현하여 처리하는 방법을 생각했다.

 

Cluster 정보 가져오기

 

아래의 코드를 통해 Cluster 슬롯에 대한 정보들을 확인할 수 있다.

const slots = await this.redisClient.cluster('slots')

[ 
  // Slot1 정보
  [ 
    0,
    5461,
    [ 
      "Primary-1 Host",
      6379,
      "xxxxxxx" 
    ],
    [ 
      "Replica1-1 Host",
      6379,
      "xxxxxxx"
    ] 
  ],
  // Slot2 정보
  [ 
    5462,
    10922,
    [ 
      "Primary-2 Host",
      6379,
      "xxxxxxx" 
    ],
    [ 
      "Replica1-2 Host",
      6379,
      "xxxxxxx"
    ]  
  ],
  // Slot3 정보  
  [ 
    10923,
    16383,
    [ 
      "Primary-3 Host",
      6379,
      "xxxxxxx" 
    ],
    [ 
      "Replica1-3 Host",
      6379,
      "xxxxxxx"
    ] 
  ]
]

Primary 노드의 Host 및 Slot 정보 추출하기

 

위의 슬롯 정보들을 통해 Primary 노드의 Host를 추출하고, startSlot, endSlot과 같은 슬롯의 범위를 추출한다.

const slots = await this.redisClient.cluster('slots')

const keysAndValuesOfShard = slots.reduce((acc, slotInfos) => {
    const primaryHost = slotInfos[2][0] // Primary Node의 Host를 가져온다.
    const startSlot = slotInfos[0] // Primary Node의 Start Slot을 가져온다.
    const endSlot = slotInfos[1] // Primary Node의 End Slot을 가져온다.
    
    acc[primaryHost] = // Key, Value를 분리한 데이터를 담는다.
    
    return acc
}, {})

요청받은 Key들을 Cluster-Key-Slot 라이브러리를 통해 분리하기

 

요청받은 Key들이 어떤 해시 슬롯에 저장되는지를 미리 파악하는 방법으로 cluster-key-slot의 라이브러리를 활용해서 CRC16 해시 알고리즘을 통해 미리 분리하는 방법을 택했다.

import calculateSlot from 'cluster-key-slot'

// 저장할 Key와 Value 리스트
const keysAndValues = [["key1", "value1"], ["key2", "value2"], ["key3", "value3"]]

const slots = await this.redisClient.cluster('slots')

// 각각의 슬롯에 대한 Key와 Value들을 분리한 데이터
const keysAndValuesOfShard = slots.reduce((acc, slotInfos) => {
    const primaryHost = slotInfos[2][0] // Primary Node의 Host를 가져온다.
    const startSlot = slotInfos[0] // Primary Node의 Start Slot을 가져온다.
    const endSlot = slotInfos[1] // Primary Node의 End Slot을 가져온다.
    
    // Key, Value 리스트를 각각의 PrimaryHost를 기준으로 분리한다. 
    acc[primaryHost] = keysAndValues.reduce((innerAcc, [key, value]) => {
        // cluster-key-slot의 CRC16 알고리즘을 통해 Key가 저장될 슬롯을 미리 구한다.
        const slot = calculateSlot(key)
        
        // 해당 Slot 범위인 경우
        if (startSlot <= slot && slot <= endSlot) {
            innerAcc.push([key, value])
        }

        return innerAcc
    }, [])
    
    return acc
}, {})

각각의 Primary 노드에 분리된 데이터를 Pipeline을 통해 저장하기

 

PrimaryHost를 기준으로 분리한 Key와 Value들을 pipeline을 통해 저장할 수 있는데, 가장 먼저 Primary 노드의 클라이언트를 호출하고, pipeline을 구성해서 한 번에 저장한다.

const keysAndValuesOfShard = // Primary 노드 Host를 기준으로 분리된 Key와 Value들 (위의 로직 참고)

// Primary 노드 클라이언트를 호출한다.
const primaryNodes = await this.redisClient.nodes('master')

// Primary 노드 클라이언트에 대해서 Loop를 돌면서 저장한다.
primaryNodes.forEach((primaryNode) => {
    const { host } = primaryNode.options // Primary 노드의 Host 정보
    const pipeline = primaryNode.pipeline() // Primary 노드 pipeline 기능

    // host에 해당되는 Key와 Value 리스트를 Loop를 돌면서 set 명령을 전달한다.
    keysAndValuesOfShard[host].forEach(([key, value]) =>
        pipeline.set(key, Buffer.from(JSON.stringify(value)), 'EX', expireTime),
    )

    // exec() 메소드를 통해 실제로 명령이 처리된다.
    pipeline.exec()
})

정리

지금까지 내용을 간단하게 정리해보자면, 다음과 같다.

 

1. Primary 노드의 Host와 Slots 정보들을 추출한다.
2. 요청받은 Key와 Value 리스트에 대해서 CRC16 해시 알고리즘을 통해 Key가 저장될 해시 슬롯을 미리 구하고, Primary 노드의 Host를 기준으로 Key와 Value 리스트를 분리한다.
3. 분리된 Key와 Value 리스트를 각각의 Primary 노드 Pipeline 기능으로 한 번에 저장한다.