做迁移的一些体会

发表于 2016-10-13 更新于 2024-10-27 分类于模式，架构模式阅读次数：本文字数： 4.1k 阅读时长 ≈ 4 分钟

难点：

最近公司做用户数据迁移，4TB的量，租借了20Gb的带宽，完整迁移需要25天，给的迁移期只有30天，整个迁移过程的容错性和可靠性要非常强

另外由于数据的特殊性，用户数据的错乱比起迁移失败等等无疑是更严重的问题，所以迁移过程要保证强独立性

然而迁移过程的开发时间只有5天，测试时间5天，非常仓促（实际上线后确实遇到不少问题，又花了5天才全部解决）

分析和解决：

同事给的建议是迁移工具尽量的简单。迁移是根据用户ID列表来迁移的，因此可以遵循简单的生产者消费者模型

生产者作为用户ID分发程序，等待消费者拉取

消费者拉取消息，并且在完成任务后，向生产者汇报成功或者失败（失败后生产者重新生产该ID）

消费者是简单的单进程逻辑，一次只处理一个用户

这样的模型是最简单可靠的，每次循环中，对不同的用户ID，new出不同的task结构，并且不同的迁移程序是独立的进程，能保证迁移过程不会错乱用户数据（强独立性）

容错性是由每个用户ID的可重入性来保证的，对于用户的每个目录和每个文件来说，先扫描是否已经迁移过，迁移过的目录文件跳过，这样重新进行消费就能完成该用户ID的任务

可靠性是由crontab来保证的，每个进程挂掉以后由crontab拉起

具体踩过的坑

监控脚本

#!/bin/sh

export PATH="/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/usr/games:/usr/local/games"

nums="p01 p02 p03 p04 p05 p06 p07 p08 p09 p10 p11 p12 p13 p14 p15 p16 p17 p18 p19 p20 p21 p22 p23 p24 p25 p26 p27 p28 p29 p30 p31 p32 p33 p34 p35 p36 p37 p38 p39 p40 p41 p42 p43 p44 p45 p46 p47 p48 p49 p50 p51 p52 p53 p54 p55 p56 p57 p58 p59 p60 p61 p62 p63 p64 p65 p66 p67"
for num in $nums
do

        count=`pgrep -f "/root/photo_migrate/main -alsologtostderr -num $num"|wc -l`

        echo "process $num,count $count"

        if [ $count -lt 1 ];then
                time=`date`
                echo "process $num,restart at $time" >> /root/photo_migrate/restart.log
                nohup /root/photo_migrate/main -alsologtostderr -num $num 2>&1 >> /root/photo_migrate/run_$num.log & 
        fi

done

该脚本能完成监控机器上所有迁移进程的任务，然而由crontab启动以后，会导致没有目录输出，该问题目前仍未解决，目前是直接命令启动的

分布式SSH命令

对于各种SSH工具来说，同时管理多个节点，命令的执行是并行的，节点数量过多后会很不方便，SSH的并发能力是有限的。

这一次迁移过程使用了50个节点，使用传统SSH来执行命令，特别是用SCP批量的从某台机器复制文件，导致SSH RESET的情况发生。

fabric这个工具能解决这个问题，这是一个远程串行执行命令的python工具

脚本非常简单

from fabric.api import *

def putfile():
        put("localfile", "/root/photo_migrate/localfile")
def start():
        remote_dir = '/root/photo_migrate/'
        with cd(remote_dir):
                run("./monitor.sh")

启动也很简单，-H后面跟iplist，每个ip逗号分割，-u后是帐号，-p后跟密码

1	fab start -Hip1,ip2,... -uroot -ppassword

client超时问题

在实际测试下载文件的时候，发现超过一定大小的文件就会超时

http.Client的Timeout字段设置的30s，已经足够大了，为什么会出现这个问题呢

看了下Timeout的注释

60     // Timeout specifies a time limit for requests made by this                 
61     // Client. The timeout includes connection time, any                        
62     // redirects, and reading the response body. The timer remains              
63     // running after Get, Head, Post, or Do return and will                     
64     // interrupt reading of the Response.Body.                                  
65     //                                                                          
66     // A Timeout of zero means no timeout.                                      
67     //                                                                          
68     // The Client's Transport must support the CancelRequest                    
69     // method or Client will return errors when attempting to make              
70     // a request with Get, Head, Post, or Do. Client's default                  
71     // Transport (DefaultTransport) supports CancelRequest.

这里说道，Timeout字段是包括连接时间，重定向时间，和读取body时间的

当下载文件时，不断读取body内容，计时器会不断计时，这个Timeout是个很简单的timeout，完全不符合我的需求

仔细看了下资料，在http.Client的Transport字段可以设置connect timeout,ResponseHeaderTimeout,read write timeout

首先封装一个Request

type Request struct {
    ...
	ConnectTimeout        time.Duration
	RWTimeout             time.Duration
	ResponseHeaderTimeout time.Duration
    ...
}
```  

Transport的Dial参数是一个函数，在这里可以用net.DialTimeout来设置连接超时时间

这个函数的返回值是一个net.Conn接口，对TCPConn做一个封装，对每次调用read和write都加一个超时计时器，这样的超时级别就足堪大用了

```c  
transport = &http.Transport{
		Dial: func(netw, addr string) (net.Conn, error) {
			conn, err := net.DialTimeout(netw, addr, r.ConnectTimeout)
			if err != nil {
				return nil, err
			}
			if r.RWTimeout > 0 {
				return &rwTimeoutConn{
					TCPConn:   conn.(*net.TCPConn),
					rwTimeout: r.RWTimeout,
				}, nil
			} else {
				return conn, nil
			}
		},
		ResponseHeaderTimeout: r.ResponseHeaderTimeout,
		MaxIdleConnsPerHost:   2000,
	}

type rwTimeoutConn struct {
	*net.TCPConn
	rwTimeout time.Duration
}

func (this *rwTimeoutConn) Read(b []byte) (int, error) {
	err := this.TCPConn.SetDeadline(time.Now().Add(this.rwTimeout))
	if err != nil {
		return 0, err
	}
	return this.TCPConn.Read(b)
}

func (this *rwTimeoutConn) Write(b []byte) (int, error) {
	err := this.TCPConn.SetDeadline(time.Now().Add(this.rwTimeout))
	if err != nil {
		return 0, err
	}
	return this.TCPConn.Write(b)
}
```  

另外Transport的ResponseHeaderTimeout也不要忘记设置

110 // ResponseHeaderTimeout, if non-zero, specifies the amount of
111 // time to wait for a server's response headers after fully
112 // writing the request (including its body, if any). This
113 // time does not include the time to read the response body.
```

这个超时从完整发出请求后计时，一段时间收不到服务端的response就超时（不包含读的时间）