func MustParseMemberIDFromKey(key string) types.ID {
	id, err := types.IDFromString(path.Base(key))
	if err != nil {
		plog.Panicf("unexpected parse member id error: %v", err)
	}
	return id
}

func (h *pipelineHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	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(r.Header, h.cid); err != nil {
		http.Error(w, err.Error(), http.StatusPreconditionFailed)
		return
	}

	if from, err := types.IDFromString(r.Header.Get("X-Server-From")); err != nil {
		if urls := r.Header.Get("X-PeerURLs"); urls != "" {
			h.tr.AddRemote(from, strings.Split(urls, ","))
		}
	}

	// 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.
	limitedr := pioutil.NewLimitedBufferReader(r.Body, connReadLimitByte)
	b, err := ioutil.ReadAll(limitedr)
	if err != nil {
		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 {
		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)))

	if err := h.r.Process(context.TODO(), m); err != nil {
		switch v := err.(type) {
		case writerToResponse:
			v.WriteTo(w)
		default:
			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)
}

func (h *streamHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	if r.Method != "GET" {
		w.Header().Set("Allow", "GET")
		http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
		return
	}

	fromStr := strings.TrimPrefix(r.URL.Path, RaftStreamPrefix+"/")
	from, err := types.IDFromString(fromStr)
	if err != nil {
		log.Printf("rafthttp: path %s cannot be parsed", fromStr)
		http.Error(w, "invalid path", http.StatusNotFound)
		return
	}
	p := h.tr.Peer(from)
	if p == nil {
		log.Printf("rafthttp: fail to find sender %s", from)
		http.Error(w, "error sender not found", http.StatusNotFound)
		return
	}

	wcid := h.cid.String()
	if gcid := r.Header.Get("X-Etcd-Cluster-ID"); gcid != wcid {
		log.Printf("rafthttp: streaming request ignored due to cluster ID mismatch got %s want %s", gcid, wcid)
		http.Error(w, "clusterID mismatch", http.StatusPreconditionFailed)
		return
	}

	wto := h.id.String()
	if gto := r.Header.Get("X-Raft-To"); gto != wto {
		log.Printf("rafthttp: streaming request ignored due to ID mismatch got %s want %s", gto, wto)
		http.Error(w, "to field mismatch", http.StatusPreconditionFailed)
		return
	}

	termStr := r.Header.Get("X-Raft-Term")
	term, err := strconv.ParseUint(termStr, 10, 64)
	if err != nil {
		log.Printf("rafthttp: streaming request ignored due to parse term %s error: %v", termStr, err)
		http.Error(w, "invalid term field", http.StatusBadRequest)
		return
	}

	sw := newStreamWriter(from, term)
	err = p.attachStream(sw)
	if err != nil {
		log.Printf("rafthttp: %v", err)
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}

	w.WriteHeader(http.StatusOK)
	w.(http.Flusher).Flush()
	go sw.handle(w.(WriteFlusher))
	<-sw.stopNotify()
}

func (h *streamHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	if r.Method != "GET" {
		w.Header().Set("Allow", "GET")
		http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
		return
	}

	fromStr := strings.TrimPrefix(r.URL.Path, RaftStreamPrefix+"/")
	from, err := types.IDFromString(fromStr)
	if err != nil {
		log.Printf("rafthttp: path %s cannot be parsed", fromStr)
		http.Error(w, "invalid path", http.StatusNotFound)
		return
	}
	s := h.finder.Sender(from)
	if s == nil {
		log.Printf("rafthttp: fail to find sender %s", from)
		http.Error(w, "error sender not found", http.StatusNotFound)
		return
	}

	wcid := h.cid.String()
	if gcid := r.Header.Get("X-Etcd-Cluster-ID"); gcid != wcid {
		log.Printf("rafthttp: streaming request ignored due to cluster ID mismatch got %s want %s", gcid, wcid)
		http.Error(w, "clusterID mismatch", http.StatusPreconditionFailed)
		return
	}

	wto := h.id.String()
	if gto := r.Header.Get("X-Raft-To"); gto != wto {
		log.Printf("rafthttp: streaming request ignored due to ID mismatch got %s want %s", gto, wto)
		http.Error(w, "to field mismatch", http.StatusPreconditionFailed)
		return
	}

	termStr := r.Header.Get("X-Raft-Term")
	term, err := strconv.ParseUint(termStr, 10, 64)
	if err != nil {
		log.Printf("rafthttp: streaming request ignored due to parse term %s error: %v", termStr, err)
		http.Error(w, "invalid term field", http.StatusBadRequest)
		return
	}

	w.WriteHeader(http.StatusOK)
	w.(http.Flusher).Flush()

	done, err := s.StartStreaming(w.(WriteFlusher), from, term)
	if err != nil {
		log.Printf("rafthttp: streaming request ignored due to start streaming error: %v", err)
		// TODO: consider http status and info here
		http.Error(w, "error enable streaming", http.StatusInternalServerError)
		return
	}
	<-done
}

func getID(p string, w http.ResponseWriter) (types.ID, bool) {
	idStr := trimPrefix(p, membersPrefix)
	if idStr == "" {
		http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
		return 0, false
	}
	id, err := types.IDFromString(idStr)
	if err != nil {
		writeError(w, nil, httptypes.NewHTTPError(http.StatusNotFound, fmt.Sprintf("No such member: %s", idStr)))
		return 0, false
	}
	return id, true
}

// If logerr is true, it prints out more error messages.
func getClusterFromRemotePeers(urls []string, timeout time.Duration, logerr bool, rt http.RoundTripper) (*membership.RaftCluster, error) {
	cc := &http.Client{
		Transport: rt,
		Timeout:   timeout,
	}
	for _, u := range urls {
		resp, err := cc.Get(u + "/members")
		if err != nil {
			if logerr {
				plog.Warningf("could not get cluster response from %s: %v", u, err)
			}
			continue
		}
		b, err := ioutil.ReadAll(resp.Body)
		resp.Body.Close()
		if err != nil {
			if logerr {
				plog.Warningf("could not read the body of cluster response: %v", err)
			}
			continue
		}
		var membs []*membership.Member
		if err = json.Unmarshal(b, &membs); err != nil {
			if logerr {
				plog.Warningf("could not unmarshal cluster response: %v", err)
			}
			continue
		}
		id, err := types.IDFromString(resp.Header.Get("X-Etcd-Cluster-ID"))
		if err != nil {
			if logerr {
				plog.Warningf("could not parse the cluster ID from cluster res: %v", err)
			}
			continue
		}

		// check the length of membership members
		// if the membership members are present then prepare and return raft cluster
		// if membership members are not present then the raft cluster formed will be
		// an invalid empty cluster hence return failed to get raft cluster member(s) from the given urls error
		if len(membs) > 0 {
			return membership.NewClusterFromMembers("", id, membs), nil
		}

		return nil, fmt.Errorf("failed to get raft cluster member(s) from the given urls.")
	}
	return nil, fmt.Errorf("could not retrieve cluster information from the given urls")
}

// If logerr is true, it prints out more error messages.
func getClusterFromPeers(urls []string, logerr bool) (*Cluster, error) {
	cc := &http.Client{
		Transport: &http.Transport{
			ResponseHeaderTimeout: 500 * time.Millisecond,
		},
		Timeout: time.Second,
	}
	for _, u := range urls {
		resp, err := cc.Get(u + "/members")
		if err != nil {
			if logerr {
				log.Printf("etcdserver: could not get cluster response from %s: %v", u, err)
			}
			continue
		}
		b, err := ioutil.ReadAll(resp.Body)
		if err != nil {
			if logerr {
				log.Printf("etcdserver: could not read the body of cluster response: %v", err)
			}
			continue
		}
		var membs []*Member
		if err := json.Unmarshal(b, &membs); err != nil {
			if logerr {
				log.Printf("etcdserver: could not unmarshal cluster response: %v", err)
			}
			continue
		}
		id, err := types.IDFromString(resp.Header.Get("X-Etcd-Cluster-ID"))
		if err != nil {
			if logerr {
				log.Printf("etcdserver: could not parse the cluster ID from cluster res: %v", err)
			}
			continue
		}
		return NewClusterFromMembers("", id, membs), nil
	}
	return nil, fmt.Errorf("etcdserver: could not retrieve cluster information from the given urls")
}

// If logerr is true, it prints out more error messages.
func getClusterFromRemotePeers(urls []string, timeout time.Duration, logerr bool, rt http.RoundTripper) (*membership.RaftCluster, error) {
	cc := &http.Client{
		Transport: rt,
		Timeout:   timeout,
	}
	for _, u := range urls {
		resp, err := cc.Get(u + "/members")
		if err != nil {
			if logerr {
				plog.Warningf("could not get cluster response from %s: %v", u, err)
			}
			continue
		}
		b, err := ioutil.ReadAll(resp.Body)
		resp.Body.Close()
		if err != nil {
			if logerr {
				plog.Warningf("could not read the body of cluster response: %v", err)
			}
			continue
		}
		var membs []*membership.Member
		if err = json.Unmarshal(b, &membs); err != nil {
			if logerr {
				plog.Warningf("could not unmarshal cluster response: %v", err)
			}
			continue
		}
		id, err := types.IDFromString(resp.Header.Get("X-Etcd-Cluster-ID"))
		if err != nil {
			if logerr {
				plog.Warningf("could not parse the cluster ID from cluster res: %v", err)
			}
			continue
		}
		return membership.NewClusterFromMembers("", id, membs), nil
	}
	return nil, fmt.Errorf("could not retrieve cluster information from the given urls")
}

func (h *streamHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	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(r.Header, h.cid); err != nil {
		http.Error(w, err.Error(), http.StatusPreconditionFailed)
		return
	}

	var t streamType
	switch path.Dir(r.URL.Path) {
	case streamTypeMsgAppV2.endpoint():
		t = streamTypeMsgAppV2
	case streamTypeMessage.endpoint():
		t = streamTypeMessage
	default:
		plog.Debugf("ignored unexpected streaming request path %s", r.URL.Path)
		http.Error(w, "invalid path", http.StatusNotFound)
		return
	}

	fromStr := path.Base(r.URL.Path)
	from, err := types.IDFromString(fromStr)
	if err != nil {
		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)) {
		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)
	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.
		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()
	if gto := r.Header.Get("X-Raft-To"); gto != wto {
		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)
	w.(http.Flusher).Flush()

	c := newCloseNotifier()
	conn := &outgoingConn{
		t:       t,
		Writer:  w,
		Flusher: w.(http.Flusher),
		Closer:  c,
	}
	p.attachOutgoingConn(conn)
	<-c.closeNotify()
}