1// Copyright 2015 The etcd Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15package rafthttp
16
17import (
18	"fmt"
19	"io"
20	"net"
21	"net/http"
22	"net/url"
23	"strings"
24	"time"
25
26	"go.etcd.io/etcd/pkg/transport"
27	"go.etcd.io/etcd/pkg/types"
28	"go.etcd.io/etcd/version"
29
30	"github.com/coreos/go-semver/semver"
31)
32
33var (
34	errMemberRemoved  = fmt.Errorf("the member has been permanently removed from the cluster")
35	errMemberNotFound = fmt.Errorf("member not found")
36)
37
38// NewListener returns a listener for raft message transfer between peers.
39// It uses timeout listener to identify broken streams promptly.
40func NewListener(u url.URL, tlsinfo *transport.TLSInfo) (net.Listener, error) {
41	return transport.NewTimeoutListener(u.Host, u.Scheme, tlsinfo, ConnReadTimeout, ConnWriteTimeout)
42}
43
44// NewRoundTripper returns a roundTripper used to send requests
45// to rafthttp listener of remote peers.
46func NewRoundTripper(tlsInfo transport.TLSInfo, dialTimeout time.Duration) (http.RoundTripper, error) {
47	// It uses timeout transport to pair with remote timeout listeners.
48	// It sets no read/write timeout, because message in requests may
49	// take long time to write out before reading out the response.
50	return transport.NewTimeoutTransport(tlsInfo, dialTimeout, 0, 0)
51}
52
53// newStreamRoundTripper returns a roundTripper used to send stream requests
54// to rafthttp listener of remote peers.
55// Read/write timeout is set for stream roundTripper to promptly
56// find out broken status, which minimizes the number of messages
57// sent on broken connection.
58func newStreamRoundTripper(tlsInfo transport.TLSInfo, dialTimeout time.Duration) (http.RoundTripper, error) {
59	return transport.NewTimeoutTransport(tlsInfo, dialTimeout, ConnReadTimeout, ConnWriteTimeout)
60}
61
62// createPostRequest creates a HTTP POST request that sends raft message.
63func createPostRequest(u url.URL, path string, body io.Reader, ct string, urls types.URLs, from, cid types.ID) *http.Request {
64	uu := u
65	uu.Path = path
66	req, err := http.NewRequest("POST", uu.String(), body)
67	if err != nil {
68		plog.Panicf("unexpected new request error (%v)", err)
69	}
70	req.Header.Set("Content-Type", ct)
71	req.Header.Set("X-Server-From", from.String())
72	req.Header.Set("X-Server-Version", version.Version)
73	req.Header.Set("X-Min-Cluster-Version", version.MinClusterVersion)
74	req.Header.Set("X-Etcd-Cluster-ID", cid.String())
75	setPeerURLsHeader(req, urls)
76
77	return req
78}
79
80// checkPostResponse checks the response of the HTTP POST request that sends
81// raft message.
82func checkPostResponse(resp *http.Response, body []byte, req *http.Request, to types.ID) error {
83	switch resp.StatusCode {
84	case http.StatusPreconditionFailed:
85		switch strings.TrimSuffix(string(body), "\n") {
86		case errIncompatibleVersion.Error():
87			plog.Errorf("request sent was ignored by peer %s (server version incompatible)", to)
88			return errIncompatibleVersion
89		case errClusterIDMismatch.Error():
90			plog.Errorf("request sent was ignored (cluster ID mismatch: remote[%s]=%s, local=%s)",
91				to, resp.Header.Get("X-Etcd-Cluster-ID"), req.Header.Get("X-Etcd-Cluster-ID"))
92			return errClusterIDMismatch
93		default:
94			return fmt.Errorf("unhandled error %q when precondition failed", string(body))
95		}
96	case http.StatusForbidden:
97		return errMemberRemoved
98	case http.StatusNoContent:
99		return nil
100	default:
101		return fmt.Errorf("unexpected http status %s while posting to %q", http.StatusText(resp.StatusCode), req.URL.String())
102	}
103}
104
105// reportCriticalError reports the given error through sending it into
106// the given error channel.
107// If the error channel is filled up when sending error, it drops the error
108// because the fact that error has happened is reported, which is
109// good enough.
110func reportCriticalError(err error, errc chan<- error) {
111	select {
112	case errc <- err:
113	default:
114	}
115}
116
117// compareMajorMinorVersion returns an integer comparing two versions based on
118// their major and minor version. The result will be 0 if a==b, -1 if a < b,
119// and 1 if a > b.
120func compareMajorMinorVersion(a, b *semver.Version) int {
121	na := &semver.Version{Major: a.Major, Minor: a.Minor}
122	nb := &semver.Version{Major: b.Major, Minor: b.Minor}
123	switch {
124	case na.LessThan(*nb):
125		return -1
126	case nb.LessThan(*na):
127		return 1
128	default:
129		return 0
130	}
131}
132
133// serverVersion returns the server version from the given header.
134func serverVersion(h http.Header) *semver.Version {
135	verStr := h.Get("X-Server-Version")
136	// backward compatibility with etcd 2.0
137	if verStr == "" {
138		verStr = "2.0.0"
139	}
140	return semver.Must(semver.NewVersion(verStr))
141}
142
143// serverVersion returns the min cluster version from the given header.
144func minClusterVersion(h http.Header) *semver.Version {
145	verStr := h.Get("X-Min-Cluster-Version")
146	// backward compatibility with etcd 2.0
147	if verStr == "" {
148		verStr = "2.0.0"
149	}
150	return semver.Must(semver.NewVersion(verStr))
151}
152
153// checkVersionCompatibility checks whether the given version is compatible
154// with the local version.
155func checkVersionCompatibility(name string, server, minCluster *semver.Version) (
156	localServer *semver.Version,
157	localMinCluster *semver.Version,
158	err error) {
159	localServer = semver.Must(semver.NewVersion(version.Version))
160	localMinCluster = semver.Must(semver.NewVersion(version.MinClusterVersion))
161	if compareMajorMinorVersion(server, localMinCluster) == -1 {
162		return localServer, localMinCluster, fmt.Errorf("remote version is too low: remote[%s]=%s, local=%s", name, server, localServer)
163	}
164	if compareMajorMinorVersion(minCluster, localServer) == 1 {
165		return localServer, localMinCluster, fmt.Errorf("local version is too low: remote[%s]=%s, local=%s", name, server, localServer)
166	}
167	return localServer, localMinCluster, nil
168}
169
170// setPeerURLsHeader reports local urls for peer discovery
171func setPeerURLsHeader(req *http.Request, urls types.URLs) {
172	if urls == nil {
173		// often not set in unit tests
174		return
175	}
176	peerURLs := make([]string, urls.Len())
177	for i := range urls {
178		peerURLs[i] = urls[i].String()
179	}
180	req.Header.Set("X-PeerURLs", strings.Join(peerURLs, ","))
181}
182
183// addRemoteFromRequest adds a remote peer according to an http request header
184func addRemoteFromRequest(tr Transporter, r *http.Request) {
185	if from, err := types.IDFromString(r.Header.Get("X-Server-From")); err == nil {
186		if urls := r.Header.Get("X-PeerURLs"); urls != "" {
187			tr.AddRemote(from, strings.Split(urls, ","))
188		}
189	}
190}
191