rafthttp

Documentation

Overview

Package rafthttp implements HTTP transportation layer for etcd/raft pkg.

Index

• Constants
• Variables
• func NewListener(u url.URL, tlsinfo *transport.TLSInfo) (net.Listener, error)
• func NewRoundTripper(tlsInfo transport.TLSInfo, dialTimeout time.Duration) (http.RoundTripper, error)
• type Pausable
• type Peer
• type Raft
• type Transport
  • func (t *Transport) ActivePeers() (cnt int)
  • func (t *Transport) ActiveSince(id types.ID) time.Time
  • func (t *Transport) AddPeer(id types.ID, us []string)
  • func (t *Transport) AddRemote(id types.ID, us []string)
  • func (t *Transport) CutPeer(id types.ID)
  • func (t *Transport) Get(id types.ID) Peer
  • func (t *Transport) Handler() http.Handler
  • func (t *Transport) MendPeer(id types.ID)
  • func (t *Transport) Pause()
  • func (t *Transport) RemoveAllPeers()
  • func (t *Transport) RemovePeer(id types.ID)
  • func (t *Transport) Resume()
  • func (t *Transport) Send(msgs []raftpb.Message)
  • func (t *Transport) SendSnapshot(m snap.Message)
  • func (t *Transport) Start() error
  • func (t *Transport) Stop()
  • func (t *Transport) UpdatePeer(id types.ID, us []string)
• type Transporter
  • func NewNopTransporter() Transporter
  • func NewSnapTransporter(snapDir string) (Transporter, <-chan snap.Message)

Constants

const (
	// ConnReadTimeout and ConnWriteTimeout are the i/o timeout set on each connection rafthttp pkg creates.
	// A 5 seconds timeout is good enough for recycling bad connections. Or we have to wait for
	// tcp keepalive failing to detect a bad connection, which is at minutes level.
	// For long term streaming connections, rafthttp pkg sends application level linkHeartbeatMessage
	// to keep the connection alive.
	// For short term pipeline connections, the connection MUST be killed to avoid it being
	// put back to http pkg connection pool.
	ConnReadTimeout  = 5 * time.Second
	ConnWriteTimeout = 5 * time.Second
)

const (
	// RoundTripperNameRaftMessage is the name of round-tripper that sends
	// all other Raft messages, other than "snap.Message".
	RoundTripperNameRaftMessage = "ROUND_TRIPPER_RAFT_MESSAGE"
	// RoundTripperNameSnapshot is the name of round-tripper that sends merged snapshot message.
	RoundTripperNameSnapshot = "ROUND_TRIPPER_SNAPSHOT"
)

Variables

var (
	RaftPrefix         = "/raft"
	ProbingPrefix      = path.Join(RaftPrefix, "probing")
	RaftStreamPrefix   = path.Join(RaftPrefix, "stream")
	RaftSnapshotPrefix = path.Join(RaftPrefix, "snapshot")
)

var (
	ErrExceedSizeLimit = errors.New("rafthttp: error limit exceeded")
)

Functions

func NewListener

func NewListener(u url.URL, tlsinfo *transport.TLSInfo) (net.Listener, error)

NewListener returns a listener for raft message transfer between peers. It uses timeout listener to identify broken streams promptly.

func NewRoundTripper

func NewRoundTripper(tlsInfo transport.TLSInfo, dialTimeout time.Duration) (http.RoundTripper, error)

NewRoundTripper returns a roundTripper used to send requests to rafthttp listener of remote peers.

Types

type Pausable

type Pausable interface {
	Pause()
	Resume()
}

Pausable is a testing interface for pausing transport traffic.

type Peer

type Peer interface {
	// contains filtered or unexported methods
}

type Raft

type Raft interface {
	Process(ctx context.Context, m raftpb.Message) error  //将指定的消息实例传递到底层的etcd-raft模块进行处理
	IsIDRemoved(id uint64) bool                           //检测指定节点是否从当前集群中移出
	ReportUnreachable(id uint64)                          //通知底层etcd-raft模块，当前节点与指定的节点无法连通
	ReportSnapshot(id uint64, status raft.SnapshotStatus) //通知底层etcd-raft模块，快照数据是否发送成功
}

type Transport

type Transport struct {
	DialTimeout time.Duration // maximum duration before timing out dial of the request
	// 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

	TLSInfo transport.TLSInfo // TLS information used when creating connection

	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
	// used to record transportation statistics with followers when
	// performing as leader in raft protocol
	LeaderStats *stats.LeaderStats
	// 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
	// contains filtered or unexported fields
}

Transport implements Transporter interface. It provides the functionality to send raft messages to peers, and receive raft messages from peers. User should call Handler method to get a handler to serve requests received from peerURLs. User needs to call Start before calling other functions, and call Stop when the Transport is no longer used.

func (*Transport) ActivePeers

func (t *Transport) ActivePeers() (cnt int)

ActivePeers returns a channel that closes when an initial peer connection has been established. Use this to wait until the first peer connection becomes active.

func (*Transport) ActiveSince

func (t *Transport) ActiveSince(id types.ID) time.Time

func (*Transport) AddPeer

func (t *Transport) AddPeer(id types.ID, us []string)

该函数的主要工作是创建并启动对应节点的Peer实例

func (*Transport) AddRemote

func (t *Transport) AddRemote(id types.ID, us []string)

func (*Transport) CutPeer

func (t *Transport) CutPeer(id types.ID)

CutPeer drops messages to the specified peer.

func (*Transport) Get

func (t *Transport) Get(id types.ID) Peer

func (*Transport) Handler

func (t *Transport) Handler() http.Handler

该方法主要负责创建Stream消息通道和Pipeline消息通道用到的Handler实例，并注册到相应的请求路径。

func (*Transport) MendPeer

func (t *Transport) MendPeer(id types.ID)

MendPeer recovers the message dropping behavior of the given peer.

func (*Transport) Pause

func (t *Transport) Pause()

func (*Transport) RemoveAllPeers

func (t *Transport) RemoveAllPeers()

func (*Transport) RemovePeer

func (t *Transport) RemovePeer(id types.ID)

func (*Transport) Resume

func (t *Transport) Resume()

func (*Transport) Send

func (t *Transport) Send(msgs []raftpb.Message)

该方法负责发送指定的raftpb.Message消息，其中首先尝试使用目标节点对应的Peer实例发送消息，如果没有找到对应的Peer实例，则尝试使用对应的remote实例发送消息。

func (*Transport) SendSnapshot

func (t *Transport) SendSnapshot(m snap.Message)

func (*Transport) Start

func (t *Transport) Start() error

func (*Transport) Stop

func (t *Transport) Stop()

func (*Transport) UpdatePeer

func (t *Transport) UpdatePeer(id types.ID, us []string)

type Transporter

type Transporter interface {
	// Start starts the given Transporter.
	// Start MUST be called before calling other functions in the interface.
	Start() error //初始化操作
	// Handler returns the HTTP handler of the transporter.
	// A transporter HTTP handler handles the HTTP requests
	// from remote peers.
	// The handler MUST be used to handle RaftPrefix(/raft)
	// endpoint.
	Handler() http.Handler //创建Handler实例，并关联到指定的URL上
	// Send sends out the given messages to the remote peers.
	// Each message has a To field, which is an id that maps
	// to an existing peer in the transport.
	// If the id cannot be found in the transport, the message
	// will be ignored.
	Send(m []raftpb.Message) //发送消息
	// SendSnapshot sends out the given snapshot message to a remote peer.
	// The behavior of SendSnapshot is similar to Send.
	SendSnapshot(m snap.Message) //发送快照数据
	// AddRemote adds a remote with given peer urls into the transport.
	// A remote helps newly joined member to catch up the progress of cluster,
	// and will not be used after that.
	// It is the caller's responsibility to ensure the urls are all valid,
	// or it panics.
	AddRemote(id types.ID, urls []string) //在集群中添加一个节点时，其他节点会通过该方法添加该新加入节点的信息
	// AddPeer adds a peer with given peer urls into the transport.
	// It is the caller's responsibility to ensure the urls are all valid,
	// or it panics.
	// Peer urls are used to connect to the remote peer.
	AddPeer(id types.ID, urls []string)
	// RemovePeer removes the peer with given id.
	RemovePeer(id types.ID)
	// RemoveAllPeers removes all the existing peers in the transport.
	RemoveAllPeers()
	// UpdatePeer updates the peer urls of the peer with the given id.
	// It is the caller's responsibility to ensure the urls are all valid,
	// or it panics.
	UpdatePeer(id types.ID, urls []string)
	// ActiveSince returns the time that the connection with the peer
	// of the given id becomes active.
	// If the connection is active since peer was added, it returns the adding time.
	// If the connection is currently inactive, it returns zero time.
	ActiveSince(id types.ID) time.Time
	// ActivePeers returns the number of active peers.
	ActivePeers() int
	// Stop closes the connections and stops the transporter.
	Stop() //关闭操作，该方法会关闭全部的网络连接
}

func NewNopTransporter

func NewNopTransporter() Transporter

func NewSnapTransporter

func NewSnapTransporter(snapDir string) (Transporter, <-chan snap.Message)