etcdserver: introduce Server interface

This changes etcdserver.Server to an interface, with the former Server
(now "EtcdServer") becoming the canonical/production implementation.
This will facilitate better testing of the http server et al with mock
implementations of the interface.
It also more clearly defines the boundary for users of the Server.

etcdserver/etcdhttp/http.go

@@ -38,7 +38,7 @@ var errClosed = errors.New("etcdhttp: client closed connection")
 // raft communication.
 type Handler struct {
 	Timeout time.Duration
-	Server  *etcdserver.Server
+	Server  etcdserver.Server
 	// TODO: dynamic configuration may make this outdated. take care of it.
 	// TODO: dynamic configuration may introduce race also.
 	Peers Peers
@@ -127,9 +127,12 @@ func (h Handler) serveRaft(ctx context.Context, w http.ResponseWriter, r *http.R
 		log.Println("etcdhttp: error unmarshaling raft message:", err)
 	}
 	log.Printf("etcdhttp: raft recv message from %#x: %+v", m.From, m)
-	if err := h.Server.Node.Step(ctx, m); err != nil {
-		log.Println("etcdhttp: error stepping raft messages:", err)
+	if err := h.Server.Process(ctx, m); err != nil {
+		log.Println("etcdhttp: error processing raft message:", err)
+		writeError(w, err)
+		return
 	}
+	w.WriteHeader(http.StatusOK)
 }
 
 // genID generates a random id that is: n < 0 < n.

etcdserver/server.go

@@ -18,6 +18,7 @@ var (
 )
 
 type SendFunc func(m []raftpb.Message)
+type SaveFunc func(st raftpb.State, ents []raftpb.Entry)
 
 type Response struct {
 	Event   *store.Event
@@ -25,7 +26,24 @@ type Response struct {
 	err     error
 }
 
-type Server struct {
+type Server interface {
+	// Start performs any initialization of the Server necessary for it to
+	// begin serving requests. It must be called before Do or Process.
+	// Start must be non-blocking; any long-running server functionality
+	// should be implemented in goroutines.
+	Start()
+	// Stop terminates the Server and performs any necessary finalization.
+	// Do and Process cannot be called after Stop has been invoked.
+	Stop()
+	// Do takes a request and attempts to fulfil it, returning a Response.
+	Do(ctx context.Context, r pb.Request) (Response, error)
+	// Process takes a raft message and applies it to the server's raft state
+	// machine, respecting any timeout of the given context.
+	Process(ctx context.Context, m raftpb.Message) error
+}
+
+// EtcdServer is the production implementation of the Server interface
+type EtcdServer struct {
 	w    wait.Wait
 	done chan struct{}
 
@@ -34,27 +52,31 @@ type Server struct {
 
 	// Send specifies the send function for sending msgs to peers. Send
 	// MUST NOT block. It is okay to drop messages, since clients should
-	// timeout and reissue their messages.  If Send is nil, Server will
+	// timeout and reissue their messages.  If Send is nil, server will
 	// panic.
 	Send SendFunc
 
 	// Save specifies the save function for saving ents to stable storage.
 	// Save MUST block until st and ents are on stable storage.  If Send is
-	// nil, Server will panic.
+	// nil, server will panic.
 	Save func(st raftpb.State, ents []raftpb.Entry)
 
 	Ticker <-chan time.Time
 }
 
 // Start prepares and starts server in a new goroutine. It is no longer safe to
-// modify a Servers fields after it has been sent to Start.
-func Start(s *Server) {
+// modify a server's fields after it has been sent to Start.
+func (s *EtcdServer) Start() {
 	s.w = wait.New()
 	s.done = make(chan struct{})
 	go s.run()
 }
 
-func (s *Server) run() {
+func (s *EtcdServer) Process(ctx context.Context, m raftpb.Message) error {
+	return s.Node.Step(ctx, m)
+}
+
+func (s *EtcdServer) run() {
 	for {
 		select {
 		case <-s.Ticker:
@@ -79,9 +101,9 @@ func (s *Server) run() {
 	}
 }
 
-// Stop stops the server, and shutsdown the running goroutine. Stop should be
-// called after a Start(s), otherwise it will panic.
-func (s *Server) Stop() {
+// Stop stops the server, and shuts down the running goroutine. Stop should be
+// called after a Start(s), otherwise it will block forever.
+func (s *EtcdServer) Stop() {
 	s.Node.Stop()
 	close(s.done)
 }
@@ -91,7 +113,7 @@ func (s *Server) Stop() {
 // Quorum == true, r will be sent through consensus before performing its
 // respective operation. Do will block until an action is performed or there is
 // an error.
-func (s *Server) Do(ctx context.Context, r pb.Request) (Response, error) {
+func (s *EtcdServer) Do(ctx context.Context, r pb.Request) (Response, error) {
 	if r.Id == 0 {
 		panic("r.Id cannot be 0")
 	}
@@ -137,7 +159,7 @@ func (s *Server) Do(ctx context.Context, r pb.Request) (Response, error) {
 }
 
 // apply interprets r as a call to store.X and returns an Response interpreted from store.Event
-func (s *Server) apply(r pb.Request) Response {
+func (s *EtcdServer) apply(r pb.Request) Response {
 	f := func(ev *store.Event, err error) Response {
 		return Response{Event: ev, err: err}
 	}

etcdserver/server_test.go

@@ -39,7 +39,7 @@ func TestDoLocalAction(t *testing.T) {
 	}
 	for i, tt := range tests {
 		store := &storeRecorder{}
-		srv := &Server{Store: store}
+		srv := &EtcdServer{Store: store}
 		resp, err := srv.Do(context.TODO(), tt.req)
 
 		if err != tt.werr {
@@ -117,7 +117,7 @@ func TestApply(t *testing.T) {
 
 	for i, tt := range tests {
 		store := &storeRecorder{}
-		srv := &Server{Store: store}
+		srv := &EtcdServer{Store: store}
 		resp := srv.apply(tt.req)
 
 		if !reflect.DeepEqual(resp, tt.wresp) {
@@ -136,7 +136,7 @@ func testServer(t *testing.T, ns int64) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	ss := make([]*Server, ns)
+	ss := make([]*EtcdServer, ns)
 
 	send := func(msgs []raftpb.Message) {
 		for _, m := range msgs {
@@ -155,14 +155,14 @@ func testServer(t *testing.T, ns int64) {
 		n := raft.Start(id, peers, 10, 1)
 		tk := time.NewTicker(10 * time.Millisecond)
 		defer tk.Stop()
-		srv := &Server{
+		srv := &EtcdServer{
 			Node:   n,
 			Store:  store.New(),
 			Send:   send,
 			Save:   func(_ raftpb.State, _ []raftpb.Entry) {},
 			Ticker: tk.C,
 		}
-		Start(srv)
+		srv.Start()
 		// TODO(xiangli): randomize election timeout
 		// then remove this sleep.
 		time.Sleep(1 * time.Millisecond)
@@ -224,14 +224,14 @@ func TestDoProposal(t *testing.T) {
 		tk := make(chan time.Time)
 		// this makes <-tk always successful, which accelerates internal clock
 		close(tk)
-		srv := &Server{
+		srv := &EtcdServer{
 			Node:   n,
 			Store:  st,
 			Send:   func(_ []raftpb.Message) {},
 			Save:   func(_ raftpb.State, _ []raftpb.Entry) {},
 			Ticker: tk,
 		}
-		Start(srv)
+		srv.Start()
 		resp, err := srv.Do(ctx, tt)
 		srv.Stop()
 
@@ -254,7 +254,7 @@ func TestDoProposalCancelled(t *testing.T) {
 	n := raft.Start(0xBAD0, []int64{0xBAD0, 0xBAD1}, 10, 1)
 	st := &storeRecorder{}
 	wait := &waitRecorder{}
-	srv := &Server{
+	srv := &EtcdServer{
 		// TODO: use fake node for better testability
 		Node:  n,
 		Store: st,
@@ -291,7 +291,7 @@ func TestDoProposalStopped(t *testing.T) {
 	tk := make(chan time.Time)
 	// this makes <-tk always successful, which accelarates internal clock
 	close(tk)
-	srv := &Server{
+	srv := &EtcdServer{
 		// TODO: use fake node for better testability
 		Node:   n,
 		Store:  st,
@@ -299,7 +299,7 @@ func TestDoProposalStopped(t *testing.T) {
 		Save:   func(_ raftpb.State, _ []raftpb.Entry) {},
 		Ticker: tk,
 	}
-	Start(srv)
+	srv.Start()
 
 	done := make(chan struct{})
 	var err error

functional/http_functional_test.go

@@ -24,18 +24,16 @@ func TestSet(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	st := store.New()
-
 	n := raft.Start(1, []int64{1}, 0, 0)
 	n.Campaign(ctx)
 
-	srv := &etcdserver.Server{
+	srv := &etcdserver.EtcdServer{
+		Store: store.New(),
 		Node:  n,
-		Store: st,
-		Send:  etcdserver.SendFunc(nopSend),
 		Save:  func(st raftpb.State, ents []raftpb.Entry) {},
+		Send:  etcdserver.SendFunc(nopSend),
 	}
-	etcdserver.Start(srv)
+	srv.Start()
 	defer srv.Stop()
 
 	h := etcdhttp.Handler{

main.go

@@ -75,15 +75,14 @@ func startEtcd() http.Handler {
 
 	n, w := startRaft(id, peers.IDs(), path.Join(*dir, "wal"))
 
-	tk := time.NewTicker(100 * time.Millisecond)
-	s := &etcdserver.Server{
+	s := &etcdserver.EtcdServer{
 		Store:  store.New(),
 		Node:   n,
 		Save:   w.Save,
 		Send:   etcdhttp.Sender(*peers),
-		Ticker: tk.C,
+		Ticker: time.Tick(100 * time.Millisecond),
 	}
-	etcdserver.Start(s)
+	s.Start()
 
 	h := etcdhttp.Handler{
 		Timeout: *timeout,