kazoo: Using zookeeper api with python

```
pip3 install kazoo
```
https://kazoo.readthedocs.io/en/latest/basic_usage.html
https://kazoo.readthedocs.io/en/latest/api/client.html

### 접속

```
from kazoo.client import KazooClient

zk = KazooClient(hosts='127.0.0.1:2181')
zk.start()
```
호스트에 접속할 때 default port는 2181이다.

### 상태 Listener
zookeeper의 상태가 변하는 것을 알 수 있다.
```
from kazoo.client import KazooState

def my_listener(state):
    if state == KazooState.LOST:
        # Register somewhere that the session was lost
    elif state == KazooState.SUSPENDED:
        # Handle being disconnected from Zookeeper
    else:
        # Handle being connected/reconnected to Zookeeper

zk.add_listener(my_listener)
```

LOST
CONNECTED
SUSPENDED
의 state를 가질 수 있다.

가능한 상태 변화
- LOST -> CONNECTED
- CONNECTED -> SUSPENDED
- CONNECTED -> LOST
- SUSPENDED -> LOST
- SUSPENDED -> CONNECTED

### CRUD

- `zk.create("/my/favorite/node", b"a value")`
`create(path, value='', acl=None, ephemeral=False, sequence=False, makepath=False)`
makepath 옵션이 True가 아니라면 경로가 존재해야만 가능하다.
data는 byte string을 받는다.
Returns: 새로 생성된 노드의 path. string type.
이미 있는 노드면 NodeExistsError
1MB 이상의 크기의 노드를 생성하려 하면 ZookeeperError
- `if zk.exists("/my/favorite"):`
- `data, stat = zk.get("/my/favorite")`
stat에 노드 정보가 있고 data에 데이터가 있다.
- `children = zk.get_children("/my/favorite")`
children의 리스트를 받는다.
- `zk.set("/my/favorite", b"some data")`
노드의 데이터를 업데이트한다. 
- `zk.delete("/my/favorite/node", recursive=True)`
모든 children을 지우려면 recursive 옵션을 넣는다.
- `result = zk.retry(zk.get, "/path/to/node")`
kazoo는 기본적으로 retry을 하지 않기 때문에 retry()를 이용해서 연결 문제에 대해 처리할 수 있다.


### Watchers

watch function을 설정하면 노드나 노드의 children에 변화가 있으면 trigger된다.

아래와 같이 get(), get_children(), exists() 함수에 watch function을 넘길 수 있다.
```
def my_func(event):
    # check to see what the children are now

# Call my_func when the children change
children = zk.get_children("/my/favorite/node", watch=my_func)
```
해당 노드 데이터가 변하거나 노드 자체가 삭제될 때 WatchedEvenet instance로 전달된다.

ChildrenWatch, DataWatch 메소드는 변화가 있을 때마다 불리거나 function이 False를 return할 때까지 불린다.
allow_session_lost가 True가 되면 함수는 세션이 끊겼을 때 더 이상 불리지 않는다.
```
@zk.ChildrenWatch("/my/favorite/node")
def watch_children(children):
    print("Children are now: %s" % children)
# Above function called immediately, and from then on

@zk.DataWatch("/my/favorite")
def watch_node(data, stat):
    print("Version: %s, data: %s" % (stat.version, data.decode("utf-8")))
```



---

 ## 예제

### example

```
(env) [irteam@css1381 kazoo]$ cat zk.py
from kazoo.client import KazooClient
import sys
import time
import datetime

zk = KazooClient(hosts="zk-etcd1.bdp.bdata.ai,zk-etcd2.bdp.bdata.ai,zk-etcd3.bdp.bdata.ai")
zk.start()

print("zk state is")
print(zk.state)


lt = time.localtime(time.time())
current_time = str(lt.tm_year) + str(lt.tm_mon) +str(lt.tm_mday) +str(lt.tm_hour) +str(lt.tm_min) +str(lt.tm_sec)

new_path = zk.create("/c3-assigner/1381", current_time.encode())
print(new_path)

if zk.exists('/c3-assigner'):
    children = zk.get_children('/c3-assigner')
    print(children)

zk.stop()
```

```
(env) [irteam@css1381 kazoo]$ python3 zk.py
zk state is
CONNECTED
/c3-assigner/1381
['1381']
```

```
[zk: localhost:2181(CONNECTED) 2] ls /c3-assigner
[1381]
[zk: localhost:2181(CONNECTED) 3] get /c3-assigner/1381
202093202824
cZxid = 0x1001e057ca
ctime = Thu Sep 03 20:28:24 KST 2020
mZxid = 0x1001e057ca
mtime = Thu Sep 03 20:28:24 KST 2020
pZxid = 0x1001e057ca
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 12
numChildren = 0
```

만든 순서대로가 지켜질까 리스트에? 그러면 list[0]으로 락 확인할 수 있다.




## 분산 lock 구현하기

## Apache document: Locks

https://zookeeper.apache.org/doc/r3.4.6/recipes.html#sc_recipes_Locks

### 구현 방법

1. "_locknode_/lock-" 의 위치에 create() 를 실행한다.
sequence와 ephemeral 플래그가 설정된다.
2. watch flag 없이 getChildren()을 lock node에 실행한다.
3. 1번 단계에서 만들어진 pathname이 가장 낮은 인덱스에 있다면 그 클라이언트는 락을 갖게 되고 프로토콜을 빠져나온다.(작업 수행 후 락을 버리라는 뜻인듯)
4. 클라이언트는 그 lock 디렉토리에서 그 다음으로 낮은 인덱스에 exists() 를 watch flag와 함께 호출한다.
5. exists()가 false를 return하면 2번 단계로 간다. 아니면 알림이 오고나서 2번으로 간다.

lock을 갖고 있던 클라이언트가 lock을 버리는 건 단순히 1번에서 만들어진 노드를 delete하면 된다.

특징이 몇 가지 있다.
- 노드 삭제가 일어나면 오직 하나의 클라이언트에게만 신호가 간다. 왜냐하면 각 노드는 하나의 클라이언트에게만 watch 가 되고 있기 때문이다. 이렇게 함으로써 herd effect를 피할 수 있다.
- polling이나 timeout이 발생하지 않는다.
- lock을 구성함으로써 lock contention, break locks 등을 확인하기가 쉽다.

### 참고 - znode 종류에 대해
https://blog.seulgi.kim/2014/05/zookeeper-1-znode-zookeeper-data.html

- persistent vs ephemeral
life cycle에 관한 설정이다. 
persistent는 직접 삭제하기 전까지는 없어지지 않는다. 
ephemeral은 클라이언트와 서버 사이의 connection이 종료되면 자동으로 사라진다.
- sequence
uniqueness를 보장하기 위한 옵션이다.
sequence 모드로 만들어진 znode는 주어진 읾 뒤에 int 범위의 숫자 10개가 postfix로 붙는다.
이 숫자는 atomic하게 증가하여, 같은 이름으로 만든 znode라도 다른 이름의 znode로 만들어진다.

ephemeral + sequence로 노드를 생성하는 것은 문제가 생길 수 있다. 
생성을 요청하고 connection이 끊어진다면 생성이 성공했는지 실패했는지 응답을 받지 못하고 timeout이 발생한다.
znode가 생성되었는지를 확인하면 성공인지 실패인지 알 수 있는데 노드 이름을 서버에서 정하기 때문에 클라이언트는 성공 여부를 알 수가 없다.
curator 라이브러리에서 이를 위해 protect mode라는 것을 도입했다고 한다.


### shared lock

lock protocol에서 몇 가지만 바꾸면 shared lock을 구현할 수 있다.


Reference

https://kazoo.readthedocs.io/en/latest/basic_usage.html

https://kazoo.readthedocs.io/en/latest/api/client.html


댓글

댓글 쓰기

이 블로그의 인기 게시물

논문 정리 - MapReduce: Simplified Data Processing on Large Clusters

Abstract 맵리듀스는 큰 데이터 셋을 다루는 프로그래밍 모델이다 . 사용자는 map 함수를 통해서 키 / 밸류 쌍을 이용하여 intermediate key/value pair 를 만들고 reduce 함수를 통해서 그 intermediate pair 들을 같은 key 별로 합쳐준다 . 많은 일들이 이런 모델을 통해 표현될 수 있다 . 이 모델은 분산 시스템이나 병렬 시스템에 대한 경험이 부족하더라도 분산 시스템의 자원을 쉽게 사용할 수 있게 해준다 . 1. introduction 데이터가 많아지면서 reasonable time 안에 작업을 끝내기 위해 수백 혹은 수천대의 기계에 분산되어야 했다 . 이 때 어떻게 연산을 병렬화하고 , 데이터를 분산하고 , 실패 작업을 어떻게 처리할지가 이슈가 되었다 . 이런 복잡한 문제들을 해결하기 위해 병렬화의 세세한 부분들은 숨기고 fault tolerance 하며 데이터를 분산시켜주고 로드 밸런싱을 해주는 라이브러리를 통해서 연산을 맵과 리듀스로 간단히 표현하도록 했다 . 2. Programming Model 사용자가 직접 만드는 맵 함수는 입력으로 key/value pairs 를 받아서 intermediate key/value pairs 를 만든다 . 그러면 intermediate key/value pairs 에 대해서 맵리듀스 라이브러리가 같은 key 별로 value 들을 묶어서 리듀스 함수로 보내준다 . 마찬가지로 사용자가 직접 만드는 리듀스 함수는 이 데이터를 받아서 key 별로 묶인 value 들을 합쳐서 표현한다 . 리듀스 함수에 전달되는 intermediate 데이터는 iterator 의 형태로 전달되기 때문에 메모리에 다 안 들어갈 정도로 큰 데이터도 처리할 수 있다 . 예를 들어서 , 어떤 문서에 있는 각 단어들이 몇 번 나왔는지 세는 프로그램을 생각해보자 . 그렇다면 맵 함수는 입력으로 < 문서 이름 , 문서 내용 > 의...
자세한 내용 보기

논문 정리 - The Google File System

The Google File System 1. Introduction 새로운 관점 component failure 은 예외가 아니라 일반적인 상황이다 . 그래서 constant monitoring, error detection, fault tolerance, automatic recovery 가 필요하다 . 파일의 크기가 커졌다 . I/O 작업이나 block 크기 등이 고려되어야 한다 . 원래 있는 데이터에 덮어쓰는 작업보다는 새로운 데이터를 추가한다 . 따라서 appending 이 성능 최적화에 중요해졌다 . 2. Design Overview 2.1 assumptions 시스템은 잘 고장나는 저렴한 장비들로 구성되어 있다 . 시스템은 큰 파일이 있다 . 작은 파일들도 있지만 optimizing 할 때는 고려하지 않는다 . 로드는 주로 큰 스트리밍 읽기와 작은 랜덤 읽기로 구성되어 있다 . 로드는 또한 파일에 데이터를 추가하는 많은 sequential write 가 있다 . 같은 파일에 여러 사용자가 동시에 데이터를 추가하는 경우도 잘 고려해야 한다 . 빠른 반응 속도 보다는 high sustained bandwidth 가 더 중요하다 . 대부분의 타겟 애플리케이션은 데이터 뭉텅이를 빠른 시간 내에 처리하는 것을 중요하게 여긴다 . 2.2 Interface 파일은 hierarchically , 디렉토리에 정렬돼있고 path name 으로 구분된다 . 또한 GFS 는 snapshot 과 record append 기능이 있다 . snapshot 은 파일이나 디렉토리 트리를 복사한다 . record append 는 다수 클라이언트가 데이터를 하나의 파일에 동시에 추가할 수 있도록 해준다 . 각 append 의 atomicity 를 보장하면서 . 2.3 Architecture 하나의 마스터와 여러개의 chunkserver 로 구성되어 있다 . 그리고 다수의...
자세한 내용 보기