client.go 23 KB

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