zookeeper应用场景之配置文件同步

zookeeper应用场景有关于分布式集群配置文件同步问题的描述,本文介绍如何把zk应用到配置文件分发的场景。

假设有三个角色

  • trigger:发布最新的配置文件数据,发送指令和数据给zk_agent,实现是下面的trigger.py
  • zk_agent:接收来自trigger.py触发的指令和数据,并且把数据更新到zk service上,从而触发zk_app来获取最新的配置数据,实现是下面的zk_agent.py
  • zk_app:部署在每台worker上的注册监听zk中配置文件所在znode的变化,从而获取最新的配置文件,应用到worker中,实现是下面的zk_app.py

    image

配置文件同步到:zk_agent

实现逻辑:
* 初始化连接到zk service,首先竞选出master
* master create一个配置文件管理的PERSISTENT类型的znode,比如是/Applications/NginxConf,
* 启动一个线程,专门接收trigger发送的指令,一收到指令,就create一个"conf-"开头的SEQUENCE|PERSISTENT类型的节点,指定znode数据内容是从trigger收到的数据
* 第一次刚更新会创建节点/Applications/NginxConf/conf-0000000000,以后每次更新新的配置文件编号就会增大。

vim zk_agent.py

#!/usr/bin/env python2.7
# -*- coding: UTF-8 -*-
 
# author : firefoxbug
# E-Mail : wanghuafire@gmail.com
# Blog   : www.firefoxbug.net

import logging
from os.path import basename, join

from zkclient import ZKClient, zookeeper, watchmethod

import os
import sys
import threading
import signal
import time

logging.basicConfig(
    level = logging.DEBUG,
    format = "[%(asctime)s] %(levelname)-8s %(message)s"
)

log = logging

class GJZookeeper(object):

    ZK_HOST = "localhost:2181"
    ROOT = "/Roles"
    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.APP_ROOT = "/Applications"
        self.APP_CONF = join(self.APP_ROOT,"NginxConf")

        self.zk = ZKClient(self.ZK_HOST, timeout = self.TIMEOUT)
        self.say("login zookeeper successfully!")
        # init
        self.create_roles_znode()
        # register
        self.register()

    def create_roles_znode(self):
        """
        create the zookeeper node if not exist
        |-Roles
             |-workers
        """
        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,znode type : EPHEMERAL | SEQUENCE
        |-Roles
             |-workers
                     |-worker000000000x         ==>>master
                     |-worker000000000x+1       ==>>worker
                     ....
        """
        self.path = self.zk.create(self.WORKERS_PATH + "/worker", "1", flags=zookeeper.EPHEMERAL | zookeeper.SEQUENCE)
        self.path = basename(self.path)
        self.say("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!")
            self.create_app_znode()
        else:
            self.say("%s is masters, I'm slave" % self.masters)

    def create_app_znode(self):
        """
        create the zookeeper node if not exist
        |-Applications
                    |-NginxConf
        """
        nodes = (self.APP_ROOT, self.APP_CONF)
        for node in nodes: 
            if not self.zk.exists(node):
                try:
                    self.say("Create znode [%s] ..."%(node))
                    self.zk.create(node, "")
                except:
                    pass

    def create_conf_znode(self,data):
        """
        create the zookeeper node's children if not exist,contents is conf data
        |-Applications
                    |-NginxConf
                            |-item-000000000x => data
        """
        self.child_node = join(self.APP_CONF,"conf-")
        path = self.zk.create(self.child_node,data, flags=zookeeper.SEQUENCE|zookeeper.PERSISTENT)
        self.say("create znode %s"%path)

    def say(self, msg):
        """
        print messages to screen
        """
        if self.VERBOSE:
            if self.path:
                log.info("[ %s(%s) ] %s" % (self.path, "master" if self.is_master else "slave", msg))
            else:
                log.info(msg)
        
class Watcher:
    def __init__(self):
        """ Creates a child thread, which returns.  The parent
            thread waits for a KeyboardInterrupt and then kills
            the child thread.
        """
        self.child = os.fork()
        if self.child == 0:
            return
        else:
            self.watch()

    def watch(self):
        try:
            os.wait()
        except KeyboardInterrupt:
            print ' exit...'
            self.kill()
        sys.exit()

    def kill(self):
        try:
            os.kill(self.child, signal.SIGKILL)
        except OSError:
            pass


def start_zk_worker():
    """
        连接到zookeeper执行初始化
    """
    gj_zookeeper = GJZookeeper()
    th1 = threading.Thread(target = start_agent_worker, name = "thread_1", args = (gj_zookeeper,))

    th1.start()

    th1.join()

def start_agent_worker(gj_zookeeper):
    """
        监听配置文件变更信息,解析指令
    """
    import socket

    address = ('', 8877)
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # s = socket.socket()
    s.bind(address)
    s.listen(5)
    print "listen on port 8877 ..."

    while True:
        ss, addr = s.accept()
        print "receive connetcion from " ,addr
        content = ""
        while True:
            try :
                data = ss.recv(512)
                if not data:
                    print "close connetcion " ,addr
                    ss.close()
                    break
                content = content + data
                print "receive message from %s : %s"%(addr[0],data)
            except Exception,e:
                print "receive error from %s : %s"%(addr[0],str(e))
                ss.close()
                break   
        parse_trigger_package(content,gj_zookeeper)

    s.close()

def parse_trigger_package(data,gj_zookeeper):
    try:
        cmd = data.split('|')[0]
        content = data.split('|')[1]
    except Exception, e:
        print "ERROR :",str(e)
        return 
    if cmd == "ADD":
        gj_zookeeper.create_conf_znode(content)
    else:
        pass


def main():
    Watcher()
    start_zk_worker()

if __name__ == "__main__":
    main()
    import time
    time.sleep(20000)

配置文件接收应用:zk_app

zk_app逻辑如下,它工作在每台worker上

  • 初始化连接到zk service
  • 获取所有/Applications/NginxConf的znode列表并且设置Watcher
  • 找到列表中最大的znode节点,记录它的编号,然后getData获取它的数据,这里就拿到了最新的配置信息
  • 每次又Watcher被触发,就获取列表中编号最大的节点,然后把上一次保存的编号和它比较,一旦又更新就重新获取数据,应用到worker,如果相同就不需要到zk上获取数据。

vim zk_app.py

#!/usr/bin/env python2.7
# -*- coding: UTF-8 -*-

# author : firefoxbug
# E-Mail : wanghuafire@gmail.com
# Blog   : www.firefoxbug.net

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"
	TIMEOUT = 10000

	def __init__(self, verbose = True):
		self.VERBOSE = verbose
		self.masters = []
		self.is_master = False
		self.path = None

		self.APP_ROOT = "/Applications"
		self.APP_CONF = join(self.APP_ROOT,"NginxConf")
		self.cur_id = "conf-"
		self.zk = ZKClient(self.ZK_HOST, timeout = self.TIMEOUT)
		self.say("login ok!")
		# register
		self.add_children_watch_on()

	def add_children_watch_on(self):
		"""
		get children, and check who is the smallest child
		"""
		@watchmethod
		def watcher(event):
			self.say("child changed, try to get master again.")
			self.add_children_watch_on()

		children = self.zk.get_children(self.APP_CONF, watcher)
		children.sort()
		self.max_id = children[-1]
		self.say("%s's children: %s" % (self.APP_CONF, children))
		if cmp(self.max_id,self.cur_id) > 0:
			self.cur_id = self.max_id
			self.say("max zid in zookeeper %s is %s"%(self.APP_CONF,self.max_id))
			znode = join(self.APP_CONF,self.max_id)
			value = self.zk.get(znode)
			print value

	def say(self, msg):
		"""
		print messages to screen
		"""
		if self.VERBOSE:
			if self.path:
				log.info("[ %s(%s) ] %s" % (self.path, "master" if self.is_master else "slave", msg))
			else:
				log.info(msg)

def main():
	gj_zookeeper = GJZookeeper()

if __name__ == "__main__":
	main()
	import time
	time.sleep(20000)

配置文件发送:trigger

实现的逻辑很简单,就是模拟给zk_agent发送数据包,格式如下

指令|数据
ADD|helloworld

vim trigger.py

#!/usr/bin/python
import socket  
import sys

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  
sock.connect(('127.0.0.1', 8877))
str = sys.argv[1] + '|' + sys.argv[2]
sock.send(str)  
sock.close()

运行

$ python zk_agent.py
$ python zk_app.py
$ python trigger.py

标签:Zookeeper