etcd网络层(五)——Transporter接口实现

待我称王封你为后i 2022-01-27 01:25 661阅读 0赞

### 一、Transport结构体 ###

type Transport struct {
    	Logger *zap.Logger
    
    	DialTimeout time.Duration // maximum duration before timing out dial of the request，dial的超时时间
    	// DialRetryFrequency defines the frequency of streamReader dial retrial attempts;
    	// a distinct rate limiter is created per every peer (default value: 10 events/sec)
    	DialRetryFrequency rate.Limit   //定义streamReader dial 重试尝试的频率；每一个peer都有不同频率的限制
    
    	TLSInfo transport.TLSInfo // TLS information used when creating connection  tls信息，创建连接时使用
    
    	ID          types.ID   // local member ID  当前节点自己的ID
    	URLs        types.URLs // local peer URLs  当前节点与集群中其他节点交互时使用的URL地址
    	ClusterID   types.ID   // raft cluster ID for request validation  当前节点所在的集群的ID
    	/*
    	Raft是一个接口，其实现的底层封装了前面介绍的etcd-raft模块，当rafthttp.Transport收到消息之后，会将其交给Raft实例进行交互
    	*/
    	Raft        Raft       // raft state machine, to which the Transport forwards received messages and reports status
    	Snapshotter *snap.Snapshotter  //负责管理快照文件
    	
    	ServerStats *stats.ServerStats // used to record general transportation statistics   用于统计一般的transportation统计
    	// used to record transportation statistics with followers when
    	// performing as leader in raft protocol
    	LeaderStats *stats.LeaderStats    //raft协议中的leader节点统计followers节点的 transportation 状态
    	// ErrorC is used to report detected critical errors, e.g.,
    	// the member has been permanently removed from the cluster
    	// When an error is received from ErrorC, user should stop raft state
    	// machine and thus stop the Transport.
    	ErrorC chan error
    
    	streamRt   http.RoundTripper // roundTripper used by streams   Stream消息通道中使用http.RoundTripper实例,HTTP长连接
    	pipelineRt http.RoundTripper // roundTripper used by pipelines  Pipeline消息通道中使用的http.RoundTripper实例,传输完成后会立即关闭连接，传输数据量较大、发送频率较低的消息，如MsgSnap消息
    
    	mu      sync.RWMutex         // protect the remote and peer map
    	//remote中只封装了pipeline实例，remote主要负责发送快照数据，帮助新加入的节点快速追上其他节点的数据
    	remotes map[types.ID]*remote // remotes map that helps newly joined member to catch up
    	/*
    	Peer接口是当前节点对集群中其他节点的抽象表示。对于当前节点来说，集群中其他节点在本地都会有一个Peer实例与之对应，
    	peers字段维护了节点ID到对应Peer实例之间的映射关系
    	*/
    	peers   map[types.ID]Peer    // peers map
    	//用于探测Pipeline消息通道是否可用。
    	prober probing.Prober
    }

Transport实现了Transporter接口。它提供了向对等端发送raft消息的功能，并且接受其他节点的消息。

ID记录当前节点的ID

ClusterID记录当前节点的集群ID

Raft接口，用于和raft模块进行交互，etcd-raft前面已经介绍。

streamRt和pipelineRt 底层都是http.RoundTripper的默认实现http.Transport，完成HTTP事务。

remote remote中只封装了pipeline实例，remote主要负责发送快照数据，帮助新加入的节点快速追上其他节点的数据

peers 记录当前节点和集群中其他节点的对应关系(网络层)，Peer接口是当前节点对集群中其他节点的抽象表示。对于当前节点来说，集群中其他节点在本地都会有一个Peer实例与之对应，peers字段维护了节点ID到对应Peer实例之间的映射关系。

### 二、Transport初始化 ###

Transport的初始化过程在Start方法中实现

func (t *Transport) Start() error {
    	var err error
    	
    	t.streamRt, err = newStreamRoundTripper(t.TLSInfo, t.DialTimeout)
    	if err != nil {
    		return err
    	}
    
    	
    	t.pipelineRt, err = NewRoundTripper(t.TLSInfo, t.DialTimeout)
    	if err != nil {
    		return err
    	}
    
    	/*
    	初始化remotes和peers两个个map字段
    	*/
    	t.remotes = make(map[types.ID]*remote)
    	t.peers = make(map[types.ID]Peer)
    	//该prober实例主要用于探测Pipeline消息通道是否可用
    	t.prober = probing.NewProber(t.pipelineRt)
    
    	// If client didn't provide dial retry frequency, use the default
    	// (100ms backoff between attempts to create a new stream),
    	// so it doesn't bring too much overhead when retry.
    	if t.DialRetryFrequency == 0 {
    		t.DialRetryFrequency = rate.Every(100 * time.Millisecond)
    	}
    	return nil
    }

streamRt创建Stream消息通道使用的http.RoundTripper实例，底层实际上是创建http.Transport实例，这里需要注意几个参数：创建连接的超时时间（根据配置指定）、读写请求的超时时间（默认5s）、keepAlive时间（默认30s）

pipelineRt创建Pipeline消息通道使用的http.RoundTripper实例,与StreamRt不同的是，读写请求的超时时间设置成了永不超时

流程大致如下：

1.初始化Stream消息通道streamRt

2.初始化Pipeline消息通道实例pipelineRt

3.初始化remotes和peers两个个map字段

4.初始化prober实例

5.初始化DialRetryFrequency

### 三、Handler方法 ###

http.Handler封装了处理客户端的请求逻辑及创建返回响应的逻辑

type Handler interface {
    	ServeHTTP(ResponseWriter, *Request)
    }

ServeHTTP(ResponseWriter, \*Request)是用于处理客户端请求及根据业务逻辑创建返回的响应信息。实现Handler接口需要实现ServeHTTP函数。

func (t *Transport) Handler() http.Handler {
    	pipelineHandler := newPipelineHandler(t, t.Raft, t.ClusterID)
    	streamHandler := newStreamHandler(t, t, t.Raft, t.ID, t.ClusterID)
    	snapHandler := newSnapshotHandler(t, t.Raft, t.Snapshotter, t.ClusterID)
    	mux := http.NewServeMux()
    	mux.Handle(RaftPrefix, pipelineHandler)
    	mux.Handle(RaftStreamPrefix+"/", streamHandler)
    	mux.Handle(RaftSnapshotPrefix, snapHandler)
    	mux.Handle(ProbingPrefix, probing.NewHandler())
    	return mux
    }

创建pipelineHandler、streamHandler 、snapHandler 三个实例，这三个实例都实现了Handler接口

mux是多路复用器，ServeMux主要通过m字段(map\[string\]muxEntry)存储URL和Handler实例之间的映射关系，设置URL和Handler之间的对应关系。

url为 /raft，则用pipelineHandler处理收到的快照数据

url为/raft/stream/开头的，都用streamHandler处理收到的消息

url为/raft/snapshot,则使用snapHandler处理收到的快照数据(应该是v3版本使用，之前的版本使用pipelineHandler处理收到的快照)。

probing是用于探活的，这里就省略了。

接下来我们主要看下pipelineHandler、streamHandler 、snapHandler是如何实现Handler接口的。

**pipelineHandler的ServeHTTP方法实现**

func (h *pipelineHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    	//进行请求数据检验，如请求的Method是否为POST，检测集群中ID是否合法
    	if r.Method != "POST" {
    		w.Header().Set("Allow", "POST")
    		http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
    		return
    	}
    
    	w.Header().Set("X-Etcd-Cluster-ID", h.cid.String())
    
    	if err := checkClusterCompatibilityFromHeader(h.lg, h.localID, r.Header, h.cid); err != nil {
    		http.Error(w, err.Error(), http.StatusPreconditionFailed)
    		return
    	}
    
    	addRemoteFromRequest(h.tr, r)
    
    	// Limit the data size that could be read from the request body, which ensures that read from
    	// connection will not time out accidentally due to possible blocking in underlying implementation.
    	/*
    	限制每次从底层连接读取的字节数上线，默认是64KB，因为快照数据可能非常大，为了防止读取超时，
    	只能每次读取一部分数据到缓冲区中，最后将全部数据拼接起来，得到完整的快照数据
    	*/
    	limitedr := pioutil.NewLimitedBufferReader(r.Body, connReadLimitByte)
    	b, err := ioutil.ReadAll(limitedr)   //读取HTTP请求的Body的全部内容
    	if err != nil {
    		if h.lg != nil {
    			h.lg.Warn(
    				"failed to read Raft message",
    				zap.String("local-member-id", h.localID.String()),
    				zap.Error(err),
    			)
    		} else {
    			plog.Errorf("failed to read raft message (%v)", err)
    		}
    		http.Error(w, "error reading raft message", http.StatusBadRequest)
    		recvFailures.WithLabelValues(r.RemoteAddr).Inc()
    		return
    	}
    
    	var m raftpb.Message
    	if err := m.Unmarshal(b); err != nil { //反序列化得到raftpb.Message实例
    		if h.lg != nil {
    			h.lg.Warn(
    				"failed to unmarshal Raft message",
    				zap.String("local-member-id", h.localID.String()),
    				zap.Error(err),
    			)
    		} else {
    			plog.Errorf("failed to unmarshal raft message (%v)", err)
    		}
    		http.Error(w, "error unmarshaling raft message", http.StatusBadRequest)
    		recvFailures.WithLabelValues(r.RemoteAddr).Inc()
    		return
    	}
    
    	receivedBytes.WithLabelValues(types.ID(m.From).String()).Add(float64(len(b)))
    
    	//将读到的消息实例交给底层的Raft状态机进行处理
    	if err := h.r.Process(context.TODO(), m); err != nil {
    		switch v := err.(type) {
    		case writerToResponse:
    			v.WriteTo(w)
    		default:
    			if h.lg != nil {
    				h.lg.Warn(
    					"failed to process Raft message",
    					zap.String("local-member-id", h.localID.String()),
    					zap.Error(err),
    				)
    			} else {
    				plog.Warningf("failed to process raft message (%v)", err)
    			}
    			http.Error(w, "error processing raft message", http.StatusInternalServerError)
    			w.(http.Flusher).Flush()
    			// disconnect the http stream
    			panic(err)
    		}
    		return
    	}
    
    	// Write StatusNoContent header after the message has been processed by
    	// raft, which facilitates the client to report MsgSnap status.
    	w.WriteHeader(http.StatusNoContent)  //向对端节点返回合适的状态码，表示请求已经被处理
    }

主要作用及流程：

pipelineHandler的ServeHTTP方法主要作用是 读取其他节点发送过来的请求数据，然后交给底层的etcd-raft模块进行处理。

1.读取对端发送过来的数据，

读取数据的时候，限制每次从底层连接读取的字节数上线，默认是64KB，因为快照数据可能非常大，为了防止读取超时，只能每次读取一部分数据到缓冲区中，最后将全部数据拼接起来，得到完整的快照数据。

2.将读取到的消息发送给底层的raft模块

3.返回对端节点状态码

**streamHandler的ServeHTTP方法实现**

func (h *streamHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    	//请求参数校验，如Method是否是GET，检验集群ID
    	if r.Method != "GET" {
    		w.Header().Set("Allow", "GET")
    		http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
    		return
    	}
    
    	w.Header().Set("X-Server-Version", version.Version)
    	w.Header().Set("X-Etcd-Cluster-ID", h.cid.String())
    
    	if err := checkClusterCompatibilityFromHeader(h.lg, h.tr.ID, r.Header, h.cid); err != nil {
    		http.Error(w, err.Error(), http.StatusPreconditionFailed)
    		return
    	}
    
    	//根据不同版本获取streamType
    	var t streamType
    	switch path.Dir(r.URL.Path) {
    	case streamTypeMsgAppV2.endpoint():
    		t = streamTypeMsgAppV2
    	case streamTypeMessage.endpoint():
    		t = streamTypeMessage
    	default:
    		if h.lg != nil {
    			h.lg.Debug(
    				"ignored unexpected streaming request path",
    				zap.String("local-member-id", h.tr.ID.String()),
    				zap.String("remote-peer-id-stream-handler", h.id.String()),
    				zap.String("path", r.URL.Path),
    			)
    		} else {
    			plog.Debugf("ignored unexpected streaming request path %s", r.URL.Path)
    		}
    		http.Error(w, "invalid path", http.StatusNotFound)
    		return
    	}
    
    	//获取对端节点的ID
    	fromStr := path.Base(r.URL.Path)
    	from, err := types.IDFromString(fromStr)
    	if err != nil {
    		if h.lg != nil {
    			h.lg.Warn(
    				"failed to parse path into ID",
    				zap.String("local-member-id", h.tr.ID.String()),
    				zap.String("remote-peer-id-stream-handler", h.id.String()),
    				zap.String("path", fromStr),
    				zap.Error(err),
    			)
    		} else {
    			plog.Errorf("failed to parse from %s into ID (%v)", fromStr, err)
    		}
    		http.Error(w, "invalid from", http.StatusNotFound)
    		return
    	}
    	if h.r.IsIDRemoved(uint64(from)) {
    		if h.lg != nil {
    			h.lg.Warn(
    				"rejected stream from remote peer because it was removed",
    				zap.String("local-member-id", h.tr.ID.String()),
    				zap.String("remote-peer-id-stream-handler", h.id.String()),
    				zap.String("remote-peer-id-from", from.String()),
    			)
    		} else {
    			plog.Warningf("rejected the stream from peer %s since it was removed", from)
    		}
    		http.Error(w, "removed member", http.StatusGone)
    		return
    	}
    	p := h.peerGetter.Get(from)  //根据对端节点ID获取对应的Peer实例
    	if p == nil {
    		// This may happen in following cases:
    		// 1. user starts a remote peer that belongs to a different cluster
    		// with the same cluster ID.
    		// 2. local etcd falls behind of the cluster, and cannot recognize
    		// the members that joined after its current progress.
    		if urls := r.Header.Get("X-PeerURLs"); urls != "" {
    			h.tr.AddRemote(from, strings.Split(urls, ","))
    		}
    		if h.lg != nil {
    			h.lg.Warn(
    				"failed to find remote peer in cluster",
    				zap.String("local-member-id", h.tr.ID.String()),
    				zap.String("remote-peer-id-stream-handler", h.id.String()),
    				zap.String("remote-peer-id-from", from.String()),
    				zap.String("cluster-id", h.cid.String()),
    			)
    		} else {
    			plog.Errorf("failed to find member %s in cluster %s", from, h.cid)
    		}
    		http.Error(w, "error sender not found", http.StatusNotFound)
    		return
    	}
    
    	wto := h.id.String()  //获取当前节点的ID
    	if gto := r.Header.Get("X-Raft-To"); gto != wto {
    		if h.lg != nil {
    			h.lg.Warn(
    				"ignored streaming request; ID mismatch",
    				zap.String("local-member-id", h.tr.ID.String()),
    				zap.String("remote-peer-id-stream-handler", h.id.String()),
    				zap.String("remote-peer-id-header", gto),
    				zap.String("remote-peer-id-from", from.String()),
    				zap.String("cluster-id", h.cid.String()),
    			)
    		} else {
    			plog.Errorf("streaming request ignored (ID mismatch got %s want %s)", gto, wto)
    		}
    		http.Error(w, "to field mismatch", http.StatusPreconditionFailed)
    		return
    	}
    
    	w.WriteHeader(http.StatusOK)  //返回状态码 200
    	w.(http.Flusher).Flush()  //调用Flush()方法将响应数据发送到对端节点
    
    	c := newCloseNotifier()
    	conn := &outgoingConn{  //创建outgoingConn实例
    		t:       t,
    		Writer:  w,
    		Flusher: w.(http.Flusher),
    		Closer:  c,
    		localID: h.tr.ID,
    		peerID:  h.id,
    	}
    	p.attachOutgoingConn(conn)   //建立连接,将outgoingConn实例与对应的streamWriter实例绑定
    	<-c.closeNotify()
    }

由于stream通道是长连接，当获取到对端请求后，将连接封装成outgoingConn实例，有streamWriter发送心跳来维护长连接。并给对端响应状态码。