client.go 22 KB

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