Merge pull request #8315 from heyitsanthony/experimental-ordering

add experimental serializable ordering feature to grpcproxy

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`.

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

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

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)