client.go 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773
  1. // Copyright 2015 CoreOS, Inc.
  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. package etcdhttp
  15. import (
  16. "encoding/json"
  17. "errors"
  18. "expvar"
  19. "fmt"
  20. "io/ioutil"
  21. "net/http"
  22. "net/url"
  23. "path"
  24. "strconv"
  25. "strings"
  26. "time"
  27. "github.com/coreos/etcd/Godeps/_workspace/src/github.com/coreos/pkg/capnslog"
  28. "github.com/coreos/etcd/Godeps/_workspace/src/github.com/jonboulle/clockwork"
  29. "github.com/coreos/etcd/Godeps/_workspace/src/github.com/prometheus/client_golang/prometheus"
  30. "github.com/coreos/etcd/Godeps/_workspace/src/golang.org/x/net/context"
  31. etcdErr "github.com/coreos/etcd/error"
  32. "github.com/coreos/etcd/etcdserver"
  33. "github.com/coreos/etcd/etcdserver/auth"
  34. "github.com/coreos/etcd/etcdserver/etcdhttp/httptypes"
  35. "github.com/coreos/etcd/etcdserver/etcdserverpb"
  36. "github.com/coreos/etcd/etcdserver/stats"
  37. "github.com/coreos/etcd/pkg/types"
  38. "github.com/coreos/etcd/raft"
  39. "github.com/coreos/etcd/store"
  40. "github.com/coreos/etcd/version"
  41. )
  42. const (
  43. authPrefix = "/v2/auth"
  44. keysPrefix = "/v2/keys"
  45. deprecatedMachinesPrefix = "/v2/machines"
  46. membersPrefix = "/v2/members"
  47. statsPrefix = "/v2/stats"
  48. varsPath = "/debug/vars"
  49. metricsPath = "/metrics"
  50. healthPath = "/health"
  51. versionPath = "/version"
  52. configPath = "/config"
  53. )
  54. // NewClientHandler generates a muxed http.Handler with the given parameters to serve etcd client requests.
  55. func NewClientHandler(server *etcdserver.EtcdServer, timeout time.Duration) http.Handler {
  56. go capabilityLoop(server)
  57. sec := auth.NewStore(server, timeout)
  58. kh := &keysHandler{
  59. sec: sec,
  60. server: server,
  61. cluster: server.Cluster(),
  62. timer: server,
  63. timeout: timeout,
  64. }
  65. sh := &statsHandler{
  66. stats: server,
  67. }
  68. mh := &membersHandler{
  69. sec: sec,
  70. server: server,
  71. cluster: server.Cluster(),
  72. timeout: timeout,
  73. clock: clockwork.NewRealClock(),
  74. }
  75. dmh := &deprecatedMachinesHandler{
  76. cluster: server.Cluster(),
  77. }
  78. sech := &authHandler{
  79. sec: sec,
  80. cluster: server.Cluster(),
  81. }
  82. mux := http.NewServeMux()
  83. mux.HandleFunc("/", http.NotFound)
  84. mux.Handle(healthPath, healthHandler(server))
  85. mux.HandleFunc(versionPath, versionHandler(server.Cluster(), serveVersion))
  86. mux.Handle(keysPrefix, kh)
  87. mux.Handle(keysPrefix+"/", kh)
  88. mux.HandleFunc(statsPrefix+"/store", sh.serveStore)
  89. mux.HandleFunc(statsPrefix+"/self", sh.serveSelf)
  90. mux.HandleFunc(statsPrefix+"/leader", sh.serveLeader)
  91. mux.HandleFunc(varsPath, serveVars)
  92. mux.HandleFunc(configPath+"/local/log", logHandleFunc)
  93. mux.Handle(metricsPath, prometheus.Handler())
  94. mux.Handle(membersPrefix, mh)
  95. mux.Handle(membersPrefix+"/", mh)
  96. mux.Handle(deprecatedMachinesPrefix, dmh)
  97. handleAuth(mux, sech)
  98. return requestLogger(mux)
  99. }
  100. type keysHandler struct {
  101. sec auth.Store
  102. server etcdserver.Server
  103. cluster etcdserver.Cluster
  104. timer etcdserver.RaftTimer
  105. timeout time.Duration
  106. }
  107. func (h *keysHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
  108. if !allowMethod(w, r.Method, "HEAD", "GET", "PUT", "POST", "DELETE") {
  109. return
  110. }
  111. w.Header().Set("X-Etcd-Cluster-ID", h.cluster.ID().String())
  112. ctx, cancel := context.WithTimeout(context.Background(), h.timeout)
  113. defer cancel()
  114. clock := clockwork.NewRealClock()
  115. startTime := clock.Now()
  116. rr, err := parseKeyRequest(r, clock)
  117. if err != nil {
  118. writeKeyError(w, err)
  119. return
  120. }
  121. // The path must be valid at this point (we've parsed the request successfully).
  122. if !hasKeyPrefixAccess(h.sec, r, r.URL.Path[len(keysPrefix):], rr.Recursive) {
  123. writeKeyNoAuth(w)
  124. return
  125. }
  126. if !rr.Wait {
  127. reportRequestReceived(rr)
  128. }
  129. resp, err := h.server.Do(ctx, rr)
  130. if err != nil {
  131. err = trimErrorPrefix(err, etcdserver.StoreKeysPrefix)
  132. writeKeyError(w, err)
  133. reportRequestFailed(rr, err)
  134. return
  135. }
  136. switch {
  137. case resp.Event != nil:
  138. if err := writeKeyEvent(w, resp.Event, h.timer); err != nil {
  139. // Should never be reached
  140. plog.Errorf("error writing event (%v)", err)
  141. }
  142. reportRequestCompleted(rr, resp, startTime)
  143. case resp.Watcher != nil:
  144. ctx, cancel := context.WithTimeout(context.Background(), defaultWatchTimeout)
  145. defer cancel()
  146. handleKeyWatch(ctx, w, resp.Watcher, rr.Stream, h.timer)
  147. default:
  148. writeKeyError(w, errors.New("received response with no Event/Watcher!"))
  149. }
  150. }
  151. type deprecatedMachinesHandler struct {
  152. cluster etcdserver.Cluster
  153. }
  154. func (h *deprecatedMachinesHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
  155. if !allowMethod(w, r.Method, "GET", "HEAD") {
  156. return
  157. }
  158. endpoints := h.cluster.ClientURLs()
  159. w.Write([]byte(strings.Join(endpoints, ", ")))
  160. }
  161. type membersHandler struct {
  162. sec auth.Store
  163. server etcdserver.Server
  164. cluster etcdserver.Cluster
  165. timeout time.Duration
  166. clock clockwork.Clock
  167. }
  168. func (h *membersHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
  169. if !allowMethod(w, r.Method, "GET", "POST", "DELETE", "PUT") {
  170. return
  171. }
  172. if !hasWriteRootAccess(h.sec, r) {
  173. writeNoAuth(w, r)
  174. return
  175. }
  176. w.Header().Set("X-Etcd-Cluster-ID", h.cluster.ID().String())
  177. ctx, cancel := context.WithTimeout(context.Background(), h.timeout)
  178. defer cancel()
  179. switch r.Method {
  180. case "GET":
  181. switch trimPrefix(r.URL.Path, membersPrefix) {
  182. case "":
  183. mc := newMemberCollection(h.cluster.Members())
  184. w.Header().Set("Content-Type", "application/json")
  185. if err := json.NewEncoder(w).Encode(mc); err != nil {
  186. plog.Warningf("failed to encode members response (%v)", err)
  187. }
  188. case "leader":
  189. id := h.server.Leader()
  190. if id == 0 {
  191. writeError(w, r, httptypes.NewHTTPError(http.StatusServiceUnavailable, "During election"))
  192. return
  193. }
  194. m := newMember(h.cluster.Member(id))
  195. w.Header().Set("Content-Type", "application/json")
  196. if err := json.NewEncoder(w).Encode(m); err != nil {
  197. plog.Warningf("failed to encode members response (%v)", err)
  198. }
  199. default:
  200. writeError(w, r, httptypes.NewHTTPError(http.StatusNotFound, "Not found"))
  201. }
  202. case "POST":
  203. req := httptypes.MemberCreateRequest{}
  204. if ok := unmarshalRequest(r, &req, w); !ok {
  205. return
  206. }
  207. now := h.clock.Now()
  208. m := etcdserver.NewMember("", req.PeerURLs, "", &now)
  209. err := h.server.AddMember(ctx, *m)
  210. switch {
  211. case err == etcdserver.ErrIDExists || err == etcdserver.ErrPeerURLexists:
  212. writeError(w, r, httptypes.NewHTTPError(http.StatusConflict, err.Error()))
  213. return
  214. case err != nil:
  215. plog.Errorf("error adding member %s (%v)", m.ID, err)
  216. writeError(w, r, err)
  217. return
  218. }
  219. res := newMember(m)
  220. w.Header().Set("Content-Type", "application/json")
  221. w.WriteHeader(http.StatusCreated)
  222. if err := json.NewEncoder(w).Encode(res); err != nil {
  223. plog.Warningf("failed to encode members response (%v)", err)
  224. }
  225. case "DELETE":
  226. id, ok := getID(r.URL.Path, w)
  227. if !ok {
  228. return
  229. }
  230. err := h.server.RemoveMember(ctx, uint64(id))
  231. switch {
  232. case err == etcdserver.ErrIDRemoved:
  233. writeError(w, r, httptypes.NewHTTPError(http.StatusGone, fmt.Sprintf("Member permanently removed: %s", id)))
  234. case err == etcdserver.ErrIDNotFound:
  235. writeError(w, r, httptypes.NewHTTPError(http.StatusNotFound, fmt.Sprintf("No such member: %s", id)))
  236. case err != nil:
  237. plog.Errorf("error removing member %s (%v)", id, err)
  238. writeError(w, r, err)
  239. default:
  240. w.WriteHeader(http.StatusNoContent)
  241. }
  242. case "PUT":
  243. id, ok := getID(r.URL.Path, w)
  244. if !ok {
  245. return
  246. }
  247. req := httptypes.MemberUpdateRequest{}
  248. if ok := unmarshalRequest(r, &req, w); !ok {
  249. return
  250. }
  251. m := etcdserver.Member{
  252. ID: id,
  253. RaftAttributes: etcdserver.RaftAttributes{PeerURLs: req.PeerURLs.StringSlice()},
  254. }
  255. err := h.server.UpdateMember(ctx, m)
  256. switch {
  257. case err == etcdserver.ErrPeerURLexists:
  258. writeError(w, r, httptypes.NewHTTPError(http.StatusConflict, err.Error()))
  259. case err == etcdserver.ErrIDNotFound:
  260. writeError(w, r, httptypes.NewHTTPError(http.StatusNotFound, fmt.Sprintf("No such member: %s", id)))
  261. case err != nil:
  262. plog.Errorf("error updating member %s (%v)", m.ID, err)
  263. writeError(w, r, err)
  264. default:
  265. w.WriteHeader(http.StatusNoContent)
  266. }
  267. }
  268. }
  269. type statsHandler struct {
  270. stats stats.Stats
  271. }
  272. func (h *statsHandler) serveStore(w http.ResponseWriter, r *http.Request) {
  273. if !allowMethod(w, r.Method, "GET") {
  274. return
  275. }
  276. w.Header().Set("Content-Type", "application/json")
  277. w.Write(h.stats.StoreStats())
  278. }
  279. func (h *statsHandler) serveSelf(w http.ResponseWriter, r *http.Request) {
  280. if !allowMethod(w, r.Method, "GET") {
  281. return
  282. }
  283. w.Header().Set("Content-Type", "application/json")
  284. w.Write(h.stats.SelfStats())
  285. }
  286. func (h *statsHandler) serveLeader(w http.ResponseWriter, r *http.Request) {
  287. if !allowMethod(w, r.Method, "GET") {
  288. return
  289. }
  290. stats := h.stats.LeaderStats()
  291. if stats == nil {
  292. writeError(w, r, httptypes.NewHTTPError(http.StatusForbidden, "not current leader"))
  293. return
  294. }
  295. w.Header().Set("Content-Type", "application/json")
  296. w.Write(stats)
  297. }
  298. func serveVars(w http.ResponseWriter, r *http.Request) {
  299. w.Header().Set("Content-Type", "application/json; charset=utf-8")
  300. fmt.Fprintf(w, "{\n")
  301. first := true
  302. expvar.Do(func(kv expvar.KeyValue) {
  303. if !first {
  304. fmt.Fprintf(w, ",\n")
  305. }
  306. first = false
  307. fmt.Fprintf(w, "%q: %s", kv.Key, kv.Value)
  308. })
  309. fmt.Fprintf(w, "\n}\n")
  310. }
  311. // TODO: change etcdserver to raft interface when we have it.
  312. // add test for healthHeadler when we have the interface ready.
  313. func healthHandler(server *etcdserver.EtcdServer) http.HandlerFunc {
  314. return func(w http.ResponseWriter, r *http.Request) {
  315. if !allowMethod(w, r.Method, "GET") {
  316. return
  317. }
  318. if uint64(server.Leader()) == raft.None {
  319. http.Error(w, `{"health": "false"}`, http.StatusServiceUnavailable)
  320. return
  321. }
  322. // wait for raft's progress
  323. index := server.Index()
  324. for i := 0; i < 3; i++ {
  325. time.Sleep(250 * time.Millisecond)
  326. if server.Index() > index {
  327. w.WriteHeader(http.StatusOK)
  328. w.Write([]byte(`{"health": "true"}`))
  329. return
  330. }
  331. }
  332. http.Error(w, `{"health": "false"}`, http.StatusServiceUnavailable)
  333. return
  334. }
  335. }
  336. func versionHandler(c etcdserver.Cluster, fn func(http.ResponseWriter, *http.Request, string)) http.HandlerFunc {
  337. return func(w http.ResponseWriter, r *http.Request) {
  338. v := c.Version()
  339. if v != nil {
  340. fn(w, r, v.String())
  341. } else {
  342. fn(w, r, "not_decided")
  343. }
  344. }
  345. }
  346. func serveVersion(w http.ResponseWriter, r *http.Request, clusterV string) {
  347. if !allowMethod(w, r.Method, "GET") {
  348. return
  349. }
  350. vs := version.Versions{
  351. Server: version.Version,
  352. Cluster: clusterV,
  353. }
  354. w.Header().Set("Content-Type", "application/json")
  355. b, err := json.Marshal(&vs)
  356. if err != nil {
  357. plog.Panicf("cannot marshal versions to json (%v)", err)
  358. }
  359. w.Write(b)
  360. }
  361. func logHandleFunc(w http.ResponseWriter, r *http.Request) {
  362. if !allowMethod(w, r.Method, "PUT") {
  363. return
  364. }
  365. in := struct{ Level string }{}
  366. d := json.NewDecoder(r.Body)
  367. if err := d.Decode(&in); err != nil {
  368. writeError(w, r, httptypes.NewHTTPError(http.StatusBadRequest, "Invalid json body"))
  369. return
  370. }
  371. logl, err := capnslog.ParseLevel(strings.ToUpper(in.Level))
  372. if err != nil {
  373. writeError(w, r, httptypes.NewHTTPError(http.StatusBadRequest, "Invalid log level "+in.Level))
  374. return
  375. }
  376. plog.Noticef("globalLogLevel set to %q", logl.String())
  377. capnslog.SetGlobalLogLevel(logl)
  378. w.WriteHeader(http.StatusNoContent)
  379. }
  380. // parseKeyRequest converts a received http.Request on keysPrefix to
  381. // a server Request, performing validation of supplied fields as appropriate.
  382. // If any validation fails, an empty Request and non-nil error is returned.
  383. func parseKeyRequest(r *http.Request, clock clockwork.Clock) (etcdserverpb.Request, error) {
  384. emptyReq := etcdserverpb.Request{}
  385. err := r.ParseForm()
  386. if err != nil {
  387. return emptyReq, etcdErr.NewRequestError(
  388. etcdErr.EcodeInvalidForm,
  389. err.Error(),
  390. )
  391. }
  392. if !strings.HasPrefix(r.URL.Path, keysPrefix) {
  393. return emptyReq, etcdErr.NewRequestError(
  394. etcdErr.EcodeInvalidForm,
  395. "incorrect key prefix",
  396. )
  397. }
  398. p := path.Join(etcdserver.StoreKeysPrefix, r.URL.Path[len(keysPrefix):])
  399. var pIdx, wIdx uint64
  400. if pIdx, err = getUint64(r.Form, "prevIndex"); err != nil {
  401. return emptyReq, etcdErr.NewRequestError(
  402. etcdErr.EcodeIndexNaN,
  403. `invalid value for "prevIndex"`,
  404. )
  405. }
  406. if wIdx, err = getUint64(r.Form, "waitIndex"); err != nil {
  407. return emptyReq, etcdErr.NewRequestError(
  408. etcdErr.EcodeIndexNaN,
  409. `invalid value for "waitIndex"`,
  410. )
  411. }
  412. var rec, sort, wait, dir, quorum, stream bool
  413. if rec, err = getBool(r.Form, "recursive"); err != nil {
  414. return emptyReq, etcdErr.NewRequestError(
  415. etcdErr.EcodeInvalidField,
  416. `invalid value for "recursive"`,
  417. )
  418. }
  419. if sort, err = getBool(r.Form, "sorted"); err != nil {
  420. return emptyReq, etcdErr.NewRequestError(
  421. etcdErr.EcodeInvalidField,
  422. `invalid value for "sorted"`,
  423. )
  424. }
  425. if wait, err = getBool(r.Form, "wait"); err != nil {
  426. return emptyReq, etcdErr.NewRequestError(
  427. etcdErr.EcodeInvalidField,
  428. `invalid value for "wait"`,
  429. )
  430. }
  431. // TODO(jonboulle): define what parameters dir is/isn't compatible with?
  432. if dir, err = getBool(r.Form, "dir"); err != nil {
  433. return emptyReq, etcdErr.NewRequestError(
  434. etcdErr.EcodeInvalidField,
  435. `invalid value for "dir"`,
  436. )
  437. }
  438. if quorum, err = getBool(r.Form, "quorum"); err != nil {
  439. return emptyReq, etcdErr.NewRequestError(
  440. etcdErr.EcodeInvalidField,
  441. `invalid value for "quorum"`,
  442. )
  443. }
  444. if stream, err = getBool(r.Form, "stream"); err != nil {
  445. return emptyReq, etcdErr.NewRequestError(
  446. etcdErr.EcodeInvalidField,
  447. `invalid value for "stream"`,
  448. )
  449. }
  450. if wait && r.Method != "GET" {
  451. return emptyReq, etcdErr.NewRequestError(
  452. etcdErr.EcodeInvalidField,
  453. `"wait" can only be used with GET requests`,
  454. )
  455. }
  456. pV := r.FormValue("prevValue")
  457. if _, ok := r.Form["prevValue"]; ok && pV == "" {
  458. return emptyReq, etcdErr.NewRequestError(
  459. etcdErr.EcodePrevValueRequired,
  460. `"prevValue" cannot be empty`,
  461. )
  462. }
  463. // TTL is nullable, so leave it null if not specified
  464. // or an empty string
  465. var ttl *uint64
  466. if len(r.FormValue("ttl")) > 0 {
  467. i, err := getUint64(r.Form, "ttl")
  468. if err != nil {
  469. return emptyReq, etcdErr.NewRequestError(
  470. etcdErr.EcodeTTLNaN,
  471. `invalid value for "ttl"`,
  472. )
  473. }
  474. ttl = &i
  475. }
  476. // prevExist is nullable, so leave it null if not specified
  477. var pe *bool
  478. if _, ok := r.Form["prevExist"]; ok {
  479. bv, err := getBool(r.Form, "prevExist")
  480. if err != nil {
  481. return emptyReq, etcdErr.NewRequestError(
  482. etcdErr.EcodeInvalidField,
  483. "invalid value for prevExist",
  484. )
  485. }
  486. pe = &bv
  487. }
  488. rr := etcdserverpb.Request{
  489. Method: r.Method,
  490. Path: p,
  491. Val: r.FormValue("value"),
  492. Dir: dir,
  493. PrevValue: pV,
  494. PrevIndex: pIdx,
  495. PrevExist: pe,
  496. Wait: wait,
  497. Since: wIdx,
  498. Recursive: rec,
  499. Sorted: sort,
  500. Quorum: quorum,
  501. Stream: stream,
  502. }
  503. if pe != nil {
  504. rr.PrevExist = pe
  505. }
  506. // Null TTL is equivalent to unset Expiration
  507. if ttl != nil {
  508. expr := time.Duration(*ttl) * time.Second
  509. rr.Expiration = clock.Now().Add(expr).UnixNano()
  510. }
  511. return rr, nil
  512. }
  513. // writeKeyEvent trims the prefix of key path in a single Event under
  514. // StoreKeysPrefix, serializes it and writes the resulting JSON to the given
  515. // ResponseWriter, along with the appropriate headers.
  516. func writeKeyEvent(w http.ResponseWriter, ev *store.Event, rt etcdserver.RaftTimer) error {
  517. if ev == nil {
  518. return errors.New("cannot write empty Event!")
  519. }
  520. w.Header().Set("Content-Type", "application/json")
  521. w.Header().Set("X-Etcd-Index", fmt.Sprint(ev.EtcdIndex))
  522. w.Header().Set("X-Raft-Index", fmt.Sprint(rt.Index()))
  523. w.Header().Set("X-Raft-Term", fmt.Sprint(rt.Term()))
  524. if ev.IsCreated() {
  525. w.WriteHeader(http.StatusCreated)
  526. }
  527. ev = trimEventPrefix(ev, etcdserver.StoreKeysPrefix)
  528. return json.NewEncoder(w).Encode(ev)
  529. }
  530. func writeKeyNoAuth(w http.ResponseWriter) {
  531. e := etcdErr.NewError(etcdErr.EcodeUnauthorized, "Insufficient credentials", 0)
  532. e.WriteTo(w)
  533. }
  534. // writeKeyError logs and writes the given Error to the ResponseWriter.
  535. // If Error is not an etcdErr, the error will be converted to an etcd error.
  536. func writeKeyError(w http.ResponseWriter, err error) {
  537. if err == nil {
  538. return
  539. }
  540. switch e := err.(type) {
  541. case *etcdErr.Error:
  542. e.WriteTo(w)
  543. default:
  544. switch err {
  545. case etcdserver.ErrTimeoutDueToLeaderFail, etcdserver.ErrTimeoutDueToConnectionLost:
  546. plog.Error(err)
  547. default:
  548. plog.Errorf("got unexpected response error (%v)", err)
  549. }
  550. ee := etcdErr.NewError(etcdErr.EcodeRaftInternal, err.Error(), 0)
  551. ee.WriteTo(w)
  552. }
  553. }
  554. func handleKeyWatch(ctx context.Context, w http.ResponseWriter, wa store.Watcher, stream bool, rt etcdserver.RaftTimer) {
  555. defer wa.Remove()
  556. ech := wa.EventChan()
  557. var nch <-chan bool
  558. if x, ok := w.(http.CloseNotifier); ok {
  559. nch = x.CloseNotify()
  560. }
  561. w.Header().Set("Content-Type", "application/json")
  562. w.Header().Set("X-Etcd-Index", fmt.Sprint(wa.StartIndex()))
  563. w.Header().Set("X-Raft-Index", fmt.Sprint(rt.Index()))
  564. w.Header().Set("X-Raft-Term", fmt.Sprint(rt.Term()))
  565. w.WriteHeader(http.StatusOK)
  566. // Ensure headers are flushed early, in case of long polling
  567. w.(http.Flusher).Flush()
  568. for {
  569. select {
  570. case <-nch:
  571. // Client closed connection. Nothing to do.
  572. return
  573. case <-ctx.Done():
  574. // Timed out. net/http will close the connection for us, so nothing to do.
  575. return
  576. case ev, ok := <-ech:
  577. if !ok {
  578. // If the channel is closed this may be an indication of
  579. // that notifications are much more than we are able to
  580. // send to the client in time. Then we simply end streaming.
  581. return
  582. }
  583. ev = trimEventPrefix(ev, etcdserver.StoreKeysPrefix)
  584. if err := json.NewEncoder(w).Encode(ev); err != nil {
  585. // Should never be reached
  586. plog.Warningf("error writing event (%v)", err)
  587. return
  588. }
  589. if !stream {
  590. return
  591. }
  592. w.(http.Flusher).Flush()
  593. }
  594. }
  595. }
  596. func trimEventPrefix(ev *store.Event, prefix string) *store.Event {
  597. if ev == nil {
  598. return nil
  599. }
  600. // Since the *Event may reference one in the store history
  601. // history, we must copy it before modifying
  602. e := ev.Clone()
  603. e.Node = trimNodeExternPrefix(e.Node, prefix)
  604. e.PrevNode = trimNodeExternPrefix(e.PrevNode, prefix)
  605. return e
  606. }
  607. func trimNodeExternPrefix(n *store.NodeExtern, prefix string) *store.NodeExtern {
  608. if n == nil {
  609. return nil
  610. }
  611. n.Key = strings.TrimPrefix(n.Key, prefix)
  612. for _, nn := range n.Nodes {
  613. nn = trimNodeExternPrefix(nn, prefix)
  614. }
  615. return n
  616. }
  617. func trimErrorPrefix(err error, prefix string) error {
  618. if e, ok := err.(*etcdErr.Error); ok {
  619. e.Cause = strings.TrimPrefix(e.Cause, prefix)
  620. }
  621. return err
  622. }
  623. func unmarshalRequest(r *http.Request, req json.Unmarshaler, w http.ResponseWriter) bool {
  624. ctype := r.Header.Get("Content-Type")
  625. if ctype != "application/json" {
  626. writeError(w, r, httptypes.NewHTTPError(http.StatusUnsupportedMediaType, fmt.Sprintf("Bad Content-Type %s, accept application/json", ctype)))
  627. return false
  628. }
  629. b, err := ioutil.ReadAll(r.Body)
  630. if err != nil {
  631. writeError(w, r, httptypes.NewHTTPError(http.StatusBadRequest, err.Error()))
  632. return false
  633. }
  634. if err := req.UnmarshalJSON(b); err != nil {
  635. writeError(w, r, httptypes.NewHTTPError(http.StatusBadRequest, err.Error()))
  636. return false
  637. }
  638. return true
  639. }
  640. func getID(p string, w http.ResponseWriter) (types.ID, bool) {
  641. idStr := trimPrefix(p, membersPrefix)
  642. if idStr == "" {
  643. http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
  644. return 0, false
  645. }
  646. id, err := types.IDFromString(idStr)
  647. if err != nil {
  648. writeError(w, nil, httptypes.NewHTTPError(http.StatusNotFound, fmt.Sprintf("No such member: %s", idStr)))
  649. return 0, false
  650. }
  651. return id, true
  652. }
  653. // getUint64 extracts a uint64 by the given key from a Form. If the key does
  654. // not exist in the form, 0 is returned. If the key exists but the value is
  655. // badly formed, an error is returned. If multiple values are present only the
  656. // first is considered.
  657. func getUint64(form url.Values, key string) (i uint64, err error) {
  658. if vals, ok := form[key]; ok {
  659. i, err = strconv.ParseUint(vals[0], 10, 64)
  660. }
  661. return
  662. }
  663. // getBool extracts a bool by the given key from a Form. If the key does not
  664. // exist in the form, false is returned. If the key exists but the value is
  665. // badly formed, an error is returned. If multiple values are present only the
  666. // first is considered.
  667. func getBool(form url.Values, key string) (b bool, err error) {
  668. if vals, ok := form[key]; ok {
  669. b, err = strconv.ParseBool(vals[0])
  670. }
  671. return
  672. }
  673. // trimPrefix removes a given prefix and any slash following the prefix
  674. // e.g.: trimPrefix("foo", "foo") == trimPrefix("foo/", "foo") == ""
  675. func trimPrefix(p, prefix string) (s string) {
  676. s = strings.TrimPrefix(p, prefix)
  677. s = strings.TrimPrefix(s, "/")
  678. return
  679. }
  680. func newMemberCollection(ms []*etcdserver.Member) *httptypes.MemberCollection {
  681. c := httptypes.MemberCollection(make([]httptypes.Member, len(ms)))
  682. for i, m := range ms {
  683. c[i] = newMember(m)
  684. }
  685. return &c
  686. }
  687. func newMember(m *etcdserver.Member) httptypes.Member {
  688. tm := httptypes.Member{
  689. ID: m.ID.String(),
  690. Name: m.Name,
  691. PeerURLs: make([]string, len(m.PeerURLs)),
  692. ClientURLs: make([]string, len(m.ClientURLs)),
  693. }
  694. copy(tm.PeerURLs, m.PeerURLs)
  695. copy(tm.ClientURLs, m.ClientURLs)
  696. return tm
  697. }