mirror of
https://github.com/etcd-io/etcd.git
synced 2024-09-27 06:25:44 +00:00
clientv3: embed api version in metadata
ref. https://github.com/etcd-io/etcd/pull/11687 Signed-off-by: Gyuho Lee <leegyuho@amazon.com> clientv3: fix racy writes to context key === RUN TestWatchOverlapContextCancel ================== WARNING: DATA RACE Write at 0x00c42110dd40 by goroutine 99: runtime.mapassign() /usr/local/go/src/runtime/hashmap.go:485 +0x0 github.com/coreos/etcd/clientv3.metadataSet() /go/src/github.com/coreos/etcd/gopath/src/github.com/coreos/etcd/clientv3/ctx.go:61 +0x8c github.com/coreos/etcd/clientv3.withVersion() /go/src/github.com/coreos/etcd/gopath/src/github.com/coreos/etcd/clientv3/ctx.go:47 +0x137 github.com/coreos/etcd/clientv3.newStreamClientInterceptor.func1() /go/src/github.com/coreos/etcd/gopath/src/github.com/coreos/etcd/clientv3/client.go:309 +0x81 google.golang.org/grpc.NewClientStream() /go/src/github.com/coreos/etcd/gopath/src/google.golang.org/grpc/stream.go:101 +0x10e github.com/coreos/etcd/etcdserver/etcdserverpb.(*watchClient).Watch() /go/src/github.com/coreos/etcd/gopath/src/github.com/coreos/etcd/etcdserver/etcdserverpb/rpc.pb.go:3193 +0xe9 github.com/coreos/etcd/clientv3.(*watchGrpcStream).openWatchClient() /go/src/github.com/coreos/etcd/gopath/src/github.com/coreos/etcd/clientv3/watch.go:788 +0x143 github.com/coreos/etcd/clientv3.(*watchGrpcStream).newWatchClient() /go/src/github.com/coreos/etcd/gopath/src/github.com/coreos/etcd/clientv3/watch.go:700 +0x5c3 github.com/coreos/etcd/clientv3.(*watchGrpcStream).run() /go/src/github.com/coreos/etcd/gopath/src/github.com/coreos/etcd/clientv3/watch.go:431 +0x12b Previous read at 0x00c42110dd40 by goroutine 130: reflect.maplen() /usr/local/go/src/runtime/hashmap.go:1165 +0x0 reflect.Value.MapKeys() /usr/local/go/src/reflect/value.go:1090 +0x43b fmt.(*pp).printValue() /usr/local/go/src/fmt/print.go:741 +0x1885 fmt.(*pp).printArg() /usr/local/go/src/fmt/print.go:682 +0x1b1 fmt.(*pp).doPrintf() /usr/local/go/src/fmt/print.go:998 +0x1cad fmt.Sprintf() /usr/local/go/src/fmt/print.go:196 +0x77 github.com/coreos/etcd/clientv3.streamKeyFromCtx() /go/src/github.com/coreos/etcd/gopath/src/github.com/coreos/etcd/clientv3/watch.go:825 +0xc8 github.com/coreos/etcd/clientv3.(*watcher).Watch() /go/src/github.com/coreos/etcd/gopath/src/github.com/coreos/etcd/clientv3/watch.go:265 +0x426 github.com/coreos/etcd/clientv3/integration.testWatchOverlapContextCancel.func1() /go/src/github.com/coreos/etcd/gopath/src/github.com/coreos/etcd/clientv3/integration/watch_test.go:959 +0x23e Goroutine 99 (running) created at: github.com/coreos/etcd/clientv3.(*watcher).newWatcherGrpcStream() /go/src/github.com/coreos/etcd/gopath/src/github.com/coreos/etcd/clientv3/watch.go:236 +0x59d github.com/coreos/etcd/clientv3.(*watcher).Watch() /go/src/github.com/coreos/etcd/gopath/src/github.com/coreos/etcd/clientv3/watch.go:278 +0xbb6 github.com/coreos/etcd/clientv3/integration.testWatchOverlapContextCancel.func1() /go/src/github.com/coreos/etcd/gopath/src/github.com/coreos/etcd/clientv3/integration/watch_test.go:959 +0x23e Goroutine 130 (running) created at: github.com/coreos/etcd/clientv3/integration.testWatchOverlapContextCancel() /go/src/github.com/coreos/etcd/gopath/src/github.com/coreos/etcd/clientv3/integration/watch_test.go:979 +0x76d github.com/coreos/etcd/clientv3/integration.TestWatchOverlapContextCancel() /go/src/github.com/coreos/etcd/gopath/src/github.com/coreos/etcd/clientv3/integration/watch_test.go:922 +0x44 testing.tRunner() /usr/local/go/src/testing/testing.go:657 +0x107 ================== Signed-off-by: Gyuho Lee <leegyuho@amazon.com>
This commit is contained in:
Gyuho Lee
parent
42d749057d
commit
20b27a887d
@ -32,7 +32,6 @@ import (
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/credentials"
|
||||
"google.golang.org/grpc/keepalive"
|
||||
"google.golang.org/grpc/metadata"
|
||||
"google.golang.org/grpc/status"
|
||||
)
|
||||
|
||||
@ -269,7 +268,11 @@ func (c *Client) dialSetupOpts(endpoint string, dopts ...grpc.DialOption) (opts
|
||||
}
|
||||
return conn, err
|
||||
}
|
||||
opts = append(opts, grpc.WithDialer(f))
|
||||
opts = append(opts,
|
||||
grpc.WithDialer(f),
|
||||
grpc.WithUnaryInterceptor(newUnaryClientInterceptor()),
|
||||
grpc.WithStreamInterceptor(newStreamClientInterceptor()),
|
||||
)
|
||||
|
||||
creds := c.creds
|
||||
if _, _, scheme := parseEndpoint(endpoint); len(scheme) != 0 {
|
||||
@ -284,6 +287,30 @@ func (c *Client) dialSetupOpts(endpoint string, dopts ...grpc.DialOption) (opts
|
||||
return opts
|
||||
}
|
||||
|
||||
func newUnaryClientInterceptor() grpc.UnaryClientInterceptor {
|
||||
return func(ctx context.Context,
|
||||
method string,
|
||||
req, reply interface{},
|
||||
cc *grpc.ClientConn,
|
||||
invoker grpc.UnaryInvoker,
|
||||
opts ...grpc.CallOption) error {
|
||||
ctx = withVersion(ctx)
|
||||
return invoker(ctx, method, req, reply, cc, opts...)
|
||||
}
|
||||
}
|
||||
|
||||
func newStreamClientInterceptor() grpc.StreamClientInterceptor {
|
||||
return func(ctx context.Context,
|
||||
desc *grpc.StreamDesc,
|
||||
cc *grpc.ClientConn,
|
||||
method string,
|
||||
streamer grpc.Streamer,
|
||||
opts ...grpc.CallOption) (grpc.ClientStream, error) {
|
||||
ctx = withVersion(ctx)
|
||||
return streamer(ctx, desc, cc, method, opts...)
|
||||
}
|
||||
}
|
||||
|
||||
// Dial connects to a single endpoint using the client's config.
|
||||
func (c *Client) Dial(endpoint string) (*grpc.ClientConn, error) {
|
||||
return c.dial(endpoint)
|
||||
@ -356,13 +383,6 @@ func (c *Client) dial(endpoint string, dopts ...grpc.DialOption) (*grpc.ClientCo
|
||||
return conn, nil
|
||||
}
|
||||
|
||||
// WithRequireLeader requires client requests to only succeed
|
||||
// when the cluster has a leader.
|
||||
func WithRequireLeader(ctx context.Context) context.Context {
|
||||
md := metadata.Pairs(rpctypes.MetadataRequireLeaderKey, rpctypes.MetadataHasLeader)
|
||||
return metadata.NewOutgoingContext(ctx, md)
|
||||
}
|
||||
|
||||
func newClient(cfg *Config) (*Client, error) {
|
||||
if cfg == nil {
|
||||
cfg = &Config{}
|
||||
|
||||
64
clientv3/ctx.go
Normal file
64
clientv3/ctx.go
Normal file
@ -0,0 +1,64 @@
|
||||
// Copyright 2020 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 clientv3
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
|
||||
"github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
|
||||
"github.com/coreos/etcd/version"
|
||||
"google.golang.org/grpc/metadata"
|
||||
)
|
||||
|
||||
// WithRequireLeader requires client requests to only succeed
|
||||
// when the cluster has a leader.
|
||||
func WithRequireLeader(ctx context.Context) context.Context {
|
||||
md, ok := metadata.FromOutgoingContext(ctx)
|
||||
if !ok { // no outgoing metadata ctx key, create one
|
||||
md = metadata.Pairs(rpctypes.MetadataRequireLeaderKey, rpctypes.MetadataHasLeader)
|
||||
return metadata.NewOutgoingContext(ctx, md)
|
||||
}
|
||||
copied := md.Copy() // avoid racey updates
|
||||
// overwrite/add 'hasleader' key/value
|
||||
metadataSet(copied, rpctypes.MetadataRequireLeaderKey, rpctypes.MetadataHasLeader)
|
||||
return metadata.NewOutgoingContext(ctx, copied)
|
||||
}
|
||||
|
||||
// embeds client version
|
||||
func withVersion(ctx context.Context) context.Context {
|
||||
md, ok := metadata.FromOutgoingContext(ctx)
|
||||
if !ok { // no outgoing metadata ctx key, create one
|
||||
md = metadata.Pairs(rpctypes.MetadataClientAPIVersionKey, version.APIVersion)
|
||||
return metadata.NewOutgoingContext(ctx, md)
|
||||
}
|
||||
copied := md.Copy() // avoid racey updates
|
||||
// overwrite/add version key/value
|
||||
metadataSet(copied, rpctypes.MetadataClientAPIVersionKey, version.APIVersion)
|
||||
return metadata.NewOutgoingContext(ctx, copied)
|
||||
}
|
||||
|
||||
func metadataGet(md metadata.MD, k string) []string {
|
||||
k = strings.ToLower(k)
|
||||
return md[k]
|
||||
}
|
||||
|
||||
func metadataSet(md metadata.MD, k string, vals ...string) {
|
||||
if len(vals) == 0 {
|
||||
return
|
||||
}
|
||||
k = strings.ToLower(k)
|
||||
md[k] = vals
|
||||
}
|
||||
67
clientv3/ctx_test.go
Normal file
67
clientv3/ctx_test.go
Normal file
@ -0,0 +1,67 @@
|
||||
// Copyright 2020 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 clientv3
|
||||
|
||||
import (
|
||||
"context"
|
||||
"reflect"
|
||||
"testing"
|
||||
|
||||
"github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
|
||||
"github.com/coreos/etcd/version"
|
||||
"google.golang.org/grpc/metadata"
|
||||
)
|
||||
|
||||
func TestMetadataWithRequireLeader(t *testing.T) {
|
||||
ctx := context.TODO()
|
||||
md, ok := metadata.FromOutgoingContext(ctx)
|
||||
if ok {
|
||||
t.Fatal("expected no outgoing metadata ctx key")
|
||||
}
|
||||
|
||||
// add a conflicting key with some other value
|
||||
md = metadata.Pairs(rpctypes.MetadataRequireLeaderKey, "invalid")
|
||||
// add a key, and expect not be overwritten
|
||||
metadataSet(md, "hello", "1", "2")
|
||||
ctx = metadata.NewOutgoingContext(ctx, md)
|
||||
|
||||
// expect overwrites but still keep other keys
|
||||
ctx = WithRequireLeader(ctx)
|
||||
md, ok = metadata.FromOutgoingContext(ctx)
|
||||
if !ok {
|
||||
t.Fatal("expected outgoing metadata ctx key")
|
||||
}
|
||||
if ss := metadataGet(md, rpctypes.MetadataRequireLeaderKey); !reflect.DeepEqual(ss, []string{rpctypes.MetadataHasLeader}) {
|
||||
t.Fatalf("unexpected metadata for %q %v", rpctypes.MetadataRequireLeaderKey, ss)
|
||||
}
|
||||
if ss := metadataGet(md, "hello"); !reflect.DeepEqual(ss, []string{"1", "2"}) {
|
||||
t.Fatalf("unexpected metadata for 'hello' %v", ss)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMetadataWithClientAPIVersion(t *testing.T) {
|
||||
ctx := withVersion(WithRequireLeader(context.TODO()))
|
||||
|
||||
md, ok := metadata.FromOutgoingContext(ctx)
|
||||
if !ok {
|
||||
t.Fatal("expected outgoing metadata ctx key")
|
||||
}
|
||||
if ss := metadataGet(md, rpctypes.MetadataRequireLeaderKey); !reflect.DeepEqual(ss, []string{rpctypes.MetadataHasLeader}) {
|
||||
t.Fatalf("unexpected metadata for %q %v", rpctypes.MetadataRequireLeaderKey, ss)
|
||||
}
|
||||
if ss := metadataGet(md, rpctypes.MetadataClientAPIVersionKey); !reflect.DeepEqual(ss, []string{version.APIVersion}) {
|
||||
t.Fatalf("unexpected metadata for %q %v", rpctypes.MetadataClientAPIVersionKey, ss)
|
||||
}
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user