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