Dependencies for the affinity router and the
affinity routing daemon.
Change-Id: Icda72c3594ef7f8f0bc0c33dc03087a4c25529ca
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/capability.go b/vendor/github.com/coreos/etcd/etcdserver/api/capability.go
new file mode 100644
index 0000000..eb34383
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/capability.go
@@ -0,0 +1,87 @@
+// 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 api
+
+import (
+ "sync"
+
+ "github.com/coreos/etcd/version"
+ "github.com/coreos/go-semver/semver"
+ "github.com/coreos/pkg/capnslog"
+)
+
+type Capability string
+
+const (
+ AuthCapability Capability = "auth"
+ V3rpcCapability Capability = "v3rpc"
+)
+
+var (
+ plog = capnslog.NewPackageLogger("github.com/coreos/etcd", "etcdserver/api")
+
+ // capabilityMaps is a static map of version to capability map.
+ capabilityMaps = map[string]map[Capability]bool{
+ "3.0.0": {AuthCapability: true, V3rpcCapability: true},
+ "3.1.0": {AuthCapability: true, V3rpcCapability: true},
+ "3.2.0": {AuthCapability: true, V3rpcCapability: true},
+ "3.3.0": {AuthCapability: true, V3rpcCapability: true},
+ }
+
+ enableMapMu sync.RWMutex
+ // enabledMap points to a map in capabilityMaps
+ enabledMap map[Capability]bool
+
+ curVersion *semver.Version
+)
+
+func init() {
+ enabledMap = map[Capability]bool{
+ AuthCapability: true,
+ V3rpcCapability: true,
+ }
+}
+
+// UpdateCapability updates the enabledMap when the cluster version increases.
+func UpdateCapability(v *semver.Version) {
+ if v == nil {
+ // if recovered but version was never set by cluster
+ return
+ }
+ enableMapMu.Lock()
+ if curVersion != nil && !curVersion.LessThan(*v) {
+ enableMapMu.Unlock()
+ return
+ }
+ curVersion = v
+ enabledMap = capabilityMaps[curVersion.String()]
+ enableMapMu.Unlock()
+ plog.Infof("enabled capabilities for version %s", version.Cluster(v.String()))
+}
+
+func IsCapabilityEnabled(c Capability) bool {
+ enableMapMu.RLock()
+ defer enableMapMu.RUnlock()
+ if enabledMap == nil {
+ return false
+ }
+ return enabledMap[c]
+}
+
+func EnableCapability(c Capability) {
+ enableMapMu.Lock()
+ defer enableMapMu.Unlock()
+ enabledMap[c] = true
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/cluster.go b/vendor/github.com/coreos/etcd/etcdserver/api/cluster.go
new file mode 100644
index 0000000..654c258
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/cluster.go
@@ -0,0 +1,38 @@
+// Copyright 2016 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 api
+
+import (
+ "github.com/coreos/etcd/etcdserver/membership"
+ "github.com/coreos/etcd/pkg/types"
+
+ "github.com/coreos/go-semver/semver"
+)
+
+// Cluster is an interface representing a collection of members in one etcd cluster.
+type Cluster interface {
+ // ID returns the cluster ID
+ ID() types.ID
+ // ClientURLs returns an aggregate set of all URLs on which this
+ // cluster is listening for client requests
+ ClientURLs() []string
+ // Members returns a slice of members sorted by their ID
+ Members() []*membership.Member
+ // Member retrieves a particular member based on ID, or nil if the
+ // member does not exist in the cluster
+ Member(id types.ID) *membership.Member
+ // Version is the cluster-wide minimum major.minor version.
+ Version() *semver.Version
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/doc.go b/vendor/github.com/coreos/etcd/etcdserver/api/doc.go
new file mode 100644
index 0000000..f44881b
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/doc.go
@@ -0,0 +1,16 @@
+// Copyright 2016 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 api manages the capabilities and features that are exposed to clients by the etcd cluster.
+package api
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/etcdhttp/base.go b/vendor/github.com/coreos/etcd/etcdserver/api/etcdhttp/base.go
new file mode 100644
index 0000000..f0d3b0b
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/etcdhttp/base.go
@@ -0,0 +1,158 @@
+// 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 etcdhttp
+
+import (
+ "encoding/json"
+ "expvar"
+ "fmt"
+ "net/http"
+ "strings"
+
+ etcdErr "github.com/coreos/etcd/error"
+ "github.com/coreos/etcd/etcdserver"
+ "github.com/coreos/etcd/etcdserver/api"
+ "github.com/coreos/etcd/etcdserver/api/v2http/httptypes"
+ "github.com/coreos/etcd/pkg/logutil"
+ "github.com/coreos/etcd/version"
+ "github.com/coreos/pkg/capnslog"
+)
+
+var (
+ plog = capnslog.NewPackageLogger("github.com/coreos/etcd", "etcdserver/api/etcdhttp")
+ mlog = logutil.NewMergeLogger(plog)
+)
+
+const (
+ configPath = "/config"
+ varsPath = "/debug/vars"
+ versionPath = "/version"
+)
+
+// HandleBasic adds handlers to a mux for serving JSON etcd client requests
+// that do not access the v2 store.
+func HandleBasic(mux *http.ServeMux, server etcdserver.ServerPeer) {
+ mux.HandleFunc(varsPath, serveVars)
+ mux.HandleFunc(configPath+"/local/log", logHandleFunc)
+ HandleMetricsHealth(mux, server)
+ mux.HandleFunc(versionPath, versionHandler(server.Cluster(), serveVersion))
+}
+
+func versionHandler(c api.Cluster, fn func(http.ResponseWriter, *http.Request, string)) http.HandlerFunc {
+ return func(w http.ResponseWriter, r *http.Request) {
+ v := c.Version()
+ if v != nil {
+ fn(w, r, v.String())
+ } else {
+ fn(w, r, "not_decided")
+ }
+ }
+}
+
+func serveVersion(w http.ResponseWriter, r *http.Request, clusterV string) {
+ if !allowMethod(w, r, "GET") {
+ return
+ }
+ vs := version.Versions{
+ Server: version.Version,
+ Cluster: clusterV,
+ }
+
+ w.Header().Set("Content-Type", "application/json")
+ b, err := json.Marshal(&vs)
+ if err != nil {
+ plog.Panicf("cannot marshal versions to json (%v)", err)
+ }
+ w.Write(b)
+}
+
+func logHandleFunc(w http.ResponseWriter, r *http.Request) {
+ if !allowMethod(w, r, "PUT") {
+ return
+ }
+
+ in := struct{ Level string }{}
+
+ d := json.NewDecoder(r.Body)
+ if err := d.Decode(&in); err != nil {
+ WriteError(w, r, httptypes.NewHTTPError(http.StatusBadRequest, "Invalid json body"))
+ return
+ }
+
+ logl, err := capnslog.ParseLevel(strings.ToUpper(in.Level))
+ if err != nil {
+ WriteError(w, r, httptypes.NewHTTPError(http.StatusBadRequest, "Invalid log level "+in.Level))
+ return
+ }
+
+ plog.Noticef("globalLogLevel set to %q", logl.String())
+ capnslog.SetGlobalLogLevel(logl)
+ w.WriteHeader(http.StatusNoContent)
+}
+
+func serveVars(w http.ResponseWriter, r *http.Request) {
+ if !allowMethod(w, r, "GET") {
+ return
+ }
+
+ w.Header().Set("Content-Type", "application/json; charset=utf-8")
+ fmt.Fprintf(w, "{\n")
+ first := true
+ expvar.Do(func(kv expvar.KeyValue) {
+ if !first {
+ fmt.Fprintf(w, ",\n")
+ }
+ first = false
+ fmt.Fprintf(w, "%q: %s", kv.Key, kv.Value)
+ })
+ fmt.Fprintf(w, "\n}\n")
+}
+
+func allowMethod(w http.ResponseWriter, r *http.Request, m string) bool {
+ if m == r.Method {
+ return true
+ }
+ w.Header().Set("Allow", m)
+ http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
+ return false
+}
+
+// WriteError logs and writes the given Error to the ResponseWriter
+// If Error is an etcdErr, it is rendered to the ResponseWriter
+// Otherwise, it is assumed to be a StatusInternalServerError
+func WriteError(w http.ResponseWriter, r *http.Request, err error) {
+ if err == nil {
+ return
+ }
+ switch e := err.(type) {
+ case *etcdErr.Error:
+ e.WriteTo(w)
+ case *httptypes.HTTPError:
+ if et := e.WriteTo(w); et != nil {
+ plog.Debugf("error writing HTTPError (%v) to %s", et, r.RemoteAddr)
+ }
+ default:
+ switch err {
+ case etcdserver.ErrTimeoutDueToLeaderFail, etcdserver.ErrTimeoutDueToConnectionLost, etcdserver.ErrNotEnoughStartedMembers, etcdserver.ErrUnhealthy:
+ mlog.MergeError(err)
+ default:
+ mlog.MergeErrorf("got unexpected response error (%v)", err)
+ }
+ herr := httptypes.NewHTTPError(http.StatusInternalServerError, "Internal Server Error")
+ if et := herr.WriteTo(w); et != nil {
+ plog.Debugf("error writing HTTPError (%v) to %s", et, r.RemoteAddr)
+ }
+ }
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/etcdhttp/doc.go b/vendor/github.com/coreos/etcd/etcdserver/api/etcdhttp/doc.go
new file mode 100644
index 0000000..a03b626
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/etcdhttp/doc.go
@@ -0,0 +1,16 @@
+// Copyright 2017 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 etcdhttp implements HTTP transportation layer for etcdserver.
+package etcdhttp
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/etcdhttp/metrics.go b/vendor/github.com/coreos/etcd/etcdserver/api/etcdhttp/metrics.go
new file mode 100644
index 0000000..aeaf350
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/etcdhttp/metrics.go
@@ -0,0 +1,123 @@
+// Copyright 2017 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 etcdhttp
+
+import (
+ "context"
+ "encoding/json"
+ "net/http"
+ "time"
+
+ "github.com/coreos/etcd/etcdserver"
+ "github.com/coreos/etcd/etcdserver/etcdserverpb"
+ "github.com/coreos/etcd/raft"
+
+ "github.com/prometheus/client_golang/prometheus"
+ "github.com/prometheus/client_golang/prometheus/promhttp"
+)
+
+const (
+ pathMetrics = "/metrics"
+ PathHealth = "/health"
+)
+
+// HandleMetricsHealth registers metrics and health handlers.
+func HandleMetricsHealth(mux *http.ServeMux, srv etcdserver.ServerV2) {
+ mux.Handle(pathMetrics, promhttp.Handler())
+ mux.Handle(PathHealth, NewHealthHandler(func() Health { return checkHealth(srv) }))
+}
+
+// HandlePrometheus registers prometheus handler on '/metrics'.
+func HandlePrometheus(mux *http.ServeMux) {
+ mux.Handle(pathMetrics, promhttp.Handler())
+}
+
+// NewHealthHandler handles '/health' requests.
+func NewHealthHandler(hfunc func() Health) http.HandlerFunc {
+ return func(w http.ResponseWriter, r *http.Request) {
+ if r.Method != http.MethodGet {
+ w.Header().Set("Allow", http.MethodGet)
+ http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
+ return
+ }
+ h := hfunc()
+ d, _ := json.Marshal(h)
+ if h.Health != "true" {
+ http.Error(w, string(d), http.StatusServiceUnavailable)
+ return
+ }
+ w.WriteHeader(http.StatusOK)
+ w.Write(d)
+ }
+}
+
+var (
+ healthSuccess = prometheus.NewCounter(prometheus.CounterOpts{
+ Namespace: "etcd",
+ Subsystem: "server",
+ Name: "health_success",
+ Help: "The total number of successful health checks",
+ })
+ healthFailed = prometheus.NewCounter(prometheus.CounterOpts{
+ Namespace: "etcd",
+ Subsystem: "server",
+ Name: "health_failures",
+ Help: "The total number of failed health checks",
+ })
+)
+
+func init() {
+ prometheus.MustRegister(healthSuccess)
+ prometheus.MustRegister(healthFailed)
+}
+
+// Health defines etcd server health status.
+// TODO: remove manual parsing in etcdctl cluster-health
+type Health struct {
+ Health string `json:"health"`
+}
+
+// TODO: server NOSPACE, etcdserver.ErrNoLeader in health API
+
+func checkHealth(srv etcdserver.ServerV2) Health {
+ h := Health{Health: "true"}
+
+ as := srv.Alarms()
+ if len(as) > 0 {
+ h.Health = "false"
+ }
+
+ if h.Health == "true" {
+ if uint64(srv.Leader()) == raft.None {
+ h.Health = "false"
+ }
+ }
+
+ if h.Health == "true" {
+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+ _, err := srv.Do(ctx, etcdserverpb.Request{Method: "QGET"})
+ cancel()
+ if err != nil {
+ h.Health = "false"
+ }
+ }
+
+ if h.Health == "true" {
+ healthSuccess.Inc()
+ } else {
+ healthFailed.Inc()
+ }
+ return h
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/etcdhttp/peer.go b/vendor/github.com/coreos/etcd/etcdserver/api/etcdhttp/peer.go
new file mode 100644
index 0000000..0a9213b
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/etcdhttp/peer.go
@@ -0,0 +1,73 @@
+// 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 etcdhttp
+
+import (
+ "encoding/json"
+ "net/http"
+
+ "github.com/coreos/etcd/etcdserver"
+ "github.com/coreos/etcd/etcdserver/api"
+ "github.com/coreos/etcd/lease/leasehttp"
+ "github.com/coreos/etcd/rafthttp"
+)
+
+const (
+ peerMembersPrefix = "/members"
+)
+
+// NewPeerHandler generates an http.Handler to handle etcd peer requests.
+func NewPeerHandler(s etcdserver.ServerPeer) http.Handler {
+ return newPeerHandler(s.Cluster(), s.RaftHandler(), s.LeaseHandler())
+}
+
+func newPeerHandler(cluster api.Cluster, raftHandler http.Handler, leaseHandler http.Handler) http.Handler {
+ mh := &peerMembersHandler{
+ cluster: cluster,
+ }
+
+ mux := http.NewServeMux()
+ mux.HandleFunc("/", http.NotFound)
+ mux.Handle(rafthttp.RaftPrefix, raftHandler)
+ mux.Handle(rafthttp.RaftPrefix+"/", raftHandler)
+ mux.Handle(peerMembersPrefix, mh)
+ if leaseHandler != nil {
+ mux.Handle(leasehttp.LeasePrefix, leaseHandler)
+ mux.Handle(leasehttp.LeaseInternalPrefix, leaseHandler)
+ }
+ mux.HandleFunc(versionPath, versionHandler(cluster, serveVersion))
+ return mux
+}
+
+type peerMembersHandler struct {
+ cluster api.Cluster
+}
+
+func (h *peerMembersHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+ if !allowMethod(w, r, "GET") {
+ return
+ }
+ w.Header().Set("X-Etcd-Cluster-ID", h.cluster.ID().String())
+
+ if r.URL.Path != peerMembersPrefix {
+ http.Error(w, "bad path", http.StatusBadRequest)
+ return
+ }
+ ms := h.cluster.Members()
+ w.Header().Set("Content-Type", "application/json")
+ if err := json.NewEncoder(w).Encode(ms); err != nil {
+ plog.Warningf("failed to encode members response (%v)", err)
+ }
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v2http/capability.go b/vendor/github.com/coreos/etcd/etcdserver/api/v2http/capability.go
new file mode 100644
index 0000000..fa0bcca
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v2http/capability.go
@@ -0,0 +1,40 @@
+// 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 (
+ "fmt"
+ "net/http"
+
+ "github.com/coreos/etcd/etcdserver/api"
+ "github.com/coreos/etcd/etcdserver/api/v2http/httptypes"
+)
+
+func capabilityHandler(c api.Capability, fn func(http.ResponseWriter, *http.Request)) http.HandlerFunc {
+ return func(w http.ResponseWriter, r *http.Request) {
+ if !api.IsCapabilityEnabled(c) {
+ notCapable(w, r, c)
+ return
+ }
+ fn(w, r)
+ }
+}
+
+func notCapable(w http.ResponseWriter, r *http.Request, c api.Capability) {
+ herr := httptypes.NewHTTPError(http.StatusInternalServerError, fmt.Sprintf("Not capable of accessing %s feature during rolling upgrades.", c))
+ if err := herr.WriteTo(w); err != nil {
+ plog.Debugf("error writing HTTPError (%v) to %s", err, r.RemoteAddr)
+ }
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v2http/client.go b/vendor/github.com/coreos/etcd/etcdserver/api/v2http/client.go
new file mode 100644
index 0000000..6aaf3db
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v2http/client.go
@@ -0,0 +1,719 @@
+// 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"
+
+ etcdErr "github.com/coreos/etcd/error"
+ "github.com/coreos/etcd/etcdserver"
+ "github.com/coreos/etcd/etcdserver/api"
+ "github.com/coreos/etcd/etcdserver/api/etcdhttp"
+ "github.com/coreos/etcd/etcdserver/api/v2http/httptypes"
+ "github.com/coreos/etcd/etcdserver/auth"
+ "github.com/coreos/etcd/etcdserver/etcdserverpb"
+ "github.com/coreos/etcd/etcdserver/membership"
+ "github.com/coreos/etcd/etcdserver/stats"
+ "github.com/coreos/etcd/pkg/types"
+ "github.com/coreos/etcd/store"
+
+ "github.com/jonboulle/clockwork"
+)
+
+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(server etcdserver.ServerPeer, timeout time.Duration) http.Handler {
+ mux := http.NewServeMux()
+ etcdhttp.HandleBasic(mux, server)
+ handleV2(mux, server, timeout)
+ return requestLogger(mux)
+}
+
+func handleV2(mux *http.ServeMux, server etcdserver.ServerV2, timeout time.Duration) {
+ sec := auth.NewStore(server, timeout)
+ kh := &keysHandler{
+ sec: sec,
+ server: server,
+ cluster: server.Cluster(),
+ timeout: timeout,
+ clientCertAuthEnabled: server.ClientCertAuthEnabled(),
+ }
+
+ sh := &statsHandler{
+ stats: server,
+ }
+
+ mh := &membersHandler{
+ sec: sec,
+ server: server,
+ cluster: server.Cluster(),
+ timeout: timeout,
+ clock: clockwork.NewRealClock(),
+ clientCertAuthEnabled: server.ClientCertAuthEnabled(),
+ }
+
+ mah := &machinesHandler{cluster: server.Cluster()}
+
+ sech := &authHandler{
+ 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 {
+ sec auth.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(w, err)
+ return
+ }
+ // The path must be valid at this point (we've parsed the request successfully).
+ if !hasKeyPrefixAccess(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(w, err)
+ reportRequestFailed(rr, err)
+ return
+ }
+ switch {
+ case resp.Event != nil:
+ if err := writeKeyEvent(w, resp, noValueOnSuccess); err != nil {
+ // Should never be reached
+ plog.Errorf("error writing event (%v)", err)
+ }
+ reportRequestCompleted(rr, resp, startTime)
+ case resp.Watcher != nil:
+ ctx, cancel := context.WithTimeout(context.Background(), defaultWatchTimeout)
+ defer cancel()
+ handleKeyWatch(ctx, w, resp, rr.Stream)
+ default:
+ writeKeyError(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 {
+ sec auth.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.sec, r, h.clientCertAuthEnabled) {
+ writeNoAuth(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 {
+ plog.Warningf("failed to encode members response (%v)", err)
+ }
+ case "leader":
+ id := h.server.Leader()
+ if id == 0 {
+ writeError(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 {
+ plog.Warningf("failed to encode members response (%v)", err)
+ }
+ default:
+ writeError(w, r, httptypes.NewHTTPError(http.StatusNotFound, "Not found"))
+ }
+ case "POST":
+ req := httptypes.MemberCreateRequest{}
+ if ok := unmarshalRequest(r, &req, w); !ok {
+ return
+ }
+ now := h.clock.Now()
+ m := membership.NewMember("", req.PeerURLs, "", &now)
+ _, err := h.server.AddMember(ctx, *m)
+ switch {
+ case err == membership.ErrIDExists || err == membership.ErrPeerURLexists:
+ writeError(w, r, httptypes.NewHTTPError(http.StatusConflict, err.Error()))
+ return
+ case err != nil:
+ plog.Errorf("error adding member %s (%v)", m.ID, err)
+ writeError(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 {
+ plog.Warningf("failed to encode members response (%v)", err)
+ }
+ case "DELETE":
+ id, ok := getID(r.URL.Path, w)
+ if !ok {
+ return
+ }
+ _, err := h.server.RemoveMember(ctx, uint64(id))
+ switch {
+ case err == membership.ErrIDRemoved:
+ writeError(w, r, httptypes.NewHTTPError(http.StatusGone, fmt.Sprintf("Member permanently removed: %s", id)))
+ case err == membership.ErrIDNotFound:
+ writeError(w, r, httptypes.NewHTTPError(http.StatusNotFound, fmt.Sprintf("No such member: %s", id)))
+ case err != nil:
+ plog.Errorf("error removing member %s (%v)", id, err)
+ writeError(w, r, err)
+ default:
+ w.WriteHeader(http.StatusNoContent)
+ }
+ case "PUT":
+ id, ok := getID(r.URL.Path, w)
+ if !ok {
+ return
+ }
+ req := httptypes.MemberUpdateRequest{}
+ if ok := unmarshalRequest(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(w, r, httptypes.NewHTTPError(http.StatusConflict, err.Error()))
+ case err == membership.ErrIDNotFound:
+ writeError(w, r, httptypes.NewHTTPError(http.StatusNotFound, fmt.Sprintf("No such member: %s", id)))
+ case err != nil:
+ plog.Errorf("error updating member %s (%v)", m.ID, err)
+ writeError(w, r, err)
+ default:
+ w.WriteHeader(http.StatusNoContent)
+ }
+ }
+}
+
+type statsHandler struct {
+ 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(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, etcdErr.NewRequestError(
+ etcdErr.EcodeInvalidForm,
+ err.Error(),
+ )
+ }
+
+ if !strings.HasPrefix(r.URL.Path, keysPrefix) {
+ return emptyReq, false, etcdErr.NewRequestError(
+ etcdErr.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, etcdErr.NewRequestError(
+ etcdErr.EcodeIndexNaN,
+ `invalid value for "prevIndex"`,
+ )
+ }
+ if wIdx, err = getUint64(r.Form, "waitIndex"); err != nil {
+ return emptyReq, false, etcdErr.NewRequestError(
+ etcdErr.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, etcdErr.NewRequestError(
+ etcdErr.EcodeInvalidField,
+ `invalid value for "recursive"`,
+ )
+ }
+ if sort, err = getBool(r.Form, "sorted"); err != nil {
+ return emptyReq, false, etcdErr.NewRequestError(
+ etcdErr.EcodeInvalidField,
+ `invalid value for "sorted"`,
+ )
+ }
+ if wait, err = getBool(r.Form, "wait"); err != nil {
+ return emptyReq, false, etcdErr.NewRequestError(
+ etcdErr.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, etcdErr.NewRequestError(
+ etcdErr.EcodeInvalidField,
+ `invalid value for "dir"`,
+ )
+ }
+ if quorum, err = getBool(r.Form, "quorum"); err != nil {
+ return emptyReq, false, etcdErr.NewRequestError(
+ etcdErr.EcodeInvalidField,
+ `invalid value for "quorum"`,
+ )
+ }
+ if stream, err = getBool(r.Form, "stream"); err != nil {
+ return emptyReq, false, etcdErr.NewRequestError(
+ etcdErr.EcodeInvalidField,
+ `invalid value for "stream"`,
+ )
+ }
+
+ if wait && r.Method != "GET" {
+ return emptyReq, false, etcdErr.NewRequestError(
+ etcdErr.EcodeInvalidField,
+ `"wait" can only be used with GET requests`,
+ )
+ }
+
+ pV := r.FormValue("prevValue")
+ if _, ok := r.Form["prevValue"]; ok && pV == "" {
+ return emptyReq, false, etcdErr.NewRequestError(
+ etcdErr.EcodePrevValueRequired,
+ `"prevValue" cannot be empty`,
+ )
+ }
+
+ if noValueOnSuccess, err = getBool(r.Form, "noValueOnSuccess"); err != nil {
+ return emptyReq, false, etcdErr.NewRequestError(
+ etcdErr.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, etcdErr.NewRequestError(
+ etcdErr.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, etcdErr.NewRequestError(
+ etcdErr.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, etcdErr.NewRequestError(
+ etcdErr.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, etcdErr.NewRequestError(
+ etcdErr.EcodeRefreshValue,
+ `A value was provided on a refresh`,
+ )
+ }
+ if ttl == nil {
+ return emptyReq, false, etcdErr.NewRequestError(
+ etcdErr.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 == store.Set || ev.Action == store.CompareAndSwap ||
+ ev.Action == store.Create || ev.Action == store.Update) {
+ ev.Node = nil
+ ev.PrevNode = nil
+ }
+ return json.NewEncoder(w).Encode(ev)
+}
+
+func writeKeyNoAuth(w http.ResponseWriter) {
+ e := etcdErr.NewError(etcdErr.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(w http.ResponseWriter, err error) {
+ if err == nil {
+ return
+ }
+ switch e := err.(type) {
+ case *etcdErr.Error:
+ e.WriteTo(w)
+ default:
+ switch err {
+ case etcdserver.ErrTimeoutDueToLeaderFail, etcdserver.ErrTimeoutDueToConnectionLost:
+ mlog.MergeError(err)
+ default:
+ mlog.MergeErrorf("got unexpected response error (%v)", err)
+ }
+ ee := etcdErr.NewError(etcdErr.EcodeRaftInternal, err.Error(), 0)
+ ee.WriteTo(w)
+ }
+}
+
+func handleKeyWatch(ctx context.Context, 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
+ plog.Warningf("error writing event (%v)", err)
+ return
+ }
+ if !stream {
+ return
+ }
+ w.(http.Flusher).Flush()
+ }
+ }
+}
+
+func trimEventPrefix(ev *store.Event, prefix string) *store.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 *store.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.(*etcdErr.Error); ok {
+ e.Cause = strings.TrimPrefix(e.Cause, prefix)
+ }
+ return err
+}
+
+func unmarshalRequest(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(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(w, r, httptypes.NewHTTPError(http.StatusBadRequest, err.Error()))
+ return false
+ }
+ if err := req.UnmarshalJSON(b); err != nil {
+ writeError(w, r, httptypes.NewHTTPError(http.StatusBadRequest, err.Error()))
+ return false
+ }
+ return true
+}
+
+func getID(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(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
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v2http/client_auth.go b/vendor/github.com/coreos/etcd/etcdserver/api/v2http/client_auth.go
new file mode 100644
index 0000000..606e2e0
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v2http/client_auth.go
@@ -0,0 +1,543 @@
+// 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 (
+ "encoding/json"
+ "net/http"
+ "path"
+ "strings"
+
+ "github.com/coreos/etcd/etcdserver/api"
+ "github.com/coreos/etcd/etcdserver/api/v2http/httptypes"
+ "github.com/coreos/etcd/etcdserver/auth"
+)
+
+type authHandler struct {
+ sec auth.Store
+ cluster api.Cluster
+ clientCertAuthEnabled bool
+}
+
+func hasWriteRootAccess(sec auth.Store, r *http.Request, clientCertAuthEnabled bool) bool {
+ if r.Method == "GET" || r.Method == "HEAD" {
+ return true
+ }
+ return hasRootAccess(sec, r, clientCertAuthEnabled)
+}
+
+func userFromBasicAuth(sec auth.Store, r *http.Request) *auth.User {
+ username, password, ok := r.BasicAuth()
+ if !ok {
+ plog.Warningf("auth: malformed basic auth encoding")
+ return nil
+ }
+ user, err := sec.GetUser(username)
+ if err != nil {
+ return nil
+ }
+
+ ok = sec.CheckPassword(user, password)
+ if !ok {
+ plog.Warningf("auth: incorrect password for user: %s", username)
+ return nil
+ }
+ return &user
+}
+
+func userFromClientCertificate(sec auth.Store, r *http.Request) *auth.User {
+ if r.TLS == nil {
+ return nil
+ }
+
+ for _, chains := range r.TLS.VerifiedChains {
+ for _, chain := range chains {
+ plog.Debugf("auth: found common name %s.\n", chain.Subject.CommonName)
+ user, err := sec.GetUser(chain.Subject.CommonName)
+ if err == nil {
+ plog.Debugf("auth: authenticated user %s by cert common name.", user.User)
+ return &user
+ }
+ }
+ }
+ return nil
+}
+
+func hasRootAccess(sec auth.Store, r *http.Request, clientCertAuthEnabled bool) bool {
+ if sec == nil {
+ // No store means no auth available, eg, tests.
+ return true
+ }
+ if !sec.AuthEnabled() {
+ return true
+ }
+
+ var rootUser *auth.User
+ if r.Header.Get("Authorization") == "" && clientCertAuthEnabled {
+ rootUser = userFromClientCertificate(sec, r)
+ if rootUser == nil {
+ return false
+ }
+ } else {
+ rootUser = userFromBasicAuth(sec, r)
+ if rootUser == nil {
+ return false
+ }
+ }
+
+ for _, role := range rootUser.Roles {
+ if role == auth.RootRoleName {
+ return true
+ }
+ }
+ plog.Warningf("auth: user %s does not have the %s role for resource %s.", rootUser.User, auth.RootRoleName, r.URL.Path)
+ return false
+}
+
+func hasKeyPrefixAccess(sec auth.Store, r *http.Request, key string, recursive, clientCertAuthEnabled bool) bool {
+ if sec == nil {
+ // No store means no auth available, eg, tests.
+ return true
+ }
+ if !sec.AuthEnabled() {
+ return true
+ }
+
+ var user *auth.User
+ if r.Header.Get("Authorization") == "" {
+ if clientCertAuthEnabled {
+ user = userFromClientCertificate(sec, r)
+ }
+ if user == nil {
+ return hasGuestAccess(sec, r, key)
+ }
+ } else {
+ user = userFromBasicAuth(sec, r)
+ if user == nil {
+ return false
+ }
+ }
+
+ writeAccess := r.Method != "GET" && r.Method != "HEAD"
+ for _, roleName := range user.Roles {
+ role, err := sec.GetRole(roleName)
+ if err != nil {
+ continue
+ }
+ if recursive {
+ if role.HasRecursiveAccess(key, writeAccess) {
+ return true
+ }
+ } else if role.HasKeyAccess(key, writeAccess) {
+ return true
+ }
+ }
+ plog.Warningf("auth: invalid access for user %s on key %s.", user.User, key)
+ return false
+}
+
+func hasGuestAccess(sec auth.Store, r *http.Request, key string) bool {
+ writeAccess := r.Method != "GET" && r.Method != "HEAD"
+ role, err := sec.GetRole(auth.GuestRoleName)
+ if err != nil {
+ return false
+ }
+ if role.HasKeyAccess(key, writeAccess) {
+ return true
+ }
+ plog.Warningf("auth: invalid access for unauthenticated user on resource %s.", key)
+ return false
+}
+
+func writeNoAuth(w http.ResponseWriter, r *http.Request) {
+ herr := httptypes.NewHTTPError(http.StatusUnauthorized, "Insufficient credentials")
+ if err := herr.WriteTo(w); err != nil {
+ plog.Debugf("error writing HTTPError (%v) to %s", err, r.RemoteAddr)
+ }
+}
+
+func handleAuth(mux *http.ServeMux, sh *authHandler) {
+ mux.HandleFunc(authPrefix+"/roles", capabilityHandler(api.AuthCapability, sh.baseRoles))
+ mux.HandleFunc(authPrefix+"/roles/", capabilityHandler(api.AuthCapability, sh.handleRoles))
+ mux.HandleFunc(authPrefix+"/users", capabilityHandler(api.AuthCapability, sh.baseUsers))
+ mux.HandleFunc(authPrefix+"/users/", capabilityHandler(api.AuthCapability, sh.handleUsers))
+ mux.HandleFunc(authPrefix+"/enable", capabilityHandler(api.AuthCapability, sh.enableDisable))
+}
+
+func (sh *authHandler) baseRoles(w http.ResponseWriter, r *http.Request) {
+ if !allowMethod(w, r.Method, "GET") {
+ return
+ }
+ if !hasRootAccess(sh.sec, r, sh.clientCertAuthEnabled) {
+ writeNoAuth(w, r)
+ return
+ }
+
+ w.Header().Set("X-Etcd-Cluster-ID", sh.cluster.ID().String())
+ w.Header().Set("Content-Type", "application/json")
+
+ roles, err := sh.sec.AllRoles()
+ if err != nil {
+ writeError(w, r, err)
+ return
+ }
+ if roles == nil {
+ roles = make([]string, 0)
+ }
+
+ err = r.ParseForm()
+ if err != nil {
+ writeError(w, r, err)
+ return
+ }
+
+ var rolesCollections struct {
+ Roles []auth.Role `json:"roles"`
+ }
+ for _, roleName := range roles {
+ var role auth.Role
+ role, err = sh.sec.GetRole(roleName)
+ if err != nil {
+ writeError(w, r, err)
+ return
+ }
+ rolesCollections.Roles = append(rolesCollections.Roles, role)
+ }
+ err = json.NewEncoder(w).Encode(rolesCollections)
+
+ if err != nil {
+ plog.Warningf("baseRoles error encoding on %s", r.URL)
+ writeError(w, r, err)
+ return
+ }
+}
+
+func (sh *authHandler) handleRoles(w http.ResponseWriter, r *http.Request) {
+ subpath := path.Clean(r.URL.Path[len(authPrefix):])
+ // Split "/roles/rolename/command".
+ // First item is an empty string, second is "roles"
+ pieces := strings.Split(subpath, "/")
+ if len(pieces) == 2 {
+ sh.baseRoles(w, r)
+ return
+ }
+ if len(pieces) != 3 {
+ writeError(w, r, httptypes.NewHTTPError(http.StatusBadRequest, "Invalid path"))
+ return
+ }
+ sh.forRole(w, r, pieces[2])
+}
+
+func (sh *authHandler) forRole(w http.ResponseWriter, r *http.Request, role string) {
+ if !allowMethod(w, r.Method, "GET", "PUT", "DELETE") {
+ return
+ }
+ if !hasRootAccess(sh.sec, r, sh.clientCertAuthEnabled) {
+ writeNoAuth(w, r)
+ return
+ }
+ w.Header().Set("X-Etcd-Cluster-ID", sh.cluster.ID().String())
+ w.Header().Set("Content-Type", "application/json")
+
+ switch r.Method {
+ case "GET":
+ data, err := sh.sec.GetRole(role)
+ if err != nil {
+ writeError(w, r, err)
+ return
+ }
+ err = json.NewEncoder(w).Encode(data)
+ if err != nil {
+ plog.Warningf("forRole error encoding on %s", r.URL)
+ return
+ }
+ return
+ case "PUT":
+ var in auth.Role
+ err := json.NewDecoder(r.Body).Decode(&in)
+ if err != nil {
+ writeError(w, r, httptypes.NewHTTPError(http.StatusBadRequest, "Invalid JSON in request body."))
+ return
+ }
+ if in.Role != role {
+ writeError(w, r, httptypes.NewHTTPError(http.StatusBadRequest, "Role JSON name does not match the name in the URL"))
+ return
+ }
+
+ var out auth.Role
+
+ // create
+ if in.Grant.IsEmpty() && in.Revoke.IsEmpty() {
+ err = sh.sec.CreateRole(in)
+ if err != nil {
+ writeError(w, r, err)
+ return
+ }
+ w.WriteHeader(http.StatusCreated)
+ out = in
+ } else {
+ if !in.Permissions.IsEmpty() {
+ writeError(w, r, httptypes.NewHTTPError(http.StatusBadRequest, "Role JSON contains both permissions and grant/revoke"))
+ return
+ }
+ out, err = sh.sec.UpdateRole(in)
+ if err != nil {
+ writeError(w, r, err)
+ return
+ }
+ w.WriteHeader(http.StatusOK)
+ }
+
+ err = json.NewEncoder(w).Encode(out)
+ if err != nil {
+ plog.Warningf("forRole error encoding on %s", r.URL)
+ return
+ }
+ return
+ case "DELETE":
+ err := sh.sec.DeleteRole(role)
+ if err != nil {
+ writeError(w, r, err)
+ return
+ }
+ }
+}
+
+type userWithRoles struct {
+ User string `json:"user"`
+ Roles []auth.Role `json:"roles,omitempty"`
+}
+
+type usersCollections struct {
+ Users []userWithRoles `json:"users"`
+}
+
+func (sh *authHandler) baseUsers(w http.ResponseWriter, r *http.Request) {
+ if !allowMethod(w, r.Method, "GET") {
+ return
+ }
+ if !hasRootAccess(sh.sec, r, sh.clientCertAuthEnabled) {
+ writeNoAuth(w, r)
+ return
+ }
+ w.Header().Set("X-Etcd-Cluster-ID", sh.cluster.ID().String())
+ w.Header().Set("Content-Type", "application/json")
+
+ users, err := sh.sec.AllUsers()
+ if err != nil {
+ writeError(w, r, err)
+ return
+ }
+ if users == nil {
+ users = make([]string, 0)
+ }
+
+ err = r.ParseForm()
+ if err != nil {
+ writeError(w, r, err)
+ return
+ }
+
+ ucs := usersCollections{}
+ for _, userName := range users {
+ var user auth.User
+ user, err = sh.sec.GetUser(userName)
+ if err != nil {
+ writeError(w, r, err)
+ return
+ }
+
+ uwr := userWithRoles{User: user.User}
+ for _, roleName := range user.Roles {
+ var role auth.Role
+ role, err = sh.sec.GetRole(roleName)
+ if err != nil {
+ continue
+ }
+ uwr.Roles = append(uwr.Roles, role)
+ }
+
+ ucs.Users = append(ucs.Users, uwr)
+ }
+ err = json.NewEncoder(w).Encode(ucs)
+
+ if err != nil {
+ plog.Warningf("baseUsers error encoding on %s", r.URL)
+ writeError(w, r, err)
+ return
+ }
+}
+
+func (sh *authHandler) handleUsers(w http.ResponseWriter, r *http.Request) {
+ subpath := path.Clean(r.URL.Path[len(authPrefix):])
+ // Split "/users/username".
+ // First item is an empty string, second is "users"
+ pieces := strings.Split(subpath, "/")
+ if len(pieces) == 2 {
+ sh.baseUsers(w, r)
+ return
+ }
+ if len(pieces) != 3 {
+ writeError(w, r, httptypes.NewHTTPError(http.StatusBadRequest, "Invalid path"))
+ return
+ }
+ sh.forUser(w, r, pieces[2])
+}
+
+func (sh *authHandler) forUser(w http.ResponseWriter, r *http.Request, user string) {
+ if !allowMethod(w, r.Method, "GET", "PUT", "DELETE") {
+ return
+ }
+ if !hasRootAccess(sh.sec, r, sh.clientCertAuthEnabled) {
+ writeNoAuth(w, r)
+ return
+ }
+ w.Header().Set("X-Etcd-Cluster-ID", sh.cluster.ID().String())
+ w.Header().Set("Content-Type", "application/json")
+
+ switch r.Method {
+ case "GET":
+ u, err := sh.sec.GetUser(user)
+ if err != nil {
+ writeError(w, r, err)
+ return
+ }
+
+ err = r.ParseForm()
+ if err != nil {
+ writeError(w, r, err)
+ return
+ }
+
+ uwr := userWithRoles{User: u.User}
+ for _, roleName := range u.Roles {
+ var role auth.Role
+ role, err = sh.sec.GetRole(roleName)
+ if err != nil {
+ writeError(w, r, err)
+ return
+ }
+ uwr.Roles = append(uwr.Roles, role)
+ }
+ err = json.NewEncoder(w).Encode(uwr)
+
+ if err != nil {
+ plog.Warningf("forUser error encoding on %s", r.URL)
+ return
+ }
+ return
+ case "PUT":
+ var u auth.User
+ err := json.NewDecoder(r.Body).Decode(&u)
+ if err != nil {
+ writeError(w, r, httptypes.NewHTTPError(http.StatusBadRequest, "Invalid JSON in request body."))
+ return
+ }
+ if u.User != user {
+ writeError(w, r, httptypes.NewHTTPError(http.StatusBadRequest, "User JSON name does not match the name in the URL"))
+ return
+ }
+
+ var (
+ out auth.User
+ created bool
+ )
+
+ if len(u.Grant) == 0 && len(u.Revoke) == 0 {
+ // create or update
+ if len(u.Roles) != 0 {
+ out, err = sh.sec.CreateUser(u)
+ } else {
+ // if user passes in both password and roles, we are unsure about his/her
+ // intention.
+ out, created, err = sh.sec.CreateOrUpdateUser(u)
+ }
+
+ if err != nil {
+ writeError(w, r, err)
+ return
+ }
+ } else {
+ // update case
+ if len(u.Roles) != 0 {
+ writeError(w, r, httptypes.NewHTTPError(http.StatusBadRequest, "User JSON contains both roles and grant/revoke"))
+ return
+ }
+ out, err = sh.sec.UpdateUser(u)
+ if err != nil {
+ writeError(w, r, err)
+ return
+ }
+ }
+
+ if created {
+ w.WriteHeader(http.StatusCreated)
+ } else {
+ w.WriteHeader(http.StatusOK)
+ }
+
+ out.Password = ""
+
+ err = json.NewEncoder(w).Encode(out)
+ if err != nil {
+ plog.Warningf("forUser error encoding on %s", r.URL)
+ return
+ }
+ return
+ case "DELETE":
+ err := sh.sec.DeleteUser(user)
+ if err != nil {
+ writeError(w, r, err)
+ return
+ }
+ }
+}
+
+type enabled struct {
+ Enabled bool `json:"enabled"`
+}
+
+func (sh *authHandler) enableDisable(w http.ResponseWriter, r *http.Request) {
+ if !allowMethod(w, r.Method, "GET", "PUT", "DELETE") {
+ return
+ }
+ if !hasWriteRootAccess(sh.sec, r, sh.clientCertAuthEnabled) {
+ writeNoAuth(w, r)
+ return
+ }
+ w.Header().Set("X-Etcd-Cluster-ID", sh.cluster.ID().String())
+ w.Header().Set("Content-Type", "application/json")
+ isEnabled := sh.sec.AuthEnabled()
+ switch r.Method {
+ case "GET":
+ jsonDict := enabled{isEnabled}
+ err := json.NewEncoder(w).Encode(jsonDict)
+ if err != nil {
+ plog.Warningf("error encoding auth state on %s", r.URL)
+ }
+ case "PUT":
+ err := sh.sec.EnableAuth()
+ if err != nil {
+ writeError(w, r, err)
+ return
+ }
+ case "DELETE":
+ err := sh.sec.DisableAuth()
+ if err != nil {
+ writeError(w, r, err)
+ return
+ }
+ }
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v2http/doc.go b/vendor/github.com/coreos/etcd/etcdserver/api/v2http/doc.go
new file mode 100644
index 0000000..475c4b1
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v2http/doc.go
@@ -0,0 +1,16 @@
+// 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 provides etcd client and server implementations.
+package v2http
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v2http/http.go b/vendor/github.com/coreos/etcd/etcdserver/api/v2http/http.go
new file mode 100644
index 0000000..589c172
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v2http/http.go
@@ -0,0 +1,74 @@
+// 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 (
+ "math"
+ "net/http"
+ "strings"
+ "time"
+
+ "github.com/coreos/etcd/etcdserver/api/etcdhttp"
+ "github.com/coreos/etcd/etcdserver/api/v2http/httptypes"
+ "github.com/coreos/etcd/etcdserver/auth"
+ "github.com/coreos/etcd/pkg/logutil"
+
+ "github.com/coreos/pkg/capnslog"
+)
+
+const (
+ // time to wait for a Watch request
+ defaultWatchTimeout = time.Duration(math.MaxInt64)
+)
+
+var (
+ plog = capnslog.NewPackageLogger("github.com/coreos/etcd", "etcdserver/api/v2http")
+ mlog = logutil.NewMergeLogger(plog)
+)
+
+func writeError(w http.ResponseWriter, r *http.Request, err error) {
+ if err == nil {
+ return
+ }
+ if e, ok := err.(auth.Error); ok {
+ herr := httptypes.NewHTTPError(e.HTTPStatus(), e.Error())
+ if et := herr.WriteTo(w); et != nil {
+ plog.Debugf("error writing HTTPError (%v) to %s", et, r.RemoteAddr)
+ }
+ return
+ }
+ etcdhttp.WriteError(w, r, err)
+}
+
+// allowMethod verifies that the given method is one of the allowed methods,
+// and if not, it writes an error to w. A boolean is returned indicating
+// whether or not the method is allowed.
+func allowMethod(w http.ResponseWriter, m string, ms ...string) bool {
+ for _, meth := range ms {
+ if m == meth {
+ return true
+ }
+ }
+ w.Header().Set("Allow", strings.Join(ms, ","))
+ http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
+ return false
+}
+
+func requestLogger(handler http.Handler) http.Handler {
+ return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ plog.Debugf("[%s] %s remote:%s", r.Method, r.RequestURI, r.RemoteAddr)
+ handler.ServeHTTP(w, r)
+ })
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v2http/httptypes/errors.go b/vendor/github.com/coreos/etcd/etcdserver/api/v2http/httptypes/errors.go
new file mode 100644
index 0000000..0657604
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v2http/httptypes/errors.go
@@ -0,0 +1,56 @@
+// 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 httptypes
+
+import (
+ "encoding/json"
+ "net/http"
+
+ "github.com/coreos/pkg/capnslog"
+)
+
+var (
+ plog = capnslog.NewPackageLogger("github.com/coreos/etcd", "etcdserver/api/v2http/httptypes")
+)
+
+type HTTPError struct {
+ Message string `json:"message"`
+ // Code is the HTTP status code
+ Code int `json:"-"`
+}
+
+func (e HTTPError) Error() string {
+ return e.Message
+}
+
+func (e HTTPError) WriteTo(w http.ResponseWriter) error {
+ w.Header().Set("Content-Type", "application/json")
+ w.WriteHeader(e.Code)
+ b, err := json.Marshal(e)
+ if err != nil {
+ plog.Panicf("marshal HTTPError should never fail (%v)", err)
+ }
+ if _, err := w.Write(b); err != nil {
+ return err
+ }
+ return nil
+}
+
+func NewHTTPError(code int, m string) *HTTPError {
+ return &HTTPError{
+ Message: m,
+ Code: code,
+ }
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v2http/httptypes/member.go b/vendor/github.com/coreos/etcd/etcdserver/api/v2http/httptypes/member.go
new file mode 100644
index 0000000..738d744
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v2http/httptypes/member.go
@@ -0,0 +1,69 @@
+// 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 httptypes defines how etcd's HTTP API entities are serialized to and
+// deserialized from JSON.
+package httptypes
+
+import (
+ "encoding/json"
+
+ "github.com/coreos/etcd/pkg/types"
+)
+
+type Member struct {
+ ID string `json:"id"`
+ Name string `json:"name"`
+ PeerURLs []string `json:"peerURLs"`
+ ClientURLs []string `json:"clientURLs"`
+}
+
+type MemberCreateRequest struct {
+ PeerURLs types.URLs
+}
+
+type MemberUpdateRequest struct {
+ MemberCreateRequest
+}
+
+func (m *MemberCreateRequest) UnmarshalJSON(data []byte) error {
+ s := struct {
+ PeerURLs []string `json:"peerURLs"`
+ }{}
+
+ err := json.Unmarshal(data, &s)
+ if err != nil {
+ return err
+ }
+
+ urls, err := types.NewURLs(s.PeerURLs)
+ if err != nil {
+ return err
+ }
+
+ m.PeerURLs = urls
+ return nil
+}
+
+type MemberCollection []Member
+
+func (c *MemberCollection) MarshalJSON() ([]byte, error) {
+ d := struct {
+ Members []Member `json:"members"`
+ }{
+ Members: []Member(*c),
+ }
+
+ return json.Marshal(d)
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v2http/metrics.go b/vendor/github.com/coreos/etcd/etcdserver/api/v2http/metrics.go
new file mode 100644
index 0000000..fdfb0c6
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v2http/metrics.go
@@ -0,0 +1,96 @@
+// 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 (
+ "strconv"
+ "time"
+
+ "net/http"
+
+ etcdErr "github.com/coreos/etcd/error"
+ "github.com/coreos/etcd/etcdserver"
+ "github.com/coreos/etcd/etcdserver/api/v2http/httptypes"
+ "github.com/coreos/etcd/etcdserver/etcdserverpb"
+ "github.com/prometheus/client_golang/prometheus"
+)
+
+var (
+ incomingEvents = prometheus.NewCounterVec(
+ prometheus.CounterOpts{
+ Namespace: "etcd",
+ Subsystem: "http",
+ Name: "received_total",
+ Help: "Counter of requests received into the system (successfully parsed and authd).",
+ }, []string{"method"})
+
+ failedEvents = prometheus.NewCounterVec(
+ prometheus.CounterOpts{
+ Namespace: "etcd",
+ Subsystem: "http",
+ Name: "failed_total",
+ Help: "Counter of handle failures of requests (non-watches), by method (GET/PUT etc.) and code (400, 500 etc.).",
+ }, []string{"method", "code"})
+
+ successfulEventsHandlingTime = prometheus.NewHistogramVec(
+ prometheus.HistogramOpts{
+ Namespace: "etcd",
+ Subsystem: "http",
+ Name: "successful_duration_seconds",
+ Help: "Bucketed histogram of processing time (s) of successfully handled requests (non-watches), by method (GET/PUT etc.).",
+ Buckets: prometheus.ExponentialBuckets(0.0005, 2, 13),
+ }, []string{"method"})
+)
+
+func init() {
+ prometheus.MustRegister(incomingEvents)
+ prometheus.MustRegister(failedEvents)
+ prometheus.MustRegister(successfulEventsHandlingTime)
+}
+
+func reportRequestReceived(request etcdserverpb.Request) {
+ incomingEvents.WithLabelValues(methodFromRequest(request)).Inc()
+}
+
+func reportRequestCompleted(request etcdserverpb.Request, response etcdserver.Response, startTime time.Time) {
+ method := methodFromRequest(request)
+ successfulEventsHandlingTime.WithLabelValues(method).Observe(time.Since(startTime).Seconds())
+}
+
+func reportRequestFailed(request etcdserverpb.Request, err error) {
+ method := methodFromRequest(request)
+ failedEvents.WithLabelValues(method, strconv.Itoa(codeFromError(err))).Inc()
+}
+
+func methodFromRequest(request etcdserverpb.Request) string {
+ if request.Method == "GET" && request.Quorum {
+ return "QGET"
+ }
+ return request.Method
+}
+
+func codeFromError(err error) int {
+ if err == nil {
+ return http.StatusInternalServerError
+ }
+ switch e := err.(type) {
+ case *etcdErr.Error:
+ return (*etcdErr.Error)(e).StatusCode()
+ case *httptypes.HTTPError:
+ return (*httptypes.HTTPError)(e).Code
+ default:
+ return http.StatusInternalServerError
+ }
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v2v3/cluster.go b/vendor/github.com/coreos/etcd/etcdserver/api/v2v3/cluster.go
new file mode 100644
index 0000000..b53e6d7
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v2v3/cluster.go
@@ -0,0 +1,31 @@
+// Copyright 2017 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 v2v3
+
+import (
+ "github.com/coreos/etcd/etcdserver/membership"
+ "github.com/coreos/etcd/pkg/types"
+
+ "github.com/coreos/go-semver/semver"
+)
+
+func (s *v2v3Server) ID() types.ID {
+ // TODO: use an actual member ID
+ return types.ID(0xe7cd2f00d)
+}
+func (s *v2v3Server) ClientURLs() []string { panic("STUB") }
+func (s *v2v3Server) Members() []*membership.Member { panic("STUB") }
+func (s *v2v3Server) Member(id types.ID) *membership.Member { panic("STUB") }
+func (s *v2v3Server) Version() *semver.Version { panic("STUB") }
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v2v3/doc.go b/vendor/github.com/coreos/etcd/etcdserver/api/v2v3/doc.go
new file mode 100644
index 0000000..2ff372f
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v2v3/doc.go
@@ -0,0 +1,16 @@
+// Copyright 2017 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 v2v3 provides a ServerV2 implementation backed by clientv3.Client.
+package v2v3
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v2v3/server.go b/vendor/github.com/coreos/etcd/etcdserver/api/v2v3/server.go
new file mode 100644
index 0000000..2ef63ce
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v2v3/server.go
@@ -0,0 +1,117 @@
+// Copyright 2017 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 v2v3
+
+import (
+ "context"
+ "net/http"
+ "time"
+
+ "github.com/coreos/etcd/clientv3"
+ "github.com/coreos/etcd/etcdserver"
+ "github.com/coreos/etcd/etcdserver/api"
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+ "github.com/coreos/etcd/etcdserver/membership"
+ "github.com/coreos/etcd/pkg/types"
+
+ "github.com/coreos/go-semver/semver"
+)
+
+type fakeStats struct{}
+
+func (s *fakeStats) SelfStats() []byte { return nil }
+func (s *fakeStats) LeaderStats() []byte { return nil }
+func (s *fakeStats) StoreStats() []byte { return nil }
+
+type v2v3Server struct {
+ c *clientv3.Client
+ store *v2v3Store
+ fakeStats
+}
+
+func NewServer(c *clientv3.Client, pfx string) etcdserver.ServerPeer {
+ return &v2v3Server{c: c, store: newStore(c, pfx)}
+}
+
+func (s *v2v3Server) ClientCertAuthEnabled() bool { return false }
+
+func (s *v2v3Server) LeaseHandler() http.Handler { panic("STUB: lease handler") }
+func (s *v2v3Server) RaftHandler() http.Handler { panic("STUB: raft handler") }
+
+func (s *v2v3Server) Leader() types.ID {
+ ctx, cancel := context.WithTimeout(context.TODO(), 5*time.Second)
+ defer cancel()
+ resp, err := s.c.Status(ctx, s.c.Endpoints()[0])
+ if err != nil {
+ return 0
+ }
+ return types.ID(resp.Leader)
+}
+
+func (s *v2v3Server) AddMember(ctx context.Context, memb membership.Member) ([]*membership.Member, error) {
+ resp, err := s.c.MemberAdd(ctx, memb.PeerURLs)
+ if err != nil {
+ return nil, err
+ }
+ return v3MembersToMembership(resp.Members), nil
+}
+
+func (s *v2v3Server) RemoveMember(ctx context.Context, id uint64) ([]*membership.Member, error) {
+ resp, err := s.c.MemberRemove(ctx, id)
+ if err != nil {
+ return nil, err
+ }
+ return v3MembersToMembership(resp.Members), nil
+}
+
+func (s *v2v3Server) UpdateMember(ctx context.Context, m membership.Member) ([]*membership.Member, error) {
+ resp, err := s.c.MemberUpdate(ctx, uint64(m.ID), m.PeerURLs)
+ if err != nil {
+ return nil, err
+ }
+ return v3MembersToMembership(resp.Members), nil
+}
+
+func v3MembersToMembership(v3membs []*pb.Member) []*membership.Member {
+ membs := make([]*membership.Member, len(v3membs))
+ for i, m := range v3membs {
+ membs[i] = &membership.Member{
+ ID: types.ID(m.ID),
+ RaftAttributes: membership.RaftAttributes{
+ PeerURLs: m.PeerURLs,
+ },
+ Attributes: membership.Attributes{
+ Name: m.Name,
+ ClientURLs: m.ClientURLs,
+ },
+ }
+ }
+ return membs
+}
+
+func (s *v2v3Server) ClusterVersion() *semver.Version { return s.Version() }
+func (s *v2v3Server) Cluster() api.Cluster { return s }
+func (s *v2v3Server) Alarms() []*pb.AlarmMember { return nil }
+
+func (s *v2v3Server) Do(ctx context.Context, r pb.Request) (etcdserver.Response, error) {
+ applier := etcdserver.NewApplierV2(s.store, nil)
+ reqHandler := etcdserver.NewStoreRequestV2Handler(s.store, applier)
+ req := (*etcdserver.RequestV2)(&r)
+ resp, err := req.Handle(ctx, reqHandler)
+ if resp.Err != nil {
+ return resp, resp.Err
+ }
+ return resp, err
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v2v3/store.go b/vendor/github.com/coreos/etcd/etcdserver/api/v2v3/store.go
new file mode 100644
index 0000000..444f93f
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v2v3/store.go
@@ -0,0 +1,620 @@
+// Copyright 2017 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 v2v3
+
+import (
+ "context"
+ "fmt"
+ "path"
+ "strings"
+ "time"
+
+ "github.com/coreos/etcd/clientv3"
+ "github.com/coreos/etcd/clientv3/concurrency"
+ etcdErr "github.com/coreos/etcd/error"
+ "github.com/coreos/etcd/mvcc/mvccpb"
+ "github.com/coreos/etcd/store"
+)
+
+// store implements the Store interface for V2 using
+// a v3 client.
+type v2v3Store struct {
+ c *clientv3.Client
+ // pfx is the v3 prefix where keys should be stored.
+ pfx string
+ ctx context.Context
+}
+
+const maxPathDepth = 63
+
+var errUnsupported = fmt.Errorf("TTLs are unsupported")
+
+func NewStore(c *clientv3.Client, pfx string) store.Store { return newStore(c, pfx) }
+
+func newStore(c *clientv3.Client, pfx string) *v2v3Store { return &v2v3Store{c, pfx, c.Ctx()} }
+
+func (s *v2v3Store) Index() uint64 { panic("STUB") }
+
+func (s *v2v3Store) Get(nodePath string, recursive, sorted bool) (*store.Event, error) {
+ key := s.mkPath(nodePath)
+ resp, err := s.c.Txn(s.ctx).Then(
+ clientv3.OpGet(key+"/"),
+ clientv3.OpGet(key),
+ ).Commit()
+ if err != nil {
+ return nil, err
+ }
+
+ if kvs := resp.Responses[0].GetResponseRange().Kvs; len(kvs) != 0 || isRoot(nodePath) {
+ nodes, err := s.getDir(nodePath, recursive, sorted, resp.Header.Revision)
+ if err != nil {
+ return nil, err
+ }
+ cidx, midx := uint64(0), uint64(0)
+ if len(kvs) > 0 {
+ cidx, midx = mkV2Rev(kvs[0].CreateRevision), mkV2Rev(kvs[0].ModRevision)
+ }
+ return &store.Event{
+ Action: store.Get,
+ Node: &store.NodeExtern{
+ Key: nodePath,
+ Dir: true,
+ Nodes: nodes,
+ CreatedIndex: cidx,
+ ModifiedIndex: midx,
+ },
+ EtcdIndex: mkV2Rev(resp.Header.Revision),
+ }, nil
+ }
+
+ kvs := resp.Responses[1].GetResponseRange().Kvs
+ if len(kvs) == 0 {
+ return nil, etcdErr.NewError(etcdErr.EcodeKeyNotFound, nodePath, mkV2Rev(resp.Header.Revision))
+ }
+
+ return &store.Event{
+ Action: store.Get,
+ Node: s.mkV2Node(kvs[0]),
+ EtcdIndex: mkV2Rev(resp.Header.Revision),
+ }, nil
+}
+
+func (s *v2v3Store) getDir(nodePath string, recursive, sorted bool, rev int64) ([]*store.NodeExtern, error) {
+ rootNodes, err := s.getDirDepth(nodePath, 1, rev)
+ if err != nil || !recursive {
+ return rootNodes, err
+ }
+ nextNodes := rootNodes
+ nodes := make(map[string]*store.NodeExtern)
+ // Breadth walk the subdirectories
+ for i := 2; len(nextNodes) > 0; i++ {
+ for _, n := range nextNodes {
+ nodes[n.Key] = n
+ if parent := nodes[path.Dir(n.Key)]; parent != nil {
+ parent.Nodes = append(parent.Nodes, n)
+ }
+ }
+ if nextNodes, err = s.getDirDepth(nodePath, i, rev); err != nil {
+ return nil, err
+ }
+ }
+ return rootNodes, nil
+}
+
+func (s *v2v3Store) getDirDepth(nodePath string, depth int, rev int64) ([]*store.NodeExtern, error) {
+ pd := s.mkPathDepth(nodePath, depth)
+ resp, err := s.c.Get(s.ctx, pd, clientv3.WithPrefix(), clientv3.WithRev(rev))
+ if err != nil {
+ return nil, err
+ }
+
+ nodes := make([]*store.NodeExtern, len(resp.Kvs))
+ for i, kv := range resp.Kvs {
+ nodes[i] = s.mkV2Node(kv)
+ }
+ return nodes, nil
+}
+
+func (s *v2v3Store) Set(
+ nodePath string,
+ dir bool,
+ value string,
+ expireOpts store.TTLOptionSet,
+) (*store.Event, error) {
+ if expireOpts.Refresh || !expireOpts.ExpireTime.IsZero() {
+ return nil, errUnsupported
+ }
+
+ if isRoot(nodePath) {
+ return nil, etcdErr.NewError(etcdErr.EcodeRootROnly, nodePath, 0)
+ }
+
+ ecode := 0
+ applyf := func(stm concurrency.STM) error {
+ parent := path.Dir(nodePath)
+ if !isRoot(parent) && stm.Rev(s.mkPath(parent)+"/") == 0 {
+ ecode = etcdErr.EcodeKeyNotFound
+ return nil
+ }
+
+ key := s.mkPath(nodePath)
+ if dir {
+ if stm.Rev(key) != 0 {
+ // exists as non-dir
+ ecode = etcdErr.EcodeNotDir
+ return nil
+ }
+ key = key + "/"
+ } else if stm.Rev(key+"/") != 0 {
+ ecode = etcdErr.EcodeNotFile
+ return nil
+ }
+ stm.Put(key, value, clientv3.WithPrevKV())
+ stm.Put(s.mkActionKey(), store.Set)
+ return nil
+ }
+
+ resp, err := s.newSTM(applyf)
+ if err != nil {
+ return nil, err
+ }
+ if ecode != 0 {
+ return nil, etcdErr.NewError(ecode, nodePath, mkV2Rev(resp.Header.Revision))
+ }
+
+ createRev := resp.Header.Revision
+ var pn *store.NodeExtern
+ if pkv := prevKeyFromPuts(resp); pkv != nil {
+ pn = s.mkV2Node(pkv)
+ createRev = pkv.CreateRevision
+ }
+
+ vp := &value
+ if dir {
+ vp = nil
+ }
+ return &store.Event{
+ Action: store.Set,
+ Node: &store.NodeExtern{
+ Key: nodePath,
+ Value: vp,
+ Dir: dir,
+ ModifiedIndex: mkV2Rev(resp.Header.Revision),
+ CreatedIndex: mkV2Rev(createRev),
+ },
+ PrevNode: pn,
+ EtcdIndex: mkV2Rev(resp.Header.Revision),
+ }, nil
+}
+
+func (s *v2v3Store) Update(nodePath, newValue string, expireOpts store.TTLOptionSet) (*store.Event, error) {
+ if isRoot(nodePath) {
+ return nil, etcdErr.NewError(etcdErr.EcodeRootROnly, nodePath, 0)
+ }
+
+ if expireOpts.Refresh || !expireOpts.ExpireTime.IsZero() {
+ return nil, errUnsupported
+ }
+
+ key := s.mkPath(nodePath)
+ ecode := 0
+ applyf := func(stm concurrency.STM) error {
+ if rev := stm.Rev(key + "/"); rev != 0 {
+ ecode = etcdErr.EcodeNotFile
+ return nil
+ }
+ if rev := stm.Rev(key); rev == 0 {
+ ecode = etcdErr.EcodeKeyNotFound
+ return nil
+ }
+ stm.Put(key, newValue, clientv3.WithPrevKV())
+ stm.Put(s.mkActionKey(), store.Update)
+ return nil
+ }
+
+ resp, err := s.newSTM(applyf)
+ if err != nil {
+ return nil, err
+ }
+ if ecode != 0 {
+ return nil, etcdErr.NewError(etcdErr.EcodeNotFile, nodePath, mkV2Rev(resp.Header.Revision))
+ }
+
+ pkv := prevKeyFromPuts(resp)
+ return &store.Event{
+ Action: store.Update,
+ Node: &store.NodeExtern{
+ Key: nodePath,
+ Value: &newValue,
+ ModifiedIndex: mkV2Rev(resp.Header.Revision),
+ CreatedIndex: mkV2Rev(pkv.CreateRevision),
+ },
+ PrevNode: s.mkV2Node(pkv),
+ EtcdIndex: mkV2Rev(resp.Header.Revision),
+ }, nil
+}
+
+func (s *v2v3Store) Create(
+ nodePath string,
+ dir bool,
+ value string,
+ unique bool,
+ expireOpts store.TTLOptionSet,
+) (*store.Event, error) {
+ if isRoot(nodePath) {
+ return nil, etcdErr.NewError(etcdErr.EcodeRootROnly, nodePath, 0)
+ }
+ if expireOpts.Refresh || !expireOpts.ExpireTime.IsZero() {
+ return nil, errUnsupported
+ }
+ ecode := 0
+ applyf := func(stm concurrency.STM) error {
+ ecode = 0
+ key := s.mkPath(nodePath)
+ if unique {
+ // append unique item under the node path
+ for {
+ key = nodePath + "/" + fmt.Sprintf("%020s", time.Now())
+ key = path.Clean(path.Join("/", key))
+ key = s.mkPath(key)
+ if stm.Rev(key) == 0 {
+ break
+ }
+ }
+ }
+ if stm.Rev(key) > 0 || stm.Rev(key+"/") > 0 {
+ ecode = etcdErr.EcodeNodeExist
+ return nil
+ }
+ // build path if any directories in path do not exist
+ dirs := []string{}
+ for p := path.Dir(nodePath); !isRoot(p); p = path.Dir(p) {
+ pp := s.mkPath(p)
+ if stm.Rev(pp) > 0 {
+ ecode = etcdErr.EcodeNotDir
+ return nil
+ }
+ if stm.Rev(pp+"/") == 0 {
+ dirs = append(dirs, pp+"/")
+ }
+ }
+ for _, d := range dirs {
+ stm.Put(d, "")
+ }
+
+ if dir {
+ // directories marked with extra slash in key name
+ key += "/"
+ }
+ stm.Put(key, value)
+ stm.Put(s.mkActionKey(), store.Create)
+ return nil
+ }
+
+ resp, err := s.newSTM(applyf)
+ if err != nil {
+ return nil, err
+ }
+ if ecode != 0 {
+ return nil, etcdErr.NewError(ecode, nodePath, mkV2Rev(resp.Header.Revision))
+ }
+
+ var v *string
+ if !dir {
+ v = &value
+ }
+
+ return &store.Event{
+ Action: store.Create,
+ Node: &store.NodeExtern{
+ Key: nodePath,
+ Value: v,
+ Dir: dir,
+ ModifiedIndex: mkV2Rev(resp.Header.Revision),
+ CreatedIndex: mkV2Rev(resp.Header.Revision),
+ },
+ EtcdIndex: mkV2Rev(resp.Header.Revision),
+ }, nil
+}
+
+func (s *v2v3Store) CompareAndSwap(
+ nodePath string,
+ prevValue string,
+ prevIndex uint64,
+ value string,
+ expireOpts store.TTLOptionSet,
+) (*store.Event, error) {
+ if isRoot(nodePath) {
+ return nil, etcdErr.NewError(etcdErr.EcodeRootROnly, nodePath, 0)
+ }
+ if expireOpts.Refresh || !expireOpts.ExpireTime.IsZero() {
+ return nil, errUnsupported
+ }
+
+ key := s.mkPath(nodePath)
+ resp, err := s.c.Txn(s.ctx).If(
+ s.mkCompare(nodePath, prevValue, prevIndex)...,
+ ).Then(
+ clientv3.OpPut(key, value, clientv3.WithPrevKV()),
+ clientv3.OpPut(s.mkActionKey(), store.CompareAndSwap),
+ ).Else(
+ clientv3.OpGet(key),
+ clientv3.OpGet(key+"/"),
+ ).Commit()
+
+ if err != nil {
+ return nil, err
+ }
+ if !resp.Succeeded {
+ return nil, compareFail(nodePath, prevValue, prevIndex, resp)
+ }
+
+ pkv := resp.Responses[0].GetResponsePut().PrevKv
+ return &store.Event{
+ Action: store.CompareAndSwap,
+ Node: &store.NodeExtern{
+ Key: nodePath,
+ Value: &value,
+ CreatedIndex: mkV2Rev(pkv.CreateRevision),
+ ModifiedIndex: mkV2Rev(resp.Header.Revision),
+ },
+ PrevNode: s.mkV2Node(pkv),
+ EtcdIndex: mkV2Rev(resp.Header.Revision),
+ }, nil
+}
+
+func (s *v2v3Store) Delete(nodePath string, dir, recursive bool) (*store.Event, error) {
+ if isRoot(nodePath) {
+ return nil, etcdErr.NewError(etcdErr.EcodeRootROnly, nodePath, 0)
+ }
+ if !dir && !recursive {
+ return s.deleteNode(nodePath)
+ }
+ if !recursive {
+ return s.deleteEmptyDir(nodePath)
+ }
+
+ dels := make([]clientv3.Op, maxPathDepth+1)
+ dels[0] = clientv3.OpDelete(s.mkPath(nodePath)+"/", clientv3.WithPrevKV())
+ for i := 1; i < maxPathDepth; i++ {
+ dels[i] = clientv3.OpDelete(s.mkPathDepth(nodePath, i), clientv3.WithPrefix())
+ }
+ dels[maxPathDepth] = clientv3.OpPut(s.mkActionKey(), store.Delete)
+
+ resp, err := s.c.Txn(s.ctx).If(
+ clientv3.Compare(clientv3.Version(s.mkPath(nodePath)+"/"), ">", 0),
+ clientv3.Compare(clientv3.Version(s.mkPathDepth(nodePath, maxPathDepth)+"/"), "=", 0),
+ ).Then(
+ dels...,
+ ).Commit()
+ if err != nil {
+ return nil, err
+ }
+ if !resp.Succeeded {
+ return nil, etcdErr.NewError(etcdErr.EcodeNodeExist, nodePath, mkV2Rev(resp.Header.Revision))
+ }
+ dresp := resp.Responses[0].GetResponseDeleteRange()
+ return &store.Event{
+ Action: store.Delete,
+ PrevNode: s.mkV2Node(dresp.PrevKvs[0]),
+ EtcdIndex: mkV2Rev(resp.Header.Revision),
+ }, nil
+}
+
+func (s *v2v3Store) deleteEmptyDir(nodePath string) (*store.Event, error) {
+ resp, err := s.c.Txn(s.ctx).If(
+ clientv3.Compare(clientv3.Version(s.mkPathDepth(nodePath, 1)), "=", 0).WithPrefix(),
+ ).Then(
+ clientv3.OpDelete(s.mkPath(nodePath)+"/", clientv3.WithPrevKV()),
+ clientv3.OpPut(s.mkActionKey(), store.Delete),
+ ).Commit()
+ if err != nil {
+ return nil, err
+ }
+ if !resp.Succeeded {
+ return nil, etcdErr.NewError(etcdErr.EcodeDirNotEmpty, nodePath, mkV2Rev(resp.Header.Revision))
+ }
+ dresp := resp.Responses[0].GetResponseDeleteRange()
+ if len(dresp.PrevKvs) == 0 {
+ return nil, etcdErr.NewError(etcdErr.EcodeNodeExist, nodePath, mkV2Rev(resp.Header.Revision))
+ }
+ return &store.Event{
+ Action: store.Delete,
+ PrevNode: s.mkV2Node(dresp.PrevKvs[0]),
+ EtcdIndex: mkV2Rev(resp.Header.Revision),
+ }, nil
+}
+
+func (s *v2v3Store) deleteNode(nodePath string) (*store.Event, error) {
+ resp, err := s.c.Txn(s.ctx).If(
+ clientv3.Compare(clientv3.Version(s.mkPath(nodePath)+"/"), "=", 0),
+ ).Then(
+ clientv3.OpDelete(s.mkPath(nodePath), clientv3.WithPrevKV()),
+ clientv3.OpPut(s.mkActionKey(), store.Delete),
+ ).Commit()
+ if err != nil {
+ return nil, err
+ }
+ if !resp.Succeeded {
+ return nil, etcdErr.NewError(etcdErr.EcodeNotFile, nodePath, mkV2Rev(resp.Header.Revision))
+ }
+ pkvs := resp.Responses[0].GetResponseDeleteRange().PrevKvs
+ if len(pkvs) == 0 {
+ return nil, etcdErr.NewError(etcdErr.EcodeKeyNotFound, nodePath, mkV2Rev(resp.Header.Revision))
+ }
+ pkv := pkvs[0]
+ return &store.Event{
+ Action: store.Delete,
+ Node: &store.NodeExtern{
+ Key: nodePath,
+ CreatedIndex: mkV2Rev(pkv.CreateRevision),
+ ModifiedIndex: mkV2Rev(resp.Header.Revision),
+ },
+ PrevNode: s.mkV2Node(pkv),
+ EtcdIndex: mkV2Rev(resp.Header.Revision),
+ }, nil
+}
+
+func (s *v2v3Store) CompareAndDelete(nodePath, prevValue string, prevIndex uint64) (*store.Event, error) {
+ if isRoot(nodePath) {
+ return nil, etcdErr.NewError(etcdErr.EcodeRootROnly, nodePath, 0)
+ }
+
+ key := s.mkPath(nodePath)
+ resp, err := s.c.Txn(s.ctx).If(
+ s.mkCompare(nodePath, prevValue, prevIndex)...,
+ ).Then(
+ clientv3.OpDelete(key, clientv3.WithPrevKV()),
+ clientv3.OpPut(s.mkActionKey(), store.CompareAndDelete),
+ ).Else(
+ clientv3.OpGet(key),
+ clientv3.OpGet(key+"/"),
+ ).Commit()
+
+ if err != nil {
+ return nil, err
+ }
+ if !resp.Succeeded {
+ return nil, compareFail(nodePath, prevValue, prevIndex, resp)
+ }
+
+ // len(pkvs) > 1 since txn only succeeds when key exists
+ pkv := resp.Responses[0].GetResponseDeleteRange().PrevKvs[0]
+ return &store.Event{
+ Action: store.CompareAndDelete,
+ Node: &store.NodeExtern{
+ Key: nodePath,
+ CreatedIndex: mkV2Rev(pkv.CreateRevision),
+ ModifiedIndex: mkV2Rev(resp.Header.Revision),
+ },
+ PrevNode: s.mkV2Node(pkv),
+ EtcdIndex: mkV2Rev(resp.Header.Revision),
+ }, nil
+}
+
+func compareFail(nodePath, prevValue string, prevIndex uint64, resp *clientv3.TxnResponse) error {
+ if dkvs := resp.Responses[1].GetResponseRange().Kvs; len(dkvs) > 0 {
+ return etcdErr.NewError(etcdErr.EcodeNotFile, nodePath, mkV2Rev(resp.Header.Revision))
+ }
+ kvs := resp.Responses[0].GetResponseRange().Kvs
+ if len(kvs) == 0 {
+ return etcdErr.NewError(etcdErr.EcodeKeyNotFound, nodePath, mkV2Rev(resp.Header.Revision))
+ }
+ kv := kvs[0]
+ indexMatch := (prevIndex == 0 || kv.ModRevision == int64(prevIndex))
+ valueMatch := (prevValue == "" || string(kv.Value) == prevValue)
+ var cause string
+ switch {
+ case indexMatch && !valueMatch:
+ cause = fmt.Sprintf("[%v != %v]", prevValue, string(kv.Value))
+ case valueMatch && !indexMatch:
+ cause = fmt.Sprintf("[%v != %v]", prevIndex, kv.ModRevision)
+ default:
+ cause = fmt.Sprintf("[%v != %v] [%v != %v]", prevValue, string(kv.Value), prevIndex, kv.ModRevision)
+ }
+ return etcdErr.NewError(etcdErr.EcodeTestFailed, cause, mkV2Rev(resp.Header.Revision))
+}
+
+func (s *v2v3Store) mkCompare(nodePath, prevValue string, prevIndex uint64) []clientv3.Cmp {
+ key := s.mkPath(nodePath)
+ cmps := []clientv3.Cmp{clientv3.Compare(clientv3.Version(key), ">", 0)}
+ if prevIndex != 0 {
+ cmps = append(cmps, clientv3.Compare(clientv3.ModRevision(key), "=", mkV3Rev(prevIndex)))
+ }
+ if prevValue != "" {
+ cmps = append(cmps, clientv3.Compare(clientv3.Value(key), "=", prevValue))
+ }
+ return cmps
+}
+
+func (s *v2v3Store) JsonStats() []byte { panic("STUB") }
+func (s *v2v3Store) DeleteExpiredKeys(cutoff time.Time) { panic("STUB") }
+
+func (s *v2v3Store) Version() int { return 2 }
+
+// TODO: move this out of the Store interface?
+
+func (s *v2v3Store) Save() ([]byte, error) { panic("STUB") }
+func (s *v2v3Store) Recovery(state []byte) error { panic("STUB") }
+func (s *v2v3Store) Clone() store.Store { panic("STUB") }
+func (s *v2v3Store) SaveNoCopy() ([]byte, error) { panic("STUB") }
+func (s *v2v3Store) HasTTLKeys() bool { panic("STUB") }
+
+func (s *v2v3Store) mkPath(nodePath string) string { return s.mkPathDepth(nodePath, 0) }
+
+func (s *v2v3Store) mkNodePath(p string) string {
+ return path.Clean(p[len(s.pfx)+len("/k/000/"):])
+}
+
+// mkPathDepth makes a path to a key that encodes its directory depth
+// for fast directory listing. If a depth is provided, it is added
+// to the computed depth.
+func (s *v2v3Store) mkPathDepth(nodePath string, depth int) string {
+ normalForm := path.Clean(path.Join("/", nodePath))
+ n := strings.Count(normalForm, "/") + depth
+ return fmt.Sprintf("%s/%03d/k/%s", s.pfx, n, normalForm)
+}
+
+func (s *v2v3Store) mkActionKey() string { return s.pfx + "/act" }
+
+func isRoot(s string) bool { return len(s) == 0 || s == "/" || s == "/0" || s == "/1" }
+
+func mkV2Rev(v3Rev int64) uint64 {
+ if v3Rev == 0 {
+ return 0
+ }
+ return uint64(v3Rev - 1)
+}
+
+func mkV3Rev(v2Rev uint64) int64 {
+ if v2Rev == 0 {
+ return 0
+ }
+ return int64(v2Rev + 1)
+}
+
+// mkV2Node creates a V2 NodeExtern from a V3 KeyValue
+func (s *v2v3Store) mkV2Node(kv *mvccpb.KeyValue) *store.NodeExtern {
+ if kv == nil {
+ return nil
+ }
+ n := &store.NodeExtern{
+ Key: string(s.mkNodePath(string(kv.Key))),
+ Dir: kv.Key[len(kv.Key)-1] == '/',
+ CreatedIndex: mkV2Rev(kv.CreateRevision),
+ ModifiedIndex: mkV2Rev(kv.ModRevision),
+ }
+ if !n.Dir {
+ v := string(kv.Value)
+ n.Value = &v
+ }
+ return n
+}
+
+// prevKeyFromPuts gets the prev key that is being put; ignores
+// the put action response.
+func prevKeyFromPuts(resp *clientv3.TxnResponse) *mvccpb.KeyValue {
+ for _, r := range resp.Responses {
+ pkv := r.GetResponsePut().PrevKv
+ if pkv != nil && pkv.CreateRevision > 0 {
+ return pkv
+ }
+ }
+ return nil
+}
+
+func (s *v2v3Store) newSTM(applyf func(concurrency.STM) error) (*clientv3.TxnResponse, error) {
+ return concurrency.NewSTM(s.c, applyf, concurrency.WithIsolation(concurrency.Serializable))
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v2v3/watcher.go b/vendor/github.com/coreos/etcd/etcdserver/api/v2v3/watcher.go
new file mode 100644
index 0000000..1c2680e
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v2v3/watcher.go
@@ -0,0 +1,140 @@
+// Copyright 2017 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 v2v3
+
+import (
+ "context"
+ "strings"
+
+ "github.com/coreos/etcd/clientv3"
+ etcdErr "github.com/coreos/etcd/error"
+ "github.com/coreos/etcd/store"
+)
+
+func (s *v2v3Store) Watch(prefix string, recursive, stream bool, sinceIndex uint64) (store.Watcher, error) {
+ ctx, cancel := context.WithCancel(s.ctx)
+ wch := s.c.Watch(
+ ctx,
+ // TODO: very pricey; use a single store-wide watch in future
+ s.pfx,
+ clientv3.WithPrefix(),
+ clientv3.WithRev(int64(sinceIndex)),
+ clientv3.WithCreatedNotify(),
+ clientv3.WithPrevKV())
+ resp, ok := <-wch
+ if err := resp.Err(); err != nil || !ok {
+ cancel()
+ return nil, etcdErr.NewError(etcdErr.EcodeRaftInternal, prefix, 0)
+ }
+
+ evc, donec := make(chan *store.Event), make(chan struct{})
+ go func() {
+ defer func() {
+ close(evc)
+ close(donec)
+ }()
+ for resp := range wch {
+ for _, ev := range s.mkV2Events(resp) {
+ k := ev.Node.Key
+ if recursive {
+ if !strings.HasPrefix(k, prefix) {
+ continue
+ }
+ // accept events on hidden keys given in prefix
+ k = strings.Replace(k, prefix, "/", 1)
+ // ignore hidden keys deeper than prefix
+ if strings.Contains(k, "/_") {
+ continue
+ }
+ }
+ if !recursive && k != prefix {
+ continue
+ }
+ select {
+ case evc <- ev:
+ case <-ctx.Done():
+ return
+ }
+ if !stream {
+ return
+ }
+ }
+ }
+ }()
+
+ return &v2v3Watcher{
+ startRev: resp.Header.Revision,
+ evc: evc,
+ donec: donec,
+ cancel: cancel,
+ }, nil
+}
+
+func (s *v2v3Store) mkV2Events(wr clientv3.WatchResponse) (evs []*store.Event) {
+ ak := s.mkActionKey()
+ for _, rev := range mkRevs(wr) {
+ var act, key *clientv3.Event
+ for _, ev := range rev {
+ if string(ev.Kv.Key) == ak {
+ act = ev
+ } else if key != nil && len(key.Kv.Key) < len(ev.Kv.Key) {
+ // use longest key to ignore intermediate new
+ // directories from Create.
+ key = ev
+ } else if key == nil {
+ key = ev
+ }
+ }
+ v2ev := &store.Event{
+ Action: string(act.Kv.Value),
+ Node: s.mkV2Node(key.Kv),
+ PrevNode: s.mkV2Node(key.PrevKv),
+ EtcdIndex: mkV2Rev(wr.Header.Revision),
+ }
+ evs = append(evs, v2ev)
+ }
+ return evs
+}
+
+func mkRevs(wr clientv3.WatchResponse) (revs [][]*clientv3.Event) {
+ var curRev []*clientv3.Event
+ for _, ev := range wr.Events {
+ if curRev != nil && ev.Kv.ModRevision != curRev[0].Kv.ModRevision {
+ revs = append(revs, curRev)
+ curRev = nil
+ }
+ curRev = append(curRev, ev)
+ }
+ if curRev != nil {
+ revs = append(revs, curRev)
+ }
+ return revs
+}
+
+type v2v3Watcher struct {
+ startRev int64
+ evc chan *store.Event
+ donec chan struct{}
+ cancel context.CancelFunc
+}
+
+func (w *v2v3Watcher) StartIndex() uint64 { return mkV2Rev(w.startRev) }
+
+func (w *v2v3Watcher) Remove() {
+ w.cancel()
+ <-w.donec
+}
+
+func (w *v2v3Watcher) EventChan() chan *store.Event { return w.evc }
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v3client/doc.go b/vendor/github.com/coreos/etcd/etcdserver/api/v3client/doc.go
new file mode 100644
index 0000000..310715f
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v3client/doc.go
@@ -0,0 +1,45 @@
+// Copyright 2017 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 v3client provides clientv3 interfaces from an etcdserver.
+//
+// Use v3client by creating an EtcdServer instance, then wrapping it with v3client.New:
+//
+// import (
+// "context"
+//
+// "github.com/coreos/etcd/embed"
+// "github.com/coreos/etcd/etcdserver/api/v3client"
+// )
+//
+// ...
+//
+// // create an embedded EtcdServer from the default configuration
+// cfg := embed.NewConfig()
+// cfg.Dir = "default.etcd"
+// e, err := embed.StartEtcd(cfg)
+// if err != nil {
+// // handle error!
+// }
+//
+// // wrap the EtcdServer with v3client
+// cli := v3client.New(e.Server)
+//
+// // use like an ordinary clientv3
+// resp, err := cli.Put(context.TODO(), "some-key", "it works!")
+// if err != nil {
+// // handle error!
+// }
+//
+package v3client
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v3client/v3client.go b/vendor/github.com/coreos/etcd/etcdserver/api/v3client/v3client.go
new file mode 100644
index 0000000..ab48ea7
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v3client/v3client.go
@@ -0,0 +1,66 @@
+// Copyright 2017 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 v3client
+
+import (
+ "context"
+ "time"
+
+ "github.com/coreos/etcd/clientv3"
+ "github.com/coreos/etcd/etcdserver"
+ "github.com/coreos/etcd/etcdserver/api/v3rpc"
+ "github.com/coreos/etcd/proxy/grpcproxy/adapter"
+)
+
+// New creates a clientv3 client that wraps an in-process EtcdServer. Instead
+// of making gRPC calls through sockets, the client makes direct function calls
+// to the etcd server through its api/v3rpc function interfaces.
+func New(s *etcdserver.EtcdServer) *clientv3.Client {
+ c := clientv3.NewCtxClient(context.Background())
+
+ kvc := adapter.KvServerToKvClient(v3rpc.NewQuotaKVServer(s))
+ c.KV = clientv3.NewKVFromKVClient(kvc, c)
+
+ lc := adapter.LeaseServerToLeaseClient(v3rpc.NewQuotaLeaseServer(s))
+ c.Lease = clientv3.NewLeaseFromLeaseClient(lc, c, time.Second)
+
+ wc := adapter.WatchServerToWatchClient(v3rpc.NewWatchServer(s))
+ c.Watcher = &watchWrapper{clientv3.NewWatchFromWatchClient(wc, c)}
+
+ mc := adapter.MaintenanceServerToMaintenanceClient(v3rpc.NewMaintenanceServer(s))
+ c.Maintenance = clientv3.NewMaintenanceFromMaintenanceClient(mc, c)
+
+ clc := adapter.ClusterServerToClusterClient(v3rpc.NewClusterServer(s))
+ c.Cluster = clientv3.NewClusterFromClusterClient(clc, c)
+
+ // TODO: implement clientv3.Auth interface?
+
+ return c
+}
+
+// BlankContext implements Stringer on a context so the ctx string doesn't
+// depend on the context's WithValue data, which tends to be unsynchronized
+// (e.g., x/net/trace), causing ctx.String() to throw data races.
+type blankContext struct{ context.Context }
+
+func (*blankContext) String() string { return "(blankCtx)" }
+
+// watchWrapper wraps clientv3 watch calls to blank out the context
+// to avoid races on trace data.
+type watchWrapper struct{ clientv3.Watcher }
+
+func (ww *watchWrapper) Watch(ctx context.Context, key string, opts ...clientv3.OpOption) clientv3.WatchChan {
+ return ww.Watcher.Watch(&blankContext{ctx}, key, opts...)
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v3election/doc.go b/vendor/github.com/coreos/etcd/etcdserver/api/v3election/doc.go
new file mode 100644
index 0000000..d6fefd7
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v3election/doc.go
@@ -0,0 +1,16 @@
+// Copyright 2017 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 v3election provides a v3 election service from an etcdserver.
+package v3election
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v3election/election.go b/vendor/github.com/coreos/etcd/etcdserver/api/v3election/election.go
new file mode 100644
index 0000000..c66d7a3
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v3election/election.go
@@ -0,0 +1,134 @@
+// Copyright 2017 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 v3election
+
+import (
+ "context"
+ "errors"
+
+ "github.com/coreos/etcd/clientv3"
+ "github.com/coreos/etcd/clientv3/concurrency"
+ epb "github.com/coreos/etcd/etcdserver/api/v3election/v3electionpb"
+)
+
+// ErrMissingLeaderKey is returned when election API request
+// is missing the "leader" field.
+var ErrMissingLeaderKey = errors.New(`"leader" field must be provided`)
+
+type electionServer struct {
+ c *clientv3.Client
+}
+
+func NewElectionServer(c *clientv3.Client) epb.ElectionServer {
+ return &electionServer{c}
+}
+
+func (es *electionServer) Campaign(ctx context.Context, req *epb.CampaignRequest) (*epb.CampaignResponse, error) {
+ s, err := es.session(ctx, req.Lease)
+ if err != nil {
+ return nil, err
+ }
+ e := concurrency.NewElection(s, string(req.Name))
+ if err = e.Campaign(ctx, string(req.Value)); err != nil {
+ return nil, err
+ }
+ return &epb.CampaignResponse{
+ Header: e.Header(),
+ Leader: &epb.LeaderKey{
+ Name: req.Name,
+ Key: []byte(e.Key()),
+ Rev: e.Rev(),
+ Lease: int64(s.Lease()),
+ },
+ }, nil
+}
+
+func (es *electionServer) Proclaim(ctx context.Context, req *epb.ProclaimRequest) (*epb.ProclaimResponse, error) {
+ if req.Leader == nil {
+ return nil, ErrMissingLeaderKey
+ }
+ s, err := es.session(ctx, req.Leader.Lease)
+ if err != nil {
+ return nil, err
+ }
+ e := concurrency.ResumeElection(s, string(req.Leader.Name), string(req.Leader.Key), req.Leader.Rev)
+ if err := e.Proclaim(ctx, string(req.Value)); err != nil {
+ return nil, err
+ }
+ return &epb.ProclaimResponse{Header: e.Header()}, nil
+}
+
+func (es *electionServer) Observe(req *epb.LeaderRequest, stream epb.Election_ObserveServer) error {
+ s, err := es.session(stream.Context(), -1)
+ if err != nil {
+ return err
+ }
+ e := concurrency.NewElection(s, string(req.Name))
+ ch := e.Observe(stream.Context())
+ for stream.Context().Err() == nil {
+ select {
+ case <-stream.Context().Done():
+ case resp, ok := <-ch:
+ if !ok {
+ return nil
+ }
+ lresp := &epb.LeaderResponse{Header: resp.Header, Kv: resp.Kvs[0]}
+ if err := stream.Send(lresp); err != nil {
+ return err
+ }
+ }
+ }
+ return stream.Context().Err()
+}
+
+func (es *electionServer) Leader(ctx context.Context, req *epb.LeaderRequest) (*epb.LeaderResponse, error) {
+ s, err := es.session(ctx, -1)
+ if err != nil {
+ return nil, err
+ }
+ l, lerr := concurrency.NewElection(s, string(req.Name)).Leader(ctx)
+ if lerr != nil {
+ return nil, lerr
+ }
+ return &epb.LeaderResponse{Header: l.Header, Kv: l.Kvs[0]}, nil
+}
+
+func (es *electionServer) Resign(ctx context.Context, req *epb.ResignRequest) (*epb.ResignResponse, error) {
+ if req.Leader == nil {
+ return nil, ErrMissingLeaderKey
+ }
+ s, err := es.session(ctx, req.Leader.Lease)
+ if err != nil {
+ return nil, err
+ }
+ e := concurrency.ResumeElection(s, string(req.Leader.Name), string(req.Leader.Key), req.Leader.Rev)
+ if err := e.Resign(ctx); err != nil {
+ return nil, err
+ }
+ return &epb.ResignResponse{Header: e.Header()}, nil
+}
+
+func (es *electionServer) session(ctx context.Context, lease int64) (*concurrency.Session, error) {
+ s, err := concurrency.NewSession(
+ es.c,
+ concurrency.WithLease(clientv3.LeaseID(lease)),
+ concurrency.WithContext(ctx),
+ )
+ if err != nil {
+ return nil, err
+ }
+ s.Orphan()
+ return s, nil
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v3election/v3electionpb/gw/v3election.pb.gw.go b/vendor/github.com/coreos/etcd/etcdserver/api/v3election/v3electionpb/gw/v3election.pb.gw.go
new file mode 100644
index 0000000..58368bb
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v3election/v3electionpb/gw/v3election.pb.gw.go
@@ -0,0 +1,313 @@
+// Code generated by protoc-gen-grpc-gateway. DO NOT EDIT.
+// source: etcdserver/api/v3election/v3electionpb/v3election.proto
+
+/*
+Package v3electionpb is a reverse proxy.
+
+It translates gRPC into RESTful JSON APIs.
+*/
+package gw
+
+import (
+ "github.com/coreos/etcd/etcdserver/api/v3election/v3electionpb"
+ "io"
+ "net/http"
+
+ "github.com/golang/protobuf/proto"
+ "github.com/grpc-ecosystem/grpc-gateway/runtime"
+ "github.com/grpc-ecosystem/grpc-gateway/utilities"
+ "golang.org/x/net/context"
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/grpclog"
+ "google.golang.org/grpc/status"
+)
+
+var _ codes.Code
+var _ io.Reader
+var _ status.Status
+var _ = runtime.String
+var _ = utilities.NewDoubleArray
+
+func request_Election_Campaign_0(ctx context.Context, marshaler runtime.Marshaler, client v3electionpb.ElectionClient, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
+ var protoReq v3electionpb.CampaignRequest
+ var metadata runtime.ServerMetadata
+
+ if err := marshaler.NewDecoder(req.Body).Decode(&protoReq); err != nil {
+ return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err)
+ }
+
+ msg, err := client.Campaign(ctx, &protoReq, grpc.Header(&metadata.HeaderMD), grpc.Trailer(&metadata.TrailerMD))
+ return msg, metadata, err
+
+}
+
+func request_Election_Proclaim_0(ctx context.Context, marshaler runtime.Marshaler, client v3electionpb.ElectionClient, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
+ var protoReq v3electionpb.ProclaimRequest
+ var metadata runtime.ServerMetadata
+
+ if err := marshaler.NewDecoder(req.Body).Decode(&protoReq); err != nil {
+ return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err)
+ }
+
+ msg, err := client.Proclaim(ctx, &protoReq, grpc.Header(&metadata.HeaderMD), grpc.Trailer(&metadata.TrailerMD))
+ return msg, metadata, err
+
+}
+
+func request_Election_Leader_0(ctx context.Context, marshaler runtime.Marshaler, client v3electionpb.ElectionClient, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
+ var protoReq v3electionpb.LeaderRequest
+ var metadata runtime.ServerMetadata
+
+ if err := marshaler.NewDecoder(req.Body).Decode(&protoReq); err != nil {
+ return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err)
+ }
+
+ msg, err := client.Leader(ctx, &protoReq, grpc.Header(&metadata.HeaderMD), grpc.Trailer(&metadata.TrailerMD))
+ return msg, metadata, err
+
+}
+
+func request_Election_Observe_0(ctx context.Context, marshaler runtime.Marshaler, client v3electionpb.ElectionClient, req *http.Request, pathParams map[string]string) (v3electionpb.Election_ObserveClient, runtime.ServerMetadata, error) {
+ var protoReq v3electionpb.LeaderRequest
+ var metadata runtime.ServerMetadata
+
+ if err := marshaler.NewDecoder(req.Body).Decode(&protoReq); err != nil {
+ return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err)
+ }
+
+ stream, err := client.Observe(ctx, &protoReq)
+ if err != nil {
+ return nil, metadata, err
+ }
+ header, err := stream.Header()
+ if err != nil {
+ return nil, metadata, err
+ }
+ metadata.HeaderMD = header
+ return stream, metadata, nil
+
+}
+
+func request_Election_Resign_0(ctx context.Context, marshaler runtime.Marshaler, client v3electionpb.ElectionClient, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
+ var protoReq v3electionpb.ResignRequest
+ var metadata runtime.ServerMetadata
+
+ if err := marshaler.NewDecoder(req.Body).Decode(&protoReq); err != nil {
+ return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err)
+ }
+
+ msg, err := client.Resign(ctx, &protoReq, grpc.Header(&metadata.HeaderMD), grpc.Trailer(&metadata.TrailerMD))
+ return msg, metadata, err
+
+}
+
+// RegisterElectionHandlerFromEndpoint is same as RegisterElectionHandler but
+// automatically dials to "endpoint" and closes the connection when "ctx" gets done.
+func RegisterElectionHandlerFromEndpoint(ctx context.Context, mux *runtime.ServeMux, endpoint string, opts []grpc.DialOption) (err error) {
+ conn, err := grpc.Dial(endpoint, opts...)
+ if err != nil {
+ return err
+ }
+ defer func() {
+ if err != nil {
+ if cerr := conn.Close(); cerr != nil {
+ grpclog.Printf("Failed to close conn to %s: %v", endpoint, cerr)
+ }
+ return
+ }
+ go func() {
+ <-ctx.Done()
+ if cerr := conn.Close(); cerr != nil {
+ grpclog.Printf("Failed to close conn to %s: %v", endpoint, cerr)
+ }
+ }()
+ }()
+
+ return RegisterElectionHandler(ctx, mux, conn)
+}
+
+// RegisterElectionHandler registers the http handlers for service Election to "mux".
+// The handlers forward requests to the grpc endpoint over "conn".
+func RegisterElectionHandler(ctx context.Context, mux *runtime.ServeMux, conn *grpc.ClientConn) error {
+ return RegisterElectionHandlerClient(ctx, mux, v3electionpb.NewElectionClient(conn))
+}
+
+// RegisterElectionHandler registers the http handlers for service Election to "mux".
+// The handlers forward requests to the grpc endpoint over the given implementation of "ElectionClient".
+// Note: the gRPC framework executes interceptors within the gRPC handler. If the passed in "ElectionClient"
+// doesn't go through the normal gRPC flow (creating a gRPC client etc.) then it will be up to the passed in
+// "ElectionClient" to call the correct interceptors.
+func RegisterElectionHandlerClient(ctx context.Context, mux *runtime.ServeMux, client v3electionpb.ElectionClient) error {
+
+ mux.Handle("POST", pattern_Election_Campaign_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+ ctx, cancel := context.WithCancel(req.Context())
+ defer cancel()
+ if cn, ok := w.(http.CloseNotifier); ok {
+ go func(done <-chan struct{}, closed <-chan bool) {
+ select {
+ case <-done:
+ case <-closed:
+ cancel()
+ }
+ }(ctx.Done(), cn.CloseNotify())
+ }
+ inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+ rctx, err := runtime.AnnotateContext(ctx, mux, req)
+ if err != nil {
+ runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+ return
+ }
+ resp, md, err := request_Election_Campaign_0(rctx, inboundMarshaler, client, req, pathParams)
+ ctx = runtime.NewServerMetadataContext(ctx, md)
+ if err != nil {
+ runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+ return
+ }
+
+ forward_Election_Campaign_0(ctx, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
+
+ })
+
+ mux.Handle("POST", pattern_Election_Proclaim_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+ ctx, cancel := context.WithCancel(req.Context())
+ defer cancel()
+ if cn, ok := w.(http.CloseNotifier); ok {
+ go func(done <-chan struct{}, closed <-chan bool) {
+ select {
+ case <-done:
+ case <-closed:
+ cancel()
+ }
+ }(ctx.Done(), cn.CloseNotify())
+ }
+ inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+ rctx, err := runtime.AnnotateContext(ctx, mux, req)
+ if err != nil {
+ runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+ return
+ }
+ resp, md, err := request_Election_Proclaim_0(rctx, inboundMarshaler, client, req, pathParams)
+ ctx = runtime.NewServerMetadataContext(ctx, md)
+ if err != nil {
+ runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+ return
+ }
+
+ forward_Election_Proclaim_0(ctx, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
+
+ })
+
+ mux.Handle("POST", pattern_Election_Leader_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+ ctx, cancel := context.WithCancel(req.Context())
+ defer cancel()
+ if cn, ok := w.(http.CloseNotifier); ok {
+ go func(done <-chan struct{}, closed <-chan bool) {
+ select {
+ case <-done:
+ case <-closed:
+ cancel()
+ }
+ }(ctx.Done(), cn.CloseNotify())
+ }
+ inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+ rctx, err := runtime.AnnotateContext(ctx, mux, req)
+ if err != nil {
+ runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+ return
+ }
+ resp, md, err := request_Election_Leader_0(rctx, inboundMarshaler, client, req, pathParams)
+ ctx = runtime.NewServerMetadataContext(ctx, md)
+ if err != nil {
+ runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+ return
+ }
+
+ forward_Election_Leader_0(ctx, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
+
+ })
+
+ mux.Handle("POST", pattern_Election_Observe_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+ ctx, cancel := context.WithCancel(req.Context())
+ defer cancel()
+ if cn, ok := w.(http.CloseNotifier); ok {
+ go func(done <-chan struct{}, closed <-chan bool) {
+ select {
+ case <-done:
+ case <-closed:
+ cancel()
+ }
+ }(ctx.Done(), cn.CloseNotify())
+ }
+ inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+ rctx, err := runtime.AnnotateContext(ctx, mux, req)
+ if err != nil {
+ runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+ return
+ }
+ resp, md, err := request_Election_Observe_0(rctx, inboundMarshaler, client, req, pathParams)
+ ctx = runtime.NewServerMetadataContext(ctx, md)
+ if err != nil {
+ runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+ return
+ }
+
+ forward_Election_Observe_0(ctx, mux, outboundMarshaler, w, req, func() (proto.Message, error) { return resp.Recv() }, mux.GetForwardResponseOptions()...)
+
+ })
+
+ mux.Handle("POST", pattern_Election_Resign_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+ ctx, cancel := context.WithCancel(req.Context())
+ defer cancel()
+ if cn, ok := w.(http.CloseNotifier); ok {
+ go func(done <-chan struct{}, closed <-chan bool) {
+ select {
+ case <-done:
+ case <-closed:
+ cancel()
+ }
+ }(ctx.Done(), cn.CloseNotify())
+ }
+ inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+ rctx, err := runtime.AnnotateContext(ctx, mux, req)
+ if err != nil {
+ runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+ return
+ }
+ resp, md, err := request_Election_Resign_0(rctx, inboundMarshaler, client, req, pathParams)
+ ctx = runtime.NewServerMetadataContext(ctx, md)
+ if err != nil {
+ runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+ return
+ }
+
+ forward_Election_Resign_0(ctx, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
+
+ })
+
+ return nil
+}
+
+var (
+ pattern_Election_Campaign_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2}, []string{"v3beta", "election", "campaign"}, ""))
+
+ pattern_Election_Proclaim_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2}, []string{"v3beta", "election", "proclaim"}, ""))
+
+ pattern_Election_Leader_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2}, []string{"v3beta", "election", "leader"}, ""))
+
+ pattern_Election_Observe_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2}, []string{"v3beta", "election", "observe"}, ""))
+
+ pattern_Election_Resign_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2}, []string{"v3beta", "election", "resign"}, ""))
+)
+
+var (
+ forward_Election_Campaign_0 = runtime.ForwardResponseMessage
+
+ forward_Election_Proclaim_0 = runtime.ForwardResponseMessage
+
+ forward_Election_Leader_0 = runtime.ForwardResponseMessage
+
+ forward_Election_Observe_0 = runtime.ForwardResponseStream
+
+ forward_Election_Resign_0 = runtime.ForwardResponseMessage
+)
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v3election/v3electionpb/v3election.proto b/vendor/github.com/coreos/etcd/etcdserver/api/v3election/v3electionpb/v3election.proto
new file mode 100644
index 0000000..cb475b8
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v3election/v3electionpb/v3election.proto
@@ -0,0 +1,119 @@
+syntax = "proto3";
+package v3electionpb;
+
+import "gogoproto/gogo.proto";
+import "etcd/etcdserver/etcdserverpb/rpc.proto";
+import "etcd/mvcc/mvccpb/kv.proto";
+
+// for grpc-gateway
+import "google/api/annotations.proto";
+
+option (gogoproto.marshaler_all) = true;
+option (gogoproto.unmarshaler_all) = true;
+
+// The election service exposes client-side election facilities as a gRPC interface.
+service Election {
+ // Campaign waits to acquire leadership in an election, returning a LeaderKey
+ // representing the leadership if successful. The LeaderKey can then be used
+ // to issue new values on the election, transactionally guard API requests on
+ // leadership still being held, and resign from the election.
+ rpc Campaign(CampaignRequest) returns (CampaignResponse) {
+ option (google.api.http) = {
+ post: "/v3beta/election/campaign"
+ body: "*"
+ };
+ }
+ // Proclaim updates the leader's posted value with a new value.
+ rpc Proclaim(ProclaimRequest) returns (ProclaimResponse) {
+ option (google.api.http) = {
+ post: "/v3beta/election/proclaim"
+ body: "*"
+ };
+ }
+ // Leader returns the current election proclamation, if any.
+ rpc Leader(LeaderRequest) returns (LeaderResponse) {
+ option (google.api.http) = {
+ post: "/v3beta/election/leader"
+ body: "*"
+ };
+ }
+ // Observe streams election proclamations in-order as made by the election's
+ // elected leaders.
+ rpc Observe(LeaderRequest) returns (stream LeaderResponse) {
+ option (google.api.http) = {
+ post: "/v3beta/election/observe"
+ body: "*"
+ };
+ }
+ // Resign releases election leadership so other campaigners may acquire
+ // leadership on the election.
+ rpc Resign(ResignRequest) returns (ResignResponse) {
+ option (google.api.http) = {
+ post: "/v3beta/election/resign"
+ body: "*"
+ };
+ }
+}
+
+message CampaignRequest {
+ // name is the election's identifier for the campaign.
+ bytes name = 1;
+ // lease is the ID of the lease attached to leadership of the election. If the
+ // lease expires or is revoked before resigning leadership, then the
+ // leadership is transferred to the next campaigner, if any.
+ int64 lease = 2;
+ // value is the initial proclaimed value set when the campaigner wins the
+ // election.
+ bytes value = 3;
+}
+
+message CampaignResponse {
+ etcdserverpb.ResponseHeader header = 1;
+ // leader describes the resources used for holding leadereship of the election.
+ LeaderKey leader = 2;
+}
+
+message LeaderKey {
+ // name is the election identifier that correponds to the leadership key.
+ bytes name = 1;
+ // key is an opaque key representing the ownership of the election. If the key
+ // is deleted, then leadership is lost.
+ bytes key = 2;
+ // rev is the creation revision of the key. It can be used to test for ownership
+ // of an election during transactions by testing the key's creation revision
+ // matches rev.
+ int64 rev = 3;
+ // lease is the lease ID of the election leader.
+ int64 lease = 4;
+}
+
+message LeaderRequest {
+ // name is the election identifier for the leadership information.
+ bytes name = 1;
+}
+
+message LeaderResponse {
+ etcdserverpb.ResponseHeader header = 1;
+ // kv is the key-value pair representing the latest leader update.
+ mvccpb.KeyValue kv = 2;
+}
+
+message ResignRequest {
+ // leader is the leadership to relinquish by resignation.
+ LeaderKey leader = 1;
+}
+
+message ResignResponse {
+ etcdserverpb.ResponseHeader header = 1;
+}
+
+message ProclaimRequest {
+ // leader is the leadership hold on the election.
+ LeaderKey leader = 1;
+ // value is an update meant to overwrite the leader's current value.
+ bytes value = 2;
+}
+
+message ProclaimResponse {
+ etcdserverpb.ResponseHeader header = 1;
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v3lock/doc.go b/vendor/github.com/coreos/etcd/etcdserver/api/v3lock/doc.go
new file mode 100644
index 0000000..e0a1008
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v3lock/doc.go
@@ -0,0 +1,16 @@
+// Copyright 2017 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 v3lock provides a v3 locking service from an etcdserver.
+package v3lock
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v3lock/lock.go b/vendor/github.com/coreos/etcd/etcdserver/api/v3lock/lock.go
new file mode 100644
index 0000000..a5efcba
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v3lock/lock.go
@@ -0,0 +1,56 @@
+// Copyright 2017 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 v3lock
+
+import (
+ "context"
+
+ "github.com/coreos/etcd/clientv3"
+ "github.com/coreos/etcd/clientv3/concurrency"
+ "github.com/coreos/etcd/etcdserver/api/v3lock/v3lockpb"
+)
+
+type lockServer struct {
+ c *clientv3.Client
+}
+
+func NewLockServer(c *clientv3.Client) v3lockpb.LockServer {
+ return &lockServer{c}
+}
+
+func (ls *lockServer) Lock(ctx context.Context, req *v3lockpb.LockRequest) (*v3lockpb.LockResponse, error) {
+ s, err := concurrency.NewSession(
+ ls.c,
+ concurrency.WithLease(clientv3.LeaseID(req.Lease)),
+ concurrency.WithContext(ctx),
+ )
+ if err != nil {
+ return nil, err
+ }
+ s.Orphan()
+ m := concurrency.NewMutex(s, string(req.Name))
+ if err = m.Lock(ctx); err != nil {
+ return nil, err
+ }
+ return &v3lockpb.LockResponse{Header: m.Header(), Key: []byte(m.Key())}, nil
+}
+
+func (ls *lockServer) Unlock(ctx context.Context, req *v3lockpb.UnlockRequest) (*v3lockpb.UnlockResponse, error) {
+ resp, err := ls.c.Delete(ctx, string(req.Key))
+ if err != nil {
+ return nil, err
+ }
+ return &v3lockpb.UnlockResponse{Header: resp.Header}, nil
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v3lock/v3lockpb/gw/v3lock.pb.gw.go b/vendor/github.com/coreos/etcd/etcdserver/api/v3lock/v3lockpb/gw/v3lock.pb.gw.go
new file mode 100644
index 0000000..efecc45
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v3lock/v3lockpb/gw/v3lock.pb.gw.go
@@ -0,0 +1,167 @@
+// Code generated by protoc-gen-grpc-gateway. DO NOT EDIT.
+// source: etcdserver/api/v3lock/v3lockpb/v3lock.proto
+
+/*
+Package v3lockpb is a reverse proxy.
+
+It translates gRPC into RESTful JSON APIs.
+*/
+package gw
+
+import (
+ "github.com/coreos/etcd/etcdserver/api/v3lock/v3lockpb"
+ "io"
+ "net/http"
+
+ "github.com/golang/protobuf/proto"
+ "github.com/grpc-ecosystem/grpc-gateway/runtime"
+ "github.com/grpc-ecosystem/grpc-gateway/utilities"
+ "golang.org/x/net/context"
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/grpclog"
+ "google.golang.org/grpc/status"
+)
+
+var _ codes.Code
+var _ io.Reader
+var _ status.Status
+var _ = runtime.String
+var _ = utilities.NewDoubleArray
+
+func request_Lock_Lock_0(ctx context.Context, marshaler runtime.Marshaler, client v3lockpb.LockClient, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
+ var protoReq v3lockpb.LockRequest
+ var metadata runtime.ServerMetadata
+
+ if err := marshaler.NewDecoder(req.Body).Decode(&protoReq); err != nil {
+ return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err)
+ }
+
+ msg, err := client.Lock(ctx, &protoReq, grpc.Header(&metadata.HeaderMD), grpc.Trailer(&metadata.TrailerMD))
+ return msg, metadata, err
+
+}
+
+func request_Lock_Unlock_0(ctx context.Context, marshaler runtime.Marshaler, client v3lockpb.LockClient, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
+ var protoReq v3lockpb.UnlockRequest
+ var metadata runtime.ServerMetadata
+
+ if err := marshaler.NewDecoder(req.Body).Decode(&protoReq); err != nil {
+ return nil, metadata, status.Errorf(codes.InvalidArgument, "%v", err)
+ }
+
+ msg, err := client.Unlock(ctx, &protoReq, grpc.Header(&metadata.HeaderMD), grpc.Trailer(&metadata.TrailerMD))
+ return msg, metadata, err
+
+}
+
+// RegisterLockHandlerFromEndpoint is same as RegisterLockHandler but
+// automatically dials to "endpoint" and closes the connection when "ctx" gets done.
+func RegisterLockHandlerFromEndpoint(ctx context.Context, mux *runtime.ServeMux, endpoint string, opts []grpc.DialOption) (err error) {
+ conn, err := grpc.Dial(endpoint, opts...)
+ if err != nil {
+ return err
+ }
+ defer func() {
+ if err != nil {
+ if cerr := conn.Close(); cerr != nil {
+ grpclog.Printf("Failed to close conn to %s: %v", endpoint, cerr)
+ }
+ return
+ }
+ go func() {
+ <-ctx.Done()
+ if cerr := conn.Close(); cerr != nil {
+ grpclog.Printf("Failed to close conn to %s: %v", endpoint, cerr)
+ }
+ }()
+ }()
+
+ return RegisterLockHandler(ctx, mux, conn)
+}
+
+// RegisterLockHandler registers the http handlers for service Lock to "mux".
+// The handlers forward requests to the grpc endpoint over "conn".
+func RegisterLockHandler(ctx context.Context, mux *runtime.ServeMux, conn *grpc.ClientConn) error {
+ return RegisterLockHandlerClient(ctx, mux, v3lockpb.NewLockClient(conn))
+}
+
+// RegisterLockHandler registers the http handlers for service Lock to "mux".
+// The handlers forward requests to the grpc endpoint over the given implementation of "LockClient".
+// Note: the gRPC framework executes interceptors within the gRPC handler. If the passed in "LockClient"
+// doesn't go through the normal gRPC flow (creating a gRPC client etc.) then it will be up to the passed in
+// "LockClient" to call the correct interceptors.
+func RegisterLockHandlerClient(ctx context.Context, mux *runtime.ServeMux, client v3lockpb.LockClient) error {
+
+ mux.Handle("POST", pattern_Lock_Lock_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+ ctx, cancel := context.WithCancel(req.Context())
+ defer cancel()
+ if cn, ok := w.(http.CloseNotifier); ok {
+ go func(done <-chan struct{}, closed <-chan bool) {
+ select {
+ case <-done:
+ case <-closed:
+ cancel()
+ }
+ }(ctx.Done(), cn.CloseNotify())
+ }
+ inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+ rctx, err := runtime.AnnotateContext(ctx, mux, req)
+ if err != nil {
+ runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+ return
+ }
+ resp, md, err := request_Lock_Lock_0(rctx, inboundMarshaler, client, req, pathParams)
+ ctx = runtime.NewServerMetadataContext(ctx, md)
+ if err != nil {
+ runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+ return
+ }
+
+ forward_Lock_Lock_0(ctx, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
+
+ })
+
+ mux.Handle("POST", pattern_Lock_Unlock_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+ ctx, cancel := context.WithCancel(req.Context())
+ defer cancel()
+ if cn, ok := w.(http.CloseNotifier); ok {
+ go func(done <-chan struct{}, closed <-chan bool) {
+ select {
+ case <-done:
+ case <-closed:
+ cancel()
+ }
+ }(ctx.Done(), cn.CloseNotify())
+ }
+ inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+ rctx, err := runtime.AnnotateContext(ctx, mux, req)
+ if err != nil {
+ runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+ return
+ }
+ resp, md, err := request_Lock_Unlock_0(rctx, inboundMarshaler, client, req, pathParams)
+ ctx = runtime.NewServerMetadataContext(ctx, md)
+ if err != nil {
+ runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+ return
+ }
+
+ forward_Lock_Unlock_0(ctx, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
+
+ })
+
+ return nil
+}
+
+var (
+ pattern_Lock_Lock_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 1}, []string{"v3beta", "lock"}, ""))
+
+ pattern_Lock_Unlock_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2}, []string{"v3beta", "lock", "unlock"}, ""))
+)
+
+var (
+ forward_Lock_Lock_0 = runtime.ForwardResponseMessage
+
+ forward_Lock_Unlock_0 = runtime.ForwardResponseMessage
+)
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v3lock/v3lockpb/v3lock.proto b/vendor/github.com/coreos/etcd/etcdserver/api/v3lock/v3lockpb/v3lock.proto
new file mode 100644
index 0000000..44b698d
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v3lock/v3lockpb/v3lock.proto
@@ -0,0 +1,65 @@
+syntax = "proto3";
+package v3lockpb;
+
+import "gogoproto/gogo.proto";
+import "etcd/etcdserver/etcdserverpb/rpc.proto";
+
+// for grpc-gateway
+import "google/api/annotations.proto";
+
+option (gogoproto.marshaler_all) = true;
+option (gogoproto.unmarshaler_all) = true;
+
+// The lock service exposes client-side locking facilities as a gRPC interface.
+service Lock {
+ // Lock acquires a distributed shared lock on a given named lock.
+ // On success, it will return a unique key that exists so long as the
+ // lock is held by the caller. This key can be used in conjunction with
+ // transactions to safely ensure updates to etcd only occur while holding
+ // lock ownership. The lock is held until Unlock is called on the key or the
+ // lease associate with the owner expires.
+ rpc Lock(LockRequest) returns (LockResponse) {
+ option (google.api.http) = {
+ post: "/v3beta/lock/lock"
+ body: "*"
+ };
+ }
+
+ // Unlock takes a key returned by Lock and releases the hold on lock. The
+ // next Lock caller waiting for the lock will then be woken up and given
+ // ownership of the lock.
+ rpc Unlock(UnlockRequest) returns (UnlockResponse) {
+ option (google.api.http) = {
+ post: "/v3beta/lock/unlock"
+ body: "*"
+ };
+ }
+}
+
+message LockRequest {
+ // name is the identifier for the distributed shared lock to be acquired.
+ bytes name = 1;
+ // lease is the ID of the lease that will be attached to ownership of the
+ // lock. If the lease expires or is revoked and currently holds the lock,
+ // the lock is automatically released. Calls to Lock with the same lease will
+ // be treated as a single acquistion; locking twice with the same lease is a
+ // no-op.
+ int64 lease = 2;
+}
+
+message LockResponse {
+ etcdserverpb.ResponseHeader header = 1;
+ // key is a key that will exist on etcd for the duration that the Lock caller
+ // owns the lock. Users should not modify this key or the lock may exhibit
+ // undefined behavior.
+ bytes key = 2;
+}
+
+message UnlockRequest {
+ // key is the lock ownership key granted by Lock.
+ bytes key = 1;
+}
+
+message UnlockResponse {
+ etcdserverpb.ResponseHeader header = 1;
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/auth.go b/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/auth.go
new file mode 100644
index 0000000..ca8e53a
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/auth.go
@@ -0,0 +1,158 @@
+// Copyright 2016 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 v3rpc
+
+import (
+ "context"
+
+ "github.com/coreos/etcd/etcdserver"
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+)
+
+type AuthServer struct {
+ authenticator etcdserver.Authenticator
+}
+
+func NewAuthServer(s *etcdserver.EtcdServer) *AuthServer {
+ return &AuthServer{authenticator: s}
+}
+
+func (as *AuthServer) AuthEnable(ctx context.Context, r *pb.AuthEnableRequest) (*pb.AuthEnableResponse, error) {
+ resp, err := as.authenticator.AuthEnable(ctx, r)
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+ return resp, nil
+}
+
+func (as *AuthServer) AuthDisable(ctx context.Context, r *pb.AuthDisableRequest) (*pb.AuthDisableResponse, error) {
+ resp, err := as.authenticator.AuthDisable(ctx, r)
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+ return resp, nil
+}
+
+func (as *AuthServer) Authenticate(ctx context.Context, r *pb.AuthenticateRequest) (*pb.AuthenticateResponse, error) {
+ resp, err := as.authenticator.Authenticate(ctx, r)
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+ return resp, nil
+}
+
+func (as *AuthServer) RoleAdd(ctx context.Context, r *pb.AuthRoleAddRequest) (*pb.AuthRoleAddResponse, error) {
+ resp, err := as.authenticator.RoleAdd(ctx, r)
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+ return resp, nil
+}
+
+func (as *AuthServer) RoleDelete(ctx context.Context, r *pb.AuthRoleDeleteRequest) (*pb.AuthRoleDeleteResponse, error) {
+ resp, err := as.authenticator.RoleDelete(ctx, r)
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+ return resp, nil
+}
+
+func (as *AuthServer) RoleGet(ctx context.Context, r *pb.AuthRoleGetRequest) (*pb.AuthRoleGetResponse, error) {
+ resp, err := as.authenticator.RoleGet(ctx, r)
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+ return resp, nil
+}
+
+func (as *AuthServer) RoleList(ctx context.Context, r *pb.AuthRoleListRequest) (*pb.AuthRoleListResponse, error) {
+ resp, err := as.authenticator.RoleList(ctx, r)
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+ return resp, nil
+}
+
+func (as *AuthServer) RoleRevokePermission(ctx context.Context, r *pb.AuthRoleRevokePermissionRequest) (*pb.AuthRoleRevokePermissionResponse, error) {
+ resp, err := as.authenticator.RoleRevokePermission(ctx, r)
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+ return resp, nil
+}
+
+func (as *AuthServer) RoleGrantPermission(ctx context.Context, r *pb.AuthRoleGrantPermissionRequest) (*pb.AuthRoleGrantPermissionResponse, error) {
+ resp, err := as.authenticator.RoleGrantPermission(ctx, r)
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+ return resp, nil
+}
+
+func (as *AuthServer) UserAdd(ctx context.Context, r *pb.AuthUserAddRequest) (*pb.AuthUserAddResponse, error) {
+ resp, err := as.authenticator.UserAdd(ctx, r)
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+ return resp, nil
+}
+
+func (as *AuthServer) UserDelete(ctx context.Context, r *pb.AuthUserDeleteRequest) (*pb.AuthUserDeleteResponse, error) {
+ resp, err := as.authenticator.UserDelete(ctx, r)
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+ return resp, nil
+}
+
+func (as *AuthServer) UserGet(ctx context.Context, r *pb.AuthUserGetRequest) (*pb.AuthUserGetResponse, error) {
+ resp, err := as.authenticator.UserGet(ctx, r)
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+ return resp, nil
+}
+
+func (as *AuthServer) UserList(ctx context.Context, r *pb.AuthUserListRequest) (*pb.AuthUserListResponse, error) {
+ resp, err := as.authenticator.UserList(ctx, r)
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+ return resp, nil
+}
+
+func (as *AuthServer) UserGrantRole(ctx context.Context, r *pb.AuthUserGrantRoleRequest) (*pb.AuthUserGrantRoleResponse, error) {
+ resp, err := as.authenticator.UserGrantRole(ctx, r)
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+ return resp, nil
+}
+
+func (as *AuthServer) UserRevokeRole(ctx context.Context, r *pb.AuthUserRevokeRoleRequest) (*pb.AuthUserRevokeRoleResponse, error) {
+ resp, err := as.authenticator.UserRevokeRole(ctx, r)
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+ return resp, nil
+}
+
+func (as *AuthServer) UserChangePassword(ctx context.Context, r *pb.AuthUserChangePasswordRequest) (*pb.AuthUserChangePasswordResponse, error) {
+ resp, err := as.authenticator.UserChangePassword(ctx, r)
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+ return resp, nil
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/codec.go b/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/codec.go
new file mode 100644
index 0000000..17a2c87
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/codec.go
@@ -0,0 +1,34 @@
+// Copyright 2016 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 v3rpc
+
+import "github.com/gogo/protobuf/proto"
+
+type codec struct{}
+
+func (c *codec) Marshal(v interface{}) ([]byte, error) {
+ b, err := proto.Marshal(v.(proto.Message))
+ sentBytes.Add(float64(len(b)))
+ return b, err
+}
+
+func (c *codec) Unmarshal(data []byte, v interface{}) error {
+ receivedBytes.Add(float64(len(data)))
+ return proto.Unmarshal(data, v.(proto.Message))
+}
+
+func (c *codec) String() string {
+ return "proto"
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/grpc.go b/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/grpc.go
new file mode 100644
index 0000000..c97e746
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/grpc.go
@@ -0,0 +1,76 @@
+// Copyright 2016 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 v3rpc
+
+import (
+ "crypto/tls"
+ "math"
+
+ "github.com/coreos/etcd/etcdserver"
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+
+ "github.com/grpc-ecosystem/go-grpc-middleware"
+ "github.com/grpc-ecosystem/go-grpc-prometheus"
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/credentials"
+ "google.golang.org/grpc/health"
+ healthpb "google.golang.org/grpc/health/grpc_health_v1"
+)
+
+const (
+ grpcOverheadBytes = 512 * 1024
+ maxStreams = math.MaxUint32
+ maxSendBytes = math.MaxInt32
+)
+
+func Server(s *etcdserver.EtcdServer, tls *tls.Config, gopts ...grpc.ServerOption) *grpc.Server {
+ var opts []grpc.ServerOption
+ opts = append(opts, grpc.CustomCodec(&codec{}))
+ if tls != nil {
+ opts = append(opts, grpc.Creds(credentials.NewTLS(tls)))
+ }
+ opts = append(opts, grpc.UnaryInterceptor(grpc_middleware.ChainUnaryServer(
+ newLogUnaryInterceptor(s),
+ newUnaryInterceptor(s),
+ grpc_prometheus.UnaryServerInterceptor,
+ )))
+ opts = append(opts, grpc.StreamInterceptor(grpc_middleware.ChainStreamServer(
+ newStreamInterceptor(s),
+ grpc_prometheus.StreamServerInterceptor,
+ )))
+ opts = append(opts, grpc.MaxRecvMsgSize(int(s.Cfg.MaxRequestBytes+grpcOverheadBytes)))
+ opts = append(opts, grpc.MaxSendMsgSize(maxSendBytes))
+ opts = append(opts, grpc.MaxConcurrentStreams(maxStreams))
+ grpcServer := grpc.NewServer(append(opts, gopts...)...)
+
+ pb.RegisterKVServer(grpcServer, NewQuotaKVServer(s))
+ pb.RegisterWatchServer(grpcServer, NewWatchServer(s))
+ pb.RegisterLeaseServer(grpcServer, NewQuotaLeaseServer(s))
+ pb.RegisterClusterServer(grpcServer, NewClusterServer(s))
+ pb.RegisterAuthServer(grpcServer, NewAuthServer(s))
+ pb.RegisterMaintenanceServer(grpcServer, NewMaintenanceServer(s))
+
+ // server should register all the services manually
+ // use empty service name for all etcd services' health status,
+ // see https://github.com/grpc/grpc/blob/master/doc/health-checking.md for more
+ hsrv := health.NewServer()
+ hsrv.SetServingStatus("", healthpb.HealthCheckResponse_SERVING)
+ healthpb.RegisterHealthServer(grpcServer, hsrv)
+
+ // set zero values for metrics registered for this grpc server
+ grpc_prometheus.Register(grpcServer)
+
+ return grpcServer
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/header.go b/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/header.go
new file mode 100644
index 0000000..75da52f
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/header.go
@@ -0,0 +1,49 @@
+// Copyright 2016 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 v3rpc
+
+import (
+ "github.com/coreos/etcd/etcdserver"
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+)
+
+type header struct {
+ clusterID int64
+ memberID int64
+ raftTimer etcdserver.RaftTimer
+ rev func() int64
+}
+
+func newHeader(s *etcdserver.EtcdServer) header {
+ return header{
+ clusterID: int64(s.Cluster().ID()),
+ memberID: int64(s.ID()),
+ raftTimer: s,
+ rev: func() int64 { return s.KV().Rev() },
+ }
+}
+
+// fill populates pb.ResponseHeader using etcdserver information
+func (h *header) fill(rh *pb.ResponseHeader) {
+ if rh == nil {
+ plog.Panic("unexpected nil resp.Header")
+ }
+ rh.ClusterId = uint64(h.clusterID)
+ rh.MemberId = uint64(h.memberID)
+ rh.RaftTerm = h.raftTimer.Term()
+ if rh.Revision == 0 {
+ rh.Revision = h.rev()
+ }
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/interceptor.go b/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/interceptor.go
new file mode 100644
index 0000000..d594ae7
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/interceptor.go
@@ -0,0 +1,263 @@
+// Copyright 2016 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 v3rpc
+
+import (
+ "context"
+ "sync"
+ "time"
+
+ "github.com/coreos/etcd/etcdserver"
+ "github.com/coreos/etcd/etcdserver/api"
+ "github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
+ "github.com/coreos/etcd/pkg/types"
+ "github.com/coreos/etcd/raft"
+
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+ "go.uber.org/zap"
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/metadata"
+ "google.golang.org/grpc/peer"
+)
+
+const (
+ maxNoLeaderCnt = 3
+)
+
+type streamsMap struct {
+ mu sync.Mutex
+ streams map[grpc.ServerStream]struct{}
+}
+
+func newUnaryInterceptor(s *etcdserver.EtcdServer) grpc.UnaryServerInterceptor {
+ return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
+ if !api.IsCapabilityEnabled(api.V3rpcCapability) {
+ return nil, rpctypes.ErrGRPCNotCapable
+ }
+
+ md, ok := metadata.FromIncomingContext(ctx)
+ if ok {
+ if ks := md[rpctypes.MetadataRequireLeaderKey]; len(ks) > 0 && ks[0] == rpctypes.MetadataHasLeader {
+ if s.Leader() == types.ID(raft.None) {
+ return nil, rpctypes.ErrGRPCNoLeader
+ }
+ }
+ }
+
+ return handler(ctx, req)
+ }
+}
+
+func newLogUnaryInterceptor(s *etcdserver.EtcdServer) grpc.UnaryServerInterceptor {
+ return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
+ startTime := time.Now()
+ resp, err := handler(ctx, req)
+ defer logUnaryRequestStats(ctx, nil, info, startTime, req, resp)
+ return resp, err
+ }
+}
+
+func logUnaryRequestStats(ctx context.Context, lg *zap.Logger, info *grpc.UnaryServerInfo, startTime time.Time, req interface{}, resp interface{}) {
+ duration := time.Since(startTime)
+ remote := "No remote client info."
+ peerInfo, ok := peer.FromContext(ctx)
+ if ok {
+ remote = peerInfo.Addr.String()
+ }
+ var responseType string = info.FullMethod
+ var reqCount, respCount int64
+ var reqSize, respSize int
+ var reqContent string
+ switch _resp := resp.(type) {
+ case *pb.RangeResponse:
+ _req, ok := req.(*pb.RangeRequest)
+ if ok {
+ reqCount = 0
+ reqSize = _req.Size()
+ reqContent = _req.String()
+ }
+ if _resp != nil {
+ respCount = _resp.GetCount()
+ respSize = _resp.Size()
+ }
+ case *pb.PutResponse:
+ _req, ok := req.(*pb.PutRequest)
+ if ok {
+ reqCount = 1
+ reqSize = _req.Size()
+ reqContent = pb.NewLoggablePutRequest(_req).String()
+ // redact value field from request content, see PR #9821
+ }
+ if _resp != nil {
+ respCount = 0
+ respSize = _resp.Size()
+ }
+ case *pb.DeleteRangeResponse:
+ _req, ok := req.(*pb.DeleteRangeRequest)
+ if ok {
+ reqCount = 0
+ reqSize = _req.Size()
+ reqContent = _req.String()
+ }
+ if _resp != nil {
+ respCount = _resp.GetDeleted()
+ respSize = _resp.Size()
+ }
+ case *pb.TxnResponse:
+ _req, ok := req.(*pb.TxnRequest)
+ if ok && _resp != nil {
+ if _resp.GetSucceeded() { // determine the 'actual' count and size of request based on success or failure
+ reqCount = int64(len(_req.GetSuccess()))
+ reqSize = 0
+ for _, r := range _req.GetSuccess() {
+ reqSize += r.Size()
+ }
+ } else {
+ reqCount = int64(len(_req.GetFailure()))
+ reqSize = 0
+ for _, r := range _req.GetFailure() {
+ reqSize += r.Size()
+ }
+ }
+ reqContent = pb.NewLoggableTxnRequest(_req).String()
+ // redact value field from request content, see PR #9821
+ }
+ if _resp != nil {
+ respCount = 0
+ respSize = _resp.Size()
+ }
+ default:
+ reqCount = -1
+ reqSize = -1
+ respCount = -1
+ respSize = -1
+ }
+
+ logGenericRequestStats(lg, startTime, duration, remote, responseType, reqCount, reqSize, respCount, respSize, reqContent)
+}
+
+func logGenericRequestStats(lg *zap.Logger, startTime time.Time, duration time.Duration, remote string, responseType string,
+ reqCount int64, reqSize int, respCount int64, respSize int, reqContent string) {
+ if lg == nil {
+ plog.Debugf("start time = %v, "+
+ "time spent = %v, "+
+ "remote = %s, "+
+ "response type = %s, "+
+ "request count = %d, "+
+ "request size = %d, "+
+ "response count = %d, "+
+ "response size = %d, "+
+ "request content = %s",
+ startTime, duration, remote, responseType, reqCount, reqSize, respCount, respSize, reqContent,
+ )
+ } else {
+ lg.Debug("request stats",
+ zap.Time("start time", startTime),
+ zap.Duration("time spent", duration),
+ zap.String("remote", remote),
+ zap.String("response type", responseType),
+ zap.Int64("request count", reqCount),
+ zap.Int("request size", reqSize),
+ zap.Int64("response count", respCount),
+ zap.Int("response size", respSize),
+ zap.String("request content", reqContent),
+ )
+ }
+}
+
+func newStreamInterceptor(s *etcdserver.EtcdServer) grpc.StreamServerInterceptor {
+ smap := monitorLeader(s)
+
+ return func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
+ if !api.IsCapabilityEnabled(api.V3rpcCapability) {
+ return rpctypes.ErrGRPCNotCapable
+ }
+
+ md, ok := metadata.FromIncomingContext(ss.Context())
+ if ok {
+ if ks := md[rpctypes.MetadataRequireLeaderKey]; len(ks) > 0 && ks[0] == rpctypes.MetadataHasLeader {
+ if s.Leader() == types.ID(raft.None) {
+ return rpctypes.ErrGRPCNoLeader
+ }
+
+ cctx, cancel := context.WithCancel(ss.Context())
+ ss = serverStreamWithCtx{ctx: cctx, cancel: &cancel, ServerStream: ss}
+
+ smap.mu.Lock()
+ smap.streams[ss] = struct{}{}
+ smap.mu.Unlock()
+
+ defer func() {
+ smap.mu.Lock()
+ delete(smap.streams, ss)
+ smap.mu.Unlock()
+ cancel()
+ }()
+
+ }
+ }
+
+ return handler(srv, ss)
+ }
+}
+
+type serverStreamWithCtx struct {
+ grpc.ServerStream
+ ctx context.Context
+ cancel *context.CancelFunc
+}
+
+func (ssc serverStreamWithCtx) Context() context.Context { return ssc.ctx }
+
+func monitorLeader(s *etcdserver.EtcdServer) *streamsMap {
+ smap := &streamsMap{
+ streams: make(map[grpc.ServerStream]struct{}),
+ }
+
+ go func() {
+ election := time.Duration(s.Cfg.TickMs) * time.Duration(s.Cfg.ElectionTicks) * time.Millisecond
+ noLeaderCnt := 0
+
+ for {
+ select {
+ case <-s.StopNotify():
+ return
+ case <-time.After(election):
+ if s.Leader() == types.ID(raft.None) {
+ noLeaderCnt++
+ } else {
+ noLeaderCnt = 0
+ }
+
+ // We are more conservative on canceling existing streams. Reconnecting streams
+ // cost much more than just rejecting new requests. So we wait until the member
+ // cannot find a leader for maxNoLeaderCnt election timeouts to cancel existing streams.
+ if noLeaderCnt >= maxNoLeaderCnt {
+ smap.mu.Lock()
+ for ss := range smap.streams {
+ if ssWithCtx, ok := ss.(serverStreamWithCtx); ok {
+ (*ssWithCtx.cancel)()
+ <-ss.Context().Done()
+ }
+ }
+ smap.streams = make(map[grpc.ServerStream]struct{})
+ smap.mu.Unlock()
+ }
+ }
+ }
+ }()
+
+ return smap
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/key.go b/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/key.go
new file mode 100644
index 0000000..9781bdd
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/key.go
@@ -0,0 +1,277 @@
+// 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 v3rpc implements etcd v3 RPC system based on gRPC.
+package v3rpc
+
+import (
+ "context"
+
+ "github.com/coreos/etcd/etcdserver"
+ "github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+ "github.com/coreos/etcd/pkg/adt"
+
+ "github.com/coreos/pkg/capnslog"
+)
+
+var (
+ plog = capnslog.NewPackageLogger("github.com/coreos/etcd", "etcdserver/api/v3rpc")
+)
+
+type kvServer struct {
+ hdr header
+ kv etcdserver.RaftKV
+ // maxTxnOps is the max operations per txn.
+ // e.g suppose maxTxnOps = 128.
+ // Txn.Success can have at most 128 operations,
+ // and Txn.Failure can have at most 128 operations.
+ maxTxnOps uint
+}
+
+func NewKVServer(s *etcdserver.EtcdServer) pb.KVServer {
+ return &kvServer{hdr: newHeader(s), kv: s, maxTxnOps: s.Cfg.MaxTxnOps}
+}
+
+func (s *kvServer) Range(ctx context.Context, r *pb.RangeRequest) (*pb.RangeResponse, error) {
+ if err := checkRangeRequest(r); err != nil {
+ return nil, err
+ }
+
+ resp, err := s.kv.Range(ctx, r)
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+
+ s.hdr.fill(resp.Header)
+ return resp, nil
+}
+
+func (s *kvServer) Put(ctx context.Context, r *pb.PutRequest) (*pb.PutResponse, error) {
+ if err := checkPutRequest(r); err != nil {
+ return nil, err
+ }
+
+ resp, err := s.kv.Put(ctx, r)
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+
+ s.hdr.fill(resp.Header)
+ return resp, nil
+}
+
+func (s *kvServer) DeleteRange(ctx context.Context, r *pb.DeleteRangeRequest) (*pb.DeleteRangeResponse, error) {
+ if err := checkDeleteRequest(r); err != nil {
+ return nil, err
+ }
+
+ resp, err := s.kv.DeleteRange(ctx, r)
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+
+ s.hdr.fill(resp.Header)
+ return resp, nil
+}
+
+func (s *kvServer) Txn(ctx context.Context, r *pb.TxnRequest) (*pb.TxnResponse, error) {
+ if err := checkTxnRequest(r, int(s.maxTxnOps)); err != nil {
+ return nil, err
+ }
+ // check for forbidden put/del overlaps after checking request to avoid quadratic blowup
+ if _, _, err := checkIntervals(r.Success); err != nil {
+ return nil, err
+ }
+ if _, _, err := checkIntervals(r.Failure); err != nil {
+ return nil, err
+ }
+
+ resp, err := s.kv.Txn(ctx, r)
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+
+ s.hdr.fill(resp.Header)
+ return resp, nil
+}
+
+func (s *kvServer) Compact(ctx context.Context, r *pb.CompactionRequest) (*pb.CompactionResponse, error) {
+ resp, err := s.kv.Compact(ctx, r)
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+
+ s.hdr.fill(resp.Header)
+ return resp, nil
+}
+
+func checkRangeRequest(r *pb.RangeRequest) error {
+ if len(r.Key) == 0 {
+ return rpctypes.ErrGRPCEmptyKey
+ }
+ return nil
+}
+
+func checkPutRequest(r *pb.PutRequest) error {
+ if len(r.Key) == 0 {
+ return rpctypes.ErrGRPCEmptyKey
+ }
+ if r.IgnoreValue && len(r.Value) != 0 {
+ return rpctypes.ErrGRPCValueProvided
+ }
+ if r.IgnoreLease && r.Lease != 0 {
+ return rpctypes.ErrGRPCLeaseProvided
+ }
+ return nil
+}
+
+func checkDeleteRequest(r *pb.DeleteRangeRequest) error {
+ if len(r.Key) == 0 {
+ return rpctypes.ErrGRPCEmptyKey
+ }
+ return nil
+}
+
+func checkTxnRequest(r *pb.TxnRequest, maxTxnOps int) error {
+ opc := len(r.Compare)
+ if opc < len(r.Success) {
+ opc = len(r.Success)
+ }
+ if opc < len(r.Failure) {
+ opc = len(r.Failure)
+ }
+ if opc > maxTxnOps {
+ return rpctypes.ErrGRPCTooManyOps
+ }
+
+ for _, c := range r.Compare {
+ if len(c.Key) == 0 {
+ return rpctypes.ErrGRPCEmptyKey
+ }
+ }
+ for _, u := range r.Success {
+ if err := checkRequestOp(u, maxTxnOps-opc); err != nil {
+ return err
+ }
+ }
+ for _, u := range r.Failure {
+ if err := checkRequestOp(u, maxTxnOps-opc); err != nil {
+ return err
+ }
+ }
+
+ return nil
+}
+
+// checkIntervals tests whether puts and deletes overlap for a list of ops. If
+// there is an overlap, returns an error. If no overlap, return put and delete
+// sets for recursive evaluation.
+func checkIntervals(reqs []*pb.RequestOp) (map[string]struct{}, adt.IntervalTree, error) {
+ var dels adt.IntervalTree
+
+ // collect deletes from this level; build first to check lower level overlapped puts
+ for _, req := range reqs {
+ tv, ok := req.Request.(*pb.RequestOp_RequestDeleteRange)
+ if !ok {
+ continue
+ }
+ dreq := tv.RequestDeleteRange
+ if dreq == nil {
+ continue
+ }
+ var iv adt.Interval
+ if len(dreq.RangeEnd) != 0 {
+ iv = adt.NewStringAffineInterval(string(dreq.Key), string(dreq.RangeEnd))
+ } else {
+ iv = adt.NewStringAffinePoint(string(dreq.Key))
+ }
+ dels.Insert(iv, struct{}{})
+ }
+
+ // collect children puts/deletes
+ puts := make(map[string]struct{})
+ for _, req := range reqs {
+ tv, ok := req.Request.(*pb.RequestOp_RequestTxn)
+ if !ok {
+ continue
+ }
+ putsThen, delsThen, err := checkIntervals(tv.RequestTxn.Success)
+ if err != nil {
+ return nil, dels, err
+ }
+ putsElse, delsElse, err := checkIntervals(tv.RequestTxn.Failure)
+ if err != nil {
+ return nil, dels, err
+ }
+ for k := range putsThen {
+ if _, ok := puts[k]; ok {
+ return nil, dels, rpctypes.ErrGRPCDuplicateKey
+ }
+ if dels.Intersects(adt.NewStringAffinePoint(k)) {
+ return nil, dels, rpctypes.ErrGRPCDuplicateKey
+ }
+ puts[k] = struct{}{}
+ }
+ for k := range putsElse {
+ if _, ok := puts[k]; ok {
+ // if key is from putsThen, overlap is OK since
+ // either then/else are mutually exclusive
+ if _, isSafe := putsThen[k]; !isSafe {
+ return nil, dels, rpctypes.ErrGRPCDuplicateKey
+ }
+ }
+ if dels.Intersects(adt.NewStringAffinePoint(k)) {
+ return nil, dels, rpctypes.ErrGRPCDuplicateKey
+ }
+ puts[k] = struct{}{}
+ }
+ dels.Union(delsThen, adt.NewStringAffineInterval("\x00", ""))
+ dels.Union(delsElse, adt.NewStringAffineInterval("\x00", ""))
+ }
+
+ // collect and check this level's puts
+ for _, req := range reqs {
+ tv, ok := req.Request.(*pb.RequestOp_RequestPut)
+ if !ok || tv.RequestPut == nil {
+ continue
+ }
+ k := string(tv.RequestPut.Key)
+ if _, ok := puts[k]; ok {
+ return nil, dels, rpctypes.ErrGRPCDuplicateKey
+ }
+ if dels.Intersects(adt.NewStringAffinePoint(k)) {
+ return nil, dels, rpctypes.ErrGRPCDuplicateKey
+ }
+ puts[k] = struct{}{}
+ }
+ return puts, dels, nil
+}
+
+func checkRequestOp(u *pb.RequestOp, maxTxnOps int) error {
+ // TODO: ensure only one of the field is set.
+ switch uv := u.Request.(type) {
+ case *pb.RequestOp_RequestRange:
+ return checkRangeRequest(uv.RequestRange)
+ case *pb.RequestOp_RequestPut:
+ return checkPutRequest(uv.RequestPut)
+ case *pb.RequestOp_RequestDeleteRange:
+ return checkDeleteRequest(uv.RequestDeleteRange)
+ case *pb.RequestOp_RequestTxn:
+ return checkTxnRequest(uv.RequestTxn, maxTxnOps)
+ default:
+ // empty op / nil entry
+ return rpctypes.ErrGRPCKeyNotFound
+ }
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/lease.go b/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/lease.go
new file mode 100644
index 0000000..5b4f2b1
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/lease.go
@@ -0,0 +1,148 @@
+// Copyright 2016 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 v3rpc
+
+import (
+ "context"
+ "io"
+
+ "github.com/coreos/etcd/etcdserver"
+ "github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+ "github.com/coreos/etcd/lease"
+)
+
+type LeaseServer struct {
+ hdr header
+ le etcdserver.Lessor
+}
+
+func NewLeaseServer(s *etcdserver.EtcdServer) pb.LeaseServer {
+ return &LeaseServer{le: s, hdr: newHeader(s)}
+}
+
+func (ls *LeaseServer) LeaseGrant(ctx context.Context, cr *pb.LeaseGrantRequest) (*pb.LeaseGrantResponse, error) {
+ resp, err := ls.le.LeaseGrant(ctx, cr)
+
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+ ls.hdr.fill(resp.Header)
+ return resp, nil
+}
+
+func (ls *LeaseServer) LeaseRevoke(ctx context.Context, rr *pb.LeaseRevokeRequest) (*pb.LeaseRevokeResponse, error) {
+ resp, err := ls.le.LeaseRevoke(ctx, rr)
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+ ls.hdr.fill(resp.Header)
+ return resp, nil
+}
+
+func (ls *LeaseServer) LeaseTimeToLive(ctx context.Context, rr *pb.LeaseTimeToLiveRequest) (*pb.LeaseTimeToLiveResponse, error) {
+ resp, err := ls.le.LeaseTimeToLive(ctx, rr)
+ if err != nil && err != lease.ErrLeaseNotFound {
+ return nil, togRPCError(err)
+ }
+ if err == lease.ErrLeaseNotFound {
+ resp = &pb.LeaseTimeToLiveResponse{
+ Header: &pb.ResponseHeader{},
+ ID: rr.ID,
+ TTL: -1,
+ }
+ }
+ ls.hdr.fill(resp.Header)
+ return resp, nil
+}
+
+func (ls *LeaseServer) LeaseLeases(ctx context.Context, rr *pb.LeaseLeasesRequest) (*pb.LeaseLeasesResponse, error) {
+ resp, err := ls.le.LeaseLeases(ctx, rr)
+ if err != nil && err != lease.ErrLeaseNotFound {
+ return nil, togRPCError(err)
+ }
+ if err == lease.ErrLeaseNotFound {
+ resp = &pb.LeaseLeasesResponse{
+ Header: &pb.ResponseHeader{},
+ Leases: []*pb.LeaseStatus{},
+ }
+ }
+ ls.hdr.fill(resp.Header)
+ return resp, nil
+}
+
+func (ls *LeaseServer) LeaseKeepAlive(stream pb.Lease_LeaseKeepAliveServer) (err error) {
+ errc := make(chan error, 1)
+ go func() {
+ errc <- ls.leaseKeepAlive(stream)
+ }()
+ select {
+ case err = <-errc:
+ case <-stream.Context().Done():
+ // the only server-side cancellation is noleader for now.
+ err = stream.Context().Err()
+ if err == context.Canceled {
+ err = rpctypes.ErrGRPCNoLeader
+ }
+ }
+ return err
+}
+
+func (ls *LeaseServer) leaseKeepAlive(stream pb.Lease_LeaseKeepAliveServer) error {
+ for {
+ req, err := stream.Recv()
+ if err == io.EOF {
+ return nil
+ }
+ if err != nil {
+ if isClientCtxErr(stream.Context().Err(), err) {
+ plog.Debugf("failed to receive lease keepalive request from gRPC stream (%q)", err.Error())
+ } else {
+ plog.Warningf("failed to receive lease keepalive request from gRPC stream (%q)", err.Error())
+ }
+ return err
+ }
+
+ // Create header before we sent out the renew request.
+ // This can make sure that the revision is strictly smaller or equal to
+ // when the keepalive happened at the local server (when the local server is the leader)
+ // or remote leader.
+ // Without this, a lease might be revoked at rev 3 but client can see the keepalive succeeded
+ // at rev 4.
+ resp := &pb.LeaseKeepAliveResponse{ID: req.ID, Header: &pb.ResponseHeader{}}
+ ls.hdr.fill(resp.Header)
+
+ ttl, err := ls.le.LeaseRenew(stream.Context(), lease.LeaseID(req.ID))
+ if err == lease.ErrLeaseNotFound {
+ err = nil
+ ttl = 0
+ }
+
+ if err != nil {
+ return togRPCError(err)
+ }
+
+ resp.TTL = ttl
+ err = stream.Send(resp)
+ if err != nil {
+ if isClientCtxErr(stream.Context().Err(), err) {
+ plog.Debugf("failed to send lease keepalive response to gRPC stream (%q)", err.Error())
+ } else {
+ plog.Warningf("failed to send lease keepalive response to gRPC stream (%q)", err.Error())
+ }
+ return err
+ }
+ }
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/maintenance.go b/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/maintenance.go
new file mode 100644
index 0000000..c9df180
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/maintenance.go
@@ -0,0 +1,229 @@
+// Copyright 2016 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 v3rpc
+
+import (
+ "context"
+ "crypto/sha256"
+ "io"
+
+ "github.com/coreos/etcd/auth"
+ "github.com/coreos/etcd/etcdserver"
+ "github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+ "github.com/coreos/etcd/mvcc"
+ "github.com/coreos/etcd/mvcc/backend"
+ "github.com/coreos/etcd/pkg/types"
+ "github.com/coreos/etcd/version"
+)
+
+type KVGetter interface {
+ KV() mvcc.ConsistentWatchableKV
+}
+
+type BackendGetter interface {
+ Backend() backend.Backend
+}
+
+type Alarmer interface {
+ Alarm(ctx context.Context, ar *pb.AlarmRequest) (*pb.AlarmResponse, error)
+}
+
+type LeaderTransferrer interface {
+ MoveLeader(ctx context.Context, lead, target uint64) error
+}
+
+type RaftStatusGetter interface {
+ etcdserver.RaftTimer
+ ID() types.ID
+ Leader() types.ID
+}
+
+type AuthGetter interface {
+ AuthInfoFromCtx(ctx context.Context) (*auth.AuthInfo, error)
+ AuthStore() auth.AuthStore
+}
+
+type maintenanceServer struct {
+ rg RaftStatusGetter
+ kg KVGetter
+ bg BackendGetter
+ a Alarmer
+ lt LeaderTransferrer
+ hdr header
+}
+
+func NewMaintenanceServer(s *etcdserver.EtcdServer) pb.MaintenanceServer {
+ srv := &maintenanceServer{rg: s, kg: s, bg: s, a: s, lt: s, hdr: newHeader(s)}
+ return &authMaintenanceServer{srv, s}
+}
+
+func (ms *maintenanceServer) Defragment(ctx context.Context, sr *pb.DefragmentRequest) (*pb.DefragmentResponse, error) {
+ plog.Noticef("starting to defragment the storage backend...")
+ err := ms.bg.Backend().Defrag()
+ if err != nil {
+ plog.Errorf("failed to defragment the storage backend (%v)", err)
+ return nil, err
+ }
+ plog.Noticef("finished defragmenting the storage backend")
+ return &pb.DefragmentResponse{}, nil
+}
+
+func (ms *maintenanceServer) Snapshot(sr *pb.SnapshotRequest, srv pb.Maintenance_SnapshotServer) error {
+ snap := ms.bg.Backend().Snapshot()
+ pr, pw := io.Pipe()
+
+ defer pr.Close()
+
+ go func() {
+ snap.WriteTo(pw)
+ if err := snap.Close(); err != nil {
+ plog.Errorf("error closing snapshot (%v)", err)
+ }
+ pw.Close()
+ }()
+
+ // send file data
+ h := sha256.New()
+ br := int64(0)
+ buf := make([]byte, 32*1024)
+ sz := snap.Size()
+ for br < sz {
+ n, err := io.ReadFull(pr, buf)
+ if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
+ return togRPCError(err)
+ }
+ br += int64(n)
+ resp := &pb.SnapshotResponse{
+ RemainingBytes: uint64(sz - br),
+ Blob: buf[:n],
+ }
+ if err = srv.Send(resp); err != nil {
+ return togRPCError(err)
+ }
+ h.Write(buf[:n])
+ }
+
+ // send sha
+ sha := h.Sum(nil)
+ hresp := &pb.SnapshotResponse{RemainingBytes: 0, Blob: sha}
+ if err := srv.Send(hresp); err != nil {
+ return togRPCError(err)
+ }
+
+ return nil
+}
+
+func (ms *maintenanceServer) Hash(ctx context.Context, r *pb.HashRequest) (*pb.HashResponse, error) {
+ h, rev, err := ms.kg.KV().Hash()
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+ resp := &pb.HashResponse{Header: &pb.ResponseHeader{Revision: rev}, Hash: h}
+ ms.hdr.fill(resp.Header)
+ return resp, nil
+}
+
+func (ms *maintenanceServer) HashKV(ctx context.Context, r *pb.HashKVRequest) (*pb.HashKVResponse, error) {
+ h, rev, compactRev, err := ms.kg.KV().HashByRev(r.Revision)
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+
+ resp := &pb.HashKVResponse{Header: &pb.ResponseHeader{Revision: rev}, Hash: h, CompactRevision: compactRev}
+ ms.hdr.fill(resp.Header)
+ return resp, nil
+}
+
+func (ms *maintenanceServer) Alarm(ctx context.Context, ar *pb.AlarmRequest) (*pb.AlarmResponse, error) {
+ return ms.a.Alarm(ctx, ar)
+}
+
+func (ms *maintenanceServer) Status(ctx context.Context, ar *pb.StatusRequest) (*pb.StatusResponse, error) {
+ resp := &pb.StatusResponse{
+ Header: &pb.ResponseHeader{Revision: ms.hdr.rev()},
+ Version: version.Version,
+ DbSize: ms.bg.Backend().Size(),
+ Leader: uint64(ms.rg.Leader()),
+ RaftIndex: ms.rg.Index(),
+ RaftTerm: ms.rg.Term(),
+ }
+ ms.hdr.fill(resp.Header)
+ return resp, nil
+}
+
+func (ms *maintenanceServer) MoveLeader(ctx context.Context, tr *pb.MoveLeaderRequest) (*pb.MoveLeaderResponse, error) {
+ if ms.rg.ID() != ms.rg.Leader() {
+ return nil, rpctypes.ErrGRPCNotLeader
+ }
+
+ if err := ms.lt.MoveLeader(ctx, uint64(ms.rg.Leader()), tr.TargetID); err != nil {
+ return nil, togRPCError(err)
+ }
+ return &pb.MoveLeaderResponse{}, nil
+}
+
+type authMaintenanceServer struct {
+ *maintenanceServer
+ ag AuthGetter
+}
+
+func (ams *authMaintenanceServer) isAuthenticated(ctx context.Context) error {
+ authInfo, err := ams.ag.AuthInfoFromCtx(ctx)
+ if err != nil {
+ return err
+ }
+
+ return ams.ag.AuthStore().IsAdminPermitted(authInfo)
+}
+
+func (ams *authMaintenanceServer) Defragment(ctx context.Context, sr *pb.DefragmentRequest) (*pb.DefragmentResponse, error) {
+ if err := ams.isAuthenticated(ctx); err != nil {
+ return nil, err
+ }
+
+ return ams.maintenanceServer.Defragment(ctx, sr)
+}
+
+func (ams *authMaintenanceServer) Snapshot(sr *pb.SnapshotRequest, srv pb.Maintenance_SnapshotServer) error {
+ if err := ams.isAuthenticated(srv.Context()); err != nil {
+ return err
+ }
+
+ return ams.maintenanceServer.Snapshot(sr, srv)
+}
+
+func (ams *authMaintenanceServer) Hash(ctx context.Context, r *pb.HashRequest) (*pb.HashResponse, error) {
+ if err := ams.isAuthenticated(ctx); err != nil {
+ return nil, err
+ }
+
+ return ams.maintenanceServer.Hash(ctx, r)
+}
+
+func (ams *authMaintenanceServer) HashKV(ctx context.Context, r *pb.HashKVRequest) (*pb.HashKVResponse, error) {
+ if err := ams.isAuthenticated(ctx); err != nil {
+ return nil, err
+ }
+ return ams.maintenanceServer.HashKV(ctx, r)
+}
+
+func (ams *authMaintenanceServer) Status(ctx context.Context, ar *pb.StatusRequest) (*pb.StatusResponse, error) {
+ return ams.maintenanceServer.Status(ctx, ar)
+}
+
+func (ams *authMaintenanceServer) MoveLeader(ctx context.Context, tr *pb.MoveLeaderRequest) (*pb.MoveLeaderResponse, error) {
+ return ams.maintenanceServer.MoveLeader(ctx, tr)
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/member.go b/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/member.go
new file mode 100644
index 0000000..cbe7b47
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/member.go
@@ -0,0 +1,101 @@
+// Copyright 2016 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 v3rpc
+
+import (
+ "context"
+ "time"
+
+ "github.com/coreos/etcd/etcdserver"
+ "github.com/coreos/etcd/etcdserver/api"
+ "github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+ "github.com/coreos/etcd/etcdserver/membership"
+ "github.com/coreos/etcd/pkg/types"
+)
+
+type ClusterServer struct {
+ cluster api.Cluster
+ server etcdserver.ServerV3
+}
+
+func NewClusterServer(s etcdserver.ServerV3) *ClusterServer {
+ return &ClusterServer{
+ cluster: s.Cluster(),
+ server: s,
+ }
+}
+
+func (cs *ClusterServer) MemberAdd(ctx context.Context, r *pb.MemberAddRequest) (*pb.MemberAddResponse, error) {
+ urls, err := types.NewURLs(r.PeerURLs)
+ if err != nil {
+ return nil, rpctypes.ErrGRPCMemberBadURLs
+ }
+
+ now := time.Now()
+ m := membership.NewMember("", urls, "", &now)
+ membs, merr := cs.server.AddMember(ctx, *m)
+ if merr != nil {
+ return nil, togRPCError(merr)
+ }
+
+ return &pb.MemberAddResponse{
+ Header: cs.header(),
+ Member: &pb.Member{ID: uint64(m.ID), PeerURLs: m.PeerURLs},
+ Members: membersToProtoMembers(membs),
+ }, nil
+}
+
+func (cs *ClusterServer) MemberRemove(ctx context.Context, r *pb.MemberRemoveRequest) (*pb.MemberRemoveResponse, error) {
+ membs, err := cs.server.RemoveMember(ctx, r.ID)
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+ return &pb.MemberRemoveResponse{Header: cs.header(), Members: membersToProtoMembers(membs)}, nil
+}
+
+func (cs *ClusterServer) MemberUpdate(ctx context.Context, r *pb.MemberUpdateRequest) (*pb.MemberUpdateResponse, error) {
+ m := membership.Member{
+ ID: types.ID(r.ID),
+ RaftAttributes: membership.RaftAttributes{PeerURLs: r.PeerURLs},
+ }
+ membs, err := cs.server.UpdateMember(ctx, m)
+ if err != nil {
+ return nil, togRPCError(err)
+ }
+ return &pb.MemberUpdateResponse{Header: cs.header(), Members: membersToProtoMembers(membs)}, nil
+}
+
+func (cs *ClusterServer) MemberList(ctx context.Context, r *pb.MemberListRequest) (*pb.MemberListResponse, error) {
+ membs := membersToProtoMembers(cs.cluster.Members())
+ return &pb.MemberListResponse{Header: cs.header(), Members: membs}, nil
+}
+
+func (cs *ClusterServer) header() *pb.ResponseHeader {
+ return &pb.ResponseHeader{ClusterId: uint64(cs.cluster.ID()), MemberId: uint64(cs.server.ID()), RaftTerm: cs.server.Term()}
+}
+
+func membersToProtoMembers(membs []*membership.Member) []*pb.Member {
+ protoMembs := make([]*pb.Member, len(membs))
+ for i := range membs {
+ protoMembs[i] = &pb.Member{
+ Name: membs[i].Name,
+ ID: uint64(membs[i].ID),
+ PeerURLs: membs[i].PeerURLs,
+ ClientURLs: membs[i].ClientURLs,
+ }
+ }
+ return protoMembs
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/metrics.go b/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/metrics.go
new file mode 100644
index 0000000..6cb41a6
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/metrics.go
@@ -0,0 +1,38 @@
+// Copyright 2016 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 v3rpc
+
+import "github.com/prometheus/client_golang/prometheus"
+
+var (
+ sentBytes = prometheus.NewCounter(prometheus.CounterOpts{
+ Namespace: "etcd",
+ Subsystem: "network",
+ Name: "client_grpc_sent_bytes_total",
+ Help: "The total number of bytes sent to grpc clients.",
+ })
+
+ receivedBytes = prometheus.NewCounter(prometheus.CounterOpts{
+ Namespace: "etcd",
+ Subsystem: "network",
+ Name: "client_grpc_received_bytes_total",
+ Help: "The total number of bytes received from grpc clients.",
+ })
+)
+
+func init() {
+ prometheus.MustRegister(sentBytes)
+ prometheus.MustRegister(receivedBytes)
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/quota.go b/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/quota.go
new file mode 100644
index 0000000..02d9960
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/quota.go
@@ -0,0 +1,90 @@
+// Copyright 2016 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 v3rpc
+
+import (
+ "context"
+
+ "github.com/coreos/etcd/etcdserver"
+ "github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+ "github.com/coreos/etcd/pkg/types"
+)
+
+type quotaKVServer struct {
+ pb.KVServer
+ qa quotaAlarmer
+}
+
+type quotaAlarmer struct {
+ q etcdserver.Quota
+ a Alarmer
+ id types.ID
+}
+
+// check whether request satisfies the quota. If there is not enough space,
+// ignore request and raise the free space alarm.
+func (qa *quotaAlarmer) check(ctx context.Context, r interface{}) error {
+ if qa.q.Available(r) {
+ return nil
+ }
+ req := &pb.AlarmRequest{
+ MemberID: uint64(qa.id),
+ Action: pb.AlarmRequest_ACTIVATE,
+ Alarm: pb.AlarmType_NOSPACE,
+ }
+ qa.a.Alarm(ctx, req)
+ return rpctypes.ErrGRPCNoSpace
+}
+
+func NewQuotaKVServer(s *etcdserver.EtcdServer) pb.KVServer {
+ return "aKVServer{
+ NewKVServer(s),
+ quotaAlarmer{etcdserver.NewBackendQuota(s), s, s.ID()},
+ }
+}
+
+func (s *quotaKVServer) Put(ctx context.Context, r *pb.PutRequest) (*pb.PutResponse, error) {
+ if err := s.qa.check(ctx, r); err != nil {
+ return nil, err
+ }
+ return s.KVServer.Put(ctx, r)
+}
+
+func (s *quotaKVServer) Txn(ctx context.Context, r *pb.TxnRequest) (*pb.TxnResponse, error) {
+ if err := s.qa.check(ctx, r); err != nil {
+ return nil, err
+ }
+ return s.KVServer.Txn(ctx, r)
+}
+
+type quotaLeaseServer struct {
+ pb.LeaseServer
+ qa quotaAlarmer
+}
+
+func (s *quotaLeaseServer) LeaseGrant(ctx context.Context, cr *pb.LeaseGrantRequest) (*pb.LeaseGrantResponse, error) {
+ if err := s.qa.check(ctx, cr); err != nil {
+ return nil, err
+ }
+ return s.LeaseServer.LeaseGrant(ctx, cr)
+}
+
+func NewQuotaLeaseServer(s *etcdserver.EtcdServer) pb.LeaseServer {
+ return "aLeaseServer{
+ NewLeaseServer(s),
+ quotaAlarmer{etcdserver.NewBackendQuota(s), s, s.ID()},
+ }
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/util.go b/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/util.go
new file mode 100644
index 0000000..799c119
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/util.go
@@ -0,0 +1,117 @@
+// Copyright 2016 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 v3rpc
+
+import (
+ "context"
+ "strings"
+
+ "github.com/coreos/etcd/auth"
+ "github.com/coreos/etcd/etcdserver"
+ "github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
+ "github.com/coreos/etcd/etcdserver/membership"
+ "github.com/coreos/etcd/lease"
+ "github.com/coreos/etcd/mvcc"
+
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/status"
+)
+
+var toGRPCErrorMap = map[error]error{
+ membership.ErrIDRemoved: rpctypes.ErrGRPCMemberNotFound,
+ membership.ErrIDNotFound: rpctypes.ErrGRPCMemberNotFound,
+ membership.ErrIDExists: rpctypes.ErrGRPCMemberExist,
+ membership.ErrPeerURLexists: rpctypes.ErrGRPCPeerURLExist,
+ etcdserver.ErrNotEnoughStartedMembers: rpctypes.ErrMemberNotEnoughStarted,
+
+ mvcc.ErrCompacted: rpctypes.ErrGRPCCompacted,
+ mvcc.ErrFutureRev: rpctypes.ErrGRPCFutureRev,
+ etcdserver.ErrRequestTooLarge: rpctypes.ErrGRPCRequestTooLarge,
+ etcdserver.ErrNoSpace: rpctypes.ErrGRPCNoSpace,
+ etcdserver.ErrTooManyRequests: rpctypes.ErrTooManyRequests,
+
+ etcdserver.ErrNoLeader: rpctypes.ErrGRPCNoLeader,
+ etcdserver.ErrNotLeader: rpctypes.ErrGRPCNotLeader,
+ etcdserver.ErrStopped: rpctypes.ErrGRPCStopped,
+ etcdserver.ErrTimeout: rpctypes.ErrGRPCTimeout,
+ etcdserver.ErrTimeoutDueToLeaderFail: rpctypes.ErrGRPCTimeoutDueToLeaderFail,
+ etcdserver.ErrTimeoutDueToConnectionLost: rpctypes.ErrGRPCTimeoutDueToConnectionLost,
+ etcdserver.ErrUnhealthy: rpctypes.ErrGRPCUnhealthy,
+ etcdserver.ErrKeyNotFound: rpctypes.ErrGRPCKeyNotFound,
+ etcdserver.ErrCorrupt: rpctypes.ErrGRPCCorrupt,
+
+ lease.ErrLeaseNotFound: rpctypes.ErrGRPCLeaseNotFound,
+ lease.ErrLeaseExists: rpctypes.ErrGRPCLeaseExist,
+ lease.ErrLeaseTTLTooLarge: rpctypes.ErrGRPCLeaseTTLTooLarge,
+
+ auth.ErrRootUserNotExist: rpctypes.ErrGRPCRootUserNotExist,
+ auth.ErrRootRoleNotExist: rpctypes.ErrGRPCRootRoleNotExist,
+ auth.ErrUserAlreadyExist: rpctypes.ErrGRPCUserAlreadyExist,
+ auth.ErrUserEmpty: rpctypes.ErrGRPCUserEmpty,
+ auth.ErrUserNotFound: rpctypes.ErrGRPCUserNotFound,
+ auth.ErrRoleAlreadyExist: rpctypes.ErrGRPCRoleAlreadyExist,
+ auth.ErrRoleNotFound: rpctypes.ErrGRPCRoleNotFound,
+ auth.ErrAuthFailed: rpctypes.ErrGRPCAuthFailed,
+ auth.ErrPermissionDenied: rpctypes.ErrGRPCPermissionDenied,
+ auth.ErrRoleNotGranted: rpctypes.ErrGRPCRoleNotGranted,
+ auth.ErrPermissionNotGranted: rpctypes.ErrGRPCPermissionNotGranted,
+ auth.ErrAuthNotEnabled: rpctypes.ErrGRPCAuthNotEnabled,
+ auth.ErrInvalidAuthToken: rpctypes.ErrGRPCInvalidAuthToken,
+ auth.ErrInvalidAuthMgmt: rpctypes.ErrGRPCInvalidAuthMgmt,
+}
+
+func togRPCError(err error) error {
+ // let gRPC server convert to codes.Canceled, codes.DeadlineExceeded
+ if err == context.Canceled || err == context.DeadlineExceeded {
+ return err
+ }
+ grpcErr, ok := toGRPCErrorMap[err]
+ if !ok {
+ return status.Error(codes.Unknown, err.Error())
+ }
+ return grpcErr
+}
+
+func isClientCtxErr(ctxErr error, err error) bool {
+ if ctxErr != nil {
+ return true
+ }
+
+ ev, ok := status.FromError(err)
+ if !ok {
+ return false
+ }
+
+ switch ev.Code() {
+ case codes.Canceled, codes.DeadlineExceeded:
+ // client-side context cancel or deadline exceeded
+ // "rpc error: code = Canceled desc = context canceled"
+ // "rpc error: code = DeadlineExceeded desc = context deadline exceeded"
+ return true
+ case codes.Unavailable:
+ msg := ev.Message()
+ // client-side context cancel or deadline exceeded with TLS ("http2.errClientDisconnected")
+ // "rpc error: code = Unavailable desc = client disconnected"
+ if msg == "client disconnected" {
+ return true
+ }
+ // "grpc/transport.ClientTransport.CloseStream" on canceled streams
+ // "rpc error: code = Unavailable desc = stream error: stream ID 21; CANCEL")
+ if strings.HasPrefix(msg, "stream error: ") && strings.HasSuffix(msg, "; CANCEL") {
+ return true
+ }
+ }
+ return false
+}
diff --git a/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/watch.go b/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/watch.go
new file mode 100644
index 0000000..dd4f329
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/etcdserver/api/v3rpc/watch.go
@@ -0,0 +1,447 @@
+// 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 v3rpc
+
+import (
+ "context"
+ "io"
+ "sync"
+ "time"
+
+ "github.com/coreos/etcd/auth"
+ "github.com/coreos/etcd/etcdserver"
+ "github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+ "github.com/coreos/etcd/mvcc"
+ "github.com/coreos/etcd/mvcc/mvccpb"
+)
+
+type watchServer struct {
+ clusterID int64
+ memberID int64
+ raftTimer etcdserver.RaftTimer
+ watchable mvcc.WatchableKV
+
+ ag AuthGetter
+}
+
+func NewWatchServer(s *etcdserver.EtcdServer) pb.WatchServer {
+ return &watchServer{
+ clusterID: int64(s.Cluster().ID()),
+ memberID: int64(s.ID()),
+ raftTimer: s,
+ watchable: s.Watchable(),
+ ag: s,
+ }
+}
+
+var (
+ // External test can read this with GetProgressReportInterval()
+ // and change this to a small value to finish fast with
+ // SetProgressReportInterval().
+ progressReportInterval = 10 * time.Minute
+ progressReportIntervalMu sync.RWMutex
+)
+
+func GetProgressReportInterval() time.Duration {
+ progressReportIntervalMu.RLock()
+ defer progressReportIntervalMu.RUnlock()
+ return progressReportInterval
+}
+
+func SetProgressReportInterval(newTimeout time.Duration) {
+ progressReportIntervalMu.Lock()
+ defer progressReportIntervalMu.Unlock()
+ progressReportInterval = newTimeout
+}
+
+const (
+ // We send ctrl response inside the read loop. We do not want
+ // send to block read, but we still want ctrl response we sent to
+ // be serialized. Thus we use a buffered chan to solve the problem.
+ // A small buffer should be OK for most cases, since we expect the
+ // ctrl requests are infrequent.
+ ctrlStreamBufLen = 16
+)
+
+// serverWatchStream is an etcd server side stream. It receives requests
+// from client side gRPC stream. It receives watch events from mvcc.WatchStream,
+// and creates responses that forwarded to gRPC stream.
+// It also forwards control message like watch created and canceled.
+type serverWatchStream struct {
+ clusterID int64
+ memberID int64
+ raftTimer etcdserver.RaftTimer
+
+ watchable mvcc.WatchableKV
+
+ gRPCStream pb.Watch_WatchServer
+ watchStream mvcc.WatchStream
+ ctrlStream chan *pb.WatchResponse
+
+ // mu protects progress, prevKV
+ mu sync.Mutex
+ // progress tracks the watchID that stream might need to send
+ // progress to.
+ // TODO: combine progress and prevKV into a single struct?
+ progress map[mvcc.WatchID]bool
+ prevKV map[mvcc.WatchID]bool
+
+ // closec indicates the stream is closed.
+ closec chan struct{}
+
+ // wg waits for the send loop to complete
+ wg sync.WaitGroup
+
+ ag AuthGetter
+}
+
+func (ws *watchServer) Watch(stream pb.Watch_WatchServer) (err error) {
+ sws := serverWatchStream{
+ clusterID: ws.clusterID,
+ memberID: ws.memberID,
+ raftTimer: ws.raftTimer,
+
+ watchable: ws.watchable,
+
+ gRPCStream: stream,
+ watchStream: ws.watchable.NewWatchStream(),
+ // chan for sending control response like watcher created and canceled.
+ ctrlStream: make(chan *pb.WatchResponse, ctrlStreamBufLen),
+ progress: make(map[mvcc.WatchID]bool),
+ prevKV: make(map[mvcc.WatchID]bool),
+ closec: make(chan struct{}),
+
+ ag: ws.ag,
+ }
+
+ sws.wg.Add(1)
+ go func() {
+ sws.sendLoop()
+ sws.wg.Done()
+ }()
+
+ errc := make(chan error, 1)
+ // Ideally recvLoop would also use sws.wg to signal its completion
+ // but when stream.Context().Done() is closed, the stream's recv
+ // may continue to block since it uses a different context, leading to
+ // deadlock when calling sws.close().
+ go func() {
+ if rerr := sws.recvLoop(); rerr != nil {
+ if isClientCtxErr(stream.Context().Err(), rerr) {
+ plog.Debugf("failed to receive watch request from gRPC stream (%q)", rerr.Error())
+ } else {
+ plog.Warningf("failed to receive watch request from gRPC stream (%q)", rerr.Error())
+ }
+ errc <- rerr
+ }
+ }()
+ select {
+ case err = <-errc:
+ close(sws.ctrlStream)
+ case <-stream.Context().Done():
+ err = stream.Context().Err()
+ // the only server-side cancellation is noleader for now.
+ if err == context.Canceled {
+ err = rpctypes.ErrGRPCNoLeader
+ }
+ }
+ sws.close()
+ return err
+}
+
+func (sws *serverWatchStream) isWatchPermitted(wcr *pb.WatchCreateRequest) bool {
+ authInfo, err := sws.ag.AuthInfoFromCtx(sws.gRPCStream.Context())
+ if err != nil {
+ return false
+ }
+ if authInfo == nil {
+ // if auth is enabled, IsRangePermitted() can cause an error
+ authInfo = &auth.AuthInfo{}
+ }
+
+ return sws.ag.AuthStore().IsRangePermitted(authInfo, wcr.Key, wcr.RangeEnd) == nil
+}
+
+func (sws *serverWatchStream) recvLoop() error {
+ for {
+ req, err := sws.gRPCStream.Recv()
+ if err == io.EOF {
+ return nil
+ }
+ if err != nil {
+ return err
+ }
+
+ switch uv := req.RequestUnion.(type) {
+ case *pb.WatchRequest_CreateRequest:
+ if uv.CreateRequest == nil {
+ break
+ }
+
+ creq := uv.CreateRequest
+ if len(creq.Key) == 0 {
+ // \x00 is the smallest key
+ creq.Key = []byte{0}
+ }
+ if len(creq.RangeEnd) == 0 {
+ // force nil since watchstream.Watch distinguishes
+ // between nil and []byte{} for single key / >=
+ creq.RangeEnd = nil
+ }
+ if len(creq.RangeEnd) == 1 && creq.RangeEnd[0] == 0 {
+ // support >= key queries
+ creq.RangeEnd = []byte{}
+ }
+
+ if !sws.isWatchPermitted(creq) {
+ wr := &pb.WatchResponse{
+ Header: sws.newResponseHeader(sws.watchStream.Rev()),
+ WatchId: -1,
+ Canceled: true,
+ Created: true,
+ CancelReason: rpctypes.ErrGRPCPermissionDenied.Error(),
+ }
+
+ select {
+ case sws.ctrlStream <- wr:
+ case <-sws.closec:
+ }
+ return nil
+ }
+
+ filters := FiltersFromRequest(creq)
+
+ wsrev := sws.watchStream.Rev()
+ rev := creq.StartRevision
+ if rev == 0 {
+ rev = wsrev + 1
+ }
+ id := sws.watchStream.Watch(creq.Key, creq.RangeEnd, rev, filters...)
+ if id != -1 {
+ sws.mu.Lock()
+ if creq.ProgressNotify {
+ sws.progress[id] = true
+ }
+ if creq.PrevKv {
+ sws.prevKV[id] = true
+ }
+ sws.mu.Unlock()
+ }
+ wr := &pb.WatchResponse{
+ Header: sws.newResponseHeader(wsrev),
+ WatchId: int64(id),
+ Created: true,
+ Canceled: id == -1,
+ }
+ select {
+ case sws.ctrlStream <- wr:
+ case <-sws.closec:
+ return nil
+ }
+ case *pb.WatchRequest_CancelRequest:
+ if uv.CancelRequest != nil {
+ id := uv.CancelRequest.WatchId
+ err := sws.watchStream.Cancel(mvcc.WatchID(id))
+ if err == nil {
+ sws.ctrlStream <- &pb.WatchResponse{
+ Header: sws.newResponseHeader(sws.watchStream.Rev()),
+ WatchId: id,
+ Canceled: true,
+ }
+ sws.mu.Lock()
+ delete(sws.progress, mvcc.WatchID(id))
+ delete(sws.prevKV, mvcc.WatchID(id))
+ sws.mu.Unlock()
+ }
+ }
+ default:
+ // we probably should not shutdown the entire stream when
+ // receive an valid command.
+ // so just do nothing instead.
+ continue
+ }
+ }
+}
+
+func (sws *serverWatchStream) sendLoop() {
+ // watch ids that are currently active
+ ids := make(map[mvcc.WatchID]struct{})
+ // watch responses pending on a watch id creation message
+ pending := make(map[mvcc.WatchID][]*pb.WatchResponse)
+
+ interval := GetProgressReportInterval()
+ progressTicker := time.NewTicker(interval)
+
+ defer func() {
+ progressTicker.Stop()
+ // drain the chan to clean up pending events
+ for ws := range sws.watchStream.Chan() {
+ mvcc.ReportEventReceived(len(ws.Events))
+ }
+ for _, wrs := range pending {
+ for _, ws := range wrs {
+ mvcc.ReportEventReceived(len(ws.Events))
+ }
+ }
+ }()
+
+ for {
+ select {
+ case wresp, ok := <-sws.watchStream.Chan():
+ if !ok {
+ return
+ }
+
+ // TODO: evs is []mvccpb.Event type
+ // either return []*mvccpb.Event from the mvcc package
+ // or define protocol buffer with []mvccpb.Event.
+ evs := wresp.Events
+ events := make([]*mvccpb.Event, len(evs))
+ sws.mu.Lock()
+ needPrevKV := sws.prevKV[wresp.WatchID]
+ sws.mu.Unlock()
+ for i := range evs {
+ events[i] = &evs[i]
+
+ if needPrevKV {
+ opt := mvcc.RangeOptions{Rev: evs[i].Kv.ModRevision - 1}
+ r, err := sws.watchable.Range(evs[i].Kv.Key, nil, opt)
+ if err == nil && len(r.KVs) != 0 {
+ events[i].PrevKv = &(r.KVs[0])
+ }
+ }
+ }
+
+ canceled := wresp.CompactRevision != 0
+ wr := &pb.WatchResponse{
+ Header: sws.newResponseHeader(wresp.Revision),
+ WatchId: int64(wresp.WatchID),
+ Events: events,
+ CompactRevision: wresp.CompactRevision,
+ Canceled: canceled,
+ }
+
+ if _, hasId := ids[wresp.WatchID]; !hasId {
+ // buffer if id not yet announced
+ wrs := append(pending[wresp.WatchID], wr)
+ pending[wresp.WatchID] = wrs
+ continue
+ }
+
+ mvcc.ReportEventReceived(len(evs))
+ if err := sws.gRPCStream.Send(wr); err != nil {
+ if isClientCtxErr(sws.gRPCStream.Context().Err(), err) {
+ plog.Debugf("failed to send watch response to gRPC stream (%q)", err.Error())
+ } else {
+ plog.Warningf("failed to send watch response to gRPC stream (%q)", err.Error())
+ }
+ return
+ }
+
+ sws.mu.Lock()
+ if len(evs) > 0 && sws.progress[wresp.WatchID] {
+ // elide next progress update if sent a key update
+ sws.progress[wresp.WatchID] = false
+ }
+ sws.mu.Unlock()
+
+ case c, ok := <-sws.ctrlStream:
+ if !ok {
+ return
+ }
+
+ if err := sws.gRPCStream.Send(c); err != nil {
+ if isClientCtxErr(sws.gRPCStream.Context().Err(), err) {
+ plog.Debugf("failed to send watch control response to gRPC stream (%q)", err.Error())
+ } else {
+ plog.Warningf("failed to send watch control response to gRPC stream (%q)", err.Error())
+ }
+ return
+ }
+
+ // track id creation
+ wid := mvcc.WatchID(c.WatchId)
+ if c.Canceled {
+ delete(ids, wid)
+ continue
+ }
+ if c.Created {
+ // flush buffered events
+ ids[wid] = struct{}{}
+ for _, v := range pending[wid] {
+ mvcc.ReportEventReceived(len(v.Events))
+ if err := sws.gRPCStream.Send(v); err != nil {
+ if isClientCtxErr(sws.gRPCStream.Context().Err(), err) {
+ plog.Debugf("failed to send pending watch response to gRPC stream (%q)", err.Error())
+ } else {
+ plog.Warningf("failed to send pending watch response to gRPC stream (%q)", err.Error())
+ }
+ return
+ }
+ }
+ delete(pending, wid)
+ }
+ case <-progressTicker.C:
+ sws.mu.Lock()
+ for id, ok := range sws.progress {
+ if ok {
+ sws.watchStream.RequestProgress(id)
+ }
+ sws.progress[id] = true
+ }
+ sws.mu.Unlock()
+ case <-sws.closec:
+ return
+ }
+ }
+}
+
+func (sws *serverWatchStream) close() {
+ sws.watchStream.Close()
+ close(sws.closec)
+ sws.wg.Wait()
+}
+
+func (sws *serverWatchStream) newResponseHeader(rev int64) *pb.ResponseHeader {
+ return &pb.ResponseHeader{
+ ClusterId: uint64(sws.clusterID),
+ MemberId: uint64(sws.memberID),
+ Revision: rev,
+ RaftTerm: sws.raftTimer.Term(),
+ }
+}
+
+func filterNoDelete(e mvccpb.Event) bool {
+ return e.Type == mvccpb.DELETE
+}
+
+func filterNoPut(e mvccpb.Event) bool {
+ return e.Type == mvccpb.PUT
+}
+
+func FiltersFromRequest(creq *pb.WatchCreateRequest) []mvcc.FilterFunc {
+ filters := make([]mvcc.FilterFunc, 0, len(creq.Filters))
+ for _, ft := range creq.Filters {
+ switch ft {
+ case pb.WatchCreateRequest_NOPUT:
+ filters = append(filters, filterNoPut)
+ case pb.WatchCreateRequest_NODELETE:
+ filters = append(filters, filterNoDelete)
+ default:
+ }
+ }
+ return filters
+}