集羣間文件併發同步

條件:

    有一臺跳板機, 能夠ssh到全部機器

    知道其餘機器的用戶名和密碼

思路:

    1. 在跳板機先簡單for循環把文件從本地拷貝到遠程10臺機器( 由於咱們設置了併發Pool(10))

    2. 而後 調用 do_peer_copy 併發10臺同時拷貝

#!/usr/bin/env python
# -*- coding:utf-8 -*-

import logging
import multiprocessing

import invoke
import redis
from fabric import Connection

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
fh = logging.FileHandler('distributeCopy.log')
fh.setLevel(logging.INFO)
ch = logging.StreamHandler()
ch.setLevel(logging.ERROR)
formatter = logging.Formatter(fmt='%(asctime)s %(processName)s-%(threadName)s - %(name)s - %(levelname)s - %(message)s',
                              datefmt='%Y-%m-%d %H:%M:%S')
fh.setFormatter(formatter)
ch.setFormatter(formatter)
logger.addHandler(fh)
logger.addHandler(ch)

passwd = 'ssh_password'
def get_server_ip():
    """
    獲取全部機器列表
    """
    pass


def get_remote_model():
    remote_path = '/data1/online/model/final_model/model'
    local_path = '/data1/rsync_data/ctr_model_webapi'
    invoke.run('rsync -a 192.168.10.100:%s %s' % (remote_path, local_path))


def do_local_copy(dst_ip, dst_path, queue):
    """
    腳本運行在跳板機上(能夠免密碼登陸其它機器的), 由於設置了Pool(10), 因此事先須要先準備10臺已經同步好了文件
    """
    src_path = '/data1/rsync_data/ctr_model_webapi/model'
    result = Connection(dst_ip).put(src_path, remote=dst_path)
    queue.put(dst_ip)
    logger.info('do_local_copy %s: %s', dst_ip, result)


def do_peer_copy(args):
    """
    從跳板機ssh到server-Y, 在server-Y上 rsync 文件
    """
    dst_ip = args[0]
    queue = args[1]
    print 'before qsize: %s' % queue.qsize()
    src_ip = queue.get()
    cmd = """sshpass -p "%s" rsync -e 'ssh -o "StrictHostKeyChecking no"' -a username@%s:%s %s""" % \
          (passwd, src_ip, '/tmp/model', '/tmp')
    result = Connection(dst_ip).run(cmd)
    queue.put(dst_ip)
    print 'end qsize: %s' % queue.qsize()
    logger.info('do_peer_copy %s => %s, %s', src_ip, dst_ip, result)
    #return result 此處不要return任何東西 由於會觸發python的一個issue https://github.com/joblib/joblib/issues/818#issuecomment-445865581


if __name__ == '__main__':
    manager = multiprocessing.Manager()
    q = manager.Queue()
    pool = multiprocessing.Pool(10)

    ips = get_server_ip()

    for ip in ips[:10]:
        do_local_copy(ip, '/tmp', q)

    pool.map(do_peer_copy, [(ip, q) for ip in ips[10:]])
    pool.close()
    pool.join()
    logger.info('finished')