zookeeperクラスタのpythonコードテスト

前の記事では、zookeeperのpythonクライアントをインストールする方法について説明しました.次に、ネット上で検索した例です.例えば、アプリケーション環境は次のとおりです.
1.2つ以上のサービスが実行され、同意時間が1つのサービスのみが要求(作業)を受け入れる場合、他のサービスは待機する.
2.要求を受けたサービスが異常に停止した場合、残りの待機サービスから1つのサービスが選出されて要求を受ける.
次に、直接例を挙げると、2つのファイルからなる1.zkclient.py   2.zktest.py
 

# coding: utf-8

# modfied from https://github.com/phunt/zk-smoketest/blob/master/zkclient.py

# zkclient.py



import zookeeper, time, threading

from collections import namedtuple



DEFAULT_TIMEOUT = 30000

VERBOSE = True



ZOO_OPEN_ACL_UNSAFE = {"perms":0x1f, "scheme":"world", "id" :"anyone"}



# Mapping of connection state values to human strings.

STATE_NAME_MAPPING = {

    zookeeper.ASSOCIATING_STATE: "associating",

    zookeeper.AUTH_FAILED_STATE: "auth-failed",

    zookeeper.CONNECTED_STATE: "connected",

    zookeeper.CONNECTING_STATE: "connecting",

    zookeeper.EXPIRED_SESSION_STATE: "expired",

}



# Mapping of event type to human string.

TYPE_NAME_MAPPING = {

    zookeeper.NOTWATCHING_EVENT: "not-watching",

    zookeeper.SESSION_EVENT: "session",

    zookeeper.CREATED_EVENT: "created",

    zookeeper.DELETED_EVENT: "deleted",

    zookeeper.CHANGED_EVENT: "changed",

    zookeeper.CHILD_EVENT: "child", 

}



class ZKClientError(Exception):

    def __init__(self, value):

        self.value = value

    def __str__(self):

        return repr(self.value)



class ClientEvent(namedtuple("ClientEvent", 'type, connection_state, path')):

    """

    A client event is returned when a watch deferred fires. It denotes

    some event on the zookeeper client that the watch was requested on.

    """



    @property

    def type_name(self):

        return TYPE_NAME_MAPPING[self.type]



    @property

    def state_name(self):

        return STATE_NAME_MAPPING[self.connection_state]



    def __repr__(self):

        return  "<ClientEvent %s at %r state: %s>" % (

            self.type_name, self.path, self.state_name)





def watchmethod(func):

    def decorated(handle, atype, state, path):

        event = ClientEvent(atype, state, path)

        return func(event)

    return decorated



class ZKClient(object):

    def __init__(self, servers, timeout=DEFAULT_TIMEOUT):

        self.timeout = timeout

        self.connected = False

        self.conn_cv = threading.Condition( )

        self.handle = -1



        self.conn_cv.acquire()

        if VERBOSE: print("Connecting to %s" % (servers))

        start = time.time()

        self.handle = zookeeper.init(servers, self.connection_watcher, timeout)

        self.conn_cv.wait(timeout/1000)

        self.conn_cv.release()



        if not self.connected:

            raise ZKClientError("Unable to connect to %s" % (servers))



        if VERBOSE:

            print("Connected in %d ms, handle is %d"

                  % (int((time.time() - start) * 1000), self.handle))



    def connection_watcher(self, h, type, state, path):

        self.handle = h

        self.conn_cv.acquire()

        self.connected = True

        self.conn_cv.notifyAll()

        self.conn_cv.release()



    def close(self):

        return zookeeper.close(self.handle)

    

    def create(self, path, data="", flags=0, acl=[ZOO_OPEN_ACL_UNSAFE]):

        start = time.time()

        result = zookeeper.create(self.handle, path, data, acl, flags)

        if VERBOSE:

            print("Node %s created in %d ms"

                  % (path, int((time.time() - start) * 1000)))

        return result



    def delete(self, path, version=-1):

        start = time.time()

        result = zookeeper.delete(self.handle, path, version)

        if VERBOSE:

            print("Node %s deleted in %d ms"

                  % (path, int((time.time() - start) * 1000)))

        return result



    def get(self, path, watcher=None):

        return zookeeper.get(self.handle, path, watcher)



    def exists(self, path, watcher=None):

        return zookeeper.exists(self.handle, path, watcher)



    def set(self, path, data="", version=-1):

        return zookeeper.set(self.handle, path, data, version)



    def set2(self, path, data="", version=-1):

        return zookeeper.set2(self.handle, path, data, version)





    def get_children(self, path, watcher=None):

        return zookeeper.get_children(self.handle, path, watcher)



    def async(self, path = "/"):

        return zookeeper.async(self.handle, path)



    def acreate(self, path, callback, data="", flags=0, acl=[ZOO_OPEN_ACL_UNSAFE]):

        result = zookeeper.acreate(self.handle, path, data, acl, flags, callback)

        return result



    def adelete(self, path, callback, version=-1):

        return zookeeper.adelete(self.handle, path, version, callback)



    def aget(self, path, callback, watcher=None):

        return zookeeper.aget(self.handle, path, watcher, callback)



    def aexists(self, path, callback, watcher=None):

        return zookeeper.aexists(self.handle, path, watcher, callback)



    def aset(self, path, callback, data="", version=-1):

        return zookeeper.aset(self.handle, path, data, version, callback)



watch_count = 0



"""Callable watcher that counts the number of notifications"""

class CountingWatcher(object):

    def __init__(self):

        self.count = 0

        global watch_count

        self.id = watch_count

        watch_count += 1



    def waitForExpected(self, count, maxwait):

        """Wait up to maxwait for the specified count,

        return the count whether or not maxwait reached.



        Arguments:

        - `count`: expected count

        - `maxwait`: max milliseconds to wait

        """

        waited = 0

        while (waited < maxwait):

            if self.count >= count:

                return self.count

            time.sleep(1.0);

            waited += 1000

        return self.count



    def __call__(self, handle, typ, state, path):

        self.count += 1

        if VERBOSE:

            print("handle %d got watch for %s in watcher %d, count %d" %

                  (handle, path, self.id, self.count))



"""Callable watcher that counts the number of notifications

and verifies that the paths are sequential"""

class SequentialCountingWatcher(CountingWatcher):

    def __init__(self, child_path):

        CountingWatcher.__init__(self)

        self.child_path = child_path



    def __call__(self, handle, typ, state, path):

        if not self.child_path(self.count) == path:

            raise ZKClientError("handle %d invalid path order %s" % (handle, path))

        CountingWatcher.__call__(self, handle, typ, state, path)



class Callback(object):

    def __init__(self):

        self.cv = threading.Condition()

        self.callback_flag = False

        self.rc = -1



    def callback(self, handle, rc, handler):

        self.cv.acquire()

        self.callback_flag = True

        self.handle = handle

        self.rc = rc

        handler()

        self.cv.notify()

        self.cv.release()



    def waitForSuccess(self):

        while not self.callback_flag:

            self.cv.wait()

        self.cv.release()



        if not self.callback_flag == True:

            raise ZKClientError("asynchronous operation timed out on handle %d" %

                             (self.handle))

        if not self.rc == zookeeper.OK:

            raise ZKClientError(

                "asynchronous operation failed on handle %d with rc %d" %

                (self.handle, self.rc))





class GetCallback(Callback):

    def __init__(self):

        Callback.__init__(self)



    def __call__(self, handle, rc, value, stat):

        def handler():

            self.value = value

            self.stat = stat

        self.callback(handle, rc, handler)



class SetCallback(Callback):

    def __init__(self):

        Callback.__init__(self)



    def __call__(self, handle, rc, stat):

        def handler():

            self.stat = stat

        self.callback(handle, rc, handler)



class ExistsCallback(SetCallback):

    pass



class CreateCallback(Callback):

    def __init__(self):

        Callback.__init__(self)



    def __call__(self, handle, rc, path):

        def handler():

            self.path = path

        self.callback(handle, rc, handler)



class DeleteCallback(Callback):

    def __init__(self):

        Callback.__init__(self)



    def __call__(self, handle, rc):

        def handler():

            pass

        self.callback(handle, rc, handler)

上のファイルは他の人がカプセル化したzookeeperインタフェースで、下はテストコードで、上のパッケージに依存する必要があります.
 
 

# coding: utf-8

# zktest.py



import logging

from os.path import basename, join



from zkclient import ZKClient, zookeeper, watchmethod



logging.basicConfig(

    level = logging.DEBUG,

    format = "[%(asctime)s] %(levelname)-8s %(message)s"

)



log = logging



class GJZookeeper(object):



    ZK_HOST = "localhost:2181"

    ROOT = "/app"

    WORKERS_PATH = join(ROOT, "workers")

    MASTERS_NUM = 1

    TIMEOUT = 10000



    def __init__(self, verbose = True):

        self.VERBOSE = verbose

        self.masters = []

        self.is_master = False

        self.path = None



        self.zk = ZKClient(self.ZK_HOST, timeout = self.TIMEOUT)

        self.say("login ok!")

        # init

        self.__init_zk()

        # register

        self.register()



    def __init_zk(self):

        """

        create the zookeeper node if not exist

        """

        nodes = (self.ROOT, self.WORKERS_PATH)

        for node in nodes:

            if not self.zk.exists(node):

                try:

                    self.zk.create(node, "")

                except:

                    pass



    @property

    def is_slave(self):

        return not self.is_master



    def register(self):

        """

        register a node for this worker

        """

        self.path = self.zk.create(self.WORKERS_PATH + "/worker", "1", flags=zookeeper.EPHEMERAL | zookeeper.SEQUENCE)

        self.path = basename(self.path)

        self.say("register ok! I'm %s" % self.path)

        # check who is the master

        self.get_master()



    def get_master(self):

        """

        get children, and check who is the smallest child

        """

        @watchmethod

        def watcher(event):

            self.say("child changed, try to get master again.")

            self.get_master()



        children = self.zk.get_children(self.WORKERS_PATH, watcher)

        children.sort()

        self.say("%s's children: %s" % (self.WORKERS_PATH, children))



        # check if I'm master

        self.masters = children[:self.MASTERS_NUM]

        if self.path in self.masters:

            self.is_master = True

            self.say("I've become master!")

        else:

            self.say("%s is masters, I'm slave" % self.masters)





    def say(self, msg):

        """

        print messages to screen

        """

        if self.VERBOSE:

            if self.path:

                if self.is_master:

                    log.info("[ %s(%s) ] %s" % (self.path, "master" , msg))

                else:

                    log.info("[ %s(%s) ] %s" % (self.path, "slave", msg))

            else:

                log.info(msg)



def main():

    gj_zookeeper = GJZookeeper()



if __name__ == "__main__":

    main()

    import time

    time.sleep(1000)

次の2つのマシン(同じマシンでも実行できます)で上記のスクリプトを実行すると、次の情報が表示されます.
 
 

[2013-07-03 14:26:12,192] INFO     login ok!

Node /app/workers/worker created in 1 ms

[2013-07-03 14:26:12,195] INFO     [ worker0000000016(slave) ] register ok! I'm worker0000000016

[2013-07-03 14:26:12,196] INFO     [ worker0000000016(slave) ] /app/workers's children: ['worker0000000016']

[2013-07-03 14:26:12,196] INFO     [ worker0000000016(master) ] I've become master!

および
 
 

[2013-07-03 14:26:58,277] INFO     login ok!

Node /app/workers/worker created in 2 ms

[2013-07-03 14:26:58,281] INFO     [ worker0000000017(slave) ] register ok! I'm worker0000000017

[2013-07-03 14:26:58,282] INFO     [ worker0000000017(slave) ] /app/workers's children: ['worker0000000016', 'worker0000000017']

[2013-07-03 14:26:58,282] INFO     [ worker0000000017(slave) ] ['worker0000000016'] is masters, I'm slave

まず最初の機械がmasterを作りました.
 
次に、最初のプロセスをオフにします.2台目はtimeout(10 s)時間後、次の情報が表示されます.
 

[2013-07-03 14:28:02,204] INFO     [ worker0000000017(slave) ] child changed, try to get master again.

[2013-07-03 14:28:02,205] INFO     [ worker0000000017(slave) ] /app/workers's children: ['worker0000000017']

[2013-07-03 14:28:02,206] INFO     [ worker0000000017(master) ] I've become master!

明らかに、2台目の機械のプログラムはmasterになりました.これが私たちが実現したい機能です.
 
もちろん複数のプロセスを開くこともでき、プログラムの中でmasterの数を選択することができます.同時に、timeout時間を小さくすることをお勧めします.私たちが先に環境に行くのは2 sです.
その後、zookeeperの他のアプリケーション環境の例が更新されます.