From 5088ae3e67e5035cbf13259ab1a2c75a45f24b94 Mon Sep 17 00:00:00 2001 From: Anthony Romano Date: Wed, 26 Jul 2017 16:56:29 -0700 Subject: [PATCH 1/3] docs: add ordering wrapper as experimental feature --- Documentation/dev-guide/experimental_apis.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Documentation/dev-guide/experimental_apis.md b/Documentation/dev-guide/experimental_apis.md index aefaeebceb1..dde2f09d6f6 100644 --- a/Documentation/dev-guide/experimental_apis.md +++ b/Documentation/dev-guide/experimental_apis.md @@ -4,4 +4,4 @@ For the most part, the etcd project is stable, but we are still moving fast! We ## The current experimental API/features are: -(none currently) +- [KV ordering](https://godoc.org/github.com/coreos/etcd/clientv3/ordering) wrapper. When an etcd client switches endpoints, responses to serializable reads may go backward in time if the new endpoint is lagging behind the rest of the cluster. The ordering wrapper caches the current cluster revision from response headers. If a response revision is less than the cached revision, the client selects another endpoint and reissues the read. Enable in grpcproxy with `--experimental-serializable-ordering`. From fca56f132a2eae42158f29e78787236a5a24e074 Mon Sep 17 00:00:00 2001 From: Anthony Romano Date: Wed, 26 Jul 2017 17:08:35 -0700 Subject: [PATCH 2/3] ordering: use x/net/context and add doc.go Compilation troubles when using the "context" package. --- clientv3/ordering/doc.go | 42 ++++++++++++++++++++++++++++++++++++++++ clientv3/ordering/kv.go | 2 +- 2 files changed, 43 insertions(+), 1 deletion(-) create mode 100644 clientv3/ordering/doc.go diff --git a/clientv3/ordering/doc.go b/clientv3/ordering/doc.go new file mode 100644 index 00000000000..856f3305801 --- /dev/null +++ b/clientv3/ordering/doc.go @@ -0,0 +1,42 @@ +// 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 ordering is a clientv3 wrapper that caches response header revisions +// to detect ordering violations from stale responses. Users may define a +// policy on how to handle the ordering violation, but typically the client +// should connect to another endpoint and reissue the request. +// +// The most common situation where an ordering violation happens is a client +// reconnects to a partitioned member and issues a serializable read. Since the +// partitioned member is likely behind the last member, it may return a Get +// response based on a store revision older than the store revision used to +// service a prior Get on the former endpoint. +// +// First, create a client: +// +// cli, err := clientv3.New(clientv3.Config{Endpoints: []string{"localhost:2379"}}) +// if err != nil { +// // handle error! +// } +// +// Next, override the client interface with the ordering wrapper: +// +// vf := func(op clientv3.Op, resp clientv3.OpResponse, prevRev int64) error { +// return fmt.Errorf("ordering: issued %+v, got %+v, expected rev=%v", op, resp, prevRev) +// } +// cli.KV = ordering.NewKV(cli.KV, vf) +// +// Now calls using 'cli' will reject order violations with an error. +// +package ordering diff --git a/clientv3/ordering/kv.go b/clientv3/ordering/kv.go index e8bf07b8c74..0001ce0131c 100644 --- a/clientv3/ordering/kv.go +++ b/clientv3/ordering/kv.go @@ -15,10 +15,10 @@ package ordering import ( - "context" "sync" "github.com/coreos/etcd/clientv3" + "golang.org/x/net/context" ) // kvOrdering ensures that serialized requests do not return From f6acd0316cf3d988d7b4a52af48f7a14c5ea2021 Mon Sep 17 00:00:00 2001 From: Anthony Romano Date: Wed, 26 Jul 2017 17:10:10 -0700 Subject: [PATCH 3/3] etcdmain: add --experimental-serializable-ordering to grpc proxy Connect to another endpoint on stale reads. --- etcdmain/grpc_proxy.go | 22 +++++++++++++++++++++- 1 file changed, 21 insertions(+), 1 deletion(-) diff --git a/etcdmain/grpc_proxy.go b/etcdmain/grpc_proxy.go index 4cc7910e164..bee6755ff64 100644 --- a/etcdmain/grpc_proxy.go +++ b/etcdmain/grpc_proxy.go @@ -15,6 +15,7 @@ package etcdmain import ( + "context" "fmt" "math" "net" @@ -26,6 +27,7 @@ import ( "github.com/coreos/etcd/clientv3" "github.com/coreos/etcd/clientv3/namespace" + "github.com/coreos/etcd/clientv3/ordering" "github.com/coreos/etcd/etcdserver/api/etcdhttp" "github.com/coreos/etcd/etcdserver/api/v3election/v3electionpb" "github.com/coreos/etcd/etcdserver/api/v3lock/v3lockpb" @@ -69,7 +71,8 @@ var ( grpcProxyNamespace string - grpcProxyEnablePprof bool + grpcProxyEnablePprof bool + grpcProxyEnableOrdering bool ) func init() { @@ -119,6 +122,9 @@ func newGRPCProxyStartCommand() *cobra.Command { cmd.Flags().BoolVar(&grpcProxyListenAutoTLS, "auto-tls", false, "proxy TLS using generated certificates") cmd.Flags().StringVar(&grpcProxyListenCRL, "client-crl-file", "", "proxy client certificate revocation list file.") + // experimental flags + cmd.Flags().BoolVar(&grpcProxyEnableOrdering, "experimental-serializable-ordering", false, "Ensure serializable reads have monotonically increasing store revisions across endpoints.") + return &cmd } @@ -255,6 +261,20 @@ func mustListenCMux(tlsinfo *transport.TLSInfo) cmux.CMux { } func newGRPCProxyServer(client *clientv3.Client) *grpc.Server { + if grpcProxyEnableOrdering { + vf := ordering.NewOrderViolationSwitchEndpointClosure(*client) + client.KV = ordering.NewKV(client.KV, vf) + plog.Infof("waiting for linearized read from cluster to recover ordering") + for { + _, err := client.KV.Get(context.TODO(), "_", clientv3.WithKeysOnly()) + if err == nil { + break + } + plog.Warningf("ordering recovery failed, retrying in 1s (%v)", err) + time.Sleep(time.Second) + } + } + if len(grpcProxyNamespace) > 0 { client.KV = namespace.NewKV(client.KV, grpcProxyNamespace) client.Watcher = namespace.NewWatcher(client.Watcher, grpcProxyNamespace)