mirror of
https://github.com/etcd-io/etcd.git
synced 2024-09-27 06:25:44 +00:00
604bc04f70
Added IsLearner field to etcdserver internal Member type. Routed learner MemberAdd request from server API to raft. Apply learner MemberAdd result to server after the request is passed through Raft.
789 lines
22 KiB
Go
789 lines
22 KiB
Go
// Copyright 2015 The etcd Authors
|
|
//
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
package v2http
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io/ioutil"
|
|
"net/http"
|
|
"net/url"
|
|
"path"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"go.etcd.io/etcd/v3/etcdserver"
|
|
"go.etcd.io/etcd/v3/etcdserver/api"
|
|
"go.etcd.io/etcd/v3/etcdserver/api/etcdhttp"
|
|
"go.etcd.io/etcd/v3/etcdserver/api/membership"
|
|
"go.etcd.io/etcd/v3/etcdserver/api/v2auth"
|
|
"go.etcd.io/etcd/v3/etcdserver/api/v2error"
|
|
"go.etcd.io/etcd/v3/etcdserver/api/v2http/httptypes"
|
|
stats "go.etcd.io/etcd/v3/etcdserver/api/v2stats"
|
|
"go.etcd.io/etcd/v3/etcdserver/api/v2store"
|
|
"go.etcd.io/etcd/v3/etcdserver/etcdserverpb"
|
|
"go.etcd.io/etcd/v3/pkg/types"
|
|
|
|
"github.com/jonboulle/clockwork"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
const (
|
|
authPrefix = "/v2/auth"
|
|
keysPrefix = "/v2/keys"
|
|
machinesPrefix = "/v2/machines"
|
|
membersPrefix = "/v2/members"
|
|
statsPrefix = "/v2/stats"
|
|
)
|
|
|
|
// NewClientHandler generates a muxed http.Handler with the given parameters to serve etcd client requests.
|
|
func NewClientHandler(lg *zap.Logger, server etcdserver.ServerPeer, timeout time.Duration) http.Handler {
|
|
mux := http.NewServeMux()
|
|
etcdhttp.HandleBasic(mux, server)
|
|
handleV2(lg, mux, server, timeout)
|
|
return requestLogger(lg, mux)
|
|
}
|
|
|
|
func handleV2(lg *zap.Logger, mux *http.ServeMux, server etcdserver.ServerV2, timeout time.Duration) {
|
|
sec := v2auth.NewStore(lg, server, timeout)
|
|
kh := &keysHandler{
|
|
lg: lg,
|
|
sec: sec,
|
|
server: server,
|
|
cluster: server.Cluster(),
|
|
timeout: timeout,
|
|
clientCertAuthEnabled: server.ClientCertAuthEnabled(),
|
|
}
|
|
|
|
sh := &statsHandler{
|
|
lg: lg,
|
|
stats: server,
|
|
}
|
|
|
|
mh := &membersHandler{
|
|
lg: lg,
|
|
sec: sec,
|
|
server: server,
|
|
cluster: server.Cluster(),
|
|
timeout: timeout,
|
|
clock: clockwork.NewRealClock(),
|
|
clientCertAuthEnabled: server.ClientCertAuthEnabled(),
|
|
}
|
|
|
|
mah := &machinesHandler{cluster: server.Cluster()}
|
|
|
|
sech := &authHandler{
|
|
lg: lg,
|
|
sec: sec,
|
|
cluster: server.Cluster(),
|
|
clientCertAuthEnabled: server.ClientCertAuthEnabled(),
|
|
}
|
|
mux.HandleFunc("/", http.NotFound)
|
|
mux.Handle(keysPrefix, kh)
|
|
mux.Handle(keysPrefix+"/", kh)
|
|
mux.HandleFunc(statsPrefix+"/store", sh.serveStore)
|
|
mux.HandleFunc(statsPrefix+"/self", sh.serveSelf)
|
|
mux.HandleFunc(statsPrefix+"/leader", sh.serveLeader)
|
|
mux.Handle(membersPrefix, mh)
|
|
mux.Handle(membersPrefix+"/", mh)
|
|
mux.Handle(machinesPrefix, mah)
|
|
handleAuth(mux, sech)
|
|
}
|
|
|
|
type keysHandler struct {
|
|
lg *zap.Logger
|
|
sec v2auth.Store
|
|
server etcdserver.ServerV2
|
|
cluster api.Cluster
|
|
timeout time.Duration
|
|
clientCertAuthEnabled bool
|
|
}
|
|
|
|
func (h *keysHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|
if !allowMethod(w, r.Method, "HEAD", "GET", "PUT", "POST", "DELETE") {
|
|
return
|
|
}
|
|
|
|
w.Header().Set("X-Etcd-Cluster-ID", h.cluster.ID().String())
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), h.timeout)
|
|
defer cancel()
|
|
clock := clockwork.NewRealClock()
|
|
startTime := clock.Now()
|
|
rr, noValueOnSuccess, err := parseKeyRequest(r, clock)
|
|
if err != nil {
|
|
writeKeyError(h.lg, w, err)
|
|
return
|
|
}
|
|
// The path must be valid at this point (we've parsed the request successfully).
|
|
if !hasKeyPrefixAccess(h.lg, h.sec, r, r.URL.Path[len(keysPrefix):], rr.Recursive, h.clientCertAuthEnabled) {
|
|
writeKeyNoAuth(w)
|
|
return
|
|
}
|
|
if !rr.Wait {
|
|
reportRequestReceived(rr)
|
|
}
|
|
resp, err := h.server.Do(ctx, rr)
|
|
if err != nil {
|
|
err = trimErrorPrefix(err, etcdserver.StoreKeysPrefix)
|
|
writeKeyError(h.lg, w, err)
|
|
reportRequestFailed(rr, err)
|
|
return
|
|
}
|
|
switch {
|
|
case resp.Event != nil:
|
|
if err := writeKeyEvent(w, resp, noValueOnSuccess); err != nil {
|
|
// Should never be reached
|
|
if h.lg != nil {
|
|
h.lg.Warn("failed to write key event", zap.Error(err))
|
|
} else {
|
|
plog.Errorf("error writing event (%v)", err)
|
|
}
|
|
}
|
|
reportRequestCompleted(rr, startTime)
|
|
case resp.Watcher != nil:
|
|
ctx, cancel := context.WithTimeout(context.Background(), defaultWatchTimeout)
|
|
defer cancel()
|
|
handleKeyWatch(ctx, h.lg, w, resp, rr.Stream)
|
|
default:
|
|
writeKeyError(h.lg, w, errors.New("received response with no Event/Watcher"))
|
|
}
|
|
}
|
|
|
|
type machinesHandler struct {
|
|
cluster api.Cluster
|
|
}
|
|
|
|
func (h *machinesHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|
if !allowMethod(w, r.Method, "GET", "HEAD") {
|
|
return
|
|
}
|
|
endpoints := h.cluster.ClientURLs()
|
|
w.Write([]byte(strings.Join(endpoints, ", ")))
|
|
}
|
|
|
|
type membersHandler struct {
|
|
lg *zap.Logger
|
|
sec v2auth.Store
|
|
server etcdserver.ServerV2
|
|
cluster api.Cluster
|
|
timeout time.Duration
|
|
clock clockwork.Clock
|
|
clientCertAuthEnabled bool
|
|
}
|
|
|
|
func (h *membersHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|
if !allowMethod(w, r.Method, "GET", "POST", "DELETE", "PUT") {
|
|
return
|
|
}
|
|
if !hasWriteRootAccess(h.lg, h.sec, r, h.clientCertAuthEnabled) {
|
|
writeNoAuth(h.lg, w, r)
|
|
return
|
|
}
|
|
w.Header().Set("X-Etcd-Cluster-ID", h.cluster.ID().String())
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), h.timeout)
|
|
defer cancel()
|
|
|
|
switch r.Method {
|
|
case "GET":
|
|
switch trimPrefix(r.URL.Path, membersPrefix) {
|
|
case "":
|
|
mc := newMemberCollection(h.cluster.Members())
|
|
w.Header().Set("Content-Type", "application/json")
|
|
if err := json.NewEncoder(w).Encode(mc); err != nil {
|
|
if h.lg != nil {
|
|
h.lg.Warn("failed to encode members response", zap.Error(err))
|
|
} else {
|
|
plog.Warningf("failed to encode members response (%v)", err)
|
|
}
|
|
}
|
|
case "leader":
|
|
id := h.server.Leader()
|
|
if id == 0 {
|
|
writeError(h.lg, w, r, httptypes.NewHTTPError(http.StatusServiceUnavailable, "During election"))
|
|
return
|
|
}
|
|
m := newMember(h.cluster.Member(id))
|
|
w.Header().Set("Content-Type", "application/json")
|
|
if err := json.NewEncoder(w).Encode(m); err != nil {
|
|
if h.lg != nil {
|
|
h.lg.Warn("failed to encode members response", zap.Error(err))
|
|
} else {
|
|
plog.Warningf("failed to encode members response (%v)", err)
|
|
}
|
|
}
|
|
default:
|
|
writeError(h.lg, w, r, httptypes.NewHTTPError(http.StatusNotFound, "Not found"))
|
|
}
|
|
|
|
case "POST":
|
|
req := httptypes.MemberCreateRequest{}
|
|
if ok := unmarshalRequest(h.lg, r, &req, w); !ok {
|
|
return
|
|
}
|
|
now := h.clock.Now()
|
|
m := membership.NewMember("", req.PeerURLs, "", &now, false) // does not support adding learner via v2http
|
|
_, err := h.server.AddMember(ctx, *m)
|
|
switch {
|
|
case err == membership.ErrIDExists || err == membership.ErrPeerURLexists:
|
|
writeError(h.lg, w, r, httptypes.NewHTTPError(http.StatusConflict, err.Error()))
|
|
return
|
|
case err != nil:
|
|
if h.lg != nil {
|
|
h.lg.Warn(
|
|
"failed to add a member",
|
|
zap.String("member-id", m.ID.String()),
|
|
zap.Error(err),
|
|
)
|
|
} else {
|
|
plog.Errorf("error adding member %s (%v)", m.ID, err)
|
|
}
|
|
writeError(h.lg, w, r, err)
|
|
return
|
|
}
|
|
res := newMember(m)
|
|
w.Header().Set("Content-Type", "application/json")
|
|
w.WriteHeader(http.StatusCreated)
|
|
if err := json.NewEncoder(w).Encode(res); err != nil {
|
|
if h.lg != nil {
|
|
h.lg.Warn("failed to encode members response", zap.Error(err))
|
|
} else {
|
|
plog.Warningf("failed to encode members response (%v)", err)
|
|
}
|
|
}
|
|
|
|
case "DELETE":
|
|
id, ok := getID(h.lg, r.URL.Path, w)
|
|
if !ok {
|
|
return
|
|
}
|
|
_, err := h.server.RemoveMember(ctx, uint64(id))
|
|
switch {
|
|
case err == membership.ErrIDRemoved:
|
|
writeError(h.lg, w, r, httptypes.NewHTTPError(http.StatusGone, fmt.Sprintf("Member permanently removed: %s", id)))
|
|
case err == membership.ErrIDNotFound:
|
|
writeError(h.lg, w, r, httptypes.NewHTTPError(http.StatusNotFound, fmt.Sprintf("No such member: %s", id)))
|
|
case err != nil:
|
|
if h.lg != nil {
|
|
h.lg.Warn(
|
|
"failed to remove a member",
|
|
zap.String("member-id", id.String()),
|
|
zap.Error(err),
|
|
)
|
|
} else {
|
|
plog.Errorf("error removing member %s (%v)", id, err)
|
|
}
|
|
writeError(h.lg, w, r, err)
|
|
default:
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}
|
|
|
|
case "PUT":
|
|
id, ok := getID(h.lg, r.URL.Path, w)
|
|
if !ok {
|
|
return
|
|
}
|
|
req := httptypes.MemberUpdateRequest{}
|
|
if ok := unmarshalRequest(h.lg, r, &req, w); !ok {
|
|
return
|
|
}
|
|
m := membership.Member{
|
|
ID: id,
|
|
RaftAttributes: membership.RaftAttributes{PeerURLs: req.PeerURLs.StringSlice()},
|
|
}
|
|
_, err := h.server.UpdateMember(ctx, m)
|
|
switch {
|
|
case err == membership.ErrPeerURLexists:
|
|
writeError(h.lg, w, r, httptypes.NewHTTPError(http.StatusConflict, err.Error()))
|
|
case err == membership.ErrIDNotFound:
|
|
writeError(h.lg, w, r, httptypes.NewHTTPError(http.StatusNotFound, fmt.Sprintf("No such member: %s", id)))
|
|
case err != nil:
|
|
if h.lg != nil {
|
|
h.lg.Warn(
|
|
"failed to update a member",
|
|
zap.String("member-id", m.ID.String()),
|
|
zap.Error(err),
|
|
)
|
|
} else {
|
|
plog.Errorf("error updating member %s (%v)", m.ID, err)
|
|
}
|
|
writeError(h.lg, w, r, err)
|
|
default:
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}
|
|
}
|
|
}
|
|
|
|
type statsHandler struct {
|
|
lg *zap.Logger
|
|
stats stats.Stats
|
|
}
|
|
|
|
func (h *statsHandler) serveStore(w http.ResponseWriter, r *http.Request) {
|
|
if !allowMethod(w, r.Method, "GET") {
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
w.Write(h.stats.StoreStats())
|
|
}
|
|
|
|
func (h *statsHandler) serveSelf(w http.ResponseWriter, r *http.Request) {
|
|
if !allowMethod(w, r.Method, "GET") {
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
w.Write(h.stats.SelfStats())
|
|
}
|
|
|
|
func (h *statsHandler) serveLeader(w http.ResponseWriter, r *http.Request) {
|
|
if !allowMethod(w, r.Method, "GET") {
|
|
return
|
|
}
|
|
stats := h.stats.LeaderStats()
|
|
if stats == nil {
|
|
etcdhttp.WriteError(h.lg, w, r, httptypes.NewHTTPError(http.StatusForbidden, "not current leader"))
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
w.Write(stats)
|
|
}
|
|
|
|
// parseKeyRequest converts a received http.Request on keysPrefix to
|
|
// a server Request, performing validation of supplied fields as appropriate.
|
|
// If any validation fails, an empty Request and non-nil error is returned.
|
|
func parseKeyRequest(r *http.Request, clock clockwork.Clock) (etcdserverpb.Request, bool, error) {
|
|
var noValueOnSuccess bool
|
|
emptyReq := etcdserverpb.Request{}
|
|
|
|
err := r.ParseForm()
|
|
if err != nil {
|
|
return emptyReq, false, v2error.NewRequestError(
|
|
v2error.EcodeInvalidForm,
|
|
err.Error(),
|
|
)
|
|
}
|
|
|
|
if !strings.HasPrefix(r.URL.Path, keysPrefix) {
|
|
return emptyReq, false, v2error.NewRequestError(
|
|
v2error.EcodeInvalidForm,
|
|
"incorrect key prefix",
|
|
)
|
|
}
|
|
p := path.Join(etcdserver.StoreKeysPrefix, r.URL.Path[len(keysPrefix):])
|
|
|
|
var pIdx, wIdx uint64
|
|
if pIdx, err = getUint64(r.Form, "prevIndex"); err != nil {
|
|
return emptyReq, false, v2error.NewRequestError(
|
|
v2error.EcodeIndexNaN,
|
|
`invalid value for "prevIndex"`,
|
|
)
|
|
}
|
|
if wIdx, err = getUint64(r.Form, "waitIndex"); err != nil {
|
|
return emptyReq, false, v2error.NewRequestError(
|
|
v2error.EcodeIndexNaN,
|
|
`invalid value for "waitIndex"`,
|
|
)
|
|
}
|
|
|
|
var rec, sort, wait, dir, quorum, stream bool
|
|
if rec, err = getBool(r.Form, "recursive"); err != nil {
|
|
return emptyReq, false, v2error.NewRequestError(
|
|
v2error.EcodeInvalidField,
|
|
`invalid value for "recursive"`,
|
|
)
|
|
}
|
|
if sort, err = getBool(r.Form, "sorted"); err != nil {
|
|
return emptyReq, false, v2error.NewRequestError(
|
|
v2error.EcodeInvalidField,
|
|
`invalid value for "sorted"`,
|
|
)
|
|
}
|
|
if wait, err = getBool(r.Form, "wait"); err != nil {
|
|
return emptyReq, false, v2error.NewRequestError(
|
|
v2error.EcodeInvalidField,
|
|
`invalid value for "wait"`,
|
|
)
|
|
}
|
|
// TODO(jonboulle): define what parameters dir is/isn't compatible with?
|
|
if dir, err = getBool(r.Form, "dir"); err != nil {
|
|
return emptyReq, false, v2error.NewRequestError(
|
|
v2error.EcodeInvalidField,
|
|
`invalid value for "dir"`,
|
|
)
|
|
}
|
|
if quorum, err = getBool(r.Form, "quorum"); err != nil {
|
|
return emptyReq, false, v2error.NewRequestError(
|
|
v2error.EcodeInvalidField,
|
|
`invalid value for "quorum"`,
|
|
)
|
|
}
|
|
if stream, err = getBool(r.Form, "stream"); err != nil {
|
|
return emptyReq, false, v2error.NewRequestError(
|
|
v2error.EcodeInvalidField,
|
|
`invalid value for "stream"`,
|
|
)
|
|
}
|
|
|
|
if wait && r.Method != "GET" {
|
|
return emptyReq, false, v2error.NewRequestError(
|
|
v2error.EcodeInvalidField,
|
|
`"wait" can only be used with GET requests`,
|
|
)
|
|
}
|
|
|
|
pV := r.FormValue("prevValue")
|
|
if _, ok := r.Form["prevValue"]; ok && pV == "" {
|
|
return emptyReq, false, v2error.NewRequestError(
|
|
v2error.EcodePrevValueRequired,
|
|
`"prevValue" cannot be empty`,
|
|
)
|
|
}
|
|
|
|
if noValueOnSuccess, err = getBool(r.Form, "noValueOnSuccess"); err != nil {
|
|
return emptyReq, false, v2error.NewRequestError(
|
|
v2error.EcodeInvalidField,
|
|
`invalid value for "noValueOnSuccess"`,
|
|
)
|
|
}
|
|
|
|
// TTL is nullable, so leave it null if not specified
|
|
// or an empty string
|
|
var ttl *uint64
|
|
if len(r.FormValue("ttl")) > 0 {
|
|
i, err := getUint64(r.Form, "ttl")
|
|
if err != nil {
|
|
return emptyReq, false, v2error.NewRequestError(
|
|
v2error.EcodeTTLNaN,
|
|
`invalid value for "ttl"`,
|
|
)
|
|
}
|
|
ttl = &i
|
|
}
|
|
|
|
// prevExist is nullable, so leave it null if not specified
|
|
var pe *bool
|
|
if _, ok := r.Form["prevExist"]; ok {
|
|
bv, err := getBool(r.Form, "prevExist")
|
|
if err != nil {
|
|
return emptyReq, false, v2error.NewRequestError(
|
|
v2error.EcodeInvalidField,
|
|
"invalid value for prevExist",
|
|
)
|
|
}
|
|
pe = &bv
|
|
}
|
|
|
|
// refresh is nullable, so leave it null if not specified
|
|
var refresh *bool
|
|
if _, ok := r.Form["refresh"]; ok {
|
|
bv, err := getBool(r.Form, "refresh")
|
|
if err != nil {
|
|
return emptyReq, false, v2error.NewRequestError(
|
|
v2error.EcodeInvalidField,
|
|
"invalid value for refresh",
|
|
)
|
|
}
|
|
refresh = &bv
|
|
if refresh != nil && *refresh {
|
|
val := r.FormValue("value")
|
|
if _, ok := r.Form["value"]; ok && val != "" {
|
|
return emptyReq, false, v2error.NewRequestError(
|
|
v2error.EcodeRefreshValue,
|
|
`A value was provided on a refresh`,
|
|
)
|
|
}
|
|
if ttl == nil {
|
|
return emptyReq, false, v2error.NewRequestError(
|
|
v2error.EcodeRefreshTTLRequired,
|
|
`No TTL value set`,
|
|
)
|
|
}
|
|
}
|
|
}
|
|
|
|
rr := etcdserverpb.Request{
|
|
Method: r.Method,
|
|
Path: p,
|
|
Val: r.FormValue("value"),
|
|
Dir: dir,
|
|
PrevValue: pV,
|
|
PrevIndex: pIdx,
|
|
PrevExist: pe,
|
|
Wait: wait,
|
|
Since: wIdx,
|
|
Recursive: rec,
|
|
Sorted: sort,
|
|
Quorum: quorum,
|
|
Stream: stream,
|
|
}
|
|
|
|
if pe != nil {
|
|
rr.PrevExist = pe
|
|
}
|
|
|
|
if refresh != nil {
|
|
rr.Refresh = refresh
|
|
}
|
|
|
|
// Null TTL is equivalent to unset Expiration
|
|
if ttl != nil {
|
|
expr := time.Duration(*ttl) * time.Second
|
|
rr.Expiration = clock.Now().Add(expr).UnixNano()
|
|
}
|
|
|
|
return rr, noValueOnSuccess, nil
|
|
}
|
|
|
|
// writeKeyEvent trims the prefix of key path in a single Event under
|
|
// StoreKeysPrefix, serializes it and writes the resulting JSON to the given
|
|
// ResponseWriter, along with the appropriate headers.
|
|
func writeKeyEvent(w http.ResponseWriter, resp etcdserver.Response, noValueOnSuccess bool) error {
|
|
ev := resp.Event
|
|
if ev == nil {
|
|
return errors.New("cannot write empty Event")
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
w.Header().Set("X-Etcd-Index", fmt.Sprint(ev.EtcdIndex))
|
|
w.Header().Set("X-Raft-Index", fmt.Sprint(resp.Index))
|
|
w.Header().Set("X-Raft-Term", fmt.Sprint(resp.Term))
|
|
|
|
if ev.IsCreated() {
|
|
w.WriteHeader(http.StatusCreated)
|
|
}
|
|
|
|
ev = trimEventPrefix(ev, etcdserver.StoreKeysPrefix)
|
|
if noValueOnSuccess &&
|
|
(ev.Action == v2store.Set || ev.Action == v2store.CompareAndSwap ||
|
|
ev.Action == v2store.Create || ev.Action == v2store.Update) {
|
|
ev.Node = nil
|
|
ev.PrevNode = nil
|
|
}
|
|
return json.NewEncoder(w).Encode(ev)
|
|
}
|
|
|
|
func writeKeyNoAuth(w http.ResponseWriter) {
|
|
e := v2error.NewError(v2error.EcodeUnauthorized, "Insufficient credentials", 0)
|
|
e.WriteTo(w)
|
|
}
|
|
|
|
// writeKeyError logs and writes the given Error to the ResponseWriter.
|
|
// If Error is not an etcdErr, the error will be converted to an etcd error.
|
|
func writeKeyError(lg *zap.Logger, w http.ResponseWriter, err error) {
|
|
if err == nil {
|
|
return
|
|
}
|
|
switch e := err.(type) {
|
|
case *v2error.Error:
|
|
e.WriteTo(w)
|
|
default:
|
|
switch err {
|
|
case etcdserver.ErrTimeoutDueToLeaderFail, etcdserver.ErrTimeoutDueToConnectionLost:
|
|
if lg != nil {
|
|
lg.Warn(
|
|
"v2 response error",
|
|
zap.String("internal-server-error", err.Error()),
|
|
)
|
|
} else {
|
|
mlog.MergeError(err)
|
|
}
|
|
default:
|
|
if lg != nil {
|
|
lg.Warn(
|
|
"unexpected v2 response error",
|
|
zap.String("internal-server-error", err.Error()),
|
|
)
|
|
} else {
|
|
mlog.MergeErrorf("got unexpected response error (%v)", err)
|
|
}
|
|
}
|
|
ee := v2error.NewError(v2error.EcodeRaftInternal, err.Error(), 0)
|
|
ee.WriteTo(w)
|
|
}
|
|
}
|
|
|
|
func handleKeyWatch(ctx context.Context, lg *zap.Logger, w http.ResponseWriter, resp etcdserver.Response, stream bool) {
|
|
wa := resp.Watcher
|
|
defer wa.Remove()
|
|
ech := wa.EventChan()
|
|
var nch <-chan bool
|
|
if x, ok := w.(http.CloseNotifier); ok {
|
|
nch = x.CloseNotify()
|
|
}
|
|
|
|
w.Header().Set("Content-Type", "application/json")
|
|
w.Header().Set("X-Etcd-Index", fmt.Sprint(wa.StartIndex()))
|
|
w.Header().Set("X-Raft-Index", fmt.Sprint(resp.Index))
|
|
w.Header().Set("X-Raft-Term", fmt.Sprint(resp.Term))
|
|
w.WriteHeader(http.StatusOK)
|
|
|
|
// Ensure headers are flushed early, in case of long polling
|
|
w.(http.Flusher).Flush()
|
|
|
|
for {
|
|
select {
|
|
case <-nch:
|
|
// Client closed connection. Nothing to do.
|
|
return
|
|
case <-ctx.Done():
|
|
// Timed out. net/http will close the connection for us, so nothing to do.
|
|
return
|
|
case ev, ok := <-ech:
|
|
if !ok {
|
|
// If the channel is closed this may be an indication of
|
|
// that notifications are much more than we are able to
|
|
// send to the client in time. Then we simply end streaming.
|
|
return
|
|
}
|
|
ev = trimEventPrefix(ev, etcdserver.StoreKeysPrefix)
|
|
if err := json.NewEncoder(w).Encode(ev); err != nil {
|
|
// Should never be reached
|
|
if lg != nil {
|
|
lg.Warn("failed to encode event", zap.Error(err))
|
|
} else {
|
|
plog.Warningf("error writing event (%v)", err)
|
|
}
|
|
return
|
|
}
|
|
if !stream {
|
|
return
|
|
}
|
|
w.(http.Flusher).Flush()
|
|
}
|
|
}
|
|
}
|
|
|
|
func trimEventPrefix(ev *v2store.Event, prefix string) *v2store.Event {
|
|
if ev == nil {
|
|
return nil
|
|
}
|
|
// Since the *Event may reference one in the store history
|
|
// history, we must copy it before modifying
|
|
e := ev.Clone()
|
|
trimNodeExternPrefix(e.Node, prefix)
|
|
trimNodeExternPrefix(e.PrevNode, prefix)
|
|
return e
|
|
}
|
|
|
|
func trimNodeExternPrefix(n *v2store.NodeExtern, prefix string) {
|
|
if n == nil {
|
|
return
|
|
}
|
|
n.Key = strings.TrimPrefix(n.Key, prefix)
|
|
for _, nn := range n.Nodes {
|
|
trimNodeExternPrefix(nn, prefix)
|
|
}
|
|
}
|
|
|
|
func trimErrorPrefix(err error, prefix string) error {
|
|
if e, ok := err.(*v2error.Error); ok {
|
|
e.Cause = strings.TrimPrefix(e.Cause, prefix)
|
|
}
|
|
return err
|
|
}
|
|
|
|
func unmarshalRequest(lg *zap.Logger, r *http.Request, req json.Unmarshaler, w http.ResponseWriter) bool {
|
|
ctype := r.Header.Get("Content-Type")
|
|
semicolonPosition := strings.Index(ctype, ";")
|
|
if semicolonPosition != -1 {
|
|
ctype = strings.TrimSpace(strings.ToLower(ctype[0:semicolonPosition]))
|
|
}
|
|
if ctype != "application/json" {
|
|
writeError(lg, w, r, httptypes.NewHTTPError(http.StatusUnsupportedMediaType, fmt.Sprintf("Bad Content-Type %s, accept application/json", ctype)))
|
|
return false
|
|
}
|
|
b, err := ioutil.ReadAll(r.Body)
|
|
if err != nil {
|
|
writeError(lg, w, r, httptypes.NewHTTPError(http.StatusBadRequest, err.Error()))
|
|
return false
|
|
}
|
|
if err := req.UnmarshalJSON(b); err != nil {
|
|
writeError(lg, w, r, httptypes.NewHTTPError(http.StatusBadRequest, err.Error()))
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
|
|
func getID(lg *zap.Logger, p string, w http.ResponseWriter) (types.ID, bool) {
|
|
idStr := trimPrefix(p, membersPrefix)
|
|
if idStr == "" {
|
|
http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
|
|
return 0, false
|
|
}
|
|
id, err := types.IDFromString(idStr)
|
|
if err != nil {
|
|
writeError(lg, w, nil, httptypes.NewHTTPError(http.StatusNotFound, fmt.Sprintf("No such member: %s", idStr)))
|
|
return 0, false
|
|
}
|
|
return id, true
|
|
}
|
|
|
|
// getUint64 extracts a uint64 by the given key from a Form. If the key does
|
|
// not exist in the form, 0 is returned. If the key exists but the value is
|
|
// badly formed, an error is returned. If multiple values are present only the
|
|
// first is considered.
|
|
func getUint64(form url.Values, key string) (i uint64, err error) {
|
|
if vals, ok := form[key]; ok {
|
|
i, err = strconv.ParseUint(vals[0], 10, 64)
|
|
}
|
|
return
|
|
}
|
|
|
|
// getBool extracts a bool by the given key from a Form. If the key does not
|
|
// exist in the form, false is returned. If the key exists but the value is
|
|
// badly formed, an error is returned. If multiple values are present only the
|
|
// first is considered.
|
|
func getBool(form url.Values, key string) (b bool, err error) {
|
|
if vals, ok := form[key]; ok {
|
|
b, err = strconv.ParseBool(vals[0])
|
|
}
|
|
return
|
|
}
|
|
|
|
// trimPrefix removes a given prefix and any slash following the prefix
|
|
// e.g.: trimPrefix("foo", "foo") == trimPrefix("foo/", "foo") == ""
|
|
func trimPrefix(p, prefix string) (s string) {
|
|
s = strings.TrimPrefix(p, prefix)
|
|
s = strings.TrimPrefix(s, "/")
|
|
return
|
|
}
|
|
|
|
func newMemberCollection(ms []*membership.Member) *httptypes.MemberCollection {
|
|
c := httptypes.MemberCollection(make([]httptypes.Member, len(ms)))
|
|
|
|
for i, m := range ms {
|
|
c[i] = newMember(m)
|
|
}
|
|
|
|
return &c
|
|
}
|
|
|
|
func newMember(m *membership.Member) httptypes.Member {
|
|
tm := httptypes.Member{
|
|
ID: m.ID.String(),
|
|
Name: m.Name,
|
|
PeerURLs: make([]string, len(m.PeerURLs)),
|
|
ClientURLs: make([]string, len(m.ClientURLs)),
|
|
}
|
|
|
|
copy(tm.PeerURLs, m.PeerURLs)
|
|
copy(tm.ClientURLs, m.ClientURLs)
|
|
|
|
return tm
|
|
}
|